[audit-high] Failure lab could bless a baseline that never did the work

gateway/failure_lab/claims.py

@@ -165,7 +165,8 @@ def from_dict(cls, document: Mapping[str, Any]) -> ClaimRecord:
         if reported is None:
             matches = bool(expected) and not rows_diverge
         else:
-            matches = bool(reported) and not rows_diverge
+            # Only a real boolean true counts. The string "false" is truthy.
+            matches = reported is True and not rows_diverge
         status, status_lines = _observed_status(
             str(document.get("status", "")), observed
         )

gateway/failure_lab/effect_ledger.py

@@ -281,9 +281,15 @@ def execution_count(self, operation_id: str | None = None) -> int:
             return int(connection.execute(sql, params).fetchone()[0])
 
     def snapshot(self) -> list[dict[str, Any]]:
-        """The PRD's ledger shape: one entry per effect with its execution count."""
+        """The PRD's ledger shape: one entry per effect with its execution count.
+
+        Counts and rows come from one read. A second read can contain an
+        operation the first read never counted, and the snapshot would then
+        raise or mix two moments.
+        """
+        records = self.effects()
         counts: dict[str, int] = {}
-        for record in self.effects():
+        for record in records:
             counts[record.operation_id] = counts.get(record.operation_id, 0) + 1
         return [
             {
@@ -299,7 +305,7 @@ def snapshot(self) -> list[dict[str, Any]]:
                 "worker_pid": record.worker_pid,
                 "configuration": record.configuration,
             }
-            for record in self.effects()
+            for record in records
         ]
 
     def reset(self) -> None:

gateway/failure_lab/evidence.py

@@ -62,7 +62,7 @@
 from dataclasses import dataclass
 from dataclasses import fields as dataclass_fields
 from datetime import datetime, timezone
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 from typing import TYPE_CHECKING, Any
 
 from failure_lab import TEST_DEFINITION_VERSION
@@ -1266,6 +1266,36 @@ def as_dict(self) -> dict[str, Any]:
         }
 
 
+def _unsafe_bundle_path(root: Path, relative: str) -> str | None:
+    """Reject a manifest path that is not a regular file inside the bundle.
+
+    Joining an absolute path, or a path with ``..``, drops the bundle
+    directory. A symlink can point at the same outside file. Either one
+    would let a matching hash pass for bytes that are not in the bundle.
+    """
+    if (
+        not isinstance(relative, str)
+        or not relative
+        or relative.startswith(("/", "\\"))
+        or re.match(r"^[A-Za-z]:", relative)
+        or "\\" in relative
+        or "\x00" in relative
+    ):
+        return "path escapes the bundle and was not read"
+    pure = PurePosixPath(relative)
+    if pure.is_absolute() or ".." in pure.parts:
+        return "path escapes the bundle and was not read"
+    try:
+        root_real = root.resolve()
+        candidate = root_real.joinpath(*pure.parts)
+        if candidate.is_symlink():
+            return "path is a symlink and was not read"
+        candidate.resolve().relative_to(root_real)
+    except (RuntimeError, OSError, ValueError):
+        return "path escapes the bundle and was not read"
+    return None
+
+
 def verify_bundle_integrity(directory: Path | str) -> IntegrityReport:
     """Re-hash every file in a bundle against its manifest.
 
@@ -1279,7 +1309,15 @@ def verify_bundle_integrity(directory: Path | str) -> IntegrityReport:
     """
     root = Path(directory)
     manifest_path = root / MANIFEST_NAME
-    if not manifest_path.is_file():
+    try:
+        manifest_is_regular = (
+            not root.is_symlink()
+            and not manifest_path.is_symlink()
+            and manifest_path.is_file()
+        )
+    except (RuntimeError, OSError, ValueError):
+        manifest_is_regular = False
+    if not manifest_is_regular:
         return IntegrityReport(
             directory=root,
             ok=False,
@@ -1291,7 +1329,7 @@ def verify_bundle_integrity(directory: Path | str) -> IntegrityReport:
         )
     try:
         manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
-    except ValueError as exc:
+    except (RuntimeError, OSError, ValueError) as exc:
         return IntegrityReport(
             directory=root,
             ok=False,
@@ -1302,39 +1340,83 @@ def verify_bundle_integrity(directory: Path | str) -> IntegrityReport:
             problems=[f"{MANIFEST_NAME} is not valid JSON: {exc}"],
         )
 
-    entries = manifest.get("files") or []
+    if not isinstance(manifest, dict) or not isinstance(manifest.get("files"), list):
+        return IntegrityReport(
+            directory=root,
+            ok=False,
+            checked=0,
+            changed=[],
+            missing=[],
+            unexpected=[],
+            problems=[f"{MANIFEST_NAME} must contain a files list"],
+        )
+    entries = manifest["files"]
     changed: list[str] = []
     missing: list[str] = []
     problems: list[str] = []
     checked = 0
     listed: set[str] = set()
 
     for entry in entries:
