fix(security): v0.40.0 — path-boundary scope, more secret redaction, ingestion bounds, run-executor truth, provider errors

CHANGELOG.md

@@ -6,6 +6,76 @@ follow semantic versioning once it reaches 1.0.
 
 ## [Unreleased]
 
+## [0.40.0] - 2026-07-20
+
+_Security-floor and correctness hardening from an adversarial mining round: three
+more secret shapes are redacted, prefix scoping now matches at a path boundary,
+ingestion is bounded against oversized/hostile inputs, run executors persist and
+degrade truthfully, and provider-credential failures surface cleanly instead of
+as opaque 500s._
+
+### Security
+
+- **Prefix scope now matches at a path boundary, not a raw prefix.** A provider
+  restricted to `allowed_prefixes=["logs"]` previously admitted
+  `logs-private/secret` via a plain `startswith` — a different top-level path.
+  `check_scope` now matches only an exact equal or a `/`-delimited child
+  (`logs` / `logs/…`, never `logs-private/…`), and drops empty-string prefix
+  entries so a stray `""` can't silently unrestrict the bucket.
+- **Three more secret shapes are redacted** in logs/reports/traces/UI: Azure
+  Blob **SAS `sig=`** HMAC query params, temporary-credential **session tokens**
+  (`FQoG`/`FwoG`/`IQoJ…`), and `private_key = <value>` assignments (GCP
+  service-account / TLS keys) — labels kept, values masked.
+- The session agent's appended provider block (name + endpoint URL, added after
+  `build_session_context` and therefore outside `assert_no_secrets_in_context`)
+  now routes both fields through `redact_text`, closing the one place a
+  credential-bearing endpoint string could reach the prompt unredacted.
+
+### Ingestion bounds
+
+- `_nonempty_lines` reads via a bounded `readline(_MAX_LINE_CHARS)` loop with a
+  cumulative-bytes ceiling, so a single multi-GiB line (or a giant file) can no
+  longer be slurped whole into one Python string and OOM the sidecar.
+- Free-text group-by labels in the access-log and inventory analyzers are clipped
+  to `_LABEL_LEN` before they reach the model context / report prose.
+- Finding cells are HTML-escaped in **both** report writers (`analysis_report`
+  and `report`), so a crafted finding title/detail can't inject markup
+  (`<img onerror=…>`) into a generated report.
+- A **future-dated** object (clock skew or a garbage `9999` date) now lands in
+  the `unknown` age bucket instead of being mis-bucketed as freshly modified
+  (`0-7d`).
+- The per-run DuckDB session applies a best-effort `memory_limit` / `threads`
+  ceiling, so an unbounded analytical query can't exhaust host RAM and take the
+  desktop app down.
+- A managed evidence combine now enforces a **cumulative decompressed-output
+  budget** across all parts (`_COMBINE_MAX_OUT_BYTES`), closing the gap where
+  thousands of individually-under-cap gzip members could still sum to a
+  disk-filling total (decompression-bomb defense in depth).
+
+### Run executors
+
+- `recent_run_ids_for_provider` joins on `runs.status = 'completed'`, so a
+  crashed/partial survey's snapshot is never read back as the newest — which had
+  made "what changed" report un-scanned buckets as removed and posture answers
+  come from a truncated set.
+- Per-bucket persistence in the account-discovery run is wrapped in a
+  savepoint-style try/except: one bucket's write failure rolls back that bucket
+  and records a warning finding instead of aborting the whole survey.
+- Report generation in the diagnostic, config-review, and account-discovery runs
+  is wrapped in `require_success`, so a failed report write fails the run loudly
+  instead of silently completing without an artifact.
+- A region-mismatch HeadBucket (`301` / `PermanentRedirect`) surfaces a distinct
+  `region_mismatch` status instead of a generic access failure.
+
+### Provider credentials
+
+- The encrypted-vault temp file is forced to `0600` via `fchmod` before any
+  ciphertext is written, so a leftover `.tmp` (whose pre-existing perms `O_CREAT`
+  would otherwise keep) can never be briefly group/world-readable.
+- A malformed stored credential reference now raises a clear
+  `CredentialResolutionError` ("re-enter the credential") instead of a raw
+  `ValueError` surfacing as an opaque 500.
+
 ## [0.39.0] - 2026-07-17
 
 _Test coverage, not behavior: a frontend test harness (the frontend had zero

sidecar/app/agent_runtime/session_agent.py

@@ -665,8 +665,14 @@ def _build_prompt(
     if conn is not None:
         try:
             from ..repositories import cloud_providers as cloud_repo
-            providers = [{"provider_id": p.id, "name": p.name, "type": p.provider_type,
-                          "region": p.region, "endpoint": p.endpoint_url}
+            # redact_text the operator-controlled name/endpoint: an endpoint URL
+            # configured with embedded basic-auth (https://KEY:SECRET@host) would
+            # otherwise leak into the prompt verbatim — this block is appended
+            # AFTER build_session_context, so assert_no_secrets_in_context (which
+            # guards only `context`) does not cover it.
+            providers = [{"provider_id": p.id, "name": redact_text(p.name or ""),
+                          "type": p.provider_type, "region": p.region,
+                          "endpoint": redact_text(p.endpoint_url or "")}
                          for p in cloud_repo.list_all(conn)]
         except Exception:  # noqa: BLE001
             providers = []

sidecar/app/analysis/access_logs.py

@@ -183,10 +183,29 @@ def _open_text(path: str | Path):
     return open(p, "r", encoding="utf-8", errors="replace")
 
 
+# A real access-log line is well under a few KB. MAX_INGEST_ROWS bounds the line
+# COUNT, but not the length of any one line — so a 2 GiB single-line upload was
+# read into one giant str (OOM), and redact_text then ran its regex set over the
+# whole thing. Cap bytes per readline AND total bytes so a pathological file
+# degrades to bounded junk rows (the parsed-fraction truth guard flags them),
+# never an OOM.
+_MAX_LINE_CHARS = 1 << 20        # 1 MiB per line
+_MAX_TOTAL_CHARS = 3 << 30       # ~3 GiB total (above the 2 GiB upload cap, below OOM)
+
+
 def _nonempty_lines(path: str | Path, limit: int | None = None) -> list[str]:
     out: list[str] = []
+    total = 0
     with _open_text(path) as fh:
-        for line in fh:
+        while True:
+            # readline(size) reads AT MOST _MAX_LINE_CHARS chars, so a giant single
+            # line comes back as bounded chunks instead of one huge str.
+            line = fh.readline(_MAX_LINE_CHARS)
+            if not line:
+                break
+            total += len(line)
+            if total > _MAX_TOTAL_CHARS:
+                break
             s = line.strip()
             if not s:
                 continue
@@ -401,8 +420,19 @@ def import_access_logs(raw_path: str | Path, duckdb_path: str | Path, fmt: str)
 # --- tool 3: analyze_access_logs --------------------------------------------
 
 
+# A single object key / user-agent can be arbitrarily long (bounded only by the
+# 2 GiB upload cap). Cap every distribution LABEL so one pathological value can't
+# blow up the model context / report prose. Mirrors aggregate._LABEL_LEN.
+_LABEL_LEN = 300
+
+
+def _clip(v: object) -> str:
+    s = str(v)
+    return s if len(s) <= _LABEL_LEN else s[:_LABEL_LEN] + "…"
+
+
 def _dist(con, sql: str) -> list[dict[str, Any]]:
-    return [{"value": str(v), "count": int(c)} for v, c in con.execute(sql).fetchall()]
+    return [{"value": _clip(v), "count": int(c)} for v, c in con.execute(sql).fetchall()]
 
 
 def analyze_access_logs(duckdb_path: str | Path) -> dict[str, Any]:

sidecar/app/analysis/duck.py

@@ -15,6 +15,16 @@ def _configure(con: duckdb.DuckDBPyConnection) -> duckdb.DuckDBPyConnection:
     # day when the sidecar runs outside UTC. UTC keeps every comparison consistent
     # with how the data was stored.
     con.execute("SET TimeZone='UTC'")
+    # Defense-in-depth memory ceiling: the analysis runs inside the desktop app's
+    # sidecar, so an unbounded DuckDB query (a future path using read_csv_auto on a
+    # crafted file, a huge aggregate) must not be able to exhaust host RAM and take
+    # the app down. Bounded threads likewise keep a background analysis from
+    # starving the UI. Both are best-effort — ignored if the build rejects them.
+    for pragma in ("SET memory_limit='2GB'", "SET threads=4"):
+        try:
+            con.execute(pragma)
+        except duckdb.Error:
+            pass
     return con
 
 

sidecar/app/analysis/inventory.py

@@ -41,6 +41,17 @@ def _norm(name: str) -> str:
     return name.lower().replace(" ", "").replace("_", "")
 
 
+# Cap every key/prefix/storage-class LABEL: a single object key is bounded only by
+# the 2 GiB upload cap, so one pathological value could blow up the model context
+# and the report prose. Mirrors aggregate._LABEL_LEN / access_logs._LABEL_LEN.
+_LABEL_LEN = 300
+
+
+def _clip(v: object) -> str:
+    s = str(v)
+    return s if len(s) <= _LABEL_LEN else s[:_LABEL_LEN] + "…"
+
+
 # Storage-class tokens used to recognize the storage-class column in a HEADERLESS
 # inventory CSV (S3 + common S3-compatible names).
 _KNOWN_STORAGE = {
@@ -290,6 +301,10 @@ def series_for(field: str) -> pd.Series:
 _AGE_CASE = """
 CASE
   WHEN try_cast(last_modified AS TIMESTAMP) IS NULL THEN 'unknown'
+  -- A future last_modified (clock skew, or a garbage 9999 date) yields a NEGATIVE
+  -- age; without this guard it satisfies '<= 7' and is mis-bucketed as freshly
+  -- modified ('0-7d'). Treat future-dated objects as 'unknown', not recent.
+  WHEN datediff('day', try_cast(last_modified AS TIMESTAMP), current_timestamp) < 0 THEN 'unknown'
   WHEN datediff('day', try_cast(last_modified AS TIMESTAMP), current_timestamp) <= 7 THEN '0-7d'
   WHEN datediff('day', try_cast(last_modified AS TIMESTAMP), current_timestamp) <= 30 THEN '8-30d'
   WHEN datediff('day', try_cast(last_modified AS TIMESTAMP), current_timestamp) <= 90 THEN '31-90d'
@@ -329,20 +344,21 @@ def analyze_inventory(duckdb_path: str | Path) -> dict[str, Any]:
             ).fetchall()
         ]
         prefix_dist = [
-            {"value": str(p), "count": int(c), "size": int(s or 0)}
+            {"value": _clip(p), "count": int(c), "size": int(s or 0)}
             for p, c, s in con.execute(
                 f"SELECT prefix, count(*) c, COALESCE(sum(size), 0) s FROM {TABLE_NAME} "
                 f"GROUP BY prefix ORDER BY s DESC LIMIT {SAMPLE_LIMIT}"
             ).fetchall()
         ]
         storage_dist = [
-            {"value": str(sc), "count": int(c)}
+            {"value": _clip(sc), "count": int(c)}
             for sc, c in con.execute(
                 f"SELECT storage_class, count(*) c FROM {TABLE_NAME} GROUP BY storage_class ORDER BY c DESC"
             ).fetchall()
         ]
         top_large = [
-            {"key": str(k), "size": int(sz or 0), "storage_class": str(sc) if sc is not None else None}
+            {"key": _clip(k), "size": int(sz or 0),
+             "storage_class": _clip(sc) if sc is not None else None}
             for k, sz, sc in con.execute(
                 f"SELECT key, size, storage_class FROM {TABLE_NAME} "
                 f"ORDER BY size DESC NULLS LAST LIMIT {SAMPLE_LIMIT}"

sidecar/app/evidence/managed_import.py

@@ -46,6 +46,36 @@
 # bounds the compressed input, and this ratio bounds the decompressed output.
 _GUNZIP_MAX_RATIO = 1000
 _GUNZIP_MIN_OUT_CAP = 4 * _CHUNK
+# Cumulative decompression-bomb backstop for a whole combine. `_gunzip_out_cap`
+# bounds ONE gzip member, but a confirmed selection can carry thousands of parts,
+# and the per-file 4-MiB floor (`_GUNZIP_MIN_OUT_CAP`) means many small members
+# each stay "under cap" while their decompressed SUM fills the disk. This absolute
+# ceiling bounds the TOTAL bytes one import writes to the combined file across all
+# parts — legitimate inventory/log combines land far below it; a bomb field trips
+# it. Set generously above any real analytical import (the DuckDB engine that reads
+# the result runs under a 2 GiB memory ceiling).
+_COMBINE_MAX_OUT_BYTES = 16 * 1024 * 1024 * 1024  # 16 GiB
+
+
+class _OutBudget:
+    """Shared cumulative-output budget for one combine (see _COMBINE_MAX_OUT_BYTES).
+
+    Threaded through every part so the running decompressed/copied total is bounded
+    across the whole selection, not just per-file."""
+
+    __slots__ = ("remaining",)
+
+    def __init__(self, cap: int = _COMBINE_MAX_OUT_BYTES) -> None:
+        self.remaining = cap
+
+    def take(self, n: int) -> None:
+        self.remaining -= n
+        if self.remaining < 0:
+            raise LimitExceeded(
+                "combined evidence exceeds the total decompression budget "
+                f"({_COMBINE_MAX_OUT_BYTES} bytes across all parts); refusing a "
+                "possible decompression bomb"
+            )
 
 
 class ImportError_(Exception):
@@ -427,21 +457,23 @@ def write(self, data: bytes) -> None:
         self._inner.write(data)
 
 
-def _append_maybe_gunzip(src: Path, out) -> None:
+def _append_maybe_gunzip(src: Path, out, budget: "_OutBudget | None" = None) -> None:
     """Stream ``src`` into ``out``, transparently gunzipping (bounded).
 
     Non-gzip content (and content that merely starts with the gzip magic but is
     not a valid stream) is copied through verbatim, matching the old
     best-effort behavior. Decompressed output is capped at
     ``_GUNZIP_MAX_RATIO`` x the compressed size — beyond that it is treated as
-    a decompression bomb and the import fails with LimitExceeded.
+    a decompression bomb and the import fails with LimitExceeded. When ``budget``
+    is supplied, every emitted byte also draws down a cumulative cap shared across
+    all parts of the combine (``_COMBINE_MAX_OUT_BYTES``).
     """
     size = src.stat().st_size
     with src.open("rb") as fh:
         magic = fh.read(2)
         fh.seek(0)
         if magic != b"\x1f\x8b":
-            _copy_stream(fh, out)
+            _copy_stream(fh, out, budget)
             return
         out_cap = _gunzip_out_cap(size)
         d = zlib.decompressobj(wbits=31)  # gzip container
@@ -464,6 +496,8 @@ def _append_maybe_gunzip(src: Path, out) -> None:
                             f"bound ({out_cap} bytes for {size} compressed bytes); "
                             "refusing a possible decompression bomb"
                         )
+                    if budget is not None:
+                        budget.take(len(piece))
                     out.write(piece)
                     data = d.unconsumed_tail
                     if d.eof:
@@ -493,26 +527,29 @@ def _append_maybe_gunzip(src: Path, out) -> None:
             _copy_stream(fh, out)
 
 
-def _copy_stream(fh: BinaryIO, out) -> None:
+def _copy_stream(fh: BinaryIO, out, budget: "_OutBudget | None" = None) -> None:
     while True:
         chunk = fh.read(_CHUNK)
         if not chunk:
             break
+        if budget is not None:
+            budget.take(len(chunk))
         out.write(chunk)
 
 
-def _append_with_newline(src: Path, out: _TrackedWriter) -> None:
-    _append_maybe_gunzip(src, out)
+def _append_with_newline(src: Path, out: _TrackedWriter, budget: "_OutBudget | None" = None) -> None:
+    _append_maybe_gunzip(src, out, budget)
     if out.wrote and out.last != b"\n":
         out.write(b"\n")
 
 
 def _combine_access_logs(parts: list[Path], dest_dir: Path) -> Path:
     out_path = dest_dir / "combined.log"
+    budget = _OutBudget()
     with out_path.open("wb") as fh:
         for part in parts:
             writer = _TrackedWriter(fh)
-            _append_with_newline(part, writer)
+            _append_with_newline(part, writer, budget)
     return out_path
 
 
@@ -523,13 +560,14 @@ def _combine_inventory(parts: list[Path], fmt: str, schema: str | None, dest_dir
         import duckdb
 
         plain: list[Path] = []
+        budget = _OutBudget()
         for i, part in enumerate(parts):
             with part.open("rb") as fh:
                 is_gz = fh.read(2) == b"\x1f\x8b"
             if is_gz:
                 unpacked = part.with_name(part.name + ".parquet")
                 with unpacked.open("wb") as fh:
-                    _append_maybe_gunzip(part, fh)
+                    _append_maybe_gunzip(part, fh, budget)
                 plain.append(unpacked)
             else:
                 plain.append(part)
@@ -550,13 +588,14 @@ def _combine_inventory(parts: list[Path], fmt: str, schema: str | None, dest_dir
     # provides the column names, which we write as the header so the existing
     # header-based importer can map columns.
     out_path = dest_dir / "combined.csv"
+    budget = _OutBudget()
     with out_path.open("wb") as fh:
         if schema:
             header = ",".join(c.strip() for c in schema.split(","))
             fh.write(header.encode("utf-8") + b"\n")
             for part in parts:
                 writer = _TrackedWriter(fh)
-                _append_with_newline(part, writer)
+                _append_with_newline(part, writer, budget)
         else:
             # No schema (prefix-listing fallback, no manifest). Only skip the
             # first line of subsequent parts when the parts REALLY carry a
@@ -568,7 +607,7 @@ def _combine_inventory(parts: list[Path], fmt: str, schema: str | None, dest_dir
             for i, part in enumerate(parts):
                 tracked = _TrackedWriter(fh)
                 writer = tracked if (i == 0 or not headered) else _SkipFirstLineWriter(tracked)
-                _append_maybe_gunzip(part, writer)
+                _append_maybe_gunzip(part, writer, budget)
                 if writer.wrote and writer.last != b"\n":
                     tracked.write(b"\n")
     return out_path

sidecar/app/repositories/account_discovery.py

@@ -176,12 +176,22 @@ def get_profile(conn: sqlite3.Connection, run_id: str) -> dict[str, Any] | None:
 def recent_run_ids_for_provider(
     conn: sqlite3.Connection, provider_id: str, limit: int = 2
 ) -> list[str]:
-    """The run_ids of a provider's most recent account snapshots, newest first —
-    across sessions (the account is the same account). Used to diff 'what changed
-    since last time'."""
+    """The run_ids of a provider's most recent COMPLETED account snapshots, newest
+    first — across sessions (the account is the same account). Used to diff 'what
+    changed since last time' and to answer cross-bucket posture.
+
+    Only ``runs.status = 'completed'`` snapshots are returned: ``create_snapshot``
+    commits the snapshot row (with ``processed_count`` = the full selection) BEFORE
+    the per-bucket loop, so a run that crashed/was killed mid-survey leaves a
+    snapshot with only k of N bucket rows. Surfacing that as the newest survey made
+    ``compare_to_last_survey`` report the un-scanned buckets as 'removed' and
+    ``query_account_profile`` answer posture from a truncated set. Joining on the
+    run's terminal status excludes those partials."""
     rows = conn.execute(
-        "SELECT run_id FROM account_snapshots WHERE provider_id = ? "
-        "ORDER BY created_at DESC, rowid DESC LIMIT ?",
+        "SELECT s.run_id FROM account_snapshots s "
+        "JOIN runs r ON r.id = s.run_id "
+        "WHERE s.provider_id = ? AND r.status = 'completed' "
+        "ORDER BY s.created_at DESC, s.rowid DESC LIMIT ?",
         (provider_id, max(1, int(limit))),
     ).fetchall()
     return [r["run_id"] for r in rows]