Coverage for node / src / stigmem_node / routes / quarantine.py: 84%
89 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Quarantine admin API — spec §19.5.
3GET /v1/quarantine — list quarantined facts (all or by garden)
4POST /v1/quarantine/{fact_id}/admit — shorthand: promote fact to main fabric
5POST /v1/quarantine/{fact_id}/reject — shorthand: reject a quarantined fact
7These endpoints are convenience wrappers over the garden-level promote/reject
8endpoints (POST /v1/gardens/:id/promote|reject). They operate node-globally:
9the caller addresses a fact by ID without needing to know its quarantine garden.
10Requires node admin (write) permission.
11"""
13from __future__ import annotations
15import uuid
16from datetime import UTC, datetime
17from typing import Annotated, Any
19from fastapi import APIRouter, Depends, HTTPException, Query, status
21from ..audit_event import INSTRUCTION_PROMOTED, emit_instruction_event_if_applicable
22from ..auth import Identity, resolve_identity
23from ..db import db
24from ..garden_acl import get_garden_by_slug_or_id, require_quarantine_moderator_or_admin
25from ..lifecycle.immutability import (
26 set_fact_garden_membership,
27 set_fact_quarantine_status,
28 set_fact_validity_override,
29)
30from ..models.constants import QUARANTINE_PENDING
31from ..models.gardens import (
32 QuarantineListResponse,
33 QuarantineRecord,
34)
36router = APIRouter(prefix="/v1/quarantine", tags=["quarantine"])
39def _require_write(identity: Identity) -> None:
40 if not identity.can_write(): 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true
41 raise HTTPException(
42 status_code=status.HTTP_403_FORBIDDEN, detail="write permission required"
43 )
46# ---------------------------------------------------------------------------
47# List quarantined facts
48# ---------------------------------------------------------------------------
51@router.get("", response_model=QuarantineListResponse)
52def list_quarantined_facts(
53 identity: Annotated[Identity, Depends(resolve_identity)],
54 garden_id: str | None = Query(None, description="Filter by quarantine garden UUID or slug"),
55 quarantine_status: str | None = Query(
56 None, description="Filter by status: pending, promoted, rejected"
57 ),
58 limit: int = Query(100, ge=1, le=1000),
59 offset: int = Query(0, ge=0),
60) -> QuarantineListResponse:
61 """List facts in the quarantine system (Spec-08-Quarantine-Garden).
63 Scoped to the caller's tenant: admins see all quarantined facts within their
64 own tenant. Other callers see facts only in quarantine gardens where they
65 hold a member role. No caller can read another tenant's quarantine.
66 """
67 if not identity.can_read(): 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true
68 raise HTTPException(
69 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required"
70 )
72 projection_joins = (
73 " LEFT JOIN fact_quarantine_status fqs ON fqs.fact_id = f.id"
74 " LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = f.id"
75 )
76 quarantine_garden_expr = "COALESCE(fqs.quarantine_garden_id, f.quarantine_garden_id)"
77 quarantine_status_expr = "COALESCE(fqs.quarantine_status, f.quarantine_status)"
78 # Tenant scope: applies to ALL callers (admins included). A per-NODE admin
79 # is NOT cross-tenant — quarantine moderation is per-TENANT (R-2 / F-SBOLA5).
80 # Kept as a LITERAL leading predicate in the WHERE f-string below (NOT appended
81 # to `filters`) so the static fact-query tenant-scope guard can verify it.
82 params: list[Any] = [identity.tenant_id]
83 filters: list[str] = [f"{quarantine_garden_expr} IS NOT NULL"]
85 if quarantine_status: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true
86 filters.append(f"{quarantine_status_expr} = ?")
87 params.append(quarantine_status)
88 else:
89 filters.append(f"{quarantine_status_expr} IS NOT NULL")
91 if garden_id:
92 # Resolve slug to UUID
93 garden = get_garden_by_slug_or_id(garden_id, tenant_id=identity.tenant_id)
94 if garden is None: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true
95 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="garden not found")
96 filters.append(f"{quarantine_garden_expr} = ?")
97 params.append(garden["id"])
98 elif not identity.can_write(): 98 ↛ 100line 98 didn't jump to line 100 because the condition on line 98 was never true
99 # Non-admins: only see facts in gardens they're members of
100 filters.append(
101 f"{quarantine_garden_expr} IN (" # nosec B608
102 " SELECT gm.garden_id FROM garden_members gm"
103 " WHERE gm.entity_uri = ?"
104 ")"
105 )
106 params.append(identity.entity_uri)
108 where_clause = " AND ".join(filters)
110 with db() as conn:
111 count_row = conn.execute(
112 f"SELECT COUNT(*) FROM facts f {projection_joins}"
113 f" WHERE f.tenant_id = ? AND {where_clause}", # nosec B608
114 params,
115 ).fetchone()
116 total: int = count_row[0] if count_row else 0
118 rows = conn.execute(
119 f"""SELECT f.id, f.entity, f.relation, f.source,
120 {quarantine_status_expr} AS quarantine_status,
121 {quarantine_garden_expr} AS quarantine_garden_id,
122 COALESCE(fqs.quarantine_reason, f.quarantine_reason) AS quarantine_reason,
123 COALESCE(fqs.quarantine_acted_by, f.quarantine_acted_by)
124 AS quarantine_acted_by,
125 COALESCE(fqs.quarantine_acted_at, f.quarantine_acted_at)
126 AS quarantine_acted_at,
127 f.source_trust, f.received_from, f.timestamp
128 FROM facts f {projection_joins}
129 WHERE f.tenant_id = ? AND {where_clause}
130 ORDER BY f.timestamp DESC
131 LIMIT ? OFFSET ?""", # nosec B608
132 [*params, limit, offset],
133 ).fetchall()
135 items = [
136 QuarantineRecord(
137 fact_id=r["id"],
138 entity=r["entity"],
139 relation=r["relation"],
140 source=r["source"],
141 quarantine_status=r["quarantine_status"] or "",
142 quarantine_garden_id=r["quarantine_garden_id"],
143 quarantine_reason=r["quarantine_reason"],
144 quarantine_acted_by=r["quarantine_acted_by"],
145 quarantine_acted_at=r["quarantine_acted_at"],
146 source_trust=float(r["source_trust"]) if r["source_trust"] is not None else None,
147 received_from=r["received_from"],
148 timestamp=r["timestamp"],
149 )
150 for r in rows
151 ]
153 return QuarantineListResponse(items=items, total=total)
156# ---------------------------------------------------------------------------
157# Admit (promote) a fact from quarantine to the main fabric
158# ---------------------------------------------------------------------------
161@router.post("/{fact_id}/admit", status_code=status.HTTP_200_OK)
162def admit_fact(
163 fact_id: str,
164 identity: Annotated[Identity, Depends(resolve_identity)],
165 target_garden_id: str | None = Query(None, description="Target garden UUID or slug"),
166 reason: str = Query("", description="Reason for admission"),
167) -> dict[str, Any]:
168 """Promote a quarantined fact to the main fabric (or a specific target garden).
170 Node admins may admit quarantined facts as last-resort moderation
171 authority. Other callers require quarantine:moderator or admin role in the
172 fact's quarantine garden.
173 """
174 _require_write(identity)
176 fact_row, garden = _get_quarantined_fact(fact_id, identity)
177 now = datetime.now(UTC).isoformat()
179 # Resolve target garden
180 target_db_id: str | None = None
181 if target_garden_id:
182 tg = get_garden_by_slug_or_id(target_garden_id, tenant_id=identity.tenant_id)
183 # The helper now tenant-scopes its UUID branch, so a cross-tenant target
184 # garden resolves to None. The explicit tenant check is kept as
185 # defense-in-depth; either way reject as 404 to avoid revealing existence.
186 if tg is None or tg["tenant_id"] != identity.tenant_id: 186 ↛ 190line 186 didn't jump to line 190 because the condition on line 186 was always true
187 raise HTTPException(
188 status_code=status.HTTP_404_NOT_FOUND, detail="target garden not found"
189 )
190 target_db_id = tg["id"]
192 with db() as conn:
193 set_fact_garden_membership(
194 conn,
195 fact_id=fact_id,
196 garden_id=target_db_id,
197 updated_by=identity.entity_uri,
198 )
199 set_fact_quarantine_status(
200 conn,
201 fact_id=fact_id,
202 quarantine_garden_id=garden["id"],
203 quarantine_status="promoted",
204 quarantine_reason=reason or "admitted via admin API",
205 quarantine_acted_by=identity.entity_uri,
206 quarantine_acted_at=now,
207 )
208 _write_quarantine_audit(conn, fact_id, "quarantine_promote", identity, now)
209 emit_instruction_event_if_applicable(
210 INSTRUCTION_PROMOTED,
211 fact_id=fact_id,
212 fact_entity=fact_row["entity"],
213 fact_relation=fact_row["relation"],
214 fact_interpret_as=fact_row["interpret_as"],
215 actor_uri=identity.entity_uri,
216 tenant_id=identity.tenant_id,
217 oidc_sub=identity.oidc_sub,
218 source=identity.entity_uri,
219 detail={
220 "reason": reason or "admitted via admin API",
221 "quarantine_garden_id": garden["id"],
222 "target_garden_id": target_db_id,
223 },
224 conn=conn,
225 )
227 return {
228 "fact_id": fact_id,
229 "action": "admitted",
230 "target_garden_id": target_db_id,
231 "acted_by": identity.entity_uri,
232 "acted_at": now,
233 }
236# ---------------------------------------------------------------------------
237# Reject a quarantined fact
238# ---------------------------------------------------------------------------
241@router.post("/{fact_id}/reject", status_code=status.HTTP_200_OK)
242def reject_fact(
243 fact_id: str,
244 identity: Annotated[Identity, Depends(resolve_identity)],
245 reason: str = Query("", description="Reason for rejection"),
246) -> dict[str, Any]:
247 """Permanently reject a quarantined fact.
249 Sets confidence = 0.0 and quarantine_status = 'rejected'.
250 Node admins may reject quarantined facts as last-resort moderation
251 authority. Other callers require quarantine:moderator or admin role in the
252 fact's quarantine garden.
253 """
254 _require_write(identity)
256 _fact_row, garden = _get_quarantined_fact(fact_id, identity)
257 now = datetime.now(UTC).isoformat()
259 with db() as conn:
260 set_fact_validity_override(
261 conn,
262 fact_id=fact_id,
263 confidence=0.0,
264 reason=reason or "rejected via admin API",
265 updated_by=identity.entity_uri,
266 )
267 set_fact_quarantine_status(
268 conn,
269 fact_id=fact_id,
270 quarantine_garden_id=garden["id"],
271 quarantine_status="rejected",
272 quarantine_reason=reason or "rejected via admin API",
273 quarantine_acted_by=identity.entity_uri,
274 quarantine_acted_at=now,
275 )
276 # Append-only retraction log (§24.2.1 c.3)
277 conn.execute(
278 "INSERT INTO fact_retractions"
279 " (id, fact_id, retracted_at, retracted_by) VALUES (?,?,?,?)",
280 (str(uuid.uuid4()), fact_id, now, identity.entity_uri),
281 )
282 _write_quarantine_audit(conn, fact_id, "quarantine_reject", identity, now)
284 return {
285 "fact_id": fact_id,
286 "action": "rejected",
287 "acted_by": identity.entity_uri,
288 "acted_at": now,
289 }
292# ---------------------------------------------------------------------------
293# Helpers
294# ---------------------------------------------------------------------------
297def _get_quarantined_fact(
298 fact_id: str, identity: Identity
299) -> tuple[dict[str, Any], dict[str, Any]]:
300 """Return (fact_row, garden_row) for a pending quarantined fact.
302 Scoped to the caller's tenant: a fact in another tenant returns 404 (R-2 /
303 F-SBOLA5). Within the caller's tenant, admins are the last-resort moderation
304 authority; garden-scoped moderators must hold quarantine:moderator or admin
305 role in the fact's quarantine garden.
306 """
307 with db() as conn:
308 row = conn.execute(
309 """SELECT f.*,
310 COALESCE(fqs.quarantine_garden_id, f.quarantine_garden_id)
311 AS projected_quarantine_garden_id,
312 COALESCE(fqs.quarantine_status, f.quarantine_status)
313 AS projected_quarantine_status,
314 COALESCE(fqs.quarantine_reason, f.quarantine_reason)
315 AS projected_quarantine_reason
316 FROM facts f
317 LEFT JOIN fact_quarantine_status fqs ON fqs.fact_id = f.id
318 WHERE f.id = ?
319 AND f.tenant_id = ?
320 AND COALESCE(fqs.quarantine_garden_id, f.quarantine_garden_id) IS NOT NULL""",
321 (fact_id, identity.tenant_id),
322 ).fetchone()
324 if row is None:
325 raise HTTPException(
326 status_code=status.HTTP_404_NOT_FOUND, detail="quarantined fact not found"
327 )
329 if row["projected_quarantine_status"] != QUARANTINE_PENDING:
330 raise HTTPException(
331 status_code=status.HTTP_409_CONFLICT,
332 detail="fact_not_quarantine_pending",
333 )
335 garden = get_garden_by_slug_or_id(
336 row["projected_quarantine_garden_id"], tenant_id=identity.tenant_id
337 )
338 if garden is None: 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 raise HTTPException(
340 status_code=status.HTTP_404_NOT_FOUND, detail="quarantine garden not found"
341 )
343 if not identity.can_write(): 343 ↛ 344line 343 didn't jump to line 344 because the condition on line 343 was never true
344 require_quarantine_moderator_or_admin(garden, identity)
346 return dict(row), garden
349def _write_quarantine_audit(
350 conn: Any, fact_id: str, event_type: str, identity: Identity, now: str
351) -> None:
352 audit_id = str(uuid.uuid4())
353 conn.execute(
354 "INSERT INTO fact_audit_log"
355 " (id, fact_id, event_type, entity_uri, oidc_sub, source,"
356 " attested_key_id, ts)"
357 " VALUES (?,?,?,?,?,?,?,?)",
358 (
359 audit_id,
360 fact_id,
361 event_type,
362 identity.entity_uri,
363 identity.oidc_sub,
364 identity.entity_uri,
365 None,
366 now,
367 ),
368 )