Detect Attack-Surface Drift

import json
import os
from pathlib import Path

import requests

API_KEY = os.environ["FULLHUNT_API_KEY"]
DOMAIN = os.environ["FULLHUNT_DOMAIN"]
BASELINE = Path(os.environ.get("FULLHUNT_BASELINE", "attack-surface-baseline.json"))


def fetch(path):
    response = requests.get(
        f"https://fullhunt.io/api/v1{path}",
        headers={"X-API-KEY": API_KEY},
        timeout=60,
    )
    response.raise_for_status()
    return response.json()


def normalized_hosts(details, subdomains):
    candidates = details.get("hosts") or details.get("results", {}).get("hosts") or []
    normalized = {}

    for item in candidates:
        if not isinstance(item, dict):
            continue
        host = item.get("host") or item.get("hostname")
        if not host:
            continue
        normalized[host] = {
            "ip": item.get("ip_address") or item.get("ip"),
            "ports": sorted(item.get("ports") or item.get("open_ports") or []),
            "products": sorted(
                product.get("name", "") if isinstance(product, dict) else str(product)
                for product in (item.get("products") or [])
            ),
            "web": sorted(item.get("web") or item.get("web_technologies") or []),
        }

    listed = subdomains.get("hosts") or subdomains.get("results", {}).get("hosts") or []
    for host in listed:
        if isinstance(host, str):
            normalized.setdefault(host, {"ip": None, "ports": [], "products": [], "web": []})
    return normalized


details = fetch(f"/domain/{DOMAIN}/details")
subdomains = fetch(f"/domain/{DOMAIN}/subdomains")
current = {
    "domain": DOMAIN,
    "hosts": normalized_hosts(details, subdomains),
    "source_metadata": {
        "details": details.get("metadata"),
        "subdomains": subdomains.get("metadata"),
    },
}

if BASELINE.exists():
    previous = json.loads(BASELINE.read_text(encoding="utf-8"))
    old_hosts = previous.get("hosts", {})
    new_hosts = current["hosts"]
    report = {
        "added": sorted(set(new_hosts) - set(old_hosts)),
        "removed": sorted(set(old_hosts) - set(new_hosts)),
        "changed": sorted(
            host
            for host in set(old_hosts) & set(new_hosts)
            if old_hosts[host] != new_hosts[host]
        ),
    }
    print(json.dumps(report, indent=2))
else:
    print("No baseline found; creating the initial baseline.")

temporary = BASELINE.with_suffix(BASELINE.suffix + ".tmp")
temporary.write_text(json.dumps(current, indent=2) + "\n", encoding="utf-8")
temporary.replace(BASELINE)