Coverage for node / src / stigmem_node / session_graph.py: 83%

1"""Per-session read/write graph controls for R-21 feedback-loop defense.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import uuid 

7from datetime import UTC, datetime 

8from typing import Any 

9 

10from fastapi import HTTPException, status 

11 

12from .auth import Identity 

13 

14SESSION_HEADER = "Stigmem-Session" 

15SUMMARIZE_WITH_PROVENANCE = "summarize_with_provenance" 

16 

17 

18def normalize_session_id(session_id: str | None) -> str | None: 

19 """Return a bounded session id, or None when the caller did not opt in.""" 

20 if session_id is None: 

21 return None 

22 normalized = session_id.strip() 

23 if not normalized: 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true

24 return None 

25 if len(normalized) > 128: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true

26 raise HTTPException( 

27 status_code=status.HTTP_400_BAD_REQUEST, 

28 detail="session_id_too_long", 

29 ) 

30 return normalized 

31 

32 

33def _now_iso() -> str: 

34 return datetime.now(UTC).isoformat() 

35 

36 

37def record_read_scopes( 

38 conn: Any, 

39 *, 

40 identity: Identity, 

41 session_id: str | None, 

42 scopes: set[str], 

43) -> None: 

44 """Record scopes read by a caller in a session.""" 

45 normalized = normalize_session_id(session_id) 

46 if normalized is None: 

47 return 

48 now = _now_iso() 

49 for scope in scopes: 

50 conn.execute( 

51 """INSERT OR IGNORE INTO session_scope_access 

52 (id, session_id, entity_uri, tenant_id, scope, access_type, ts) 

53 VALUES (?, ?, ?, ?, ?, 'read', ?)""", 

54 ( 

55 str(uuid.uuid4()), 

56 normalized, 

57 identity.entity_uri, 

58 identity.tenant_id, 

59 scope, 

60 now, 

61 ), 

62 ) 

63 

64 

65def record_write_scope( 

66 conn: Any, 

67 *, 

68 identity: Identity, 

69 session_id: str | None, 

70 scope: str, 

71) -> None: 

72 """Record a scope written by a caller in a session.""" 

73 normalized = normalize_session_id(session_id) 

74 if normalized is None: 

75 return 

76 conn.execute( 

77 """INSERT OR IGNORE INTO session_scope_access 

78 (id, session_id, entity_uri, tenant_id, scope, access_type, ts) 

79 VALUES (?, ?, ?, ?, ?, 'write', ?)""", 

80 ( 

81 str(uuid.uuid4()), 

82 normalized, 

83 identity.entity_uri, 

84 identity.tenant_id, 

85 scope, 

86 _now_iso(), 

87 ), 

88 ) 

89 

90 

91def _read_scopes_for_session(conn: Any, *, identity: Identity, session_id: str) -> set[str]: 

92 rows = conn.execute( 

93 """SELECT scope FROM session_scope_access 

94 WHERE session_id = ? 

95 AND entity_uri = ? 

96 AND tenant_id = ? 

97 AND access_type = 'read'""", 

98 (session_id, identity.entity_uri, identity.tenant_id), 

99 ).fetchall() 

100 return {row["scope"] for row in rows} 

101 

102 

103def _provenance_scopes( 

104 conn: Any, derived_from: list[dict[str, Any]], tenant_id: str 

105) -> set[str]: 

106 scopes: set[str] = set() 

107 for entry in derived_from: 

108 fact_id = entry.get("fact_id") 

109 hash_val = entry.get("hash") 

110 row = None 

111 if fact_id: 111 ↛ 116line 111 didn't jump to line 116 because the condition on line 111 was always true

112 row = conn.execute( 

113 "SELECT scope FROM facts WHERE id = ? AND tenant_id = ?", 

114 (fact_id, tenant_id), 

115 ).fetchone() 

116 elif isinstance(hash_val, str) and hash_val.startswith("sha256:"): 

117 alias = conn.execute( 

118 "SELECT fact_id FROM fact_cid_aliases WHERE cid = ? AND tenant_id = ?", 

119 (hash_val, tenant_id), 

120 ).fetchone() 

121 if alias is not None: 

122 row = conn.execute( 

123 "SELECT scope FROM facts WHERE id = ? AND tenant_id = ?", 

124 (alias["fact_id"], tenant_id), 

125 ).fetchone() 

126 if row is not None: 126 ↛ 107line 126 didn't jump to line 107 because the condition on line 126 was always true

127 scopes.add(row["scope"]) 

128 return scopes 

129 

130 

131def ensure_write_allowed( 

132 conn: Any, 

133 *, 

134 identity: Identity, 

135 session_id: str | None, 

136 target_scope: str, 

137 write_mode: str, 

138 derived_from: list[dict[str, Any]], 

139) -> None: 

140 """Reject read-then-write same-scope loops unless provenance is carried forward.""" 

141 normalized = normalize_session_id(session_id) 

142 if normalized is None: 

143 return 

144 

145 read_scopes = _read_scopes_for_session(conn, identity=identity, session_id=normalized) 

146 if target_scope not in read_scopes: 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true

147 return 

148 

149 if write_mode == SUMMARIZE_WITH_PROVENANCE and target_scope in _provenance_scopes( 

150 conn, derived_from, identity.tenant_id 

151 ): 

152 return 

153 

154 raise HTTPException( 

155 status_code=status.HTTP_403_FORBIDDEN, 

156 detail={ 

157 "code": "feedback_loop_provenance_required", 

158 "message": ( 

159 "writes into scopes read earlier in the same session require " 

160 "write_mode='summarize_with_provenance' and source provenance" 

161 ), 

162 "session_id": normalized, 

163 "scope": target_scope, 

164 }, 

165 ) 

166 

167 

168def encode_derived_from(derived_from: list[dict[str, Any]]) -> str | None: 

169 if not derived_from: 

170 return None 

171 return json.dumps(derived_from, sort_keys=True, separators=(",", ":"))