Coverage for node / src / stigmem_node / routes / subscriptions.py: 88%
117 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Subscription CRUD and replay-window routes — spec §20.3, §20.5.
3POST /v1/subscriptions — create (subscriber_identity = caller; BOLA §20.3.2)
4GET /v1/subscriptions — list caller's subscriptions
5GET /v1/subscriptions/{id} — get one (BOLA: 404 if not owner)
6DELETE /v1/subscriptions/{id} — cancel (BOLA: 404 if not owner)
7GET /v1/subscriptions/{id}/events — replay window (§20.5)
8"""
10from __future__ import annotations
12import json
13import uuid
14from datetime import UTC, datetime, timedelta
15from typing import Annotated, Any
17from fastapi import APIRouter, Depends, HTTPException, Query, status
19from .. import settings as _settings_pkg
20from ..auth import Identity, resolve_identity
21from ..db import db
22from ..garden_acl import get_garden_by_garden_uri, require_garden_read
23from ..models.constants import VALID_SCOPES
24from ..models.subscriptions import (
25 SubscriptionCreateRequest,
26 SubscriptionEventRecord,
27 SubscriptionEventsResponse,
28 SubscriptionListResponse,
29 SubscriptionRecord,
30)
31from ..net_util import assert_safe_url
32from ..subscription_delivery import _sanitize_payload as _sanitize_event_payload
34router = APIRouter(prefix="/v1/subscriptions", tags=["subscriptions"])
37def _target_kind(target: str) -> str:
38 return "scope" if target in VALID_SCOPES else "entity"
41def _row_to_record(row: Any) -> SubscriptionRecord:
42 return SubscriptionRecord(
43 id=row["id"],
44 subscriber_identity=row["subscriber_identity"],
45 target=row["target"],
46 target_kind=row["target_kind"],
47 on_change=row["on_change"],
48 delivery_address=row["delivery_address"],
49 idempotency_key=row["idempotency_key"],
50 created_at=row["created_at"],
51 last_delivered_at=row["last_delivered_at"],
52 circuit_open=bool(row["circuit_open"]),
53 consecutive_failures=row["consecutive_failures"],
54 )
57def _event_row_to_record(row: Any) -> SubscriptionEventRecord:
58 return SubscriptionEventRecord(
59 id=row["id"],
60 subscription_id=row["subscription_id"],
61 event_type=row["event_type"],
62 entity_uri=row["entity_uri"],
63 fact_id=row["fact_id"],
64 payload=json.loads(row["payload"]),
65 created_at=row["created_at"],
66 delivered_at=row["delivered_at"],
67 delivery_status=row["delivery_status"],
68 delivery_attempts=row["delivery_attempts"],
69 )
72def _event_row_to_record_with_payload(row: Any, payload_json: str) -> SubscriptionEventRecord:
73 """Like _event_row_to_record but substitutes a pre-sanitized payload JSON string."""
74 return SubscriptionEventRecord(
75 id=row["id"],
76 subscription_id=row["subscription_id"],
77 event_type=row["event_type"],
78 entity_uri=row["entity_uri"],
79 fact_id=row["fact_id"],
80 payload=json.loads(payload_json),
81 created_at=row["created_at"],
82 delivered_at=row["delivered_at"],
83 delivery_status=row["delivery_status"],
84 delivery_attempts=row["delivery_attempts"],
85 )
88# ---------------------------------------------------------------------------
89# CRUD
90# ---------------------------------------------------------------------------
93@router.post("", response_model=SubscriptionRecord, status_code=status.HTTP_201_CREATED)
94def create_subscription(
95 req: SubscriptionCreateRequest,
96 identity: Annotated[Identity, Depends(resolve_identity)],
97) -> SubscriptionRecord:
98 """Create a subscription (Spec-X7-Subscriptions).
100 subscriber_identity is always the caller (BOLA).
101 """
102 if not identity.can_read(): 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 raise HTTPException(
104 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required"
105 )
107 # §17 garden ACL: if target is a garden URI, caller must be a member
108 if req.target.startswith("stigmem://") and "/garden/" in req.target: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true
109 garden = get_garden_by_garden_uri(req.target, tenant_id=identity.tenant_id)
110 if garden is None:
111 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="garden not found")
112 require_garden_read(garden, identity)
114 # GHSA-5p3m-vhh6-9236: validate a webhook delivery_address BEFORE persisting,
115 # so an unsafe URL is rejected (400) rather than stored-then-blocked-at-delivery.
116 # https-only by default; http requires the explicit operator opt-in. A `wake`
117 # subscription's delivery_address is an identity URI (not a fetchable URL), so
118 # it is deliberately NOT subjected to the SSRF guard.
119 if req.on_change == "webhook":
120 try:
121 assert_safe_url(
122 req.delivery_address,
123 allow_schemes=_settings_pkg.settings.webhook_allowed_schemes,
124 )
125 except ValueError as exc:
126 raise HTTPException(
127 status_code=status.HTTP_400_BAD_REQUEST,
128 detail=f"unsafe webhook delivery_address: {exc}",
129 ) from exc
131 # Idempotency key: return existing subscription if key matches THIS caller.
132 # Scoped to subscriber and tenant to prevent cross-entity metadata leakage (R2).
133 if req.idempotency_key:
134 with db() as conn:
135 existing = conn.execute(
136 """SELECT * FROM subscriptions
137 WHERE idempotency_key=? AND tenant_id=? AND subscriber_identity=?""",
138 (req.idempotency_key, identity.tenant_id, identity.entity_uri),
139 ).fetchone()
140 if existing is not None:
141 return _row_to_record(existing)
143 # Natural dedup: same (subscriber, target, on_change, delivery_address) → return existing
144 with db() as conn:
145 dupe = conn.execute(
146 """SELECT * FROM subscriptions
147 WHERE subscriber_identity=? AND target=? AND on_change=?
148 AND delivery_address=? AND tenant_id=?""",
149 (
150 identity.entity_uri,
151 req.target,
152 req.on_change,
153 req.delivery_address,
154 identity.tenant_id,
155 ),
156 ).fetchone()
157 if dupe is not None:
158 return _row_to_record(dupe)
160 # F-AVAIL-3: cap active subscriptions per principal to bound fan-out/DoS.
161 _sub_cap = _settings_pkg.settings.max_subscriptions_per_principal
162 if _sub_cap > 0:
163 with db() as conn:
164 held = conn.execute(
165 "SELECT COUNT(*) AS n FROM subscriptions "
166 "WHERE subscriber_identity=? AND tenant_id=?",
167 (identity.entity_uri, identity.tenant_id),
168 ).fetchone()["n"]
169 if held >= _sub_cap:
170 raise HTTPException(
171 status_code=status.HTTP_429_TOO_MANY_REQUESTS,
172 detail=f"subscription limit reached ({_sub_cap})",
173 )
175 sub_id = str(uuid.uuid4())
176 now = datetime.now(UTC).isoformat()
177 target_kind = _target_kind(req.target)
179 with db() as conn:
180 conn.execute(
181 """INSERT INTO subscriptions
182 (id, subscriber_identity, target, target_kind, on_change,
183 delivery_address, idempotency_key, created_at, tenant_id)
184 VALUES (?,?,?,?,?,?,?,?,?)""",
185 (
186 sub_id,
187 identity.entity_uri,
188 req.target,
189 target_kind,
190 req.on_change,
191 req.delivery_address,
192 req.idempotency_key,
193 now,
194 identity.tenant_id,
195 ),
196 )
197 row = conn.execute("SELECT * FROM subscriptions WHERE id=?", (sub_id,)).fetchone()
199 return _row_to_record(row)
202@router.get("", response_model=SubscriptionListResponse)
203def list_subscriptions(
204 identity: Annotated[Identity, Depends(resolve_identity)],
205) -> SubscriptionListResponse:
206 """List the authenticated caller's subscriptions (BOLA: own only)."""
207 if not identity.can_read(): 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 raise HTTPException(
209 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required"
210 )
212 with db() as conn:
213 rows = conn.execute(
214 """SELECT * FROM subscriptions
215 WHERE subscriber_identity=? AND tenant_id=?
216 ORDER BY created_at DESC""",
217 (identity.entity_uri, identity.tenant_id),
218 ).fetchall()
220 return SubscriptionListResponse(
221 subscriptions=[_row_to_record(r) for r in rows],
222 total=len(rows),
223 )
226@router.get("/{subscription_id}", response_model=SubscriptionRecord)
227def get_subscription(
228 subscription_id: str,
229 identity: Annotated[Identity, Depends(resolve_identity)],
230) -> SubscriptionRecord:
231 """Get a subscription by ID (BOLA: returns 404 if caller does not own it)."""
232 if not identity.can_read(): 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true
233 raise HTTPException(
234 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required"
235 )
237 with db() as conn:
238 row = conn.execute(
239 "SELECT * FROM subscriptions WHERE id=? AND tenant_id=?",
240 (subscription_id, identity.tenant_id),
241 ).fetchone()
243 if row is None or row["subscriber_identity"] != identity.entity_uri:
244 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="subscription not found")
246 return _row_to_record(row)
249@router.delete("/{subscription_id}", status_code=status.HTTP_204_NO_CONTENT)
250def delete_subscription(
251 subscription_id: str,
252 identity: Annotated[Identity, Depends(resolve_identity)],
253) -> None:
254 """Cancel a subscription (BOLA: only the owner may delete)."""
255 if not identity.can_read(): 255 ↛ 256line 255 didn't jump to line 256 because the condition on line 255 was never true
256 raise HTTPException(
257 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required"
258 )
260 with db() as conn:
261 row = conn.execute(
262 "SELECT id, subscriber_identity FROM subscriptions WHERE id=? AND tenant_id=?",
263 (subscription_id, identity.tenant_id),
264 ).fetchone()
266 if row is None or row["subscriber_identity"] != identity.entity_uri:
267 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="subscription not found")
269 with db() as conn:
270 conn.execute("DELETE FROM subscriptions WHERE id=?", (subscription_id,))
273# ---------------------------------------------------------------------------
274# Replay window
275# ---------------------------------------------------------------------------
278@router.get("/{subscription_id}/events", response_model=SubscriptionEventsResponse)
279def list_subscription_events(
280 subscription_id: str,
281 identity: Annotated[Identity, Depends(resolve_identity)],
282 since: str | None = Query(
283 None,
284 description="ISO 8601 timestamp; return events created at or after this time",
285 ),
286 cursor: str | None = Query(None, description="Opaque pagination cursor (event id)"),
287 limit: int = Query(50, ge=1, le=500),
288) -> SubscriptionEventsResponse:
289 """Replay window: return delivery events for a subscription (Spec-X7-Subscriptions).
291 Results are bounded to the configured replay window (default 24 h).
292 """
293 if not identity.can_read(): 293 ↛ 294line 293 didn't jump to line 294 because the condition on line 293 was never true
294 raise HTTPException(
295 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required"
296 )
298 with db() as conn:
299 sub_row = conn.execute(
300 "SELECT id, subscriber_identity FROM subscriptions WHERE id=? AND tenant_id=?",
301 (subscription_id, identity.tenant_id),
302 ).fetchone()
304 if sub_row is None or sub_row["subscriber_identity"] != identity.entity_uri:
305 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="subscription not found")
307 replay_s = _settings_pkg.settings.subscription_replay_s
308 window_cutoff = (datetime.now(UTC) - timedelta(seconds=replay_s)).isoformat()
309 effective_since = max(since, window_cutoff) if since is not None else window_cutoff
311 if cursor:
312 sql = (
313 "SELECT rowid, * FROM subscription_events "
314 "WHERE subscription_id = ? AND created_at >= ? AND rowid > ? "
315 "ORDER BY rowid ASC LIMIT ?"
316 )
317 params: list[Any] = [subscription_id, effective_since, int(cursor)]
318 else:
319 sql = (
320 "SELECT rowid, * FROM subscription_events "
321 "WHERE subscription_id = ? AND created_at >= ? "
322 "ORDER BY rowid ASC LIMIT ?"
323 )
324 params = [subscription_id, effective_since]
325 params.append(limit + 1)
327 with db() as conn:
328 rows = conn.execute(sql, params).fetchall()
330 has_more = len(rows) > limit
331 rows = rows[:limit]
332 next_cursor = str(rows[-1]["rowid"]) if has_more and rows else None
334 # S3 fix: re-apply §17 garden ACL and §19 sanitizer at replay time.
335 # subscription_events.payload stores the raw pre-delivery payload; without this
336 # re-check a subscriber could retrieve garden-scoped facts they've since been
337 # removed from, or sanitizer-redacted content, via the replay window.
338 event_ctx = {
339 "subscriber_identity": sub_row["subscriber_identity"],
340 "tenant_id": identity.tenant_id,
341 }
342 safe_events: list[SubscriptionEventRecord] = []
343 for r in rows:
344 raw_payload = json.loads(r["payload"])
345 sanitized = _sanitize_event_payload(event_ctx, raw_payload)
346 if sanitized is None:
347 # ACL/sanitizer blocked — return redacted placeholder
348 safe_payload = json.dumps({"fact_id": raw_payload.get("id"), "redacted": True})
349 else:
350 safe_payload = json.dumps(sanitized)
351 safe_events.append(_event_row_to_record_with_payload(r, safe_payload))
353 return SubscriptionEventsResponse(
354 events=safe_events,
355 total=len(safe_events),
356 cursor=next_cursor,
357 )