From bdcde5b1a937eaca500a612ee2f79e19021063c1 Mon Sep 17 00:00:00 2001 From: Oleg Bk Date: Tue, 14 Jul 2026 18:09:43 +0200 Subject: [PATCH 1/2] feat(mcp-proxy): consume paid install safety check Call the private paid install advisory endpoint during paid activation before package authorization and download. Continue on review recommendations with bounded output, fail closed on blocked/malformed/unavailable responses, and keep existing provider-absent fallback behavior. Implemented with assistance from Codex. --- .../agentveil_mcp_proxy/paid_activation.py | 26 +- .../agentveil_mcp_proxy/paid_install.py | 213 ++++++++++++++- .../tests/test_mcp_proxy_paid_install.py | 247 ++++++++++++++++++ 3 files changed, 474 insertions(+), 12 deletions(-) diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/paid_activation.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/paid_activation.py index aeea843..9e72954 100644 --- a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/paid_activation.py +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/paid_activation.py @@ -204,6 +204,7 @@ def _envelope( activation: Mapping[str, Any], provider: PaidProviderSnapshot, install_state: Mapping[str, Any] | None = None, + install_safety_advisory: str | None = None, ) -> dict[str, Any]: payload = { "ok": True, @@ -223,7 +224,11 @@ def _envelope( "public_fallback_available": install_state.get("public_fallback_available"), "error_code": install_state.get("error_code"), "last_installed_at": install_state.get("last_installed_at"), + "install_safety_state": install_state.get("install_safety_state"), + "install_safety_reason": install_state.get("install_safety_reason"), } + if install_safety_advisory: + payload["install_safety_advisory"] = install_safety_advisory return payload @@ -233,15 +238,17 @@ def format_paid_human_output(payload: Mapping[str, Any]) -> str: install = payload.get("install") or {} if activation.get("status") == STATUS_ACTIVE and install: fallback = "available" if activation.get("public_fallback_available") else "unavailable" - return "\n".join( - [ - f"Status: {activation['status']}", - f"Provider: {install.get('provider_id') or provider.get('provider_id') or 'private_v1'}", - f"Installed package: {install.get('package_name')}", - f"Installed version: {install.get('package_version')}", - f"Public fallback: {fallback}", - ] - ) + lines = [ + f"Status: {activation['status']}", + f"Provider: {install.get('provider_id') or provider.get('provider_id') or 'private_v1'}", + f"Installed package: {install.get('package_name')}", + f"Installed version: {install.get('package_version')}", + f"Public fallback: {fallback}", + ] + advisory = payload.get("install_safety_advisory") + if advisory: + lines.insert(0, str(advisory)) + return "\n".join(lines) lines = [ f"Paid provider: {'present' if payload['paid_provider_present'] else 'absent'}", @@ -317,6 +324,7 @@ def build_paid_activate_payload(*, license_key: str, home: Path | None) -> dict[ activation=activation, provider=result.provider, install_state=result.install_state, + install_safety_advisory=result.install_safety_advisory, ) provider = activate_with_paid_provider(license_key=license_key.strip()) diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/paid_install.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/paid_install.py index 0c91577..8c45599 100644 --- a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/paid_install.py +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/paid_install.py @@ -41,6 +41,46 @@ "public_fallback_available", "error_code", "last_installed_at", + "install_safety_state", + "install_safety_reason", + } +) + +INSTALL_SAFETY_OPERATION = "install" +INSTALL_SAFETY_SOURCE_REF = "src_private_policy_artifact" +INSTALL_SAFETY_SOURCE_REF_KIND = "workspace_registry" +INSTALL_SAFETY_REQUESTED_PACKAGE = "pkg_agentveil_private_policy" +INSTALL_SAFETY_EXPECTED_PACKAGE = "pkg_agentveil_private_policy" +PROVENANCE_INTENT_SOURCE_USER_DIRECT = "user_direct" +PROVENANCE_TARGET_SOURCE_WORKSPACE_REGISTRY = "workspace_registry" +PROVENANCE_TOOL_SOURCE_APPROVED_REGISTRY = "approved_registry" +PROVENANCE_METADATA_INFLUENCE_NONE = "none" +INSTALL_SAFETY_STATE_VERIFIED = "verified" +INSTALL_SAFETY_STATE_REVIEW_RECOMMENDED = "review_recommended" +# claim-check: allow "blocked" is a bounded backend response state label. +INSTALL_SAFETY_STATE_BLOCKED = "blocked" +INSTALL_SAFETY_STATE_MALFORMED = "malformed" +INSTALL_SAFETY_DECISION_ALLOW = "allow" +INSTALL_SAFETY_DECISION_REDIRECT = "redirect" +INSTALL_SAFETY_DECISION_BLOCK = "block" +INSTALL_SAFETY_LIVE_ENFORCEMENT_HOLD = "HOLD" +INSTALL_SAFETY_ALLOWED_REQUEST_KEYS = frozenset( + { + "entitlement_token", + "operation", + "source_ref", + "source_ref_kind", + "user_pinned_source", + "intent_source", + "target_source", + "tool_source", + "metadata_influence", + "requested_package", + "expected_package", + "package_namespace", + "expected_hash", + "resource_hash", + "payload_hash", } ) @@ -67,6 +107,8 @@ ERROR_PACKAGE_NAME_MISMATCH = "package_name_mismatch" ERROR_VERSION_MISMATCH = "package_version_mismatch" ERROR_INSTALL_FAILED = "install_failed" +ERROR_INSTALL_SAFETY_BLOCKED = "install_safety_blocked" +ERROR_INSTALL_SAFETY_MALFORMED = "install_safety_malformed" class PaidInstallError(ValueError): @@ -96,6 +138,17 @@ class EntitlementResult: expires_at: str | None +@dataclass(frozen=True) +class InstallSafetyResult: + ok: bool + decision: str | None + reason_code: str | None + install_safety_state: str | None + live_enforcement: str | None + public_warning: str | None + error_code: str | None + + @dataclass(frozen=True) class PackageAuthorizeResult: download_authorized: bool @@ -122,6 +175,7 @@ class PaidActivateInstallResult: install_state: dict[str, Any] public_fallback_available: bool license_id: str + install_safety_advisory: str | None = None class PaidBackendClient(Protocol): @@ -137,6 +191,12 @@ def issue_entitlement( ) -> EntitlementResult: ... + def check_install_safety( + self, + entitlement_token: str, + ) -> InstallSafetyResult: + ... + def authorize_package( self, entitlement_token: str, @@ -193,6 +253,81 @@ def current_python_version() -> str: return f"{sys.version_info.major}.{sys.version_info.minor}" +def format_install_safety_advisory_line(reason_code: str | None) -> str: + code = (reason_code or "unknown").strip() or "unknown" + return f"Install check: review recommended ({code})" + + +def format_install_safety_blocked_message(reason_code: str | None) -> str: + # claim-check: allow "blocked" is user-facing bounded status from backend. + code = (reason_code or "blocked").strip() or "blocked" + # claim-check: allow "blocked" is a bounded backend response state label. + return f"install check blocked ({code})" + + +def build_install_safety_check_request(entitlement_token: str) -> dict[str, Any]: + """Bounded request body aligned with private InstallSafetyCheckRequestSchema.""" + + return { + "entitlement_token": entitlement_token, + "operation": INSTALL_SAFETY_OPERATION, + "source_ref": INSTALL_SAFETY_SOURCE_REF, + "source_ref_kind": INSTALL_SAFETY_SOURCE_REF_KIND, + "user_pinned_source": False, + "intent_source": PROVENANCE_INTENT_SOURCE_USER_DIRECT, + "target_source": PROVENANCE_TARGET_SOURCE_WORKSPACE_REGISTRY, + "tool_source": PROVENANCE_TOOL_SOURCE_APPROVED_REGISTRY, + "metadata_influence": PROVENANCE_METADATA_INFLUENCE_NONE, + "requested_package": INSTALL_SAFETY_REQUESTED_PACKAGE, + "expected_package": INSTALL_SAFETY_EXPECTED_PACKAGE, + } + + +def parse_install_safety_result(payload: Mapping[str, Any]) -> InstallSafetyResult: + if not isinstance(payload, Mapping): + raise PaidInstallError(ERROR_INSTALL_SAFETY_MALFORMED, exit_code=1) + decision = _optional_str(payload.get("decision")) + install_safety_state = _optional_str(payload.get("install_safety_state")) + reason_code = _optional_str(payload.get("reason_code")) + if not decision or not install_safety_state or not reason_code: + raise PaidInstallError(ERROR_INSTALL_SAFETY_MALFORMED, exit_code=1) + return InstallSafetyResult( + ok=bool(payload.get("ok", True)), + decision=decision, + reason_code=reason_code, + install_safety_state=install_safety_state, + live_enforcement=_optional_str(payload.get("live_enforcement")), + public_warning=_optional_str(payload.get("public_warning")), + error_code=_optional_str(payload.get("error_code")), + ) + + +def evaluate_install_safety(result: InstallSafetyResult) -> tuple[str | None, str | None, str | None]: + """Return advisory line, persisted state, persisted reason; or raise on deny.""" + + state = (result.install_safety_state or "").strip().lower() + decision = (result.decision or "").strip().lower() + if not state: + raise PaidInstallError(ERROR_INSTALL_SAFETY_MALFORMED, exit_code=1) + + if state in {INSTALL_SAFETY_STATE_BLOCKED, INSTALL_SAFETY_STATE_MALFORMED}: + reason = result.reason_code or result.error_code or state + raise PaidInstallError(format_install_safety_blocked_message(reason), exit_code=1) + if decision == INSTALL_SAFETY_DECISION_BLOCK: + # claim-check: allow "blocked" is a bounded fallback reason label. + reason = result.reason_code or result.error_code or "blocked" + raise PaidInstallError(format_install_safety_blocked_message(reason), exit_code=1) + + if state == INSTALL_SAFETY_STATE_REVIEW_RECOMMENDED: + advisory = result.public_warning or format_install_safety_advisory_line(result.reason_code) + return advisory, state, result.reason_code + + if state == INSTALL_SAFETY_STATE_VERIFIED: + return None, state, result.reason_code + + raise PaidInstallError(ERROR_INSTALL_SAFETY_MALFORMED, exit_code=1) + + def assert_install_metadata_bounded(data: Mapping[str, Any]) -> None: extra = set(data) - BOUNDED_INSTALL_KEYS if extra: @@ -381,8 +516,12 @@ def _post_json(self, path: str, payload: Mapping[str, Any]) -> dict[str, Any]: with urllib.request.urlopen(request, timeout=self._timeout) as response: raw = response.read() except urllib.error.HTTPError as exc: - detail = exc.read().decode("utf-8", errors="replace") - scan_paid_output_for_leaks(detail) + try: + detail = exc.read().decode("utf-8", errors="replace") + except OSError: + detail = "" + if detail: + scan_paid_output_for_leaks(detail) raise PaidInstallError(ERROR_BACKEND_UNAVAILABLE, exit_code=1) from exc except urllib.error.URLError as exc: raise PaidInstallError(ERROR_BACKEND_UNAVAILABLE, exit_code=1) from exc @@ -412,6 +551,54 @@ def _post_bytes(self, path: str, payload: Mapping[str, Any]) -> bytes: except urllib.error.URLError as exc: raise PaidInstallError(ERROR_BACKEND_UNAVAILABLE, exit_code=1) from exc + def _post_install_safety_json( + self, + payload: Mapping[str, Any], + *, + entitlement_token: str, + ) -> dict[str, Any]: + body = json.dumps(dict(payload)).encode("utf-8") + request = urllib.request.Request( + # claim-check: allow "safety" is the private advisory endpoint name. + f"{self._base_url}/v1/paid/install/safety-check", + data=body, + headers={"Content-Type": "application/json", "Accept": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=self._timeout) as response: + raw = response.read() + except urllib.error.HTTPError as exc: + raw = b"" + try: + raw = exc.read() + except OSError: + raw = b"" + if raw and exc.code in {403, 422}: + try: + parsed = json.loads(raw.decode("utf-8")) + except json.JSONDecodeError as decode_exc: + raise PaidInstallError(ERROR_BACKEND_UNAVAILABLE, exit_code=1) from decode_exc + if isinstance(parsed, dict): + scan_paid_output_for_leaks(json.dumps(parsed), secrets=(entitlement_token,)) + return parsed + if exc.code in {403, 422}: + raise PaidInstallError(ERROR_INSTALL_SAFETY_BLOCKED, exit_code=1) from exc + detail = raw.decode("utf-8", errors="replace") if raw else "" + if detail: + scan_paid_output_for_leaks(detail) + raise PaidInstallError(ERROR_BACKEND_UNAVAILABLE, exit_code=1) from exc + except urllib.error.URLError as exc: + raise PaidInstallError(ERROR_BACKEND_UNAVAILABLE, exit_code=1) from exc + try: + parsed = json.loads(raw.decode("utf-8")) + except json.JSONDecodeError as exc: + raise PaidInstallError(ERROR_BACKEND_UNAVAILABLE, exit_code=1) from exc + if not isinstance(parsed, dict): + raise PaidInstallError(ERROR_BACKEND_UNAVAILABLE, exit_code=1) + scan_paid_output_for_leaks(json.dumps(parsed), secrets=(entitlement_token,)) + return parsed + def validate_activation(self, license_key: str) -> ActivationValidateResult: payload = self._post_json("/v1/paid/activate/validate", {"license_key": license_key}) return ActivationValidateResult( @@ -445,6 +632,17 @@ def issue_entitlement( expires_at=_optional_str(payload.get("expires_at")), ) + def check_install_safety( + self, + entitlement_token: str, + ) -> InstallSafetyResult: + request_payload = build_install_safety_check_request(entitlement_token) + response_payload = self._post_install_safety_json( + request_payload, + entitlement_token=entitlement_token, + ) + return parse_install_safety_result(response_payload) + def authorize_package( self, entitlement_token: str, @@ -517,9 +715,15 @@ def run_paid_activate_install_flow( ) entitlement = backend.issue_entitlement(license_key, validation) + resolved_artifact_id = artifact_id or os.environ.get("AVP_PAID_ARTIFACT_ID", DEFAULT_ARTIFACT_ID) + # claim-check: allow "safety" is advisory-only; install still verifies hash/metadata. + install_check = backend.check_install_safety(entitlement.entitlement_token) + install_safety_advisory, install_safety_state, install_safety_reason = evaluate_install_safety( + install_check, + ) authorization = backend.authorize_package( entitlement.entitlement_token, - artifact_id=artifact_id or os.environ.get("AVP_PAID_ARTIFACT_ID", DEFAULT_ARTIFACT_ID), + artifact_id=resolved_artifact_id, platform_name=current_platform_name(), python_version=current_python_version(), ) @@ -567,6 +771,8 @@ def run_paid_activate_install_flow( "public_fallback_available": authorization.public_fallback_available, "error_code": None, "last_installed_at": utc_now_iso(), + "install_safety_state": install_safety_state, + "install_safety_reason": install_safety_reason, } write_install_state(install_state_path(home), install_state) @@ -589,6 +795,7 @@ def run_paid_activate_install_flow( install_state=install_state, public_fallback_available=authorization.public_fallback_available, license_id=synthetic_license_id(license_key), + install_safety_advisory=install_safety_advisory, ) diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_paid_install.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_paid_install.py index 9faddc3..637947a 100644 --- a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_paid_install.py +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_paid_install.py @@ -23,9 +23,18 @@ ERROR_PACKAGE_NAME_MISMATCH, ERROR_VERSION_MISMATCH, INSTALL_FILENAME, + INSTALL_SAFETY_ALLOWED_REQUEST_KEYS, + INSTALL_SAFETY_LIVE_ENFORCEMENT_HOLD, + INSTALL_SAFETY_STATE_BLOCKED, + INSTALL_SAFETY_STATE_MALFORMED, + INSTALL_SAFETY_STATE_REVIEW_RECOMMENDED, + INSTALL_SAFETY_STATE_VERIFIED, HttpPaidBackendClient, PaidInstallError, + build_install_safety_check_request, + evaluate_install_safety, install_wheel_to_vendor, + parse_install_safety_result, parse_wheel_metadata, scan_paid_output_for_leaks, set_paid_backend_client, @@ -53,6 +62,11 @@ class _MockBackendState: artifact_hash: str artifact_size: int last_authorize_artifact_id: str | None = None + safety_decision: str = "allow" + safety_reason_code: str | None = None + post_paths: list[str] | None = None + authorize_call_count: int = 0 + download_call_count: int = 0 class _MockPaidBackendHandler(BaseHTTPRequestHandler): @@ -75,6 +89,8 @@ def _send_json(self, payload: dict, *, status: int = 200) -> None: self.wfile.write(body) def do_POST(self) -> None: # noqa: N802 + self.state.post_paths = self.state.post_paths or [] + self.state.post_paths.append(self.path) if self.path == "/v1/paid/activate/validate": payload = self._read_json() if payload.get("license_key") != RAW_LICENSE_KEY: @@ -108,7 +124,79 @@ def do_POST(self) -> None: # noqa: N802 } ) return + # claim-check: allow "safety" is the advisory endpoint label under test. + if self.path == "/v1/paid/install/safety-check": + payload = self._read_json() + extra = set(payload) - INSTALL_SAFETY_ALLOWED_REQUEST_KEYS + if extra: + self._send_json({"detail": "request_invalid"}, status=422) + return + expected = build_install_safety_check_request(ENTITLEMENT_TOKEN) + for key, value in expected.items(): + if payload.get(key) != value: + self._send_json({"detail": "request_invalid"}, status=422) + return + if payload.get("entitlement_token") != ENTITLEMENT_TOKEN: + self._send_json({"detail": "request_invalid"}, status=422) + return + decision = self.state.safety_decision + # claim-check: allow "blocked" is mock backend response state. + if decision == "blocked": + self._send_json( + { + "ok": False, + "decision": "block", + "reason_code": self.state.safety_reason_code or "hash_unverified", + "redirect_ref": None, + "install_safety_state": INSTALL_SAFETY_STATE_BLOCKED, + "live_enforcement": INSTALL_SAFETY_LIVE_ENFORCEMENT_HOLD, + "source_truth": {"resource_hash": None}, + }, + status=403, + ) + return + if decision == "malformed": + self._send_json( + { + "ok": False, + "decision": "malformed", + "reason_code": "malformed_response", + "redirect_ref": None, + "install_safety_state": INSTALL_SAFETY_STATE_MALFORMED, + "live_enforcement": INSTALL_SAFETY_LIVE_ENFORCEMENT_HOLD, + "source_truth": {}, + }, + status=422, + ) + return + if decision == "redirect": + reason = self.state.safety_reason_code or "model_suggested_source" + self._send_json( + { + "ok": False, + "decision": "redirect", + "reason_code": reason, + "redirect_ref": "src_model_pkg_001", + "install_safety_state": INSTALL_SAFETY_STATE_REVIEW_RECOMMENDED, + "live_enforcement": INSTALL_SAFETY_LIVE_ENFORCEMENT_HOLD, + "source_truth": {"decision": "redirect"}, + } + ) + return + self._send_json( + { + "ok": True, + "decision": "allow", + "reason_code": "workspace_registry_trusted", + "redirect_ref": None, + "install_safety_state": INSTALL_SAFETY_STATE_VERIFIED, + "live_enforcement": INSTALL_SAFETY_LIVE_ENFORCEMENT_HOLD, + "source_truth": {"decision": "allow"}, + } + ) + return if self.path == "/v1/paid/packages/authorize": + self.state.authorize_call_count += 1 payload = self._read_json() self.state.last_authorize_artifact_id = payload.get("artifact_id") if payload.get("entitlement_token") != ENTITLEMENT_TOKEN: @@ -142,6 +230,7 @@ def do_POST(self) -> None: # noqa: N802 ) return if self.path == "/v1/paid/packages/download": + self.state.download_call_count += 1 payload = self._read_json() if payload.get("download_authorization_id") != "dlauth_test_001": self.send_response(403) @@ -194,6 +283,7 @@ def mock_paid_backend(tmp_path): wheel_bytes=wheel_bytes, artifact_hash=artifact_hash, artifact_size=len(wheel_bytes), + post_paths=[], ) _MockPaidBackendHandler.state = state server = ThreadingHTTPServer(("127.0.0.1", 0), _MockPaidBackendHandler) @@ -292,6 +382,12 @@ def test_paid_activate_install_local_flow(tmp_path, mock_paid_backend): _assert_no_paid_leaks("", file_text=activation_file.read_text(encoding="utf-8")) _assert_no_paid_leaks("", file_text=install_file.read_text(encoding="utf-8")) assert json.loads(install_file.read_text(encoding="utf-8"))["status"] == STATUS_ACTIVE + assert _MockPaidBackendHandler.state.post_paths is not None + # claim-check: allow "safety" is the advisory endpoint label under test. + safety_index = _MockPaidBackendHandler.state.post_paths.index("/v1/paid/install/safety-check") + authorize_index = _MockPaidBackendHandler.state.post_paths.index("/v1/paid/packages/authorize") + download_index = _MockPaidBackendHandler.state.post_paths.index("/v1/paid/packages/download") + assert safety_index < authorize_index < download_index code, out, err = _run_main_with_backend( ["paid", "status", "--home", str(home)], @@ -496,3 +592,154 @@ def authorize_with_evil_name(self): # noqa: ANN001 assert code == 1 assert ERROR_PACKAGE_NAME_MISMATCH in err or "package_name_mismatch" in err _assert_no_paid_leaks(out + err) + + +def test_paid_activate_install_safety_review_continues_with_warning(tmp_path, mock_paid_backend): + _MockPaidBackendHandler.state.safety_decision = "redirect" + _MockPaidBackendHandler.state.safety_reason_code = "model_suggested_source" + home = tmp_path / "avp-home" + code, out, err = _run_main_with_backend( + ["paid", "activate", "--license-key-stdin", "--home", str(home)], + home=home, + base_url=mock_paid_backend, + stdin_text=f"{RAW_LICENSE_KEY}\n", + ) + assert code == 0, err + assert "Install check: review recommended (model_suggested_source)" in out + assert "Status: active" in out + install_file = home / "paid" / INSTALL_FILENAME + install_state = json.loads(install_file.read_text(encoding="utf-8")) + assert install_state["install_safety_state"] == "review_recommended" + assert install_state["install_safety_reason"] == "model_suggested_source" + _assert_no_paid_leaks(out + err, file_text=install_file.read_text(encoding="utf-8")) + + +def test_paid_activate_install_safety_blocked_stops_before_download(tmp_path, mock_paid_backend): + # claim-check: allow "blocked" is bounded test fixture state. + _MockPaidBackendHandler.state.safety_decision = "blocked" + _MockPaidBackendHandler.state.safety_reason_code = "hash_unverified" + home = tmp_path / "avp-home" + code, out, err = _run_main_with_backend( + ["paid", "activate", "--license-key-stdin", "--home", str(home)], + home=home, + base_url=mock_paid_backend, + stdin_text=f"{RAW_LICENSE_KEY}\n", + ) + assert code == 1 + # claim-check: allow "blocked" is bounded CLI error text. + assert "install check blocked (hash_unverified)" in err.lower() + assert _MockPaidBackendHandler.state.authorize_call_count == 0 + assert _MockPaidBackendHandler.state.download_call_count == 0 + assert not (home / "paid" / INSTALL_FILENAME).exists() + _assert_no_paid_leaks(out + err) + + +def test_paid_activate_install_safety_malformed_fails_closed(tmp_path, mock_paid_backend): + _MockPaidBackendHandler.state.safety_decision = "malformed" + home = tmp_path / "avp-home" + code, out, err = _run_main_with_backend( + ["paid", "activate", "--license-key-stdin", "--home", str(home)], + home=home, + base_url=mock_paid_backend, + stdin_text=f"{RAW_LICENSE_KEY}\n", + ) + assert code == 1 + # claim-check: allow "blocked" is bounded CLI error text. + assert "install check blocked" in err.lower() + assert _MockPaidBackendHandler.state.authorize_call_count == 0 + _assert_no_paid_leaks(out + err) + + +def test_paid_activate_install_safety_backend_unavailable_fails_closed(tmp_path, mock_paid_backend): + original_post = _MockPaidBackendHandler.do_POST + + def safety_server_error(self): # noqa: ANN001 + # claim-check: allow "safety" is the advisory endpoint label under test. + if self.path == "/v1/paid/install/safety-check": + self._send_json({"ok": False, "error_code": "unavailable"}, status=503) + return + return original_post(self) + + _MockPaidBackendHandler.do_POST = safety_server_error + try: + home = tmp_path / "avp-home" + code, out, err = _run_main_with_backend( + ["paid", "activate", "--license-key-stdin", "--home", str(home)], + home=home, + base_url=mock_paid_backend, + stdin_text=f"{RAW_LICENSE_KEY}\n", + ) + finally: + _MockPaidBackendHandler.do_POST = original_post + + assert code == 1 + assert "paid_backend_unavailable" in err.lower() or "ERROR:" in err + assert _MockPaidBackendHandler.state.authorize_call_count == 0 + _assert_no_paid_leaks(out + err) + + +def test_build_install_safety_check_request_matches_private_schema(): + payload = build_install_safety_check_request(ENTITLEMENT_TOKEN) + assert set(payload) <= INSTALL_SAFETY_ALLOWED_REQUEST_KEYS + assert "artifact_id" not in payload + assert payload["user_pinned_source"] is False + assert payload["intent_source"] == "user_direct" + assert payload["target_source"] == "workspace_registry" + assert payload["tool_source"] == "approved_registry" + assert payload["metadata_influence"] == "none" + + +def test_evaluate_install_safety_accepts_redirect_advisory_continue(): + result = parse_install_safety_result( + { + "ok": False, + "decision": "redirect", + "reason_code": "model_suggested_source", + "install_safety_state": INSTALL_SAFETY_STATE_REVIEW_RECOMMENDED, + "live_enforcement": INSTALL_SAFETY_LIVE_ENFORCEMENT_HOLD, + } + ) + advisory, state, reason = evaluate_install_safety(result) + assert advisory == "Install check: review recommended (model_suggested_source)" + assert state == INSTALL_SAFETY_STATE_REVIEW_RECOMMENDED + assert reason == "model_suggested_source" + + +def test_paid_activate_rejects_extra_safety_check_field(tmp_path, mock_paid_backend): + original_post = _MockPaidBackendHandler.do_POST + + def reject_artifact_id(self): # noqa: ANN001 + # claim-check: allow "safety" is the advisory endpoint label under test. + if self.path == "/v1/paid/install/safety-check": + payload = self._read_json() + if "artifact_id" in payload: + self._send_json({"detail": "request_invalid"}, status=422) + return + return original_post(self) + + _MockPaidBackendHandler.do_POST = reject_artifact_id + import agentveil_mcp_proxy.paid_install as paid_install_module + + original_build = paid_install_module.build_install_safety_check_request + try: + + def build_with_artifact(token: str) -> dict[str, object]: + body = dict(original_build(token)) + body["artifact_id"] = ARTIFACT_ID + return body + + paid_install_module.build_install_safety_check_request = build_with_artifact + home = tmp_path / "avp-home" + code, out, err = _run_main_with_backend( + ["paid", "activate", "--license-key-stdin", "--home", str(home)], + home=home, + base_url=mock_paid_backend, + stdin_text=f"{RAW_LICENSE_KEY}\n", + ) + finally: + _MockPaidBackendHandler.do_POST = original_post + paid_install_module.build_install_safety_check_request = original_build + + assert code == 1 + assert _MockPaidBackendHandler.state.authorize_call_count == 0 + _assert_no_paid_leaks(out + err) From 06265f9f5cf3fe9cf6fccc8c57565a3934178af1 Mon Sep 17 00:00:00 2001 From: Oleg Bk Date: Tue, 14 Jul 2026 19:24:20 +0200 Subject: [PATCH 2/2] feat(mcp-proxy): send install clone advisory context Add bounded install_clone_context validation for Runtime Gate package install and clone advisory metadata. Wire MCP package mutation tools to send only bounded refs and cover direct SDK privacy rejection plus signed advisory receipt verification. Implemented with assistance from Codex. --- agentveil/agent.py | 10 + agentveil/runtime_install_clone.py | 200 ++++++++++++ .../agentveil_mcp_proxy/classification.py | 43 ++- .../agentveil_mcp_proxy/runtime_gate.py | 35 ++- .../tests/test_mcp_proxy_classification.py | 51 ++++ .../tests/test_mcp_proxy_runtime_gate.py | 288 ++++++++++++++++++ tests/test_runtime_install_clone_context.py | 119 ++++++++ 7 files changed, 736 insertions(+), 10 deletions(-) create mode 100644 agentveil/runtime_install_clone.py create mode 100644 tests/test_runtime_install_clone_context.py diff --git a/agentveil/agent.py b/agentveil/agent.py index 78c928e..595a1f0 100644 --- a/agentveil/agent.py +++ b/agentveil/agent.py @@ -35,6 +35,7 @@ AVPValidationError, AVPServerError, ) +from agentveil.runtime_install_clone import validate_install_clone_context log = logging.getLogger("agentveil") @@ -1335,6 +1336,7 @@ def runtime_evaluate( payload_hash: Optional[str] = None, risk_class: Optional[str] = None, policy_context_hash: Optional[str] = None, + install_clone_context: Optional[dict] = None, ) -> dict: """ Evaluate whether this agent may perform one action right now. @@ -1342,6 +1344,10 @@ def runtime_evaluate( The agent DID is always derived from this SDK identity; callers cannot override it. The returned decision is one of ALLOW, BLOCK, or WAITING_FOR_HUMAN_APPROVAL. + + Optional ``install_clone_context`` is bounded advisory metadata for + package install/clone tools. It must not contain raw paths, URLs, + prompts, secrets, or package-manager command payloads. """ body_data = { "agent_did": self._did, @@ -1360,6 +1366,10 @@ def runtime_evaluate( body_data["risk_class"] = risk_class if policy_context_hash is not None: body_data["policy_context_hash"] = policy_context_hash + if install_clone_context is not None: + body_data["install_clone_context"] = validate_install_clone_context( + install_clone_context + ) return self._post_json("/v1/runtime/evaluate", body_data) def get_runtime_decision(self, audit_id: str) -> dict: diff --git a/agentveil/runtime_install_clone.py b/agentveil/runtime_install_clone.py new file mode 100644 index 0000000..9357fbd --- /dev/null +++ b/agentveil/runtime_install_clone.py @@ -0,0 +1,200 @@ +"""Bounded install/clone advisory context for Runtime Gate requests. + +Public clients must not forward raw paths, URLs, prompts, package-manager +commands, or secrets in ``install_clone_context``. This module validates a +caller-supplied dict against the private-compatible bounded vocabulary before +any HTTP body is built. +""" + +from __future__ import annotations + +import json +import re +from typing import Any, Mapping + +from agentveil.exceptions import AVPValidationError + +_SOURCE_REF_PATTERN = re.compile(r"^src_[a-z0-9_-]{1,58}$") +_NAMESPACE_REF_PATTERN = re.compile(r"^ns_[a-z0-9_-]{1,58}$") +_PACKAGE_REF_PATTERN = re.compile(r"^pkg_[a-z0-9_-]{1,58}$") +_HASH_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") + +_SOURCE_REF_KINDS = frozenset({ + "user_pinned", + "workspace_registry", + "internal_registry", + "approved_namespace", + "model_suggested", + "metadata_suggested", + "tool_output_suggested", + "unknown", +}) +_INTENT_SOURCES = frozenset({ + "user_direct", + "user_pinned", + "workspace_policy", + "agent_plan", + "metadata", + "tool_output", + "remote_context", + "unknown", +}) +_TARGET_SOURCES = frozenset({ + "user_pinned", + "workspace_registry", + "internal_registry", + "tool_output", + "metadata", + "model_suggested", + "remote_context", + "unknown", +}) +_TOOL_SOURCES = frozenset({ + "approved_registry", + "workspace_registry", + "local_manifest", + "mcp_server_schema", + "model_suggested", + "unknown", +}) +_METADATA_INFLUENCES = frozenset({ + "none", + "low", + "medium", + "high", + "unknown", +}) + +_REQUIRED_KEYS = frozenset({ + "operation", + "source_ref", + "source_ref_kind", + "user_pinned_source", + "intent_source", + "target_source", + "tool_source", + "metadata_influence", +}) +_OPTIONAL_KEYS = frozenset({ + "package_namespace", + "requested_package", + "expected_package", + "expected_hash", + "resource_hash", + "payload_hash", +}) +_ALLOWED_KEYS = _REQUIRED_KEYS | _OPTIONAL_KEYS + +_FORBIDDEN_SUBSTRINGS = ( + "://", + "/Users/", + "/private/", + "/var/folders/", + "/home/", + "\\Users\\", + "X-Amz-", + "sk_live_", + "sk_test_", + "ghp_", + "-----BEGIN", +) + + +def validate_install_clone_context(context: Mapping[str, Any]) -> dict[str, Any]: + """Return a bounded copy of ``install_clone_context`` or raise. + + Rejects unknown keys, invalid vocabulary, and values that look like raw + paths, URLs, or secret markers. Returns a fresh bounded copy. + """ + + if not isinstance(context, Mapping): + raise AVPValidationError("install_clone_context must be an object") + + keys = set(context) + extra = keys - _ALLOWED_KEYS + if extra: + raise AVPValidationError( + f"install_clone_context contains forbidden keys: {sorted(extra)}" + ) + missing = _REQUIRED_KEYS - keys + if missing: + raise AVPValidationError( + f"install_clone_context missing required keys: {sorted(missing)}" + ) + + operation = context["operation"] + if operation not in {"install", "clone"}: + raise AVPValidationError("install_clone_context.operation invalid") + + source_ref = _require_str(context["source_ref"], "source_ref") + if not _SOURCE_REF_PATTERN.fullmatch(source_ref): + raise AVPValidationError("install_clone_context.source_ref invalid") + + source_ref_kind = _require_str(context["source_ref_kind"], "source_ref_kind") + if source_ref_kind not in _SOURCE_REF_KINDS: + raise AVPValidationError("install_clone_context.source_ref_kind invalid") + + if not isinstance(context["user_pinned_source"], bool): + raise AVPValidationError("install_clone_context.user_pinned_source must be bool") + + intent_source = _require_str(context["intent_source"], "intent_source") + if intent_source not in _INTENT_SOURCES: + raise AVPValidationError("install_clone_context.intent_source invalid") + + target_source = _require_str(context["target_source"], "target_source") + if target_source not in _TARGET_SOURCES: + raise AVPValidationError("install_clone_context.target_source invalid") + + tool_source = _require_str(context["tool_source"], "tool_source") + if tool_source not in _TOOL_SOURCES: + raise AVPValidationError("install_clone_context.tool_source invalid") + + metadata_influence = _require_str(context["metadata_influence"], "metadata_influence") + if metadata_influence not in _METADATA_INFLUENCES: + raise AVPValidationError("install_clone_context.metadata_influence invalid") + + bounded: dict[str, Any] = { + "operation": operation, + "source_ref": source_ref, + "source_ref_kind": source_ref_kind, + "user_pinned_source": context["user_pinned_source"], + "intent_source": intent_source, + "target_source": target_source, + "tool_source": tool_source, + "metadata_influence": metadata_influence, + } + + for key in ("package_namespace", "requested_package", "expected_package"): + if key not in context or context[key] is None: + continue + value = _require_str(context[key], key) + pattern = _NAMESPACE_REF_PATTERN if key == "package_namespace" else _PACKAGE_REF_PATTERN + if not pattern.fullmatch(value): + raise AVPValidationError(f"install_clone_context.{key} invalid") + bounded[key] = value + + for key in ("expected_hash", "resource_hash", "payload_hash"): + if key not in context or context[key] is None: + continue + value = _require_str(context[key], key) + if not _HASH_PATTERN.fullmatch(value): + raise AVPValidationError(f"install_clone_context.{key} invalid") + bounded[key] = value + + serialized = json.dumps(bounded, sort_keys=True) + lowered = serialized.lower() + for marker in _FORBIDDEN_SUBSTRINGS: + if marker.lower() in lowered: + raise AVPValidationError( + "install_clone_context contains forbidden raw path/url/secret marker" + ) + return bounded + + +def _require_str(value: Any, field: str) -> str: + if not isinstance(value, str): + raise AVPValidationError(f"install_clone_context.{field} must be a string") + text = value.strip() + if not text or len(text) > 71: + raise AVPValidationError(f"install_clone_context.{field} invalid") + return text diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/classification.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/classification.py index 099a8d1..1551d0c 100644 --- a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/classification.py +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/classification.py @@ -171,6 +171,17 @@ "pip_run_script": RiskClass.DESTRUCTIVE, } +# Package install/clone-relevant tools that may attach bounded Runtime Gate +# advisory context. Read-only package status tools are intentionally excluded. +_PACKAGE_INSTALL_CLONE_CONTEXT_TOOLS = frozenset({ + "pip_install", + "pip_uninstall", + "pip_update", + "pip_run_script", +}) +INSTALL_CLONE_SOURCE_REF = "src_package_route_builtin" +INSTALL_CLONE_PACKAGE_REF = "pkg_package_route_builtin" + # Fetch/network MCP tools (e.g. the official MCP "fetch" server's `fetch` tool) # take a URL argument. The tool name `fetch` matches the generic _READ prefix, # so a benign public fetch already infers READ. The risk that this prefix misses @@ -277,7 +288,7 @@ class ClassifiedToolCall: def backend_metadata(self) -> dict[str, Any]: """Return privacy-filtered metadata intended for later backend calls.""" - return { + metadata = { "action": self.action, "action_hash": self.action_hash if self.action == self.action_hash else None, "resource": self.resource, @@ -291,6 +302,10 @@ def backend_metadata(self) -> dict[str, Any]: else self.policy_evaluation.would_decision.value ), } + install_clone_context = build_install_clone_context(self.tool) + if install_clone_context is not None: + metadata["install_clone_context"] = install_clone_context + return metadata def local_evidence_metadata(self) -> dict[str, Any]: """Return local-only metadata for future evidence slices.""" @@ -399,6 +414,29 @@ def extract_resource(arguments: Mapping[str, Any]) -> str | None: return None +def build_install_clone_context(tool: str) -> dict[str, Any] | None: + """Return bounded install/clone advisory context for package mutation tools. + + Returns ``None`` for non-package tools. Includes only stable bounded refs, + without raw package names, paths, URLs, prompts, source, or secrets. + """ + + if tool not in _PACKAGE_INSTALL_CLONE_CONTEXT_TOOLS: + return None + return { + "operation": "install", + "source_ref": INSTALL_CLONE_SOURCE_REF, + "source_ref_kind": "workspace_registry", + "user_pinned_source": False, + "intent_source": "user_direct", + "target_source": "workspace_registry", + "tool_source": "approved_registry", + "metadata_influence": "none", + "requested_package": INSTALL_CLONE_PACKAGE_REF, + "expected_package": INSTALL_CLONE_PACKAGE_REF, + } + + def infer_action_family(tool: str) -> str: # claim-check: allow "privacy-safe" describes a coarse label helper, not full data safety. """Return a coarse, privacy-safe action family label for one MCP tool name.""" @@ -515,8 +553,11 @@ def _normalize_json(value: Any) -> Any: __all__ = [ "ClassifiedToolCall", "HASH_PREFIX", + "INSTALL_CLONE_PACKAGE_REF", + "INSTALL_CLONE_SOURCE_REF", "REDACTED", "ToolCallClassifier", + "build_install_clone_context", "extract_resource", "infer_action_family", "infer_risk_class", diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/runtime_gate.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/runtime_gate.py index 3605d03..0801e3f 100644 --- a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/runtime_gate.py +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/runtime_gate.py @@ -20,7 +20,9 @@ from agentveil.agent import AVPAgent from agentveil.data_integrity import DataIntegrityError, verify_eddsa_jcs_2022 from agentveil.delegation import DelegationInvalid, verify_delegation +from agentveil.exceptions import AVPValidationError from agentveil.proof import ProofVerificationError, verify_signed_jcs +from agentveil.runtime_install_clone import validate_install_clone_context from agentveil_mcp_proxy.circuit_breaker import ( CircuitBreaker, CircuitBreakerOpenError, @@ -88,6 +90,7 @@ class _RuntimeGateRequest: payload_hash: str risk_class: str policy_context_hash: str + install_clone_context: Mapping[str, Any] | None = None class RuntimeGateClient: @@ -191,15 +194,20 @@ def evaluate(self, classification: ClassifiedToolCall) -> RuntimeGateDecision: raise RuntimeGateUnavailableError("runtime gate circuit breaker open") from exc try: try: - response = self.agent.runtime_evaluate( - action=request.action, - resource=request.resource, - environment=request.environment, - delegation_receipt=self.control_grant, - payload_hash=request.payload_hash, - risk_class=request.risk_class, - policy_context_hash=request.policy_context_hash, - ) + evaluate_kwargs: dict[str, Any] = { + "action": request.action, + "resource": request.resource, + "environment": request.environment, + "delegation_receipt": self.control_grant, + "payload_hash": request.payload_hash, + "risk_class": request.risk_class, + "policy_context_hash": request.policy_context_hash, + } + if request.install_clone_context is not None: + evaluate_kwargs["install_clone_context"] = dict( + request.install_clone_context + ) + response = self.agent.runtime_evaluate(**evaluate_kwargs) except Exception as exc: raise RuntimeGateUnavailableError("runtime gate request failed") from exc if not isinstance(response, Mapping): @@ -243,6 +251,14 @@ def seen_receipt_cache_size(self) -> int: def _build_request(self, classification: ClassifiedToolCall) -> _RuntimeGateRequest: metadata = classification.backend_metadata() + install_clone_context = metadata.get("install_clone_context") + if install_clone_context is not None: + if not isinstance(install_clone_context, Mapping): + raise RuntimeGateUnavailableError("install_clone_context invalid") + try: + install_clone_context = validate_install_clone_context(install_clone_context) + except AVPValidationError as exc: + raise RuntimeGateUnavailableError("install_clone_context invalid") from exc return _RuntimeGateRequest( action=_runtime_field(metadata.get("action")), resource=_runtime_field(metadata.get("resource")), @@ -253,6 +269,7 @@ def _build_request(self, classification: ClassifiedToolCall) -> _RuntimeGateRequ metadata.get("policy_context_hash"), "policy_context_hash", ), + install_clone_context=install_clone_context, ) def _decision_receipt_jcs(self, response: Mapping[str, Any]) -> str: diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_classification.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_classification.py index 3e7a4a4..2d9b276 100644 --- a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_classification.py +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_classification.py @@ -322,6 +322,57 @@ def test_infer_risk_class_recognizes_pip_run_script_as_destructive(): assert infer_risk_class("package.pip_run_script", tool="pip_run_script") is RiskClass.DESTRUCTIVE +def test_build_install_clone_context_for_package_mutation_tools_only(): + from agentveil_mcp_proxy.classification import ( + INSTALL_CLONE_PACKAGE_REF, + INSTALL_CLONE_SOURCE_REF, + build_install_clone_context, + ) + + for tool in ("pip_install", "pip_uninstall", "pip_update", "pip_run_script"): + context = build_install_clone_context(tool) + assert context is not None + assert context["operation"] == "install" + assert context["source_ref"] == INSTALL_CLONE_SOURCE_REF + assert context["source_ref_kind"] == "workspace_registry" + assert context["user_pinned_source"] is False + assert context["intent_source"] == "user_direct" + assert context["target_source"] == "workspace_registry" + assert context["tool_source"] == "approved_registry" + assert context["metadata_influence"] == "none" + assert context["requested_package"] == INSTALL_CLONE_PACKAGE_REF + assert context["expected_package"] == INSTALL_CLONE_PACKAGE_REF + text = json.dumps(context, sort_keys=True) + assert "/Users/" not in text + assert "http" not in text + assert "secret" not in text.lower() + + for tool in ("package_list_manifest", "create_issue", "read_file", "git_status"): + assert build_install_clone_context(tool) is None + + +def test_package_tool_backend_metadata_includes_install_clone_context(): + from agentveil_mcp_proxy.classification import INSTALL_CLONE_SOURCE_REF + + classified = ToolCallClassifier(_config(), server_name="package").classify( + tool="pip_install", + arguments={"package_name": "raw-secret-package-name", "project_path": "/Users/secret/proj"}, + ) + metadata = classified.backend_metadata() + assert "install_clone_context" in metadata + context = metadata["install_clone_context"] + assert context["source_ref"] == INSTALL_CLONE_SOURCE_REF + text = json.dumps(metadata, sort_keys=True) + assert "raw-secret-package-name" not in text + assert "/Users/secret" not in text + + non_package = ToolCallClassifier(_config(), server_name="github").classify( + tool="create_issue", + arguments={"title": "x"}, + ) + assert "install_clone_context" not in non_package.backend_metadata() + + def test_no_official_mcp_git_tool_falls_back_to_unknown(): # Tool list from https://github.com/modelcontextprotocol/servers/tree/main/src/git official_git_tools = ( diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_runtime_gate.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_runtime_gate.py index db1f7c3..2cedb94 100644 --- a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_runtime_gate.py +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_runtime_gate.py @@ -103,6 +103,7 @@ def _config(*, privacy: dict | None = None, fallback: dict | None = None) -> Pro "read": "allow", "write": "approval", "destructive": "block", + # claim-check: allow "production" is a fallback risk-class key in test config. "production": "block", "financial": "block", "unknown": "approval", @@ -598,6 +599,293 @@ def mock_post(url, **kwargs): assert forbidden not in body_text +def _package_classification(config: ProxyConfig): + return ToolCallClassifier(config, server_name="package").classify( + tool="pip_install", + arguments={ + "package_name": "raw-secret-package-name", + "project_path": "/Users/secret/proj", + "url": "https://evil.example/pkg.whl", + "token": "sk_live_do_not_leak", + }, + ) + + +def _package_ask_backend_config() -> ProxyConfig: + return ProxyConfig.from_dict({ + "proxy_config_schema_version": 1, + "avp": { + "base_url": "https://agentveil.dev", + "agent_name": "agentveil-mcp-proxy", + "trusted_signer_dids": [BACKEND_DID], + }, + "mode": "protect", + "privacy": { + "action": "redacted", + "resource": "hash", + "payload": "hash_only", + "evidence_upload": False, + }, + "fallback": { + "read": "allow", + "write": "approval", + "destructive": "block", + # claim-check: allow "production" is a fallback risk-class key in test config. + "production": "block", + "financial": "block", + "unknown": "approval", + }, + "approval": {}, + "policy": { + "id": "runtime-gate-package-test", + "policy_schema_version": 1, + "default_decision": "ask_backend", + "default_risk_class": "write", + "rules": [ + { + "id": "ask-runtime-gate-pip-install", + "source": "user", + "decision": "ask_backend", + "match": {"server": ["package"], "tool": ["pip_install"]}, + "risk_class": "write", + } + ], + }, + "downstream": {}, + }) + + +def test_package_install_runtime_request_includes_bounded_install_clone_context(): + config = _package_ask_backend_config() + agent = RecordingAgent() + client = RuntimeGateClient(agent=agent, config=config, control_grant={"id": "grant"}) + + result = client.evaluate(_package_classification(config)) + + assert result.decision == "ALLOW" + call = agent.calls[0] + assert "install_clone_context" in call + context = call["install_clone_context"] + assert context["operation"] == "install" + assert context["source_ref"] == "src_package_route_builtin" + assert context["source_ref_kind"] == "workspace_registry" + assert context["tool_source"] == "approved_registry" + assert context["requested_package"] == "pkg_package_route_builtin" + body_text = json.dumps(call, sort_keys=True) + for forbidden in ( + "raw-secret-package-name", + "/Users/secret", + "evil.example", + "sk_live_do_not_leak", + ): + assert forbidden not in body_text + + +def test_non_package_runtime_request_omits_install_clone_context(): + config = _config() + agent = RecordingAgent() + client = RuntimeGateClient(agent=agent, config=config, control_grant={"id": "grant"}) + client.evaluate(_classification(config)) + assert "install_clone_context" not in agent.calls[0] + + +def test_package_install_wire_body_includes_install_clone_context_without_raw_args(): + config = _package_ask_backend_config() + agent = AVPAgent("https://agentveil.dev", AGENT_SEED, name="wire-package", timeout=2.0) + client = RuntimeGateClient(agent=agent, config=config, control_grant={"id": "grant"}) + captured: dict[str, object] = {} + + def mock_post(url, **kwargs): + body = json.loads(kwargs["content"]) + captured["url"] = url + captured["body"] = body + # Receipt verifies core fields only; advisory may be present in response. + receipt_request = { + key: body[key] + for key in ( + "action", + "resource", + "environment", + "payload_hash", + "risk_class", + "policy_context_hash", + ) + } + receipt_jcs = _decision_receipt(receipt_request, decision="ALLOW") + response = MagicMock(spec=httpx.Response) + response.status_code = 200 + response.json.return_value = { + "audit_id": AUDIT_ID, + "decision": "ALLOW", + "decision_receipt_jcs": receipt_jcs, + "install_clone_advisory": { + "ok": True, + "decision": "allow", + "reason_code": "workspace_registry_trusted", + "advisory_state": "verified", + "live_enforcement": "HOLD", + "source_truth": {"decision": "allow"}, + }, + } + return response + + with patch.object(httpx.Client, "post", side_effect=mock_post): + result = client.evaluate(_package_classification(config)) + + assert result.decision == "ALLOW" + body = captured["body"] + assert body["install_clone_context"]["source_ref"] == "src_package_route_builtin" + assert set(body) >= { + "agent_did", + "action", + "resource", + "environment", + "receipt", + "payload_hash", + "risk_class", + "policy_context_hash", + "install_clone_context", + } + body_text = json.dumps(body, sort_keys=True) + for forbidden in ("raw-secret-package-name", "/Users/secret", "sk_live_do_not_leak"): + assert forbidden not in body_text + + +def test_receipt_with_install_clone_advisory_still_verifies(): + config = _package_ask_backend_config() + + class AdvisoryReceiptAgent(RecordingAgent): + def runtime_evaluate(self, **kwargs): + self.calls.append(kwargs) + receipt_request = { + key: kwargs[key] + for key in ( + "action", + "resource", + "environment", + "payload_hash", + "risk_class", + "policy_context_hash", + ) + } + receipt_jcs = _decision_receipt(receipt_request, decision="ALLOW") + # Inject advisory into signed body after signing would break verification; + # advisory is response-level, receipt core fields unchanged. + return { + "audit_id": AUDIT_ID, + "decision": "ALLOW", + "decision_receipt_jcs": receipt_jcs, + "install_clone_advisory": { + "ok": False, + "decision": "redirect", + "reason_code": "model_suggested_source", + "advisory_state": "review_recommended", + "live_enforcement": "HOLD", + "source_truth": {"decision": "redirect"}, + }, + } + + agent = AdvisoryReceiptAgent() + client = RuntimeGateClient(agent=agent, config=config, control_grant={"id": "grant"}) + result = client.evaluate(_package_classification(config)) + assert result.decision == "ALLOW" + assert "install_clone_context" in agent.calls[0] + + +def test_signed_receipt_body_with_install_clone_advisory_still_verifies(): + """P2: advisory may also appear inside signed DecisionReceipt body.""" + + config = _package_ask_backend_config() + + class SignedAdvisoryReceiptAgent(RecordingAgent): + def runtime_evaluate(self, **kwargs): + self.calls.append(kwargs) + receipt_request = { + key: kwargs[key] + for key in ( + "action", + "resource", + "environment", + "payload_hash", + "risk_class", + "policy_context_hash", + ) + } + body = { + "schema_version": "decision_receipt/2", + "audit_id": AUDIT_ID, + "agent_did": AGENT_DID, + "action": receipt_request["action"], + "resource": receipt_request["resource"], + "environment": receipt_request["environment"], + "decision": "ALLOW", + "payload_hash": receipt_request["payload_hash"], + "risk_class": "unknown", + "policy_context_hash": "b" * 64, + "client_risk_class": receipt_request["risk_class"], + "client_policy_context_hash": receipt_request["policy_context_hash"], + "install_clone_advisory": { + "ok": False, + "decision": "redirect", + "reason_code": "model_suggested_source", + "advisory_state": "review_recommended", + "live_enforcement": "HOLD", + "source_truth": {"decision": "redirect"}, + }, + } + return { + "audit_id": AUDIT_ID, + "decision": "ALLOW", + "decision_receipt_jcs": _sign_jcs(body), + } + + agent = SignedAdvisoryReceiptAgent() + client = RuntimeGateClient(agent=agent, config=config, control_grant={"id": "grant"}) + result = client.evaluate(_package_classification(config)) + assert result.decision == "ALLOW" + assert result.receipt_body["install_clone_advisory"]["advisory_state"] == "review_recommended" + + +def test_avp_agent_rejects_raw_install_clone_context_before_post(): + config = _package_ask_backend_config() + agent = AVPAgent("https://agentveil.dev", AGENT_SEED, name="raw-context", timeout=2.0) + client = RuntimeGateClient(agent=agent, config=config, control_grant={"id": "grant"}) + from agentveil.exceptions import AVPValidationError + + posted = {"called": False} + + def mock_post(*_args, **_kwargs): + posted["called"] = True + raise AssertionError("must not post raw install_clone_context") + + # Bypass RuntimeGateClient validation by calling agent directly. + with patch.object(httpx.Client, "post", side_effect=mock_post): + with pytest.raises(AVPValidationError): + agent.runtime_evaluate( + action="redacted", + resource="sha256:" + ("a" * 64), + environment="mcp_proxy", + delegation_receipt={"id": "grant"}, + payload_hash="sha256:" + ("c" * 64), + risk_class="write", + policy_context_hash="sha256:" + ("d" * 64), + install_clone_context={ + "operation": "install", + "source_ref": "https://evil.example/pkg.whl", + "source_ref_kind": "workspace_registry", + "user_pinned_source": False, + "intent_source": "user_direct", + "target_source": "workspace_registry", + "tool_source": "approved_registry", + "metadata_influence": "none", + "requested_package": "/Users/secret/proj", + }, + ) + assert posted["called"] is False + # Keep client path green for valid package classification. + assert client is not None + + def test_verified_allow_forwards_downstream(tmp_path): config = _config() gate = StaticGate(RuntimeGateDecision( diff --git a/tests/test_runtime_install_clone_context.py b/tests/test_runtime_install_clone_context.py new file mode 100644 index 0000000..b5b6067 --- /dev/null +++ b/tests/test_runtime_install_clone_context.py @@ -0,0 +1,119 @@ +"""Client-side bounding for Runtime Gate install_clone_context.""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +import httpx +import pytest +from nacl.signing import SigningKey + +from agentveil.agent import AVPAgent, _public_key_to_did +from agentveil.exceptions import AVPValidationError +from agentveil.runtime_install_clone import validate_install_clone_context + +AGENT_SEED = bytes.fromhex("44" * 32) + + +def _bounded_context(**overrides): + payload = { + "operation": "install", + "source_ref": "src_package_route_builtin", + "source_ref_kind": "workspace_registry", + "user_pinned_source": False, + "intent_source": "user_direct", + "target_source": "workspace_registry", + "tool_source": "approved_registry", + "metadata_influence": "none", + "requested_package": "pkg_package_route_builtin", + "expected_package": "pkg_package_route_builtin", + } + payload.update(overrides) + return payload + + +def test_validate_install_clone_context_accepts_bounded_payload(): + bounded = validate_install_clone_context(_bounded_context()) + assert bounded["source_ref"] == "src_package_route_builtin" + assert "artifact_id" not in bounded + + +def test_validate_install_clone_context_rejects_extra_keys(): + with pytest.raises(AVPValidationError, match="forbidden keys"): + validate_install_clone_context(_bounded_context(artifact_id="art_x")) + + +@pytest.mark.parametrize( + "overrides", + [ + {"source_ref": "https://evil.example/pkg.whl"}, + {"source_ref": "/Users/secret/proj"}, + {"requested_package": "raw-secret-package-name"}, + {"intent_source": "not_a_real_intent"}, + {"operation": "download"}, + ], +) +def test_validate_install_clone_context_rejects_raw_or_invalid(overrides): + with pytest.raises(AVPValidationError): + validate_install_clone_context(_bounded_context(**overrides)) + + +def test_avp_agent_runtime_evaluate_rejects_raw_context_before_http_post(): + agent = AVPAgent("https://agentveil.dev", AGENT_SEED, name="privacy-probe", timeout=2.0) + posted = {"called": False} + + def mock_post(*_args, **_kwargs): + posted["called"] = True + raise AssertionError("HTTP post must not run for invalid install_clone_context") + + with patch.object(httpx.Client, "post", side_effect=mock_post): + with pytest.raises(AVPValidationError): + agent.runtime_evaluate( + action="package.pip_install", + resource="redacted", + environment="mcp_proxy", + delegation_receipt={"id": "grant"}, + payload_hash="sha256:" + ("a" * 64), + risk_class="write", + policy_context_hash="sha256:" + ("b" * 64), + install_clone_context=_bounded_context( + source_ref="https://evil.example/pkg.whl", + requested_package="/Users/secret/proj", + ), + ) + assert posted["called"] is False + + +def test_avp_agent_runtime_evaluate_posts_only_bounded_context(): + agent = AVPAgent("https://agentveil.dev", AGENT_SEED, name="bounded-probe", timeout=2.0) + captured: dict[str, object] = {} + + def mock_post(url, **kwargs): + body = json.loads(kwargs["content"]) + captured["url"] = url + captured["body"] = body + response = MagicMock(spec=httpx.Response) + response.status_code = 200 + response.json.return_value = {"decision": "ALLOW", "audit_id": "urn:uuid:x"} + return response + + with patch.object(httpx.Client, "post", side_effect=mock_post): + agent.runtime_evaluate( + action="package.pip_install", + resource="redacted", + environment="mcp_proxy", + delegation_receipt={"id": "grant"}, + payload_hash="sha256:" + ("a" * 64), + risk_class="write", + policy_context_hash="sha256:" + ("b" * 64), + install_clone_context=_bounded_context(), + ) + + body = captured["body"] + assert captured["url"] == "/v1/runtime/evaluate" + assert body["install_clone_context"]["source_ref"] == "src_package_route_builtin" + body_text = json.dumps(body, sort_keys=True) + assert "https://" not in body_text + assert "/Users/" not in body_text + assert _public_key_to_did(bytes(SigningKey(AGENT_SEED).verify_key)) == body["agent_did"]