test_mcp_hardening.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_mcp/test_mcp_hardening.py
 
"""
Purpose
-------
Phase 5C programmatic sign-off of blueprint section 12 hardening
checklist.
 
Responsibilities
----------------
- Verify gateway controls (rate limit, tool manifest).
- Verify input validation (extra="forbid", path traversal, shell
  metacharacters).
- Verify policy enforcement (write-tool confirm, oct_format double-guard,
  strict profile).
- Verify sandbox constraints (no shell=True, cwd pinning, env
  sanitisation, timeout, output cap).
- Verify redaction (secrets, project path, truncation).
- Verify audit records (JSONL per call, session/request IDs).
 
Diagnostics
-----------
Domain: MCP-TESTS
Levels:
    L2 — test lifecycle
    L3 — assertion details
    L4 — deep tracing
 
Contracts
---------
- Tests use mock sandbox and ``tmp_path``; no real subprocess is
  spawned.
"""
 
from __future__ import annotations
 
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
 
import pytest
from pydantic import ConfigDict, ValidationError
 
from oct.mcp.audit import McpAuditLogger
from oct.mcp.config import McpConfig
from oct.mcp.executor import ToolExecutor
from oct.mcp.metrics import McpMetrics, PROMETHEUS_AVAILABLE
from oct.mcp.policy import McpPolicy, TOOL_MANIFEST
from oct.mcp.redactor import Redactor
from oct.mcp.safety import HITL_SENTINEL, SafetyGate
from oct.mcp.sandbox import SandboxExecutor, SandboxResult, _sanitise_env
from oct.mcp.server import RateLimiter, ServerState
from oct.mcp.tools import _run_tool, _run_tool_write
from oct.mcp.validator import _TOOL_MODEL_MAP, validate_tool_args
 
 
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
 
def _isolated_metrics() -> McpMetrics:
    """McpMetrics on a private CollectorRegistry — avoids
    ``Duplicated timeseries`` collisions across tests."""
    if not PROMETHEUS_AVAILABLE:
        return McpMetrics()
    from prometheus_client import CollectorRegistry
    return McpMetrics(registry=CollectorRegistry())
 
 
def _make_state(
    tmp_path: Path,
    sandbox_output: str = '{"exit_code": 0}',
    sandbox_exit_code: int = 0,
    profile: str = "default",
    safe_mode: bool = False,
    rate_limit: int = 30,
) -> tuple[ServerState, McpAuditLogger]:
    (tmp_path / "pyproject.toml").write_text(
        '[project]\nname="test"\nversion="0.1"\n'
    )
    mock_sandbox = MagicMock(spec=SandboxExecutor)
    mock_sandbox.run.return_value = SandboxResult(
        exit_code=sandbox_exit_code, stdout=sandbox_output, stderr=""
    )
    audit_log = tmp_path / "audit.log"
    audit_logger = McpAuditLogger(audit_log_path=audit_log)
    config = McpConfig()
    config.safe_mode = safe_mode
    state = ServerState(
        project_root=tmp_path,
        config=config,
        policy=McpPolicy(profile=profile, safe_mode=safe_mode),
        executor=ToolExecutor(sandbox=mock_sandbox),
        redactor=Redactor(),
        audit_logger=audit_logger,
        rate_limiter=RateLimiter(limit_per_minute=rate_limit),
        safety=SafetyGate(config),
        metrics=_isolated_metrics(),
        session_id="hardening-test",
    )
    return state, audit_logger
 
 
def _last_audit(logger: McpAuditLogger) -> dict:
    assert logger.current_path is not None
    lines = logger.current_path.read_text(encoding="utf-8").strip().splitlines()
    return json.loads(lines[-1])
 
 
# ---------------------------------------------------------------------------
# §12.1 Gateway
# ---------------------------------------------------------------------------
 
class TestGateway:
    def test_unknown_tool_denied_by_policy(self, tmp_path: Path):
        """Deny-by-default: unknown tool names are rejected by McpPolicy."""
        policy = McpPolicy(profile="default", safe_mode=False)
        decision = policy.check("evil_unknown_tool")
        assert not decision.allowed
 
    def test_rate_limiter_returns_false_when_exhausted(self):
        """RateLimiter.consume() returns False after token bucket is drained."""
        limiter = RateLimiter(limit_per_minute=2)
        limiter.consume()
        limiter.consume()
        assert not limiter.consume()
 
    def test_rate_limiter_returns_true_within_limit(self):
        limiter = RateLimiter(limit_per_minute=10)
        assert limiter.consume() is True
 
    def test_unknown_tool_error_returned_not_raised(self, tmp_path: Path):
        state, _ = _make_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool("oct_evil_unknown_tool", {})
        assert "[oct-mcp error]" in result
        assert isinstance(result, str)
 
 
