From d2742ec6fb29b53c026d6e4ed021a1c8a1f7bc23 Mon Sep 17 00:00:00 2001 From: Guillaume Bourgeois Date: Fri, 9 Oct 2026 22:33:42 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20CIRCL=20hashlookup,=20RDAP=20and=20crt.?= =?UTF-8?q?sh=20=E2=80=94=20free=20IOC=20lookups,=20no=20API=20key?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three keyless sources for the questions an analyst asks first: - circl-hashlookup 1.0.0: is this hash known software (NSRL, distro packages, Windows updates)? `score` 0 for a known file, 100 when CIRCL flags it KnownMalicious, so an enrichment rule with benign 0 marks it benign; an unknown hash is an HTTP 404, read as "no record". Single and bulk lookups; base_url for a self-hosted instance. - rdap 1.0.0: who registered a domain, IP block or AS, and when — with age_days for the phishing question. Authoritative server from the IANA bootstrap, rdap.org as fallback; a host name walks up to its registered domain; GDPR placeholders are not reported as registrants; a TLD with no RDAP (.ch) is an error, not a fake "unregistered". - crtsh 1.0.0: Certificate Transparency — sub-domains with first/last seen, issuers, recent certificates, and a starts_with mode for look-alikes of a brand (crt.sh no longer searches with a leading wildcard). Retries the 502/503/504 its shared front throws under load. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AmWT9wtWHGtZpMS64Hp4q5 --- integrations/circl-hashlookup/manifest.yaml | 48 ++++ .../circl-hashlookup/scripts/bulk_lookup.py | 106 +++++++++ .../circl-hashlookup/scripts/lookup_hash.py | 73 ++++++ .../scripts/test_connection.py | 73 ++++++ integrations/crtsh/manifest.yaml | 45 ++++ .../crtsh/scripts/search_certificates.py | 133 +++++++++++ integrations/crtsh/scripts/test_connection.py | 60 +++++ integrations/rdap/manifest.yaml | 65 ++++++ integrations/rdap/scripts/asn_lookup.py | 190 ++++++++++++++++ integrations/rdap/scripts/domain_lookup.py | 213 ++++++++++++++++++ integrations/rdap/scripts/ip_lookup.py | 185 +++++++++++++++ integrations/rdap/scripts/test_connection.py | 163 ++++++++++++++ 12 files changed, 1354 insertions(+) create mode 100644 integrations/circl-hashlookup/manifest.yaml create mode 100644 integrations/circl-hashlookup/scripts/bulk_lookup.py create mode 100644 integrations/circl-hashlookup/scripts/lookup_hash.py create mode 100644 integrations/circl-hashlookup/scripts/test_connection.py create mode 100644 integrations/crtsh/manifest.yaml create mode 100644 integrations/crtsh/scripts/search_certificates.py create mode 100644 integrations/crtsh/scripts/test_connection.py create mode 100644 integrations/rdap/manifest.yaml create mode 100644 integrations/rdap/scripts/asn_lookup.py create mode 100644 integrations/rdap/scripts/domain_lookup.py create mode 100644 integrations/rdap/scripts/ip_lookup.py create mode 100644 integrations/rdap/scripts/test_connection.py diff --git a/integrations/circl-hashlookup/manifest.yaml b/integrations/circl-hashlookup/manifest.yaml new file mode 100644 index 0000000..9487aad --- /dev/null +++ b/integrations/circl-hashlookup/manifest.yaml @@ -0,0 +1,48 @@ +id: circl_hashlookup +name: CIRCL hashlookup +version: 1.0.0 +description: "CIRCL hashlookup — tells whether a file hash belongs to known software: NSRL, Linux distribution packages, Windows updates and other trusted sets (6.5 billion hashes). The fastest way to rule out a false positive on a file alert. Free public service, no API key; the server is open source and can be self-hosted (set base_url). Stdlib-only. As an enrichment rule: command lookup_hash, score field `score`, benign threshold 0 and malicious threshold 100 — a known file reads benign, a record CIRCL flags as KnownMalicious reads malicious, and an unknown hash is an answer (not found), never a failure." +changelog: "1.0.0 — Initial release: single lookup (MD5, SHA-1, SHA-256), bulk lookup (MD5, SHA-1), test connection." +category: enrichment + +config_schema: + properties: + base_url: + type: string + description: "hashlookup server (default https://hashlookup.circl.lu — set it to a self-hosted instance to keep hashes in-house)" + default: https://hashlookup.circl.lu + insecure: + type: boolean + description: "Trust any TLS certificate (not secure — a self-hosted instance with a private CA only)" + default: false + required: [] + +commands: + - id: lookup_hash + name: hashlookup-lookup-hash + description: "Look up one MD5, SHA-1 or SHA-256 hash. Returns the CIRCL record (file name, product, source set, hashlookup:trust) plus `known: true` and `score` (0 = known file, 100 = flagged KnownMalicious). An unknown hash returns HTTP 404, which Riposte's enrichment reads as 'no record'." + risk: read + inputs_schema: + properties: + hash: { type: string, description: "MD5 (32 hex), SHA-1 (40 hex) or SHA-256 (64 hex)" } + required: [hash] + outputs_schema: { properties: {} } + + - id: bulk_lookup + name: hashlookup-bulk-lookup + description: "Look up many MD5 / SHA-1 hashes at once (the bulk API does not take SHA-256; those are looked up one by one). Returns `known` (records found) and `unknown` (hashes CIRCL has no record of)." + risk: read + inputs_schema: + properties: + hashes: { type: string, description: "Hashes separated by commas, spaces or new lines (a JSON array also works)" } + required: [hashes] + outputs_schema: { properties: {} } + + - id: test_connection + name: hashlookup-test-connection + description: "Reach the server's /info endpoint (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/circl-hashlookup/scripts/bulk_lookup.py b/integrations/circl-hashlookup/scripts/bulk_lookup.py new file mode 100644 index 0000000..884d17f --- /dev/null +++ b/integrations/circl-hashlookup/scripts/bulk_lookup.py @@ -0,0 +1,106 @@ +import json, os, re, ssl, sys, urllib.error, urllib.request + +DEFAULT_BASE = "https://hashlookup.circl.lu" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "").strip() or DEFAULT_BASE).rstrip("/") + + +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 call(cfg, path, body=None): + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Accept": "application/json", "User-Agent": "riposte-soar"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(_base(cfg) + path, data=data, headers=headers, method="POST" if data is not None else "GET") + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + return json.loads(raw) if raw 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")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +KINDS = {32: "md5", 40: "sha1", 64: "sha256"} + + +def kind_of(h): + h = h.strip().lower() + if not re.fullmatch(r"[0-9a-f]+", h) or len(h) not in KINDS: + raise Exception("not an MD5, SHA-1 or SHA-256 hash: " + h) + return KINDS[len(h)], h + + +def annotate(rec): + # score: 0 for a known file, 100 when CIRCL itself flags the record as + # malicious — what an enrichment rule maps with benign 0 / malicious 100. + rec = dict(rec) + rec["known"] = True + rec["score"] = 100 if rec.get("KnownMalicious") else 0 + return rec + + +def parse_list(v): + if isinstance(v, list): + return [str(x) for x in v] + s = str(v or "").strip() + if s.startswith("["): + return [str(x) for x in json.loads(s)] + return [x for x in re.split(r"[\s,;]+", s) if x] + + +def main(cfg, inputs): + groups = {"md5": [], "sha1": [], "sha256": []} + for raw in parse_list(inputs.get("hashes")): + kind, h = kind_of(raw) + if h not in groups[kind]: + groups[kind].append(h) + known, unknown = [], [] + for kind in ("md5", "sha1"): + hs = groups[kind] + for i in range(0, len(hs), 500): + chunk = hs[i:i + 500] + res = call(cfg, "/bulk/" + kind, {"hashes": chunk}) + found = set() + for rec in res if isinstance(res, list) else []: + key = str(rec.get("MD5" if kind == "md5" else "SHA-1", "")).lower() + if key and "message" not in rec: + found.add(key) + known.append(annotate(rec)) + unknown.extend(h for h in chunk if h not in found) + for h in groups["sha256"]: + try: + known.append(annotate(call(cfg, "/lookup/sha256/" + h))) + except urllib.error.HTTPError as e: + if e.code != 404: + raise + unknown.append(h) + return {"known": known, "unknown": unknown, "known_count": len(known), "unknown_count": len(unknown)} + + +_run(main) diff --git a/integrations/circl-hashlookup/scripts/lookup_hash.py b/integrations/circl-hashlookup/scripts/lookup_hash.py new file mode 100644 index 0000000..6b52e3b --- /dev/null +++ b/integrations/circl-hashlookup/scripts/lookup_hash.py @@ -0,0 +1,73 @@ +import json, os, re, ssl, sys, urllib.error, urllib.request + +DEFAULT_BASE = "https://hashlookup.circl.lu" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "").strip() or DEFAULT_BASE).rstrip("/") + + +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 call(cfg, path, body=None): + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Accept": "application/json", "User-Agent": "riposte-soar"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(_base(cfg) + path, data=data, headers=headers, method="POST" if data is not None else "GET") + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + return json.loads(raw) if raw 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")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +KINDS = {32: "md5", 40: "sha1", 64: "sha256"} + + +def kind_of(h): + h = h.strip().lower() + if not re.fullmatch(r"[0-9a-f]+", h) or len(h) not in KINDS: + raise Exception("not an MD5, SHA-1 or SHA-256 hash: " + h) + return KINDS[len(h)], h + + +def annotate(rec): + # score: 0 for a known file, 100 when CIRCL itself flags the record as + # malicious — what an enrichment rule maps with benign 0 / malicious 100. + rec = dict(rec) + rec["known"] = True + rec["score"] = 100 if rec.get("KnownMalicious") else 0 + return rec + + +def main(cfg, inputs): + kind, h = kind_of(str(inputs.get("hash", ""))) + return annotate(call(cfg, "/lookup/%s/%s" % (kind, h))) + + +_run(main) diff --git a/integrations/circl-hashlookup/scripts/test_connection.py b/integrations/circl-hashlookup/scripts/test_connection.py new file mode 100644 index 0000000..b83a64a --- /dev/null +++ b/integrations/circl-hashlookup/scripts/test_connection.py @@ -0,0 +1,73 @@ +import json, os, re, ssl, sys, urllib.error, urllib.request + +DEFAULT_BASE = "https://hashlookup.circl.lu" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "").strip() or DEFAULT_BASE).rstrip("/") + + +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 call(cfg, path, body=None): + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Accept": "application/json", "User-Agent": "riposte-soar"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(_base(cfg) + path, data=data, headers=headers, method="POST" if data is not None else "GET") + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + return json.loads(raw) if raw 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")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +KINDS = {32: "md5", 40: "sha1", 64: "sha256"} + + +def kind_of(h): + h = h.strip().lower() + if not re.fullmatch(r"[0-9a-f]+", h) or len(h) not in KINDS: + raise Exception("not an MD5, SHA-1 or SHA-256 hash: " + h) + return KINDS[len(h)], h + + +def annotate(rec): + # score: 0 for a known file, 100 when CIRCL itself flags the record as + # malicious — what an enrichment rule maps with benign 0 / malicious 100. + rec = dict(rec) + rec["known"] = True + rec["score"] = 100 if rec.get("KnownMalicious") else 0 + return rec + + +def main(cfg, inputs): + info = call(cfg, "/info") + return {"ok": True, "info": info} + + +_run(main) diff --git a/integrations/crtsh/manifest.yaml b/integrations/crtsh/manifest.yaml new file mode 100644 index 0000000..e473e25 --- /dev/null +++ b/integrations/crtsh/manifest.yaml @@ -0,0 +1,45 @@ +id: crtsh +name: crt.sh (Certificate Transparency) +version: 1.0.0 +description: "crt.sh by Sectigo — searches the public Certificate Transparency logs, where every publicly trusted TLS certificate is recorded. Lists a domain's sub-domains and who issued their certificates, shows when a host first got a certificate (a phishing site usually gets one the day it goes live), and finds look-alike domains that start with a client's brand. Free, no API key; stdlib-only. crt.sh is a shared public service: it is slow on big domains and a starts_with search on a short word can time out — keep searches narrow." +changelog: "1.0.0 — Initial release: certificate search (sub-domains, exact name, starts with), test connection." +category: enrichment + +config_schema: + properties: + base_url: + type: string + description: "crt.sh server (default https://crt.sh)" + default: https://crt.sh + insecure: + type: boolean + description: "Trust any TLS certificate (not secure)" + default: false + required: [] + +commands: + - id: search_certificates + name: crtsh-search-certificates + description: "Certificates logged for a domain. Returns `names` (each DNS name seen, with the first and last time a certificate covered it, newest first, at most 1000 — `name_count` counts them all), `issuers` (certificate count per CA) and `certificates` (the most recent ones, with a crt.sh link). mode: subdomains (default — the domain and every name under it), exact (that name only), starts_with (every logged name beginning with the text, any TLD: look-alikes of a brand such as brand-login.com or brand.co; crt.sh no longer searches for text in the middle of a name, so my-brand.com is not found)." + risk: read + inputs_schema: + properties: + domain: { type: string, description: "Domain, or the brand to look for with mode=starts_with" } + mode: + type: string + description: "subdomains | exact | starts_with (default subdomains)" + enum: [subdomains, exact, starts_with] + exclude_expired: { type: boolean, description: "Ignore expired certificates (default true)" } + since_days: { type: number, description: "Only certificates issued in the last N days (0 = no limit, default 0)" } + limit: { type: number, description: "How many certificates to return in `certificates` (default 50, max 500); `names` and `issuers` always count them all" } + required: [domain] + outputs_schema: { properties: {} } + + - id: test_connection + name: crtsh-test-connection + description: "Run a tiny search against crt.sh (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/crtsh/scripts/search_certificates.py b/integrations/crtsh/scripts/search_certificates.py new file mode 100644 index 0000000..a778a80 --- /dev/null +++ b/integrations/crtsh/scripts/search_certificates.py @@ -0,0 +1,133 @@ +import datetime, json, os, ssl, sys, time, urllib.error, urllib.parse, urllib.request + +DEFAULT_BASE = "https://crt.sh" + + +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 search(cfg, q, exclude_expired): + base = (str(cfg.get("base_url") or "").strip() or DEFAULT_BASE).rstrip("/") + params = {"q": q, "output": "json"} + if exclude_expired: + params["exclude"] = "expired" + req = urllib.request.Request(base + "/?" + urllib.parse.urlencode(params), + headers={"Accept": "application/json", "User-Agent": "riposte-soar"}) + # crt.sh's database is shared by the whole internet and its front answers + # 502/503/504 when a query queues too long; the same query usually goes + # through a few seconds later. + for attempt in range(4): + try: + with urllib.request.urlopen(req, timeout=120, context=_ctx(cfg)) as r: + raw = r.read() + return json.loads(raw) if raw.strip() else [] + except urllib.error.HTTPError as e: + if e.code not in (502, 503, 504) or attempt == 3: + raise + time.sleep(3 * (attempt + 1)) + + +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) + + +def truthy(v, default): + if v is None or v == "": + return default + return v not in (False, "false", "False", 0, "0") + + +def main(cfg, inputs): + domain = str(inputs.get("domain", "")).strip().lower().rstrip(".") + if not domain: + raise Exception("domain is required") + mode = str(inputs.get("mode") or "subdomains").strip().lower() + # crt.sh no longer runs a search that starts with a wildcard (%brand% comes + # back empty), so a look-alike hunt is a prefix search: brand% finds + # brand.co, brand-login.com and brandpay.ch, not my-brand.com. + q = {"subdomains": "%." + domain, "exact": domain, "starts_with": domain + "%"}.get(mode) + if q is None: + raise Exception("mode must be subdomains, exact or starts_with") + rows = search(cfg, q, truthy(inputs.get("exclude_expired"), True)) + if mode == "subdomains": + # %.domain leaves the apex out; it is part of what "the domain" means. + rows += [r for r in search(cfg, domain, truthy(inputs.get("exclude_expired"), True)) if r.get("id") not in {x.get("id") for x in rows}] + + since = int(float(inputs.get("since_days") or 0)) + if since > 0: + floor = (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=since)).strftime("%Y-%m-%dT%H:%M:%S") + rows = [r for r in rows if str(r.get("not_before", "")) >= floor] + + # crt.sh lists a certificate once per log entry (precertificate and final): + # the same id twice, which would double every count below. + seen, certs = set(), [] + for r in rows: + if r.get("id") in seen: + continue + seen.add(r.get("id")) + certs.append(r) + certs.sort(key=lambda r: str(r.get("not_before", "")), reverse=True) + + names, issuers = {}, {} + for r in certs: + nb, na = str(r.get("not_before", "")), str(r.get("not_after", "")) + for n in str(r.get("name_value", "")).lower().split("\n"): + n = n.strip() + if not n: + continue + e = names.setdefault(n, {"name": n, "first_seen": nb, "last_seen": nb, "valid_until": na, "certificates": 0}) + e["certificates"] += 1 + e["first_seen"] = min(e["first_seen"], nb) + e["last_seen"] = max(e["last_seen"], nb) + e["valid_until"] = max(e["valid_until"], na) + issuer = str(r.get("issuer_name", "")) + cn = [p.split("=", 1)[1] for p in issuer.split(", ") if p.startswith("CN=")] + o = [p.split("=", 1)[1] for p in issuer.split(", ") if p.startswith("O=")] + label = " / ".join(x for x in (o[:1] + cn[:1]) if x) or issuer + issuers[label] = issuers.get(label, 0) + 1 + + limit = max(1, min(500, int(float(inputs.get("limit") or 50)))) + base = (str(cfg.get("base_url") or "").strip() or DEFAULT_BASE).rstrip("/") + return { + "query": q, + "certificate_count": len(certs), + "name_count": len(names), + # Newest first and capped: a big domain logs thousands of names, and the + # recent ones are what an analyst is looking for. name_count counts all. + "names": sorted(names.values(), key=lambda e: e["first_seen"], reverse=True)[:1000], + "issuers": [{"issuer": k, "certificates": v} for k, v in sorted(issuers.items(), key=lambda kv: -kv[1])], + "certificates": [{ + "id": r.get("id"), + "common_name": r.get("common_name", ""), + "names": [n for n in str(r.get("name_value", "")).lower().split("\n") if n], + "issuer": r.get("issuer_name", ""), + "not_before": r.get("not_before", ""), + "not_after": r.get("not_after", ""), + "serial_number": r.get("serial_number", ""), + "url": base + "/?id=" + str(r.get("id")), + } for r in certs[:limit]], + } + + +_run(main) diff --git a/integrations/crtsh/scripts/test_connection.py b/integrations/crtsh/scripts/test_connection.py new file mode 100644 index 0000000..2a82803 --- /dev/null +++ b/integrations/crtsh/scripts/test_connection.py @@ -0,0 +1,60 @@ +import datetime, json, os, ssl, sys, time, urllib.error, urllib.parse, urllib.request + +DEFAULT_BASE = "https://crt.sh" + + +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 search(cfg, q, exclude_expired): + base = (str(cfg.get("base_url") or "").strip() or DEFAULT_BASE).rstrip("/") + params = {"q": q, "output": "json"} + if exclude_expired: + params["exclude"] = "expired" + req = urllib.request.Request(base + "/?" + urllib.parse.urlencode(params), + headers={"Accept": "application/json", "User-Agent": "riposte-soar"}) + # crt.sh's database is shared by the whole internet and its front answers + # 502/503/504 when a query queues too long; the same query usually goes + # through a few seconds later. + for attempt in range(4): + try: + with urllib.request.urlopen(req, timeout=120, context=_ctx(cfg)) as r: + raw = r.read() + return json.loads(raw) if raw.strip() else [] + except urllib.error.HTTPError as e: + if e.code not in (502, 503, 504) or attempt == 3: + raise + time.sleep(3 * (attempt + 1)) + + +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) + + +def main(cfg, inputs): + rows = search(cfg, "crt.sh", True) + return {"ok": True, "certificates": len(rows)} + + +_run(main) diff --git a/integrations/rdap/manifest.yaml b/integrations/rdap/manifest.yaml new file mode 100644 index 0000000..efe232d --- /dev/null +++ b/integrations/rdap/manifest.yaml @@ -0,0 +1,65 @@ +id: rdap +name: RDAP (WHOIS) +version: 1.0.0 +description: "RDAP, the IETF successor to WHOIS — who registered a domain, an IP block or an AS number, and when. Answers the first phishing question (how old is this domain?) with `age_days`, plus registrar, status, name servers and the abuse contact. Queries the authoritative registry found through the IANA bootstrap files; free, no API key; stdlib-only. Personal data is usually redacted by registries (GDPR) — expect the registrant to be empty for most domains." +changelog: "1.0.0 — Initial release: domain, IP and ASN lookups through the IANA bootstrap, test connection." +category: enrichment + +config_schema: + properties: + bootstrap_url: + type: string + description: "Where the IANA bootstrap files live (default https://data.iana.org/rdap)" + default: https://data.iana.org/rdap + fallback_url: + type: string + description: "RDAP server asked when the bootstrap has no entry for a TLD (default https://rdap.org, a public redirector). Leave empty to fail instead." + default: https://rdap.org + insecure: + type: boolean + description: "Trust any TLS certificate (not secure)" + default: false + required: [] + +commands: + - id: domain_lookup + name: rdap-domain-lookup + description: "Registration data for a domain. A host name (www.mail.example.com) is walked up to the registered domain. Returns a summary — registrar, created, updated, expires, age_days, status, nameservers, abuse_email, registrant (when not redacted), dnssec — and the registry's `rdap` record. An unregistered domain returns HTTP 404." + risk: read + inputs_schema: + properties: + domain: { type: string, description: "Domain or host name" } + include_raw: { type: boolean, description: "Include the registry's full RDAP record under `rdap` (default true)" } + required: [domain] + outputs_schema: { properties: {} } + + - id: ip_lookup + name: rdap-ip-lookup + description: "The network an IPv4 / IPv6 address belongs to, from its regional registry (RIPE, ARIN, APNIC, LACNIC, AFRINIC): network name, CIDR, country, holder organisation, abuse contact, allocation date." + risk: read + inputs_schema: + properties: + ip: { type: string, description: "IPv4 or IPv6 address" } + include_raw: { type: boolean, description: "Include the registry's full RDAP record under `rdap` (default true)" } + required: [ip] + outputs_schema: { properties: {} } + + - id: asn_lookup + name: rdap-asn-lookup + description: "The holder of an autonomous system number: name, organisation, country, abuse contact, registration date." + risk: read + inputs_schema: + properties: + asn: { type: string, description: "AS number (13335 or AS13335)" } + include_raw: { type: boolean, description: "Include the registry's full RDAP record under `rdap` (default true)" } + required: [asn] + outputs_schema: { properties: {} } + + - id: test_connection + name: rdap-test-connection + description: "Fetch the IANA domain bootstrap file (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/rdap/scripts/asn_lookup.py b/integrations/rdap/scripts/asn_lookup.py new file mode 100644 index 0000000..6232ef0 --- /dev/null +++ b/integrations/rdap/scripts/asn_lookup.py @@ -0,0 +1,190 @@ +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("asn", "")).strip().upper() + if raw.startswith("AS"): + raw = raw[2:] + if not raw.isdigit(): + raise Exception("not an AS number: " + str(inputs.get("asn", ""))) + n = int(raw) + server = "" + for ranges, urls in bootstrap(cfg, "asn"): + for r in ranges: + lo, _, hi = r.partition("-") + if int(lo) <= n <= int(hi or lo): + server = server_url(urls) + break + if server: + break + server = server or fallback(cfg) + if not server: + raise Exception("no RDAP server for this AS number and no fallback_url set") + rec = get(cfg, server + "/autnum/" + str(n)) + found = entities_by_role(rec) + out = {"asn": n} + out.update(common(rec, server)) + out["name"] = rec.get("name", "") + out["country"] = rec.get("country", "") + out["holder"] = first(found, "registrant", "org") or first(found, "registrant", "fn") + out["registrant"] = out["holder"] + if want_raw(inputs): + out["rdap"] = strip(rec) + return out + + +_run(main) diff --git a/integrations/rdap/scripts/domain_lookup.py b/integrations/rdap/scripts/domain_lookup.py new file mode 100644 index 0000000..3d91c08 --- /dev/null +++ b/integrations/rdap/scripts/domain_lookup.py @@ -0,0 +1,213 @@ +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) diff --git a/integrations/rdap/scripts/ip_lookup.py b/integrations/rdap/scripts/ip_lookup.py new file mode 100644 index 0000000..c57f46a --- /dev/null +++ b/integrations/rdap/scripts/ip_lookup.py @@ -0,0 +1,185 @@ +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) diff --git a/integrations/rdap/scripts/test_connection.py b/integrations/rdap/scripts/test_connection.py new file mode 100644 index 0000000..0bfa977 --- /dev/null +++ b/integrations/rdap/scripts/test_connection.py @@ -0,0 +1,163 @@ +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): + services = bootstrap(cfg, "dns") + return {"ok": True, "tlds_served": sum(len(t) for t, _ in services)} + + +_run(main)