Coverage for node / src / stigmem_node / garden_acl.py: 91%
57 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Garden ACL enforcement — spec §17.3, §19.5.3.
3Gardens are named, ACL'd partitions above scope (v0.9).
4ACL is checked at fact read and write time in addition to scope enforcement.
5Quarantine gardens extend this with the quarantine:moderator role (v1.1).
6"""
8from __future__ import annotations
10from typing import Any
12from fastapi import HTTPException, status
14from .auth import Identity
15from .db import db
18def get_garden_by_slug_or_id(
19 slug_or_id: str, tenant_id: str | None = None
20) -> dict[str, Any] | None:
21 """Return a garden row by slug or by its UUID id.
23 When tenant_id is provided, BOTH the slug lookup and the UUID lookup are
24 scoped to that tenant. Slug scoping lets the same slug used by different
25 tenants resolve to the correct garden; UUID scoping prevents a caller from
26 resolving (and then linking a fact into) another tenant's garden by passing
27 its globally-unique raw UUID. No caller legitimately needs cross-tenant UUID
28 resolution, so scoping the UUID branch here closes that class for every
29 call site at once (fact-target operations such as quarantine admit/promote
30 in particular). When tenant_id is None (no multi-tenant context) the lookup
31 is unscoped.
32 """
33 with db() as conn:
34 if tenant_id is not None: 34 ↛ 42line 34 didn't jump to line 42 because the condition on line 34 was always true
35 # Both slug and UUID lookups are scoped to the caller's tenant.
36 row = conn.execute(
37 "SELECT * FROM gardens"
38 " WHERE (slug = ? AND tenant_id = ?) OR (id = ? AND tenant_id = ?)",
39 (slug_or_id, tenant_id, slug_or_id, tenant_id),
40 ).fetchone()
41 else:
42 row = conn.execute(
43 "SELECT * FROM gardens WHERE slug = ? OR id = ?",
44 (slug_or_id, slug_or_id),
45 ).fetchone()
46 return dict(row) if row is not None else None
49def get_garden_by_garden_uri(
50 garden_uri: str, tenant_id: str | None = None
51) -> dict[str, Any] | None:
52 """Return a garden row by its stigmem://authority/garden/{slug} URI."""
53 # Extract slug from URI: stigmem://authority/garden/{slug}
54 parts = garden_uri.split("/garden/", 1)
55 if len(parts) != 2 or not parts[1]: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true
56 return None
57 slug = parts[1].rstrip("/")
58 return get_garden_by_slug_or_id(slug, tenant_id=tenant_id)
61def get_member_role(garden_id: str, entity_uri: str) -> str | None:
62 """Return the role of entity_uri in the given garden UUID, or None if not a member."""
63 with db() as conn:
64 row = conn.execute(
65 "SELECT role FROM garden_members WHERE garden_id = ? AND entity_uri = ?",
66 (garden_id, entity_uri),
67 ).fetchone()
68 return row["role"] if row is not None else None
71def require_garden_write(garden: dict[str, Any], identity: Identity) -> None:
72 """Raise 403 if identity cannot write facts into this garden (spec §17.3)."""
73 role = get_member_role(garden["id"], identity.entity_uri)
74 if role not in ("admin", "writer"):
75 if role == "reader":
76 raise HTTPException(
77 status_code=status.HTTP_403_FORBIDDEN,
78 detail="write permission required — you are a reader in this garden",
79 )
80 raise HTTPException(
81 status_code=status.HTTP_403_FORBIDDEN,
82 detail="not a member of this garden",
83 )
86def require_garden_read(garden: dict[str, Any], identity: Identity) -> None:
87 """Raise 403 if identity cannot read facts from this garden (spec §17.3)."""
88 role = get_member_role(garden["id"], identity.entity_uri)
89 if role is None:
90 raise HTTPException(
91 status_code=status.HTTP_403_FORBIDDEN,
92 detail="not a member of this garden",
93 )
96def require_garden_admin(garden: dict[str, Any], identity: Identity) -> None:
97 """Raise 403 if identity is not an admin of this garden."""
98 role = get_member_role(garden["id"], identity.entity_uri)
99 if role != "admin":
100 raise HTTPException(
101 status_code=status.HTTP_403_FORBIDDEN,
102 detail="garden admin permission required",
103 )
106def caller_can_see_garden(garden_id: str, identity: Identity) -> bool:
107 """Return True if identity holds any role in the garden (for query-time filtering)."""
108 role = get_member_role(garden_id, identity.entity_uri)
109 return role is not None
112def require_fact_garden_read(conn: Any, fact_id: str, tenant_id: str, identity: Identity) -> None:
113 """Raise 404 if a fact-by-id is in a (projected) garden the caller cannot see.
115 Shared gate for fact-by-id read surfaces (single get, provenance, cid verify)
116 so a restricted-garden fact's existence/content/lineage stays hidden from
117 same-tenant non-members (spec §17.3). Uses the PROJECTED garden
118 ``COALESCE(fact_garden_membership.garden_id, facts.garden_id)`` — a fact
119 promoted into a garden has raw ``garden_id`` NULL. Hides as 404 (not 403) so
120 existence is not revealed. No-op for garden-less facts or unknown ids.
121 """
122 row = conn.execute(
123 "SELECT COALESCE(fgm.garden_id, f.garden_id) AS gid FROM facts f"
124 " LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = f.id"
125 " WHERE f.id = ? AND f.tenant_id = ?",
126 (fact_id, tenant_id),
127 ).fetchone()
128 if row is None or row["gid"] is None:
129 return
130 if not caller_can_see_garden(row["gid"], identity): 130 ↛ exitline 130 didn't return from function 'require_fact_garden_read' because the condition on line 130 was always true
131 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="fact not found")
134def is_node_admin(identity: Identity) -> bool:
135 """Node admin: any identity with write permission (spec §5.15)."""
136 return identity.can_write()
139def require_quarantine_moderator_or_admin(garden: dict[str, Any], identity: Identity) -> None:
140 """Raise 403 if identity cannot promote/reject quarantined facts (spec §19.5.3).
142 This helper checks garden-scoped moderation roles only: 'admin' or
143 'quarantine:moderator' in the quarantine garden membership table.
144 Route-level callers intentionally allow node admins through before this
145 helper runs. Node-admin bypass is intentional because node admins are the
146 system's last-resort moderation authority; garden-scoped moderators must
147 still be members of the specific quarantine garden.
148 """
149 role = get_member_role(garden["id"], identity.entity_uri)
150 if role not in ("admin", "quarantine:moderator"):
151 raise HTTPException(
152 status_code=status.HTTP_403_FORBIDDEN,
153 detail="quarantine:moderator or admin role required to promote/reject facts",
154 )
157def has_elevated_quarantine_role(garden: dict[str, Any], identity: Identity) -> bool:
158 """True if identity holds admin or quarantine:moderator in a quarantine garden."""
159 role = get_member_role(garden["id"], identity.entity_uri)
160 return role in ("admin", "quarantine:moderator")
163def quarantine_garden_has_pending_facts(garden_uuid: str) -> bool:
164 """True if the quarantine garden holds at least one fact with quarantine_status='pending'."""
165 with db() as conn:
166 row = conn.execute(
167 "SELECT f.id FROM facts f"
168 " LEFT JOIN fact_quarantine_status fqs ON fqs.fact_id = f.id"
169 " WHERE COALESCE(fqs.quarantine_garden_id, f.quarantine_garden_id) = ?"
170 " AND COALESCE(fqs.quarantine_status, f.quarantine_status) = 'pending'"
171 " LIMIT 1",
172 (garden_uuid,),
173 ).fetchone()
174 return row is not None