|
| 1 | +"""FULL LOCAL receipt verification — Ed25519, offline, no CodeRifts call. |
| 2 | +
|
| 3 | + pip install coderifts-sdk[verify] |
| 4 | +
|
| 5 | + from coderifts import verify_receipt, keyring_from_document |
| 6 | + v = verify_receipt(token, keyring_from_document(keys_doc)) |
| 7 | + if not v["valid"]: |
| 8 | + raise RuntimeError(f'{v["status"]}: {v.get("reason")}') |
| 9 | +
|
| 10 | +── WHY THE CRYPTO IS AN EXTRA ────────────────────────────────────────────────────────────── |
| 11 | +
|
| 12 | +This package's install profile is ``requests`` and nothing else, which is why it works on hosts |
| 13 | +that cannot build a wheel. Most callers here talk to the API and never verify anything. Making the |
| 14 | +signature path optional keeps that install intact and gives the callers who DO verify a real one. |
| 15 | +
|
| 16 | +``cryptography`` over ``pynacl``, measured: |
| 17 | +
|
| 18 | +* ``coderifts-verifier`` — this ecosystem's dedicated Python verifier — already depends on |
| 19 | + ``cryptography>=41``. Choosing ``pynacl`` here would mean ONE receipt format verified by TWO |
| 20 | + crypto stacks inside one ecosystem, and would make the eventual consolidation (this module |
| 21 | + delegating to that package once it is published) a dependency swap for every user. |
| 22 | +* ``cryptography`` ships wheels for every platform this SDK targets and is already present in most |
| 23 | + environments that have a Python toolchain at all; ``pynacl`` is thinner but adds a second native |
| 24 | + stack next to one that is usually already there. |
| 25 | +* ``pynacl`` is the smaller binding and that is a real advantage. It is not worth splitting the |
| 26 | + ecosystem's crypto for. |
| 27 | +
|
| 28 | +── WHY THIS CODE EXISTS AT ALL, SAID PLAINLY ─────────────────────────────────────────────── |
| 29 | +
|
| 30 | +It is a SECOND Python implementation of a format ``coderifts-verifier`` already verifies. That is |
| 31 | +a cost, not a feature: two verifiers of one format disagreeing is a defect this ecosystem has |
| 32 | +already paid for once. |
| 33 | +
|
| 34 | +MEASURED before writing it: ``coderifts-verifier`` is NOT on PyPI (``pip index versions`` finds no |
| 35 | +distribution), so an extra that merely depended on it could not resolve for anyone. When it is |
| 36 | +published, the honest move is to make ``[verify]`` depend on it and delete this module's signature |
| 37 | +path — the parity suite below is what will make that swap safe. |
| 38 | +
|
| 39 | +── WHAT IS COVERED, AND WHAT IS NOT ──────────────────────────────────────────────────────── |
| 40 | +
|
| 41 | +Covered: the FROZEN signed-bytes rule (RECEIPT_FORMAT.md §3) for v1–v4, unknown/retired/revoked |
| 42 | +key status, expiry with the 30s clock-skew leeway, and the structural refusals. Proven against the |
| 43 | +node core's own 16-vector corpus, not against this file's idea of the format. |
| 44 | +
|
| 45 | +NOT covered, and it is not covered by any local verifier: REVOCATION as of now, and the issuer's |
| 46 | +clock. A key compromised a minute ago still verifies here. |
| 47 | +""" |
| 48 | + |
| 49 | +from __future__ import annotations |
| 50 | + |
| 51 | +import base64 |
| 52 | +import binascii |
| 53 | +import hashlib |
| 54 | +import json |
| 55 | +import time |
| 56 | +from typing import Any, Dict, Mapping, Optional |
| 57 | + |
| 58 | +SIGNED_PREFIX = "crchain.v1" |
| 59 | +MAX_SUPPORTED_V = 4 |
| 60 | +#: The same 30s the node core applies. A receipt that expired within the leeway is still current. |
| 61 | +CLOCK_SKEW_LEEWAY_MS = 30_000 |
| 62 | +#: Statuses this verifier understands. Anything else FAILS CLOSED — an operator who marked a key |
| 63 | +#: with a word this build does not know must not be told the key is fine. |
| 64 | +KNOWN_STATUSES = frozenset({"active", "retired", "revoked", None}) |
| 65 | + |
| 66 | +_INSTALL_HINT = ( |
| 67 | + "coderifts-sdk was installed without the signature-verification extra.\n" |
| 68 | + " pip install 'coderifts-sdk[verify]'\n" |
| 69 | + "This function does NOT fall back to the digest cross-check: a partial check reported as a " |
| 70 | + "verification is worse than no verification, because it is counted as one. If you want the " |
| 71 | + "cross-check, call cross_check_receipt() and read what it says it is." |
| 72 | +) |
| 73 | + |
| 74 | + |
| 75 | +def _ed25519(): |
| 76 | + """The crypto backend, imported at call time so a requests-only install stays importable.""" |
| 77 | + try: |
| 78 | + from cryptography.hazmat.primitives.asymmetric.ed25519 import ( # noqa: PLC0415 |
| 79 | + Ed25519PublicKey, |
| 80 | + ) |
| 81 | + from cryptography.hazmat.primitives.serialization import ( # noqa: PLC0415 |
| 82 | + load_pem_public_key, |
| 83 | + ) |
| 84 | + except ImportError as exc: # pragma: no cover - exercised by the no-extra test |
| 85 | + raise ImportError(_INSTALL_HINT) from exc |
| 86 | + return Ed25519PublicKey, load_pem_public_key |
| 87 | + |
| 88 | + |
| 89 | +def _b64url(segment: str) -> bytes: |
| 90 | + return base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4)) |
| 91 | + |
| 92 | + |
| 93 | +def _fail(status: str, reason: str, payload: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: |
| 94 | + out: Dict[str, Any] = {"valid": False, "status": status, "reason": reason} |
| 95 | + if payload is not None: |
| 96 | + out["payload"] = dict(payload) |
| 97 | + return out |
| 98 | + |
| 99 | + |
| 100 | +def reconstruct_signed_input(body: Mapping[str, Any]) -> str: |
| 101 | + """The FROZEN signed bytes (RECEIPT_FORMAT.md §3), rebuilt from the body's own fields. |
| 102 | +
|
| 103 | + Rebuilt, never read back from the encoded segment: the encoded payload is what an attacker |
| 104 | + hands you, and signing over it would verify that they encoded their own bytes consistently. |
| 105 | + """ |
| 106 | + v = body.get("v") |
| 107 | + parts = [ |
| 108 | + SIGNED_PREFIX, |
| 109 | + str(body.get("kid", "")), |
| 110 | + str(body.get("fp", "")), |
| 111 | + str(body.get("prev", "")), |
| 112 | + str(body.get("caller", "")), |
| 113 | + str(body.get("ts", "")), |
| 114 | + ] |
| 115 | + if isinstance(v, int) and v >= 2: |
| 116 | + parts.append(str(body.get("reg", ""))) |
| 117 | + if isinstance(v, int) and v >= 3: |
| 118 | + parts.append(str(body.get("ir", ""))) |
| 119 | + if isinstance(v, int) and v >= 4: |
| 120 | + parts.append(str(body.get("expires_at", ""))) |
| 121 | + parts.append(str(body.get("bh", ""))) |
| 122 | + return "|".join(parts) |
| 123 | + |
| 124 | + |
| 125 | +def _has_delimiter(body: Mapping[str, Any]) -> bool: |
| 126 | + """A signed field containing `|` would let a v4 body re-split as a v3 one. |
| 127 | +
|
| 128 | + This is the cross-version re-split guard, and it is why the corpus's `downgrade_v4_as_v3` |
| 129 | + vector is refused as INVALID_SIGNATURE/delimiter_in_field rather than as a bad signature. |
| 130 | + """ |
| 131 | + for key in ("kid", "fp", "prev", "caller", "ts", "reg", "ir", "expires_at", "bh"): |
| 132 | + value = body.get(key) |
| 133 | + if isinstance(value, str) and "|" in value: |
| 134 | + return True |
| 135 | + return False |
| 136 | + |
| 137 | + |
| 138 | +def keyring_from_document(doc: Mapping[str, Any]) -> Dict[str, Dict[str, Any]]: |
| 139 | + """kid → entry, from a `.well-known/coderifts-keys.json`-shaped document. |
| 140 | +
|
| 141 | + The keys are PINNED BY THE CALLER and never fetched here. A verifier that downloads the key it |
| 142 | + is about to trust has verified nothing an attacker on the path could not arrange. |
| 143 | + """ |
| 144 | + out: Dict[str, Dict[str, Any]] = {} |
| 145 | + for k in (doc or {}).get("keys", []) or []: |
| 146 | + if not isinstance(k, Mapping): |
| 147 | + continue |
| 148 | + kid, pem = k.get("kid"), k.get("public_key_pem") |
| 149 | + if not isinstance(kid, str) or not isinstance(pem, str): |
| 150 | + continue |
| 151 | + out[kid] = {"public_key_pem": pem, "status": k.get("status", "active")} |
| 152 | + return out |
| 153 | + |
| 154 | + |
| 155 | +def verify_receipt( |
| 156 | + token: str, |
| 157 | + keyring: Mapping[str, Mapping[str, Any]], |
| 158 | + *, |
| 159 | + expected_kid: Optional[str] = None, |
| 160 | + now: Optional[int] = None, |
| 161 | +) -> Dict[str, Any]: |
| 162 | + """Verify a chain receipt LOCALLY. No network, no API key, full Ed25519. |
| 163 | +
|
| 164 | + :param token: the receipt token |
| 165 | + :param keyring: kid → {public_key_pem, status}; see :func:`keyring_from_document` |
| 166 | + :param now: milliseconds since epoch, for expiry. Absent = the host clock. |
| 167 | + :raises ImportError: when installed without the ``[verify]`` extra. It does NOT quietly |
| 168 | + degrade to the digest cross-check — a partial check reported as a verification is worse |
| 169 | + than none, because it is counted as one. |
| 170 | + """ |
| 171 | + Ed25519PublicKey, load_pem_public_key = _ed25519() |
| 172 | + |
| 173 | + if not isinstance(token, str) or not token: |
| 174 | + return _fail("MALFORMED", "malformed_structure") |
| 175 | + segments = token.split(".") |
| 176 | + if len(segments) != 2 or not all(segments): |
| 177 | + return _fail("MALFORMED", "malformed_structure") |
| 178 | + try: |
| 179 | + body = json.loads(_b64url(segments[0])) |
| 180 | + except (ValueError, binascii.Error): |
| 181 | + return _fail("MALFORMED", "bad_json") |
| 182 | + if not isinstance(body, dict): |
| 183 | + return _fail("MALFORMED", "bad_json") |
| 184 | + |
| 185 | + v = body.get("v") |
| 186 | + if isinstance(v, int) and v > MAX_SUPPORTED_V: |
| 187 | + # NO `reason`, matching the node core exactly. MEASURED against its 16-vector corpus: it |
| 188 | + # emits `{valid, status}` here and on VERIFIED_EXPIRED, and a Python that helpfully added |
| 189 | + # one would make a consumer branching on `reason` see a value from one implementation and |
| 190 | + # nothing from the other. That is the divergence class two verifiers of one format are for. |
| 191 | + return {"valid": False, "status": "UNSUPPORTED_VERSION", "payload": dict(body)} |
| 192 | + if _has_delimiter(body): |
| 193 | + return _fail("INVALID_SIGNATURE", "delimiter_in_field", body) |
| 194 | + |
| 195 | + kid = body.get("kid") |
| 196 | + if expected_kid is not None and kid != expected_kid: |
| 197 | + return _fail("UNKNOWN_KEY", "unknown_kid", body) |
| 198 | + entry = (keyring or {}).get(kid) |
| 199 | + if not entry: |
| 200 | + return _fail("UNKNOWN_KEY", "unknown_kid", body) |
| 201 | + status = entry.get("status", "active") |
| 202 | + if status not in KNOWN_STATUSES: |
| 203 | + return _fail("UNKNOWN_KEY_STATUS", "unknown_key_status", body) |
| 204 | + if status == "revoked": |
| 205 | + return _fail("UNKNOWN_KEY", "revoked_kid", body) |
| 206 | + |
| 207 | + try: |
| 208 | + key = load_pem_public_key(str(entry["public_key_pem"]).encode("utf-8")) |
| 209 | + except Exception: # noqa: BLE001 - a key we cannot load is a key we do not trust |
| 210 | + return _fail("UNKNOWN_KEY", "unloadable_key", body) |
| 211 | + if not isinstance(key, Ed25519PublicKey): |
| 212 | + return _fail("UNKNOWN_KEY", "not_ed25519", body) |
| 213 | + |
| 214 | + message = reconstruct_signed_input(body).encode("utf-8") |
| 215 | + try: |
| 216 | + key.verify(_b64url(segments[1]), message) |
| 217 | + except (binascii.Error, ValueError): |
| 218 | + return _fail("INVALID_SIGNATURE", "signature_mismatch", body) |
| 219 | + except Exception: # noqa: BLE001 - cryptography raises InvalidSignature |
| 220 | + return _fail("INVALID_SIGNATURE", "signature_mismatch", body) |
| 221 | + |
| 222 | + if status == "retired": |
| 223 | + # The signature is authentic and the key is no longer issuing. Historical, not current. |
| 224 | + return {"valid": False, "status": "RETIRED_KEY_VALID_AT_ISSUE", |
| 225 | + "reason": "retired_kid", "payload": dict(body)} |
| 226 | + |
| 227 | + expires_at = body.get("expires_at") |
| 228 | + if isinstance(expires_at, str) and expires_at: |
| 229 | + try: |
| 230 | + from datetime import datetime # noqa: PLC0415 |
| 231 | + exp_ms = int(datetime.fromisoformat( |
| 232 | + expires_at.replace("Z", "+00:00")).timestamp() * 1000) |
| 233 | + except ValueError: |
| 234 | + return _fail("MALFORMED", "bad_expires_at", body) |
| 235 | + current = int(time.time() * 1000) if now is None else int(now) |
| 236 | + if exp_ms + CLOCK_SKEW_LEEWAY_MS < current: |
| 237 | + # No `reason` — see UNSUPPORTED_VERSION above. |
| 238 | + return {"valid": False, "status": "VERIFIED_EXPIRED", "payload": dict(body)} |
| 239 | + |
| 240 | + return {"valid": True, "status": "VERIFIED_CURRENT", "payload": dict(body), |
| 241 | + "does_not_prove": [ |
| 242 | + "that the receipt AUTHORIZES the action you are about to take — a valid signature " |
| 243 | + "is authenticity, not authorization", |
| 244 | + "that the signing key is still trusted — REVOCATION and the issuer's clock are " |
| 245 | + "invisible to any local verifier, including this one", |
| 246 | + "that these bytes came from the run you think they did — this checks ONE token, " |
| 247 | + "and 'one run' is a property of a set (cr.evidence.root.v1)", |
| 248 | + ]} |
| 249 | + |
| 250 | + |
| 251 | +def verify_available() -> bool: |
| 252 | + """True when the ``[verify]`` extra is installed. Never used to soften a verdict.""" |
| 253 | + try: |
| 254 | + _ed25519() |
| 255 | + return True |
| 256 | + except ImportError: |
| 257 | + return False |
| 258 | + |
| 259 | + |
| 260 | +__all__ = ["verify_receipt", "keyring_from_document", "reconstruct_signed_input", |
| 261 | + "verify_available", "CLOCK_SKEW_LEEWAY_MS"] |
0 commit comments