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

1"""Opt-in gates for the experimental Memory Garden advanced ACL plugin.""" 

2 

3from __future__ import annotations 

4 

5import os 

6from logging import Logger 

7 

8from .db import db 

9from .plugins import get_registry 

10 

11PLUGIN_NAME = "stigmem-plugin-memory-garden-acl" 

12_ENV_PREFIX = "STIGMEM_MEMORY_GARDEN_ACL_" 

13_TRUE_VALUES = {"1", "true", "yes", "on"} 

14 

15 

16def _env_bool(name: str) -> bool: 

17 return os.environ.get(f"{_ENV_PREFIX}{name}", "").strip().lower() in _TRUE_VALUES 

18 

19 

20def plugin_registered() -> bool: 

21 """Return True when the experimental advanced ACL plugin is explicitly registered.""" 

22 return PLUGIN_NAME in get_registry().registered_plugins() 

23 

24 

25def _gate_enabled(flag_name: str) -> bool: 

26 return plugin_registered() and _env_bool("ENABLED") and _env_bool(flag_name) 

27 

28 

29def oidc_permission_ceiling_enabled() -> bool: 

30 """Gate membership-derived OIDC permission ceilings.""" 

31 return _gate_enabled("ENABLE_OIDC_PERMISSION_CEILING") 

32 

33 

34def recall_filter_enabled() -> bool: 

35 """Gate cross-surface garden ACL filtering for recall, graph, and subscriptions.""" 

36 return _gate_enabled("APPLY_RECALL_FILTER") 

37 

38 

39def memory_garden_acl_filtering_state() -> str: 

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

41 

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

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

44 subscription-delivery filtering are not all enabled. 

45 """ 

46 if not plugin_registered() or not _env_bool("ENABLED"): 46 ↛ 48line 46 didn't jump to line 48 because the condition on line 46 was always true

47 return "disabled" 

48 if recall_filter_enabled() and oidc_permission_ceiling_enabled(): 

49 return "enabled-full" 

50 return "enabled-partial" 

51 

52 

53def gardens_with_members_exist() -> bool: 

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

55 with db() as conn: 

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

57 return row is not None 

58 

59 

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

61 """Warn once at startup when gardens exist but advanced ACL filtering is off.""" 

62 if plugin_registered() or not gardens_with_members_exist(): 

63 return 

64 logger.warning( 

65 "SECURITY WARNING: Garden ACL filtering is disabled " 

66 "(stigmem-plugin-memory-garden-acl not registered). Restricted gardens " 

67 "do not filter tenant-wide queries, recall ranking, push subscriptions, " 

68 "or graph traversal. Install and enable the plugin, or accept the " 

69 "documented opt-in Memory Garden ACL posture." 

70 )