f3nris/riposte-marketplace
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(crtsh): Cert Spotter fallback and a 60 s time budget (1.1.0)

Browse Source 
crt.sh often answers 502 for minutes at a time. When it is down,
sub-domain and exact searches are now answered by SSLMate Cert Spotter,
which reads the same CT logs; its rows are reshaped into crt.sh's so the
output keeps its shape, plus 'source' and 'complete'. starts_with has no
fallback (Cert Spotter cannot search by prefix).

Riposte kills an integration script at 60 s, and the old retry loop
(4 x 120 s) could never finish. Every call now takes its timeout from a
55 s budget, with 15 s kept for the fallback; only 5xx are retried.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AmWT9wtWHGtZpMS64Hp4q5
This commit is contained in:
Guillaume Bourgeois
2026-10-10 21:12:15 +00:00
parent 94142f4739
commit 856dd2321d
3 changed files with 181 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
integrations/crtsh/manifest.yaml
+16 -4
View File
@@ -1,8 +1,8 @@
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."
version: 1.1.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. When crt.sh is down (it often answers 502), sub-domain and exact searches fall back to SSLMate Cert Spotter, which reads the same logs (unexpired certificates only; set a free API key to lift its 10-requests-an-hour limit). 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.1.0 — Cert Spotter fallback: when crt.sh is down (502/503/504 or no answer), sub-domain and exact searches are answered by SSLMate Cert Spotter, from the same CT logs; the output says which (`source`) and whether it may be partial (`complete`). Every call now fits Riposte's 60 s script limit (crt.sh used to be retried for minutes and killed before it answered). 1.0.0 — Initial release: certificate search (sub-domains, exact name, starts with), test connection."
category: enrichment
config_schema:
@@ -11,6 +11,18 @@ config_schema:
type: string
description: "crt.sh server (default https://crt.sh)"
default: https://crt.sh
certspotter_fallback:
type: boolean
description: "When crt.sh is down, answer sub-domain and exact searches from SSLMate Cert Spotter (same CT logs, unexpired certificates only; no starts_with)"
default: true
certspotter_api_key:
type: string
description: "Cert Spotter API key (optional, free account at sslmate.com). Without one Cert Spotter allows about 10 requests an hour and the fallback reads at most 3 pages; with one, 10 pages"
x-soar-sensitive: true
certspotter_url:
type: string
description: "Cert Spotter API (default https://api.certspotter.com)"
default: https://api.certspotter.com
insecure:
type: boolean
description: "Trust any TLS certificate (not secure)"
@@ -20,7 +32,7 @@ config_schema:
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)."
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), plus `source` (crt.sh, or certspotter when crt.sh was down and the fallback answered) and `complete` (false when the answer may miss certificates — Cert Spotter page cap, or crt.sh too slow for the apex query). 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:
integrations/crtsh/scripts/search_certificates.py
+122 -10
View File
@@ -20,6 +20,29 @@ def _ctx(cfg):
return None
DEFAULT_CERTSPOTTER = "https://api.certspotter.com"
# Riposte stops an integration script after 60 s, whatever it is doing. Every
# network call below takes its timeout from what is left of this budget, so the
# script always gets to print an answer (or a clear error) before it is killed.
BUDGET = 55.0
# Kept back from crt.sh so the Cert Spotter fallback still has time to run.
FALLBACK_RESERVE = 15.0
_start = time.monotonic()
def _left():
return BUDGET - (time.monotonic() - _start)
class Unavailable(Exception):
"""crt.sh did not answer in time or kept answering 5xx."""
def _fallback_on(cfg):
return cfg.get("certspotter_fallback") not in (False, "false", "False", 0, "0")
def search(cfg, q, exclude_expired):
base = (str(cfg.get("base_url") or "").strip() or DEFAULT_BASE).rstrip("/")
params = {"q": q, "output": "json"}
@@ -27,18 +50,80 @@ def search(cfg, q, exclude_expired):
params["exclude"] = "expired"
req = urllib.request.Request(base + "/?" + urllib.parse.urlencode(params),
headers={"Accept": "application/json", "User-Agent": "riposte-soar"})
reserve = FALLBACK_RESERVE if _fallback_on(cfg) else 0.0
# 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):
# 502/503/504 when a query queues too long; the same query often goes
# through a few seconds later, so those are retried, three tries at most.
# A timeout or a refused connection is not: it has already cost its time,
# and the fallback is the better use of what is left.
for attempt in range(3):
timeout = min(45.0, _left() - reserve)
if timeout < 5:
raise Unavailable("crt.sh unavailable (out of time)")
try:
with urllib.request.urlopen(req, timeout=120, context=_ctx(cfg)) as r:
with urllib.request.urlopen(req, timeout=timeout, 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:
if e.code not in (502, 503, 504):
raise
time.sleep(3 * (attempt + 1))
if attempt == 2:
raise Unavailable("crt.sh unavailable (HTTP " + str(e.code) + ")")
except (urllib.error.URLError, TimeoutError, OSError) as e:
raise Unavailable("crt.sh unavailable (" + ("timeout" if "timed out" in str(e) else str(getattr(e, "reason", e))) + ")")
time.sleep(min(3.0 * (attempt + 1), max(0.0, _left() - reserve - 5)))
def certspotter(cfg, domain, include_subdomains):
"""The same Certificate Transparency logs, read through SSLMate's Cert Spotter.
Its rows are reshaped into crt.sh's, so everything after the search is shared.
Cert Spotter only lists certificates that have not expired, and without an
API key allows about 10 requests an hour, so the pages read are capped.
"""
key = str(cfg.get("certspotter_api_key") or "").strip()
base = (str(cfg.get("certspotter_url") or "").strip() or DEFAULT_CERTSPOTTER).rstrip("/")
params = [("domain", domain), ("expand", "dns_names"), ("expand", "issuer")]
if include_subdomains:
params.append(("include_subdomains", "true"))
headers = {"Accept": "application/json", "User-Agent": "riposte-soar"}
if key:
headers["Authorization"] = "Bearer " + key
max_pages = 10 if key else 3
rows, after, complete = [], None, False
for _ in range(max_pages):
timeout = _left() - 2
if timeout < 3:
break
qs = params + ([("after", after)] if after else [])
req = urllib.request.Request(base + "/v1/issuances?" + urllib.parse.urlencode(qs), headers=headers)
try:
with urllib.request.urlopen(req, timeout=timeout, context=_ctx(cfg)) as r:
page = json.loads(r.read() or b"[]")
except urllib.error.HTTPError as e:
if e.code == 429 and not rows:
raise Exception("crt.sh unavailable and Cert Spotter quota reached"
+ ("" if key else " (about 10 requests an hour without an API key — set certspotter_api_key)"))
if rows:
break
raise
for c in page:
rows.append({
"id": "cs-" + str(c.get("id")),
"common_name": (c.get("dns_names") or [""])[0],
"name_value": "\n".join(c.get("dns_names") or []),
"issuer_name": (c.get("issuer") or {}).get("name", ""),
# crt.sh dates carry no zone; keep one format for sorting and comparing.
"not_before": str(c.get("not_before", "")).rstrip("Z"),
"not_after": str(c.get("not_after", "")).rstrip("Z"),
"serial_number": "",
"cert_sha256": c.get("cert_sha256", ""),
})
if not page:
complete = True
break
after = page[-1].get("id")
return rows, complete
def _run(fn):
@@ -69,10 +154,26 @@ def main(cfg, inputs):
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":
exclude_expired = truthy(inputs.get("exclude_expired"), True)
source, complete = "crt.sh", True
try:
rows = search(cfg, q, exclude_expired)
if mode == "subdomains" and _left() > 10:
# %.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}]
# Skipped rather than risked when crt.sh was slow enough to eat the budget.
ids = {x.get("id") for x in rows}
try:
rows += [r for r in search(cfg, domain, exclude_expired) if r.get("id") not in ids]
except Unavailable:
complete = False
except Unavailable:
# Cert Spotter cannot search by prefix, so a look-alike hunt has no fallback.
if not _fallback_on(cfg):
raise
if mode == "starts_with":
raise Exception(str(sys.exc_info()[1]) + "; no fallback for starts_with — Cert Spotter cannot search by prefix")
rows, complete = certspotter(cfg, domain, mode == "subdomains")
source = "certspotter"
since = int(float(inputs.get("since_days") or 0))
if since > 0:
@@ -109,8 +210,19 @@ def main(cfg, inputs):
limit = max(1, min(500, int(float(inputs.get("limit") or 50))))
base = (str(cfg.get("base_url") or "").strip() or DEFAULT_BASE).rstrip("/")
def link(r):
if source == "certspotter":
return base + "/?q=" + str(r.get("cert_sha256", ""))
return base + "/?id=" + str(r.get("id"))
return {
"query": q,
# Which service answered: crt.sh, or Cert Spotter when crt.sh was down.
"source": source,
# False when the answer may miss certificates: Cert Spotter's page cap was
# hit, or crt.sh gave up on the apex query. Cert Spotter never lists
# expired certificates, whatever exclude_expired says.
"complete": complete,
"certificate_count": len(certs),
"name_count": len(names),
# Newest first and capped: a big domain logs thousands of names, and the
@@ -125,7 +237,7 @@ def main(cfg, inputs):
"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")),
"url": link(r),
} for r in certs[:limit]],
}
integrations/crtsh/scripts/test_connection.py
+42 -18
View File
@@ -20,25 +20,40 @@ def _ctx(cfg):
return None
def search(cfg, q, exclude_expired):
def _get(url, headers, cfg, timeout):
req = urllib.request.Request(url, headers=dict({"Accept": "application/json", "User-Agent": "riposte-soar"}, **headers))
with urllib.request.urlopen(req, timeout=timeout, context=_ctx(cfg)) as r:
return json.loads(r.read() or b"[]")
def _crtsh(cfg):
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):
# Riposte stops the script at 60 s: two tries at most, Cert Spotter keeps the rest.
last = ""
for attempt in range(2):
try:
with urllib.request.urlopen(req, timeout=120, context=_ctx(cfg)) as r:
raw = r.read()
return json.loads(raw) if raw.strip() else []
return len(_get(base + "/?" + urllib.parse.urlencode({"q": "crt.sh", "output": "json", "exclude": "expired"}), {}, cfg, 20)), ""
except urllib.error.HTTPError as e:
if e.code not in (502, 503, 504) or attempt == 3:
raise
time.sleep(3 * (attempt + 1))
last = "HTTP " + str(e.code)
if e.code not in (502, 503, 504):
break
except Exception as e:
last = str(getattr(e, "reason", e))
if attempt == 0:
time.sleep(3)
return None, last
def _certspotter(cfg):
key = str(cfg.get("certspotter_api_key") or "").strip()
base = (str(cfg.get("certspotter_url") or "").strip() or "https://api.certspotter.com").rstrip("/")
try:
_get(base + "/v1/issuances?domain=crt.sh", {"Authorization": "Bearer " + key} if key else {}, cfg, 15)
return True, ""
except urllib.error.HTTPError as e:
return False, "HTTP " + str(e.code) + (" (quota reached)" if e.code == 429 else "")
except Exception as e:
return False, str(getattr(e, "reason", e))
def _run(fn):
@@ -53,8 +68,17 @@ def _run(fn):
def main(cfg, inputs):
rows = search(cfg, "crt.sh", True)
return {"ok": True, "certificates": len(rows)}
count, crt_err = _crtsh(cfg)
if count is not None:
return {"ok": True, "crtsh": "ok", "certificates": count}
if cfg.get("certspotter_fallback") in (False, "false", "False", 0, "0"):
raise Exception("crt.sh unavailable (" + crt_err + ") and the Cert Spotter fallback is off")
ok, cs_err = _certspotter(cfg)
if not ok:
raise Exception("crt.sh unavailable (" + crt_err + ") and Cert Spotter unreachable (" + cs_err + ")")
# crt.sh being down is common and the fallback answers: the instance works.
return {"ok": True, "crtsh": "unavailable (" + crt_err + ")", "certspotter": "ok",
"note": "crt.sh is down right now; searches will be answered by Cert Spotter"}
_run(main)