docs(ioc-sources): suggest summary lines for Riposte context enrichment rules

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AmWT9wtWHGtZpMS64Hp4q5

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. Add the summary `Known file ${FileName} (${source})` (Riposte 0.299+) to see which file it is."
+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: {} }

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)

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)

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)

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. As a context enrichment rule (Riposte 0.299+, no score field) on domains: search_certificates with summary `${name_count} names, newest certificate ${certificates.0.not_before}`."
+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: {} }

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)

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)

integrations/harfanglab/mappers/get_alerts.yaml

@@ -3,7 +3,9 @@ description: "Maps a HarfangLab EDR alert (/api/data/alert/alert/Alert/, results
 field_mappings:
   title: "rule_name"
   severity: "level = 'critical' ? 5 : (level = 'high' ? 4 : (level = 'medium' ? 3 : 2))"
-  description: "alert_type"
+  # msg is the rule's own explanation of what it detects; the alert type
+  # (sigma, yara, ioc…) stands in only on the alerts that carry none.
+  description: "msg ? msg : alert_type"
 # results_path = results; source_path is JSONata over ONE alert object.
 # Paths absent from a given alert are skipped at ingestion, so extra entries are safe.
 ocsf:

integrations/harfanglab/mappers/get_security_events.yaml

@@ -3,7 +3,9 @@ description: "Maps a HarfangLab EDR security event/alert (/api/data/alert/alert/
 field_mappings:
   title: "rule_name"
   severity: "level = 'critical' ? 5 : (level = 'high' ? 4 : (level = 'medium' ? 3 : 2))"
-  description: "alert_type"
+  # msg is the rule's own explanation of what it detects; the alert type
+  # (sigma, yara, ioc…) stands in only on the alerts that carry none.
+  description: "msg ? msg : alert_type"
 # results_path = results; source_path is JSONata over ONE alert object.
 # Paths absent from a given alert are skipped at ingestion, so extra entries are safe.
 ocsf:

integrations/harfanglab/scripts/get_alerts.py

@@ -53,7 +53,11 @@ def main():
         "alert_time__gte": inputs.get("from_date"),
         "alert_time__lte": inputs.get("to_date"),
     }
-    print(json.dumps(request("GET", base + "/api/data/alert/alert/Alert/" + qs(params), headers)))
+    out = request("GET", base + "/api/data/alert/alert/Alert/" + qs(params), headers)
+    # The list endpoints echo the catalogue of filterable fields (~50 KB) on
+    # every call; it says nothing about the results, so it is not passed on.
+    out.pop("filters", None)
+    print(json.dumps(out))
 
 
 try:

integrations/harfanglab/scripts/get_security_events.py

@@ -40,7 +40,11 @@ def main():
         "alert_time__gte": inputs.get("created_after"),
     }
     url = base + "/api/data/alert/alert/Alert/" + qs(params)
-    print(json.dumps(request("GET", url, headers)))
+    out = request("GET", url, headers)
+    # The list endpoints echo the catalogue of filterable fields (~50 KB) on
+    # every call; it says nothing about the results, so it is not passed on.
+    out.pop("filters", None)
+    print(json.dumps(out))
 
 
 try:

integrations/harfanglab/scripts/get_threats.py

@@ -39,7 +39,11 @@ def main():
         "creation_date__gte": inputs.get("created_after"),
     }
     url = base + "/api/data/alert/alert/Threat/" + qs(params)
-    print(json.dumps(request("GET", url, headers)))
+    out = request("GET", url, headers)
+    # The list endpoints echo the catalogue of filterable fields (~50 KB) on
+    # every call; it says nothing about the results, so it is not passed on.
+    out.pop("filters", None)
+    print(json.dumps(out))
 
 
 try:

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. As a context enrichment rule (Riposte 0.299+, no score field): domain_lookup with summary `Registered ${age_days} days ago · ${registrar}`; ip_lookup with `${network_name} · ${holder} · ${country}`."
+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: {} }

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)

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)

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)

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)