diff --git a/benchmarks/run.py b/benchmarks/run.py index 5710a125..17b6db32 100644 --- a/benchmarks/run.py +++ b/benchmarks/run.py @@ -24,6 +24,7 @@ import tempfile import time from pathlib import Path +from typing import Any BENCH_ROOT = Path(__file__).resolve().parent @@ -124,6 +125,73 @@ def run_deterministic(corpus_dir: Path, case: dict, out, variant: str = "vulnera } +def engine_run_receipt(repo_dir: Path, scan_id: str) -> dict[str, Any] | None: + """Compact receipt of the engine run's provenance/quality record. + + Reads the emitted ``run.json`` (schema 1.1 fields when present) so eval + manifests carry comparable, evidence-backed capability and quality + receipts across runs — probed capability statuses, named preflight + degradations, scope-violation counts, and observed activity totals. + ``None`` when the run wrote no record; ``{"error": ...}`` when it is + unreadable rather than silently absent. + """ + path = repo_dir / "strix_runs" / scan_id / "run.json" + if not path.is_file(): + return None + try: + record = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + return {"error": f"run.json unreadable: {exc}"} + if not isinstance(record, dict): + return {"error": "run.json is not an object"} + + receipt: dict[str, Any] = {"schemaVersion": record.get("schema_version")} + caps = record.get("sandbox_capabilities") + if isinstance(caps, dict): + probed = caps.get("capabilities") + preflight = caps.get("preflight") + + def _controls(key: str) -> list[str]: + if not isinstance(preflight, dict): + return [] + return sorted( + str(item.get("control")) + for item in preflight.get(key) or [] + if isinstance(item, dict) and item.get("control") + ) + + receipt["sandboxCapabilities"] = { + "backend": caps.get("backend"), + "statuses": { + str(name): str(entry.get("status")) + for name, entry in probed.items() + if isinstance(entry, dict) + } + if isinstance(probed, dict) + else {}, + "preflightDegradations": _controls("degradations"), + "preflightFailures": _controls("failures"), + } + quality = record.get("scan_quality") + if isinstance(quality, dict): + observed = quality.get("observed") + receipt["scanQuality"] = { + "observed": observed if isinstance(observed, dict) else None, + "declared": quality.get("declared") + if isinstance(quality.get("declared"), dict) + else None, + "surfaces": len(quality.get("surfaces") or []), + "unassessed": len(quality.get("unassessed") or []), + } + violations = record.get("scope_violations") + if isinstance(violations, dict): + receipt["scopeViolations"] = violations.get("total") + export = record.get("evidence_export") + if isinstance(export, dict): + receipt["evidenceExport"] = export.get("status") + return receipt + + def run_engine( corpus_dir: Path, case: dict, out, run_index: int, variant: str = "vulnerable" ) -> dict: @@ -188,6 +256,7 @@ def run_engine( errors.append("invalid findings JSON") else: errors.append("missing vulnerabilities.json") + receipt = engine_run_receipt(repo_dir, scan_id) shutil.rmtree(repo_dir, ignore_errors=True) return { "case": case["id"], @@ -200,6 +269,7 @@ def run_engine( "runtimeMs": runtime_ms, "returncode": proc.returncode, "stderrTail": proc.stderr[-2000:] if proc.returncode != 0 else None, + "runReceipt": receipt, } diff --git a/benchmarks/score.py b/benchmarks/score.py index 880aae0a..cea6bedb 100644 --- a/benchmarks/score.py +++ b/benchmarks/score.py @@ -213,6 +213,32 @@ def score(results_dir: Path, corpus_dir: Path) -> dict: ) clean_runs = [r for r in manifest["runs"] if r.get("variant") == "clean"] + engine_runs_meta = [r for r in manifest["runs"] if r.get("detector") == "engine"] + receipts = [ + r["runReceipt"] + for r in engine_runs_meta + if isinstance(r.get("runReceipt"), dict) and not r["runReceipt"].get("error") + ] + # Comparable receipts: whether every engine run carried the same run.json + # provenance fields (schema, probed capability statuses). Presence is + # counted, never inferred — a run without a receipt is just absent. + engine_receipts = { + "total": len(engine_runs_meta), + "withReceipt": len(receipts), + "schemaVersions": sorted( + {str(r.get("schemaVersion")) for r in receipts if r.get("schemaVersion")} + ), + "capabilityStatuses": sorted( + { + f"{name}={status}" + for r in receipts + for name, status in ( + (r.get("sandboxCapabilities") or {}).get("statuses") or {} + ).items() + } + ), + } + summary = { "status": status, "failedRuns": len(failures), @@ -236,6 +262,7 @@ def score(results_dir: Path, corpus_dir: Path) -> dict: "engineStability": stability, "totalRuntimeMs": sum(r.get("runtimeMs", 0) for r in manifest["runs"]), "discoveryReceipts": len(discovery_receipts), + "engineReceipts": engine_receipts, "engineRevision": manifest.get("engineRevision"), } (results_dir / "results.json").write_text(json.dumps(summary, indent=2)) diff --git a/lyrashield/artifacts/quality.py b/lyrashield/artifacts/quality.py new file mode 100644 index 00000000..78be6cd5 --- /dev/null +++ b/lyrashield/artifacts/quality.py @@ -0,0 +1,258 @@ +"""``scan_quality`` — honest per-surface accounting for run.json schema 1.1. + +This block answers "what was actually exercised versus merely declared". +Every number here derives from activity the runtime observed (agent graph, +filed findings, web-search metering, replay admission decisions, probed +sandbox capabilities) or from the model-declared coverage ledger — always +labeled ``declared``. Nothing is estimated and no coverage percentage is +invented: a surface with no observed or declared activity stays +``unassessed``, and capability gaps surface as named degradations rather +than being smoothed into a score. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime +from typing import Any +from urllib.parse import urlparse + +from strix.report.coverage import agents_from_graph + + +logger = logging.getLogger(__name__) + +SCAN_QUALITY_SCHEMA = "lyrashield-scan-quality/1.0" + +_MAX_SURFACES = 200 +_MAX_SURFACE_CHARS = 200 +_MAX_UNASSESSED = 100 + +# Agent graph statuses that mean the agent did not finish cleanly. +_INCOMPLETE_AGENT_STATUSES = frozenset({"crashed", "stopped", "running", "waiting"}) + + +def _surface_key(value: Any) -> str | None: + """Normalize a surface reference to a host or bounded free-text key.""" + if not isinstance(value, str) or not value.strip(): + return None + text = value.strip()[:_MAX_SURFACE_CHARS] + candidate = text if "://" in text else f"//{text.split('/')[0]}" + try: + parsed = urlparse(candidate) + except ValueError: + return text + host = parsed.hostname + if host: + return host.lower().rstrip(".") + return text + + +def _authorized_scope_hosts(run_record: dict[str, Any]) -> list[str]: + """The recorded authorized host set — probed capability record first.""" + caps = run_record.get("sandbox_capabilities") + if isinstance(caps, dict): + hosts = caps.get("authorized_hosts") + if isinstance(hosts, list): + return sorted({str(h) for h in hosts if isinstance(h, str) and h}) + scope = run_record.get("proxy_default_scope") + if isinstance(scope, dict): + allowlist = scope.get("allowlist") + if isinstance(allowlist, list): + return sorted( + {str(p).lstrip("*.") for p in allowlist if isinstance(p, str) and p.strip("*.")} + ) + return [] + + +def _sandbox_summary(run_record: dict[str, Any]) -> dict[str, Any] | None: + """Compact capability/preflight provenance for the quality block.""" + caps = run_record.get("sandbox_capabilities") + if not isinstance(caps, dict): + return None + probed = caps.get("capabilities") + statuses = ( + { + str(name): str(entry.get("status")) + for name, entry in probed.items() + if isinstance(entry, dict) + } + if isinstance(probed, dict) + else {} + ) + preflight = caps.get("preflight") + degradations: list[str] = [] + failures: list[str] = [] + if isinstance(preflight, dict): + degradations.extend( + str(item["control"]) + for item in preflight.get("degradations") or [] + if isinstance(item, dict) and item.get("control") + ) + failures.extend( + str(item["control"]) + for item in preflight.get("failures") or [] + if isinstance(item, dict) and item.get("control") + ) + summary: dict[str, Any] = { + "backend": caps.get("backend"), + "capabilities": statuses, + "preflight_degradations": sorted(degradations), + "preflight_failures": sorted(failures), + } + return summary + + +def _surface_row() -> dict[str, Any]: + return { + "origins": set(), + "declared": 0, + "findings": 0, + "admitted_requests": 0, + "denied_requests": 0, + } + + +def _assessment(row: dict[str, Any]) -> str: + """Honest per-surface verdict — observed beats declared beats nothing.""" + if row["findings"] > 0 or row["admitted_requests"] > 0: + return "observed" + if row["declared"] > 0: + return "declared" + if row["denied_requests"] > 0: + return "denied" + return "unassessed" + + +def build_scan_quality( + *, + run_record: dict[str, Any], + agent_graph: dict[str, Any], + coverage_entries: list[dict[str, Any]], + vulnerability_reports: list[dict[str, Any]], + scope_decisions: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Assemble the ``scan_quality`` run-record block. + + All inputs are already-observed run data; this function derives counts + and per-surface verdicts only. It never estimates coverage, never + upgrades a model-declared claim to an observation, and leaves every + unexercised surface explicitly ``unassessed``. + """ + agents = agents_from_graph(agent_graph) + agents_incomplete = sum( + 1 for agent in agents if agent.get("status") in _INCOMPLETE_AGENT_STATUSES + ) + + decisions = scope_decisions if isinstance(scope_decisions, dict) else {} + violations = [v for v in decisions.get("violations") or [] if isinstance(v, dict)] + dropped_violations = int(decisions.get("dropped") or 0) + admitted_hosts = { + str(host): int(count) + for host, count in (decisions.get("admitted_hosts") or {}).items() + if isinstance(count, int) and count > 0 + } + + surfaces: dict[str, dict[str, Any]] = {} + + def _row(key: str | None) -> dict[str, Any] | None: + if not key: + return None + if key not in surfaces and len(surfaces) < _MAX_SURFACES: + surfaces[key] = _surface_row() + return surfaces.get(key) + + for host in _authorized_scope_hosts(run_record): + row = _row(host) + if row is not None: + row["origins"].add("authorized_scope") + + declared_outcomes: dict[str, int] = {} + for entry in coverage_entries: + outcome = str(entry.get("outcome") or "") + if outcome: + declared_outcomes[outcome] = declared_outcomes.get(outcome, 0) + 1 + row = _row(_surface_key(entry.get("surface"))) + if row is not None: + row["declared"] += 1 + row["origins"].add("declared") + + for report in vulnerability_reports: + key = _surface_key(report.get("endpoint")) or _surface_key(report.get("target")) + row = _row(key) + if row is not None: + row["findings"] += 1 + row["origins"].add("finding") + + for host, count in admitted_hosts.items(): + row = _row(host) + if row is not None: + row["admitted_requests"] += count + row["origins"].add("egress_admitted") + + for violation in violations: + row = _row(str(violation.get("host") or "") or None) + if row is not None: + row["denied_requests"] += 1 + row["origins"].add("egress_denied") + + surface_rows: list[dict[str, Any]] = [] + unassessed: list[str] = [] + for key in sorted(surfaces): + row = surfaces[key] + verdict = _assessment(row) + surface_rows.append( + { + "surface": key, + "origins": sorted(row["origins"]), + "declared_coverage_entries": row["declared"], + "findings": row["findings"], + "admitted_requests": row["admitted_requests"], + "denied_requests": row["denied_requests"], + "assessment": verdict, + } + ) + if verdict == "unassessed" and len(unassessed) < _MAX_UNASSESSED: + unassessed.append(key) + + web_search_usage = run_record.get("web_search_usage") + export = run_record.get("evidence_export") + document: dict[str, Any] = { + "schema": SCAN_QUALITY_SCHEMA, + "generated_at": datetime.now(UTC).isoformat(), + "run_id": run_record.get("run_id"), + "observed": { + "agents_total": len(agents), + "agents_finished": len(agents) - agents_incomplete, + "agents_incomplete": agents_incomplete, + "findings_filed": len(vulnerability_reports), + "web_search_calls": len(web_search_usage) if isinstance(web_search_usage, list) else 0, + "proxy_requests_admitted": sum(admitted_hosts.values()), + "proxy_requests_denied": len(violations) + dropped_violations, + "evidence_export": export.get("status") if isinstance(export, dict) else None, + "sandbox": _sandbox_summary(run_record), + "scan_status": run_record.get("status"), + "terminal_reason": run_record.get("terminal_reason"), + }, + "declared": { + "coverage_entries": len(coverage_entries), + "outcomes": declared_outcomes, + }, + "surfaces": surface_rows, + "unassessed": unassessed, + "note": ( + "Counts derive only from observed runtime activity and the " + "model-declared coverage ledger. 'declared' surfaces are " + "agent-reported, not machine-verified; 'unassessed' surfaces " + "were in scope but have no recorded exercise." + ), + } + if dropped_violations: + document["observed"]["scope_violations_dropped"] = dropped_violations + return document + + +__all__ = [ + "SCAN_QUALITY_SCHEMA", + "build_scan_quality", +] diff --git a/lyrashield/artifacts/state.py b/lyrashield/artifacts/state.py index 4d87783d..dcfac1f4 100644 --- a/lyrashield/artifacts/state.py +++ b/lyrashield/artifacts/state.py @@ -16,6 +16,7 @@ from agents.usage import Usage from lyrashield.artifacts import evidence as _evidence +from lyrashield.artifacts import quality as _quality from lyrashield.artifacts.sarif import write_sarif from lyrashield.artifacts.usage import LLMUsageLedger, _int_or_zero, _round_cost from lyrashield.artifacts.writer import ( @@ -72,9 +73,13 @@ def _clean_title(title: str) -> str: # structured advisory_cvss, severity_change_conditions, engine-attested # fix_verification, bounded http_exchange_ids, append-only revision history) # plus coverage.json, threat_model.json, http_exchanges.json and an inline -# result manifest. The whole surface is gated on ``LYRASHIELD_RUN_RECORD_V1_1`` -# (default off) so readers deploy before writers; a run keeps the version it -# was created with. +# result manifest. Task-12 additions: ``scope_violations`` (bounded replay-guard +# denial evidence) and ``scan_quality`` (per-surface observed-vs-declared +# accounting). ``sandbox_capabilities`` — the probed backend capability record — +# is unconditional run provenance, not part of the gated evidence surface. The +# whole surface is gated on ``LYRASHIELD_RUN_RECORD_V1_1`` (default off) so +# readers deploy before writers; a run keeps the version it was created with. +RUN_RECORD_SCHEMA_VERSION = "1.0" # Fields every run.json write must carry from its first observable appearance # (the worker parses this contract at any point in the run, not just at the @@ -269,6 +274,9 @@ def get_global_report_state() -> Optional["ReportState"]: _MAX_COLLECTION_SIZE = 1_000 _MAX_METADATA_DEPTH = 10 _MAX_FINDING_SERIALIZED_SIZE = 1_000_000 +# Persisted scope-violation entries bound (ledger bound is lower; this caps +# the durable list merged across saves). +_MAX_SCOPE_VIOLATION_ENTRIES = 500 def _truncate_text(value: str, max_length: int = _MAX_TEXT_LENGTH) -> str: @@ -584,6 +592,12 @@ def __init__(self, run_name: str | None = None): self.posthog_scan_ended_sent: bool = False self.scarf_scan_ended_sent: bool = False self.scan_ended_exit_reason: str | None = None + # How many scope-violation ledger entries have already been merged + # into run_record["scope_violations"]["entries"], and how much of the + # process-cumulative ledger overflow has already been accounted into + # the persisted ``dropped`` count. + self._scope_violations_seen = 0 + self._scope_dropped_seen = 0 def get_run_dir(self) -> Path: if self._run_dir is None: @@ -683,6 +697,24 @@ def hydrate_from_run_dir(self) -> None: self._report_artifacts_revision = restored_revision self._persisted_report_artifacts_revision = restored_revision + # Same-process resume: the caido ledger may still hold entries already + # merged into the persisted record. Seed both offsets from the live + # snapshot so _sync_scope_decisions() only processes new denials — + # never re-appends entries or re-adds previously counted overflow. + try: + from lyrashield.tools.proxy import caido_api + + snapshot = caido_api.get_scope_decisions() + except ImportError: + snapshot = None + if isinstance(snapshot, dict): + violations = snapshot.get("violations") + if isinstance(violations, list): + self._scope_violations_seen = max(self._scope_violations_seen, len(violations)) + dropped = snapshot.get("dropped") + if isinstance(dropped, int) and not isinstance(dropped, bool): + self._scope_dropped_seen = max(self._scope_dropped_seen, dropped) + def add_vulnerability_report( self, title: str, @@ -1056,6 +1088,58 @@ def update_vulnerability_report( self.vulnerability_updated_callback(sanitized) return revised + def set_sandbox_capabilities(self, capabilities: dict[str, Any]) -> None: + """Record the probed sandbox capability set as run provenance. + + The record comes from the post-start capability probe — only + capabilities the backend verifiably delivered are marked + ``supported``; ``unprobed`` entries carry named preflight + degradations so a missing guarantee is never silent. + """ + if isinstance(capabilities, dict): + self.run_record["sandbox_capabilities"] = capabilities + + def _sync_scope_decisions(self) -> dict[str, int]: + """Merge new replay-guard denials into the run record (bounded). + + Violations arrive from the process-local ledger in + ``lyrashield.tools.proxy.caido_api`` — every denied request the + replay guard saw. The persisted list is capped; overflow is counted + in ``dropped`` so the record stays honest about truncation. Returns + the per-host admitted-request counts for quality accounting. + """ + from lyrashield.tools.proxy import caido_api + + snapshot = caido_api.get_scope_decisions() + violations = snapshot["violations"] + new_entries = violations[self._scope_violations_seen :] + self._scope_violations_seen = len(violations) + + persisted = self.run_record.get("scope_violations") + existing: list[dict[str, Any]] = [] + # snapshot["dropped"] is process-cumulative; add only the new overflow + # so repeated saves cannot inflate the persisted count. + ledger_dropped = int(snapshot["dropped"]) + dropped = ledger_dropped - self._scope_dropped_seen + self._scope_dropped_seen = ledger_dropped + if isinstance(persisted, dict): + raw_entries = persisted.get("entries") + if isinstance(raw_entries, list): + existing = [e for e in raw_entries if isinstance(e, dict)] + dropped += int(persisted.get("dropped") or 0) + + keep = _MAX_SCOPE_VIOLATION_ENTRIES - len(existing) + if len(new_entries) > keep: + dropped += len(new_entries) - max(0, keep) + new_entries = new_entries[: max(0, keep)] + existing.extend(new_entries) + self.run_record["scope_violations"] = { + "entries": existing, + "dropped": dropped, + "total": len(existing) + dropped, + } + return cast("dict[str, int]", snapshot["admitted_hosts"]) + def set_evidence_export_outcome(self, outcome: dict[str, Any]) -> None: """Record the HTTP exchange evidence export result on the run record. @@ -1445,13 +1529,11 @@ def _write_evidence_artifacts(self, run_dir: Path) -> bool: ``False`` so the persisted revision stays honest, but they never block the required receipt. """ - from strix.report.coverage import read_agent_graph - persisted = True try: coverage = _evidence.build_coverage_document( run_record=self.run_record, - agent_graph=read_agent_graph(runtime_state_dir(run_dir)), + agent_graph=_read_agent_graph(runtime_state_dir(run_dir)), vulnerability_reports=self.vulnerability_reports, exit_reason=self.scan_ended_exit_reason, ) @@ -1520,6 +1602,31 @@ def _save_artifacts(self) -> bool: # touching findings), so they refresh on every save. if not self._write_evidence_artifacts(run_dir): report_artifacts_persisted = False + # Replay-guard denials and the honest quality ledger are derived + # from observed activity only — unexercised surfaces stay + # unassessed, never smoothed into a coverage number. + admitted_hosts: dict[str, int] = {} + try: + admitted_hosts = self._sync_scope_decisions() + except Exception: + logger.exception("scope_violations sync failed (non-fatal)") + try: + recorded = self.run_record.get("scope_violations") + recorded = recorded if isinstance(recorded, dict) else {} + self.run_record["scan_quality"] = _quality.build_scan_quality( + run_record=self.run_record, + agent_graph=_read_agent_graph(runtime_state_dir(run_dir)), + coverage_entries=_coverage_ledger_entries(), + vulnerability_reports=self.vulnerability_reports, + scope_decisions={ + "violations": recorded.get("entries") or [], + "dropped": recorded.get("dropped") or 0, + "admitted_hosts": admitted_hosts, + }, + ) + except Exception: + report_artifacts_persisted = False + logger.exception("scan_quality build failed (non-fatal)") # The result manifest binds every emitted artifact to this record # by checksum — the immutable link the worker verifies against. try: @@ -1656,6 +1763,20 @@ def _hydrate_llm_usage(self, raw_usage: Any) -> None: self._sync_llm_usage_record() +def _read_agent_graph(state_dir: Path) -> dict[str, Any]: + """Lazy wrapper so the substrate coverage module loads only on demand.""" + from strix.report.coverage import read_agent_graph + + return read_agent_graph(state_dir) + + +def _coverage_ledger_entries() -> list[dict[str, Any]]: + """Lazy wrapper for the model-declared coverage ledger store.""" + from strix.tools.coverage.tools import get_coverage_entries + + return cast("list[dict[str, Any]]", get_coverage_entries()) + + def _as_dict(obj: Any) -> dict[str, Any] | None: """Return *obj* as a str-keyed dict, or None if it isn't a mapping.""" if isinstance(obj, dict): diff --git a/lyrashield/lifecycle/runner.py b/lyrashield/lifecycle/runner.py index 0cd061f0..f2c4134b 100644 --- a/lyrashield/lifecycle/runner.py +++ b/lyrashield/lifecycle/runner.py @@ -382,6 +382,11 @@ async def run_strix_scan( "name": "authorized-targets", "allowlist": bundle.get("default_scope_allowlist") or [], } + capabilities = bundle.get("sandbox_capabilities") + if capabilities is not None: + report_state = get_global_report_state() + if report_state is not None: + report_state.set_sandbox_capabilities(capabilities) sandbox_session = bundle["session"] diff --git a/lyrashield/runtime/capabilities.py b/lyrashield/runtime/capabilities.py new file mode 100644 index 00000000..87fa1240 --- /dev/null +++ b/lyrashield/runtime/capabilities.py @@ -0,0 +1,394 @@ +"""Probed sandbox capability provenance and preflight evaluation. + +Session setup must never claim a control it did not verify. This module +probes the capabilities the active backend actually delivered — network +isolation, read-only mounts, exposed ports, exec, resource constraints — +immediately after the session starts, then evaluates them against the +controls the run requires. + +Statuses: + +- ``supported`` — probed and confirmed (with evidence). +- ``absent`` — probed and found missing/contradicted. +- ``unprobed`` — the backend exposes no way to verify it. An unprobed + capability is never claimed; when a required control rests on it the run + records a named degradation rather than silently proceeding. + +A required control whose capability is probed ``absent`` fails preflight +(:class:`SandboxPreflightError`), which surfaces inside the session +lifecycle so the half-started sandbox is torn down by the caller's normal +cleanup path. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from datetime import UTC, datetime +from typing import Any, cast + +from lyrashield.runtime.docker_client import _sandbox_network + + +logger = logging.getLogger(__name__) + +CAPABILITY_RECORD_SCHEMA = "lyrashield-sandbox-capabilities/1.0" + +STATUS_SUPPORTED = "supported" +STATUS_ABSENT = "absent" +STATUS_UNPROBED = "unprobed" + + +class SandboxPreflightError(RuntimeError): + """A capability a required control depends on is probed absent.""" + + +def _cap(status: str, detail: str, **evidence: Any) -> dict[str, Any]: + entry: dict[str, Any] = {"status": status, "detail": detail} + if evidence: + entry["evidence"] = evidence + return entry + + +def _container_attrs(client: Any, session: Any) -> dict[str, Any] | None: + """Live container attrs for the docker backend, else ``None``. + + ``None`` means the backend exposes no introspection — the capability is + ``unprobed``, never silently assumed. + """ + docker_client = getattr(client, "docker_client", None) + container_id = getattr(getattr(session, "_inner", session), "container_id", None) + if docker_client is None or not isinstance(container_id, str) or not container_id: + return None + try: + container = docker_client.containers.get(container_id) + attrs = getattr(container, "attrs", None) + except Exception: # introspection failure means unprobed + logger.debug("capability probe: container inspect failed", exc_info=True) + return None + return attrs if isinstance(attrs, dict) else None + + +def _probe_exec(session: Any) -> dict[str, Any]: + """``session.exec`` is how every agent tool reaches the sandbox.""" + if callable(getattr(session, "exec", None)): + return _cap(STATUS_SUPPORTED, "session.exec is available") + return _cap(STATUS_ABSENT, "session object exposes no exec entrypoint") + + +def _probe_ports(session: Any, caido_endpoint: Any) -> dict[str, Any]: + """Exposed-port resolution is required to reach the in-container proxy.""" + if not callable(getattr(session, "resolve_exposed_port", None)): + return _cap(STATUS_ABSENT, "session exposes no port resolution") + host = getattr(caido_endpoint, "host", None) + port = getattr(caido_endpoint, "port", None) + if host and port: + return _cap( + STATUS_SUPPORTED, + "proxy port resolved through the backend", + resolved=f"{host}:{port}", + ) + return _cap(STATUS_ABSENT, "proxy port resolution returned no endpoint") + + +def _probe_proxy_capture(caido_client: Any) -> dict[str, Any]: + if caido_client is not None: + return _cap(STATUS_SUPPORTED, "capture/replay client bootstrapped") + return _cap(STATUS_ABSENT, "capture/replay client unavailable") + + +def _probe_network_policy( + backend_name: str, + client: Any, + session: Any, +) -> dict[str, Any]: + """Verify deny-by-default egress from immutable container/network facts.""" + if backend_name != "docker": + return _cap( + STATUS_UNPROBED, + f"backend {backend_name!r} exposes no network introspection", + ) + attrs = _container_attrs(client, session) + if attrs is None: + return _cap(STATUS_UNPROBED, "container attributes unavailable") + configured = _sandbox_network() + if not configured: + return _cap(STATUS_ABSENT, "no deny-by-default sandbox network configured") + host_config = cast("dict[str, Any]", attrs.get("HostConfig", {}) or {}) + mode = str(host_config.get("NetworkMode", "") or "") + networks = cast( + "dict[str, Any]", + cast("dict[str, Any]", attrs.get("NetworkSettings", {}) or {}).get("Networks", {}) or {}, + ) + evidence: dict[str, Any] = { + "configured": configured, + "network_mode": mode, + "attached": sorted(str(k) for k in networks), + } + if mode != configured: + return _cap( + STATUS_ABSENT, + f"container network mode {mode!r} does not match {configured!r}", + **evidence, + ) + if set(networks.keys()) != {configured}: + return _cap( + STATUS_ABSENT, + "container is attached to networks besides the sandbox network", + **evidence, + ) + try: + network = client.docker_client.networks.get(configured) + internal = bool(cast("dict[str, Any]", getattr(network, "attrs", {}) or {}).get("Internal")) + except Exception: # introspection failure means unprobed + return _cap( + STATUS_UNPROBED, + "network object inspection failed", + **evidence, + ) + evidence["internal"] = internal + if not internal: + return _cap( + STATUS_ABSENT, + f"network {configured!r} is not internal", + **evidence, + ) + return _cap( + STATUS_SUPPORTED, + "container is attached exclusively to an internal (deny-by-default) network", + **evidence, + ) + + +def _probe_mounts( + backend_name: str, + client: Any, + session: Any, + bind_mounts: list[dict[str, Any]], +) -> dict[str, Any]: + """Verify every requested bind mount landed, with its read-only bit. + + The egress policy and relay grant are delivered read-only; a mount that + silently arrived writable (or not at all) means the in-sandbox guard has + no trustworthy policy file and must fail closed. + """ + expected = [m for m in bind_mounts if m.get("target")] + if backend_name != "docker": + return _cap( + STATUS_UNPROBED, + f"backend {backend_name!r} exposes no mount introspection", + expected=len(expected), + ) + attrs = _container_attrs(client, session) + if attrs is None: + return _cap( + STATUS_UNPROBED, + "container attributes unavailable", + expected=len(expected), + ) + mounts = attrs.get("Mounts", []) + by_target: dict[str, dict[str, Any]] = {} + if isinstance(mounts, list): + for mount in mounts: + if isinstance(mount, dict) and mount.get("Destination"): + by_target[str(mount["Destination"])] = mount + missing: list[str] = [] + writable: list[str] = [] + verified: list[str] = [] + for spec in expected: + target = str(spec["target"]) + mount = by_target.get(target) + if mount is None: + missing.append(target) + continue + wants_ro = spec.get("read_only", True) + if wants_ro and mount.get("RW") is not False: + writable.append(target) + continue + verified.append(target) + evidence = {"verified": verified, "missing": missing, "writable": writable} + if missing or writable: + parts = [] + if missing: + parts.append(f"missing: {', '.join(missing)}") + if writable: + parts.append(f"mounted writable: {', '.join(writable)}") + return _cap(STATUS_ABSENT, "; ".join(parts), **evidence) + return _cap( + STATUS_SUPPORTED, + f"{len(verified)} requested mount(s) verified", + **evidence, + ) + + +def _probe_exec_constraints( + backend_name: str, + client: Any, + session: Any, +) -> dict[str, Any]: + """Observed resource/exec constraints (cgroup caps, caps bounding, pty).""" + pty = getattr(session, "supports_pty", None) + pty_observed = pty if isinstance(pty, bool) else None + if backend_name != "docker": + return _cap( + STATUS_UNPROBED, + f"backend {backend_name!r} exposes no constraint introspection", + pty=pty_observed, + ) + attrs = _container_attrs(client, session) + if attrs is None: + return _cap( + STATUS_UNPROBED, + "container attributes unavailable", + pty=pty_observed, + ) + host_config = cast("dict[str, Any]", attrs.get("HostConfig", {}) or {}) + observed = { + "memory_bytes": host_config.get("Memory") or 0, + "nano_cpus": host_config.get("NanoCpus") or 0, + "pids_limit": host_config.get("PidsLimit") or 0, + "cap_add": sorted(str(c) for c in (host_config.get("CapAdd") or []) or []), + "security_opt": sorted(str(o) for o in (host_config.get("SecurityOpt") or []) or []), + "pty": pty_observed, + } + bounded = any( + observed[key] + for key in ("memory_bytes", "nano_cpus", "pids_limit") + if isinstance(observed[key], int | float) + ) + if not bounded: + return _cap( + STATUS_ABSENT, + "no cgroup resource limits observed on the container", + **observed, + ) + return _cap( + STATUS_SUPPORTED, + "resource limits observed on the running container", + **observed, + ) + + +def probe_session_capabilities( + *, + backend_name: str, + client: Any, + session: Any, + caido_client: Any, + caido_endpoint: Any, + bind_mounts: list[dict[str, Any]], + authorized_hosts: list[str], + relay_configured: bool, +) -> dict[str, Any]: + """Probe the live session and build the run's capability record. + + Never raises: a probe that itself fails degrades to ``unprobed`` with a + named degradation rather than crashing session setup. Preflight failures + are recorded on the record and raised by the caller. + """ + capabilities: dict[str, dict[str, Any]] = {} + probes: dict[str, Callable[[], dict[str, Any]]] = { + "exec": lambda: _probe_exec(session), + "ports": lambda: _probe_ports(session, caido_endpoint), + "proxy_capture": lambda: _probe_proxy_capture(caido_client), + "network_policy": lambda: _probe_network_policy(backend_name, client, session), + "mounts": lambda: _probe_mounts(backend_name, client, session, bind_mounts), + "exec_constraints": lambda: _probe_exec_constraints(backend_name, client, session), + } + for name, probe in probes.items(): + try: + capabilities[name] = probe() + except Exception as exc: # a failed probe is unprobed, not silent + logger.exception("capability probe %s failed", name) + capabilities[name] = _cap( + STATUS_UNPROBED, + f"probe raised {type(exc).__name__}", + ) + if relay_configured: + # The scoped relay grant rides a read-only mount into the container + # and the bridge config was verified during proxy bootstrap; both are + # probed above. A requested relay with an unverifiable grant mount is + # a degradation, not a silent assumption. + capabilities["scoped_relay"] = _cap( + STATUS_SUPPORTED + if capabilities.get("mounts", {}).get("status") == STATUS_SUPPORTED + else STATUS_UNPROBED, + "relay configured; grant mount verified via mounts probe", + ) + record: dict[str, Any] = { + "schema": CAPABILITY_RECORD_SCHEMA, + "backend": backend_name, + "probed_at": datetime.now(UTC).isoformat(), + "authorized_hosts": sorted(authorized_hosts), + "capabilities": capabilities, + } + record["preflight"] = evaluate_preflight( + capabilities, + authorized_hosts=authorized_hosts, + relay_configured=relay_configured, + ) + return record + + +def evaluate_preflight( + capabilities: dict[str, dict[str, Any]], + *, + authorized_hosts: list[str], + relay_configured: bool, +) -> dict[str, Any]: + """Map probed capabilities to the controls the run requires. + + ``failures`` are required controls resting on a probed-absent + capability — the caller must abort session setup. ``degradations`` are + named honesty markers: work continues, but the run record shows exactly + which guarantee could not be proven. + """ + failures: list[dict[str, Any]] = [] + degradations: list[dict[str, Any]] = [] + + def _entry(name: str, control: str) -> dict[str, Any]: + cap = capabilities.get(name) or {} + return { + "capability": name, + "control": control, + "status": cap.get("status"), + "detail": cap.get("detail"), + } + + def _status(name: str) -> str: + return str((capabilities.get(name) or {}).get("status") or STATUS_UNPROBED) + + # Required controls — probed absence fails the run outright. + for name, control in ( + ("exec", "agent_exec"), + ("ports", "proxy_channel"), + ("proxy_capture", "traffic_capture"), + ): + if _status(name) == STATUS_ABSENT: + failures.append(_entry(name, control)) + elif _status(name) == STATUS_UNPROBED: + degradations.append(_entry(name, control)) + + if _status("network_policy") == STATUS_ABSENT: + failures.append(_entry("network_policy", "deny_by_default_egress")) + elif _status("network_policy") == STATUS_UNPROBED: + degradations.append(_entry("network_policy", "deny_by_default_egress")) + + mounts_status = _status("mounts") + if mounts_status == STATUS_ABSENT: + # The egress policy is delivered on a read-only mount. Without it the + # in-sandbox replay guard fails closed — no scoped replay at all. + if authorized_hosts: + failures.append(_entry("mounts", "scoped_replay")) + else: + degradations.append(_entry("mounts", "egress_policy_delivery")) + elif mounts_status == STATUS_UNPROBED: + degradations.append(_entry("mounts", "egress_policy_delivery")) + + if relay_configured and mounts_status != STATUS_SUPPORTED: + degradations.append(_entry("scoped_relay", "relay_grant_delivery")) + + if _status("exec_constraints") in (STATUS_ABSENT, STATUS_UNPROBED): + degradations.append(_entry("exec_constraints", "resource_limits")) + + return {"degradations": degradations, "failures": failures} diff --git a/lyrashield/runtime/session_manager.py b/lyrashield/runtime/session_manager.py index 8f59cc6b..505ea4f1 100644 --- a/lyrashield/runtime/session_manager.py +++ b/lyrashield/runtime/session_manager.py @@ -23,6 +23,10 @@ from lyrashield.runtime.attachments import public_manifest, stage_attachments from lyrashield.runtime.backends import get_backend from lyrashield.runtime.caido_bootstrap import bootstrap_caido +from lyrashield.runtime.capabilities import ( + SandboxPreflightError, + probe_session_capabilities, +) from lyrashield.runtime.docker_client import host_gateway_enabled from lyrashield.runtime.local_dir_staging import stage_symlink_safe_dir from lyrashield.tools.proxy import caido_api @@ -436,6 +440,26 @@ def write_relay_upstream(relay_proxy: str) -> tuple[dict[str, Any], str]: return {"source": str(upstream), "target": _RELAY_UPSTREAM_TARGET, "read_only": True}, host_dir +def _enforce_sandbox_preflight(scan_id: str, capabilities: dict[str, Any]) -> None: + """Log named degradations; raise when a required control is unmet.""" + for degradation in capabilities["preflight"]["degradations"]: + logger.warning( + "sandbox preflight degradation for scan %s: %s=%s (%s)", + scan_id, + degradation.get("capability"), + degradation.get("status"), + degradation.get("detail"), + ) + failures = capabilities["preflight"]["failures"] + if failures: + raise SandboxPreflightError( + f"sandbox preflight failed for scan {scan_id}: " + + "; ".join( + f"{f['control']} (capability {f['capability']}={f['status']})" for f in failures + ) + ) + + async def create_or_reuse( # noqa: PLR0912, PLR0915 scan_id: str, *, @@ -508,6 +532,10 @@ async def create_or_reuse( # noqa: PLR0912, PLR0915 client: Any | None = None session: Any | None = None caido_client: Any | None = None + # A fresh session starts a fresh per-request scope-decision ledger: + # denials/admissions recorded by the replay guard belong to exactly + # one run's evidence. + caido_api.clear_scope_decisions() try: entries, bind_mounts, staged_dirs, extra_path_grants = build_session_entries( local_sources @@ -585,6 +613,22 @@ async def create_or_reuse( # noqa: PLR0912, PLR0915 scan_id=scan_id, authorized_hosts=authorized_hosts, ) + + # Capability probe: record what the backend actually delivered — + # never claim a control that was assumed but not probed. A + # required control resting on a probed-absent capability fails + # preflight; an unprobed one becomes a named degradation. + capabilities = probe_session_capabilities( + backend_name=backend_name, + client=client, + session=session, + caido_client=caido_client, + caido_endpoint=caido_endpoint, + bind_mounts=bind_mounts, + authorized_hosts=sorted(authorized_hosts), + relay_configured=bool(environment.get("STRIX_TARGET_RELAY")), + ) + _enforce_sandbox_preflight(scan_id, capabilities) except BaseException as startup_error: # One ownership scope: everything allocated above is released # here — the created sandbox first (the resource the reaper @@ -682,6 +726,7 @@ async def create_or_reuse( # noqa: PLR0912, PLR0915 "attachments_dir": attachments_dir, # Provenance shape of what was actually staged (host paths removed). "attachment_manifest": public_manifest(list(attachments or [])), + "sandbox_capabilities": capabilities, } async with _CACHE_LOCK: _SESSION_CACHE[scan_id] = bundle diff --git a/lyrashield/tools/proxy/caido_api.py b/lyrashield/tools/proxy/caido_api.py index 43939def..da11150f 100644 --- a/lyrashield/tools/proxy/caido_api.py +++ b/lyrashield/tools/proxy/caido_api.py @@ -6,9 +6,11 @@ import dataclasses import ipaddress import json +import logging import os import re import socket +import threading import time import urllib.request from pathlib import Path @@ -31,6 +33,8 @@ from caido_sdk_client import Client as CaidoClient +logger = logging.getLogger(__name__) + RequestPart = Literal["request", "response"] SortBy = Literal[ "timestamp", @@ -218,6 +222,155 @@ def _private_range_block_reason(hostname: str) -> str | None: return None +def _host_resolves_private(hostname: str) -> bool: + """True when the host is a private-range IP or resolves into one.""" + for raw in _resolve_hostname_ips(hostname): + try: + ip = ipaddress.ip_address(raw) + except ValueError: + continue + if any(ip in net for net in _PRIVATE_NETWORKS): + return True + return False + + +def _host_in_authorized_scope(hostname: str, authorized_hosts: frozenset[str]) -> bool: + """Match a replay destination against the recorded authorized host set. + + Mirrors the default Caido scope allowlist: a bare host admits itself and + its subdomains (``*.host``); an IP literal admits only itself — IP scope + never widens to a name. + """ + hostname = hostname.lower().rstrip(".") + for allowed in authorized_hosts: + if hostname == allowed: + return True + try: + ipaddress.ip_address(allowed) + continue + except ValueError: + pass + if hostname.endswith(f".{allowed}"): + return True + return False + + +def _authorized_scope_block_reason(hostname: str) -> str | None: + """Deny replay destinations outside the recorded authorized host set. + + Only applies when a trusted (or fail-closed) egress policy exists — + without a recorded scope there is nothing to violate and the legacy + blocklists still apply. ``allow_private_egress`` widens scope to + private-range destinations, never to arbitrary public hosts. + """ + policy = load_egress_policy() + if policy is None: + return None + if _host_in_authorized_scope(hostname, policy.authorized_hosts): + return None + if policy.allow_private_egress and _host_resolves_private(hostname): + return None + return ( + f"host {hostname!r} is outside the recorded authorized scope " + "(the run's egress policy authorizes only its own target hosts)" + ) + + +# --- Per-request scope decision ledger --------------------------------------- +# +# Every replay admission decision (admitted/denied) is recorded in-process. +# Denials are scope-violation evidence: a bounded entry list plus a dropped +# counter (the ledger must never grow without bound). Admissions are counted +# per host only — the request itself already lives in the proxy project. +# ``ReportState`` drains this ledger into run.json (schema 1.1); inside the +# sandbox the denial log line is the durable trace. +_SCOPE_VIOLATION_LIMIT = 200 +_SCOPE_HOST_LEDGER_LIMIT = 1_000 + +_scope_ledger_lock = threading.Lock() +_scope_ledger: dict[str, Any] = { + "violations": [], + "dropped": 0, + "admitted_hosts": {}, +} + + +def _evidence_url(url: str) -> str: + """URL shape for evidence: scheme/host/path only — never credentials or query.""" + try: + parsed = urlparse(url) + host = parsed.hostname or "" + port = parsed.port + except ValueError: + return "" + if not host: + return "" + netloc = f"[{host}]" if ":" in host else host + if port: + netloc = f"{netloc}:{port}" + return f"{parsed.scheme}://{netloc}{parsed.path or '/'}"[:512] + + +def _record_scope_decision( + url: str, + *, + method: str, + admitted: bool, + rule: str, + reason: str | None = None, +) -> None: + """Record one replay admission decision (bounded, process-local).""" + try: + host = (urlparse(url).hostname or "").lower().rstrip(".") + except ValueError: + host = "" + with _scope_ledger_lock: + violations: list[dict[str, Any]] = _scope_ledger["violations"] + admitted_hosts: dict[str, int] = _scope_ledger["admitted_hosts"] + if admitted: + if host and (len(admitted_hosts) < _SCOPE_HOST_LEDGER_LIMIT or host in admitted_hosts): + admitted_hosts[host] = admitted_hosts.get(host, 0) + 1 + return + if len(violations) < _SCOPE_VIOLATION_LIMIT: + violations.append( + { + "at": time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime()), + "method": method.upper()[:16], + "host": host, + "url": _evidence_url(url), + "rule": rule, + "reason": (reason or "")[:500], + } + ) + else: + _scope_ledger["dropped"] += 1 + logger.warning( + "scope-violation: replay %s %s denied (%s: %s)", + method.upper(), + host or "", + rule, + reason, + ) + + +def get_scope_decisions() -> dict[str, Any]: + """Snapshot the scope-decision ledger without mutating it.""" + with _scope_ledger_lock: + return { + "violations": [dict(v) for v in _scope_ledger["violations"]], + "dropped": _scope_ledger["dropped"], + "admitted_hosts": dict(_scope_ledger["admitted_hosts"]), + } + + +def clear_scope_decisions() -> None: + """Reset the ledger — called once per fresh sandbox session.""" + with _scope_ledger_lock: + _scope_ledger["violations"].clear() + _scope_ledger["dropped"] = 0 + _scope_ledger["admitted_hosts"].clear() + + _BLOCKED_METADATA_HOSTS = frozenset( {"metadata.google.internal", "metadata.google.internal.", "metadata.google", "metadata.google."} ) @@ -246,34 +399,49 @@ def _host_gateway_allowed() -> bool: } -def _check_replay_url_host(url: str) -> str | None: - """Return a human-readable block reason, or None if the host is allowed.""" +def _replay_denial(url: str) -> tuple[str, str] | None: + """Return ``(reason, rule)`` for a denied replay request, else ``None``.""" parsed = urlparse(url) if parsed.scheme.lower() not in {"http", "https"}: - return f"non-HTTP scheme {parsed.scheme!r}" + return (f"non-HTTP scheme {parsed.scheme!r}", "non_http_scheme") hostname = (parsed.hostname or "").lower() if not hostname: return None if hostname in _BLOCKED_METADATA_HOSTS: - return f"cloud metadata host {hostname!r}" + return (f"cloud metadata host {hostname!r}", "cloud_metadata") if not _host_gateway_allowed() and hostname in { "host.docker.internal", "host.docker.internal.", }: - return "host.docker.internal (set STRIX_SANDBOX_ALLOW_HOST_GATEWAY=1 to allow)" + return ( + "host.docker.internal (set STRIX_SANDBOX_ALLOW_HOST_GATEWAY=1 to allow)", + "host_gateway", + ) try: ip = ipaddress.ip_address(hostname) except ValueError: ip = None if ip is not None: if ip in _BLOCKED_METADATA_IPS: - return f"cloud metadata IP {ip}" + return (f"cloud metadata IP {ip}", "cloud_metadata") for net in _LINK_LOCAL_NETWORKS: if ip in net: - return f"link-local address {ip}" + return (f"link-local address {ip}", "link_local") # Private-range guard also resolves DNS names, so a hostname that points # into RFC1918/loopback space is caught the same way as a literal IP. - return _private_range_block_reason(hostname) + private_reason = _private_range_block_reason(hostname) + if private_reason is not None: + return (private_reason, "private_range") + scope_reason = _authorized_scope_block_reason(hostname) + if scope_reason is not None: + return (scope_reason, "outside_authorized_scope") + return None + + +def _check_replay_url_host(url: str) -> str | None: + """Return a human-readable block reason, or None if the host is allowed.""" + denial = _replay_denial(url) + return denial[0] if denial is not None else None def caido_url() -> str: @@ -456,9 +624,12 @@ def build_raw_request( parsed = urlparse(url) if not parsed.scheme or not parsed.netloc: raise ValueError(f"Invalid URL: {url}") - block_reason = _check_replay_url_host(url) - if block_reason: + denial = _replay_denial(url) + if denial is not None: + block_reason, rule = denial + _record_scope_decision(url, method=method, admitted=False, rule=rule, reason=block_reason) raise ValueError(f"URL is blocked ({block_reason}): {url}") + _record_scope_decision(url, method=method, admitted=True, rule="admitted") is_tls = parsed.scheme.lower() == "https" host = parsed.hostname or "" port = parsed.port or (443 if is_tls else 80) @@ -1026,8 +1197,10 @@ async def view_sitemap_entry(entry_id: str) -> dict[str, Any]: "SitemapDepth", "SortBy", "SortOrder", + "clear_scope_decisions", "close_client", "get_client", + "get_scope_decisions", "list_requests", "list_sitemap", "repeat_request", diff --git a/scripts/customer-branding-allowlist.json b/scripts/customer-branding-allowlist.json index 90e1c6c6..1d65ea8d 100644 --- a/scripts/customer-branding-allowlist.json +++ b/scripts/customer-branding-allowlist.json @@ -352,7 +352,9 @@ "from strix.config.loader import load_settings", "from strix.core.paths import run_dir_for, runtime_state_dir", " tool_version=_strix_version(),", - " from strix.report.coverage import read_agent_graph" + " from strix.report.coverage import read_agent_graph", + " from strix.tools.coverage.tools import get_coverage_entries", + "from strix.core.paths import run_dir_for" ], "lyrashield/artifacts/usage.py": [ "# Modifications \u00a9 2026 LyraShield; based on upstream Strix (Apache-2.0)" @@ -516,7 +518,8 @@ " environment[\"STRIX_TARGET_RELAY\"] = \"1\"", " if environment.get(\"STRIX_TARGET_RELAY\"):", " environment[\"STRIX_RUN_ID\"] = scan_id", - " target_relay=bool(environment.get(\"STRIX_TARGET_RELAY\"))," + " target_relay=bool(environment.get(\"STRIX_TARGET_RELAY\")),", + " relay_configured=bool(environment.get(\"STRIX_TARGET_RELAY\"))," ], "lyrashield/telemetry/README.md": [ "- `STRIX_NO_UPDATE_CHECK=1` \u2014 disables the upstream self-update network check.", @@ -569,14 +572,14 @@ "``/workspace/.strix/tool-output/.txt``; the agent sees a head + tail slice" ], "lyrashield/tools/proxy/caido_api.py": [ - " return \"host.docker.internal (set STRIX_SANDBOX_ALLOW_HOST_GATEWAY=1 to allow)\"", " # STRIX_RUN_ID is set (inside the container). A wrong-run policy is", " ``STRIX_SANDBOX_ALLOW_PRIVATE_EGRESS`` opt-in is honored). When a policy", " expected_run_id = os.environ.get(\"STRIX_RUN_ID\", \"\").strip()", " final_headers.setdefault(\"User-Agent\", \"strix\")", " return os.environ.get(\"STRIX_CAIDO_URL\", _DEFAULT_CAIDO_URL).rstrip(\"/\")", " return os.environ.get(\"STRIX_SANDBOX_ALLOW_HOST_GATEWAY\", \"\").strip().lower() in {", - "_PRIVATE_EGRESS_OPT_IN_ENV = \"STRIX_SANDBOX_ALLOW_PRIVATE_EGRESS\"" + "_PRIVATE_EGRESS_OPT_IN_ENV = \"STRIX_SANDBOX_ALLOW_PRIVATE_EGRESS\"", + " \"host.docker.internal (set STRIX_SANDBOX_ALLOW_HOST_GATEWAY=1 to allow)\"," ], "lyrashield/tools/reporting/tool.py": [ "# Modifications \u00a9 2026 LyraShield; based on upstream Strix (Apache-2.0)", @@ -754,6 +757,9 @@ "from strix.core.paths import runtime_state_dir", " from strix.report.coverage import build_coverage_document as _build_substrate", " from strix.tools.coverage.tools import get_coverage_entries" + ], + "lyrashield/artifacts/quality.py": [ + "from strix.report.coverage import agents_from_graph" ] } } diff --git a/tests/test_attachments.py b/tests/test_attachments.py index 55889370..54dad979 100644 --- a/tests/test_attachments.py +++ b/tests/test_attachments.py @@ -324,6 +324,13 @@ async def no_caido(*_args: Any, **_kwargs: Any) -> None: ) monkeypatch.setattr(session_manager, "get_backend", lambda _name: backend) monkeypatch.setattr(session_manager, "bootstrap_caido", no_caido) + # These fakes exercise mount/lifecycle wiring, not capability probing — a + # real probe would (correctly) fail the stub session on exec=absent. + monkeypatch.setattr( + session_manager, + "probe_session_capabilities", + lambda **_kwargs: {"preflight": {"degradations": [], "failures": []}}, + ) session_manager._SESSION_CACHE.pop(scan_id, None) await session_manager.create_or_reuse( diff --git a/tests/test_benchmark_scoring.py b/tests/test_benchmark_scoring.py index 082bba8e..c2b42037 100644 --- a/tests/test_benchmark_scoring.py +++ b/tests/test_benchmark_scoring.py @@ -126,6 +126,58 @@ def git(*args: str) -> None: assert after["trackedDiffSha256"] != before["trackedDiffSha256"] +def test_engine_run_receipt_surfaces_run_record_fields(tmp_path: Path) -> None: + run_dir = tmp_path / "strix_runs" / "scan-1" + run_dir.mkdir(parents=True) + (run_dir / "run.json").write_text( + json.dumps( + { + "schema_version": "1.1", + "sandbox_capabilities": { + "backend": "docker", + "capabilities": { + "exec": {"status": "supported"}, + "mounts": {"status": "unprobed"}, + }, + "preflight": { + "degradations": [{"control": "egress_policy_delivery"}], + "failures": [{"control": "agent_exec"}], + }, + }, + "scan_quality": { + "observed": {"findings_filed": 2}, + "declared": {"coverage_entries": 1}, + "surfaces": [{"surface": "a"}, {"surface": "b"}], + "unassessed": ["c"], + }, + "scope_violations": {"total": 4}, + "evidence_export": {"status": "exported"}, + } + ) + ) + receipt = run.engine_run_receipt(tmp_path, "scan-1") + assert receipt["schemaVersion"] == "1.1" + caps = receipt["sandboxCapabilities"] + assert caps["backend"] == "docker" + assert caps["statuses"] == {"exec": "supported", "mounts": "unprobed"} + assert caps["preflightDegradations"] == ["egress_policy_delivery"] + assert caps["preflightFailures"] == ["agent_exec"] + assert receipt["scanQuality"]["observed"]["findings_filed"] == 2 + assert receipt["scanQuality"]["surfaces"] == 2 + assert receipt["scanQuality"]["unassessed"] == 1 + assert receipt["scopeViolations"] == 4 + assert receipt["evidenceExport"] == "exported" + + +def test_engine_run_receipt_missing_and_unreadable(tmp_path: Path) -> None: + assert run.engine_run_receipt(tmp_path, "no-scan") is None + bad = tmp_path / "strix_runs" / "bad-scan" + bad.mkdir(parents=True) + (bad / "run.json").write_text("{not json") + receipt = run.engine_run_receipt(tmp_path, "bad-scan") + assert "error" in receipt + + def test_scoring_requires_original_corpus_and_fixture_bytes(tmp_path: Path) -> None: corpus = { "name": "fixture", diff --git a/tests/test_replay_scope_admission.py b/tests/test_replay_scope_admission.py new file mode 100644 index 00000000..997b41d6 --- /dev/null +++ b/tests/test_replay_scope_admission.py @@ -0,0 +1,152 @@ +"""Per-request scope admission against the recorded authorized host set. + +Replay requests must be checked against the egress policy recorded for the +run — not the mutable proxy config. Requests outside the recorded scope are +denied and logged as scope-violation evidence in the decision ledger. +""" + +from __future__ import annotations + +import contextlib +from pathlib import Path +from typing import TYPE_CHECKING, Any + + +if TYPE_CHECKING: + from collections.abc import Iterator + +import pytest + +from lyrashield.runtime.session_manager import write_egress_policy +from lyrashield.tools.proxy import caido_api + + +@pytest.fixture(autouse=True) +def _ledger(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + """Reset the scope-decision ledger and policy/env state per test.""" + caido_api.clear_scope_decisions() + monkeypatch.delenv("LYRASHIELD_EGRESS_POLICY", raising=False) + monkeypatch.delenv("STRIX_RUN_ID", raising=False) + monkeypatch.delenv("STRIX_SANDBOX_ALLOW_PRIVATE_EGRESS", raising=False) + # Behave as in-container: the mounted policy file is the only authority. + monkeypatch.setattr(caido_api, "_in_container", lambda: True) + monkeypatch.setattr(caido_api, "_path_on_readonly_mount", lambda _p: True) + yield + caido_api.clear_scope_decisions() + + +def _policy( + monkeypatch: pytest.MonkeyPatch, + authorized_hosts: list[str], + *, + allow_private_egress: bool = False, +) -> None: + _mount, host_dir = write_egress_policy( + "scan-scope", + set(authorized_hosts), + allow_private_egress=allow_private_egress, + ) + monkeypatch.setenv("STRIX_RUN_ID", "scan-scope") + monkeypatch.setenv("LYRASHIELD_EGRESS_POLICY", str(Path(host_dir) / "policy.json")) + + +def _send(url: str, method: str = "GET") -> Any: + return caido_api.build_raw_request(method=method, url=url, headers={}, body="") + + +def test_in_scope_host_admitted_and_recorded(monkeypatch: Any) -> None: + _policy(monkeypatch, ["app.example.com"]) + _conn, raw = _send("https://app.example.com/login") + assert raw.startswith(b"GET /login") + decisions = caido_api.get_scope_decisions() + assert decisions["admitted_hosts"] == {"app.example.com": 1} + assert decisions["violations"] == [] + + +def test_out_of_scope_public_host_denied_as_violation(monkeypatch: Any) -> None: + _policy(monkeypatch, ["app.example.com"]) + with pytest.raises(ValueError, match="outside the recorded authorized scope"): + _send("https://evil.example.net/?token=secret") + decisions = caido_api.get_scope_decisions() + assert len(decisions["violations"]) == 1 + entry = decisions["violations"][0] + assert entry["rule"] == "outside_authorized_scope" + assert entry["host"] == "evil.example.net" + # Evidence URLs never carry credentials or query strings. + assert "token=secret" not in entry["url"] + assert entry["url"] == "https://evil.example.net/" + + +def test_subdomain_of_authorized_host_in_scope(monkeypatch: Any) -> None: + _policy(monkeypatch, ["example.com"]) + _conn, _raw = _send("https://api.example.com/") + with pytest.raises(ValueError, match="outside the recorded authorized scope"): + _send("https://notexample.com/") + + +def test_authorized_ip_matches_exactly_not_by_widening(monkeypatch: Any) -> None: + _policy(monkeypatch, ["203.0.113.10"]) + _conn, _raw = _send("http://203.0.113.10/") + with pytest.raises(ValueError, match="outside the recorded authorized scope"): + _send("http://203.0.113.11/") + + +def test_empty_authorized_scope_denies_all_replay(monkeypatch: Any) -> None: + _policy(monkeypatch, []) + for url in ("https://example.com/", "https://api.target.io/"): + with pytest.raises(ValueError, match="outside the recorded authorized scope"): + _send(url) + decisions = caido_api.get_scope_decisions() + assert len(decisions["violations"]) == 2 + + +def test_private_egress_opt_in_extends_scope_to_private_only(monkeypatch: Any) -> None: + _policy(monkeypatch, [], allow_private_egress=True) + _conn, _raw = _send("http://10.0.0.5/") + with pytest.raises(ValueError, match="outside the recorded authorized scope"): + _send("https://example.com/") + + +def test_no_policy_keeps_legacy_guard() -> None: + """Without a recorded scope the blocklists still apply; public hosts pass.""" + _conn, _raw = _send("https://example.com/") + with pytest.raises(ValueError, match="private-range"): + _send("http://10.0.0.5/") + decisions = caido_api.get_scope_decisions() + assert decisions["admitted_hosts"] == {"example.com": 1} + assert decisions["violations"][0]["rule"] == "private_range" + + +def test_fail_closed_policy_denies_everything(tmp_path: Path, monkeypatch: Any) -> None: + """An untrusted/malformed policy mounts fail-closed: nothing is in scope.""" + policy_dir = tmp_path / "rw-policy" + policy_dir.mkdir(mode=0o700) + policy_dir.chmod(0o700) + (policy_dir / "policy.json").write_text('{"version": 1, "scan_id": "scan-scope"}') + monkeypatch.setenv("STRIX_RUN_ID", "scan-scope") + monkeypatch.setenv("LYRASHIELD_EGRESS_POLICY", str(policy_dir / "policy.json")) + with pytest.raises(ValueError): + _send("https://app.example.com/") + decisions = caido_api.get_scope_decisions() + assert decisions["violations"] + assert decisions["violations"][0]["rule"] == "outside_authorized_scope" + + +def test_violation_ledger_is_bounded(monkeypatch: Any) -> None: + monkeypatch.setattr(caido_api, "_SCOPE_VIOLATION_LIMIT", 5) + caido_api.clear_scope_decisions() + for i in range(8): + with contextlib.suppress(ValueError): + _send(f"http://169.254.169.{i}/") + decisions = caido_api.get_scope_decisions() + assert len(decisions["violations"]) == 5 + assert decisions["dropped"] == 3 + + +def test_clear_scope_decisions_resets_ledger(monkeypatch: Any) -> None: + _policy(monkeypatch, ["app.example.com"]) + with pytest.raises(ValueError): + _send("https://evil.example.net/") + caido_api.clear_scope_decisions() + decisions = caido_api.get_scope_decisions() + assert decisions == {"violations": [], "dropped": 0, "admitted_hosts": {}} diff --git a/tests/test_sandbox_capabilities.py b/tests/test_sandbox_capabilities.py new file mode 100644 index 00000000..537bf359 --- /dev/null +++ b/tests/test_sandbox_capabilities.py @@ -0,0 +1,323 @@ +"""Capability-probe matrix and preflight semantics for sandbox sessions. + +The probe must reflect what the backend verifiably delivered — supported, +absent, or unprobed — and never assume. A required control resting on an +absent capability fails preflight; an unprobed one becomes a named +degradation rather than a silent gap. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest + +from lyrashield.runtime import session_manager +from lyrashield.runtime.capabilities import ( + STATUS_ABSENT, + STATUS_SUPPORTED, + STATUS_UNPROBED, + SandboxPreflightError, + evaluate_preflight, + probe_session_capabilities, +) + + +_NETWORK = "lyrashield-sandbox" +_POLICY_TARGET = "/run/lyrashield-egress/policy.json" + + +def _docker_attrs( + *, + network: str = _NETWORK, + internal_attached: list[str] | None = None, + mounts: list[dict[str, Any]] | None = None, + memory: int = 2 << 30, + nano_cpus: int = 2_000_000_000, + pids: int = 512, +) -> dict[str, Any]: + return { + "HostConfig": { + "NetworkMode": network, + "Memory": memory, + "NanoCpus": nano_cpus, + "PidsLimit": pids, + "CapAdd": ["NET_RAW", "NET_ADMIN"], + "SecurityOpt": ["no-new-privileges"], + }, + "NetworkSettings": { + "Networks": {name: {} for name in (internal_attached or [network])}, + }, + "Mounts": mounts + if mounts is not None + else [{"Type": "bind", "Destination": _POLICY_TARGET, "RW": False}], + } + + +class _FakeContainers: + def __init__(self, attrs: dict[str, Any]) -> None: + self._attrs = attrs + + def get(self, _container_id: str) -> Any: + return SimpleNamespace(attrs=self._attrs) + + +class _FakeNetworks: + def __init__(self, internal: bool) -> None: + self._internal = internal + + def get(self, _name: str) -> Any: + return SimpleNamespace(attrs={"Internal": self._internal}) + + +def _docker_client(attrs: dict[str, Any], *, internal: bool = True) -> Any: + return SimpleNamespace( + docker_client=SimpleNamespace( + containers=_FakeContainers(attrs), + networks=_FakeNetworks(internal), + ) + ) + + +class _Session: + def __init__(self, container_id: str | None = "abc123") -> None: + self._inner = SimpleNamespace(container_id=container_id) + self.supports_pty = False + + async def exec(self, *_args: Any, **_kwargs: Any) -> Any: + return SimpleNamespace(ok=lambda: True, stdout=b"", stderr=b"", exit_code=0) + + async def resolve_exposed_port(self, _port: int) -> Any: + return SimpleNamespace(tls=False, host="127.0.0.1", port=48080) + + +def _probe(**overrides: Any) -> dict[str, Any]: + kwargs: dict[str, Any] = { + "backend_name": "docker", + "client": _docker_client(_docker_attrs()), + "session": _Session(), + "caido_client": SimpleNamespace(), + "caido_endpoint": SimpleNamespace(tls=False, host="127.0.0.1", port=48080), + "bind_mounts": [ + {"source": "/var/empty/policy.json", "target": _POLICY_TARGET, "read_only": True} + ], + "authorized_hosts": ["app.example.com"], + "relay_configured": False, + } + kwargs.update(overrides) + return probe_session_capabilities(**kwargs) + + +@pytest.fixture(autouse=True) +def _sandbox_network_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("STRIX_DOCKER_SANDBOX_NETWORK", _NETWORK) + + +def test_probe_all_supported_on_verified_docker() -> None: + record = _probe() + caps = record["capabilities"] + assert record["backend"] == "docker" + assert record["authorized_hosts"] == ["app.example.com"] + for name in ( + "exec", + "ports", + "proxy_capture", + "network_policy", + "mounts", + "exec_constraints", + ): + assert caps[name]["status"] == STATUS_SUPPORTED, (name, caps[name]) + assert record["preflight"] == {"degradations": [], "failures": []} + + +def test_probe_non_docker_marks_introspection_unprobed() -> None: + record = _probe(backend_name="remote", client=SimpleNamespace()) + caps = record["capabilities"] + # Session-API capabilities are still probed on any backend. + assert caps["exec"]["status"] == STATUS_SUPPORTED + assert caps["ports"]["status"] == STATUS_SUPPORTED + # Backend-introspection capabilities are unprobed — never assumed. + for name in ("network_policy", "mounts", "exec_constraints"): + assert caps[name]["status"] == STATUS_UNPROBED, (name, caps[name]) + controls = {d["control"] for d in record["preflight"]["degradations"]} + assert {"deny_by_default_egress", "egress_policy_delivery", "resource_limits"} <= controls + assert record["preflight"]["failures"] == [] + + +def test_probe_absent_exec_is_preflight_failure() -> None: + record = _probe(session=SimpleNamespace()) + assert record["capabilities"]["exec"]["status"] == STATUS_ABSENT + controls = {f["control"] for f in record["preflight"]["failures"]} + assert "agent_exec" in controls + + +def test_probe_absent_proxy_client_is_preflight_failure() -> None: + record = _probe(caido_client=None) + assert record["capabilities"]["proxy_capture"]["status"] == STATUS_ABSENT + controls = {f["control"] for f in record["preflight"]["failures"]} + assert "traffic_capture" in controls + + +def test_probe_missing_policy_mount_fails_when_targets_exist() -> None: + attrs = _docker_attrs(mounts=[]) + record = _probe(client=_docker_client(attrs), authorized_hosts=["app.example.com"]) + assert record["capabilities"]["mounts"]["status"] == STATUS_ABSENT + assert _POLICY_TARGET in record["capabilities"]["mounts"]["evidence"]["missing"] + controls = {f["control"] for f in record["preflight"]["failures"]} + assert "scoped_replay" in controls + + +def test_probe_missing_policy_mount_degrades_without_targets() -> None: + attrs = _docker_attrs(mounts=[]) + record = _probe(client=_docker_client(attrs), authorized_hosts=[]) + assert record["capabilities"]["mounts"]["status"] == STATUS_ABSENT + assert record["preflight"]["failures"] == [] + controls = {d["control"] for d in record["preflight"]["degradations"]} + assert "egress_policy_delivery" in controls + + +def test_probe_writable_policy_mount_is_absent() -> None: + attrs = _docker_attrs(mounts=[{"Type": "bind", "Destination": _POLICY_TARGET, "RW": True}]) + record = _probe(client=_docker_client(attrs)) + assert record["capabilities"]["mounts"]["status"] == STATUS_ABSENT + assert _POLICY_TARGET in record["capabilities"]["mounts"]["evidence"]["writable"] + + +def test_probe_non_internal_network_is_absent() -> None: + record = _probe(client=_docker_client(_docker_attrs(), internal=False)) + assert record["capabilities"]["network_policy"]["status"] == STATUS_ABSENT + controls = {f["control"] for f in record["preflight"]["failures"]} + assert "deny_by_default_egress" in controls + + +def test_probe_extra_network_attachment_is_absent() -> None: + attrs = _docker_attrs(internal_attached=[_NETWORK, "bridge"]) + record = _probe(client=_docker_client(attrs)) + assert record["capabilities"]["network_policy"]["status"] == STATUS_ABSENT + + +def test_probe_unbounded_resources_is_degradation_not_failure() -> None: + attrs = _docker_attrs(memory=0, nano_cpus=0, pids=0) + record = _probe(client=_docker_client(attrs)) + assert record["capabilities"]["exec_constraints"]["status"] == STATUS_ABSENT + assert record["preflight"]["failures"] == [] + controls = {d["control"] for d in record["preflight"]["degradations"]} + assert "resource_limits" in controls + + +def test_probe_exception_becomes_unprobed_not_crash() -> None: + class BrokenSession: + @property + def exec(self) -> Any: + raise RuntimeError("boom") + + async def resolve_exposed_port(self, _port: int) -> Any: + return SimpleNamespace(tls=False, host="127.0.0.1", port=48080) + + # A probe that itself raises records ``unprobed`` — never crashes setup. + record = _probe(session=BrokenSession()) + assert record["capabilities"]["exec"]["status"] == STATUS_UNPROBED + assert "probe raised" in record["capabilities"]["exec"]["detail"] + + +def test_evaluate_preflight_missing_capabilities_are_unprobed() -> None: + preflight = evaluate_preflight({}, authorized_hosts=["a.example.com"], relay_configured=False) + # Nothing probed → nothing fails outright, but every required control is + # named as degraded rather than silently assumed. + assert preflight["failures"] == [] + controls = {d["control"] for d in preflight["degradations"]} + assert { + "agent_exec", + "proxy_channel", + "traffic_capture", + "deny_by_default_egress", + "egress_policy_delivery", + "resource_limits", + } <= controls + + +@pytest.mark.asyncio +async def test_create_or_reuse_records_capabilities( + monkeypatch: pytest.MonkeyPatch, +) -> None: + attrs = _docker_attrs() + client = _docker_client(attrs) + session = _Session() + + async def backend(**_kwargs: Any) -> tuple[Any, Any]: + return client, session + + async def caido(*_args: Any, **_kwargs: Any) -> Any: + return SimpleNamespace() + + scan_id = "cap-probe-scan" + monkeypatch.setattr( + session_manager, + "load_settings", + lambda: SimpleNamespace(runtime=SimpleNamespace(backend="docker")), + ) + monkeypatch.setattr(session_manager, "get_backend", lambda _name: backend) + monkeypatch.setattr(session_manager, "bootstrap_caido", caido) + session_manager._SESSION_CACHE.pop(scan_id, None) + try: + bundle = await session_manager.create_or_reuse( + scan_id, + image="test-image", + local_sources=[], + targets=[ + {"type": "web_application", "details": {"target_url": "https://app.example.com"}} + ], + ) + finally: + session_manager._SESSION_CACHE.pop(scan_id, None) + + caps = bundle["sandbox_capabilities"] + assert caps["backend"] == "docker" + assert caps["capabilities"]["exec"]["status"] == STATUS_SUPPORTED + assert caps["capabilities"]["network_policy"]["status"] == STATUS_SUPPORTED + assert caps["capabilities"]["mounts"]["status"] == STATUS_SUPPORTED + assert caps["preflight"]["failures"] == [] + + +@pytest.mark.asyncio +async def test_create_or_reuse_preflight_failure_cleans_up( + monkeypatch: pytest.MonkeyPatch, +) -> None: + deleted: list[str] = [] + closed: list[str] = [] + + class NoExecSession: + async def resolve_exposed_port(self, _port: int) -> Any: + return SimpleNamespace(tls=False, host="127.0.0.1", port=48080) + + class Client: + async def delete(self, _session: Any) -> None: + deleted.append("deleted") + + class CaidoStub: + async def aclose(self) -> None: + closed.append("closed") + + async def backend(**_kwargs: Any) -> tuple[Any, Any]: + return Client(), NoExecSession() + + async def caido(*_args: Any, **_kwargs: Any) -> Any: + return CaidoStub() + + scan_id = "cap-preflight-fail" + monkeypatch.setattr( + session_manager, + "load_settings", + lambda: SimpleNamespace(runtime=SimpleNamespace(backend="docker")), + ) + monkeypatch.setattr(session_manager, "get_backend", lambda _name: backend) + monkeypatch.setattr(session_manager, "bootstrap_caido", caido) + session_manager._SESSION_CACHE.pop(scan_id, None) + try: + with pytest.raises(SandboxPreflightError, match="agent_exec"): + await session_manager.create_or_reuse(scan_id, image="test-image", local_sources=[]) + finally: + session_manager._SESSION_CACHE.pop(scan_id, None) + + assert deleted == ["deleted"] diff --git a/tests/test_scan_quality.py b/tests/test_scan_quality.py new file mode 100644 index 00000000..7e80c7c1 --- /dev/null +++ b/tests/test_scan_quality.py @@ -0,0 +1,303 @@ +"""Honest scan-quality accounting (run.json schema 1.1). + +``scan_quality`` reports what was actually exercised versus declared — +derived only from observed runtime activity and the model-declared coverage +ledger. Unexercised surfaces stay ``unassessed``; nothing is extrapolated. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from lyrashield.artifacts.quality import SCAN_QUALITY_SCHEMA, build_scan_quality +from lyrashield.artifacts.state import ReportState +from lyrashield.runtime.session_manager import write_egress_policy +from lyrashield.tools.proxy import caido_api +from strix.tools.coverage.tools import hydrate_coverage_from_disk + + +def _agent_graph(*agents: tuple[str, str]) -> dict[str, Any]: + return { + "statuses": dict(agents), + "names": {agent_id: agent_id for agent_id, _ in agents}, + "metadata": {}, + "parent_of": {}, + } + + +def test_quality_distinguishes_observed_declared_unassessed() -> None: + run_record = { + "run_id": "q1", + "status": "completed", + "sandbox_capabilities": { + "backend": "docker", + "authorized_hosts": ["app.example.com", "idle.example.com"], + "capabilities": {"exec": {"status": "supported"}}, + "preflight": {"degradations": [], "failures": []}, + }, + } + entries = [ + { + "surface": "https://app.example.com/login", + "risk_area": "xss", + "outcome": "no_issue_found", + }, + {"surface": "declared-only.example.com", "risk_area": "idor", "outcome": "reported"}, + ] + reports = [{"id": "v1", "endpoint": "https://app.example.com/search", "severity": "high"}] + decisions = { + "violations": [ + { + "at": "2026-01-01 00:00:00 UTC", + "method": "GET", + "host": "evil.example.net", + "url": "https://evil.example.net/", + "rule": "outside_authorized_scope", + "reason": "denied", + } + ], + "dropped": 0, + "admitted_hosts": {"app.example.com": 3}, + } + doc = build_scan_quality( + run_record=run_record, + agent_graph=_agent_graph(("a1", "finished"), ("a2", "crashed")), + coverage_entries=entries, + vulnerability_reports=reports, + scope_decisions=decisions, + ) + assert doc["schema"] == SCAN_QUALITY_SCHEMA + assert doc["observed"]["agents_total"] == 2 + assert doc["observed"]["agents_incomplete"] == 1 + assert doc["observed"]["findings_filed"] == 1 + assert doc["observed"]["proxy_requests_admitted"] == 3 + assert doc["observed"]["proxy_requests_denied"] == 1 + assert doc["declared"]["coverage_entries"] == 2 + + rows = {row["surface"]: row for row in doc["surfaces"]} + # Observed + declared + findings on one surface. + assert rows["app.example.com"]["assessment"] == "observed" + assert rows["app.example.com"]["admitted_requests"] == 3 + assert rows["app.example.com"]["findings"] == 1 + assert rows["app.example.com"]["declared_coverage_entries"] == 1 + # Declared-only surface stays declared, never upgraded. + assert rows["declared-only.example.com"]["assessment"] == "declared" + # An authorized host with no recorded exercise is honestly unassessed. + assert rows["idle.example.com"]["assessment"] == "unassessed" + assert "idle.example.com" in doc["unassessed"] + # A denied destination is evidence of an attempt, not of exercise. + assert rows["evil.example.net"]["assessment"] == "denied" + assert rows["evil.example.net"]["denied_requests"] == 1 + + +def test_quality_without_observations_marks_scope_unassessed() -> None: + doc = build_scan_quality( + run_record={ + "run_id": "q2", + "sandbox_capabilities": {"authorized_hosts": ["a.example.com"]}, + }, + agent_graph={}, + coverage_entries=[], + vulnerability_reports=[], + scope_decisions=None, + ) + assert doc["observed"]["agents_total"] == 0 + assert doc["unassessed"] == ["a.example.com"] + assert doc["surfaces"][0]["assessment"] == "unassessed" + + +# --------------------------------------------------------------------------- +# run.json wiring +# --------------------------------------------------------------------------- + + +@pytest.fixture +def state_1_1(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> ReportState: + monkeypatch.setenv("LYRASHIELD_RUN_RECORD_V1_1", "1") + monkeypatch.setenv("STRIX_SANDBOX_MODE", "local") + monkeypatch.chdir(tmp_path) + caido_api.clear_scope_decisions() + monkeypatch.delenv("LYRASHIELD_EGRESS_POLICY", raising=False) + monkeypatch.delenv("STRIX_RUN_ID", raising=False) + monkeypatch.setattr(caido_api, "_in_container", lambda: True) + monkeypatch.setattr(caido_api, "_path_on_readonly_mount", lambda _p: True) + return ReportState(run_name="quality-scan") + + +def _read_record(state: ReportState) -> dict[str, Any]: + return json.loads((state.get_run_dir() / "run.json").read_text()) + + +def test_run_record_1_1_carries_quality_and_violations( + state_1_1: ReportState, monkeypatch: pytest.MonkeyPatch +) -> None: + state = state_1_1 + state.set_sandbox_capabilities( + { + "schema": "lyrashield-sandbox-capabilities/1.0", + "backend": "docker", + "authorized_hosts": ["app.example.com"], + "capabilities": {"exec": {"status": "supported"}}, + "preflight": {"degradations": [], "failures": []}, + } + ) + # One denied out-of-scope request becomes persisted evidence. + _mount, host_dir = write_egress_policy("scan-q", {"app.example.com"}) + monkeypatch.setenv("STRIX_RUN_ID", "scan-q") + monkeypatch.setenv("LYRASHIELD_EGRESS_POLICY", str(Path(host_dir) / "policy.json")) + with pytest.raises(ValueError, match="outside the recorded authorized scope"): + caido_api.build_raw_request( + method="GET", url="https://evil.example.net/", headers={}, body="" + ) + + assert state.save_run_data() + record = _read_record(state) + assert record["sandbox_capabilities"]["backend"] == "docker" + violations = record["scope_violations"] + assert violations["total"] == 1 + assert violations["entries"][0]["host"] == "evil.example.net" + quality = record["scan_quality"] + assert quality["schema"] == SCAN_QUALITY_SCHEMA + assert quality["observed"]["proxy_requests_denied"] == 1 + denied = {r["surface"] for r in quality["surfaces"] if r["assessment"] == "denied"} + assert "evil.example.net" in denied + + +def test_run_record_1_0_omits_quality_fields( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.delenv("LYRASHIELD_RUN_RECORD_V1_1", raising=False) + monkeypatch.setenv("STRIX_SANDBOX_MODE", "local") + monkeypatch.chdir(tmp_path) + state = ReportState(run_name="legacy-quality") + state.set_sandbox_capabilities( + { + "schema": "lyrashield-sandbox-capabilities/1.0", + "backend": "docker", + "authorized_hosts": [], + "capabilities": {}, + "preflight": {"degradations": [], "failures": []}, + } + ) + assert state.save_run_data() + record = _read_record(state) + # Provenance is unconditional; the evidence/quality blocks are 1.1-only. + assert record["schema_version"] == "1.0" + assert record["sandbox_capabilities"]["backend"] == "docker" + assert "scan_quality" not in record + assert "scope_violations" not in record + + +def test_scope_violations_persist_across_saves_and_resume( + state_1_1: ReportState, monkeypatch: pytest.MonkeyPatch +) -> None: + state = state_1_1 + _mount, host_dir = write_egress_policy("scan-q2", {"app.example.com"}) + monkeypatch.setenv("STRIX_RUN_ID", "scan-q2") + monkeypatch.setenv("LYRASHIELD_EGRESS_POLICY", str(Path(host_dir) / "policy.json")) + + with pytest.raises(ValueError): + caido_api.build_raw_request( + method="GET", url="https://one.example.net/", headers={}, body="" + ) + assert state.save_run_data() + with pytest.raises(ValueError): + caido_api.build_raw_request( + method="GET", url="https://two.example.net/", headers={}, body="" + ) + assert state.save_run_data() + entries = _read_record(state)["scope_violations"]["entries"] + assert {e["host"] for e in entries} == {"one.example.net", "two.example.net"} + + # Resume: a fresh ReportState over the same run dir keeps prior entries — + # and same-process ledger offsets must not re-append them. + resumed = ReportState(run_name="quality-scan") + resumed.hydrate_from_run_dir() + with pytest.raises(ValueError): + caido_api.build_raw_request( + method="GET", url="https://three.example.net/", headers={}, body="" + ) + assert resumed.save_run_data() + violations = _read_record(resumed)["scope_violations"] + assert len(violations["entries"]) == 3 + assert violations["total"] == 3 + assert violations["dropped"] == 0 + assert {e["host"] for e in violations["entries"]} == { + "one.example.net", + "two.example.net", + "three.example.net", + } + + +def test_scope_violation_dropped_does_not_reaccumulate( + state_1_1: ReportState, monkeypatch: pytest.MonkeyPatch +) -> None: + """The ledger ``dropped`` is process-cumulative: repeated saves must merge + only the delta, never re-add the whole overflow (I-merge).""" + state = state_1_1 + _mount, host_dir = write_egress_policy("scan-q3", {"app.example.com"}) + monkeypatch.setenv("STRIX_RUN_ID", "scan-q3") + monkeypatch.setenv("LYRASHIELD_EGRESS_POLICY", str(Path(host_dir) / "policy.json")) + monkeypatch.setattr(caido_api, "_SCOPE_VIOLATION_LIMIT", 2) + + for host in ("a.example.net", "b.example.net", "c.example.net", "d.example.net"): + with pytest.raises(ValueError): + caido_api.build_raw_request(method="GET", url=f"https://{host}/", headers={}, body="") + + assert state.save_run_data() + first = _read_record(state)["scope_violations"] + assert first["dropped"] == 2 + assert first["total"] == 4 + + # Repeated saves with no new denials keep the overflow stable. + assert state.save_run_data() + assert state.save_run_data() + again = _read_record(state)["scope_violations"] + assert again["dropped"] == 2 + assert again["total"] == 4 + + +def test_quality_surfaces_are_bounded(state_1_1: ReportState) -> None: + state = state_1_1 + entries = [ + {"surface": f"https://host-{i}.example.com/", "risk_area": "xss", "outcome": "reported"} + for i in range(300) + ] + doc = build_scan_quality( + run_record=state.run_record, + agent_graph={}, + coverage_entries=entries, + vulnerability_reports=[], + scope_decisions=None, + ) + assert len(doc["surfaces"]) <= 200 + + +def test_hydrated_ledger_feeds_quality(state_1_1: ReportState) -> None: + state = state_1_1 + run_dir = state.get_run_dir() + state_dir = run_dir / ".state" + state_dir.mkdir(parents=True, exist_ok=True) + (state_dir / "coverage.json").write_text( + json.dumps( + { + "ab" * 8: { + "surface": "https://app.example.com/", + "risk_area": "sqli", + "outcome": "no_issue_found", + "created_at": "2026-01-01 00:00:00 UTC", + "agent_name": "recon", + } + } + ), + encoding="utf-8", + ) + hydrate_coverage_from_disk(state_dir) + assert state.save_run_data() + quality = _read_record(state)["scan_quality"] + assert quality["declared"]["coverage_entries"] == 1 + assert quality["declared"]["outcomes"] == {"no_issue_found": 1} diff --git a/tests/test_session_cleanup.py b/tests/test_session_cleanup.py index fc280430..66877c6f 100644 --- a/tests/test_session_cleanup.py +++ b/tests/test_session_cleanup.py @@ -69,6 +69,11 @@ def _stub_startup( "get_sandbox_container_ip": Mock(return_value=None), "resolve_sandbox_endpoint": Mock(return_value=("127.0.0.1", 8080)), "bootstrap_caido": bootstrap, + # These fakes exercise lifecycle cleanup, not capability probing — a + # real probe would (correctly) fail the stub session on exec=absent. + "probe_session_capabilities": Mock( + return_value={"preflight": {"degradations": [], "failures": []}} + ), } if environment is not None: replacements["build_sandbox_environment"] = Mock(return_value=environment) diff --git a/tests/test_session_entries.py b/tests/test_session_entries.py index 261243ba..99675187 100644 --- a/tests/test_session_entries.py +++ b/tests/test_session_entries.py @@ -161,6 +161,9 @@ async def test_create_or_reuse_passes_path_grants_to_the_manifest( captured: dict[str, Any] = {} class Session: + async def exec(self, *_args: Any, **_kwargs: Any) -> Any: + return SimpleNamespace(ok=lambda: True, stdout=b"", stderr=b"", exit_code=0) + async def resolve_exposed_port(self, _port: int) -> Any: return SimpleNamespace(tls=False, host="127.0.0.1", port=48080) @@ -168,10 +171,16 @@ async def backend(**kwargs: Any) -> tuple[Any, Any]: captured.update(kwargs) return SimpleNamespace(), Session() - async def no_caido(*_args: Any, **_kwargs: Any) -> None: - return None + async def no_caido(*_args: Any, **_kwargs: Any) -> Any: + # A real bootstrap returns a client or raises; preflight treats a + # missing capture client as a required-control failure. + return SimpleNamespace() scan_id = "manifest-grants" + # The fake backend bypasses docker admission, so give the capability + # probe a configured network — unverifiable isolation degrades rather + # than fails for a stub session without container attrs. + monkeypatch.setenv("STRIX_DOCKER_SANDBOX_NETWORK", "strix-sandbox") monkeypatch.setattr( session_manager, "load_settings",