Build an AI AWS Security Auditor with Claude API and Boto3

pip install anthropic boto3

export AWS_DEFAULT_REGION=us-east-1
export ANTHROPIC_API_KEY=sk-ant-...

import boto3
import json
import anthropic
from datetime import datetime
 
def audit_s3_public_buckets():
    s3 = boto3.client('s3')
    findings = []
    buckets = s3.list_buckets().get('Buckets', [])
 
    for bucket in buckets:
        name = bucket['Name']
        try:
            acl = s3.get_public_access_block(Bucket=name)
            config = acl['PublicAccessBlockConfiguration']
            if not all([
                config.get('BlockPublicAcls', False),
                config.get('BlockPublicPolicy', False),
                config.get('IgnorePublicAcls', False),
                config.get('RestrictPublicBuckets', False)
            ]):
                findings.append({
                    "resource": f"s3://{name}",
                    "issue": "Public access block not fully enabled",
                    "severity": "HIGH"
                })
        except s3.exceptions.NoSuchPublicAccessBlockConfiguration:
            findings.append({
                "resource": f"s3://{name}",
                "issue": "No public access block configuration set",
                "severity": "HIGH"
            })
        except Exception:
            pass
 
    return findings
 
 
def audit_security_groups():
    ec2 = boto3.client('ec2')
    findings = []
    paginator = ec2.get_paginator('describe_security_groups')
 
    for page in paginator.paginate():
        for sg in page['SecurityGroups']:
            for rule in sg.get('IpPermissions', []):
                for ip_range in rule.get('IpRanges', []):
                    if ip_range.get('CidrIp') == '0.0.0.0/0':
                        port = rule.get('FromPort', 'ALL')
                        findings.append({
                            "resource": f"sg/{sg['GroupId']} ({sg['GroupName']})",
                            "issue": f"Inbound 0.0.0.0/0 on port {port}",
                            "severity": "HIGH" if port in [22, 3389, 3306, 5432] else "MEDIUM"
                        })
 
    return findings
 
 
def audit_iam_mfa():
    iam = boto3.client('iam')
    findings = []
    paginator = iam.get_paginator('list_users')
 
    for page in paginator.paginate():
        for user in page['Users']:
            username = user['UserName']
            mfa_devices = iam.list_mfa_devices(UserName=username)['MFADevices']
            if not mfa_devices:
                # Only flag users that have console access
                try:
                    iam.get_login_profile(UserName=username)
                    findings.append({
                        "resource": f"iam/user/{username}",
                        "issue": "Console access without MFA",
                        "severity": "CRITICAL"
                    })
                except iam.exceptions.NoSuchEntityException:
                    pass  # No console access, skip
 
    return findings
 
 
def audit_rds_public():
    rds = boto3.client('rds')
    findings = []
    paginator = rds.get_paginator('describe_db_instances')
 
    for page in paginator.paginate():
        for db in page['DBInstances']:
            if db.get('PubliclyAccessible', False):
                findings.append({
                    "resource": f"rds/{db['DBInstanceIdentifier']}",
                    "issue": f"RDS instance publicly accessible ({db['Engine']})",
                    "severity": "CRITICAL"
                })
 
    return findings
 
 
def audit_cloudtrail():
    ct = boto3.client('cloudtrail')
    findings = []
 
    try:
        trails = ct.describe_trails(includeShadowTrails=False)['trailList']
        if not trails:
            findings.append({
                "resource": "cloudtrail",
                "issue": "No CloudTrail trails configured",
                "severity": "HIGH"
            })
        else:
            for trail in trails:
                status = ct.get_trail_status(Name=trail['TrailARN'])
                if not status.get('IsLogging', False):
                    findings.append({
                        "resource": f"cloudtrail/{trail['Name']}",
                        "issue": "CloudTrail trail exists but logging is disabled",
                        "severity": "HIGH"
                    })
    except Exception as e:
        findings.append({
            "resource": "cloudtrail",
            "issue": f"Could not check CloudTrail: {str(e)}",
            "severity": "UNKNOWN"
        })
 
    return findings
 
 
def analyse_with_claude(findings: list) -> str:
    client = anthropic.Anthropic()
 
    findings_text = json.dumps(findings, indent=2)
 
    message = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=2048,
        messages=[
            {
                "role": "user",
                "content": f"""You are an AWS security expert. Analyse these security findings from an automated AWS audit and provide:
 
1. Executive summary (2-3 sentences for a non-technical manager)
2. Top 3 critical risks to fix immediately, with specific AWS CLI or console steps to remediate each
3. Quick wins that can be fixed in under 10 minutes
4. Overall security score out of 10
 
Findings:
{findings_text}
 
Be specific and actionable. Include the exact AWS CLI commands where possible."""
            }
        ]
    )
 
    return message.content[0].text
 
 
def run_audit():
    print(f"Starting AWS Security Audit at {datetime.now().isoformat()}")
    print("=" * 60)
 
    all_findings = []
 
    print("Checking S3 public access...")
    all_findings.extend(audit_s3_public_buckets())
 
    print("Checking security groups...")
    all_findings.extend(audit_security_groups())
 
    print("Checking IAM MFA...")
    all_findings.extend(audit_iam_mfa())
 
    print("Checking RDS public access...")
    all_findings.extend(audit_rds_public())
 
    print("Checking CloudTrail...")
    all_findings.extend(audit_cloudtrail())
 
    print(f"\nFound {len(all_findings)} findings. Sending to Claude for analysis...\n")
 
    analysis = analyse_with_claude(all_findings)
 
    report = {
        "audit_date": datetime.now().isoformat(),
        "account_id": boto3.client('sts').get_caller_identity()['Account'],
        "total_findings": len(all_findings),
        "findings_by_severity": {
            "CRITICAL": len([f for f in all_findings if f['severity'] == 'CRITICAL']),
            "HIGH": len([f for f in all_findings if f['severity'] == 'HIGH']),
            "MEDIUM": len([f for f in all_findings if f['severity'] == 'MEDIUM']),
        },
        "raw_findings": all_findings,
        "ai_analysis": analysis
    }
 
    with open("security-report.json", "w") as f:
        json.dump(report, f, indent=2)
 
    print("=" * 60)
    print("AI ANALYSIS")
    print("=" * 60)
    print(analysis)
    print(f"\nFull report saved to security-report.json")
 
    return report
 
 
if __name__ == "__main__":
    run_audit()

name: Weekly AWS Security Audit
 
on:
  schedule:
    - cron: '0 8 * * 1'
  workflow_dispatch:
 
jobs:
  audit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
 
      - name: Install dependencies
        run: pip install anthropic boto3
 
      - name: Run audit
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          AWS_DEFAULT_REGION: us-east-1
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: python audit.py
 
      - name: Upload report
        uses: actions/upload-artifact@v4
        with:
          name: security-report-${{ github.run_id }}
          path: security-report.json