Coverage for node / src / stigmem_node / identity / trust_store.py: 84%
114 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Peer trust store — reads/writes federation_manifests table (spec §19.8).
3Public surface:
4 store_peer_manifest(entity_uri, manifest, log_entry) -> None
5 get_peer_manifest(entity_uri) -> OrgManifest | None
6 refresh_peer_manifests() -> None (periodic task)
7 cleanup_expired_tokens() -> int (background cleanup, run opportunistically)
9Security requirements:
10 H1 mitigation: when resolving a peer manifest for token verification,
11 check manifest.expires_at > now; attempt refresh; reject if still expired.
12 Rotation chain invariant 4: reject any update that would regress key_id.
13"""
15from __future__ import annotations
17import json
18import logging
19import uuid
20from datetime import UTC, datetime
21from typing import TYPE_CHECKING
22from urllib.parse import urlsplit, urlunsplit
24import httpx
26from ..net_util import resolve_pinned_address
27from ..settings import settings
28from .manifest import (
29 ManifestError,
30 OrgManifest,
31 manifest_from_dict,
32 manifest_to_dict,
33 verify_manifest,
34 verify_rotation_chain,
35)
36from .transparency_log import LogEntry
38if TYPE_CHECKING:
39 pass
41logger = logging.getLogger("stigmem.identity.trust_store")
44def store_peer_manifest(
45 entity_uri: str,
46 manifest: OrgManifest,
47 log_entry: LogEntry | None = None,
48 *,
49 trust_mode: str = "relaxed",
50) -> None:
51 """Upsert a peer manifest with rotation-chain regression check.
53 Raises ManifestError if the update would regress to a previously-used key
54 (§19.1.4 invariant 4) or if the manifest fails self-verification.
55 """
56 from ..db import db
58 # Verify the manifest before storing
59 verify_manifest(manifest, trust_mode=trust_mode)
61 now = datetime.now(UTC).isoformat()
62 log_entry_json = (
63 json.dumps(
64 {
65 "log_id": log_entry.log_id,
66 "leaf_hash": log_entry.leaf_hash,
67 "log_index": log_entry.log_index,
68 "integrated_time": log_entry.integrated_time,
69 "inclusion_proof": log_entry.inclusion_proof,
70 }
71 )
72 if log_entry is not None
73 else None
74 )
76 manifest_json = json.dumps(manifest_to_dict(manifest), separators=(",", ":"))
78 with db() as conn:
79 existing = conn.execute(
80 "SELECT id, manifest_json, key_id FROM federation_manifests WHERE entity_uri = ?",
81 (entity_uri,),
82 ).fetchone()
84 if existing is not None:
85 # Rotation chain regression check: verify that the new manifest's chain
86 # connects to (or continues from) the previously-accepted key.
87 prev_manifest = manifest_from_dict(json.loads(existing["manifest_json"]))
88 if manifest.key_id != prev_manifest.key_id:
89 try:
90 verify_rotation_chain(
91 manifest,
92 previous_key_id=prev_manifest.key_id,
93 previous_pubkey_b64=prev_manifest.public_key,
94 )
95 except ManifestError as exc:
96 raise ManifestError(f"manifest update rejected: {exc}") from exc
98 conn.execute(
99 """UPDATE federation_manifests
100 SET manifest_json = ?,
101 signature = ?,
102 key_id = ?,
103 issued_at = ?,
104 expires_at = ?,
105 log_entry_json = ?,
106 updated_at = ?
107 WHERE entity_uri = ?""",
108 (
109 manifest_json,
110 manifest.signature,
111 manifest.key_id,
112 manifest.issued_at,
113 manifest.expires_at,
114 log_entry_json,
115 now,
116 entity_uri,
117 ),
118 )
119 else:
120 conn.execute(
121 """INSERT INTO federation_manifests
122 (id, entity_uri, manifest_json, signature, key_id,
123 issued_at, expires_at, log_entry_json, created_at, updated_at)
124 VALUES (?,?,?,?,?,?,?,?,?,?)""",
125 (
126 str(uuid.uuid4()),
127 entity_uri,
128 manifest_json,
129 manifest.signature,
130 manifest.key_id,
131 manifest.issued_at,
132 manifest.expires_at,
133 log_entry_json,
134 now,
135 now,
136 ),
137 )
139 logger.info("stored manifest for %s (key_id=%s)", entity_uri, manifest.key_id)
142def get_peer_manifest(
143 entity_uri: str,
144 *,
145 refresh_if_expired: bool = True,
146 trust_mode: str = "relaxed",
147) -> OrgManifest | None:
148 """Return the stored manifest for *entity_uri*, or None if unknown.
150 H1 mitigation: if the stored manifest is expired and refresh_if_expired is True,
151 we attempt an HTTP fetch from /.well-known/stigmem-manifest.json.
152 Returns None (reject) if the manifest is expired and cannot be refreshed.
153 """
154 from ..db import db
156 with db() as conn:
157 row = conn.execute(
158 "SELECT manifest_json, log_entry_json, expires_at FROM federation_manifests "
159 "WHERE entity_uri = ?",
160 (entity_uri,),
161 ).fetchone()
163 if row is None:
164 return None
166 manifest = manifest_from_dict(json.loads(row["manifest_json"]))
168 # H1: expiry check
169 now = datetime.now(UTC)
170 expires_at = datetime.fromisoformat(row["expires_at"].replace("Z", "+00:00"))
172 if expires_at <= now:
173 if not refresh_if_expired:
174 logger.warning("manifest for %s is expired; rejecting", entity_uri)
175 return None # caller must treat as rejection
176 # Attempt refresh
177 refreshed = _try_fetch_manifest(entity_uri)
178 if refreshed is None: 178 ↛ 181line 178 didn't jump to line 181 because the condition on line 178 was always true
179 logger.warning("manifest for %s expired; refresh failed; rejecting", entity_uri)
180 return None
181 try:
182 store_peer_manifest(entity_uri, refreshed, trust_mode=trust_mode)
183 except ManifestError as exc:
184 logger.warning("refreshed manifest for %s failed validation: %s", entity_uri, exc)
185 return None
186 return refreshed
188 return manifest
191def refresh_peer_manifests() -> None:
192 """Periodic task: refresh all active peer manifests from their well-known endpoints.
194 Alerts (logs warnings) on rotation events.
195 Also runs opportunistic cleanup of expired capability tokens.
196 """
197 from ..db import db
199 with db() as conn:
200 rows = conn.execute("SELECT entity_uri, manifest_json FROM federation_manifests").fetchall()
202 for row in rows:
203 entity_uri: str = row["entity_uri"]
204 prev_manifest = manifest_from_dict(json.loads(row["manifest_json"]))
205 refreshed = _try_fetch_manifest(entity_uri)
206 if refreshed is None: 206 ↛ 208line 206 didn't jump to line 208 because the condition on line 206 was always true
207 continue
208 if refreshed.key_id != prev_manifest.key_id:
209 logger.warning(
210 "key rotation detected for %s: %s -> %s",
211 entity_uri,
212 prev_manifest.key_id,
213 refreshed.key_id,
214 )
215 try:
216 store_peer_manifest(entity_uri, refreshed)
217 except ManifestError as exc:
218 logger.warning("skipping refresh for %s: %s", entity_uri, exc)
220 cleanup_expired_tokens()
223def _try_fetch_manifest(entity_uri: str) -> OrgManifest | None:
224 """Fetch /.well-known/stigmem-manifest.json from the peer's origin."""
225 # Derive base URL from entity_uri (strip scheme-specific parts if needed)
226 # entity_uri is expected to be an https:// URI or stigmem:// URI
227 if entity_uri.startswith("https://") or entity_uri.startswith("http://"):
228 from urllib.parse import urlparse
230 parsed = urlparse(entity_uri)
231 base_url = f"{parsed.scheme}://{parsed.netloc}"
232 else:
233 return None # can't derive URL from non-HTTP URI
235 try:
236 # R-5 / F-SSRF1 anti-rebind pin. This manifest is re-fetched on the recurring
237 # ``refresh_peer_manifests`` loop over every stored peer's entity_uri, so a
238 # peer-controlled host that passed validation can later DNS-rebind to an
239 # internal/IMDS address — the same resolve-then-reconnect TOCTOU the recurring
240 # pull path closes via ``federation_pull._pinned_get``. ``_pinned_manifest_get``
241 # resolves the host ONCE (https-only, rejecting the whole URL on any private
242 # record) BEFORE opening the client and connects to that exact pinned IP, so an
243 # ``http://`` entity_uri or a rebind target fails closed with no GET. The dev
244 # bypass is ``federation_insecure`` alone, matching the recurring-pull path.
245 resp = _pinned_manifest_get(
246 f"{base_url}/.well-known/stigmem-manifest.json",
247 timeout=10.0,
248 skip_pin=settings.federation_insecure,
249 )
250 if resp.status_code != 200: 250 ↛ 252line 250 didn't jump to line 252 because the condition on line 250 was always true
251 return None
252 data = resp.json()
253 manifest = manifest_from_dict(data)
254 verify_manifest(manifest)
255 return manifest
256 except Exception as exc:
257 logger.debug("failed to fetch manifest for %s: %s", entity_uri, exc)
258 return None
261def _pinned_manifest_get(
262 url: str,
263 *,
264 timeout: float,
265 skip_pin: bool,
266) -> httpx.Response:
267 """GET *url* with the a11 anti-rebind DNS pin (R-5 / F-SSRF1), unless *skip_pin*.
269 Synchronous sibling of ``federation_pull._pinned_get``. Resolves the host ONCE
270 via ``resolve_pinned_address`` (https-only — rejects the whole URL if ANY resolved
271 record is private/loopback/IMDS, and rejects a non-https scheme), connecting to the
272 EXACT pinned IP literal while preserving the ``Host`` header + TLS SNI + cert
273 verification against the original hostname. The pin is resolved BEFORE the client is
274 opened, so a blocked/rebind target fails closed (``ValueError``) without ever issuing
275 a request. ``follow_redirects=False`` blocks a redirect from re-introducing a
276 rebindable hop. Under *skip_pin* (``federation_insecure``) the original-hostname URL
277 is passed straight through with no pin extensions (dev/test escape).
278 """
279 if skip_pin:
280 return httpx.get(url, timeout=timeout, follow_redirects=False)
282 pinned_ip = resolve_pinned_address(url, allow_schemes=frozenset({"https"}))
283 parts = urlsplit(url)
284 hostname = parts.hostname or ""
285 port = parts.port
286 ip_authority = f"[{pinned_ip}]" if ":" in pinned_ip else pinned_ip
287 if port is not None:
288 netloc = f"{ip_authority}:{port}"
289 host_header = f"{hostname}:{port}"
290 else:
291 netloc = ip_authority
292 host_header = hostname
293 pinned_url = urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
294 # extensions={"sni_hostname": ...} runs TLS SNI + cert verification against the
295 # original hostname while the socket connects to the pinned IP literal (the webhook
296 # pin shape). httpx.get forwards it to the transient Client; the type stub omits the
297 # kwarg, so the runtime-valid call needs an ignore.
298 return httpx.get( # type: ignore[call-arg]
299 pinned_url,
300 timeout=timeout,
301 follow_redirects=False,
302 headers={"Host": host_header},
303 extensions={"sni_hostname": hostname},
304 )
307def cleanup_expired_tokens() -> int:
308 """Delete capability tokens expired more than 24 hours ago. Returns count deleted."""
309 from datetime import timedelta
311 from ..db import db
313 cutoff = (datetime.now(UTC) - timedelta(hours=24)).isoformat()
314 with db() as conn:
315 cur = conn.execute(
316 "DELETE FROM capability_tokens WHERE expiry < ?",
317 (cutoff,),
318 )
319 deleted: int = cur.rowcount or 0
320 if deleted: 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 logger.info("cleaned up %d expired capability tokens", deleted)
322 return deleted