-        relative = str(entry.get("path", ""))
+        if not isinstance(entry, dict):
+            problems.append("manifest file entry must be an object")
+            continue
+        relative = entry.get("path", "")
+        if not isinstance(relative, str):
+            problems.append("path escapes the bundle and was not read")
+            continue
         listed.add(relative)
+        unsafe = _unsafe_bundle_path(root, relative)
+        if unsafe is not None:
+            problems.append(f"{relative}: {unsafe}")
+            continue
         path = root / relative
-        if not path.is_file():
+        try:
+            regular = path.is_file() and not path.is_symlink()
+        except (RuntimeError, OSError, ValueError):
+            regular = False
+        if not regular:
             missing.append(relative)
             continue
+        try:
+            actual = _sha256_file(path)
+            size = path.stat().st_size
+        except (RuntimeError, OSError, ValueError) as exc:
+            problems.append(f"{relative}: could not read bundle file: {exc}")
+            continue
         checked += 1
-        actual = _sha256_file(path)
         expected = str(entry.get("sha256", ""))
         if actual != expected:
             changed.append(relative)
             problems.append(
                 f"{relative}: sha256 is {actual[:16]}..., manifest says {expected[:16]}..."
             )
-        size = path.stat().st_size
-        if entry.get("bytes") is not None and int(entry["bytes"]) != size:
+        expected_size = entry.get("bytes")
+        if expected_size is not None and (
+            not isinstance(expected_size, int)
+            or isinstance(expected_size, bool)
+            or expected_size != size
+        ):
             problems.append(
                 f"{relative}: {size} bytes on disk, manifest says {entry['bytes']}"
             )
 
-    on_disk = {
-        str(path.relative_to(root)).replace(os.sep, "/")
-        for path in root.rglob("*")
-        if path.is_file()
-    }
+    on_disk: set[str] = set()
+    pending = [root]
+    while pending:
+        directory_to_scan = pending.pop()
+        try:
+            with os.scandir(directory_to_scan) as entries:
+                for entry in entries:
+                    path = Path(entry.path)
+                    if entry.is_dir(follow_symlinks=False):
+                        pending.append(path)
+                    else:
+                        on_disk.add(str(path.relative_to(root)).replace(os.sep, "/"))
+        except (OSError, RuntimeError) as exc:
+            problems.append(
+                f"{directory_to_scan}: could not scan bundle directory: {exc}"
+            )
     unexpected = sorted(on_disk - listed - {MANIFEST_NAME})
     problems.extend(
         f"{name}: present on disk but not listed in the manifest" for name in unexpected

gateway/failure_lab/refund_tool.py

@@ -312,7 +312,16 @@ async def refund_create(
 
     def _control_ok(request: Request) -> bool:
         provided = request.headers.get(CONTROL_HEADER)
-        return provided is not None and hmac.compare_digest(provided, control_token)
+        if not provided:
+            return False
+        try:
+            # compare_digest on str raises TypeError for non-ASCII. Compare
+            # UTF-8 bytes so a latin-1 header is a denial, not a 500.
+            return hmac.compare_digest(
+                provided.encode("utf-8"), control_token.encode("utf-8")
+            )
+        except (TypeError, ValueError, UnicodeError):
+            return False
 
     def _denied() -> JSONResponse:
         return JSONResponse({"detail": "failure_lab_control_denied"}, status_code=403)

gateway/failure_lab/report.py

@@ -81,7 +81,7 @@ class ConclusionKind(str, Enum):
         "and no comparison was made either way"
     ),
     ConclusionKind.INCONCLUSIVE.value: (
-        "too few configurations ran for this test to say anything"
+        "the measurements are not enough to say a baseline handled this test"
     ),
 }
 
