Coverage for src/local_deep_research/security/egress/classification.py: 100%

1"""Two-axis egress data classification — ADR-0007, rollout stage A. 

2 

3Pure decision logic. Each model a run touches (search engine, collection / 

4store, LLM, embeddings) is classified on two **orthogonal** axes: 

5 

6* **Sensitivity** — of the data the component *sources*: must it never leave the 

7 machine? (``SENSITIVE`` / ``NON_SENSITIVE``) 

8* **Exposure** — of the *sink* the component represents: does sending data to it 

9 put that data off-machine? (``EXPOSING`` / ``CONTAINED``) 

10 

11:func:`evaluate_run` decides whether a *set* of classified components may run 

12together, per the four-quadrant rule in ADR-0007. The core invariant: **a run 

13may not let a SENSITIVE source reach an EXPOSING sink.** Because an agentic run 

14can turn one engine's results into another engine's query (the LangGraph 

15silent-expansion bug class already in our threat model), the rule is evaluated 

16over the whole component set, not per individual call. 

17 

18This module holds the pure decision logic. It is consumed via 

19``run_classification``, which resolves live engines/providers, assembles a run, 

20and — at the run-start precheck — **enforces** the decision (rejecting a run 

21that would let a sensitive source reach an exposing sink), except under the 

22``UNPROTECTED`` escape hatch where it evaluates permissive. See 

23``docs/decisions/0007-egress-two-axis-data-classification.md``. 

24""" 

25 

26from __future__ import annotations 

27 

28from dataclasses import dataclass, field 

29from enum import Enum 

30from typing import Iterable 

31 

32 

33class Sensitivity(Enum): 

34 """Whether the data a component *sources* may leave the machine.""" 

35 

36 NON_SENSITIVE = "non_sensitive" 

37 SENSITIVE = "sensitive" 

38 

39 

40class Exposure(Enum): 

41 """Whether sending data to a component (as a sink) puts it off-machine.""" 

42 

43 CONTAINED = "contained" 

44 EXPOSING = "exposing" 

45 

46 

47class Role(Enum): 

48 """How a component participates in a run's data flow. 

49 

50 A single real component may take more than one role — a search engine is a 

51 ``SEARCH_SINK`` for the query *and* a ``SOURCE`` of results; Elasticsearch 

52 is a ``SOURCE`` of data *and* a ``SEARCH_SINK`` for the query you send it. 

53 Such a component is passed to :func:`evaluate_run` once per role it plays, 

54 **using the same** :attr:`Component.name` each time — the self-exclusion in 

55 quadrant 4 (a dual-risk store used solo) relies on that shared identity. 

56 """ 

57 

58 SOURCE = "source" # contributes data into the run 

59 SEARCH_SINK = "search_sink" # receives the query (+ maybe expanded results) 

60 INFERENCE_SINK = "inference_sink" # LLM / embeddings — sees all source data 

61 

62 

63class Mode(Enum): 

64 """Enforcement mode — SELinux vocabulary (ADR-0007).""" 

65 

66 ENFORCING = "enforcing" # the invariant is enforced; violations blocked 

67 PERMISSIVE = "permissive" # evaluated for warnings, never blocks 

68 

69 

70@dataclass(frozen=True) 

71class Label: 

72 """A component's position on the two axes.""" 

73 

74 sensitivity: Sensitivity 

75 exposure: Exposure 

76 

77 

78@dataclass(frozen=True) 

79class Component: 

80 """A classified participant in a run. 

81 

82 ``name`` is a stable identifier (engine name, ``collection_<uuid>``, 

83 ``llm:<provider>``, ``embeddings:<provider>``) used for self-exclusion and 

84 reporting. 

85 """ 

86 

87 name: str 

88 role: Role 

89 label: Label 

90 

91 

92@dataclass(frozen=True) 

93class Decision: 

94 """Outcome of :func:`evaluate_run`.""" 

95 

96 allowed: bool 

97 reason: str 

98 offending: tuple[str, ...] = field(default_factory=tuple) 

99 

100 

101def evaluate_run( 

102 components: Iterable[Component], *, mode: Mode = Mode.ENFORCING 

103) -> Decision: 

104 """Decide whether a set of classified components may run together. 

105 

106 Implements the four-quadrant rule of ADR-0007: 

107 

108 1. ``non-sensitive, contained`` → combines with anything. 

109 2. ``sensitive, contained`` → only with other contained sources. 

110 3. ``non-sensitive, exposing`` → only with non-sensitive sources. 

111 4. ``sensitive + exposing`` → solo, with contained inference. 

112 

113 The rule assumes the caller passes each real engine as BOTH a ``SOURCE`` 

114 (carrying its sensitivity) and a ``SEARCH_SINK`` (carrying its exposure) 

115 under one shared ``name`` — the quadrant-4 self-exclusion depends on that 

116 pairing (see ``run_classification.classify_run``). 

117 

118 Under :attr:`Mode.PERMISSIVE` the run is always allowed, but the underlying 

119 would-be-enforced ``reason`` / ``offending`` are preserved (prefixed 

120 ``permissive:``) so the caller can still surface a warning banner. 

121 """ 

122 components = tuple(components) 

123 decision = _decide_enforcing(components) 

124 if mode is Mode.PERMISSIVE and not decision.allowed: 

125 return Decision( 

126 True, f"permissive:{decision.reason}", offending=decision.offending 

127 ) 

128 return decision 

129 

130 

131def _decide_enforcing(components: tuple[Component, ...]) -> Decision: 

132 """The enforcing four-quadrant decision (mode-agnostic).""" 

133 sensitive_names = { 

134 c.name 

135 for c in components 

136 if c.role is Role.SOURCE 

137 and c.label.sensitivity is Sensitivity.SENSITIVE 

138 } 

139 

140 # No sensitive data in the run → nothing to protect; any sink is fine. 

141 if not sensitive_names: 

142 return Decision(True, "no_sensitive_source") 

143 

144 # (1) Sensitive content always reaches the inference sink (the LLM/embeddings 

145 # see every source's data), so an exposing inference sink leaks it. 

146 exposing_inference = tuple( 

147 sorted( 

148 c.name 

149 for c in components 

150 if c.role is Role.INFERENCE_SINK 

151 and c.label.exposure is Exposure.EXPOSING 

152 ) 

153 ) 

154 if exposing_inference: 

155 return Decision( 

156 False, 

157 "sensitive_to_exposing_inference", 

158 offending=exposing_inference, 

159 ) 

160 

161 # (2) An exposing search sink may coexist with sensitive data ONLY if it is 

162 # itself the sole sensitive source (quadrant 4, solo): returning its own 

163 # data to itself is not a new leak. Any *other* sensitive source could be 

164 # agentically expanded into a query to that sink → leak. The sink must 

165 # ALSO be sensitive itself, so a mere name collision between a 

166 # non-sensitive exposing sink and an unrelated sensitive source cannot 

167 # be mistaken for the solo dual-risk case. 

168 offending_search = tuple( 

169 sorted( 

170 { 

171 sink.name 

172 for sink in components 

173 if sink.role is Role.SEARCH_SINK 

174 and sink.label.exposure is Exposure.EXPOSING 

175 and ( 

176 (sensitive_names - {sink.name}) 

177 or sink.label.sensitivity is not Sensitivity.SENSITIVE 

178 ) 

179 } 

180 ) 

181 ) 

182 if offending_search: 

183 return Decision( 

184 False, "sensitive_to_exposing_search", offending=offending_search 

185 ) 

186 

187 return Decision(True, "sensitive_contained")