# ---------------------------------------------------------------------------
# §12.2 Input Validation
# ---------------------------------------------------------------------------
 
class TestInputValidation:
    def test_all_tool_models_have_extra_forbid(self):
        """All 20 tool Pydantic models must have model_config = ConfigDict(extra='forbid')."""
        errors = []
        for tool_name, model_cls in _TOOL_MODEL_MAP.items():
            cfg = getattr(model_cls, "model_config", {})
            extra = cfg.get("extra", None)
            if extra != "forbid":
                errors.append(f"{tool_name}: model_config.extra={extra!r} (expected 'forbid')")
        assert not errors, "\n".join(errors)
 
    def test_extra_field_rejected_by_all_models(self):
        """Passing an undeclared field should raise ValidationError for every model."""
        errors = []
        for tool_name, model_cls in _TOOL_MODEL_MAP.items():
            try:
                model_cls.model_validate({"__evil_field__": "injected"})
                errors.append(f"{tool_name}: did not raise ValidationError for extra field")
            except ValidationError:
                pass  # expected
        assert not errors, "\n".join(errors)
 
    def test_path_traversal_rejected_for_oct_lint(self, tmp_path: Path):
        with pytest.raises((ValidationError, ValueError)):
            validate_tool_args(
                "oct_lint",
                {"paths": ["../../etc/passwd"]},
                project_root=tmp_path,
            )
 
    def test_shell_metachar_semicolon_rejected_for_oct_lint(self, tmp_path: Path):
        with pytest.raises((ValidationError, ValueError)):
            validate_tool_args(
                "oct_lint",
                {"paths": ["src; rm -rf /"]},
                project_root=tmp_path,
            )
 
    def test_shell_metachar_pipe_rejected_for_oct_lint(self, tmp_path: Path):
        with pytest.raises((ValidationError, ValueError)):
            validate_tool_args(
                "oct_lint",
                {"paths": ["src | cat /etc/passwd"]},
                project_root=tmp_path,
            )
 
    def test_no_verify_rejected_for_git_commit(self, tmp_path: Path):
        """T-06: no_verify is not a declared field → ValidationError."""
        with pytest.raises(ValidationError):
            validate_tool_args(
                "oct_git_commit",
                {"message": "feat: test", "no_verify": True},
                project_root=tmp_path,
            )
 
    def test_format_jobs_field_removed(self, tmp_path: Path):
        """OI-502: FormatArgs.jobs was removed (CLI has no --jobs flag)."""
        with pytest.raises(ValidationError):
            validate_tool_args(
                "oct_format",
                {"jobs": 1},
                project_root=tmp_path,
            )
 
    def test_missing_required_field_rejected(self, tmp_path: Path):
        """oct_new requires 'path'; omitting it raises ValidationError."""
        with pytest.raises((ValidationError, ValueError)):
            validate_tool_args("oct_new", {}, project_root=tmp_path)
 
 
# ---------------------------------------------------------------------------
# §12.3 Policy enforcement
# ---------------------------------------------------------------------------
 
