Coverage for src/local_deep_research/security/egress/run_classification.py: 97%
88 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Stage B (ADR-0007): bridge the pure classification core to live engines and
2providers, assemble a run's component set, and decide its admissibility.
4Resolution reads each element's *declared* two-axis label (the class attributes
5added in stage A) and applies the dynamic refinements that depend on
6configuration:
8 * a self-hosted store's exposure flips to ``EXPOSING`` when its configured
9 URL resolves to a public host (reusing the policy's fail-up classifier), so
10 a Paperless/Elasticsearch holding private data on a public endpoint becomes
11 quadrant 4 (sensitive + exposing);
12 * a collection's sensitivity flips to ``NON_SENSITIVE`` when it is marked
13 public; the aggregate ``library`` stays sensitive.
15``audit_run`` computes and LOGS the four-quadrant decision and returns it; the
16run-start precheck AND the worker chokepoint ENFORCE on that result (rejecting a
17denied run) — except under the ``UNPROTECTED`` escape hatch, where ``audit_run``
18evaluates in permissive mode (always allowed, diagnostic preserved).
19``audit_run`` never raises: if the decision cannot be computed it **fails
20closed**, returning a denying ``Decision`` (reason ``audit_error``) so the run
21is refused visibly rather than silently degrading to the scope PEPs.
22"""
24from __future__ import annotations
26from typing import Iterable, List, Optional
28from loguru import logger
30from .classification import (
31 Component,
32 Decision,
33 Exposure,
34 Label,
35 Mode,
36 Role,
37 Sensitivity,
38 evaluate_run,
39)
42def engine_label(engine_name: str, settings_snapshot, ctx) -> Optional[Label]:
43 """Resolve a search engine's two-axis label, or ``None`` if unknown.
45 Reads the engine class's declared ``egress_sensitivity`` /
46 ``egress_exposure`` and applies the configuration-dependent refinements.
47 """
48 from .policy import (
49 _classify_engine_url,
50 _get_engine_class,
51 _resolve_collection_is_public,
52 )
54 # Collections / library: the class is not a static engine, and sensitivity
55 # comes from the per-collection is_public flag (public -> non-sensitive).
56 if engine_name == "library" or engine_name.startswith("collection_"):
57 is_public = _resolve_collection_is_public(engine_name, ctx.username)
58 sensitivity = (
59 Sensitivity.NON_SENSITIVE if is_public else Sensitivity.SENSITIVE
60 )
61 # A collection is a local store: contained. (It never gains a public
62 # URL the way Paperless/Elasticsearch can.)
63 return Label(sensitivity, Exposure.CONTAINED)
65 cls = _get_engine_class(engine_name)
66 if cls is None:
67 return None # unknown engine — the per-engine PEP still guards it
69 sensitivity = getattr(cls, "egress_sensitivity", Sensitivity.SENSITIVE)
70 exposure = getattr(cls, "egress_exposure", Exposure.EXPOSING)
72 # Exposure fail-up: a contained store whose configured URL resolves to a
73 # public host leaks the query off the box -> EXPOSING. Only ever tightens,
74 # mirroring the policy's asymmetric URL override.
75 url_setting = getattr(cls, "url_setting", None)
76 if url_setting and exposure is Exposure.CONTAINED:
77 if _classify_engine_url(url_setting, settings_snapshot, ctx) is False:
78 exposure = Exposure.EXPOSING
80 # Per-destination trust (ADR-0007 stage D): the user has explicitly vouched
81 # for this engine's endpoint, so treat it as contained (moves a dual-risk
82 # store from quadrant 4 to quadrant 2). Only ever applies to a local-nature
83 # store that failed up on its URL — NEVER to an inherently-public engine
84 # (is_public), so a trusted name can't launder a public search sink.
85 if (
86 exposure is Exposure.EXPOSING
87 and getattr(cls, "is_public", False) is not True
88 and engine_name.lower()
89 in _trusted_names("policy.trusted_search_engines", settings_snapshot)
90 ):
91 exposure = Exposure.CONTAINED
93 return Label(sensitivity, exposure)
96def _require_local_probe(ctx):
97 """A require-local EgressContext used to ask the enforcing PEPs whether a
98 provider's configured endpoint is local. Copies the caller's hostname
99 allow-list / username so classification matches the real run."""
100 from .policy import EgressContext, EgressScope
102 return EgressContext(
103 scope=EgressScope.PRIVATE_ONLY,
104 primary_engine=getattr(ctx, "primary_engine", "") or "",
105 require_local_llm=True,
106 require_local_embeddings=True,
107 local_hostnames=getattr(ctx, "local_hostnames", ()) or (),
108 username=getattr(ctx, "username", None),
109 )
112def _trusted_names(setting_key: str, settings_snapshot) -> set:
113 """Lower-cased set of names the user has explicitly trusted.
115 Reads a JSON-list setting (mirroring ``llm.allowed_local_hostnames``);
116 returns an empty set on a missing/malformed value.
117 """
118 if not settings_snapshot:
119 return set()
120 from .policy import _get_setting_value, coerce_str_list
122 _, names = coerce_str_list(
123 _get_setting_value(settings_snapshot, setting_key, ())
124 )
125 return {n.lower() for n in names}
128def _inference_label(provider, evaluate_fn, settings_snapshot, ctx) -> Label:
129 """Shared exposure resolution for an inference sink (LLM or embeddings).
131 Decided by the SAME classification the enforcing PEP uses (``evaluate_fn``
132 under a require-local probe): the provider's configured endpoint is
133 CONTAINED iff the PEP would accept it as local, else EXPOSING — keeping the
134 resolver in exact lock-step with enforcement (so a self-hosted endpoint on a
135 local URL is contained, not falsely exposing). Trust then relaxes an
136 exposing sink the user has explicitly vouched for.
137 """
138 p = (provider or "").lower()
139 snapshot = settings_snapshot if settings_snapshot is not None else {}
140 try:
141 allowed = evaluate_fn(
142 p, _require_local_probe(ctx), settings_snapshot=snapshot
143 ).allowed
144 except Exception: # noqa: silent-exception - fail closed to exposing
145 allowed = False
146 exposure = Exposure.CONTAINED if allowed else Exposure.EXPOSING
147 if exposure is Exposure.EXPOSING and p in _trusted_names(
148 "policy.trusted_inference_providers", settings_snapshot
149 ):
150 exposure = Exposure.CONTAINED
151 return Label(Sensitivity.NON_SENSITIVE, exposure)
154def llm_label(
155 provider: Optional[str], settings_snapshot=None, ctx=None
156) -> Label:
157 """Resolve an LLM provider's label (an inference sink — exposure only)."""
158 from .policy import evaluate_llm_endpoint
160 return _inference_label(
161 provider, evaluate_llm_endpoint, settings_snapshot, ctx
162 )
165def embeddings_label(
166 provider: Optional[str], settings_snapshot=None, ctx=None
167) -> Label:
168 """Resolve an embeddings provider's label (an inference sink)."""
169 from .policy import evaluate_embeddings
171 return _inference_label(
172 provider, evaluate_embeddings, settings_snapshot, ctx
173 )
176def classify_run(
177 settings_snapshot,
178 ctx,
179 *,
180 engines: Iterable[str],
181 llm_provider: Optional[str] = None,
182 embeddings_provider: Optional[str] = None,
183) -> List[Component]:
184 """Assemble the run's classified component set.
186 Each search engine contributes two components sharing its name — a
187 ``SOURCE`` (carrying its sensitivity) and a ``SEARCH_SINK`` (carrying its
188 exposure) — so the quadrant-4 self-exclusion in :func:`evaluate_run` works.
189 The LLM and embeddings providers are ``INFERENCE_SINK`` components.
190 """
191 components: List[Component] = []
192 for name in engines:
193 label = engine_label(name, settings_snapshot, ctx)
194 if label is None:
195 # Unknown engine (not in the registry — e.g. a programmatic
196 # retriever): fail CLOSED to the most restrictive quadrant rather
197 # than dropping it, so an unclassified source can never silently
198 # relax the run's admissibility.
199 label = Label(Sensitivity.SENSITIVE, Exposure.EXPOSING)
200 components.append(Component(name, Role.SOURCE, label))
201 components.append(Component(name, Role.SEARCH_SINK, label))
202 if llm_provider: 202 ↛ 210line 202 didn't jump to line 210 because the condition on line 202 was always true
203 components.append(
204 Component(
205 f"llm:{llm_provider}",
206 Role.INFERENCE_SINK,
207 llm_label(llm_provider, settings_snapshot, ctx),
208 )
209 )
210 if embeddings_provider:
211 components.append(
212 Component(
213 f"embeddings:{embeddings_provider}",
214 Role.INFERENCE_SINK,
215 embeddings_label(embeddings_provider, settings_snapshot, ctx),
216 )
217 )
218 return components
221def _fail_closed_decision(ctx) -> Decision:
222 """The decision to return when the two-axis rule cannot be computed.
224 Refuses the run (reason ``audit_error``) so the failure is visible, except
225 under the ``UNPROTECTED`` escape hatch, which must never block.
227 This is called from the ``except`` blocks that handle failures, so it must
228 NOT import ``.policy``: if the triggering failure was itself a ``.policy``
229 import error (its imports are lazy precisely to dodge a circular-import
230 hazard), re-importing here would re-raise and defeat the fail-closed
231 guarantee — ``audit_run`` would raise instead of returning a denial, and
232 the callers' outer handlers would then fail OPEN. ``EgressScope`` is a
233 ``str`` enum, so compare against the literal value with no import.
234 """
235 if getattr(ctx, "scope", None) == "unprotected":
236 return Decision(True, "unprotected")
237 return Decision(False, "audit_error")
240def audit_run(
241 settings_snapshot,
242 ctx,
243 *,
244 engines: Iterable[str],
245 llm_provider: Optional[str] = None,
246 embeddings_provider: Optional[str] = None,
247) -> Decision:
248 """Compute the four-quadrant decision for a run and LOG it.
250 Never raises. On success returns the :class:`Decision`. If the decision
251 cannot be computed it **fails closed** — returns a denying ``Decision``
252 (reason ``audit_error``) so the run is refused and the failure is visible,
253 rather than silently degrading to the scope PEPs. Under the ``UNPROTECTED``
254 escape-hatch scope the decision is evaluated in :attr:`Mode.PERMISSIVE`
255 (always allowed, diagnostic preserved), and an ``audit_error`` under that
256 scope is likewise never blocking — the escape hatch must never refuse.
257 """
258 try:
259 from .policy import EgressScope
261 mode = (
262 Mode.PERMISSIVE
263 if getattr(ctx, "scope", None) == EgressScope.UNPROTECTED
264 else Mode.ENFORCING
265 )
266 components = classify_run(
267 settings_snapshot,
268 ctx,
269 engines=engines,
270 llm_provider=llm_provider,
271 embeddings_provider=embeddings_provider,
272 )
273 decision = evaluate_run(components, mode=mode)
274 logger.bind(policy_audit=True).info(
275 "egress two-axis audit",
276 allowed=decision.allowed,
277 reason=decision.reason,
278 offending=decision.offending,
279 )
280 return decision
281 except Exception: # noqa: silent-exception - audit must never break a run
282 logger.bind(policy_audit=True).warning(
283 "egress two-axis audit could not be computed — failing closed",
284 exc_info=True,
285 )
286 return _fail_closed_decision(ctx)
289def audit_run_from_snapshot(
290 settings_snapshot, ctx, primary_engine, llm_provider=None
291):
292 """Two-axis decision for a run described by its settings snapshot.
294 Shared by the ``/api/start_research`` precheck and the worker chokepoint in
295 ``run_research_process`` so every entry point — API, follow-up, chat, queue
296 — evaluates the identical rule, then delegates to :func:`audit_run` (which
297 fails closed to a denying decision when the rule cannot be computed).
299 ``llm_provider`` is the run's per-request provider override (the
300 ``model_provider`` kwarg / request field). It **wins** over the snapshot,
301 mirroring ``get_llm`` (``config/llm_config.py``: an explicit provider is
302 used, else the snapshot default). Passing only the snapshot would audit the
303 stored default while the run builds the overridden provider — e.g. a saved
304 local ``ollama`` audited as CONTAINED while the run actually calls a cloud
305 provider chosen for that one run — admitting a leak. A **falsy** override
306 (``None`` or ``""`` — chat passes ``""`` when no provider is set) means "no
307 override": the provider is resolved from the snapshot, the SAME key/default
308 the run uses. Using ``is None`` here would let ``""`` through and
309 ``classify_run``'s truthy guard would then silently drop the LLM sink.
311 The LLM and (RAG-only) embeddings providers are read with
312 :func:`get_setting_from_snapshot` using the SAME keys/defaults the run uses,
313 so the check never resolves to a different (safer) value than the run.
314 Embeddings for a library / collection run come from
315 ``local_search_embedding_provider`` (``search_engine_library.py``), NOT the
316 unrelated ``embeddings.provider`` key.
318 Resolution runs inside the same fail-closed guard as the decision itself.
319 """
320 try:
321 from ...config.thread_settings import get_setting_from_snapshot
323 if not llm_provider:
324 # get_setting_from_snapshot only substitutes the default when the
325 # key is ABSENT; a key present-but-empty returns "" again. Fall back
326 # to the system default so the LLM sink is always classified and
327 # never silently dropped by classify_run's truthy guard.
328 llm_provider = (
329 get_setting_from_snapshot(
330 "llm.provider",
331 "ollama",
332 settings_snapshot=settings_snapshot,
333 )
334 or "ollama"
335 )
336 # Embeddings only leave the box when the run actually embeds — RAG over
337 # a collection / library. A lexical store or a web primary never embeds,
338 # so a configured cloud embedder must not falsely trip the
339 # sensitive->exposing check for those runs. Read the SAME key the RAG
340 # engine uses (search_engine_library.py: local_search_embedding_provider);
341 # embeddings.provider is a different, unused-here setting.
342 embeddings_provider = None
343 if primary_engine == "library" or (
344 isinstance(primary_engine, str)
345 and primary_engine.startswith("collection_")
346 ):
347 embeddings_provider = get_setting_from_snapshot(
348 "local_search_embedding_provider",
349 default="sentence_transformers",
350 settings_snapshot=settings_snapshot,
351 )
352 except Exception: # noqa: silent-exception - resolution must fail closed
353 logger.bind(policy_audit=True).warning(
354 "egress provider resolution failed — failing closed", exc_info=True
355 )
356 return _fail_closed_decision(ctx)
357 return audit_run(
358 settings_snapshot,
359 ctx,
360 engines=[primary_engine],
361 llm_provider=llm_provider,
362 embeddings_provider=embeddings_provider,
363 )