Build an AI Compliance Auditor with Claude API for AWS Infrastructure

pip install anthropic boto3 rich

import anthropic
import boto3
import json
from rich.console import Console
from rich.table import Table
 
client = anthropic.Anthropic()
console = Console()
 
 
def scan_s3_compliance() -> list[dict]:
    """Check S3 buckets for compliance issues."""
    s3 = boto3.client("s3")
    findings = []
 
    buckets = s3.list_buckets()["Buckets"]
 
    for bucket in buckets:
        name = bucket["Name"]
        issues = []
 
        # Check public access block
        try:
            pab = s3.get_public_access_block(Bucket=name)["PublicAccessBlockConfiguration"]
            if not all([pab.get("BlockPublicAcls"), pab.get("BlockPublicPolicy"),
                        pab.get("IgnorePublicAcls"), pab.get("RestrictPublicBuckets")]):
                issues.append("Public access block not fully enabled")
        except s3.exceptions.NoSuchPublicAccessBlockConfiguration:
            issues.append("No public access block configured — bucket may be publicly accessible")
 
        # Check encryption
        try:
            enc = s3.get_bucket_encryption(Bucket=name)
            rules = enc["ServerSideEncryptionConfiguration"]["Rules"]
            if not any(r.get("ApplyServerSideEncryptionByDefault") for r in rules):
                issues.append("Default encryption not enabled")
        except Exception:
            issues.append("Encryption not configured — data at rest unencrypted")
 
        # Check versioning
        try:
            ver = s3.get_bucket_versioning(Bucket=name)
            if ver.get("Status") != "Enabled":
                issues.append("Versioning not enabled — cannot recover deleted objects")
        except Exception:
            issues.append("Could not check versioning")
 
        # Check logging
        try:
            log = s3.get_bucket_logging(Bucket=name)
            if not log.get("LoggingEnabled"):
                issues.append("Access logging not enabled")
        except Exception:
            pass
 
        if issues:
            findings.append({
                "resource": f"s3://{name}",
                "type": "S3",
                "issues": issues
            })
 
    return findings
 
 
def scan_iam_compliance() -> list[dict]:
    """Check IAM for compliance issues."""
    iam = boto3.client("iam")
    findings = []
 
    # Check root MFA
    summary = iam.get_account_summary()["SummaryMap"]
    if summary.get("AccountMFAEnabled", 0) == 0:
        findings.append({
            "resource": "AWS Root Account",
            "type": "IAM",
            "issues": ["Root account MFA not enabled — critical security risk"]
        })
 
    # Check password policy
    try:
        policy = iam.get_account_password_policy()["PasswordPolicy"]
        issues = []
        if policy.get("MinimumPasswordLength", 0) < 14:
            issues.append("Password minimum length < 14 characters")
        if not policy.get("RequireSymbols"):
            issues.append("Password policy does not require symbols")
        if not policy.get("MaxPasswordAge"):
            issues.append("No password rotation policy set")
        if issues:
            findings.append({"resource": "IAM Password Policy", "type": "IAM", "issues": issues})
    except iam.exceptions.NoSuchEntityException:
        findings.append({
            "resource": "IAM Password Policy",
            "type": "IAM",
            "issues": ["No password policy configured"]
        })
 
    # Check users with access keys > 90 days old
    users = iam.list_users()["Users"]
    from datetime import datetime, timezone
    for user in users[:20]:    # Check first 20 users
        keys = iam.list_access_keys(UserName=user["UserName"])["AccessKeyMetadata"]
        for key in keys:
            if key["Status"] == "Active":
                age = (datetime.now(timezone.utc) - key["CreateDate"]).days
                if age > 90:
                    findings.append({
                        "resource": f"IAM User: {user['UserName']} (key: {key['AccessKeyId']})",
                        "type": "IAM",
                        "issues": [f"Access key {age} days old — exceeds 90-day rotation policy"]
                    })
 
    return findings
 
 
