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