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