Verify Cloud Decommissioning

import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

import requests

BASE_URL = "https://fullhunt.io/api/v1"
API_KEY = os.environ["FULLHUNT_API_KEY"]
DOMAIN = os.environ["FULLHUNT_DECOMMISSION_DOMAIN"]
BASELINE_PATH = Path(os.environ.get("FULLHUNT_DECOMMISSION_BASELINE", "decommission-before.json"))
REPORT_PATH = Path(os.environ.get("FULLHUNT_DECOMMISSION_REPORT", "decommission-report.json"))
HEADERS = {"X-API-KEY": API_KEY, "Content-Type": "application/json"}


def get(session, path, params=None):
    response = session.get(f"{BASE_URL}{path}", headers=HEADERS, params=params, timeout=60)
    response.raise_for_status()
    return response.json()


def post(session, path, body):
    response = session.post(f"{BASE_URL}{path}", headers=HEADERS, json=body, timeout=60)
    response.raise_for_status()
    return response.json()


def write_atomic(path, value):
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, indent=2, default=str) + "\n", encoding="utf-8")
    temporary.replace(path)


def host_names(payload):
    return {item.get("host") for item in payload.get("hosts", []) if item.get("host")}


def baseline(session):
    details = get(session, f"/domain/{DOMAIN}/details")
    write_atomic(
        BASELINE_PATH,
        {
            "domain": DOMAIN,
            "collected_at": datetime.now(timezone.utc).isoformat(),
            "domain_details": details,
        },
    )
    print(json.dumps({"baseline": str(BASELINE_PATH), "hosts": len(host_names(details))}))


def verify(session):
    before = json.loads(BASELINE_PATH.read_text(encoding="utf-8"))
    if before.get("domain") != DOMAIN:
        raise ValueError("Baseline belongs to another domain")

    scan_state = json.loads(Path(os.environ["FULLHUNT_SCAN_STATE"]).read_text(encoding="utf-8"))
    if scan_state.get("target") != DOMAIN or scan_state.get("status") != "scan_completed":
        raise ValueError("A completed OEM scan state for this domain is required")
    completed_at = int(scan_state.get("scan_info", {}).get("completed_at") or 0)

    current = get(session, f"/domain/{DOMAIN}/details")
    historical = post(
        session,
        "/oem/historical-hosts/search",
        {"type": "domain", "query": DOMAIN, "query_tags": {"workflow": "decommission-verification"}},
    )
    passive_dns = get(session, "/nexus/passive-dns/lookup", {"domain": DOMAIN})
    certificates = get(session, "/nexus/cloud-certs/dns-search", {"query": DOMAIN})

    current_hosts = current.get("hosts", [])
    fresh_live = [
        item
        for item in current_hosts
        if int(item.get("last_seen") or 0) >= completed_at and item.get("is_live") is True
    ]
    stale = [item for item in current_hosts if int(item.get("last_seen") or 0) < completed_at]
    before_names = host_names(before["domain_details"])
    current_names = host_names(current)

    report = {
        "decision": "blocked" if fresh_live else "review_required",
        "domain": DOMAIN,
        "scan_completed_at": completed_at,
        "removed_from_current_results": sorted(before_names - current_names),
        "remaining_in_current_results": sorted(current_names),
        "fresh_live_hosts": fresh_live,
        "stale_host_records": stale,
        "historical_record_count": historical.get("total_results", 0),
        "passive_dns_relationship_count": passive_dns.get("count", 0),
        "certificate_relationship_count": certificates.get("count", 0),
        "evidence": {
            "current": current,
            "historical": historical,
            "passive_dns": passive_dns,
            "certificates": certificates,
        },
    }
    write_atomic(REPORT_PATH, report)
    print(json.dumps({key: value for key, value in report.items() if key != "evidence"}, indent=2))


def main():
    if len(sys.argv) != 2 or sys.argv[1] not in {"baseline", "verify"}:
        raise SystemExit("Usage: verify_cloud_decommission.py baseline|verify")
    with requests.Session() as session:
        baseline(session) if sys.argv[1] == "baseline" else verify(session)


if __name__ == "__main__":
    main()