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/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/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_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) 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"]