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
85 lines
3.1 KiB
Python
85 lines
3.1 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
|
|
|
|
|
|
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("/")
|
|
# Riposte stops the script at 60 s: two tries at most, Cert Spotter keeps the rest.
|
|
last = ""
|
|
for attempt in range(2):
|
|
try:
|
|
return len(_get(base + "/?" + urllib.parse.urlencode({"q": "crt.sh", "output": "json", "exclude": "expired"}), {}, cfg, 20)), ""
|
|
except urllib.error.HTTPError as e:
|
|
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):
|
|
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):
|
|
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)
|