Every AI Chatbot Is a Potential
HIPAA Breach Waiting to Happen.

#!/usr/bin/env python3
"""HIPAA-specific AI tool monitoring and PHI exposure detection."""

import csv
import json
import hashlib
from datetime import datetime
from collections import defaultdict

class HIPAAComplianceMonitor:
    """Monitor AI tool access for HIPAA PHI exposure risks."""

    PHI_RISK_CATEGORIES = {
        "Text Generation":   {"risk": "CRITICAL", "phi_vector": "prompt_input"},
        "Code Generation":   {"risk": "HIGH",     "phi_vector": "code_comments_variables"},
        "Data Analytics":    {"risk": "CRITICAL", "phi_vector": "dataset_upload"},
        "Voice & Speech":    {"risk": "CRITICAL", "phi_vector": "audio_recording"},
        "Image & Video":     {"risk": "CRITICAL", "phi_vector": "medical_imaging"},
        "Healthcare AI":     {"risk": "CRITICAL", "phi_vector": "clinical_data"},
        "Customer Service":  {"risk": "HIGH",     "phi_vector": "patient_communications"},
        "Translation":       {"risk": "HIGH",     "phi_vector": "document_translation"},
    }

    BREACH_THRESHOLD_BYTES = 500
    REPORTABLE_RECORD_MIN = 500

    def __init__(self, blocklist_path: str, baa_registry_path: str):
        self.domains = {}
        self.baa_covered = set()

        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", ""),
                }

        with open(baa_registry_path) as f:
            for line in f:
                self.baa_covered.add(line.strip().lower())

    def analyze_proxy_log(self, log_path: str) -> list:
        """Scan proxy logs for PHI exposure to AI tools without BAAs."""
        incidents = []
        user_exposure = defaultdict(lambda: {
            "domains": set(), "total_bytes": 0, "events": 0
        })

        with open(log_path) as f:
            for line in f:
                parts = line.strip().split()
                if len(parts) < 8:
                    continue

                ts, user, method = parts[0], parts[2], parts[3]
                bytes_out = int(parts[4] or 0)
                domain = parts[6].lower().split("/")[0]

                if domain not in self.domains:
                    continue
                if domain in self.baa_covered:
                    continue

                info = self.domains[domain]
                cat = info["category"]
                risk = self.PHI_RISK_CATEGORIES.get(cat, {})

                is_data_submission = (
                    method == "POST"
                    and bytes_out > self.BREACH_THRESHOLD_BYTES
                )

                if is_data_submission:
                    incident = {
                        "type": "PHI_EXPOSURE_SUSPECTED",
                        "timestamp": ts,
                        "user_hash": hashlib.sha256(
                            user.encode()
                        ).hexdigest()[:12],
                        "domain": domain,
                        "tool_name": info["tool_name"],
                        "category": cat,
                        "hipaa_risk": risk.get("risk", "MEDIUM"),
                        "phi_vector": risk.get("phi_vector", "unknown"),
                        "bytes_transmitted": bytes_out,
                        "baa_status": "NONE",
                        "breach_reportable": bytes_out > 10240,
                        "security_rule_violations": [
                            "164.312(a)(1) Access Control",
                            "164.312(e)(1) Transmission Security",
                            "164.502(e) BAA Required",
                        ],
                    }
                    incidents.append(incident)
                    user_exposure[user]["domains"].add(domain)
                    user_exposure[user]["total_bytes"] += bytes_out
                    user_exposure[user]["events"] += 1

        return {
            "scan_timestamp": datetime.now().isoformat(),
            "total_incidents": len(incidents),
            "critical_incidents": sum(
                1 for i in incidents
                if i["hipaa_risk"] == "CRITICAL"
            ),
            "potentially_reportable": sum(
                1 for i in incidents
                if i["breach_reportable"]
            ),
            "unique_users_exposed": len(user_exposure),
            "incidents": incidents,
        }

# Usage
monitor = HIPAAComplianceMonitor(
    "ai_tools_blocklist.csv",
    "approved_baa_domains.txt"
)
results = monitor.analyze_proxy_log("/var/log/squid/access.log")

print(f"PHI exposure incidents: {results['total_incidents']}")
print(f"Critical (clinical data): {results['critical_incidents']}")
print(f"Potentially reportable:   {results['potentially_reportable']}")

with open("hipaa_phi_exposure_report.jsonl", "w") as f:
    for incident in results["incidents"]:
        f.write(json.dumps(incident) + "\n")