Build with the API / Brand and domain dark-web monitor

Brand and domain dark-web monitor

Are my brands or domains showing up on the dark web?

What you get

Matches a list of keywords (brand names, domains, subsidiaries) against ransomware leak-site victims and ransomware groups in one call, and prints the hits per bucket with dates and links.

Endpoints
/darkweb/keyword-hits
Credits per run
3
Key
Works on the free key

The code

brand-domain-monitor.py, 104 lines. Python 3, needs requests.

#!/usr/bin/env python3
"""brand-domain-monitor: are my brands or domains showing up on the dark web?

What it does
  Matches a list of keywords (brand names, domains, subsidiaries) against ransomware
  leak-site victims and ransomware groups in one call,
  and prints the hits per bucket with dates and links.

Endpoints
  GET /darkweb/keyword-hits?keywords=a,b,c   (3 credits, however many keywords)

Cost: 3 credits per run. Works on the free tier (7-day window, up to 5 hits per bucket).

Usage
  python3 brand-domain-monitor.py acme acme-corp.example "Acme Holdings"
  (The default keywords are generic sector words, so the demo has something to show.)
"""
import argparse
import json
import os
import sys
import time

import requests

API_BASE = os.environ.get("THREATCLUSTER_API_BASE", "https://threatcluster.io/api/public/v1").rstrip("/")
API_KEY = os.environ.get("THREATCLUSTER_API_KEY", "")
SITE = os.environ.get("THREATCLUSTER_SITE", "https://threatcluster.io")  # where the linked pages live
PACE_SECONDS = 2.1  # free keys get 30 requests/min; ~28/min keeps loops clear of 429s
_credits = {"used": 0, "remaining": None}


def _auth_headers():
    # A tc_... key is sent as X-API-Key; anything else is a short-lived bearer (e.g. from `tc login`).
    if not API_KEY:
        sys.exit("THREATCLUSTER_API_KEY is not set. Free keys: https://threatcluster.io/about/api")
    if API_KEY.startswith("tc_"):
        return {"X-API-Key": API_KEY}
    return {"Authorization": "Bearer " + API_KEY}


def api_get(path, _allow=(), **params):
    """GET one endpoint. Exits non-zero with the API's own error body, unless the
    status is listed in _allow (per-item lookups where a 404 is itself an answer)."""
    r = requests.get(API_BASE + path, headers=_auth_headers(), params=params, timeout=60)
    if "X-Request-Cost" in r.headers:
        _credits["used"] += int(r.headers["X-Request-Cost"])
    if "X-RateLimit-Remaining" in r.headers:
        _credits["remaining"] = r.headers["X-RateLimit-Remaining"]
    if r.status_code >= 400 and r.status_code not in _allow:
        try:
            body = json.dumps(r.json())
        except ValueError:
            body = r.text[:500]
        sys.exit("HTTP %d from %s: %s" % (r.status_code, path, body))
    return r


def cluster_url(cluster):
    # The free payload carries `slug` (canonical page) and `cluster_id`; the last 8 hex
    # chars of the id also resolve, via a 301 to the slug, when slug is missing.
    ident = cluster.get("slug") or cluster.get("short_id") or cluster.get("cluster_id", "").replace("-", "")[-8:]
    return "%s/cluster/%s" % (SITE, ident)


def report_credits():
    # Stderr, so it never pollutes piped output (brief.md, blocklists, JSON).
    remaining = _credits["remaining"] if _credits["remaining"] is not None else "n/a"
    sys.stderr.write("[threatcluster] credits used this run: %d, remaining today: %s\n" % (_credits["used"], remaining))

DEFAULT_KEYWORDS = ["bank", "hospital", "school"]


def matched(keywords, *fields):
    blob = " ".join(str(f or "") for f in fields).lower()
    return ", ".join(k for k in keywords if k.lower() in blob) or "(matched on other fields)"


