Coverage for node / src / stigmem_node / routes / federation / peers.py: 93%
99 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Federation peer registration and listing routes."""
3from __future__ import annotations
5import json
6from typing import Annotated, Any
8from fastapi import BackgroundTasks, Depends, HTTPException, status
9from pydantic import BaseModel, field_validator
11from ...auth import Identity, resolve_identity
12from ...db import db
13from ...models.federation import (
14 PeerApprovalRequest,
15 PeerApprovalResponse,
16 PeerRegisterRequest,
17 PeerRegisterResponse,
18)
19from ...tenant import TenantIdError, validate_tenant_id
20from .._federation_impl import approve_peer_impl, register_peer_impl
21from .common import _public_module, router
24@router.post(
25 "/v1/federation/peers",
26 response_model=PeerRegisterResponse,
27 status_code=status.HTTP_201_CREATED,
28)
29async def register_peer(
30 req: PeerRegisterRequest,
31 background_tasks: BackgroundTasks,
32 identity: Annotated[Identity, Depends(resolve_identity)],
33) -> PeerRegisterResponse:
34 """Register a peer.
36 Fetches its well-known doc and verifies declaration_sig
37 (Spec-05-Federation-Trust).
38 """
39 # Implementation lives in _federation_impl.register_peer_impl.
40 return await register_peer_impl(req, background_tasks, identity)
43@router.post(
44 "/v1/federation/peers/{peer_id}/approve",
45 response_model=PeerApprovalResponse,
46)
47def approve_peer(
48 peer_id: str,
49 req: PeerApprovalRequest,
50 background_tasks: BackgroundTasks,
51 identity: Annotated[Identity, Depends(resolve_identity)],
52) -> PeerApprovalResponse:
53 """Approve a pending peer after out-of-band public-key confirmation."""
54 return approve_peer_impl(peer_id, req.pubkey_fingerprint, background_tasks, identity)
57# ---------------------------------------------------------------------------
58# PATCH /v1/federation/peers/{peer_id} — set per-peer tenant policy (mig. 041)
59# ---------------------------------------------------------------------------
62class PeerPolicyPatch(BaseModel):
63 """Operator-settable per-peer tenant policy (migration 041)."""
65 pull_tenant: str | None = None
66 ingest_tenant: str | None = None
67 allowed_tenants: list[str] | None = None
68 trust_tier: str | None = None
69 tenant_map: dict[str, str] | None = None
71 @field_validator("trust_tier")
72 @classmethod
73 def _tier(cls, v: str | None) -> str | None:
74 if v is not None and v not in ("cross_org", "same_domain"):
75 raise ValueError("trust_tier must be 'cross_org' or 'same_domain'")
76 return v
78 @field_validator("tenant_map")
79 @classmethod
80 def _tenant_map(cls, v: dict[str, str] | None) -> dict[str, str] | None:
81 if v is None: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 return None
83 # Store the NORMALIZED tenant ids (validate_tenant_id NFKC/strip/lowercases) so the
84 # persisted map keys/values are canonical and match the canonical origin_tenant the
85 # ingest path looks up — a raw mixed-case/unicode entry would otherwise silently miss
86 # and fail-closed-deny.
87 normalized: dict[str, str] = {}
88 for origin_tenant, local_tenant in v.items():
89 try:
90 normalized[validate_tenant_id(origin_tenant)] = validate_tenant_id(local_tenant)
91 except TenantIdError as exc:
92 raise ValueError(f"invalid tenant_map entry: {exc}") from exc
93 return normalized
96@router.patch("/v1/federation/peers/{peer_id}")
97def patch_peer_policy(
98 peer_id: str,
99 req: PeerPolicyPatch,
100 identity: Annotated[Identity, Depends(resolve_identity)],
101) -> dict[str, Any]:
102 if not identity.can_admin_federation():
103 raise HTTPException(status_code=403, detail="federation admin required")
104 sets: list[str] = []
105 params: list[Any] = []
106 if req.pull_tenant is not None:
107 sets.append("pull_tenant = ?")
108 params.append(req.pull_tenant)
109 if req.ingest_tenant is not None:
110 sets.append("ingest_tenant = ?")
111 params.append(req.ingest_tenant)
112 if req.allowed_tenants is not None:
113 sets.append("allowed_tenants = ?")
114 params.append(json.dumps(req.allowed_tenants))
115 if req.trust_tier is not None:
116 sets.append("trust_tier = ?")
117 params.append(req.trust_tier)
118 if not sets and req.tenant_map is None:
119 raise HTTPException(status_code=400, detail="no policy fields provided")
120 params.append(peer_id)
121 with db() as conn:
122 if req.trust_tier == "same_domain":
123 prow = conn.execute(
124 "SELECT entity_uri FROM peers WHERE id = ?", (peer_id,)
125 ).fetchone()
126 if prow is None: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 raise HTTPException(status_code=404, detail="peer not found")
128 if not (prow["entity_uri"] or "").strip():
129 raise HTTPException(
130 status_code=422,
131 detail="trust_tier=same_domain requires a verified entity_uri (Phase 2a)",
132 )
133 if sets:
134 cur = conn.execute(
135 f"UPDATE peers SET {', '.join(sets)} WHERE id = ?", # noqa: S608 # nosec B608 — set clauses are literal column fragments; values in params
136 params,
137 )
138 if cur.rowcount == 0:
139 raise HTTPException(status_code=404, detail="peer not found")
140 elif ( 140 ↛ 145line 140 didn't jump to line 145 because the condition on line 140 was never true
141 conn.execute("SELECT 1 FROM peers WHERE id = ?", (peer_id,)).fetchone()
142 is None
143 ):
144 # tenant_map-only PATCH still requires the peer to exist.
145 raise HTTPException(status_code=404, detail="peer not found")
146 if req.tenant_map is not None:
147 # Full-replace semantics: clear then re-insert ({} clears).
148 conn.execute(
149 "DELETE FROM peer_tenant_map WHERE peer_id = ?", (peer_id,)
150 )
151 conn.executemany(
152 "INSERT INTO peer_tenant_map (peer_id, origin_tenant, local_tenant) "
153 "VALUES (?, ?, ?)",
154 [(peer_id, ot, lt) for ot, lt in req.tenant_map.items()],
155 )
156 conn.commit()
157 updated = [s.split(" =")[0] for s in sets]
158 if req.tenant_map is not None:
159 updated.append("tenant_map")
160 _public_module().write_audit_log(
161 peer_id,
162 "peer_policy_updated",
163 {"updated": updated, "by": identity.entity_uri},
164 )
165 return {"peer_id": peer_id, "updated": updated}
168# ---------------------------------------------------------------------------
169# GET /v1/federation/peers — list peers (§5.7)
170# ---------------------------------------------------------------------------
173def _decode_allowed_tenants(raw: Any) -> list[str]:
174 """Defensively json-decode ``allowed_tenants``; default ``[]`` when null/blank."""
175 if not raw:
176 return []
177 try:
178 decoded = json.loads(raw)
179 except (TypeError, ValueError):
180 return []
181 return decoded if isinstance(decoded, list) else []
184@router.get("/v1/federation/peers")
185def list_peers(
186 identity: Annotated[Identity, Depends(resolve_identity)],
187) -> dict[str, Any]:
188 if not identity.can_federate(): 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true
189 raise HTTPException(status_code=403, detail="federate permission required")
190 with db() as conn:
191 rows = conn.execute(
192 "SELECT id, node_id, node_url, status, allowed_scopes, established_at, "
193 "pull_tenant, ingest_tenant, allowed_tenants, trust_tier FROM peers"
194 ).fetchall()
195 return {
196 "peers": [
197 {
198 "peer_id": r["id"],
199 "node_id": r["node_id"],
200 "node_url": r["node_url"],
201 "status": r["status"],
202 "allowed_scopes": json.loads(r["allowed_scopes"]),
203 "established_at": r["established_at"],
204 "pull_tenant": r["pull_tenant"],
205 "ingest_tenant": r["ingest_tenant"],
206 "allowed_tenants": _decode_allowed_tenants(r["allowed_tenants"]),
207 "trust_tier": r["trust_tier"],
208 }
209 for r in rows
210 ]
211 }