riposte-marketplace/integrations/rdap/scripts/ip_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 main(cfg, inputs):
    raw = str(inputs.get("ip", "")).strip()
    addr = ipaddress.ip_address(raw)
    server = ""
    for prefixes, urls in bootstrap(cfg, "ipv4" if addr.version == 4 else "ipv6"):
        if any(addr in ipaddress.ip_network(p, strict=False) for p in prefixes):
            server = server_url(urls)
            break
    server = server or fallback(cfg)
    if not server:
        raise Exception("no RDAP server for this address and no fallback_url set")
    rec = get(cfg, server + "/ip/" + str(addr))
    out = {"ip": str(addr)}
    out.update(common(rec, server))
    found = entities_by_role(rec)
    out["network_name"] = rec.get("name", "")
    out["country"] = rec.get("country", "")
    out["range"] = "%s - %s" % (rec.get("startAddress", ""), rec.get("endAddress", ""))
    cidrs = [("%s/%s" % (c.get("v4prefix") or c.get("v6prefix"), c.get("length"))) for c in rec.get("cidr0_cidrs") or []]
    out["cidr"] = cidrs
    out["holder"] = first(found, "registrant", "org") or first(found, "registrant", "fn") or out["registrant"]
    out["registrant"] = out["holder"]
    if want_raw(inputs):
        out["rdap"] = strip(rec)
    return out


_run(main)