Your Employees Use 10x More
AI Tools Than You Think.

#!/usr/bin/env python3
"""AI Shadow IT Audit — Network Layer Analysis.
Correlates DNS and proxy logs against the AI Tools Blocklist
to produce a comprehensive audit inventory of unauthorized AI tool usage."""

import csv
import json
import sys
import re
from collections import Counter, defaultdict
from datetime import datetime
from dataclasses import dataclass, field
from typing import Dict, List, Set

@dataclass
class AuditFinding:
    domain: str = ""
    tool_name: str = ""
    category: str = ""
    risk_level: str = "medium"
    dns_query_count: int = 0
    proxy_hit_count: int = 0
    post_requests: int = 0
    bytes_uploaded: int = 0
    unique_users: Set[str] = field(default_factory=set)
    departments: Set[str] = field(default_factory=set)
    first_seen: str = ""
    last_seen: str = ""

    @property
    def data_exposure_score(self) -> int:
        """0-100 score indicating data leakage risk."""
        score = 0
        if self.post_requests > 0: score += 25
        if self.bytes_uploaded > 50_000: score += 20
        if self.bytes_uploaded > 500_000: score += 15
        if len(self.unique_users) > 5: score += 15
        if len(self.departments) > 1: score += 10
        if self.risk_level == "high": score += 15
        return min(score, 100)

class ShadowITAuditor:
    def __init__(self, blocklist_csv: str, ip_dept_map: str = None):
        self.ai_domains = self._load_blocklist(blocklist_csv)
        self.dept_map = self._load_dept_map(ip_dept_map) if ip_dept_map else {}
        self.findings: Dict[str, AuditFinding] = {}

    def _load_blocklist(self, path: str) -> dict:
        domains = {}
        with open(path) as f:
            for row in csv.DictReader(f):
                domains[row["domain"].strip().lower()] = row
        return domains

    def _load_dept_map(self, path: str) -> dict:
        """Load IP-to-department mapping from CSV (ip,department)."""
        mapping = {}
        with open(path) as f:
            for row in csv.DictReader(f):
                mapping[row["ip"]] = row["department"]
        return mapping

    def _match_domain(self, fqdn: str):
        fqdn = fqdn.lower().rstrip(".")
        if fqdn in self.ai_domains:
            return fqdn
        root = ".".join(fqdn.split(".")[-2:])
        return root if root in self.ai_domains else None

    def ingest_dns_logs(self, log_path: str):
        """Process DNS resolver logs for AI tool domain matches."""
        pattern = re.compile(
            r"(\d{2}-\w{3}-\d{4}\s+\d{2}:\d{2}:\d{2})\S*\s+"
            r".*query:\s+([\w.\-]+)\s+IN\s+\w+.*\((\d+\.\d+\.\d+\.\d+)\)"
        )
        with open(log_path) as f:
            for line in f:
                m = pattern.search(line)
                if not m:
                    continue
                ts, domain, client_ip = m.group(1), m.group(2), m.group(3)
                matched = self._match_domain(domain)
                if matched:
                    self._record_finding(matched, client_ip, ts, source="dns")

    def ingest_proxy_logs(self, log_path: str):
        """Process proxy/SWG access logs for AI tool sessions."""
        proxy_re = re.compile(
            r"(\d{4}-\d{2}-\d{2}T[\d:]+)\s+([\d.]+)\s+(\w+)\s+"
            r"(\d+)\s+(https?://\S+)"
        )
        from urllib.parse import urlparse
        with open(log_path) as f:
            for line in f:
                m = proxy_re.search(line)
                if not m:
                    continue
                ts, client_ip, method, size, url = m.groups()
                host = urlparse(url).hostname
                if not host:
                    continue
                matched = self._match_domain(host)
                if matched:
                    f = self._record_finding(matched, client_ip, ts, source="proxy")
                    if method == "POST":
                        f.post_requests += 1
                        f.bytes_uploaded += int(size)

    def _record_finding(self, domain, client_ip, ts, source):
        if domain not in self.findings:
            info = self.ai_domains[domain]
            self.findings[domain] = AuditFinding(
                domain=domain,
                tool_name=info.get("tool_name", "Unknown"),
                category=info.get("primary_category", "Uncategorized"),
                risk_level=info.get("risk_level", "medium"),
            )
        f = self.findings[domain]
        if source == "dns":
            f.dns_query_count += 1
        else:
            f.proxy_hit_count += 1
        f.unique_users.add(client_ip)
        dept = self.dept_map.get(client_ip, "Unknown")
        f.departments.add(dept)
        if not f.first_seen or ts < f.first_seen:
            f.first_seen = ts
        if not f.last_seen or ts > f.last_seen:
            f.last_seen = ts
        return f

    def generate_audit_report(self) -> dict:
        sorted_findings = sorted(
            self.findings.values(),
            key=lambda x: x.data_exposure_score, reverse=True
        )
        return {
            "audit_date": datetime.now().isoformat(),
            "total_ai_tools_found": len(self.findings),
            "high_risk_tools": sum(
                1 for f in sorted_findings
                if f.data_exposure_score >= 60
            ),
            "total_users_affected": len(set(
                ip for f in sorted_findings for ip in f.unique_users
            )),
            "departments_affected": sorted(set(
                d for f in sorted_findings for d in f.departments
            )),
            "category_summary": dict(Counter(
                f.category for f in sorted_findings
            )),
            "findings": [
                {
                    "domain": f.domain,
                    "tool_name": f.tool_name,
                    "category": f.category,
                    "risk_score": f.data_exposure_score,
                    "unique_users": len(f.unique_users),
                    "departments": sorted(f.departments),
                    "data_uploaded_kb": round(f.bytes_uploaded / 1024, 1),
                    "first_seen": f.first_seen,
                    "last_seen": f.last_seen,
                } for f in sorted_findings
            ],
        }

