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()

#!/usr/bin/env python3
"""Shadow AI prevention effectiveness dashboard — weekly metrics."""

import json
import requests
from datetime import datetime, timedelta

API_BASE = "https://aitoolsblocklist.com/api/v1"
API_KEY  = "your-api-key-here"

def get_feed_coverage() -> dict:
    """Get current feed statistics."""
    resp = requests.get(
        f"{API_BASE}/stats",
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()

def count_blocked_attempts(log_path: str, days: int = 7) -> dict:
    """Count AI tool block events from firewall deny log."""
    cutoff = datetime.now() - timedelta(days=days)
    blocks_by_category = {}
    unique_users = set()
    total = 0
    with open(log_path, "r") as f:
        for line in f:
            try:
                event = json.loads(line)
                ts = datetime.fromisoformat(event["timestamp"])
                if ts < cutoff:
                    continue
                cat = event.get("ai_category", "Unknown")
                blocks_by_category[cat] = blocks_by_category.get(cat, 0) + 1
                unique_users.add(event.get("src_user", event["src_ip"]))
                total += 1
            except (json.JSONDecodeError, KeyError):
                continue
    return {
        "total_blocks": total,
        "unique_users": len(unique_users),
        "by_category": blocks_by_category,
    }

# Generate weekly dashboard
feed = get_feed_coverage()
blocks = count_blocked_attempts("/var/log/firewall/ai-blocks.jsonl")

print(f"Shadow AI Prevention Dashboard — Week of {datetime.now().strftime('%Y-%m-%d')}")
print(f"{'='*65}")
print(f"Feed coverage:        {feed['total_domains']:,} domains across {feed['categories']} categories")
print(f"Blocked attempts:     {blocks['total_blocks']:,} (last 7 days)")
print(f"Unique users blocked: {blocks['unique_users']}")
print(f"\nBlocks by AI category:")
for cat, count in sorted(
    blocks["by_category"].items(), key=lambda x: x[1], reverse=True
):
    bar = "█" * (count // 10)
    print(f"  {cat:30s}  {count:5d}  {bar}")