How to Build a Brand Protection Monitoring System with Domain APIs

HOMOGLYPHS = {
    'a': ['а', '@'],      # Cyrillic а, at-sign
    'e': ['е', '3'],      # Cyrillic е, digit 3
    'i': ['і', '1', 'l'], # Cyrillic і, digit 1, lowercase L
    'o': ['о', '0'],      # Cyrillic о, digit 0
    's': ['$', '5'],
    'l': ['1', 'I'],
}

BRAND_KEYWORDS = [
    'login', 'secure', 'support', 'account',
    'verify', 'update', 'help', 'service', 'portal',
]

WATCHLIST_TLDS = ['.net', '.io', '.co', '.org', '.eu', '.app', '.dev']


def generate_variants(brand: str, primary_tld: str = '.com') -> list[str]:
    variants = set()
    b = brand.lower()

    # --- TLD variants ---
    for tld in WATCHLIST_TLDS:
        variants.add(f"{b}{tld}")

    # --- Keyword combinations ---
    for kw in BRAND_KEYWORDS:
        variants.add(f"{b}-{kw}{primary_tld}")
        variants.add(f"{kw}-{b}{primary_tld}")

    # --- Typosquatting: character transpositions ---
    for i in range(len(b) - 1):
        swapped = list(b)
        swapped[i], swapped[i+1] = swapped[i+1], swapped[i]
        variants.add(f"{''.join(swapped)}{primary_tld}")

    # --- Typosquatting: character omissions ---
    for i in range(len(b)):
        variants.add(f"{b[:i] + b[i+1:]}{primary_tld}")

    # --- Typosquatting: character duplications ---
    for i in range(len(b)):
        variants.add(f"{b[:i] + b[i] + b[i:]}{primary_tld}")

    # --- Homoglyph substitutions ---
    for i, char in enumerate(b):
        for replacement in HOMOGLYPHS.get(char, []):
            mutated = b[:i] + replacement + b[i+1:]
            variants.add(f"{mutated}{primary_tld}")

    # Remove the original domain itself
    variants.discard(f"{b}{primary_tld}")
    return sorted(variants)


# Example: generate_variants("acmecorp") → 80+ candidates

# pip install aiohttp
import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"
BASE    = "https://whoisjson.com/api/v1"
HEADERS = {"Authorization": f"TOKEN={API_KEY}"}


async def get(session: aiohttp.ClientSession, path: str, domain: str) -> dict:
    try:
        async with session.get(
            f"{BASE}/{path}",
            params={"domain": domain},
            headers=HEADERS,
            timeout=aiohttp.ClientTimeout(total=12),
        ) as resp:
            return await resp.json() if resp.status == 200 else {}
    except Exception:
        return {}


async def enrich_domain(session: aiohttp.ClientSession, domain: str) -> dict | None:
    """Return None for unregistered domains to skip downstream API calls."""
    avail = await get(session, "domain-availability", domain)
    if avail.get("available", True):
        return None   # not registered — skip

    whois, dns, ssl = await asyncio.gather(
        get(session, "whois",          domain),
        get(session, "nslookup",       domain),
        get(session, "ssl-cert-check", domain),
    )
    return {"domain": domain, "whois": whois, "dns": dns, "ssl": ssl}

from datetime import datetime, timezone


def score_domain(e: dict) -> dict:
    pts   = 0
    flags = []
    w = e.get("whois") or {}
    d = e.get("dns")   or {}
    s = e.get("ssl")   or {}

    # +40 — newly registered (< 30 days)
    if (w.get("age") or {}).get("isNewlyRegistered"):
        pts += 40; flags.append("newly_registered")

    # +20 — active MX record
    if d.get("MX"):
        pts += 20; flags.append("mx_active")

    # +20 — SSL cert issued within 7 days of domain creation
    ssl_from   = (s.get("valid_from") or "")[:10]
    whois_crt  = (w.get("created")    or "")[:10]
    if ssl_from and whois_crt:
        try:
            delta = (
                datetime.fromisoformat(ssl_from) -
                datetime.fromisoformat(whois_crt)
            ).days
            if 0 <= delta <= 7:
                pts += 20; flags.append("cert_issued_quickly")
        except ValueError:
            pass

    # +20 — fully redacted registrant
    owner = ((w.get("contacts") or {}).get("owner") or [{}])[0]
    if not any([owner.get("name"), owner.get("organization"), owner.get("email")]):
        pts += 20; flags.append("redacted_contacts")

    pts = min(pts, 100)
    tier = "HIGH" if pts >= 60 else ("MEDIUM" if pts >= 30 else "LOW")
    return {
        "domain":    e["domain"],
        "score":     pts,
        "risk":      tier,
        "flags":     flags,
        "created":   w.get("created"),
        "registrar": (w.get("registrar") or {}).get("name"),
        "a_records": d.get("A", []),
        "mx_records": [mx.get("exchange") for mx in d.get("MX", [])],
        "ssl_issuer": (s.get("issuer") or {}).get("O"),
        "ssl_expiry": s.get("valid_to"),
    }

