diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/agent_launcher.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/agent_launcher.py index 97209a3..65cc143 100644 --- a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/agent_launcher.py +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/agent_launcher.py @@ -1137,6 +1137,97 @@ def stop_managed_launch(*, project_dir: Path) -> dict[str, Any]: ProtectionMode = Literal["controlled MCP route", "advisory", "not protected"] McpRouteState = Literal["configured", "missing"] EvidenceObservationState = Literal["observed", "ready_no_records", "not_initialized"] +HookCoverageState = Literal["not reported", "missing", "installed", "observed", "stale"] +TRUST_BOUNDARY_HUMAN_LINE = "Protection scope: project MCP route, not host-wide" + + +def build_trust_boundary( + *, + home: Path, + project_dir: Path, + profile: RuntimeProfileSpec | None = None, + approval_center: CenterStatus | None = None, + mcp_route_state: McpRouteState | None = None, + evidence_state: EvidenceObservationState | None = None, + protection_mode: ProtectionMode | None = None, + hook_coverage: HookCoverageState = "not reported", +) -> dict[str, Any]: + """Return bounded local trust-boundary facts for status/doctor surfaces.""" + + config_exists = (proxy_dir(home) / "config.json").is_file() + resolved_route = mcp_route_state or classify_mcp_route_state(home) + resolved_evidence = evidence_state or classify_evidence_state( + home=home, + config_exists=config_exists, + ) + center = approval_center or check_approval_center_status(home) + resolved_protection = protection_mode + if resolved_protection is None and profile is not None: + resolved_protection = derive_protection_mode( + mcp_route_state=resolved_route, + center=center, + profile_id=profile.profile_id, + ) + approval_wait: ApprovalWaitState = "not applicable" + if profile is not None: + approval_wait = classify_approval_wait_mode(home, profile) + else: + approval_wait = classify_setup_approval_wait_mode(home) + return { + "scope": "project", + "host_wide_control_claim": False, + "proxy_route_state": resolved_route, + "approval_center_state": center.state, + "approval_wait_mode": approval_wait, + "local_proof_state": resolved_evidence, + "workspace_root_ref": bounded_path_ref(project_dir)["ref"] or "", + "hook_coverage": hook_coverage, + "protection_mode": resolved_protection or "not protected", + "boundary_note": ( + "AgentVeil trust applies to routed MCP actions in this project. " + "Native host tools outside the MCP route are not host-wide contained." + ), + } + + +def classify_setup_approval_wait_mode(home: Path) -> ApprovalWaitState: + """Return approval wait-mode state from project proxy config when known.""" + + config_path = proxy_dir(home) / "config.json" + if not config_path.is_file(): + return "not configured" + try: + config_payload = json.loads(config_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, UnicodeDecodeError): + return "not configured" + if not isinstance(config_payload, dict): + return "not configured" + approval = config_payload.get("approval") + if isinstance(approval, Mapping) and approval.get("wait_for_decision") is True: + return "enabled" + return "not configured" + + +def derive_hook_coverage_from_connector_status( + status: Mapping[str, Any], +) -> HookCoverageState: + """Map existing connector status hook fields to bounded hook coverage.""" + + if status.get("hook_evidence_observed") is True: + return "observed" + if status.get("mcp_route_observed") is True: + return "observed" + hook = str(status.get("hook") or "") + hook_state = str(status.get("hook_state") or "") + if hook in {"missing"} or hook_state in {"missing"}: + return "missing" + if hook in {"stale"} or hook_state in {"stale", "invalid-json"}: + return "stale" + if hook_state in {"fired", "protected"} or status.get("status") == "protected": + return "observed" + if hook in {"installed", "present", "fired"} or hook_state in {"installed", "fired"}: + return "installed" + return "missing" @dataclass(frozen=True) @@ -1151,6 +1242,7 @@ class LaunchStatusView: proof_hint: str next_step: str diagnostics: tuple[str, ...] + trust_boundary: dict[str, Any] def to_dict(self) -> dict[str, Any]: payload = self.status.to_dict() @@ -1165,6 +1257,7 @@ def to_dict(self) -> dict[str, Any]: "proof_hint": self.proof_hint, "next_step": self.next_step, "diagnostics": list(self.diagnostics), + "trust_boundary": self.trust_boundary, } ) return payload @@ -1358,6 +1451,15 @@ def build_launch_status_view( child_running=status.child_running, evidence_state=evidence_state, ) + trust_boundary = build_trust_boundary( + home=home, + project_dir=project_dir, + profile=profile, + approval_center=status.approval_center, + mcp_route_state=mcp_route_state, + evidence_state=evidence_state, + protection_mode=protection_mode, + ) return LaunchStatusView( status=status, profile=profile, @@ -1367,6 +1469,7 @@ def build_launch_status_view( proof_hint=proof_hint, next_step=next_step, diagnostics=diagnostics, + trust_boundary=trust_boundary, ) @@ -1377,6 +1480,7 @@ def format_launch_status_human(view: LaunchStatusView) -> list[str]: "AgentVeil managed runtime status", f"Profile: {view.profile.display_name} ({status.profile_id}) — {activity}", f"Protection: {view.protection_mode}", + TRUST_BOUNDARY_HUMAN_LINE, f"Controls: {view.profile.control_surface}", f"Limitation: {view.profile.known_limitations}", f"Approval center: {status.approval_center.state}", @@ -1483,6 +1587,7 @@ class LaunchDoctorReport: blocking: tuple[str, ...] next_step: str diagnostics: tuple[str, ...] + trust_boundary: dict[str, Any] def to_dict(self) -> dict[str, Any]: return { @@ -1502,6 +1607,7 @@ def to_dict(self) -> dict[str, Any]: "blocking": list(self.blocking), "next_step": self.next_step, "diagnostics": list(self.diagnostics), + "trust_boundary": self.trust_boundary, } @@ -1686,6 +1792,20 @@ def build_launch_doctor_report( blocking=blocking, ready=ready, ) + protection_mode = derive_protection_mode( + mcp_route_state=mcp_route_state, + center=approval_center, + profile_id=profile.profile_id, + ) + trust_boundary = build_trust_boundary( + home=home, + project_dir=project_dir, + profile=profile, + approval_center=approval_center, + mcp_route_state=mcp_route_state, + evidence_state=evidence_state, + protection_mode=protection_mode, + ) return LaunchDoctorReport( profile_id=profile.profile_id, profile_label=profile.display_name, @@ -1701,6 +1821,7 @@ def build_launch_doctor_report( blocking=blocking, next_step=next_step, diagnostics=diagnostics, + trust_boundary=trust_boundary, ) @@ -1713,6 +1834,7 @@ def format_launch_doctor_human(report: LaunchDoctorReport) -> list[str]: f"Approval Center: {report.approval_center}", f"Approval wait: {report.approval_wait}", f"Local proof: {report.local_proof}", + TRUST_BOUNDARY_HUMAN_LINE, ] if report.provider_key != "not applicable": lines.append(f"Provider key: {report.provider_key}") @@ -1747,6 +1869,9 @@ def format_launch_doctor_human(report: LaunchDoctorReport) -> list[str]: "build_launch_doctor_diagnostics", "build_launch_doctor_next_step", "build_launch_doctor_report", + "build_trust_boundary", + "classify_setup_approval_wait_mode", + "derive_hook_coverage_from_connector_status", "build_launch_diagnostics", "build_launch_next_step", "build_launch_proof_hint", diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/cli.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/cli.py index 93b60cd..f2f58ea 100644 --- a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/cli.py +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/cli.py @@ -1560,6 +1560,34 @@ def run_setup_wizard_cli( return 0 if result.ok or result.setup_status == "incomplete" else 2 +def _attach_connector_trust_boundary( + status: dict[str, Any], + *, + home: Path, + project_dir: Path, +) -> dict[str, Any]: + from agentveil_mcp_proxy.agent_launcher import ( + CenterStatus, + build_trust_boundary, + derive_hook_coverage_from_connector_status, + ) + from agentveil_mcp_proxy.approval.server import inspect_managed_approval_center + + enriched = dict(status) + center = inspect_managed_approval_center(home) + enriched["trust_boundary"] = build_trust_boundary( + home=home, + project_dir=project_dir, + approval_center=CenterStatus( + state=center.state, + pid=getattr(center, "pid", None), + port=getattr(center, "port", None), + ), + hook_coverage=derive_hook_coverage_from_connector_status(enriched), + ) + return enriched + + def print_setup_status_cli( *, home: Path | None = None, @@ -1584,6 +1612,14 @@ def print_setup_status_cli( proxy_config_path=config_path, ) payload = setup_status_to_dict(status) + project_dir = home.parent if home.name == ".avp" else home + from agentveil_mcp_proxy.agent_launcher import build_trust_boundary + + payload["trust_boundary"] = build_trust_boundary( + home=home, + project_dir=project_dir, + hook_coverage="not reported", + ) assert_setup_output_is_privacy_safe(payload) if output_json: _print_json(payload, sink) @@ -4497,6 +4533,7 @@ def run_setup_cursor_status_cli(*, workspace: Path | None, output_json: bool) -> status = cursor_setup.connector_status(target, home=home) if status["approval_center"] != "running" and status["status"] != "unsafe": status["status"] = "advisory" + status = _attach_connector_trust_boundary(status, home=home, project_dir=target) if output_json: _print_operator_json(status) else: @@ -5440,6 +5477,7 @@ def run_setup_codex_status_cli( center_state=center.state, proxy_command=proxy_command or _resolve_setup_proxy_command(), ) + status = _attach_connector_trust_boundary(status, home=home, project_dir=target) if output_json: _print_operator_json(status) else: @@ -5729,6 +5767,7 @@ def run_setup_gemini_status_cli( center_state=center.state, proxy_command=proxy_command or _resolve_setup_proxy_command(), ) + status = _attach_connector_trust_boundary(status, home=home, project_dir=target) if output_json: _print_operator_json(status) else: @@ -5823,6 +5862,7 @@ def run_setup_connector_status_cli(*, project_dir: Path | None, output_json: boo # status accordingly so we never say protected with a dead approval path. if center.state != "running" and status["status"] != "unsafe": status["status"] = "unsafe" if status["mcp_route"] == "missing" else "advisory" + status = _attach_connector_trust_boundary(status, home=home, project_dir=target) if output_json: _print_operator_json(status) else: diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/redirect_playbooks.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/redirect_playbooks.py index 4524e0b..602ddf3 100644 --- a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/redirect_playbooks.py +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/redirect_playbooks.py @@ -11,11 +11,16 @@ from dataclasses import dataclass from enum import Enum +from pathlib import Path from typing import Any, Literal, Mapping from agentveil_mcp_proxy.classification import ClassifiedToolCall from agentveil_mcp_proxy.policy import RiskClass -from agentveil_mcp_proxy.product_route import PRODUCT_ROUTE_POLICY_ID, PRODUCT_ROUTE_TOOL_CATALOG +from agentveil_mcp_proxy.product_route import ( + PRODUCT_ROUTE_POLICY_ID, + PRODUCT_ROUTE_TOOL_CATALOG, + SANDBOX_READ_ONLY_MCP_TOOLS, +) class RiskFamily(str, Enum): @@ -210,6 +215,47 @@ class RedirectPlaybook(str, Enum): for tool in PRODUCT_ROUTE_TOOL_CATALOG: _TOOL_RISK_FAMILIES.setdefault(tool, RiskFamily.UNKNOWN) +_SENSITIVE_PATH_MARKERS: tuple[str, ...] = ( + ".git/", + ".env", + ".ssh/", + ".avp/", + ".agentveil/", +) + +_DEFAULT_REDIRECT_TOOLS: frozenset[str] = frozenset({ + "list_workspace", + "read_file", + "get_file_info", + "instruction_surface_status", + "git_status", + "git_diff", + "git_diff_staged", + "git_diff_unstaged", + "git_log", + "git_show", + "git_branch", + "package_inspect_state", + "package_list_manifest", + "package_risk_status", +}) | SANDBOX_READ_ONLY_MCP_TOOLS + +_PLAYBOOK_PRIMARY_TOOLS: dict[str, str] = { + RedirectPlaybook.INSPECT_BEFORE_WRITE.value: "read_file", + RedirectPlaybook.INSPECT_BEFORE_DELETE.value: "read_file", + RedirectPlaybook.SHOW_GIT_STATUS_AND_DIFF.value: "git_status", + RedirectPlaybook.INSPECT_PACKAGE_RISK.value: "package_inspect_state", + RedirectPlaybook.SECRET_POSTURE_ONLY.value: "instruction_surface_status", + RedirectPlaybook.RELEASE_READINESS_CHECK.value: "list_workspace", + RedirectPlaybook.REPO_CHANGE_REVIEW.value: "list_workspace", + RedirectPlaybook.UNTRUSTED_TEXT_REVIEW.value: "instruction_surface_status", + RedirectPlaybook.WORKFLOW_REVIEW.value: "list_workspace", + RedirectPlaybook.REMOTE_COMMAND_REVIEW.value: "list_workspace", + RedirectPlaybook.STOP_AND_CLASSIFY_UNKNOWN.value: "list_workspace", +} + +RedirectOutcome = Literal["approval_required", "hard_blocked"] + @dataclass(frozen=True) class PlaybookSpec: @@ -389,20 +435,188 @@ def build_risk_family_guidance( ) -def redirect_fields_from_guidance(guidance: RiskFamilyRedirectGuidance) -> dict[str, Any]: +def redirect_fields_from_guidance( + guidance: RiskFamilyRedirectGuidance, + *, + classification: ClassifiedToolCall | None = None, + request_id: str | None = None, + available_tools: frozenset[str] | None = None, +) -> dict[str, Any]: """Return bounded redirect keys for JSON-RPC error data.""" - return { + redirect_outcome: RedirectOutcome = ( + "approval_required" if guidance.approval_required else "hard_blocked" + ) + fields: dict[str, Any] = { "risk_family": guidance.risk_family, "redirect_playbook": guidance.redirect_playbook, "requested_action": guidance.requested_action, "risk_reason": guidance.risk_reason, "safe_first_step_id": guidance.safe_first_step_id, "approval_required": guidance.approval_required, + "redirect_outcome": redirect_outcome, "target_outcome": guidance.target_outcome, "target_reached": False, "redirect_playbook_id": guidance.redirect_playbook_id, } + if classification is not None: + fields["redirect"] = build_structured_redirect_contract( + classification, + guidance, + redirect_outcome=redirect_outcome, + available_tools=available_tools, + ) + if request_id is not None: + fields["original_request_fingerprint"] = build_original_request_fingerprint( + classification, + request_id, + ) + return fields + + +def _path_is_sensitive(resource_plain: str | None) -> bool: + if resource_plain is None: + return False + normalized = resource_plain.replace("\\", "/").lower().strip() + if not normalized: + return False + if normalized.startswith("/") or normalized.startswith("~"): + return True + if any(marker in normalized for marker in _SENSITIVE_PATH_MARKERS): + return True + return any(part.startswith(".") for part in Path(normalized).parts) + + +def _bounded_relative_path(resource_plain: str | None) -> str | None: + if resource_plain is None: + return None + text = resource_plain.strip() + if not text or text.startswith("/") or text.startswith("~"): + return None + if ".." in Path(text).parts: + return None + if _path_is_sensitive(text): + return None + return text + + +def _redirect_tool_available(tool: str, available_tools: frozenset[str]) -> bool: + return tool in available_tools + + +def _next_action_for_tool( + tool: str, + classification: ClassifiedToolCall, + *, + available_tools: frozenset[str], +) -> dict[str, Any] | None: + if not _redirect_tool_available(tool, available_tools): + return None + args: dict[str, Any] = {} + if tool in {"read_file", "get_file_info"}: + path = _bounded_relative_path(classification.resource_plain) + if path is None: + return None + args = {"path": path} + return {"tool": tool, "args": args} + + +_FALLBACK_TOOL_CANDIDATES: tuple[str, ...] = ( + "list_workspace", + "instruction_surface_status", + "git_status", + "package_inspect_state", +) + + +def _first_available_tool( + candidates: tuple[str, ...], + available_tools: frozenset[str], +) -> str | None: + for tool in candidates: + if tool in available_tools: + return tool + return None + + +def build_structured_redirect_contract( + classification: ClassifiedToolCall, + guidance: RiskFamilyRedirectGuidance, + *, + redirect_outcome: RedirectOutcome, + available_tools: frozenset[str] | None = None, +) -> dict[str, Any]: + """Return bounded machine-readable redirect next action for agents.""" + + tools = _DEFAULT_REDIRECT_TOOLS if available_tools is None else available_tools + then_retry_original = redirect_outcome == "approval_required" + playbook_id = guidance.redirect_playbook_id + primary_tool = _PLAYBOOK_PRIMARY_TOOLS.get(playbook_id, "list_workspace") + fallback_tool = _first_available_tool(_FALLBACK_TOOL_CANDIDATES, tools) + + if guidance.risk_family == RiskFamily.SECRET_ACCESS.value: + primary_tool = "instruction_surface_status" + elif _path_is_sensitive(classification.resource_plain): + primary_tool = fallback_tool or primary_tool + if redirect_outcome == "hard_blocked" and guidance.risk_family in { + RiskFamily.FILE_DELETE.value, + RiskFamily.SECRET_ACCESS.value, + }: + primary_tool = fallback_tool or primary_tool + + next_action = _next_action_for_tool( + primary_tool, + classification, + available_tools=tools, + ) + fallback_next_action = None + if fallback_tool is not None: + fallback_next_action = _next_action_for_tool( + fallback_tool, + classification, + available_tools=tools, + ) + if fallback_next_action is None and _redirect_tool_available(fallback_tool, tools): + fallback_next_action = {"tool": fallback_tool, "args": {}} + if next_action is None and fallback_next_action is not None: + next_action = fallback_next_action + + kind = playbook_id + if redirect_outcome == "hard_blocked" and guidance.risk_family == RiskFamily.SECRET_ACCESS.value: + kind = "sensitive_path_blocked" + + contract: dict[str, Any] = { + "kind": kind, + "target_changed": False, + "then_retry_original": then_retry_original, + "next_action": next_action, + "fallback_next_action": fallback_next_action, + } + if next_action is None: + contract["next_action_unavailable_reason"] = ( + "no advertised read-only tool is available for this redirect" + ) + return contract + + +def build_original_request_fingerprint( + classification: ClassifiedToolCall, + request_id: str, +) -> dict[str, str]: + """Return bounded original-request identity without raw payload content.""" + + if classification.resource_hash: + target_ref = f"resource:{classification.resource_hash}" + elif classification.resource: + target_ref = f"resource:{classification.resource}" + else: + target_ref = "target:unknown" + return { + "tool": classification.tool, + "target_ref": target_ref, + "payload_hash": classification.payload_hash, + "request_id": request_id, + } def message_visible_approval_redirect( @@ -417,6 +631,7 @@ def message_visible_approval_redirect( f"Approval required for {guidance.risk_family}.\n" f"Redirect playbook: {guidance.redirect_playbook}.\n" f"Safe first step: {spec.safe_first_step}\n" # claim-check: allow bounded playbook UI string. + f"Approval can allow the exact same MCP tool call after review.\n" f"Open {approval_url}, approve or deny, then retry the same request." ) @@ -431,12 +646,21 @@ def message_visible_blocked_redirect( return ( "Secret access blocked.\n" # claim-check: allow blocked as JSON-RPC status vocabulary. f"Redirect playbook: {guidance.redirect_playbook}.\n" - f"Safe first step: {spec.safe_first_step}" # claim-check: allow bounded playbook UI string. + f"Safe first step: {spec.safe_first_step}\n" # claim-check: allow bounded playbook UI string. + "The original secret access will not be allowed; inspect posture only." + ) + if guidance.risk_family == RiskFamily.FILE_DELETE.value: + return ( + f"Action blocked for {guidance.risk_family}.\n" # claim-check: allow blocked as JSON-RPC status vocabulary. + f"Redirect playbook: {guidance.redirect_playbook}.\n" + f"Safe first step: {spec.safe_first_step}\n" # claim-check: allow bounded playbook UI string. + "The original delete will not be allowed; confirm target identity with read-only inspection only." ) return ( f"Action blocked for {guidance.risk_family}.\n" # claim-check: allow blocked as JSON-RPC status vocabulary. f"Redirect playbook: {guidance.redirect_playbook}.\n" - f"Safe first step: {spec.safe_first_step}" # claim-check: allow bounded playbook UI string. + f"Safe first step: {spec.safe_first_step}\n" # claim-check: allow bounded playbook UI string. + "The original action will not be allowed; use the safe inspection step only." ) @@ -491,7 +715,13 @@ def enrich_risk_family_error_data( """Attach risk-family redirect metadata to JSON-RPC error data.""" guidance = build_risk_family_guidance(classification, outcome=outcome) - data.update(redirect_fields_from_guidance(guidance)) + data.update( + redirect_fields_from_guidance( + guidance, + classification=classification, + request_id=original_request_id, + ) + ) if original_request_id is not None: data["original_request_id"] = original_request_id data["redirect_context"] = redirect_context_stub( @@ -651,7 +881,9 @@ def representative_tool_risk_families() -> Mapping[str, str]: "RiskFamilyRedirectGuidance", "attach_redirect_playbook_fields", "attach_redirect_playbook_fields_for_evidence_record", + "build_original_request_fingerprint", "build_risk_family_guidance", + "build_structured_redirect_contract", "enrich_risk_family_error_data", "message_visible_approval_redirect", "message_visible_blocked_redirect", diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_agent_launcher.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_agent_launcher.py index 9df1190..de2768e 100644 --- a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_agent_launcher.py +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_agent_launcher.py @@ -43,6 +43,7 @@ LaunchResult, LaunchStatus, launch_manifest_path, + TRUST_BOUNDARY_HUMAN_LINE, load_launch_manifest, normalize_child_command, parse_hermes_agentveil_stdio_config, @@ -1156,6 +1157,8 @@ def test_launch_status_view_json_fields(tmp_path): assert payload["protection_mode"] == "advisory" assert payload["mcp_route_state"] == "configured" assert payload["proof_hint"] == "" + assert payload["trust_boundary"]["host_wide_control_claim"] is False + assert payload["trust_boundary"]["proxy_route_state"] == "configured" assert str(project) not in json.dumps(payload) @@ -1201,6 +1204,43 @@ def test_launch_status_human_omits_proof_hint_when_unavailable(tmp_path): assert "Initialize the project proxy route" in text +def test_launch_status_human_includes_trust_boundary_scope_line(tmp_path): + project = tmp_path / "project" + project.mkdir() + view = build_launch_status_view( + home=project_avp_home(project), + profile=GENERIC_PROCESS_PROFILE, + project_dir=project, + ) + text = "\n".join(format_launch_status_human(view)) + payload = view.to_dict() + + assert TRUST_BOUNDARY_HUMAN_LINE in text + assert text.count(TRUST_BOUNDARY_HUMAN_LINE) == 1 + assert "protected host-wide" not in text.lower() + assert "trust_boundary" in payload + assert TRUST_BOUNDARY_HUMAN_LINE not in json.dumps(payload) + + +def test_launch_doctor_human_includes_trust_boundary_scope_line(tmp_path): + project = tmp_path / "project" + project.mkdir() + report = build_launch_doctor_report( + home=project_avp_home(project), + profile=GENERIC_PROCESS_PROFILE, + project_dir=project, + parent_env={}, + ) + text = "\n".join(format_launch_doctor_human(report)) + payload = report.to_dict() + + assert TRUST_BOUNDARY_HUMAN_LINE in text + assert text.count(TRUST_BOUNDARY_HUMAN_LINE) == 1 + assert "protected host-wide" not in text.lower() + assert "trust_boundary" in payload + assert TRUST_BOUNDARY_HUMAN_LINE not in json.dumps(payload) + + def test_launch_diagnostics_cover_stale_center_and_native_bypass(): status = LaunchStatus( profile_id="hermes-cli", @@ -1291,6 +1331,18 @@ def test_launch_doctor_empty_project_is_not_ready(tmp_path): assert report.ready is False assert report.provider_key == "not applicable" assert "project_route" in report.blocking + assert report.trust_boundary["host_wide_control_claim"] is False + assert report.trust_boundary["proxy_route_state"] == "missing" + assert report.trust_boundary["hook_coverage"] == "not reported" + + +def test_derive_hook_coverage_from_connector_status() -> None: + from agentveil_mcp_proxy.agent_launcher import derive_hook_coverage_from_connector_status + + assert derive_hook_coverage_from_connector_status({"hook": "missing"}) == "missing" + assert derive_hook_coverage_from_connector_status({"hook_evidence_observed": True}) == "observed" + assert derive_hook_coverage_from_connector_status({"hook_state": "fired"}) == "observed" + assert derive_hook_coverage_from_connector_status({"hook": "installed"}) == "installed" def test_launch_doctor_configured_route_without_evidence(tmp_path, monkeypatch): diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_cli.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_cli.py index fb6c19e..dbc88b7 100644 --- a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_cli.py +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_cli.py @@ -1978,6 +1978,8 @@ def test_cli_setup_status_bare_is_connector(tmp_path, capsys): assert payload["hook"] == "missing" assert payload["mcp_route"] == "missing" assert "next_step" in payload + assert payload["trust_boundary"]["host_wide_control_claim"] is False + assert payload["trust_boundary"]["hook_coverage"] == "missing" def test_cli_setup_status_home_routes_to_wizard(tmp_path, capsys): @@ -1990,6 +1992,45 @@ def test_cli_setup_status_home_routes_to_wizard(tmp_path, capsys): # wizard output must not carry the connector-only keys assert "mcp_route" not in payload assert "hook" not in payload or "next_step" not in payload + assert payload["trust_boundary"]["hook_coverage"] == "not reported" + + +def test_cli_setup_status_client_codex_includes_trust_boundary( + tmp_path, + monkeypatch, + capsys, +): + from agentveil_mcp_proxy import codex_setup + + monkeypatch.setattr( + codex_setup, + "connector_status", + lambda **_kwargs: { + "status": "advisory", + "hook": "installed", + "hook_state": "installed", + "mcp_route": "present", + "proxy_route": "present", + "approval_center": "running", + }, + ) + monkeypatch.setattr( + "agentveil_mcp_proxy.approval.server.inspect_managed_approval_center", + lambda _home: SimpleNamespace(state="running", pid=123, port=8765), + ) + + assert main([ + "setup", + "status", + "--client", + "codex", + "--project-dir", + str(tmp_path), + "--json", + ]) == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["trust_boundary"]["host_wide_control_claim"] is False + assert payload["trust_boundary"]["hook_coverage"] == "installed" def test_cli_setup_claude_code_preview_does_not_write(tmp_path, capsys): @@ -2512,9 +2553,19 @@ def test_cli_launch_status_bounded_without_paths(tmp_path, capsys): assert payload["protection_mode"] == "not protected" assert payload["mcp_route_state"] == "missing" assert payload["proof_hint"] == "" + assert payload["trust_boundary"]["host_wide_control_claim"] is False + assert "Protection scope: project MCP route, not host-wide" not in json.dumps(payload) assert str(tmp_path) not in json.dumps(payload) +def test_cli_launch_status_human_includes_trust_boundary_scope_line(tmp_path, capsys): + rc = main(["launch", "status", "--project-dir", str(tmp_path)]) + assert rc == 0 + text = capsys.readouterr().out + assert "Protection scope: project MCP route, not host-wide" in text + assert text.count("Protection scope: project MCP route, not host-wide") == 1 + + def test_cli_launch_doctor_requires_profile(tmp_path, capsys): rc = main(["launch", "doctor", "--project-dir", str(tmp_path)]) assert rc == 2 @@ -2538,6 +2589,7 @@ def test_cli_launch_doctor_empty_project_json(tmp_path, capsys): assert payload["local_proof"] == "not initialized" assert payload["ready"] is False assert payload["provider_key"] == "not applicable" + assert payload["trust_boundary"]["host_wide_control_claim"] is False assert str(tmp_path) not in json.dumps(payload) @@ -2575,6 +2627,7 @@ def fake_defaults(_path): text = capsys.readouterr().out assert "AgentVeil launcher doctor" in text assert "Provider key:" in text + assert "Protection scope: project MCP route, not host-wide" in text def test_cli_launch_status_reports_hermes_cli_from_manifest(tmp_path, capsys, monkeypatch): diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_redirect_playbooks.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_redirect_playbooks.py index bf08fa7..0dda436 100644 --- a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_redirect_playbooks.py +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_redirect_playbooks.py @@ -13,25 +13,33 @@ from agentveil_mcp_proxy.policy import PolicyDecision, RiskClass from agentveil_mcp_proxy.product_route import evaluate_product_route_tool from agentveil_mcp_proxy.redirect_playbooks import ( + build_original_request_fingerprint, build_risk_family_guidance, + build_structured_redirect_contract, redirect_fields_from_guidance, uses_risk_family_redirects, ) PAYLOAD_HASH = "sha256:" + "a" * 64 +RESOURCE_HASH = "sha256:" + "b" * 64 -def _product_route_classification(tool: str) -> ClassifiedToolCall: +def _product_route_classification( + tool: str, + *, + resource_plain: str | None = "notes.txt", +) -> ClassifiedToolCall: evaluation = evaluate_product_route_tool(tool) + resource_hash = None if resource_plain is None else sha256_text(resource_plain) return ClassifiedToolCall( server="product", tool=tool, action_plain=tool, action=tool, action_hash=sha256_text(tool), - resource_plain=None, - resource=None, - resource_hash=None, + resource_plain=resource_plain, + resource=resource_hash, + resource_hash=resource_hash, payload_hash=PAYLOAD_HASH, risk_class=evaluation.risk_class, policy_evaluation=evaluation, @@ -56,9 +64,9 @@ def _non_product_classification(tool: str = "write_file") -> ClassifiedToolCall: action_plain=tool, action=tool, action_hash=sha256_text(tool), - resource_plain=None, - resource=None, - resource_hash=None, + resource_plain="notes.txt", + resource=RESOURCE_HASH, + resource_hash=RESOURCE_HASH, payload_hash=PAYLOAD_HASH, risk_class=RiskClass.WRITE, policy_evaluation=evaluation, @@ -76,9 +84,79 @@ def _assert_bounded_redirect_fields(data: dict) -> None: def test_risky_product_route_action_gets_bounded_redirect_fields() -> None: guidance = build_risk_family_guidance(_product_route_classification("write_file"), outcome="approval") - fields = redirect_fields_from_guidance(guidance) + fields = redirect_fields_from_guidance( + guidance, + classification=_product_route_classification("write_file"), + request_id="req-write-1", + ) _assert_bounded_redirect_fields(fields) + assert fields["redirect_outcome"] == "approval_required" + assert fields["redirect"]["then_retry_original"] is True + assert fields["redirect"]["target_changed"] is False + assert fields["redirect"]["next_action"]["tool"] == "read_file" + assert fields["redirect"]["next_action"]["args"] == {"path": "notes.txt"} + assert fields["original_request_fingerprint"]["request_id"] == "req-write-1" + assert fields["original_request_fingerprint"]["payload_hash"] == PAYLOAD_HASH + assert "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" in str(fields) + + +def test_hard_blocked_redirect_uses_distinct_contract() -> None: + classification = _product_route_classification("delete_file") + guidance = build_risk_family_guidance(classification, outcome="block") + fields = redirect_fields_from_guidance( + guidance, + classification=classification, + request_id="req-delete-1", + ) + + assert fields["redirect_outcome"] == "hard_blocked" + assert fields["redirect"]["then_retry_original"] is False + assert fields["redirect"]["next_action"]["tool"] in {"read_file", "list_workspace"} + + +def test_secret_block_does_not_suggest_read_file() -> None: + classification = _product_route_classification("get_secret", resource_plain=".env") + guidance = build_risk_family_guidance(classification, outcome="block") + redirect = build_structured_redirect_contract( + classification, + guidance, + redirect_outcome="hard_blocked", + ) + + assert redirect["kind"] == "sensitive_path_blocked" + assert redirect["then_retry_original"] is False + assert redirect["next_action"]["tool"] != "read_file" + + +def test_redirect_contract_respects_advertised_tool_surface() -> None: + classification = _product_route_classification("write_file") + guidance = build_risk_family_guidance(classification, outcome="approval") + redirect = build_structured_redirect_contract( + classification, + guidance, + redirect_outcome="approval_required", + available_tools=frozenset({"git_status"}), + ) + + assert redirect["next_action"] is not None + assert redirect["next_action"]["tool"] == "git_status" + assert redirect["next_action"]["tool"] != "list_workspace" + + +def test_redirect_contract_null_when_no_safe_tool_advertised() -> None: + classification = _product_route_classification("write_file") + guidance = build_risk_family_guidance(classification, outcome="approval") + redirect = build_structured_redirect_contract( + classification, + guidance, + redirect_outcome="approval_required", + available_tools=frozenset(), + ) + + assert redirect["next_action"] is None + assert redirect["next_action_unavailable_reason"] + assert "list_workspace" not in str(redirect) def test_non_product_redirect_metadata_remains_bounded() -> None: @@ -113,8 +191,11 @@ def test_approval_required_error_includes_bounded_redirect_metadata() -> None: _assert_bounded_redirect_fields(data) assert data["status"] == "approval_required" + assert data["redirect_outcome"] == "approval_required" assert data["record_id"] == "approval-1" assert data["approval_url"].startswith("http://127.0.0.1/") + assert data["redirect"]["then_retry_original"] is True + assert data["original_request_fingerprint"]["tool"] == "write_file" def test_blocked_error_includes_bounded_redirect_metadata() -> None: @@ -129,10 +210,27 @@ def test_blocked_error_includes_bounded_redirect_metadata() -> None: _assert_bounded_redirect_fields(data) assert data["status"] == "blocked" + assert data["redirect_outcome"] == "hard_blocked" + assert data["redirect"]["then_retry_original"] is False def test_block_guidance_uses_bounded_fallback_fields() -> None: guidance = build_risk_family_guidance(_product_route_classification("get_secret"), outcome="block") - fields = redirect_fields_from_guidance(guidance) + fields = redirect_fields_from_guidance( + guidance, + classification=_product_route_classification("get_secret"), + request_id="req-secret", + ) _assert_bounded_redirect_fields(fields) + assert fields["redirect_outcome"] == "hard_blocked" + + +def test_original_request_fingerprint_is_bounded() -> None: + classification = _product_route_classification("write_file") + fingerprint = build_original_request_fingerprint(classification, "req-1") + + assert fingerprint["tool"] == "write_file" + assert fingerprint["target_ref"] == f"resource:{classification.resource_hash}" + assert fingerprint["payload_hash"] == PAYLOAD_HASH + assert fingerprint["request_id"] == "req-1"