@@ -293,6 +293,28 @@ def _differences(
     return added, lost
 
 
+def _baseline_is_credible(column: ConfigurationView | None) -> bool:
+    """A baseline column the report may treat as having run the operation.
+
+    Zero duplicate effects is not evidence on its own. A column that never
+    reached the one intended execution suppressed duplicates by doing
+    nothing. OBSERVED records measurements without establishing that the
+    baseline handled the test correctly; only PASS supports that claim.
+    """
+    return (
+        column is not None
+        and column.ran
+        and column.verdict == Verdict.PASS.value
+        and column.downstream_effects >= INTENDED_EXECUTIONS
+    )
+
+
+def _baseline_supports_sufficiency(column: ConfigurationView | None) -> bool:
+    if column is None or not _baseline_is_credible(column):
+        return False
+    return column.duplicate_effects == 0
+
+
 def _conclude(
     result: ScenarioResult,
     existing: ConfigurationView | None,
@@ -371,7 +393,7 @@ def _conclude(
             disadvantages,
         )
 
-    if existing is not None and existing.ran and existing.duplicate_effects == 0:
+    if _baseline_supports_sufficiency(existing):
         return Conclusion(
             ConclusionKind.EXISTING_INTEGRATION_SUFFICIENT,
             "Your existing integration handled this test correctly. We did not "
@@ -384,28 +406,42 @@ def _conclude(
             disadvantages,
         )
 
-    if native is not None and native.ran and native.duplicate_effects == 0:
-        if prevented_vs_native == 0:
-            cost = (
-                " On this test the baseline also did something the gateway did "
-                "not: see what it cost, below."
-                if disadvantages
-                else ""
-            )
+    if _baseline_supports_sufficiency(native):
+        cost = (
+            " On this test the baseline also did something the gateway did "
+            "not: see what it cost, below."
+            if disadvantages
+            else ""
+        )
+        return Conclusion(
+            ConclusionKind.NATIVE_CONTROLS_SUFFICIENT,
+            "Correct native controls handled this test on their own. We did "
+            "not observe an additional duplicate effect prevented by Agent "
+            "Middleware in this scenario. If your downstream already honors "
+            "a durable operation id the way this baseline does, this "
+            "failure mode is already covered for you." + cost,
+            prevented_vs_native,
+            prevented_vs_existing,
+            differences,
+            disadvantages,
+        )
+
+    if prevented_vs_native == 0 and prevented_vs_existing == 0:
+        if (existing_ran or native_ran) and not (
+            _baseline_is_credible(existing) or _baseline_is_credible(native)
+        ):
             return Conclusion(
-                ConclusionKind.NATIVE_CONTROLS_SUFFICIENT,
-                "Correct native controls handled this test on their own. We did "
-                "not observe an additional duplicate effect prevented by Agent "
-                "Middleware in this scenario. If your downstream already honors "
-                "a durable operation id the way this baseline does, this "
-                "failure mode is already covered for you." + cost,
-                prevented_vs_native,
-                prevented_vs_existing,
+                ConclusionKind.INCONCLUSIVE,
+                "The available baseline measurements do not establish success "
+                "in this test: none has both a PASS verdict and "
+                "the required downstream execution. A zero duplicate count "
+                "alone cannot establish that the intended business action "
+                "succeeded.",
+                0,
+                0,
                 differences,
                 disadvantages,
             )
-
-    if prevented_vs_native == 0 and prevented_vs_existing == 0:
         if differences:
             return Conclusion(
                 ConclusionKind.GATEWAY_ADDED_EVIDENCE_ONLY,
@@ -418,9 +454,9 @@ def _conclude(
                 disadvantages,
             )
         return Conclusion(
-            ConclusionKind.NATIVE_CONTROLS_SUFFICIENT,
-            "No difference was observed between the baseline and the gateway "
-            "in this scenario.",
+            ConclusionKind.INCONCLUSIVE,
+            "No difference was observed, but no baseline established that it "
+            "handled the intended business action without duplicates.",
             prevented_vs_native,
             prevented_vs_existing,
             differences,

gateway/tests/test_failure_lab_claims.py

@@ -182,6 +182,25 @@ def test_loaded_manifest_cannot_promote_negative_rows_to_pass(tmp_path, row_verd
     assert supported_claims(loaded) == []
 
 
+@pytest.mark.parametrize("reported", ["false", "true", "yes", 1])
+def test_a_non_boolean_match_flag_does_not_grant_support(tmp_path, reported):
+    """The gate accepts JSON true, not every truthy value.
+
+    The string "false" is truthy in Python. A hand-written manifest must not
+    become supportable because of that.
+    """
+    manifest = build_claims_manifest([_scenario("T01")], environment="local")
+    document = manifest.as_dict()
+    document["claims"][0]["matches_documented_expectation"] = reported
+    path = tmp_path / "claims.json"
+    path.write_text(json.dumps(document))
+    loaded = load_claims_manifest(path)
+    assert loaded.records[0].matches_documented_expectation is False
+    with pytest.raises(UnsupportedClaimError):
+        assert_claim_is_supported(loaded, CLAIM)
+    assert supported_claims(loaded) == []
+
+
 @pytest.mark.parametrize("documented", [Verdict.PASS, Verdict.FAIL])
 def test_loaded_manifest_keeps_passing_rows_and_expectation_gate(tmp_path, documented):
     manifest = build_claims_manifest(