fix(mcp-proxy): close P10.5-security defense-in-depth gaps

Enforce DecisionReceipt schema and binding semantics during offline evidence bundle verification, including orphan signed-receipt warnings. Add durable fsync handling for secure CLI JSON writes. Bound client JSON-RPC input lines and downstream response bookkeeping for malicious or noisy downstream behavior.

Implemented with assistance from Codex.

Author: oleg-bk

agentveil_mcp_proxy/cli.py

@@ -43,6 +43,7 @@
     parse_utc_timestamp,
     verify_evidence_bundle_file,
 )
+from agentveil_mcp_proxy.evidence.proof import _fsync_parent_directory
 from agentveil_mcp_proxy.identity import (
     IdentityDecryptError,
     IdentityError,
@@ -203,18 +204,31 @@ def _secure_write_json(path: Path, data: dict[str, Any], *, force: bool = False)
     if force:
         tmp_path = path.with_name(f".{path.name}.tmp")
         flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
-        with os.fdopen(os.open(tmp_path, flags, 0o600), "w", encoding="utf-8") as fh:
-            json.dump(data, fh, indent=2, sort_keys=True)
-            fh.write("\n")
-        os.replace(tmp_path, path)
-        os.chmod(path, 0o600)
+        try:
+            with os.fdopen(os.open(tmp_path, flags, 0o600), "w", encoding="utf-8") as fh:
+                json.dump(data, fh, indent=2, sort_keys=True)
+                fh.write("\n")
+                fh.flush()
+                os.fsync(fh.fileno())
+            os.replace(tmp_path, path)
+            os.chmod(path, 0o600)
+            _fsync_parent_directory(path)
+        except Exception:
+            try:
+                tmp_path.unlink()
+            except OSError:
+                pass
+            raise
         return
 
     flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
     with os.fdopen(os.open(path, flags, 0o600), "w", encoding="utf-8") as fh:
         json.dump(data, fh, indent=2, sort_keys=True)
         fh.write("\n")
+        fh.flush()
+        os.fsync(fh.fileno())
     os.chmod(path, 0o600)
+    _fsync_parent_directory(path)
 
 
 def _read_json(path: Path, label: str) -> dict[str, Any]:

agentveil_mcp_proxy/evidence/proof.py

@@ -21,6 +21,7 @@
 
 
 EVIDENCE_EXPORT_SCHEMA_VERSION = 1
+_DECISION_RECEIPT_SCHEMAS = frozenset({"decision_receipt/1", "decision_receipt/2"})
 _RECEIPT_RECORD_CROSS_CHECK_FIELDS = (
     ("payload_hash", "payload_hash"),
     ("risk_class", "client_risk_class"),
@@ -219,11 +220,14 @@ def verify_evidence_bundle(
         actual_digest = hashlib.sha256(receipt_jcs.encode("utf-8")).hexdigest()
         if digest != actual_digest:
             raise EvidenceVerificationError("signed receipt digest mismatch")
-        verified_bodies[digest] = _verify_receipt_with_pinned_signers(
+        receipt_body = _verify_receipt_with_pinned_signers(
             receipt_jcs,
             pinned_signers,
         )
+        _verify_decision_receipt_semantics(receipt_body)
+        verified_bodies[digest] = receipt_body
 
+    referenced_receipt_digests: set[str] = set()
     for record in records:
         if not isinstance(record, dict):
             continue
@@ -232,16 +236,23 @@ def verify_evidence_bundle(
             continue
         if receipt_digest not in verified_bodies:
             continue
+        referenced_receipt_digests.add(receipt_digest)
         receipt_body = verified_bodies[receipt_digest]
         for record_field, receipt_field in _RECEIPT_RECORD_CROSS_CHECK_FIELDS:
             record_value = record.get(record_field)
             receipt_value = receipt_body.get(receipt_field)
-            if record_value is None or receipt_value is None:
+            if record_value is None:
                 continue
+            if receipt_value is None:
+                raise EvidenceVerificationError(
+                    f"DecisionReceipt {receipt_field} missing for referenced record"
+                )
             if receipt_value != record_value:
                 raise EvidenceVerificationError(
                     f"DecisionReceipt {receipt_field} mismatch with record {record_field}"
                 )
+    for digest in sorted(set(verified_bodies) - referenced_receipt_digests):
+        warnings.append(f"signed receipt {digest[:16]}... not referenced by any record")
 
     return EvidenceVerificationResult(
         valid=True,
@@ -313,6 +324,13 @@ def _verify_receipt_with_pinned_signers(
     raise EvidenceVerificationError("signed receipt signer is not trusted") from last_error
 
 
+def _verify_decision_receipt_semantics(body: Mapping[str, Any]) -> None:
+    if body.get("schema_version") not in _DECISION_RECEIPT_SCHEMAS:
+        raise EvidenceVerificationError("signed receipt schema unsupported")
+    if not isinstance(body.get("audit_id"), str) or not body.get("audit_id"):
+        raise EvidenceVerificationError("signed receipt audit_id missing")
+
+
 def _compute_unverified_receipt_count(
     records: Iterable[Mapping[str, Any]],
     receipts: Mapping[str, Any],

agentveil_mcp_proxy/passthrough.py

@@ -60,6 +60,9 @@
 JSONRPC_DOWNSTREAM_TIMEOUT = -32014
 DEFAULT_DOWNSTREAM_RESPONSE_TIMEOUT_SECONDS = 30.0
 MAX_DOWNSTREAM_MESSAGE_BYTES = 1 * 1024 * 1024
+MAX_CLIENT_MESSAGE_BYTES = 1 * 1024 * 1024
+MAX_PENDING_RESPONSES = 1000
+DEFAULT_TIMED_OUT_ID_RETENTION_SECONDS = 600.0
 SAFE_ENV_KEYS = (
     "PATH",
     "HOME",
@@ -287,6 +290,46 @@ def jsonrpc_error(
     }
 
 
+def _read_bounded_line(client_in: TextIO, max_bytes: int) -> tuple[str | None, bool]:
+    read = getattr(client_in, "read", None)
+    if not callable(read):
+        try:
+            raw_line = next(client_in)  # type: ignore[arg-type]
+        except StopIteration:
+            return None, False
+        raw_bytes = raw_line.encode("utf-8", errors="replace")
+        if not raw_line.endswith("\n") or len(raw_bytes.rstrip(b"\n")) > max_bytes:
+            return "", True
+        return raw_line, False
+
+    chunks: list[str] = []
+    byte_count = 0
+    while True:
+        char = read(1)
+        if char == "":
+            if chunks:
+                return "", True
+            return None, False
+        if char == "\n":
+            return "".join(chunks) + "\n", False
+        char_size = len(char.encode("utf-8", errors="replace"))
+        if byte_count + char_size > max_bytes:
+            _discard_line_remainder(client_in)
+            return "", True
+        chunks.append(char)
+        byte_count += char_size
+
+
+def _discard_line_remainder(client_in: TextIO) -> None:
+    read = getattr(client_in, "read", None)
+    if not callable(read):
+        return
+    while True:
+        char = read(1)
+        if char in {"", "\n"}:
+            return
+
+
 def _blocked_error(
     request_id: Any,
     message: str,
@@ -356,8 +399,11 @@ def __init__(
         self._runtime_gate_startup_error: Exception | None = None
         self._runtime_gate_errors = 0
         self._downstream_timeouts = 0
+        self._client_oversized_messages = 0
+        self._unsolicited_downstream_responses = 0
         self._security_events: Deque[Mapping[str, Any]] = deque(maxlen=1000)
-        self._timed_out_response_ids: set[str] = set()
+        self._inflight_ids: set[str] = set()
+        self._timed_out_response_ids: dict[str, float] = {}
         self._windows_job: _WindowsJobObject | None = None
 
     @property
@@ -384,6 +430,18 @@ def downstream_timeouts(self) -> int:
 
         return self._downstream_timeouts
 
+    @property
+    def client_oversized_messages(self) -> int:
+        """Number of oversized or unterminated client messages rejected."""
+
+        return self._client_oversized_messages
+
+    @property
+    def unsolicited_downstream_responses(self) -> int:
+        """Number of downstream responses dropped for unknown client request IDs."""
+
+        return self._unsolicited_downstream_responses
+
     @property
     def security_events(self) -> tuple[Mapping[str, Any], ...]:
         """Sanitized in-memory security events for P5 failure handling."""
@@ -496,7 +554,21 @@ def run_stdio(self, client_in: TextIO, client_out: TextIO) -> int:
         self._notification_writer = lambda message: self._write_client(client_out, message)
         self.start()
         try:
-            for raw_line in client_in:
+            while True:
+                raw_line, rejected = _read_bounded_line(client_in, MAX_CLIENT_MESSAGE_BYTES)
+                if rejected:
+                    self._increment_client_oversized_messages()
+                    self._write_client(
+                        client_out,
+                        jsonrpc_error(
+                            None,
+                            JSONRPC_INVALID_REQUEST,
+                            "client request exceeds maximum size",
+                        ),
+                    )
+                    continue
+                if raw_line is None:
+                    break
                 if not raw_line.strip():
                     continue
                 responses = self.handle_client_line(raw_line)
@@ -528,12 +600,17 @@ def handle_client_line(self, raw_line: str) -> list[dict[str, Any]]:
             policy_error, approval_outcome = self._policy_error_response(classification, request_id)
             if policy_error is not None:
                 return [policy_error] if has_id else []
-            self._send_downstream(message)
-            if not has_id:
-                return []
-            response = self._wait_downstream_response(request_id)
-            self._record_approval_result(approval_outcome, response)
-            return [response]
+            response_key = self._register_inflight_id(request_id) if has_id else None
+            try:
+                self._send_downstream(message)
+                if not has_id:
+                    return []
+                response = self._wait_downstream_response(request_id)
+                self._record_approval_result(approval_outcome, response)
+                return [response]
+            finally:
+                if response_key is not None:
+                    self._unregister_inflight_id(response_key)
         except DownstreamTimeoutError:
             self._increment_downstream_timeouts()
             self._record_approval_error(approval_outcome, "downstream_response_timeout")
@@ -801,11 +878,31 @@ def _increment_downstream_timeouts(self) -> None:
         with self._counters_lock:
             self._downstream_timeouts += 1
 
+    def _increment_client_oversized_messages(self) -> None:
+        with self._counters_lock:
+            self._client_oversized_messages += 1
+
+    def _increment_unsolicited_downstream_responses(self) -> None:
+        with self._counters_lock:
+            self._unsolicited_downstream_responses += 1
+
+    def _register_inflight_id(self, request_id: Any) -> str:
+        response_key = self._id_key(request_id)
+        with self._stdout_condition:
+            self._inflight_ids.add(response_key)
+        return response_key
+
+    def _unregister_inflight_id(self, response_key: str) -> None:
+        with self._stdout_condition:
+            self._inflight_ids.discard(response_key)
+            self._prune_pending_responses_locked()
+
     def _wait_downstream_response(self, expected_id: Any) -> dict[str, Any]:
         response_key = self._id_key(expected_id)
         deadline = time.monotonic() + self.downstream.response_timeout_seconds
         with self._stdout_condition:
             while True:
+                self._prune_timed_out_ids_locked()
                 queued = self._responses.get(response_key)
                 if queued:
                     response = queued.pop(0)
@@ -816,7 +913,9 @@ def _wait_downstream_response(self, expected_id: Any) -> dict[str, Any]:
                     raise self._downstream_error
                 remaining = deadline - time.monotonic()
                 if remaining <= 0:
-                    self._timed_out_response_ids.add(response_key)
+                    self._timed_out_response_ids[
+                        response_key
+                    ] = time.monotonic() + DEFAULT_TIMED_OUT_ID_RETENTION_SECONDS
                     raise DownstreamTimeoutError("downstream response timed out")
                 self._stdout_condition.wait(timeout=remaining)
 
@@ -911,13 +1010,46 @@ def _handle_downstream_message(self, response: Any) -> None:
             return
         if "id" in response:
             with self._stdout_condition:
+                self._prune_timed_out_ids_locked()
                 response_key = self._id_key(response.get("id"))
                 if response_key in self._timed_out_response_ids:
-                    self._timed_out_response_ids.remove(response_key)
+                    self._timed_out_response_ids.pop(response_key, None)
+                    return
+                if response_key not in self._inflight_ids:
+                    self._increment_unsolicited_downstream_responses()
                     return
                 self._responses.setdefault(response_key, []).append(response)
+                self._prune_pending_responses_locked()
                 self._stdout_condition.notify_all()
 
+    def _prune_timed_out_ids_locked(self, now: float | None = None) -> None:
+        now = time.monotonic() if now is None else now
+        expired = [
+            response_key
+            for response_key, expires_at in self._timed_out_response_ids.items()
+            if expires_at <= now
+        ]
+        for response_key in expired:
+            self._timed_out_response_ids.pop(response_key, None)
+
+    def _prune_pending_responses_locked(self) -> None:
+        pending_count = sum(len(responses) for responses in self._responses.values())
+        while pending_count > MAX_PENDING_RESPONSES:
+            dropped = False
+            for response_key, responses in list(self._responses.items()):
+                if response_key in self._inflight_ids:
+                    continue
+                if responses:
+                    responses.pop(0)
+                    pending_count -= 1
+                    dropped = True
+                if not responses:
+                    self._responses.pop(response_key, None)
+                if pending_count <= MAX_PENDING_RESPONSES:
+                    return
+            if not dropped:
+                return
+
     def _downstream_buffer_too_large(self, buffer: str) -> bool:
         return len(buffer.encode("utf-8", errors="replace")) > MAX_DOWNSTREAM_MESSAGE_BYTES
 

tests/test_mcp_proxy_cli.py

@@ -7,9 +7,11 @@
 import json
 import os
 from pathlib import Path
+import stat
 
 import pytest
 
+import agentveil_mcp_proxy.cli as proxy_cli
 from agentveil.delegation import verify_delegation
 from agentveil_mcp_proxy.cli import (
     AGENTVEIL_DEV_SIGNER_DIDS,
@@ -97,6 +99,38 @@ def _replace_grant(
     return grant
 
 
+def test_secure_write_json_fsyncs_before_close(tmp_path, monkeypatch):
+    calls: list[int] = []
+
+    def fake_fsync(fd: int) -> None:
+        os.fstat(fd)
+        calls.append(fd)
+
+    monkeypatch.setattr(proxy_cli.os, "fsync", fake_fsync)
+
+    proxy_cli._secure_write_json(tmp_path / "config.json", {"ok": True})
+
+    assert calls
+
+
+def test_secure_write_json_force_fsyncs_parent_directory_on_posix(tmp_path, monkeypatch):
+    if os.name == "nt":
+        pytest.skip("directory fsync is POSIX-specific")
+    calls: list[bool] = []
+
+    def fake_fsync(fd: int) -> None:
+        calls.append(stat.S_ISDIR(os.fstat(fd).st_mode))
+
+    monkeypatch.setattr(proxy_cli.os, "fsync", fake_fsync)
+    path = tmp_path / "config.json"
+    proxy_cli._secure_write_json(path, {"old": True})
+
+    calls.clear()
+    proxy_cli._secure_write_json(path, {"new": True}, force=True)
+
+    assert calls[-1] is True
+
+
 def test_init_creates_identity_config_and_control_grant_with_0600(tmp_path):
     home = tmp_path / "avp-home"
     result = init_proxy(