Investigate Historical Exposure

import json
import os
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_INVESTIGATION_DOMAIN"]
CASE_ID = os.environ["FULLHUNT_CASE_ID"]
OUTPUT_PATH = Path(os.environ.get("FULLHUNT_EVIDENCE_PATH", f"evidence-{CASE_ID}.json"))
HEADERS = {"X-API-KEY": API_KEY, "Content-Type": "application/json"}
TAGS = {"case_id": CASE_ID, "workflow": "historical-exposure"}


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 historical_timeline(records):
    events = []
    for record in records:
        events.append(
            {
                "host": record.get("host"),
                "first_seen": record.get("first_seen"),
                "last_seen": record.get("last_seen"),
                "date_added": record.get("date_added"),
                "snapshot_reason": record.get("snapshot_reason"),
                "ip_address": record.get("ip_address"),
                "ports": record.get("network_ports") or [],
                "products": record.get("products") or [],
            }
        )
    return events


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 main():
    with requests.Session() as session:
        current = get(session, f"/domain/{DOMAIN}/details")
        historical = post(
            session,
            "/oem/historical-hosts/search",
            {"type": "domain", "query": DOMAIN, "query_tags": TAGS},
        )
        passive_dns = get(session, "/nexus/passive-dns/lookup", {"domain": DOMAIN})
        certificates = get(session, "/nexus/cloud-certs/dns-search", {"query": DOMAIN})

    bundle = {
        "case_id": CASE_ID,
        "query": DOMAIN,
        "collected_at": datetime.now(timezone.utc).isoformat(),
        "sources": {
            "current_domain_details": current,
            "oem_historical_hosts": historical,
            "nexus_passive_dns": passive_dns,
            "nexus_cloud_certificates": certificates,
        },
        "timeline": historical_timeline(historical.get("results", [])),
    }
    write_atomic(OUTPUT_PATH, bundle)
    print(
        json.dumps(
            {
                "evidence_file": str(OUTPUT_PATH),
                "current_hosts": len(current.get("hosts", [])),
                "historical_snapshots": historical.get("total_results", 0),
                "passive_dns_count": passive_dns.get("count", 0),
                "certificate_count": certificates.get("count", 0),
            },
            indent=2,
        )
    )


if __name__ == "__main__":
    main()