#!/usr/bin/env python3 """ ================================================================ Hybrid-Cloud Security Scanner -- with Standards-Mapped Fixes ================================================================ Scans every S3 bucket, every EC2 security group, every IAM user, and the account's CloudTrail configuration. For each misconfiguration found, prints: - what is wrong - severity - the exact NIST SP 800-53 Rev.5 control and CSA Top Threats 2025 category it relates to - the exact AWS CLI command to fix it (where a single command can fix it; MFA enrolment cannot be fully scripted, and this is noted explicitly rather than glossed over) This script is READ-ONLY -- it never changes anything. Use the suggested commands yourself, or apply them by hand. USAGE ----- python scan_and_fix.py python scan_and_fix.py --json report.json ================================================================ """ import boto3 import json import argparse from datetime import datetime from botocore.exceptions import ClientError REGION = 'eu-west-2' # ---------------------------------------------------------------- # Standards catalogue -- every issue type maps to a severity, # a NIST SP 800-53 Rev.5 control, a CSA Top Threats 2025 # category, and a copy-paste-ready fix command. # ---------------------------------------------------------------- CATALOGUE = { 's3_public_access': { 'severity': 'HIGH', 'description': 'Bucket allows public access (Block Public Access is off)', 'nist': 'SC-7 (Boundary Protection), AC-3 (Access Enforcement)', 'csa': 'Misconfiguration & Inadequate Change Control', 'fix': ('aws s3api put-public-access-block --bucket {r} ' '--public-access-block-configuration ' 'BlockPublicAcls=true,IgnorePublicAcls=true,' 'BlockPublicPolicy=true,RestrictPublicBuckets=true'), }, 's3_public_policy': { 'severity': 'HIGH', 'description': 'Bucket policy grants access to everyone (Principal: *)', 'nist': 'AC-3 (Access Enforcement), AC-6 (Least Privilege)', 'csa': 'Misconfiguration & Inadequate Change Control', 'fix': 'aws s3api delete-bucket-policy --bucket {r}', }, 's3_no_encryption': { 'severity': 'MEDIUM', 'description': 'No server-side encryption (SSE-KMS) configured', 'nist': 'SC-28 (Protection of Information at Rest)', 'csa': 'Insufficient Data Protection', 'fix': ('aws s3api put-bucket-encryption --bucket {r} ' '--server-side-encryption-configuration \'{{"Rules":[' '{{"ApplyServerSideEncryptionByDefault":' '{{"SSEAlgorithm":"aws:kms"}}}}]}}\''), }, 's3_no_versioning': { 'severity': 'LOW', 'description': 'Versioning disabled - no recovery from accidental deletion/overwrite', 'nist': 'SI-12 (Information Management and Retention), CP-9 (Backup)', 'csa': 'Insufficient Data Protection', 'fix': 'aws s3api put-bucket-versioning --bucket {r} --versioning-configuration Status=Enabled', }, 'ec2_open_ssh': { 'severity': 'HIGH', 'description': 'SSH (port 22) open to the entire internet (0.0.0.0/0)', 'nist': 'AC-17 (Remote Access), SC-7 (Boundary Protection)', 'csa': 'Insecure Interfaces and APIs', 'fix': ('aws ec2 revoke-security-group-ingress --group-id {r} ' '--protocol tcp --port 22 --cidr 0.0.0.0/0'), }, 'ec2_open_rdp': { 'severity': 'HIGH', 'description': 'RDP (port 3389) open to the entire internet (0.0.0.0/0)', 'nist': 'AC-17 (Remote Access), SC-7 (Boundary Protection)', 'csa': 'Insecure Interfaces and APIs', 'fix': ('aws ec2 revoke-security-group-ingress --group-id {r} ' '--protocol tcp --port 3389 --cidr 0.0.0.0/0'), }, 'ec2_open_all_traffic': { 'severity': 'HIGH', 'description': 'ALL traffic on ALL ports open to the entire internet', 'nist': 'AC-17 (Remote Access), SC-7 (Boundary Protection)', 'csa': 'Insecure Interfaces and APIs', 'fix': ('aws ec2 revoke-security-group-ingress --group-id {r} ' '--protocol -1 --cidr 0.0.0.0/0'), }, 'ec2_open_port': { 'severity': 'MEDIUM', 'description': 'Non-standard port {port} open to the entire internet', 'nist': 'AC-17 (Remote Access), SC-7 (Boundary Protection)', 'csa': 'Insecure Interfaces and APIs', 'fix': ('aws ec2 revoke-security-group-ingress --group-id {r} ' '--protocol tcp --port {port} --cidr 0.0.0.0/0'), }, 'ec2_instance_exposed': { 'severity': 'HIGH', 'description': 'Running instance is attached to a security group with open rules', 'nist': 'CM-7 (Least Functionality), SC-7 (Boundary Protection)', 'csa': 'Insecure Interfaces and APIs', 'fix': ('aws ec2 modify-instance-attribute --instance-id {r} ' '--groups '), }, 'iam_excessive_permissions': { 'severity': 'HIGH', 'description': 'User has AdministratorAccess attached (full account privileges)', 'nist': 'AC-6 (Least Privilege), AC-2 (Account Management)', 'csa': 'Identity and Access Management Weaknesses', 'fix': ('aws iam detach-user-policy --user-name {r} ' '--policy-arn arn:aws:iam::aws:policy/AdministratorAccess'), }, 'iam_no_mfa': { 'severity': 'HIGH', 'description': 'User has no MFA device registered', 'nist': 'IA-2 (Identification and Authentication), IA-5 (Authenticator Management)', 'csa': 'Identity and Access Management Weaknesses', 'fix': ('aws iam create-virtual-mfa-device --virtual-mfa-device-name {r}-mfa ' '--outfile {r}-qrcode.png --bootstrap-method QRCodePNG ' '# then: aws iam enable-mfa-device (needs two authenticator codes; ' 'cannot be fully scripted without a human enrolling the device)'), }, 'cloudtrail_disabled': { 'severity': 'LOW', 'description': 'No CloudTrail trail is actively logging in this account', 'nist': 'AU-2 (Event Logging), AU-12 (Audit Record Generation)', 'csa': 'Misconfiguration & Inadequate Change Control', 'fix': ('aws cloudtrail create-trail --name org-trail --s3-bucket-name ' '--is-multi-region-trail && aws cloudtrail start-logging --name org-trail'), }, } RISK_WEIGHT = {'HIGH': 3, 'MEDIUM': 2, 'LOW': 1} def rule(char='-', width=78): print(char * width) def make_issue(issue_type, resource, extra=None): """Build a formatted issue dict from the catalogue.""" entry = CATALOGUE[issue_type] desc = entry['description'] fix = entry['fix'] if extra: desc = desc.format(**extra) fix = fix.format(r=resource, **extra) else: fix = fix.format(r=resource) return { 'resource': resource, 'type': issue_type, 'severity': entry['severity'], 'description': desc, 'nist': entry['nist'], 'csa': entry['csa'], 'fix': fix, } def scan_s3(s3): print("SCANNING S3 BUCKETS") rule() issues = [] buckets = s3.list_buckets()['Buckets'] if not buckets: print(" No S3 buckets found.\n") return issues for b in buckets: name = b['Name'] found_here = [] # Block Public Access try: cfg = s3.get_public_access_block(Bucket=name)['PublicAccessBlockConfiguration'] if not all(cfg.values()): found_here.append(make_issue('s3_public_access', name)) except ClientError as e: if e.response['Error']['Code'] == 'NoSuchPublicAccessBlockConfiguration': found_here.append(make_issue('s3_public_access', name)) # Public bucket policy try: policy = json.loads(s3.get_bucket_policy(Bucket=name)['Policy']) for stmt in policy.get('Statement', []): if stmt.get('Effect') == 'Allow' and stmt.get('Principal') in ('*', {'AWS': '*'}): found_here.append(make_issue('s3_public_policy', name)) break except ClientError as e: if e.response['Error']['Code'] != 'NoSuchBucketPolicy': pass # Encryption try: enc = s3.get_bucket_encryption(Bucket=name)['ServerSideEncryptionConfiguration'] has_kms = any( rule_.get('ApplyServerSideEncryptionByDefault', {}).get('SSEAlgorithm') == 'aws:kms' for rule_ in enc.get('Rules', []) ) if not has_kms: found_here.append(make_issue('s3_no_encryption', name)) except ClientError as e: if e.response['Error']['Code'] in ( 'ServerSideEncryptionConfigurationNotFoundError', 'NoSuchEncryptionConfiguration'): found_here.append(make_issue('s3_no_encryption', name)) # Versioning v = s3.get_bucket_versioning(Bucket=name) if v.get('Status') != 'Enabled': found_here.append(make_issue('s3_no_versioning', name)) if found_here: print(f" [VULNERABLE] {name}") for i in found_here: print(f" [{i['severity']:6}] {i['description']}") else: print(f" [OK] {name} - no issues found") issues.extend(found_here) print() return issues def scan_ec2(ec2): print("SCANNING EC2 SECURITY GROUPS") rule() issues = [] sgs = ec2.describe_security_groups()['SecurityGroups'] # Build a lookup of which instances use which security groups instances = ec2.describe_instances() sg_to_instances = {} for res in instances.get('Reservations', []): for inst in res['Instances']: if inst['State']['Name'] not in ('running', 'pending'): continue for sg in inst.get('SecurityGroups', []): sg_to_instances.setdefault(sg['GroupId'], []).append(inst['InstanceId']) for sg in sgs: if sg['GroupName'] == 'default': continue found_here = [] for perm in sg.get('IpPermissions', []): proto = perm.get('IpProtocol', '') from_port = perm.get('FromPort', 0) open_world = any(r.get('CidrIp') == '0.0.0.0/0' for r in perm.get('IpRanges', [])) if not open_world: continue if proto == '-1': found_here.append(make_issue('ec2_open_all_traffic', sg['GroupId'])) elif from_port == 22: found_here.append(make_issue('ec2_open_ssh', sg['GroupId'])) elif from_port == 3389: found_here.append(make_issue('ec2_open_rdp', sg['GroupId'])) elif from_port not in (80, 443): found_here.append(make_issue('ec2_open_port', sg['GroupId'], {'port': from_port})) if found_here: print(f" [VULNERABLE] {sg['GroupName']} ({sg['GroupId']})") for i in found_here: print(f" [{i['severity']:6}] {i['description']}") # Check for attached running instances - raises real-world severity attached = sg_to_instances.get(sg['GroupId'], []) for inst_id in attached: issue = make_issue('ec2_instance_exposed', inst_id) print(f" [{issue['severity']:6}] Instance {inst_id} is running behind this open group") found_here.append(issue) else: print(f" [OK] {sg['GroupName']} ({sg['GroupId']}) - no open rules") issues.extend(found_here) print() return issues def scan_iam(iam): """ Scans every IAM user for two conditions: an attached AdministratorAccess policy, and the absence of any MFA device. Both are reported per user, since a user can carry either, neither, or both at once. """ print("SCANNING IAM USERS") rule() issues = [] users = iam.list_users()['Users'] if not users: print(" No IAM users found.\n") return issues for u in users: name = u['UserName'] found_here = [] attached = iam.list_attached_user_policies(UserName=name)['AttachedPolicies'] if any(p['PolicyArn'].endswith('/AdministratorAccess') for p in attached): found_here.append(make_issue('iam_excessive_permissions', name)) mfa_devices = iam.list_mfa_devices(UserName=name)['MFADevices'] if not mfa_devices: found_here.append(make_issue('iam_no_mfa', name)) if found_here: print(f" [VULNERABLE] {name}") for i in found_here: print(f" [{i['severity']:6}] {i['description']}") else: print(f" [OK] {name} - no issues found") issues.extend(found_here) print() return issues def scan_cloudtrail(cloudtrail): """ Checks whether any CloudTrail trail in the account is actively logging. An account can have a trail configured but stopped, which is functionally the same exposure as having none at all, so both states are reported as the same finding. """ print("SCANNING CLOUDTRAIL") rule() issues = [] trails = cloudtrail.describe_trails()['trailList'] active_trails = [] for t in trails: status = cloudtrail.get_trail_status(Name=t['TrailARN']) if status.get('IsLogging'): active_trails.append(t['Name']) if not active_trails: issue = make_issue('cloudtrail_disabled', 'account-wide') print(f" [VULNERABLE] No active trail in this account") print(f" [{issue['severity']:6}] {issue['description']}") issues.append(issue) else: print(f" [OK] {len(active_trails)} active trail(s): {', '.join(active_trails)}") print() return issues def print_summary(issues): print("=" * 78) print("SUMMARY & STANDARDS-MAPPED REMEDIATION PLAN") print("=" * 78) if not issues: print("\nNo issues found. Environment is compliant.\n") return high = [i for i in issues if i['severity'] == 'HIGH'] medium = [i for i in issues if i['severity'] == 'MEDIUM'] low = [i for i in issues if i['severity'] == 'LOW'] ders = sum(RISK_WEIGHT[i['severity']] for i in issues) print(f"\nTotal issues: {len(issues)} (HIGH: {len(high)} MEDIUM: {len(medium)} LOW: {len(low)})") print(f"Data Exposure Risk Score (DERS): {ders} (sum of issues x severity weight)\n") for idx, i in enumerate(issues, 1): rule() print(f"Issue {idx}/{len(issues)} - [{i['severity']}]") print(f" Resource : {i['resource']}") print(f" Problem : {i['description']}") print(f" NIST 800-53 : {i['nist']}") print(f" CSA category : {i['csa']}") print(f" Suggested fix:") print(f" {i['fix']}") rule('=') print("\nThese commands are SUGGESTIONS ONLY - this scanner never changes") print("anything. Review each command before running it.\n") def main(): parser = argparse.ArgumentParser(description='Hybrid-cloud security scanner (S3, EC2, IAM, CloudTrail) with standards-mapped fixes') parser.add_argument('--json', metavar='FILE', help='Save findings to a JSON file') args = parser.parse_args() session = boto3.Session(region_name=REGION) s3 = session.client('s3') ec2 = session.client('ec2') iam = session.client('iam') cloudtrail = session.client('cloudtrail') print("=" * 78) print("HYBRID-CLOUD SECURITY SCANNER") print(f"Region: {REGION} | Scanned: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("=" * 78 + "\n") all_issues = [] all_issues.extend(scan_s3(s3)) all_issues.extend(scan_ec2(ec2)) all_issues.extend(scan_iam(iam)) all_issues.extend(scan_cloudtrail(cloudtrail)) print_summary(all_issues) if args.json: with open(args.json, 'w') as f: json.dump({ 'scanned_at': datetime.now().isoformat(), 'region': REGION, 'total_issues': len(all_issues), 'ders': sum(RISK_WEIGHT[i['severity']] for i in all_issues), 'issues': all_issues, }, f, indent=2) print(f"Findings saved to {args.json}") if __name__ == '__main__': main()