riposte-marketplace/integrations/crtsh/scripts/search_certificates.py

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)