О доверии к отечественным сертификатам и о проверке российских CT логов

#!/usr/bin/env python3import sysimport jsonimport base64import hashlibimport structimport requestsfrom asn1crypto import x509, pem# ============================================================# Yandex CT log list# Static snapshot:#   version: 1.229#   log_list_timestamp: 2026-08-12T22:00:21Z# ============================================================REMOTE_LOG_LIST_URL = \    "https://browser-resources.s3.yandex.net/ctlog/ctlog.json"STATIC_LIST_VERSION = "1.229"STATIC_LIST_TIMESTAMP = "2026-08-12T22:00:21Z"LOGS = [    {        "operator": "Yandex",        "description": "Yandex Agate-2026 log",        "url": "https://ct-agate.yandex.net/2026/",        "log_id": "Vh7HRN8a9wLmJoOt38SVz3WjGwxs6KDH6gEDMJC3SIU=",        "start": "2026-01-01T00:00:00Z",        "end": "2027-01-01T00:00:00Z",    },    {        "operator": "Yandex",        "description": "Yandex Agate-2027 log",        "url": "https://ct-agate.yandex.net/2027/",        "log_id": "GukWh2bX94qcsIb4xJw8Xc64yZMAqa9L/7CbB0+BLBc=",        "start": "2027-01-01T00:00:00Z",        "end": "2028-01-01T00:00:00Z",    },    {        "operator": "VK LLC",        "description": "VK 'NCA2026' Log",        "url": "https://ctlog2026.mail.ru/nca2026/",        "log_id": "eNkSvkIKwAIcul2iVDKRU9tIywYr/EMBJB4Qq8cWnmc=",        "start": "2026-01-01T00:00:00Z",        "end": "2027-01-01T00:00:00Z",    },    {        "operator": "VK LLC",        "description": "VK 'NCA2027' Log",        "url": "https://ctlog2027.mail.ru/nca2027/",        "log_id": "1LN/8nNPJvHlUHXlhVtwT95xu30Fx1fT3OsTf2JDrSQ=",        "start": "2027-01-01T00:00:00Z",        "end": "2028-01-01T00:00:00Z",    },    {        "operator":            "The Ministry of Digital Development and Communications",        "description":            "The Ministry of Digital Development and Communications "            "'2026' Log",        "url": "https://26.ctlog.digital.gov.ru/2026/",        "log_id": "ybgRvMAcYl3KJ/TWxEfLX34Ov3LLWZGoMy+Zv5oqPko=",        "start": "2026-01-01T00:00:00Z",        "end": "2027-01-01T00:00:00Z",    },    {        "operator":            "The Ministry of Digital Development and Communications",        "description":            "The Ministry of Digital Development and Communications "            "'2027' Log",        "url": "https://27.ctlog.digital.gov.ru/2027/",        "log_id": "TJig3bv0dZNmDDboZiaHFAn5fnh4TpGeqWr4HFMnXNY=",        "start": "2027-01-01T00:00:00Z",        "end": "2028-01-01T00:00:00Z",    },]SCT_OID = "1.3.6.1.4.1.11129.2.4.2"# ============================================================# Helpers# ============================================================def read_cert(path):    with open(path, "rb") as f:        data = f.read()    if pem.detect(data):        _, _, data = pem.unarmor(data)    return x509.Certificate.load(data)def u16(b, off):    return struct.unpack(">H", b[off:off + 2])[0]def u64(b, off):    return struct.unpack(">Q", b[off:off + 8])[0]def uint24(n):    return n.to_bytes(3, "big")def b64(data):    return base64.b64encode(data).decode("ascii")# ============================================================# Check remote log list for updates# ============================================================def flatten_remote_logs(data):    result = []    for operator in data.get("operators", []):        operator_name = operator.get("name", "?")        for log in operator.get("logs", []):            interval = log.get("temporal_interval", {})            result.append({                "operator": operator_name,                "description": log.get("description", "?"),                "url": log.get("url"),                "log_id": log.get("log_id"),                "start": interval.get("start_inclusive"),                "end": interval.get("end_exclusive"),            })    return resultdef log_identity(log):    return log.get("log_id")def check_log_list_update():    """    Check Yandex ctlog.json.    Failure to download the list is NOT fatal.    The embedded static list will still be used.    """    print("Checking CT log list...")    print("  URL:", REMOTE_LOG_LIST_URL)    try:        r = requests.get(            REMOTE_LOG_LIST_URL,            timeout=5,        )        r.raise_for_status()        data = r.json()    except Exception as e:        print("  Remote list unavailable:", e)        print(            "  Using built-in list version",            STATIC_LIST_VERSION        )        return    remote_version = str(data.get("version", "?"))    remote_timestamp = data.get("log_list_timestamp", "?")    print("  Built-in version:", STATIC_LIST_VERSION)    print("  Remote version:  ", remote_version)    print("  Remote timestamp:", remote_timestamp)    remote_logs = flatten_remote_logs(data)    static_by_id = {        log_identity(x): x        for x in LOGS    }    remote_by_id = {        log_identity(x): x        for x in remote_logs    }    changed = False    # --------------------------------------------------------    # Added logs    # --------------------------------------------------------    added = (        set(remote_by_id)        - set(static_by_id)    )    if added:        changed = True        print()        print("  NEW LOGS in remote list:")        for log_id in sorted(added):            x = remote_by_id[log_id]            print()            print("    +", x["description"])            print("      URL:   ", x["url"])            print("      Log ID:", x["log_id"])    # --------------------------------------------------------    # Removed logs    # --------------------------------------------------------    removed = (        set(static_by_id)        - set(remote_by_id)    )    if removed:        changed = True        print()        print("  LOGS REMOVED from remote list:")        for log_id in sorted(removed):            x = static_by_id[log_id]            print()            print("    -", x["description"])            print("      URL:   ", x["url"])            print("      Log ID:", x["log_id"])    # --------------------------------------------------------    # Changed properties for same LogID    # --------------------------------------------------------    common = (        set(static_by_id)        & set(remote_by_id)    )    fields = [        "url",        "description",        "operator",        "start",        "end",    ]    for log_id in sorted(common):        old = static_by_id[log_id]        new = remote_by_id[log_id]        differences = []        for field in fields:            if old.get(field) != new.get(field):                differences.append(                    (                        field,                        old.get(field),                        new.get(field),                    )                )        if differences:            changed = True            print()            print(                "  CHANGED:",                old["description"]            )            print(                "    Log ID:",                log_id            )            for field, old_value, new_value in differences:                print(                    f"    {field}:"                )                print(                    f"      built-in: {old_value}"                )                print(                    f"      remote:   {new_value}"                )    # --------------------------------------------------------    # Version itself changed    # --------------------------------------------------------    if remote_version != STATIC_LIST_VERSION:        changed = True    print()    if changed:        print(            "  *** WARNING: built-in CT log list "            "may be outdated ***"        )        print(            "  The script will continue using the "            "built-in list."        )    else:        print(            "  CT log list is up to date."        )# ============================================================# SCT parsing# ============================================================def parse_sct(sct):    if len(sct) < 43:        raise RuntimeError("SCT is too short")    p = 0    version = sct[p]    p += 1    log_id = sct[p:p + 32]    p += 32    timestamp = u64(sct, p)    p += 8    ext_len = u16(sct, p)    p += 2    if p + ext_len > len(sct):        raise RuntimeError(            "Malformed SCT extensions"        )    extensions = sct[p:p + ext_len]    p += ext_len    signature = sct[p:]    return {        "version": version,        "log_id": log_id,        "timestamp": timestamp,        "extensions": extensions,        "signature": signature,        "raw": sct,    }def get_scts(cert):    extensions = cert["tbs_certificate"]["extensions"]    raw = None    for ext in extensions:        if ext["extn_id"].dotted == SCT_OID:            raw = ext["extn_value"].native            break    if raw is None:        raise RuntimeError(            "Certificate has no embedded SCT extension"        )    if len(raw) < 2:        raise RuntimeError("Malformed SCT list")    list_len = u16(raw, 0)    if list_len != len(raw) - 2:        raise RuntimeError(            f"Bad SCT list length: "            f"declared={list_len}, "            f"actual={len(raw) - 2}"        )    pos = 2    result = []    while pos < len(raw):        if pos + 2 > len(raw):            raise RuntimeError(                "Malformed SCT list"            )        sct_len = u16(raw, pos)        pos += 2        if pos + sct_len > len(raw):            raise RuntimeError(                "Malformed SCT length"            )        sct = raw[pos:pos + sct_len]        pos += sct_len        result.append(            parse_sct(sct)        )    return result# ============================================================# Match LogID -> CT log# ============================================================def find_log_by_id(log_id_bytes):    wanted = b64(log_id_bytes)    for log in LOGS:        if log["log_id"] == wanted:            return log    return None# ============================================================# Reconstruct precertificate# ============================================================def remove_sct_extension(cert):    tbs_der = cert["tbs_certificate"].dump()    tbs = x509.TbsCertificate.load(tbs_der)    extensions = tbs["extensions"]    found_index = None    for i, ext in enumerate(extensions):        if ext["extn_id"].dotted == SCT_OID:            found_index = i            break    if found_index is None:        raise RuntimeError(            "SCT extension not found"        )    del extensions[found_index]    return tbs.dump()def issuer_key_hash(issuer):    spki = (        issuer["tbs_certificate"]        ["subject_public_key_info"]        .dump()    )    return hashlib.sha256(spki).digest()def make_precert_leaf(cert, issuer, sct):    tbs = remove_sct_extension(cert)    ikh = issuer_key_hash(issuer)    leaf = bytearray()    # Version v1    leaf += b"\x00"    # MerkleLeafType = timestamped_entry    leaf += b"\x00"    # timestamp    leaf += struct.pack(        ">Q",        sct["timestamp"]    )    # LogEntryType = precert_entry    leaf += struct.pack(">H", 1)    # issuer_key_hash    leaf += ikh    # TBSCertificate    leaf += uint24(len(tbs))    leaf += tbs    # CtExtensions    leaf += struct.pack(        ">H",        len(sct["extensions"])    )    leaf += sct["extensions"]    return bytes(leaf)# ============================================================# Merkle hashing# ============================================================def leaf_hash(leaf):    return hashlib.sha256(        b"\x00" + leaf    ).digest()def node_hash(left, right):    return hashlib.sha256(        b"\x01" + left + right    ).digest()def verify_inclusion(    leaf_hash_value,    index,    tree_size,    audit_path,):    if tree_size <= 0:        return None    if index < 0 or index >= tree_size:        return None    fn = index    sn = tree_size - 1    r = leaf_hash_value    for p in audit_path:        if fn == sn or (fn & 1):            r = node_hash(p, r)            while (                fn != 0                and (fn & 1) == 0            ):                fn >>= 1                sn >>= 1        else:            r = node_hash(r, p)        fn >>= 1        sn >>= 1    if sn != 0:        return None    return r# ============================================================# CT HTTP API# ============================================================def api_url(log_url, endpoint):    return (        log_url.rstrip("/")        + "/ct/v1/"        + endpoint    )def get_sth(log_url):    r = requests.get(        api_url(log_url, "get-sth"),        timeout=20,    )    r.raise_for_status()    return r.json()def get_proof(log_url, h, tree_size):    return requests.get(        api_url(            log_url,            "get-proof-by-hash"        ),        params={            "hash": b64(h),            "tree_size": tree_size,        },        timeout=20,    )def get_entry(log_url, index):    r = requests.get(        api_url(            log_url,            "get-entries"        ),        params={            "start": index,            "end": index,        },        timeout=20,    )    r.raise_for_status()    data = r.json()    entries = data.get(        "entries",        []    )    if len(entries) != 1:        raise RuntimeError(            f"Expected 1 entry, "            f"got {len(entries)}"        )    return entries[0]# ============================================================# Check a matched SCT# ============================================================def check_sct(    log,    cert,    issuer,    sct,):    print("  Getting STH...")    sth = get_sth(log["url"])    tree_size = sth["tree_size"]    advertised_root = base64.b64decode(        sth["sha256_root_hash"]    )    print(        "  tree_size:",        tree_size    )    leaf = make_precert_leaf(        cert,        issuer,        sct    )    h = leaf_hash(leaf)    print(        "  Leaf hash:",        b64(h)    )    response = get_proof(        log["url"],        h,        tree_size,    )    print(        "  get-proof-by-hash HTTP:",        response.status_code    )    if response.status_code != 200:        print("  NOT FOUND")        text = response.text.strip()        if text:            print(                "  Response:",                text[:500]            )        return False    proof = response.json()    index = proof["leaf_index"]    path = [        base64.b64decode(x)        for x in proof["audit_path"]    ]    print("  FOUND")    print(        "  leaf_index:",        index    )    print(        "  proof nodes:",        len(path)    )    calculated_root = verify_inclusion(        h,        index,        tree_size,        path,    )    if calculated_root is None:        print(            "  Merkle proof: INVALID"        )        return False    if calculated_root != advertised_root:        print(            "  Merkle root: MISMATCH"        )        print(            "    calculated:",            calculated_root.hex()        )        print(            "    advertised:",            advertised_root.hex()        )        return False    print(        "  Merkle proof: VALID"    )    # --------------------------------------------------------    # Fetch actual CT entry    # --------------------------------------------------------    entry = get_entry(        log["url"],        index    )    server_leaf = base64.b64decode(        entry["leaf_input"]    )    if server_leaf != leaf:        print(            "  leaf_input comparison: DIFFERENT"        )        print(            "    local length: ",            len(leaf)        )        print(            "    server length:",            len(server_leaf)        )        return False    print(        "  leaf_input comparison: IDENTICAL"    )    print()    print(        "  *** INCLUSION FULLY VERIFIED ***"    )    return True# ============================================================# Main# ============================================================def main():    if len(sys.argv) != 3:        print(            f"Usage: {sys.argv[0]} "            "cert.pem issuer.pem",            file=sys.stderr,        )        sys.exit(2)    # --------------------------------------------------------    # Check whether embedded CT log list is current    # --------------------------------------------------------    check_log_list_update()    print()    print("=" * 72)    cert = read_cert(        sys.argv[1]    )    issuer = read_cert(        sys.argv[2]    )    print()    print("Reading SCTs...")    scts = get_scts(cert)    print(        f"Found {len(scts)} "        "embedded SCT(s)"    )    verified = 0    for n, sct in enumerate(scts):        print()        print("=" * 72)        print(f"SCT #{n}")        print("=" * 72)        log_id = b64(            sct["log_id"]        )        print(            "Log ID:",            log_id        )        print(            "Timestamp:",            sct["timestamp"]        )        # ----------------------------------------------------        # Find CT log from LogID        # ----------------------------------------------------        log = find_log_by_id(            sct["log_id"]        )        if log is None:            print()            print(                "CT log: UNKNOWN"            )            print(                "This LogID is not present "                "in the built-in Yandex CT log list."            )            continue        print()        print(            "CT log:",            log["description"]        )        print(            "Operator:",            log["operator"]        )        print(            "URL:",            log["url"]        )        print(            "Temporal interval:",            log["start"],            "..",            log["end"]        )        try:            ok = check_sct(                log,                cert,                issuer,                sct,            )        except Exception as e:            print()            print(                "ERROR:",                type(e).__name__ + ":",                e            )            ok = False        if ok:            verified += 1    # --------------------------------------------------------    # Summary    # --------------------------------------------------------    print()    print("=" * 72)    print("FINAL RESULT")    print("=" * 72)    print(        "Embedded SCTs:",        len(scts)    )    print(        "Fully verified:",        verified    )    if verified == len(scts):        print(            "All SCT log entries were "            "successfully verified."        )        sys.exit(0)    elif verified:        print(            "Some SCT log entries were "            "successfully verified."        )        sys.exit(0)    else:        print(            "No SCT log entry was "            "successfully verified."        )        sys.exit(1)if __name__ == "__main__":    main()