riposte-marketplace/integrations/rdap/scripts/domain_lookup.py

import datetime, ipaddress, json, os, ssl, sys, urllib.error, urllib.request

DEFAULT_BOOTSTRAP = "https://data.iana.org/rdap"
DEFAULT_FALLBACK = "https://rdap.org"


def _cfg():
    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))


def _inputs():
    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))


def _ctx(cfg):
    if cfg.get("insecure") in (True, "true", "True", 1, "1"):
        c = ssl.create_default_context()
        c.check_hostname = False
        c.verify_mode = ssl.CERT_NONE
        return c
    return None


def get(cfg, url):
    req = urllib.request.Request(url, headers={"Accept": "application/rdap+json, application/json", "User-Agent": "riposte-soar"})
    with urllib.request.urlopen(req, timeout=45, context=_ctx(cfg)) as r:
        raw = r.read()
        return json.loads(raw) if raw else {}


def bootstrap(cfg, name):
    base = (str(cfg.get("bootstrap_url") or "").strip() or DEFAULT_BOOTSTRAP).rstrip("/")
    return get(cfg, base + "/" + name + ".json").get("services", [])


def fallback(cfg):
    v = cfg.get("fallback_url")
    return DEFAULT_FALLBACK if v is None else str(v).strip().rstrip("/")


def server_url(urls):
    # Prefer https; every bootstrap entry lists its servers with a trailing slash.
    urls = sorted(urls, key=lambda u: not u.startswith("https"))
    return urls[0].rstrip("/") if urls else ""


def _run(fn):
    try:
        print(json.dumps(fn(_cfg(), _inputs())))
    except urllib.error.HTTPError as e:
        print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")[:2000]}))
        sys.exit(1)
    except Exception as e:
        print(json.dumps({"error": str(e)}))
        sys.exit(1)


# --- reading an RDAP record -------------------------------------------------

def vcard(entity):
    out = {}
    arr = entity.get("vcardArray") or []
    for prop in (arr[1] if len(arr) > 1 and isinstance(arr[1], list) else []):
        if not isinstance(prop, list) or len(prop) < 4:
            continue
        key, val = prop[0], prop[3]
        if isinstance(val, list):
            val = " ".join(str(x) for x in val if x)
        if key in ("fn", "org", "email", "tel") and val and key not in out:
            out[key] = str(val).replace("tel:", "")
        if key == "adr" and isinstance(prop[1], dict) and prop[1].get("label"):
            out.setdefault("address", prop[1]["label"])
    return out


def entities_by_role(rec):
    found = {}

    def walk(ents):
        for e in ents or []:
            card = vcard(e)
            card["handle"] = e.get("handle", "")
            for role in e.get("roles") or []:
                found.setdefault(role, []).append(card)
            walk(e.get("entities"))
    walk(rec.get("entities"))
    return found


def first(found, role, key):
    for card in found.get(role, []):
        if card.get(key):
            return card[key]
    return ""


def events(rec):
    out = {}
    for ev in rec.get("events") or []:
        a, d = ev.get("eventAction", ""), ev.get("eventDate", "")
        if a and d and a not in out:
            out[a] = d
    return out


def age_days(iso):
    if not iso:
        return None
    try:
        t = datetime.datetime.fromisoformat(iso.replace("Z", "+00:00"))
        if t.tzinfo is None:
            t = t.replace(tzinfo=datetime.timezone.utc)
        return (datetime.datetime.now(datetime.timezone.utc) - t).days
    except ValueError:
        return None


REDACTED = ("redacted", "ano nymous", "data protected", "privacy", "not disclosed", "withheld")


def disclosed(v):
    # Registries replace personal data with placeholders (GDPR); a placeholder
    # is no registrant, and naming one would read as an identity.
    return "" if any(m in str(v).lower() for m in REDACTED) else v


def common(rec, server):
    found = entities_by_role(rec)
    ev = events(rec)
    registrant = found.get("registrant", [{}])[0] if found.get("registrant") else {}
    return {
        "handle": rec.get("handle", ""),
        "status": rec.get("status") or [],
        "created": ev.get("registration", ""),
        "updated": ev.get("last changed", ""),
        "expires": ev.get("expiration", ""),
        "age_days": age_days(ev.get("registration", "")),
        "registrar": first(found, "registrar", "fn") or first(found, "registrar", "org"),
        "registrant": disclosed(registrant.get("org") or registrant.get("fn") or ""),
        "abuse_email": first(found, "abuse", "email"),
        "abuse_phone": first(found, "abuse", "tel"),
        "rdap_server": server,
    }


def strip(rec):
    # Notices and links are the registry's terms of use and navigation: kilobytes
    # of boilerplate no analyst reads, dropped so the record stays readable.
    if not isinstance(rec, dict):
        return rec
    return {k: v for k, v in rec.items() if k not in ("notices", "links", "rdapConformance")}


def want_raw(inputs):
    return inputs.get("include_raw") not in (False, "false", "False", 0, "0")


def find_server(cfg, name):
    labels = name.split(".")
    services = bootstrap(cfg, "dns")
    for i in range(len(labels)):
        suffix = ".".join(labels[i:])
        for tlds, urls in services:
            if suffix in [t.lower() for t in tlds]:
                return server_url(urls)
    return fallback(cfg)


def main(cfg, inputs):
    name = str(inputs.get("domain", "")).strip().lower().rstrip(".")
    for p in ("https://", "http://"):
        if name.startswith(p):
            name = name[len(p):]
    name = name.split("/")[0].split(":")[0]
    if "." not in name:
        raise Exception("not a domain name: " + name)
    server = find_server(cfg, name)
    if not server:
        raise Exception("no RDAP server for this TLD and no fallback_url set")
    # A host name is not a registration: walk up from the full name to the
    # registered domain, stopping at the first the registry knows. Two labels is
    # the floor — a ccTLD second level (co.uk) answers 404 and the walk ends
    # with the registry's own 404.
    labels = name.split(".")
    last = None
    for i in range(0, max(1, len(labels) - 1)):
        candidate = ".".join(labels[i:])
        try:
            rec = get(cfg, server + "/domain/" + candidate)
        except urllib.error.HTTPError as e:
            if e.code != 404:
                raise
            body = e.read().decode("utf-8", "replace")
            # The public redirector answers 404 for a TLD whose registry runs no
            # RDAP (.ch, .de…): that is not "unregistered", and must not read as
            # an empty answer in the enrichment panel.
            if "no rdap service" in body.lower():
                raise Exception("the registry of ." + labels[-1] + " offers no RDAP service")
            last = urllib.error.HTTPError(e.url, 404, e.msg, e.hdrs, None)
            last.read = lambda b=body: b.encode("utf-8")
            continue
        out = {"domain": rec.get("ldhName", candidate).lower(), "queried": name}
        out.update(common(rec, server))
        out["nameservers"] = sorted({str(n.get("ldhName", "")).lower() for n in rec.get("nameservers") or [] if n.get("ldhName")})
        sd = rec.get("secureDNS") or {}
        out["dnssec"] = bool(sd.get("delegationSigned"))
        if want_raw(inputs):
            out["rdap"] = strip(rec)
        return out
    raise last


_run(main)