Monitor Certificate Expiry and Issuers

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

import requests

BASE_URL = "https://fullhunt.io/api/v1/enterprise"
API_KEY = os.environ["FULLHUNT_API_KEY"]
ORG_ID = os.environ["FULLHUNT_ORG_ID"]
EXPIRY_DAYS = int(os.environ.get("FULLHUNT_CERT_EXPIRY_DAYS", "30"))
APPROVED_ISSUERS = {
    value.strip().lower()
    for value in os.environ.get("FULLHUNT_APPROVED_ISSUERS", "").split(",")
    if value.strip()
}
STATE_PATH = Path(os.environ.get("FULLHUNT_CERT_STATE", f"certificates-{ORG_ID}.json"))
HEADERS = {"X-API-KEY": API_KEY}


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 parse_time(value):
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value, tz=timezone.utc)
    if not value:
        return None
    normalized = str(value).strip().replace("Z", "+00:00")
    try:
        parsed = datetime.fromisoformat(normalized)
        return parsed.replace(tzinfo=parsed.tzinfo or timezone.utc)
    except ValueError:
        pass
    for pattern in ("%d-%m-%Y %H:%M:%S", "%Y-%m-%d %H:%M:%S"):
        try:
            return datetime.strptime(normalized, pattern).replace(tzinfo=timezone.utc)
        except ValueError:
            continue
    return None


def save_snapshot(snapshot):
    temporary = STATE_PATH.with_suffix(STATE_PATH.suffix + ".tmp")
    temporary.write_text(json.dumps(snapshot, indent=2) + "\n", encoding="utf-8")
    temporary.replace(STATE_PATH)


def main():
    now = datetime.now(timezone.utc)
    observations = {}

    with requests.Session() as session:
        entities = get(session, "/entities", {"org": ORG_ID})
        for index, entity in enumerate(entities):
            if index:
                time.sleep(3.1)
            hosts = get(session, "/assets", {"entity": entity["asset"]})
            for host in hosts:
                cert = host.get("cert_object") or {}
                if not cert or not any(cert.values()):
                    continue
                key = cert.get("sha256_fingerprint") or "|".join(
                    [
                        str(host.get("host") or ""),
                        str(cert.get("issuer_organization") or cert.get("issuer_common_name") or ""),
                        str(cert.get("not_after") or ""),
                    ]
                )
                expires = parse_time(cert.get("not_after"))
                issuer = cert.get("issuer_organization") or cert.get("issuer_common_name") or ""
                observations[key] = {
                    "host": host.get("host"),
                    "entity": entity["asset"],
                    "last_seen": host.get("last_seen"),
                    "dns_names": cert.get("dns_names") or [],
                    "issuer": issuer,
                    "not_after": cert.get("not_after"),
                    "days_remaining": (expires - now).days if expires else None,
                    "hostname_valid": cert.get("is_valid_hostname"),
                }

    previous = {}
    if STATE_PATH.exists():
        previous = json.loads(STATE_PATH.read_text(encoding="utf-8")).get("certificates", {})

    report = {
        "new_certificates": [value for key, value in observations.items() if key not in previous],
        "expiring": [
            value
            for value in observations.values()
            if value["days_remaining"] is not None and value["days_remaining"] <= EXPIRY_DAYS
        ],
        "unexpected_issuers": [
            value
            for value in observations.values()
            if APPROVED_ISSUERS and value["issuer"].lower() not in APPROVED_ISSUERS
        ],
        "invalid_hostnames": [
            value for value in observations.values() if value["hostname_valid"] is False
        ],
    }
    save_snapshot({"organization_id": ORG_ID, "certificates": observations})
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()