Adding a Custom AI URL Category
to Your Next-Gen Firewall

# UTM CLI — Configure external threat feed for AI tool domains

config system external-resource
    edit "AI-Tools-Feed"
        set type category
        set resource "https://edl.yourcompany.internal/ai_domains.txt" # hosted internally, regenerated daily from the API CSV
        set refresh-rate 1440    # Refresh every 1440 minutes (24 hours)
        set status enable
    next
end

# Create a web filter profile that blocks the AI category
config webfilter profile
    edit "Block-AI-Tools"
        config ftgd-wf
            config filters
                edit 1
                    set category 192    # Custom category ID for AI-Tools-Feed
                    set action block
                next
            end
        end
    next
end

# Apply the web filter profile to a firewall policy
config firewall policy
    edit 10
        set name "Block-AI-Outbound"
        set srcintf "internal"
        set dstintf "wan1"
        set action accept
        set utm-status enable
        set webfilter-profile "Block-AI-Tools"
    next
end

#!/usr/bin/env python3
# Automated AI URL category updater — multi-vendor support
# Schedule via cron: 0 3 * * * /opt/scripts/update_ai_category.py

import requests, json, csv, io, sys, logging

logging.basicConfig(level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s')
log = logging.getLogger('ai-category-updater')

# Configuration
API_KEY    = "YOUR_API_KEY"
API_BASE   = "https://www.aitoolsblocklist.com/api/database/"
STATE_FILE = "/var/lib/ai-blocklist/last_sync.json"

def fetch_delta():
    """Re-download the database when it changes; diff locally."""
    try:
        with open(STATE_FILE) as f:
            state = json.load(f)
    except FileNotFoundError:
        state = {"last_updated_unix": 0, "domains": []}

    # Poll metadata; skip the download if nothing changed
    info = requests.get(API_BASE,
        params={"action": "database_info"},
        headers={"X-API-Key": API_KEY},
        timeout=30).json()
    if info["last_updated_unix"] <= state["last_updated_unix"]:
        log.info("Database unchanged since last sync")
        return {"added": [], "removed": [], **state}

    # Download the CSV (domain,category,subcategory) and diff locally
    resp = requests.get(API_BASE,
        params={"action": "download_database"},
        headers={"X-API-Key": API_KEY},
        timeout=120)
    resp.raise_for_status()
    current = {row["domain"]
               for row in csv.DictReader(io.StringIO(resp.text))}
    previous = set(state["domains"])

    data = {"added": sorted(current - previous),
            "removed": sorted(previous - current),
            "last_updated_unix": info["last_updated_unix"],
            "domains": sorted(current)}
    log.info(f"Delta: {len(data['added'])} added, {len(data['removed'])} removed")
    return data

def update_ngfw_edl():
    """Force the firewall to refresh its EDL immediately."""
    import subprocess
    result = subprocess.run([
        "curl", "-sk",
        "https://firewall.local/api/?type=op&cmd="
        ""
        ""
        "&key=PAN_API_KEY"
    ], capture_output=True)
    log.info(f"Firewall EDL refresh: {result.returncode}")

def update_cloud_swg(added, removed):
    """Push delta to cloud SWG custom URL category."""
    session = requests.Session()
    # Authenticate to SWG API
    session.post("https://your-swg-api.example.com/api/v1/authenticatedSession",
        json={"apiKey": "SWG_KEY", "username": "admin", "password": "pass"})
    # Get current category, merge delta, PUT update
    cat = session.get("https://your-swg-api.example.com/api/v1/urlCategories/CUSTOM_CAT_ID").json()
    urls = set(cat["urls"])
    urls.update(added)
    urls -= set(removed)
    cat["urls"] = list(urls)
    session.put(f"https://your-swg-api.example.com/api/v1/urlCategories/{cat['id']}", json=cat)
    session.post("https://your-swg-api.example.com/api/v1/status/activate")
    log.info(f"Cloud SWG updated: {len(urls)} total domains")

if __name__ == "__main__":
    delta = fetch_delta()
    update_ngfw_edl()
    update_cloud_swg(delta["added"], delta["removed"])
    # Save sync state
    with open(STATE_FILE, "w") as f:
        json.dump({"last_updated_unix": delta["last_updated_unix"],
                   "domains": delta["domains"]}, f)
    log.info("All platforms updated successfully")