class TestPolicyHardening:
    _write_tools = [
        ("oct_format", {"apply": True}),
        ("oct_clean", {}),
        ("oct_git_commit", {"message": "feat: test"}),
        ("oct_docs", {}),
        ("oct_scaffold", {"name": "myproject"}),
        ("oct_export_source", {}),
        ("oct_install_hooks", {}),
        ("oct_git_hooks_install", {}),
        ("oct_git_init", {}),
        ("oct_git_changelog", {}),
    ]
 
    def test_all_write_tools_require_confirm_true(self, tmp_path: Path):
        """All Phase 5B write tools return HITL_SENTINEL when confirm=False."""
        state, _ = _make_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
 
        errors = []
        for tool_name, extra in self._write_tools:
            result = _run_tool_write(tool_name, {"confirm": False, **extra})
            if not result.startswith(HITL_SENTINEL):
                errors.append(f"{tool_name}: expected HITL sentinel, got: {result!r}")
        assert not errors, "\n".join(errors)
 
    def test_oct_format_apply_false_confirm_true_dry_run(self, tmp_path: Path):
        """oct_format with apply=False, confirm=True → --dry-run flag sent to executor."""
        state, _ = _make_state(tmp_path, sandbox_output="[DRY RUN]")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_format", {"confirm": True, "apply": False})
        call_args = state.executor._sandbox.run.call_args
        cmd = call_args[1]["cmd"] if "cmd" in call_args[1] else call_args[0][0]
        assert "--dry-run" in cmd
        assert "--fix" not in cmd
 
    def test_oct_format_apply_true_confirm_true_fix(self, tmp_path: Path):
        """oct_format with apply=True, confirm=True → --fix flag sent to executor."""
        state, _ = _make_state(tmp_path, sandbox_output="Fixed.")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_format", {"confirm": True, "apply": True})
        call_args = state.executor._sandbox.run.call_args
        cmd = call_args[1]["cmd"] if "cmd" in call_args[1] else call_args[0][0]
        assert "--fix" in cmd
        assert "--dry-run" not in cmd
 
    def test_strict_profile_denies_all_write_tools(self, tmp_path: Path):
        """In strict profile, all write tools are denied before the HITL gate."""
        state, _ = _make_state(tmp_path, profile="strict")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
 
        errors = []
        for tool_name, extra in self._write_tools:
            result = _run_tool_write(tool_name, {"confirm": True, **extra})
            if "[oct-mcp error]" not in result:
                errors.append(f"{tool_name} should be denied in strict profile")
        assert not errors, "\n".join(errors)
 
    def test_all_write_tools_are_destructive_in_manifest(self):
        """All 11 Phase 5B tools must have destructive=True in TOOL_MANIFEST."""
        write_tools = [
            "oct_format", "oct_docs", "oct_export_source", "oct_new", "oct_scaffold",
            "oct_clean", "oct_install_hooks", "oct_git_commit",
            "oct_git_hooks_install", "oct_git_init", "oct_git_changelog",
        ]
        errors = []
        for name in write_tools:
            spec = TOOL_MANIFEST.get(name)
            if spec is None:
                errors.append(f"{name}: not in TOOL_MANIFEST")
            elif not spec.destructive:
                errors.append(f"{name}: destructive=False (expected True)")
        assert not errors, "\n".join(errors)
 
    def test_all_write_tools_auto_approve_false(self):
        """All Phase 5B write tools must have auto_approve=False in TOOL_MANIFEST."""
        write_tools = [
            "oct_format", "oct_docs", "oct_export_source", "oct_new", "oct_scaffold",
            "oct_clean", "oct_install_hooks", "oct_git_commit",
            "oct_git_hooks_install", "oct_git_init", "oct_git_changelog",
        ]
        errors = []
        for name in write_tools:
            spec = TOOL_MANIFEST.get(name)
            if spec and spec.auto_approve:
                errors.append(f"{name}: auto_approve=True (expected False)")
        assert not errors, "\n".join(errors)
 
 
# ---------------------------------------------------------------------------
# §12.4 Sandbox
# ---------------------------------------------------------------------------
 
