Coverage for node / src / stigmem_node / memory_garden_acl_gate.py: 91%

1"""Garden ACL recall filtering (graduated to core) + remaining experimental gates.""" 

2 

3from __future__ import annotations 

4 

5from logging import Logger 

6from typing import Any 

7 

8from .db import db 

9 

10 

11def _live_settings() -> Any: 

12 import sys 

13 

14 return sys.modules["stigmem_node.settings"].settings 

15 

16 

17def oidc_permission_ceiling_enabled() -> bool: 

18 """Garden-membership-derived OIDC permission ceiling — graduated to core. 

19 

20 Off by default (see ``settings.oidc_permission_ceiling``): enabling it caps 

21 OIDC-issued permissions to what the caller's garden memberships grant. 

22 """ 

23 return bool(_live_settings().oidc_permission_ceiling) 

24 

25 

26def recall_filter_enabled() -> bool: 

27 """Cross-surface garden ACL recall filtering — graduated to core (default-on). 

28 

29 Closes the cross-garden read leak (F-CONF-1): tenant-wide recall, query, graph 

30 traversal, and subscription delivery are restricted to gardens the caller is a 

31 member of. Single-tenant installs are unaffected (their facts have 

32 ``garden_id`` NULL). Opt out via ``STIGMEM_MEMORY_GARDEN_ACL_RECALL_FILTER=false``. 

33 """ 

34 return bool(_live_settings().memory_garden_acl_recall_filter) 

35 

36 

37def memory_garden_acl_filtering_state() -> str: 

38 """Return the operator-visible advanced ACL filtering posture. 

39 

40 ``disabled`` means default core behavior is active: direct garden reads and 

41 writes are guarded, but tenant-wide query, recall, graph, OIDC ceiling, and 

42 subscription-delivery filtering are not all enabled. 

43 """ 

44 if not garden_acl_enforced(): 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true

45 return "disabled" 

46 if oidc_permission_ceiling_enabled(): 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true

47 return "enabled-full" 

48 return "enabled-partial" 

49 

50 

51def gardens_with_members_exist() -> bool: 

52 """Return True when at least one garden membership row exists.""" 

53 with db() as conn: 

54 row = conn.execute("SELECT 1 FROM garden_members LIMIT 1").fetchone() 

55 return row is not None 

56 

57 

58def garden_acl_enforced() -> bool: 

59 """True when the garden access boundary must be enforced on read surfaces. 

60 

61 Fail-closed: enforced when the operator flag is on OR — regardless of the 

62 flag — whenever any garden-with-members exists. The flag can never *disable* 

63 the boundary once gardens exist; it can only be a no-op when there is nothing 

64 to protect (no garden-with-members → every fact's ``garden_id`` is NULL, so 

65 filtering changes nothing). Call once per request, not per fact. 

66 """ 

67 if bool(_live_settings().memory_garden_acl_recall_filter): 

68 return True 

69 return gardens_with_members_exist() 

70 

71 

72def caller_visible_gardens(identity: Any) -> frozenset[str]: 

73 """Garden ids the caller is a member of — one query, for in-memory filtering. 

74 

75 Lets callers batch the membership check (O(1) per fact against this set) 

76 instead of one DB lookup per candidate fact, so there is no performance 

77 reason to ever disable the boundary. 

78 """ 

79 entity_uri = getattr(identity, "entity_uri", None) 

80 if entity_uri is None: 

81 return frozenset() 

82 with db() as conn: 

83 rows = conn.execute( 

84 "SELECT garden_id FROM garden_members WHERE entity_uri = ?", 

85 (entity_uri,), 

86 ).fetchall() 

87 return frozenset(row["garden_id"] for row in rows) 

88 

89 

90def warn_if_memory_garden_acl_filtering_disabled(logger: Logger) -> None: 

91 """Inform at startup when the disable flag is set but ACL stays enforced. 

92 

93 The flag cannot create a leak: once gardens-with-members exist the boundary 

94 is enforced regardless (see ``garden_acl_enforced``). This logs that the 

95 operator's opt-out is being overridden for safety, rather than warning of a 

96 leak that can no longer happen. 

97 """ 

98 flag_off = not bool(_live_settings().memory_garden_acl_recall_filter) 

99 if not flag_off or not gardens_with_members_exist(): 

100 return 

101 logger.warning( 

102 "Garden ACL recall filtering flag is OFF " 

103 "(STIGMEM_MEMORY_GARDEN_ACL_RECALL_FILTER=false) but gardens with members " 

104 "exist, so the garden access boundary remains ENFORCED on recall, query, " 

105 "graph, and subscription delivery (fail-closed: the flag cannot disable it " 

106 "once gardens exist). Remove the override to silence this notice." 

107 )