AI Tool Risk Assessment Framework
for Enterprise Security Teams

#!/usr/bin/env python3
"""Batch vendor assessment using the AI Tools Blocklist API."""

import requests
import csv
from datetime import datetime

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

# Category-level risk baselines — derived from analysis
# of data handling patterns across 17,410+ AI tools
CATEGORY_BASELINES = {
    "Text Generation":       {"data_exp": 8, "training": 7, "cap": 5},
    "Code Assistant":        {"data_exp": 9, "training": 8, "cap": 7},
    "Image Generation":      {"data_exp": 6, "training": 8, "cap": 3},
    "Voice & Audio":         {"data_exp": 7, "training": 6, "cap": 4},
    "Data Analysis":         {"data_exp": 9, "training": 6, "cap": 6},
    "Autonomous Agents":     {"data_exp": 9, "training": 7, "cap": 10},
    "Translation":           {"data_exp": 7, "training": 5, "cap": 2},
    "Transcription":         {"data_exp": 7, "training": 5, "cap": 2},
}

def assess_discovered_tools(domains: list) -> list:
    """Look up each domain in the AI tools database and
       generate preliminary risk assessments."""
    results = []
    for domain in domains:
        resp = requests.get(
            f"{API_BASE}/lookup/{domain}",
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=10,
        )
        if resp.status_code != 200:
            continue
        tool = resp.json()
        cat = tool.get("primary_category", "Unknown")
        baseline = CATEGORY_BASELINES.get(cat, {
            "data_exp": 7, "training": 7, "cap": 5
        })
        results.append({
            "domain": domain,
            "category": cat,
            "data_exposure_baseline": baseline["data_exp"],
            "training_retention_baseline": baseline["training"],
            "capability_baseline": baseline["cap"],
            "requires_manual_review": True,
            "assessed_date": datetime.now().isoformat(),
        })
    return results

# Assess tools discovered during shadow AI audit
discovered = ["novelai.net", "codeium.com", "deepseek.com"]
assessments = assess_discovered_tools(discovered)
for a in assessments:
    print(f"  {a['domain']:30s}  {a['category']:20s}  "
          f"data={a['data_exposure_baseline']}  "
          f"train={a['training_retention_baseline']}  "
          f"cap={a['capability_baseline']}")

#!/usr/bin/env python3
"""Continuous AI tool risk monitoring — detect changes and recalculate scores."""

import requests
import json
import hashlib
from datetime import datetime, timedelta

API_BASE = "https://aitoolsblocklist.com/api/v1"
API_KEY  = "your-api-key-here"
STATE_FILE = "/var/lib/ai-risk/monitor_state.json"

def load_previous_state() -> dict:
    try:
        with open(STATE_FILE, "r") as f:
            return json.load(f)
    except FileNotFoundError:
        return {"domains": {}, "last_check": None}

def fetch_feed_changes(since: str) -> list:
    """Fetch domains added or modified since last check."""
    resp = requests.get(
        f"{API_BASE}/feed/changes",
        params={"since": since, "format": "json"},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json().get("changes", [])

def detect_risk_drift(prev: dict, current: dict) -> list:
    """Compare previous and current tool metadata to find risk-relevant changes."""
    alerts = []
    for domain, data in current.items():
        old = prev.get(domain)
        if not old:
            alerts.append({
                "type": "new_tool",
                "domain": domain,
                "category": data["category"],
                "action": "Initial risk assessment required",
            })
        elif old.get("category") != data["category"]:
            alerts.append({
                "type": "category_change",
                "domain": domain,
                "old_category": old["category"],
                "new_category": data["category"],
                "action": "Risk score recalculation required",
            })
    return alerts

# Run monitoring check
state = load_previous_state()
since = state["last_check"] or (
    datetime.utcnow() - timedelta(days=1)
).isoformat()
changes = fetch_feed_changes(since)
print(f"Changes since {since}: {len(changes)} domains")

for alert in detect_risk_drift(state["domains"], {
    c["domain"]: c for c in changes
}):
    print(f"  [{alert['type'].upper()}] {alert['domain']}: {alert['action']}")