Respond to Compromised Credential Records

This workflow creates a privacy-reduced incident queue from Enterprise dark-web records. FullHunt supplies evidence. Password reset, session revocation, user notification, and case closure happen in your identity and incident-response systems.

Requirements

Enterprise access with the Dark Web Monitoring module
A monitored company domain
An approved retention and access policy for credential-exposure data
Python 3.10 or later with requests

Do not test returned credentials against an account or system. Do not copy raw passwords into logs, tickets, chat, or state files.

Evidence sources

Dataset	Endpoint	Filter	Page size
Compromised credentials	`GET /enterprise/darkweb/compromised-credentials`	`query`	1,000
Discovered emails	`GET /enterprise/darkweb/discovered-emails`	`q`	1,000

The two endpoints use different filter names and response envelopes. The collector normalizes both without retaining password, hashed_password, address, phone, IP address, or other unnecessary fields.

Run the collector

export FULLHUNT_API_KEY="YOUR_API_KEY"
export FULLHUNT_MONITORED_DOMAIN="your-owned-domain.example"
python collect_credential_exposure.py

collect_credential_exposure.py

import hashlib
import json
import os
from pathlib import Path

import requests

BASE_URL = "https://fullhunt.io/api/v1/enterprise/darkweb"
API_KEY = os.environ["FULLHUNT_API_KEY"]
DOMAIN = os.environ["FULLHUNT_MONITORED_DOMAIN"]
STATE_PATH = Path(os.environ.get("FULLHUNT_CREDENTIAL_STATE", "credential-exposure-state.json"))
BASELINE_ONLY = os.environ.get("FULLHUNT_BASELINE_ONLY") == "1"
HEADERS = {"X-API-KEY": API_KEY}
PAGE_SIZE = 1000
MAX_PAGES = 100


def mask_email(value):
    local, separator, domain = (value or "").partition("@")
    if not separator:
        return "redacted"
    visible = local[:1] if local else ""
    return f"{visible}***@{domain}"


def stable_key(dataset, record):
    if record.get("id"):
        return f"{dataset}:{record['id']}"
    material = "|".join(
        str(record.get(field) or "")
        for field in ("email", "database_name", "darkweb_metadata_name", "breach_date")
    )
    digest = hashlib.sha256(material.encode("utf-8")).hexdigest()
    return f"{dataset}:sha256:{digest}"


def fetch_compromised(session):
    records = []
    for page in range(1, MAX_PAGES + 1):
        response = session.get(
            f"{BASE_URL}/compromised-credentials",
            headers=HEADERS,
            params={"query": f"@{DOMAIN}", "page": page},
            timeout=30,
        )
        response.raise_for_status()
        payload = response.json()
        items = payload.get("results")
        if not isinstance(items, list):
            raise ValueError("Unexpected compromised-credentials response")
        records.extend(items)
        total = int(payload.get("total_results") or len(records))
        if len(items) < PAGE_SIZE or len(records) >= total:
            return records
    raise RuntimeError("Compromised-credential pagination reached its page guard")


def fetch_discovered(session):
    records = []
    for page in range(1, MAX_PAGES + 1):
        response = session.get(
            f"{BASE_URL}/discovered-emails",
            headers=HEADERS,
            params={"q": DOMAIN, "page": page},
            timeout=30,
        )
        response.raise_for_status()
        results = response.json().get("results")
        if not isinstance(results, dict) or not isinstance(results.get("items"), list):
            raise ValueError("Unexpected discovered-emails response")
        records.extend(results["items"])
        pages = int(results.get("pages") or page)
        if page >= pages:
            return records
    raise RuntimeError("Discovered-email pagination reached its page guard")


def load_seen():
    if not STATE_PATH.exists():
        return set()
    return set(json.loads(STATE_PATH.read_text(encoding="utf-8")).get("seen", []))


def save_seen(seen):
    temporary = STATE_PATH.with_suffix(STATE_PATH.suffix + ".tmp")
    temporary.write_text(json.dumps({"seen": sorted(seen)}, indent=2) + "\n", encoding="utf-8")
    temporary.replace(STATE_PATH)


def reduced_record(dataset, record):
    return {
        "dataset": dataset,
        "record_id": record.get("id"),
        "email": mask_email(record.get("email")),
        "email_sha256": hashlib.sha256((record.get("email") or "").lower().encode("utf-8")).hexdigest(),
        "breach_name": record.get("darkweb_metadata_name") or record.get("database_name"),
        "breach_date": record.get("breach_date"),
        "ingested_at": record.get("date_added"),
        "exposed_data_types": record.get("darkweb_metadata_leaked_data"),
    }


def main():
    seen = load_seen()
    updated = set(seen)
    incidents = []

    with requests.Session() as session:
        datasets = {
            "compromised-credentials": fetch_compromised(session),
            "discovered-emails": fetch_discovered(session),
        }

    for dataset, records in datasets.items():
        for record in records:
            key = stable_key(dataset, record)
            updated.add(key)
            if key not in seen and not BASELINE_ONLY:
                incidents.append(reduced_record(dataset, record))

    save_seen(updated)
    print(json.dumps({"new_incidents": incidents, "baseline_only": BASELINE_ONLY}, indent=2))


if __name__ == "__main__":
    main()

Response procedure

Match the reduced email identifier to an active identity through an access-controlled internal process.
Confirm whether the returned breach source and date are relevant to the identity.
Apply your identity policy, such as password reset, session revocation, multifactor-authentication review, or credential rotation.
Record the action and owner in the incident system without copying raw credential values.
Close the case only after the identity control confirms the required action.

The first run reports every returned record as new. Set FULLHUNT_BASELINE_ONLY=1 to create state without generating an initial incident backlog.

Data handling

Encrypt state and incident records at rest and restrict access by job role.
Retain stable FullHunt record IDs and source dates for evidence lineage.
Define deletion windows for resolved records and departed users.
Avoid storing the full API response unless a documented legal or incident-response need requires it.
Treat an empty query result as not returned, not proof that no credential exposure exists.

Knowledge Base

Guides

Requirements

Evidence sources

Run the collector

Response procedure

Data handling

​Requirements

​Evidence sources

​Run the collector

​Response procedure

​Data handling

Requirements

Evidence sources

Run the collector

Response procedure

Data handling