def scan_cloudtrail_compliance() -> list[dict]:
    """Check CloudTrail configuration."""
    ct = boto3.client("cloudtrail", region_name="ap-south-1")
    findings = []
 
    try:
        trails = ct.describe_trails()["trailList"]
 
        if not trails:
            return [{"resource": "CloudTrail", "type": "CloudTrail",
                     "issues": ["No CloudTrail configured — no audit log of API calls"]}]
 
        for trail in trails:
            issues = []
            if not trail.get("IsMultiRegionTrail"):
                issues.append("CloudTrail not multi-region — misses API calls in other regions")
            if not trail.get("LogFileValidationEnabled"):
                issues.append("Log file validation not enabled — logs can be tampered")
 
            status = ct.get_trail_status(Name=trail["TrailARN"])
            if not status.get("IsLogging"):
                issues.append("CloudTrail logging is disabled")
 
            if issues:
                findings.append({
                    "resource": f"CloudTrail: {trail['Name']}",
                    "type": "CloudTrail",
                    "issues": issues
                })
 
    except Exception as e:
        findings.append({"resource": "CloudTrail", "type": "CloudTrail",
                         "issues": [f"Could not check: {e}"]})
 
    return findings
 
 
def analyze_with_claude(all_findings: list[dict]) -> str:
    """Get Claude's compliance analysis and prioritized remediation."""
    if not all_findings:
        return "No compliance issues found. Infrastructure meets basic requirements."
 
    findings_json = json.dumps(all_findings[:30], indent=2)    # Limit for token budget
 
    prompt = f"""You are a cloud security compliance expert. Analyze these AWS infrastructure findings.
 
## Findings
{findings_json}
 
Provide a compliance report with:
1. **Overall Risk Rating**: critical/high/medium/low with reasoning
2. **Compliance Framework Mapping**: Map each finding to SOC 2, CIS AWS Benchmark, or HIPAA control
3. **Prioritized Remediation** (top 5): Exact AWS CLI or console steps to fix each issue
4. **Quick Wins**: Findings that can be fixed in < 5 minutes
5. **Estimated Remediation Time**: Total hours to address all findings
6. **Executive Summary**: 2-3 sentences for non-technical stakeholders
 
Format as a structured markdown report."""
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2500,
        messages=[{"role": "user", "content": prompt}]
    )
 
    return response.content[0].text
 
 
def run_compliance_audit():
    console.print("[bold]AWS Compliance Audit[/bold]\n")
 
    console.print("Scanning S3 buckets...")
    s3_findings = scan_s3_compliance()
 
    console.print("Scanning IAM...")
    iam_findings = scan_iam_compliance()
 
    console.print("Scanning CloudTrail...")
    ct_findings = scan_cloudtrail_compliance()
 
    all_findings = s3_findings + iam_findings + ct_findings
 
    console.print(f"\nFound [bold red]{len(all_findings)}[/bold red] compliance issues")
    console.print("Analyzing with Claude AI...\n")
 
    analysis = analyze_with_claude(all_findings)
 
    console.print("[bold green]COMPLIANCE REPORT[/bold green]")
    console.print("="*60)
    console.print(analysis)
 
    # Save report
    import datetime
    report_file = f"compliance_report_{datetime.date.today()}.md"
    with open(report_file, "w") as f:
        f.write(analysis)
    console.print(f"\n[dim]Report saved: {report_file}[/dim]")
 
 
if __name__ == "__main__":
    run_compliance_audit()

## Overall Risk Rating: HIGH
 
The infrastructure has 14 compliance findings, including 3 critical issues
that require immediate attention.
 
## Critical Findings
 
### 1. Root Account MFA Not Enabled (CIS 1.5, SOC 2 CC6.1)
**Fix (2 minutes):**
1. Go to IAM → Security credentials
2. Click "Assign MFA device" next to root account
3. Use a hardware MFA device for root (not virtual)
 
### 2. S3 Bucket myapp-data has no encryption (HIPAA § 164.312(a)(2))
**Fix (1 minute):**
aws s3api put-bucket-encryption \
  --bucket myapp-data \
  --server-side-encryption-configuration '{"Rules": [{"ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "aws:kms"}}]}'