{
  "domain":     "acmecorp-login.com",
  "score":      100,
  "risk":       "HIGH",
  "flags":      ["newly_registered", "mx_active", "cert_issued_quickly", "redacted_contacts"],
  "created":    "2026-04-26 07:12:00",
  "registrar":  "Namecheap, Inc.",
  "a_records":  ["104.21.45.89"],
  "mx_records": ["mail.acmecorp-login.com"],
  "ssl_issuer": "Let's Encrypt",
  "ssl_expiry": "2026-07-25T07:12:00.000Z"
}

# pip install aiohttp
import asyncio, json
import aiohttp
from datetime import datetime

API_KEY     = "YOUR_API_KEY"
BASE        = "https://whoisjson.com/api/v1"
HEADERS     = {"Authorization": f"TOKEN={API_KEY}"}
CONCURRENCY = 10          # tune to plan RPM / 4

HOMOGLYPHS = {'a': ['а','@'], 'e': ['е','3'], 'i': ['і','1'], 'o': ['о','0'], 's': ['$']}
KEYWORDS   = ['login', 'secure', 'support', 'account', 'verify']
TLDS       = ['.net', '.io', '.co', '.org']


def generate_variants(brand: str, primary_tld: str = '.com') -> list[str]:
    b, variants = brand.lower(), set()
    for tld in TLDS:
        variants.add(f"{b}{tld}")
    for kw in KEYWORDS:
        variants.add(f"{b}-{kw}{primary_tld}")
        variants.add(f"{kw}-{b}{primary_tld}")
    for i in range(len(b) - 1):
        s = list(b); s[i], s[i+1] = s[i+1], s[i]
        variants.add(f"{''.join(s)}{primary_tld}")
    for i in range(len(b)):
        variants.add(f"{b[:i]+b[i+1:]}{primary_tld}")
        variants.add(f"{b[:i]+b[i]+b[i:]}{primary_tld}")
    for i, c in enumerate(b):
        for r in HOMOGLYPHS.get(c, []):
            variants.add(f"{b[:i]+r+b[i+1:]}{primary_tld}")
    variants.discard(f"{b}{primary_tld}")
    return sorted(variants)


async def get(s: aiohttp.ClientSession, path: str, domain: str) -> dict:
    try:
        async with s.get(f"{BASE}/{path}", params={"domain": domain},
                         headers=HEADERS,
                         timeout=aiohttp.ClientTimeout(total=12)) as r:
            return await r.json() if r.status == 200 else {}
    except Exception:
        return {}


async def enrich(s: aiohttp.ClientSession, domain: str) -> dict | None:
    avail = await get(s, "domain-availability", domain)
    if avail.get("available", True):
        return None
    whois, dns, ssl = await asyncio.gather(
        get(s, "whois", domain),
        get(s, "nslookup", domain),
        get(s, "ssl-cert-check", domain),
    )
    return {"domain": domain, "whois": whois, "dns": dns, "ssl": ssl}


def score(e: dict) -> dict:
    pts, flags = 0, []
    w, d, s = e.get("whois") or {}, e.get("dns") or {}, e.get("ssl") or {}
    if (w.get("age") or {}).get("isNewlyRegistered"):
        pts += 40; flags.append("newly_registered")
    if d.get("MX"):
        pts += 20; flags.append("mx_active")
    ssl_from, crt = (s.get("valid_from") or "")[:10], (w.get("created") or "")[:10]
    try:
        if ssl_from and crt and 0 <= (datetime.fromisoformat(ssl_from) - datetime.fromisoformat(crt)).days <= 7:
            pts += 20; flags.append("cert_issued_quickly")
    except ValueError:
        pass
    owner = ((w.get("contacts") or {}).get("owner") or [{}])[0]
    if not any([owner.get("name"), owner.get("organization"), owner.get("email")]):
        pts += 20; flags.append("redacted_contacts")
    pts = min(pts, 100)
    return {"domain": e["domain"], "score": pts,
            "risk": "HIGH" if pts >= 60 else ("MEDIUM" if pts >= 30 else "LOW"),
            "flags": flags, "created": w.get("created"),
            "registrar": (w.get("registrar") or {}).get("name")}


def alert_if_high_risk(result: dict) -> None:
    if result["risk"] == "HIGH":
        print(f"[ALERT] HIGH-RISK domain detected: {result['domain']}")
        print(f"        Score: {result['score']}/100 — Flags: {', '.join(result['flags'])}")
        print(f"        Created: {result['created']} — Registrar: {result['registrar']}")


async def run(brand: str) -> list[dict]:
    variants = generate_variants(brand)
    sem = asyncio.Semaphore(CONCURRENCY)
    results = []

    async def bounded(domain: str):
        async with sem:
            return await enrich(session, domain)

    async with aiohttp.ClientSession() as session:
        enriched = await asyncio.gather(*[bounded(d) for d in variants])

    for e in enriched:
        if e is not None:
            r = score(e)
            alert_if_high_risk(r)
            results.append(r)

    return sorted(results, key=lambda x: x["score"], reverse=True)


if __name__ == "__main__":
    brand = "acmecorp"   # replace with your brand name
    output = asyncio.run(run(brand))
    print(json.dumps(output, indent=2, ensure_ascii=False))