def main():
    ap = argparse.ArgumentParser(description="Dark-web keyword hits for brands/domains")
    ap.add_argument("keywords", nargs="*", default=DEFAULT_KEYWORDS, help="brand names, domains, subsidiaries")
    ap.add_argument("--per-bucket", type=int, default=25, help="max hits per bucket (free tier caps at 5)")
    args = ap.parse_args()

    data = api_get("/darkweb/keyword-hits", keywords=",".join(args.keywords), per_bucket_limit=args.per_bucket).json()
    hits = data.get("hits", {})
    print("Keywords: %s" % ", ".join(data.get("keywords", args.keywords)))
    print("Total hits: %s (window: last %s days)\n" % (data.get("total", 0), data.get("lookback_days", "?")))

    for v in hits.get("victims", []):
        print("VICTIM   %s  %-14s %-32s %s  %s"
              % ((v.get("discovered") or "")[:10], v.get("group_name") or "?", (v.get("victim_name") or "?")[:32],
                 v.get("country") or "--", v.get("sector") or ""))
        print("         matched: %s  %s/dark-web/victim/%s" % (matched(args.keywords, v.get("victim_name")), SITE, v.get("id", "")))
    for g in hits.get("groups", []):
        print("GROUP    %s  %s/dark-web/group/%s" % (g.get("name") or g.get("group_name") or "?", SITE, g.get("name") or ""))
    empty = [k for k in ("victims", "groups") if not hits.get(k)]
    if empty:
        print("\nNo hits in: %s" % ", ".join(empty))
    report_credits()


if __name__ == "__main__":
    main()

The shell one-liner, brand-domain-monitor.sh:

#!/usr/bin/env bash
# brand-domain-monitor.sh: dark-web keyword hits for the comma-separated keywords in arg 1. 3 credits. Free tier OK.
# usage: ./brand-domain-monitor.sh "acme,acme-corp.example"
set -euo pipefail
: "${THREATCLUSTER_API_KEY:?set THREATCLUSTER_API_KEY (free keys: https://threatcluster.io/about/api)}"
BASE="${THREATCLUSTER_API_BASE:-https://threatcluster.io/api/public/v1}"
SITE="${THREATCLUSTER_SITE:-https://threatcluster.io}"
# A tc_... key goes in X-API-Key; anything else is a bearer (e.g. from `tc login`).
if [[ "$THREATCLUSTER_API_KEY" == tc_* ]]; then AUTH="X-API-Key: $THREATCLUSTER_API_KEY"; else AUTH="Authorization: Bearer $THREATCLUSTER_API_KEY"; fi
# tc_get <path> [curl -G args...]: prints the body; on HTTP >= 400 prints the API error to stderr and fails.
tc_get() { local path=$1; shift; local out code
  out=$(curl -sS -G -H "$AUTH" -w $'\n%{http_code}' "$BASE$path" "$@"); code=${out##*$'\n'}
  if [ "$code" -ge 400 ]; then echo "HTTP $code from $path: ${out%$'\n'*}" >&2; return 1; fi; printf '%s' "${out%$'\n'*}"; }
tc_get /darkweb/keyword-hits --data-urlencode "keywords=${1:?comma-separated keywords}" --data-urlencode per_bucket_limit=25 \
 | jq -r --arg site "$SITE" '"total hits: \(.total)",
     (.hits.victims[]  | "VICTIM  \(.discovered[:10])  \(.group_name)  \(.victim_name)  (\(.country // "--"))  \($site)/dark-web/victim/\(.id)"),
     (.hits.groups[]   | "GROUP   \(.name // .group_name)")'

Output

From a run on 3 September 2026. Yours will differ as the corpus moves.

[threatcluster] credits used this run: 3, remaining today: 62
Keywords: bank, hospital, school
Total hits: 5 (window: last 7 days)

VICTIM   2026-09-02  incransom      Westfield Public School District US  Education
         matched: school  https://threatcluster.io/dark-web/victim/37604ebc815f79f6
VICTIM   2026-09-02  incransom      Policlinico Triestino            IT  Healthcare
         matched: (matched on other fields)  https://threatcluster.io/dark-web/victim/8fcff03fdbfe25d0
VICTIM   2026-09-02  incransom      specialtytextile.com             US  Manufacturing
         matched: (matched on other fields)  https://threatcluster.io/dark-web/victim/6d5099d925516372
VICTIM   2026-09-02  aurora         Chip 1 Exchange                  US  Technology
         matched: (matched on other fields)  https://threatcluster.io/dark-web/victim/9490ece3ed4cb8e3
VICTIM   2026-09-01  thegentlemen   Nutex Health                     US  Healthcare
         matched: (matched on other fields)  https://threatcluster.io/dark-web/victim/dff02778ccad0dc8

No hits in: groups, markets, breaches

Run it

  1. Get a free key at /api. Every account has one; it covers the endpoints above.
  2. export THREATCLUSTER_API_KEY=… your key.
  3. python3 brand-domain-monitor.py

The file on GitHub: brand-domain-monitor.py and brand-domain-monitor.sh.

Goes well with

All recipes: /build. Endpoint reference: /api/public/v1/docs.