Coverage for node / src / stigmem_node / lifecycle / tombstone_signing.py: 87%
104 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Tombstone signing and verification — spec §23.2.4, rev 14.
3Signing body: JCS over TombstoneRecord with "signature" and "reason" excluded
4(field-exclusion pattern per §19.1.3). This allows reason redaction before
5federation rebroadcast without invalidating the signature.
7Public surface:
8 sign_tombstone(record) -> TombstoneRecord
9 verify_tombstone_signature(record, public_key_b64) -> None (raises on fail)
10 get_node_key_id() -> str | None
11"""
13from __future__ import annotations
15import base64
16import logging
17from typing import Any
19import canonicaljson
20from cryptography.exceptions import InvalidSignature
21from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
23from ..models.tombstones import TombstoneRecord, TombstoneRevocationRecord
25logger = logging.getLogger("stigmem.tombstone")
28def _pad(b64url: str) -> str:
29 return b64url + "=" * (-len(b64url) % 4)
32def _pubkey_from_b64(b64: str) -> Ed25519PublicKey:
33 raw = base64.urlsafe_b64decode(_pad(b64))
34 return Ed25519PublicKey.from_public_bytes(raw)
37def _signing_body(record: TombstoneRecord) -> bytes:
38 """JCS-canonical bytes over TombstoneRecord with 'signature' and 'reason' excluded (§23.2.4).
40 F-12 §23.2.3: scope arrays are sorted lexicographically before canonicalization
41 to prevent interop failures from array-order differences across JSON implementations.
42 """
43 scope_val: Any = record.scope
44 if isinstance(scope_val, list): 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true
45 scope_val = sorted(scope_val)
46 doc: dict[str, Any] = {
47 "id": record.id,
48 "entity_uri": record.entity_uri,
49 "scope": scope_val,
50 "signed_by": record.signed_by,
51 "key_id": record.key_id,
52 "created_at": record.created_at,
53 "legal_hold": record.legal_hold,
54 }
55 return canonicaljson.encode_canonical_json(doc)
58def get_node_key_id() -> str | None:
59 """Return the node's current signing key_id (SHA-256 hex, 16-char prefix), or None."""
60 from ..identity.capability import load_node_private_key
61 from ..identity.key_rotation import generate_key_id
63 priv = load_node_private_key()
64 if priv is None: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true
65 return None
66 pub = priv.public_key()
67 return generate_key_id(pub)
70def sign_tombstone(record: TombstoneRecord) -> TombstoneRecord:
71 """Sign *record* with the node's active private key. Returns a new record with signature set."""
72 from ..identity.capability import load_node_private_key
73 from ..identity.key_rotation import generate_key_id
75 priv = load_node_private_key()
76 if priv is None:
77 raise RuntimeError("STIGMEM_NODE_PRIVATE_KEY not configured; cannot sign tombstones")
79 pub = priv.public_key()
80 key_id = generate_key_id(pub)
81 record = record.model_copy(update={"key_id": key_id})
82 body = _signing_body(record)
83 sig_bytes = priv.sign(body)
84 sig_b64 = base64.urlsafe_b64encode(sig_bytes).decode().rstrip("=")
85 return record.model_copy(update={"signature": sig_b64})
88def verify_tombstone_signature(record: TombstoneRecord, public_key_b64: str) -> None:
89 """Verify tombstone signature against *public_key_b64* (base64url Ed25519).
91 Raises ValueError on failure. Per §23.4.2.1 the caller must resolve the
92 signing key from the org manifest independently of the relaying peer.
93 """
94 try:
95 pub = _pubkey_from_b64(public_key_b64)
96 body = _signing_body(record)
97 sig_bytes = base64.urlsafe_b64decode(_pad(record.signature))
98 pub.verify(sig_bytes, body)
99 except InvalidSignature as exc:
100 raise ValueError("tombstone signature verification failed") from exc
101 except Exception as exc:
102 raise ValueError(f"tombstone signature error: {exc}") from exc
105class IssuerVerificationError(ValueError):
106 """The tombstone/revocation ISSUER-signer signature could not be verified (fail-closed).
108 ``reason`` is a short machine-stable code so callers (push route HTTP wrapper, pull
109 client log line) can branch / report consistently without parsing the message.
110 """
112 def __init__(self, reason: str, message: str | None = None) -> None:
113 self.reason = reason
114 super().__init__(message or reason)
117def _resolve_pubkey_for_key_id(manifest: Any, key_id: str) -> str | None:
118 """Return base64url public key from *manifest* matching *key_id*, or None.
120 Matches the manifest's current key, else scans rotation_events for a matching
121 ``new_key_id`` (dual-trust rotation window). Shared by the push route and the pull
122 client so both resolve the issuer key identically.
123 """
124 if manifest.key_id == key_id: 124 ↛ 127line 124 didn't jump to line 127 because the condition on line 124 was always true
125 pk: str = manifest.public_key
126 return pk
127 for evt in getattr(manifest, "rotation_events", []):
128 if getattr(evt, "new_key_id", None) == key_id:
129 new_pk: str | None = getattr(evt, "new_public_key", None)
130 return new_pk
131 return None
134def resolve_and_verify_tombstone_issuer(
135 record: Any,
136 *,
137 key_id: str,
138 signer_uri: str,
139 verifier: Any,
140) -> None:
141 """Resolve the signer manifest, pick the signing key, and verify the issuer signature.
143 The single shared issuer-signer verification used by BOTH the push ingest route
144 (``routes/_federation_impl._verify_signed_artifact_or_400`` wraps this into HTTP
145 errors + audit) and the pull client (``federation_pull.pull_tombstones_from_peer_once``,
146 which closes the W6.1 gap where pull did NOT verify the issuer). Keeping it in one place
147 means push and pull can never drift.
149 Raises ``IssuerVerificationError(reason=...)`` on any failure. ``reason`` is one of:
150 ``missing_key_id`` / ``signer_manifest_missing`` / ``key_id_not_in_signer_manifest`` /
151 a verifier ValueError string (signature mismatch). Returns None == verified.
152 """
153 from ..identity.trust_store import get_peer_manifest
155 if not key_id:
156 raise IssuerVerificationError("missing_key_id", "tombstone missing key_id")
157 manifest = get_peer_manifest(signer_uri)
158 if manifest is None:
159 raise IssuerVerificationError(
160 "signer_manifest_missing", f"no manifest for signer {signer_uri!r}"
161 )
162 pubkey_b64 = _resolve_pubkey_for_key_id(manifest, key_id)
163 if pubkey_b64 is None: 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true
164 raise IssuerVerificationError(
165 "key_id_not_in_signer_manifest", "key_id not in signer manifest"
166 )
167 try:
168 verifier(record, pubkey_b64)
169 except ValueError as exc:
170 raise IssuerVerificationError(str(exc), str(exc)) from exc
173def _revocation_signing_body(record: TombstoneRevocationRecord) -> bytes:
174 """JCS-canonical bytes over TombstoneRevocationRecord with 'signature' and 'reason' excluded."""
175 doc: dict[str, Any] = {
176 "id": record.id,
177 "tombstone_id": record.tombstone_id,
178 "signed_by": record.signed_by,
179 "key_id": record.key_id,
180 "created_at": record.created_at,
181 }
182 return canonicaljson.encode_canonical_json(doc)
185def sign_revocation(record: TombstoneRevocationRecord) -> TombstoneRevocationRecord:
186 """Sign *record* with the node's active private key. Returns a new record with signature set."""
187 from ..identity.capability import load_node_private_key
188 from ..identity.key_rotation import generate_key_id
190 priv = load_node_private_key()
191 if priv is None:
192 raise RuntimeError("STIGMEM_NODE_PRIVATE_KEY not configured; cannot sign revocations")
194 pub = priv.public_key()
195 key_id = generate_key_id(pub)
196 record = record.model_copy(update={"key_id": key_id})
197 body = _revocation_signing_body(record)
198 sig_bytes = priv.sign(body)
199 sig_b64 = base64.urlsafe_b64encode(sig_bytes).decode().rstrip("=")
200 return record.model_copy(update={"signature": sig_b64})
203def verify_revocation_signature(record: TombstoneRevocationRecord, public_key_b64: str) -> None:
204 """Verify revocation signature against *public_key_b64* (base64url Ed25519).
206 Raises ValueError on failure.
207 """
208 try:
209 pub = _pubkey_from_b64(public_key_b64)
210 body = _revocation_signing_body(record)
211 sig_bytes = base64.urlsafe_b64decode(_pad(record.signature))
212 pub.verify(sig_bytes, body)
213 except InvalidSignature as exc:
214 raise ValueError("revocation signature verification failed") from exc
215 except Exception as exc:
216 raise ValueError(f"revocation signature error: {exc}") from exc