Detect Shadow AI Usage
Through Network Intelligence

#!/usr/bin/env python3
"""Shadow AI detection via DNS query log analysis.
Cross-references DNS resolver logs against the AI Tools Blocklist
to identify unauthorized AI tool usage across the enterprise."""

import csv
import re
import sys
import json
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path

def load_blocklist(csv_path: str) -> dict:
    """Load AI Tools Blocklist CSV into a domain lookup dictionary."""
    domains = {}
    with open(csv_path, "r") as f:
        reader = csv.DictReader(f)
        for row in reader:
            domains[row["domain"].strip().lower()] = {
                "category": row.get("primary_category", "Uncategorized"),
                "tool_name": row.get("tool_name", "Unknown"),
                "risk_level": row.get("risk_level", "medium"),
            }
    return domains

def parse_bind_log(log_path: str):
    """Parse BIND/named query log lines.
    Expected format: DD-Mon-YYYY HH:MM:SS.mmm ... query: domain IN A ... (client_ip)
    Yields (timestamp_str, client_ip, queried_domain) tuples."""
    pattern = re.compile(
        r"(\d{2}-\w{3}-\d{4}\s+\d{2}:\d{2}:\d{2})\.\d+\s+"
        r".*query:\s+([\w.\-]+)\s+IN\s+\w+.*\((\d+\.\d+\.\d+\.\d+)\)"
    )
    with open(log_path, "r") as f:
        for line in f:
            m = pattern.search(line)
            if m:
                yield m.group(1), m.group(3), m.group(2).lower().rstrip(".")

def extract_root_domain(fqdn: str) -> str:
    """Extract registrable domain from FQDN for matching."""
    parts = fqdn.split(".")
    if len(parts) >= 2:
        return ".".join(parts[-2:])
    return fqdn

def detect_shadow_ai(log_path: str, blocklist_path: str):
    """Main detection: correlate DNS logs against AI tools blocklist."""
    ai_domains = load_blocklist(blocklist_path)
    hits = defaultdict(lambda: {
        "clients": Counter(),
        "first_seen": None,
        "last_seen": None,
        "total_queries": 0,
    })
    category_stats = Counter()
    total_queries = 0
    matched_queries = 0

    for ts, client_ip, domain in parse_bind_log(log_path):
        total_queries += 1
        root = extract_root_domain(domain)
        match_domain = domain if domain in ai_domains else (
            root if root in ai_domains else None
        )
        if match_domain:
            matched_queries += 1
            entry = hits[match_domain]
            entry["clients"][client_ip] += 1
            entry["total_queries"] += 1
            if not entry["first_seen"] or ts < entry["first_seen"]:
                entry["first_seen"] = ts
            if not entry["last_seen"] or ts > entry["last_seen"]:
                entry["last_seen"] = ts
            category_stats[ai_domains[match_domain]["category"]] += 1

    # Generate report
    report = {
        "scan_date": datetime.now().isoformat(),
        "total_dns_queries": total_queries,
        "ai_tool_queries": matched_queries,
        "unique_ai_domains": len(hits),
        "unique_clients": len(set(
            ip for h in hits.values() for ip in h["clients"]
        )),
        "category_breakdown": dict(category_stats.most_common()),
        "top_domains": [],
    }

    for domain, data in sorted(
        hits.items(), key=lambda x: x[1]["total_queries"], reverse=True
    )[:50]:
        info = ai_domains[domain]
        report["top_domains"].append({
            "domain": domain,
            "tool_name": info["tool_name"],
            "category": info["category"],
            "risk_level": info["risk_level"],
            "total_queries": data["total_queries"],
            "unique_clients": len(data["clients"]),
            "first_seen": data["first_seen"],
            "last_seen": data["last_seen"],
        })

    return report

if __name__ == "__main__":
    report = detect_shadow_ai(sys.argv[1], sys.argv[2])
    print(json.dumps(report, indent=2))
    print(f"\n{'='*60}")
    print(f"Shadow AI Detection Report — {report['scan_date'][:10]}")
    print(f"{'='*60}")
    print(f"Total DNS queries analyzed:  {report['total_dns_queries']:,}")
    print(f"AI tool queries detected:   {report['ai_tool_queries']:,}")
    print(f"Unique AI tool domains:     {report['unique_ai_domains']}")
    print(f"Unique client IPs:          {report['unique_clients']}")
    print(f"\nTop 20 Shadow AI Tools:")
    for i, d in enumerate(report["top_domains"][:20], 1):
        print(f"  {i:2}. {d['domain']:<35} {d['category']:<25} "
              f"queries={d['total_queries']:<8} users={d['unique_clients']}")

#!/usr/bin/env python3
"""Real-time shadow AI detection — tail DNS logs and alert on AI tool queries."""

import csv
import time
import json
import re
import logging
from pathlib import Path
from datetime import datetime, timedelta
from collections import defaultdict

logging.basicConfig(level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("shadow_ai_monitor")

class ShadowAIMonitor:
    def __init__(self, blocklist_path: str, alert_threshold: int = 5):
        self.ai_domains = self._load_blocklist(blocklist_path)
        self.alert_threshold = alert_threshold
        self.window = defaultdict(lambda: defaultdict(int))
        self.window_start = datetime.now()
        self.alerted = set()
        logger.info(f"Loaded {len(self.ai_domains):,} AI tool domains")

    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 _resolve_domain(self, fqdn: str):
        """Match FQDN or root domain against blocklist."""
        if fqdn in self.ai_domains:
            return fqdn
        root = ".".join(fqdn.split(".")[-2:])
        return root if root in self.ai_domains else None

    def process_query(self, client_ip: str, domain: str):
        matched = self._resolve_domain(domain.lower().rstrip("."))
        if not matched:
            return
        self.window[matched][client_ip] += 1
        info = self.ai_domains[matched]
        total = sum(self.window[matched].values())

        if total >= self.alert_threshold and matched not in self.alerted:
            self.alerted.add(matched)
            alert = {
                "alert_type": "shadow_ai_detected",
                "domain": matched,
                "tool_name": info.get("tool_name", "Unknown"),
                "category": info.get("primary_category", "Unknown"),
                "unique_users": len(self.window[matched]),
                "total_queries": total,
                "timestamp": datetime.now().isoformat(),
            }
            logger.warning(f"SHADOW AI ALERT: {json.dumps(alert)}")
            self._send_to_siem(alert)

    def _send_to_siem(self, alert: dict):
        """Forward alert to SIEM via syslog, webhook, or file."""
        with open("/var/log/shadow_ai_alerts.json", "a") as f:
            f.write(json.dumps(alert) + "\n")

    def rotate_window(self, interval_minutes: int = 60):
        if datetime.now() - self.window_start > timedelta(minutes=interval_minutes):
            self.window.clear()
            self.alerted.clear()
            self.window_start = datetime.now()
            logger.info("Detection window rotated")