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

#!/bin/bash
# continuous_ai_monitor.sh — Hourly AI shadow IT monitoring
# Deployed as a cron job: 0 * * * * /opt/shadow-ai/continuous_ai_monitor.sh

set -euo pipefail

# Configuration
BLOCKLIST_DIR="/opt/shadow-ai/blocklist"
BLOCKLIST_CSV="${BLOCKLIST_DIR}/ai_tools_database.csv"
LOG_DIR="/var/log/shadow-ai"
ALERT_DIR="/var/log/shadow-ai/alerts"
DNS_LOG="/var/log/named/queries.log"
PROXY_LOG="/var/log/squid/access.log"
API_KEY="${AI_BLOCKLIST_API_KEY}"
SIEM_WEBHOOK="https://siem.internal/api/v1/events"
STATE_FILE="${LOG_DIR}/monitor_state.json"
KNOWN_TOOLS="${LOG_DIR}/known_ai_tools.txt"

mkdir -p "${LOG_DIR}" "${ALERT_DIR}"

# Step 1: Update the blocklist daily (skip if updated <24h ago)
BLOCKLIST_AGE=$(( $(date +%s) - $(stat -c %Y "${BLOCKLIST_CSV}" 2>/dev/null || echo 0) ))
if [ "${BLOCKLIST_AGE}" -gt 86400 ]; then
    echo "[$(date -Iseconds)] Updating AI Tools Blocklist..."
    curl -sS -H "Authorization: Bearer ${API_KEY}" \
        "https://api.aitoolsblocklist.com/v1/export/csv" \
        -o "${BLOCKLIST_CSV}.tmp"
    DOMAIN_COUNT=$(wc -l < "${BLOCKLIST_CSV}.tmp")
    if [ "${DOMAIN_COUNT}" -gt 40000 ]; then
        mv "${BLOCKLIST_CSV}.tmp" "${BLOCKLIST_CSV}"
        echo "[$(date -Iseconds)] Blocklist updated: ${DOMAIN_COUNT} domains"
    else
        echo "[$(date -Iseconds)] ERROR: Blocklist update failed validation"
        rm -f "${BLOCKLIST_CSV}.tmp"
    fi
fi

# Step 2: Extract last hour of DNS and proxy logs
HOUR_AGO=$(date -d "1 hour ago" +"%d-%b-%Y %H:%M:%S")
awk -v start="${HOUR_AGO}" '$0 >= start' "${DNS_LOG}" > "${LOG_DIR}/dns_hourly.log"
awk -v start=$(date -d "1 hour ago" +%s) \
    '$1 >= start' "${PROXY_LOG}" > "${LOG_DIR}/proxy_hourly.log"

# Step 3: Run the audit detection script
python3 /opt/shadow-ai/audit_detect.py \
    "${BLOCKLIST_CSV}" \
    /opt/shadow-ai/ip_department_map.csv \
    "${LOG_DIR}/dns_hourly.log" \
    "${LOG_DIR}/proxy_hourly.log" \
    > "${LOG_DIR}/hourly_report.json"

# Step 4: Detect NEW tools not seen before
touch "${KNOWN_TOOLS}"
python3 -c "
import json, sys
report = json.load(open('${LOG_DIR}/hourly_report.json'))
known = set(open('${KNOWN_TOOLS}').read().splitlines())
new_tools = []
for f in report.get('findings', []):
    if f['domain'] not in known:
        new_tools.append(f)
if new_tools:
    alert = {
        'alert_type': 'new_shadow_ai_tools',
        'timestamp': report['audit_date'],
        'new_tools_count': len(new_tools),
        'tools': new_tools
    }
    print(json.dumps(alert))
" > "${ALERT_DIR}/new_tools_$(date +%Y%m%d%H).json"

# Step 5: Forward alerts to SIEM
ALERT_FILE="${ALERT_DIR}/new_tools_$(date +%Y%m%d%H).json"
if [ -s "${ALERT_FILE}" ]; then
    curl -sS -X POST "${SIEM_WEBHOOK}" \
        -H "Content-Type: application/json" \
        -d @"${ALERT_FILE}"
    echo "[$(date -Iseconds)] New AI tools alert forwarded to SIEM"

    # Update known tools registry
    python3 -c "
import json
alert = json.load(open('${ALERT_FILE}'))
for t in alert.get('tools', []):
    print(t['domain'])
" >> "${KNOWN_TOOLS}"
fi

# Step 6: Cleanup hourly temp files
rm -f "${LOG_DIR}/dns_hourly.log" "${LOG_DIR}/proxy_hourly.log"
echo "[$(date -Iseconds)] Hourly shadow AI scan complete"