f3nris
856dd2321d
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
246 lines
10 KiB
Python
246 lines
10 KiB
Python
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
|
|
|
|
|
|
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"}
|
|
if 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 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=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):
|
|
raise
|
|
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):
|
|
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")
|
|
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.
|
|
# 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:
|
|
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("/")
|
|
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
|
|
# 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": link(r),
|
|
} for r in certs[:limit]],
|
|
}
|
|
|
|
|
|
_run(main)
|