class TestSandboxHardening:
    def test_no_shell_true_in_mcp_source(self):
        """Static AST check: no actual shell=True keyword arg in any oct/mcp/*.py call."""
        import ast as _ast
        mcp_dir = Path(__file__).parent.parent.parent / "oct" / "mcp"
        violations = []
        for py_file in sorted(mcp_dir.glob("*.py")):
            try:
                tree = _ast.parse(py_file.read_text(encoding="utf-8"))
            except SyntaxError:
                continue
            for node in _ast.walk(tree):
                if isinstance(node, _ast.Call):
                    for kw in node.keywords:
                        if (
                            kw.arg == "shell"
                            and isinstance(kw.value, _ast.Constant)
                            and kw.value.value is True
                        ):
                            violations.append(f"{py_file.name}:{node.lineno}")
        assert not violations, (
            "shell=True keyword arg found in MCP source calls:\n"
            + "\n".join(violations)
        )
 
    def test_sandbox_executor_passes_cwd_project_root(self, tmp_path: Path):
        """SandboxExecutor must pin cwd to the project root."""
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, "ok", "")
        from oct.mcp.sandbox import SandboxExecutor
        executor = SandboxExecutor(backend=mock_backend)
        executor.run(cmd=["echo"], cwd=tmp_path, timeout=10)
        call_kwargs = mock_backend.run.call_args[1]
        assert call_kwargs["cwd"] == tmp_path
 
    def test_sanitised_env_excludes_pythonpath(self, tmp_path: Path):
        """PYTHONPATH must be stripped from the sandbox environment."""
        with patch("os.environ", {**__import__("os").environ, "PYTHONPATH": "/evil"}):
            env = _sanitise_env(tmp_path)
        assert "PYTHONPATH" not in env
 
    def test_sanitised_env_excludes_api_key(self, tmp_path: Path):
        """API_KEY (secret substring) must be stripped from the sandbox environment."""
        with patch("os.environ", {**__import__("os").environ, "API_KEY": "exposed"}):
            env = _sanitise_env(tmp_path)
        assert "API_KEY" not in env
 
    def test_sandbox_executor_timeout_returns_timed_out(self, tmp_path: Path):
        """SandboxExecutor catches TimeoutExpired and sets timed_out=True."""
        import subprocess
        mock_backend = MagicMock()
        mock_backend.run.side_effect = subprocess.TimeoutExpired(cmd="echo", timeout=1)
        from oct.mcp.sandbox import SandboxExecutor
        executor = SandboxExecutor(backend=mock_backend)
        result = executor.run(cmd=["echo"], cwd=tmp_path, timeout=1)
        assert result.timed_out is True
        assert result.exit_code == 1
 
    def test_sandbox_executor_output_truncated_when_exceeded(self, tmp_path: Path):
        """SandboxResult.output_truncated is True when output exceeds max_output_bytes."""
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, "x" * 5000, "")
        from oct.mcp.sandbox import SandboxExecutor
        executor = SandboxExecutor(backend=mock_backend)
        result = executor.run(cmd=["echo"], cwd=tmp_path, timeout=10, max_output_bytes=100)
        assert result.output_truncated is True
 
 
# ---------------------------------------------------------------------------
# §12.5 Redaction
# ---------------------------------------------------------------------------
 
class TestRedactionHardening:
    def test_api_key_in_output_redacted(self, tmp_path: Path):
        """T-03: Secrets in sandbox stdout must be redacted before returning."""
        state, _ = _make_state(
            tmp_path, sandbox_output="Running lint\nAPI_KEY=s3cr3t_value\nDone"
        )
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool("oct_lint", {})
        assert "s3cr3t_value" not in result
 
    def test_project_root_path_replaced_with_placeholder(self, tmp_path: Path):
        """T-07: Absolute project path in output is anonymised."""
        state, _ = _make_state(
            tmp_path, sandbox_output=f"Error in {tmp_path}/src/module.py line 42"
        )
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool("oct_lint", {})
        assert str(tmp_path) not in result
        assert "[PROJECT]" in result
 
    def test_large_output_truncated_with_marker(self, tmp_path: Path):
        """Output > max_output_bytes is truncated (output_truncated flag set)."""
        large = "safe_content " * 100_000  # ~1.3 MB
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, large, "")
        from oct.mcp.sandbox import SandboxExecutor
        executor = SandboxExecutor(backend=mock_backend)
        result = executor.run(cmd=["echo"], cwd=tmp_path, timeout=10,
                              max_output_bytes=1_048_576)
        # If output exceeds limit, it should be truncated
        total_bytes = len((result.stdout + result.stderr).encode("utf-8", errors="replace"))
        assert total_bytes <= 1_048_576
 
 
# ---------------------------------------------------------------------------
# §12.6 Audit
# ---------------------------------------------------------------------------
 
class TestAuditHardening:
    def test_every_tool_call_writes_jsonl_record(self, tmp_path: Path):
        """Every _run_tool() call must write exactly one JSONL audit record."""
        state, logger = _make_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool("oct_lint", {})
        assert logger.current_path is not None
        lines = logger.current_path.read_text(encoding="utf-8").strip().splitlines()
        assert len(lines) == 1
        record = json.loads(lines[0])
        assert record["tool"] == "oct_lint"
 
    def test_audit_record_not_returned_to_caller(self, tmp_path: Path):
        """_run_tool() returns a plain string, not the audit record dict."""
        state, logger = _make_state(tmp_path, sandbox_output="lint output")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool("oct_lint", {})
        assert isinstance(result, str)
        # The result should NOT contain audit-record keys
        assert '"session_id"' not in result
        assert '"request_id"' not in result
        assert '"policy_rule"' not in result
 
    def test_audit_record_has_session_id(self, tmp_path: Path):
        state, logger = _make_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool("oct_lint", {})
        record = _last_audit(logger)
        assert record.get("session_id") == "hardening-test"
 
    def test_audit_record_has_request_id(self, tmp_path: Path):
        state, logger = _make_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool("oct_lint", {})
        record = _last_audit(logger)
        assert record.get("request_id")  # non-empty UUID string
 
    def test_policy_denial_writes_denied_record(self, tmp_path: Path):
        state, logger = _make_state(tmp_path, safe_mode=True)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool("oct_lint", {})
        record = _last_audit(logger)
        assert record["decision"] == "denied"
 
    def test_validation_failure_writes_denied_record(self, tmp_path: Path):
        state, logger = _make_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool("oct_lint", {"paths": ["../../etc/passwd"]})
        record = _last_audit(logger)
        assert record["decision"] == "denied"
        assert record["policy_rule"] == "validation_failure"
 
    def test_hitl_confirmation_writes_requires_confirmation_record(self, tmp_path: Path):
        state, logger = _make_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_clean", {"confirm": False})
        record = _last_audit(logger)
        assert record["decision"] == "requires_confirmation"
 
 
