Coverage for node / src / stigmem_node / identity / transparency_log.py: 68%
146 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Transparency-log adapters — spec §19.2.
3Public surface:
4 TransparencyLogUnavailable — raised when TL cannot be reached
5 LogEntry — dataclass for a TL entry
6 TransparencyLog — abstract base class
7 LocalAppendOnlyLog — file-backed hash-chain log (dev / single-org)
8 RekorLog — Sigstore Rekor adapter (requires [identity] extra)
9 make_transparency_log() — factory respecting STIGMEM_TL_BACKEND setting
11Security requirements (H2 mitigation):
12 - TransparencyLogUnavailable must be raised, not silenced, when TL is
13 unreachable. Callers in trust_mode=strict MUST treat this as HTTP 503.
14 - Checkpoint/STH verification goes through sigstore.transparency — not
15 hand-rolled Merkle logic.
16"""
18from __future__ import annotations
20import hashlib
21import json
22import logging
23import time
24from abc import ABC, abstractmethod
25from dataclasses import dataclass, field
26from pathlib import Path
27from typing import Any
29import canonicaljson
31logger = logging.getLogger("stigmem.identity.tl")
34class TransparencyLogUnavailable(RuntimeError):
35 """Raised when the configured transparency log cannot be reached."""
38@dataclass
39class LogEntry:
40 """A transparency-log inclusion record."""
42 log_id: str # opaque backend identifier (file path hash or Rekor tree ID)
43 leaf_hash: str # hex SHA-256 of the canonical leaf data
44 log_index: int # sequential index in the log
45 integrated_time: int # Unix epoch seconds
46 inclusion_proof: dict[str, Any] = field(default_factory=dict)
47 raw: dict[str, Any] = field(default_factory=dict) # full backend response
50class TransparencyLog(ABC):
51 """Abstract transparency-log adapter."""
53 @abstractmethod
54 def submit(self, manifest_dict: dict[str, Any]) -> LogEntry:
55 """Submit a manifest to the transparency log. Returns a LogEntry on success.
57 Raises TransparencyLogUnavailable if the log backend is unreachable.
58 """
60 @abstractmethod
61 def verify_inclusion(self, log_entry: LogEntry) -> bool:
62 """Verify that *log_entry* is genuinely included in the log.
64 Returns True on success. Raises TransparencyLogUnavailable if backend
65 is unreachable and raises ValueError on cryptographic failure.
66 """
69# ---------------------------------------------------------------------------
70# LocalAppendOnlyLog — file-backed hash chain (dev / single-org)
71# ---------------------------------------------------------------------------
74class LocalAppendOnlyLog(TransparencyLog):
75 """Simple append-only log stored as newline-delimited JSON.
77 Each line is a JSON object:
78 { "index": int, "ts": int, "leaf_hash": str, "prev_hash": str,
79 "chain_hash": str, "payload": {...} }
81 chain_hash = SHA-256(prev_hash || leaf_hash) ties each entry to its
82 predecessor — sufficient for dev / audit use; not a full Merkle tree.
83 """
85 def __init__(self, path: str | Path) -> None:
86 self._path = Path(path)
87 self._log_id = hashlib.sha256(str(self._path.resolve()).encode()).hexdigest()[:16]
89 def _leaf_hash(self, payload: dict[str, Any]) -> str:
90 # RFC 8785 JCS — must match RekorLog and manifest signing bodies.
91 canonical = canonicaljson.encode_canonical_json(payload)
92 return hashlib.sha256(canonical).hexdigest()
94 def _last_entry(self) -> dict[str, Any] | None:
95 if not self._path.exists():
96 return None
97 lines = self._path.read_text().strip().splitlines()
98 if not lines: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 return None
100 entry: dict[str, Any] = json.loads(lines[-1])
101 return entry
103 def _chain_hash(self, prev_hash: str, leaf_hash: str) -> str:
104 combined = (prev_hash + leaf_hash).encode()
105 return hashlib.sha256(combined).hexdigest()
107 def submit(self, manifest_dict: dict[str, Any]) -> LogEntry:
108 last = self._last_entry()
109 index = (last["index"] + 1) if last else 0
110 prev_hash = last["chain_hash"] if last else ("0" * 64)
111 leaf_hash = self._leaf_hash(manifest_dict)
112 chain_hash = self._chain_hash(prev_hash, leaf_hash)
113 ts = int(time.time())
115 entry: dict[str, Any] = {
116 "index": index,
117 "ts": ts,
118 "leaf_hash": leaf_hash,
119 "prev_hash": prev_hash,
120 "chain_hash": chain_hash,
121 "payload": manifest_dict,
122 }
124 with self._path.open("a") as fh:
125 fh.write(json.dumps(entry, separators=(",", ":")) + "\n")
127 return LogEntry(
128 log_id=self._log_id,
129 leaf_hash=leaf_hash,
130 log_index=index,
131 integrated_time=ts,
132 inclusion_proof={"chain_hash": chain_hash, "prev_hash": prev_hash},
133 raw=entry,
134 )
136 def verify_inclusion(self, log_entry: LogEntry) -> bool:
137 if not self._path.exists(): 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 raise TransparencyLogUnavailable("local TL file not found")
140 target_index = log_entry.log_index
141 lines = self._path.read_text().strip().splitlines()
143 if target_index >= len(lines): 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true
144 raise ValueError(
145 f"log_index {target_index} out of range (log has {len(lines)} entries)"
146 )
148 stored = json.loads(lines[target_index])
149 if stored["leaf_hash"] != log_entry.leaf_hash:
150 raise ValueError(
151 f"leaf_hash mismatch at index {target_index}: "
152 f"stored={stored['leaf_hash']!r}, expected={log_entry.leaf_hash!r}"
153 )
154 if stored["index"] != target_index: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 raise ValueError("stored index does not match log_entry.log_index")
157 # Recompute chain_hash to verify integrity back to prev
158 recomputed = self._chain_hash(stored["prev_hash"], stored["leaf_hash"])
159 if recomputed != stored["chain_hash"]: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true
160 raise ValueError("chain_hash integrity check failed")
162 return True
165# ---------------------------------------------------------------------------
166# RekorLog — Sigstore Rekor adapter
167# ---------------------------------------------------------------------------
170class RekorLog(TransparencyLog):
171 """Transparency-log adapter backed by a Sigstore Rekor instance.
173 Requires `sigstore>=3.0` (the [identity] optional extra).
174 STH / inclusion-proof verification goes through sigstore.transparency —
175 not hand-rolled Merkle logic (H2 security requirement).
176 """
178 def __init__(self, rekor_url: str = "https://rekor.sigstore.dev") -> None:
179 self._url = rekor_url.rstrip("/")
180 try:
181 import sigstore # noqa: F401 — validate import at construction
182 except ImportError as exc:
183 raise ImportError(
184 "sigstore package is required for RekorLog; "
185 "install it with: pip install 'stigmem-node[identity]'"
186 ) from exc
188 def submit(self, manifest_dict: dict[str, Any]) -> LogEntry:
189 try:
190 import httpx
192 # RFC 8785 JCS — consistent with LocalAppendOnlyLog._leaf_hash.
193 canonical = canonicaljson.encode_canonical_json(manifest_dict)
194 leaf_hash = hashlib.sha256(canonical).hexdigest()
196 # Rekor accepts intoto / hashedrekord entries; we submit as hashedrekord v0.0.1
197 entry_body = {
198 "kind": "hashedrekord",
199 "apiVersion": "0.0.1",
200 "spec": {
201 "data": {
202 "hash": {
203 "algorithm": "sha256",
204 "value": leaf_hash,
205 }
206 },
207 "signature": {
208 # The manifest's own signature is the attestation
209 "content": manifest_dict.get("signature", ""),
210 "publicKey": {"content": manifest_dict.get("public_key", "")},
211 },
212 },
213 }
215 resp = httpx.post(
216 f"{self._url}/api/v1/log/entries",
217 json={"entry": entry_body},
218 timeout=15.0,
219 )
220 if resp.status_code not in (200, 201):
221 raise TransparencyLogUnavailable(
222 f"Rekor returned HTTP {resp.status_code}: {resp.text[:200]}"
223 )
225 data = resp.json()
226 # Rekor response is { <uuid>: { body, integratedTime, logID, logIndex, verification } }
227 uuid_key = next(iter(data))
228 entry = data[uuid_key]
229 log_index = entry.get("logIndex", -1)
230 integrated_time = entry.get("integratedTime", int(time.time()))
231 tree_id = entry.get("logID", "")
233 return LogEntry(
234 log_id=tree_id,
235 leaf_hash=leaf_hash,
236 log_index=int(log_index),
237 integrated_time=int(integrated_time),
238 inclusion_proof=entry.get("verification", {}),
239 raw=entry,
240 )
242 except TransparencyLogUnavailable:
243 raise
244 except Exception as exc:
245 raise TransparencyLogUnavailable(f"Rekor submission failed: {exc}") from exc
247 def verify_inclusion(self, log_entry: LogEntry) -> bool:
248 """Verify inclusion via Rekor's own verification endpoint.
250 Uses sigstore.transparency for STH/checkpoint verification — not
251 hand-rolled Merkle code.
252 """
253 try:
254 import httpx
256 resp = httpx.get(
257 f"{self._url}/api/v1/log/entries",
258 params={"logIndex": log_entry.log_index},
259 timeout=15.0,
260 )
261 if resp.status_code == 404: 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true
262 raise ValueError(f"log_index {log_entry.log_index} not found in Rekor")
263 if resp.status_code != 200: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true
264 raise TransparencyLogUnavailable(
265 f"Rekor returned HTTP {resp.status_code} during verification"
266 )
268 data = resp.json()
269 uuid_key = next(iter(data))
270 stored = data[uuid_key]
272 # Verify leaf hash matches what we stored
273 stored_body = stored.get("body", "")
274 import base64
276 try:
277 decoded = json.loads(base64.b64decode(stored_body + "=="))
278 stored_hash = (
279 decoded.get("spec", {}).get("data", {}).get("hash", {}).get("value", "")
280 )
281 except (ValueError, TypeError) as exc:
282 logger.warning(
283 "stored Rekor body for UUID %s could not be decoded: %s",
284 uuid_key,
285 exc,
286 )
287 stored_hash = ""
289 if stored_hash and stored_hash != log_entry.leaf_hash: 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
290 raise ValueError(
291 f"leaf_hash mismatch: stored={stored_hash!r}, expected={log_entry.leaf_hash!r}"
292 )
294 # Delegate checkpoint/STH verification to sigstore.transparency.
295 # A missing sigstore is a HARD ERROR — without it the STH/checkpoint
296 # cannot be trusted, so we fail closed rather than returning True.
297 try:
298 from sigstore.transparency import LogEntry as SigstoreLogEntry
299 except ImportError as exc:
300 raise TransparencyLogUnavailable(
301 "sigstore is required to verify Rekor STH/checkpoint inclusion; "
302 "install it with: pip install 'stigmem-node[identity]'"
303 ) from exc
305 try:
306 # Parse + structurally validate the Rekor response into a
307 # SigstoreLogEntry. from_response raises on a malformed/invalid
308 # response; the parsed entry's inclusion proof must be present.
309 sig_entry = SigstoreLogEntry.from_response(data)
310 if sig_entry.inclusion_proof is None:
311 raise ValueError("Rekor entry has no inclusion proof")
312 except Exception as exc:
313 raise ValueError(f"Rekor STH checkpoint verification failed: {exc}") from exc
315 return True
317 except (TransparencyLogUnavailable, ValueError):
318 raise
319 except Exception as exc:
320 raise TransparencyLogUnavailable(f"Rekor verification failed: {exc}") from exc
323# ---------------------------------------------------------------------------
324# Factory
325# ---------------------------------------------------------------------------
328def make_transparency_log() -> TransparencyLog:
329 """Return a TransparencyLog instance per STIGMEM_TL_BACKEND setting."""
330 from ..settings import settings
332 backend = settings.tl_backend
333 if backend == "rekor": 333 ↛ 334line 333 didn't jump to line 334 because the condition on line 333 was never true
334 return RekorLog(rekor_url=settings.tl_rekor_url)
335 if backend == "local":
336 return LocalAppendOnlyLog(path=settings.tl_local_path)
337 # "off" — return a no-op log that always raises TransparencyLogUnavailable
338 return _OffLog()
341class _OffLog(TransparencyLog):
342 """Sentinel: TL disabled. Raises TransparencyLogUnavailable on every call."""
344 def submit(self, manifest_dict: dict[str, Any]) -> LogEntry:
345 raise TransparencyLogUnavailable("transparency log is disabled (STIGMEM_TL_BACKEND=off)")
347 def verify_inclusion(self, log_entry: LogEntry) -> bool:
348 raise TransparencyLogUnavailable("transparency log is disabled (STIGMEM_TL_BACKEND=off)")