if __name__ == "__main__":
    auditor = ShadowITAuditor(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else None)
    auditor.ingest_dns_logs(sys.argv[3])
    if len(sys.argv) > 4:
        auditor.ingest_proxy_logs(sys.argv[4])
    report = auditor.generate_audit_report()
    print(json.dumps(report, indent=2))

# AI Shadow IT Audit Report — Template Structure
# Adapt sections to your organization's reporting standards

report_metadata:
  title: "AI Shadow IT Audit Report"
  audit_period: "2025-06-01 to 2025-06-30"
  prepared_by: "IT Security — Shadow AI Audit Team"
  classification: "CONFIDENTIAL — Internal Use Only"
  distribution:
    - "CISO"
    - "CTO"
    - "Chief Compliance Officer"
    - "VP Engineering"
    - "General Counsel"

executive_summary:
  total_ai_tools_discovered: 187
  previously_known: 12
  high_risk_tools: 34
  departments_affected: 12
  unique_users: 847
  estimated_data_volume: "2.3 TB uploaded over 30 days"
  critical_findings:
    - "Source code submitted to 8 unauthorized code assistant tools"
    - "Customer PII identified in submissions to 3 AI data tools"
    - "Financial projections uploaded to 2 AI analysis platforms"
    - "Legal contracts submitted to 4 AI summarization services"

risk_matrix:
  critical:   # Immediate action required
    criteria: "PII/financial data exposure, regulatory violation risk"
    tools_count: 8
    action: "Block immediately via firewall/proxy, notify affected users"
  high:       # Action within 5 business days
    criteria: "Source code or IP exposure, multi-department usage"
    tools_count: 26
    action: "Evaluate for sanctioning or blocking, review data submitted"
  medium:     # Action within 30 days
    criteria: "Internal docs or business data, limited user base"
    tools_count: 71
    action: "Add to monitoring, include in policy review cycle"
  low:        # Monitor only
    criteria: "Public data only, minimal usage, low-risk category"
    tools_count: 82
    action: "Continue monitoring, review quarterly"

remediation_workflow:
  phase_1_immediate:  # Days 1-3
    - "Block critical-risk AI tool domains at firewall/proxy layer"
    - "Notify users of blocked tools with approved alternatives"
    - "Engage legal for data exposure incident assessment"
  phase_2_short_term:  # Days 4-14
    - "Evaluate high-risk tools for potential sanctioning"
    - "Deploy continuous monitoring via AI Tools Blocklist API"
    - "Implement DNS-layer blocking for all high-risk domains"
    - "Publish AI acceptable use policy to all employees"
  phase_3_sustained:  # Days 15-30+
    - "Establish AI tool approval workflow with security review"
    - "Configure automated daily blocklist updates"
    - "Schedule quarterly re-audit with expanded scope"
    - "Integrate AI tool monitoring into SOC dashboards"

department_findings:
  engineering:
    tools_found: 42
    users: 156
    top_categories: ["Code Assistants", "AI Debugging", "Documentation AI"]
    data_risk: "HIGH — proprietary source code confirmed in submissions"
  marketing:
    tools_found: 28
    users: 89
    top_categories: ["Content Generation", "Image AI", "Video AI"]
    data_risk: "MEDIUM — pre-release product details in content prompts"
  legal:
    tools_found: 11
    users: 23
    top_categories: ["Document AI", "Research AI", "Summarization"]
    data_risk: "CRITICAL — confidential contracts uploaded for analysis"