# ---------------------------------------------------------------------------
# §12.7 General / Miscellaneous
# ---------------------------------------------------------------------------
 
class TestGeneralHardening:
    def test_confirm_default_is_false_on_all_write_models(self):
        """confirm: bool = False must be the default on all write tool models."""
        write_tools = [
            "oct_format", "oct_docs", "oct_export_source",
            "oct_scaffold", "oct_clean", "oct_install_hooks",
            "oct_git_commit", "oct_git_hooks_install", "oct_git_init",
            "oct_git_changelog",
        ]
        errors = []
        for tool_name in write_tools:
            model_cls = _TOOL_MODEL_MAP.get(tool_name)
            if model_cls is None:
                errors.append(f"{tool_name}: not in _TOOL_MODEL_MAP")
                continue
            # Instantiate with bare minimum required fields (none besides confirm)
            # and check confirm defaults to False.
            try:
                # Find minimum required fields by attempting empty construction.
                # We only care about the 'confirm' default here.
                fields = model_cls.model_fields
                if "confirm" not in fields:
                    errors.append(f"{tool_name}: no 'confirm' field")
                    continue
                default = fields["confirm"].default
                if default is not False:
                    errors.append(f"{tool_name}: confirm default={default!r} (expected False)")
            except Exception as exc:
                errors.append(f"{tool_name}: error checking confirm default: {exc}")
        assert not errors, "\n".join(errors)
 
    def test_tool_manifest_has_20_entries(self):
        """TOOL_MANIFEST must contain exactly 20 tools (9 read + 11 write)."""
        assert len(TOOL_MANIFEST) == 20
 
    def test_tool_model_map_has_20_entries(self):
        """_TOOL_MODEL_MAP must contain exactly 20 tools."""
        assert len(_TOOL_MODEL_MAP) == 20
 
    def test_tool_manifest_and_model_map_keys_match(self):
        """Every key in TOOL_MANIFEST must also be in _TOOL_MODEL_MAP and vice versa."""
        manifest_keys = set(TOOL_MANIFEST.keys())
        model_keys = set(_TOOL_MODEL_MAP.keys())
        assert manifest_keys == model_keys, (
            f"Mismatch — manifest only: {manifest_keys - model_keys}, "
            f"model_map only: {model_keys - manifest_keys}"
        )
 
    def test_read_only_tools_are_non_destructive(self):
        """All 9 Phase 5A read-only tools must have destructive=False."""
        read_tools = [
            "oct_lint", "oct_health", "oct_skeleton", "oct_deps",
            "oct_test", "oct_typecheck", "oct_diag",
            "oct_git_status", "oct_git_check",
        ]
        errors = []
        for name in read_tools:
            spec = TOOL_MANIFEST.get(name)
            if spec and spec.destructive:
                errors.append(f"{name}: destructive=True (expected False)")
        assert not errors, "\n".join(errors)
 
    def test_read_only_tools_auto_approve_true(self):
        """All 9 Phase 5A read-only tools must have auto_approve=True."""
        read_tools = [
            "oct_lint", "oct_health", "oct_skeleton", "oct_deps",
            "oct_test", "oct_typecheck", "oct_diag",
            "oct_git_status", "oct_git_check",
        ]
        errors = []
        for name in read_tools:
            spec = TOOL_MANIFEST.get(name)
            if spec and not spec.auto_approve:
                errors.append(f"{name}: auto_approve=False (expected True)")
        assert not errors, "\n".join(errors)