AI Compliance Is Not Optional.
It Is Already Regulated.

#!/usr/bin/env python3
"""Generate compliance audit trail from proxy logs and AI domain feed."""

import csv
import json
import sys
from datetime import datetime
from pathlib import Path

class ComplianceAuditTrail:
    """Correlate proxy logs with AI domain feed for regulatory audit."""

    REGULATION_MAP = {
        "Text Generation":     ["GDPR", "CCPA", "DORA", "SOX"],
        "Code Generation":     ["SOX", "DORA"],
        "Image Generation":    ["GDPR", "EU AI Act"],
        "Data Analytics":      ["GDPR", "CCPA", "SOX"],
        "Voice & Speech":      ["GDPR", "CCPA", "EU AI Act"],
        "AI Agents":           ["DORA", "SOX", "EU AI Act"],
        "Healthcare AI":       ["GDPR", "HIPAA", "EU AI Act"],
    }

    def __init__(self, blocklist_path: str):
        self.domains = {}
        with open(blocklist_path) as f:
            for row in csv.DictReader(f):
                self.domains[row["domain"].lower()] = {
                    "category": row.get("primary_category", "Unknown"),
                    "tool_name": row.get("tool_name", ""),
                    "risk_score": int(row.get("risk_score", 50)),
                }

    def process_proxy_log(self, log_path: str) -> list:
        """Parse Squid/BlueCoat proxy log and generate audit events."""
        audit_events = []
        with open(log_path) as f:
            for line in f:
                parts = line.strip().split()
                if len(parts) < 8:
                    continue
                timestamp, user, domain = parts[0], parts[2], parts[6]
                method, bytes_sent = parts[3], int(parts[4] or 0)
                domain_key = domain.lower().split("/")[0]

                if domain_key in self.domains:
                    info = self.domains[domain_key]
                    regulations = self.REGULATION_MAP.get(
                        info["category"], ["GDPR", "CCPA"]
                    )
                    audit_events.append({
                        "event_type": "AI_TOOL_ACCESS",
                        "timestamp": timestamp,
                        "user": user,
                        "domain": domain_key,
                        "tool_name": info["tool_name"],
                        "category": info["category"],
                        "risk_score": info["risk_score"],
                        "method": method,
                        "bytes_sent": bytes_sent,
                        "data_submission": method == "POST" and bytes_sent > 1024,
                        "applicable_regulations": regulations,
                        "severity": "HIGH" if bytes_sent > 10240 else "MEDIUM",
                    })
        return audit_events

    def generate_report(self, events: list, output: str):
        """Write structured audit trail to JSON Lines for SIEM ingestion."""
        with open(output, "w") as f:
            for event in events:
                f.write(json.dumps(event) + "\n")
        print(f"Wrote {len(events)} audit events to {output}")
        print(f"Data submissions detected: {sum(1 for e in events if e['data_submission'])}")
        print(f"High severity events:      {sum(1 for e in events if e['severity'] == 'HIGH')}")

# Usage
trail = ComplianceAuditTrail("ai_tools_blocklist.csv")
events = trail.process_proxy_log("/var/log/squid/access.log")
trail.generate_report(events, "compliance_audit_trail.jsonl")

# regulatory_mapping.yaml
# Maps AI tool categories to regulatory obligations and required controls

regulatory_framework:

  gdpr:
    name: "General Data Protection Regulation"
    jurisdictions: ["EU", "EEA", "UK"]
    high_risk_categories:
      - "Text & Language"        # Chatbots process PII in prompts
      - "Data Analytics"          # Dataset uploads may contain PII
      - "Voice & Speech"         # Biometric data under Art. 9
      - "Healthcare AI"           # Special category health data
      - "Image & Video"          # Facial data is biometric PII
    required_controls:
      - "Data Processing Agreement with vendor"
      - "Cross-border transfer mechanism (SCCs/adequacy)"
      - "Data Protection Impact Assessment (DPIA)"
      - "Record of Processing Activities (ROPA) entry"
      - "Right to erasure procedure documented"

  ccpa:
    name: "California Consumer Privacy Act / CPRA"
    jurisdictions: ["US-CA"]
    high_risk_categories:
      - "Text & Language"
      - "Data Analytics"
      - "Customer Service AI"
      - "Marketing AI"
    required_controls:
      - "Third-party data sharing disclosure"
      - "Opt-out mechanism for data sale/sharing"
      - "45-day consumer request response process"

  dora:
    name: "Digital Operational Resilience Act"
    jurisdictions: ["EU"]
    sector: "Financial services"
    high_risk_categories:
      - "Code & Development"     # Trading/risk system code
      - "Data Analytics"          # Financial model processing
      - "AI Agents"              # Autonomous decision-making
      - "Automation"             # Process dependencies
    required_controls:
      - "ICT third-party register entry (Art. 28)"
      - "Pre-contractual risk assessment"
      - "Exit strategy documented"
      - "Concentration risk assessment"
      - "Incident notification <= 4 hours"

  sox:
    name: "Sarbanes-Oxley Act"
    jurisdictions: ["US"]
    sector: "Public companies"
    high_risk_categories:
      - "Data Analytics"          # Financial data processing
      - "Code & Development"     # Financial system changes
      - "Text & Language"        # Disclosure drafting
      - "Automation"             # Financial process automation
    required_controls:
      - "IT General Controls (ITGC) coverage"
      - "Change management audit trail"
      - "Access control documentation"
      - "Output validation procedures"

#!/usr/bin/env python3
"""Generate weekly AI compliance metrics for executive reporting."""

import json
from collections import Counter, defaultdict
from datetime import datetime, timedelta

def generate_compliance_dashboard(audit_trail_path: str, days: int = 7):
    """Produce compliance KPIs from audit trail data."""
    cutoff = datetime.now() - timedelta(days=days)
    events = []

    with open(audit_trail_path) as f:
        for line in f:
            event = json.loads(line)
            event_time = datetime.fromisoformat(event["timestamp"])
            if event_time >= cutoff:
                events.append(event)

    # KPI calculations
    total_events = len(events)
    unique_tools = len(set(e["domain"] for e in events))
    unique_users = len(set(e["user"] for e in events))
    data_submissions = sum(1 for e in events if e.get("data_submission"))
    high_severity = sum(1 for e in events if e["severity"] == "HIGH")

    # Regulatory exposure breakdown
    reg_exposure = Counter()
    for e in events:
        for reg in e.get("applicable_regulations", []):
            reg_exposure[reg] += 1

    # Category risk breakdown
    cat_risk = defaultdict(lambda: {"events": 0, "users": set()})
    for e in events:
        cat_risk[e["category"]]["events"] += 1
        cat_risk[e["category"]]["users"].add(e["user"])

    report = {
        "report_period": f"{cutoff.date()} to {datetime.now().date()}",
        "kpis": {
            "total_ai_tool_access_events": total_events,
            "unique_ai_tools_accessed": unique_tools,
            "unique_users": unique_users,
            "data_submission_events": data_submissions,
            "high_severity_events": high_severity,
            "compliance_coverage": f"{(unique_tools / total_events * 100):.1f}%",
        },
        "regulatory_exposure": dict(reg_exposure.most_common()),
        "category_risk_summary": {
            cat: {"events": d["events"], "unique_users": len(d["users"])}
            for cat, d in sorted(
                cat_risk.items(),
                key=lambda x: x[1]["events"],
                reverse=True,
            )
        },
    }

    print(json.dumps(report, indent=2))
    return report

generate_compliance_dashboard("compliance_audit_trail.jsonl")