"""Batch-3 fixture tests (no network). Run: python tests/test_batch3_patterns.py Batch-3 scope: AZURE_STORAGE — connection-string family in canon PATTERNS (deliverable: AccountName+AccountKey self-contained; probe = SharedKey-signed list). HTTPS_AUTH anchoring — diff-probe of 68 targets found 0 real creds; reject placeholder creds ($VAR, admin:admin, echo pairs), doc hosts (localhost/RFC1918/example.*) at parse time, not probe time. PRIVATE_KEY structural validation — GCP-SA lesson (broken PEM fragments, ValueError): base64 body must decode; OPENSSH magic or ASN.1 SEQUENCE. Stdlib only (no openssh_key dep — CVE-2022-31124 in <0.0.6). Coupling gate: AZURE_STORAGE must resolve in eternal_v10 allowlist, else UNKNOWN -> DEAD terminal (KH_AWS graveyard lesson). Static parse only. """ import base64 import os import random import sys HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(os.path.dirname(HERE), "src")) sys.path.insert(0, os.path.dirname(HERE)) FAILS = [] def check(name, cond, detail=""): if cond: print(f"ok {name}") else: print(f"FAIL {name} {detail}") FAILS.append(name) import key_patterns as K # noqa: E402 CANON = dict(K.PATTERNS) _r = random.Random(20261006) _B64 = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" def _b64(n): return "".join(_r.choice(_B64) for _ in range(n)) # ---- AZURE_STORAGE fixtures ---- _AZB64 = _b64(86) + "==" # 86 b64 chars + padding = 64-byte key AZ_POS = ("DefaultEndpointsProtocol=https;AccountName=acmefarm01;" "AccountKey=" + _AZB64 + ";EndpointSuffix=core.windows.net") AZ_NEG1 = ("DefaultEndpointsProtocol=https;AccountName=acmefarm01;" "AccountKey=" + _b64(85) + "==;EndpointSuffix=core.windows.net") # 85 short AZ_NEG2 = "AccountName=acmefarm01;AccountKey=" + _AZB64 # no protocol anchor AZ_NEG3 = ("DefaultEndpointsProtocol=http;AccountName=acmefarm01;" "AccountKey=" + _AZB64 + ";EndpointSuffix=core.windows.net") # http not https # ---- HTTPS_AUTH anchoring fixtures ---- HA_POS = "https://artifactory_reader:Kj8xQ2mZvR4tW9yB@acme.jfrog.io/artifactory/api" HA_JUNK_ECHO = "https://admin:admin@files.acme-cdn.io/dist.zip" HA_JUNK_PH = "https://user:password@nexus.build.corp/repo" HA_JUNK_VAR = "https://${REG_USER}:${REG_TOKEN}@registry-1.docker.io/v2/" HA_JUNK_LOCAL = "https://deploy:Kj8xQ2mZvR4tW9yB@localhost:8443/admin" HA_JUNK_RFC1918 = "https://ci:Kj8xQ2mZvR4tW9yB@192.168.1.50/pkg" HA_JUNK_EXAMPLE = "https://ci:Kj8xQ2mZvR4tW9yB@app.example.org/api" # ---- PRIVATE_KEY structural fixtures ---- def _pem(label, body_bytes): b64 = base64.b64encode(body_bytes).decode() lines = [b64[i:i + 64] for i in range(0, len(b64), 64)] return (f"-----BEGIN {label} PRIVATE KEY-----\n" + "\n".join(lines) + f"\n-----END {label} PRIVATE KEY-----") PK_POS_OPENSSH = _pem("OPENSSH", b"openssh-key-v1\x00" + bytes(range(1, 40))) PK_POS_RSA = _pem("RSA", bytes([0x30, 0x82, 0x01, 0x22]) + bytes(range(2, 44))) PK_NEG_NOMAGIC = _pem("OPENSSH", b"garbage-not-a-key-at-all" + bytes(range(9))) PK_NEG_ENCRYPTED = _pem("ENCRYPTED", bytes([0x30, 0x82, 0x01, 0x22]) + bytes(range(2, 44))) PK_NEG_B64 = ("-----BEGIN RSA PRIVATE KEY-----\n" + "!" * 60 + "\n-----END RSA PRIVATE KEY-----") # ================= pattern-level ================= rx_az = CANON.get("AZURE_STORAGE") check("AZURE_STORAGE:in-canon", rx_az is not None, "missing from PATTERNS") if rx_az: check("AZURE_STORAGE:pos", rx_az.search(AZ_POS) is not None) check("AZURE_STORAGE:neg1-short-key", rx_az.search(AZ_NEG1) is None) check("AZURE_STORAGE:neg2-no-protocol", rx_az.search(AZ_NEG2) is None) check("AZURE_STORAGE:neg3-http", rx_az.search(AZ_NEG3) is None) owners = [n for n, r in CANON.items() if r.search(AZ_POS)] check("AZURE_STORAGE:shape-owned-once", owners == ["AZURE_STORAGE"], f"owners={owners}") m = rx_az.search(AZ_POS) val = m.group(0) check("AZURE_STORAGE:value-self-contained", "AccountName=" in val and "AccountKey=" in val, f"val={val[:60]}") # ================= scan_text end-to-end ================= got = {p for p, _ in K.scan_text(AZ_POS)} check("scan_text:AZURE_STORAGE", "AZURE_STORAGE" in got, f"provs={sorted(got)}") # HTTPS_AUTH: pos survives, junk shapes rejected hp = [v for p, v in K.scan_text(HA_POS) if p == "HTTPS_AUTH"] check("https_auth:pos-emitted", bool(hp), f"scan={[p for p,_ in K.scan_text(HA_POS)]}") for nm, fix in (("echo", HA_JUNK_ECHO), ("placeholder", HA_JUNK_PH), ("var-template", HA_JUNK_VAR), ("localhost", HA_JUNK_LOCAL), ("rfc1918", HA_JUNK_RFC1918), ("example-host", HA_JUNK_EXAMPLE)): hj = [v for p, v in K.scan_text(fix) if p == "HTTPS_AUTH"] check(f"https_auth:junk-{nm}", not hj, f"emitted {hj[:1]}") # PRIVATE_KEY: structurally valid survive, broken rejected for nm, fix in (("openssh", PK_POS_OPENSSH), ("rsa", PK_POS_RSA)): pk = [v for p, v in K.scan_text(fix) if p == "PRIVATE_KEY"] check(f"private_key:pos-{nm}", bool(pk), f"scan={[p for p,_ in K.scan_text(fix)]}") for nm, fix in (("no-magic", PK_NEG_NOMAGIC), ("bad-b64", PK_NEG_B64), ("encrypted-label", PK_NEG_ENCRYPTED)): pk = [v for p, v in K.scan_text(fix) if p == "PRIVATE_KEY"] check(f"private_key:neg-{nm}", not pk, f"emitted len={len(pk)}") # ================= coupling gate ================= import ast as _ast import re as _re _allow, _provs = None, set() for _p in (os.path.join(os.path.dirname(HERE), "src", "eternal_v10.py"), os.path.join(os.path.dirname(HERE), "eternal_v10.py")): if os.path.exists(_p): _src = open(_p, encoding="utf-8").read() _m = _re.search(r"if prov_name in \[([^\]]+)\]", _src) if _m: _allow = {s.strip().strip("'\"") for s in _m.group(1).split(",")} try: for _node in _ast.walk(_ast.parse(_src)): if isinstance(_node, (_ast.Assign, _ast.AnnAssign)): _tgt = _node.targets if isinstance(_node, _ast.Assign) else [_node.target] for _t in _tgt: if (isinstance(_t, _ast.Name) and _t.id == "PROVIDERS" and isinstance(_node.value, _ast.Dict)): for _k in _node.value.keys: if isinstance(_k, _ast.Constant) and isinstance(_k.value, str): _provs.add(_k.value) except SyntaxError: pass break if _allow is None and not _provs: check("coupling:allowlist-found", False, "eternal_v10 allowlist not parsed (no PROVIDERS)") else: for _name in ("AZURE_STORAGE", "PRIVATE_KEY"): _resolved = (_allow and _name in _allow) or _name in _provs check(f"coupling:{_name}-resolvable", _resolved, f"{_name} absent -> UNKNOWN->DEAD destroys finds") print() if FAILS: print(f"BATCH3 FAILURES: {len(FAILS)}: {FAILS}") sys.exit(1) print("BATCH3 ALL GREEN")