Coverage for node / src / stigmem_node / federation / origin_signature.py: 90%
88 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Phase 2b/2c — per-fact origin signature (design §2.2, §2c W3.1).
3The origin node signs the JCS-canonical tuple
4(tv, fact_id, cid, origin_tenant, origin_node_id, origin_allowed_scopes,
5origin_allowed_tenants, origin_entity_uri, valid_until) with its Ed25519 federation key
6(== its manifest key after Phase 2a unification). Receivers verify against the pubkey set
7from resolve_origin_key(). CID binds content; this signature binds content <-> origin
8identity <-> tenant <-> authorization <-> visibility window, and the fact_id (F-1: fact_id
9is NOT in the CID, so binding it here makes the wire id tamper-evident — dedup and the
10received_from/conflict graph key on it). valid_until is included because it is excluded
11from the CID and is authorization-relevant. Verification runs regardless of trust_mode
12(including 'off') — it is the hard gate.
14Phase 2c W3.1 — HARD CUTOVER to v2.1: the tuple now ALSO binds the origin's ``entity_uri``
15(so a relay cannot lie about which origin a relayed fact came from — the receiver fetches
16and verifies the origin's manifest by this entity_uri) and a hardcoded in-body tuple
17version ``tv = "2.1"`` (forward-proofing: the signature commits to its exact field set).
18``entity_uri`` is MANDATORY — there is NO legacy 6-field verify path. An origin block whose
19``entity_uri`` is missing or empty is REJECTED, never silently verified under the old 2b
20tuple (anti-downgrade: a 6-field fallback would let an attacker strip ``entity_uri`` and
21defeat the binding). Facts signed under old 2b (no entity_uri) are simply not relayable;
22that is acceptable and fail-safe.
23"""
25from __future__ import annotations
27import base64
28from typing import Any
30import canonicaljson
31from cryptography.exceptions import InvalidSignature
32from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
35class OriginSignatureError(ValueError):
36 """Origin signature missing, malformed, or failed verification (fail-closed)."""
39# Hardcoded in-body tuple version (Phase 2c W3.1). Committing it INTO the signed bytes makes
40# the signature pin its exact field set, so a future 2.2 (different fields) can never be
41# confused with a 2.1 signature. Bump in lockstep with any change to the signed field set.
42_TUPLE_VERSION = "2.1"
44# Hardcoded in-body tuple version for the TOMBSTONE origin-attestation tuple (Phase 2c W6.3).
45# DISTINCT from the fact tuple's "2.1" so a tombstone origin signature can NEVER be confused
46# with / replayed as a fact origin signature (domain separation is a security property — the
47# two tuples share the same private key and base64url framing, only the signed bytes differ).
48_TOMBSTONE_TUPLE_VERSION = "t2.1"
50# Hardcoded in-body tuple version for the REVOCATION origin-attestation tuple (Phase 2c Rev-1).
51# DISTINCT from both the fact tuple's "2.1" and the tombstone tuple's "t2.1" so a revocation
52# origin signature can NEVER be confused with / replayed as a tombstone or fact origin
53# signature (domain separation — all three share the same key and base64url framing, only the
54# signed bytes differ).
55_REVOCATION_TUPLE_VERSION = "r2.1"
58def _pad(s: str) -> str:
59 return s + "=" * (-len(s) % 4)
62def _require_entity_uri(origin: dict[str, Any]) -> str:
63 """Return a non-empty origin ``entity_uri`` or raise (anti-downgrade, W3.1).
65 ``entity_uri`` is MANDATORY in the v2.1 tuple. A missing or empty value is a
66 hard rejection — never a fall-through to a legacy 6-field tuple.
67 """
68 entity_uri = origin.get("entity_uri")
69 if not entity_uri:
70 raise OriginSignatureError("origin entity_uri missing or empty (v2.1 requires it)")
71 return str(entity_uri)
74def canonical_origin_tuple(
75 *,
76 fact_id: str,
77 cid: str,
78 origin_tenant: str,
79 origin_node_id: str,
80 origin_allowed_scopes: list[str],
81 origin_allowed_tenants: list[str],
82 valid_until: str | None,
83 entity_uri: str,
84) -> bytes:
85 """RFC 8785 JCS bytes of the signed v2.1 origin tuple (sets sorted for determinism).
87 ``entity_uri`` and the hardcoded ``tv`` constant are bound INTO the tuple (W3.1).
88 """
89 return canonicaljson.encode_canonical_json(
90 {
91 "cid": cid,
92 "entity_uri": entity_uri,
93 "fact_id": fact_id,
94 "origin_allowed_scopes": sorted(origin_allowed_scopes),
95 "origin_allowed_tenants": sorted(origin_allowed_tenants),
96 "origin_node_id": origin_node_id,
97 "origin_tenant": origin_tenant,
98 "tv": _TUPLE_VERSION,
99 "valid_until": valid_until,
100 }
101 )
104def sign_origin(
105 private_key: Ed25519PrivateKey,
106 *,
107 fact_id: str,
108 cid: str,
109 origin: dict[str, Any],
110 valid_until: str | None,
111) -> str:
112 """Sign the origin tuple; returns base64url signature (no padding)."""
113 body = canonical_origin_tuple(
114 fact_id=fact_id,
115 cid=cid,
116 origin_tenant=origin["tenant"],
117 origin_node_id=origin["node_id"],
118 origin_allowed_scopes=origin["allowed_scopes"],
119 origin_allowed_tenants=origin["allowed_tenants"],
120 valid_until=valid_until,
121 entity_uri=_require_entity_uri(origin),
122 )
123 return base64.urlsafe_b64encode(private_key.sign(body)).decode().rstrip("=")
126def verify_origin_signature(
127 sig_b64: str,
128 *,
129 fact_id: str,
130 cid: str,
131 origin: dict[str, Any],
132 valid_until: str | None,
133 allowed_pubkeys: set[str],
134) -> None:
135 """Verify sig against ANY key in allowed_pubkeys (current + rotation window).
137 Raises OriginSignatureError on any failure. Returning None == verified.
138 """
139 if not sig_b64 or not allowed_pubkeys:
140 raise OriginSignatureError("origin signature or key set missing")
141 # Anti-downgrade (W3.1): require entity_uri BEFORE building the tuple. A missing/empty
142 # value raises here — there is NO 6-field legacy reconstruction to fall back to, so a
143 # relay cannot strip entity_uri to defeat the origin->entity binding.
144 entity_uri = _require_entity_uri(origin)
145 try:
146 body = canonical_origin_tuple(
147 fact_id=fact_id,
148 cid=cid,
149 origin_tenant=origin["tenant"],
150 origin_node_id=origin["node_id"],
151 origin_allowed_scopes=origin["allowed_scopes"],
152 origin_allowed_tenants=origin["allowed_tenants"],
153 valid_until=valid_until,
154 entity_uri=entity_uri,
155 )
156 sig = base64.urlsafe_b64decode(_pad(sig_b64))
157 except (KeyError, TypeError, ValueError) as exc:
158 raise OriginSignatureError(f"malformed origin block or signature: {exc}") from exc
159 for pub_b64 in allowed_pubkeys:
160 try:
161 pub = Ed25519PublicKey.from_public_bytes(base64.urlsafe_b64decode(_pad(pub_b64)))
162 pub.verify(sig, body)
163 return
164 except (InvalidSignature, ValueError):
165 continue
166 raise OriginSignatureError("origin signature did not verify against any allowed key")
169# ---------------------------------------------------------------------------
170# Tombstone origin-attestation signature (Phase 2c W6.3)
171#
172# SEPARATE from the existing tombstone issuer-signer signature
173# (lifecycle/tombstone_signing.py). That one says "this org issued this RTBF tombstone"; this
174# one says "this ORIGIN node relayed this tombstone under THIS propagation grant". Binding the
175# tombstone ``id``, ``entity_uri`` and ``scope`` into the signed tuple is the anti-relaunder
176# property: a relay that widens ``scope`` ("local" -> "*"), retargets ``entity_uri``, or lies
177# about its grant invalidates the signature. ``tv = "t2.1"`` (distinct from the fact tuple's
178# "2.1") gives hard domain separation so the two origin signatures are never interchangeable.
179# ---------------------------------------------------------------------------
182def canonical_tombstone_origin_tuple(
183 *,
184 tombstone_id: str,
185 entity_uri: str,
186 scope: str,
187 origin_node_id: str,
188 origin_tenant: str,
189 origin_allowed_scopes: list[str],
190 origin_allowed_tenants: list[str],
191 origin_entity_uri: str,
192) -> bytes:
193 """RFC 8785 JCS bytes of the signed tombstone origin tuple (sets sorted for determinism).
195 Binds the tombstone ``id`` (as ``tid``), subject ``entity_uri`` and ``scope`` so a relay
196 cannot relaunder or re-scope the suppression. ``tv`` is the hardcoded constant ``"t2.1"``,
197 distinct from the fact tuple's ``"2.1"`` (domain separation, W6.3).
198 """
199 return canonicaljson.encode_canonical_json(
200 {
201 "entity_uri": entity_uri,
202 "origin_allowed_scopes": sorted(origin_allowed_scopes),
203 "origin_allowed_tenants": sorted(origin_allowed_tenants),
204 "origin_entity_uri": origin_entity_uri,
205 "origin_node_id": origin_node_id,
206 "origin_tenant": origin_tenant,
207 "scope": scope,
208 "tid": tombstone_id,
209 "tv": _TOMBSTONE_TUPLE_VERSION,
210 }
211 )
214def _require_value(value: str, name: str) -> str:
215 """Return a non-empty value or raise (anti-downgrade, W6.3)."""
216 if not value:
217 raise OriginSignatureError(f"tombstone origin {name} missing or empty")
218 return value
221def sign_tombstone_origin(
222 private_key: Ed25519PrivateKey,
223 *,
224 tombstone_id: str,
225 entity_uri: str,
226 scope: str,
227 origin_node_id: str,
228 origin_tenant: str,
229 origin_allowed_scopes: list[str],
230 origin_allowed_tenants: list[str],
231 origin_entity_uri: str,
232) -> str:
233 """Sign the tombstone origin tuple; returns base64url signature (no padding)."""
234 body = canonical_tombstone_origin_tuple(
235 tombstone_id=tombstone_id,
236 entity_uri=_require_value(entity_uri, "entity_uri"),
237 scope=scope,
238 origin_node_id=origin_node_id,
239 origin_tenant=origin_tenant,
240 origin_allowed_scopes=origin_allowed_scopes,
241 origin_allowed_tenants=origin_allowed_tenants,
242 origin_entity_uri=_require_value(origin_entity_uri, "entity_uri"),
243 )
244 return base64.urlsafe_b64encode(private_key.sign(body)).decode().rstrip("=")
247def verify_tombstone_origin_signature(
248 sig_b64: str,
249 *,
250 tombstone_id: str,
251 entity_uri: str,
252 scope: str,
253 origin_node_id: str,
254 origin_tenant: str,
255 origin_allowed_scopes: list[str],
256 origin_allowed_tenants: list[str],
257 origin_entity_uri: str,
258 allowed_pubkeys: set[str],
259) -> None:
260 """Verify a tombstone origin sig against ANY key in allowed_pubkeys (rotation window).
262 Raises OriginSignatureError on any failure. Returning None == verified. Requires both the
263 subject ``entity_uri`` and the ``origin_entity_uri`` to be present/non-empty BEFORE building
264 the tuple (anti-downgrade: there is no legacy field-stripped path to fall back to).
265 """
266 if not sig_b64 or not allowed_pubkeys: 266 ↛ 267line 266 didn't jump to line 267 because the condition on line 266 was never true
267 raise OriginSignatureError("tombstone origin signature or key set missing")
268 # Anti-downgrade (W6.3): require both entity_uri fields before building the tuple.
269 entity_uri = _require_value(entity_uri, "entity_uri")
270 origin_entity_uri = _require_value(origin_entity_uri, "entity_uri")
271 try:
272 body = canonical_tombstone_origin_tuple(
273 tombstone_id=tombstone_id,
274 entity_uri=entity_uri,
275 scope=scope,
276 origin_node_id=origin_node_id,
277 origin_tenant=origin_tenant,
278 origin_allowed_scopes=origin_allowed_scopes,
279 origin_allowed_tenants=origin_allowed_tenants,
280 origin_entity_uri=origin_entity_uri,
281 )
282 sig = base64.urlsafe_b64decode(_pad(sig_b64))
283 except (KeyError, TypeError, ValueError) as exc:
284 raise OriginSignatureError(
285 f"malformed tombstone origin block or signature: {exc}"
286 ) from exc
287 for pub_b64 in allowed_pubkeys:
288 try:
289 pub = Ed25519PublicKey.from_public_bytes(base64.urlsafe_b64decode(_pad(pub_b64)))
290 pub.verify(sig, body)
291 return
292 except (InvalidSignature, ValueError):
293 continue
294 raise OriginSignatureError(
295 "tombstone origin signature did not verify against any allowed key"
296 )
299# ---------------------------------------------------------------------------
300# Revocation origin-attestation signature (Phase 2c Rev-1)
301#
302# SEPARATE from the existing revocation issuer-signer signature
303# (sign_revocation/verify_revocation_signature). That one says "this org issued this tombstone
304# REVERSAL"; this one says "this ORIGIN node relayed this revocation under THIS propagation
305# grant". A revocation has no entity_uri/scope of its own — it references a tombstone by id —
306# so the tuple binds the revocation ``id`` (as ``rid``) and the referenced ``tombstone_id``
307# instead. Binding both is the anti-relaunder property: a relay that retargets which revocation
308# or which tombstone it carries, or lies about its grant, invalidates the signature.
309# ``tv = "r2.1"`` (distinct from the fact tuple's "2.1" and the tombstone tuple's "t2.1") gives
310# hard domain separation so the three origin signatures are never interchangeable.
311# ---------------------------------------------------------------------------
314def canonical_revocation_origin_tuple(
315 *,
316 revocation_id: str,
317 tombstone_id: str,
318 origin_node_id: str,
319 origin_tenant: str,
320 origin_allowed_scopes: list[str],
321 origin_allowed_tenants: list[str],
322 origin_entity_uri: str,
323) -> bytes:
324 """RFC 8785 JCS bytes of the signed revocation origin tuple (sets sorted for determinism).
326 Binds the revocation ``id`` (as ``rid``) and the referenced ``tombstone_id`` so a relay
327 cannot retarget the reversal. ``tv`` is the hardcoded constant ``"r2.1"``, distinct from
328 the fact tuple's ``"2.1"`` and the tombstone tuple's ``"t2.1"`` (domain separation, Rev-1).
329 """
330 return canonicaljson.encode_canonical_json(
331 {
332 "origin_allowed_scopes": sorted(origin_allowed_scopes),
333 "origin_allowed_tenants": sorted(origin_allowed_tenants),
334 "origin_entity_uri": origin_entity_uri,
335 "origin_node_id": origin_node_id,
336 "origin_tenant": origin_tenant,
337 "rid": revocation_id,
338 "tombstone_id": tombstone_id,
339 "tv": _REVOCATION_TUPLE_VERSION,
340 }
341 )
344def sign_revocation_origin(
345 private_key: Ed25519PrivateKey,
346 *,
347 revocation_id: str,
348 tombstone_id: str,
349 origin_node_id: str,
350 origin_tenant: str,
351 origin_allowed_scopes: list[str],
352 origin_allowed_tenants: list[str],
353 origin_entity_uri: str,
354) -> str:
355 """Sign the revocation origin tuple; returns base64url signature (no padding)."""
356 body = canonical_revocation_origin_tuple(
357 revocation_id=revocation_id,
358 tombstone_id=tombstone_id,
359 origin_node_id=origin_node_id,
360 origin_tenant=origin_tenant,
361 origin_allowed_scopes=origin_allowed_scopes,
362 origin_allowed_tenants=origin_allowed_tenants,
363 origin_entity_uri=_require_value(origin_entity_uri, "entity_uri"),
364 )
365 return base64.urlsafe_b64encode(private_key.sign(body)).decode().rstrip("=")
368def verify_revocation_origin_signature(
369 sig_b64: str,
370 *,
371 revocation_id: str,
372 tombstone_id: str,
373 origin_node_id: str,
374 origin_tenant: str,
375 origin_allowed_scopes: list[str],
376 origin_allowed_tenants: list[str],
377 origin_entity_uri: str,
378 allowed_pubkeys: set[str],
379) -> None:
380 """Verify a revocation origin sig against ANY key in allowed_pubkeys (rotation window).
382 Raises OriginSignatureError on any failure. Returning None == verified. Requires
383 ``origin_entity_uri`` to be present/non-empty BEFORE building the tuple (anti-downgrade:
384 there is no legacy field-stripped path to fall back to).
385 """
386 if not sig_b64 or not allowed_pubkeys: 386 ↛ 387line 386 didn't jump to line 387 because the condition on line 386 was never true
387 raise OriginSignatureError("revocation origin signature or key set missing")
388 # Anti-downgrade (Rev-1): require origin_entity_uri before building the tuple.
389 origin_entity_uri = _require_value(origin_entity_uri, "entity_uri")
390 try:
391 body = canonical_revocation_origin_tuple(
392 revocation_id=revocation_id,
393 tombstone_id=tombstone_id,
394 origin_node_id=origin_node_id,
395 origin_tenant=origin_tenant,
396 origin_allowed_scopes=origin_allowed_scopes,
397 origin_allowed_tenants=origin_allowed_tenants,
398 origin_entity_uri=origin_entity_uri,
399 )
400 sig = base64.urlsafe_b64decode(_pad(sig_b64))
401 except (KeyError, TypeError, ValueError) as exc:
402 raise OriginSignatureError(
403 f"malformed revocation origin block or signature: {exc}"
404 ) from exc
405 for pub_b64 in allowed_pubkeys:
406 try:
407 pub = Ed25519PublicKey.from_public_bytes(base64.urlsafe_b64decode(_pad(pub_b64)))
408 pub.verify(sig, body)
409 return
410 except (InvalidSignature, ValueError):
411 continue
412 raise OriginSignatureError(
413 "revocation origin signature did not verify against any allowed key"
414 )