Prevent Shadow AI Adoption
Before It Becomes a Crisis

#!/usr/bin/env python3
"""Shadow AI baseline audit — cross-reference DNS logs against the AI tools feed."""

import csv
import sys
from collections import Counter, defaultdict
from datetime import datetime

def load_ai_feed(feed_path: str) -> dict:
    """Load the AI tools blocklist into a lookup dictionary."""
    domains = {}
    with open(feed_path, "r") as f:
        reader = csv.DictReader(f)
        for row in reader:
            domains[row["domain"].lower()] = {
                "category": row.get("primary_category", "Unknown"),
                "subcategory": row.get("subcategory", ""),
                "risk_level": row.get("risk_level", "medium"),
            }
    return domains

def audit_dns_logs(log_path: str, ai_feed: dict) -> dict:
    """Parse DNS query logs and flag AI tool lookups."""
    findings = defaultdict(lambda: {
        "clients": Counter(), "first_seen": None, "last_seen": None
    })
    with open(log_path, "r") as f:
        for line in f:
            parts = line.strip().split()
            if len(parts) < 8:
                continue
            ts = parts[0] + " " + parts[1]
            queried = parts[4].rstrip(".").lower()
            client_ip = parts[6].split("#")[0]
            if queried in ai_feed:
                entry = findings[queried]
                entry["clients"][client_ip] += 1
                if not entry["first_seen"]:
                    entry["first_seen"] = ts
                entry["last_seen"] = ts
    return findings

# Run baseline audit
ai_feed = load_ai_feed("ai_tools_database.csv")
findings = audit_dns_logs("/var/log/named/query.log", ai_feed)

print(f"Shadow AI Baseline Audit — {datetime.now().strftime('%Y-%m-%d')}")
print(f"{'='*65}")
print(f"Total AI tool domains detected: {len(findings)}")
total_clients = len(set(
    ip for f in findings.values() for ip in f["clients"]
))
print(f"Unique client IPs involved:     {total_clients}")
print()

for domain, data in sorted(
    findings.items(),
    key=lambda x: sum(x[1]["clients"].values()),
    reverse=True
)[:25]:
    info = ai_feed[domain]
    queries = sum(data["clients"].values())
    users = len(data["clients"])
    print(f"  {domain}")
    print(f"    Category:   {info['category']} / {info['subcategory']}")
    print(f"    Risk:       {info['risk_level'].upper()}")
    print(f"    Queries:    {queries}  |  Users: {users}")
    print(f"    First seen: {data['first_seen']}")
    print()