358 lines
13 KiB
Python
358 lines
13 KiB
Python
# #region Test.McpApprovals [C:5] [TYPE Module] [SEMANTICS test,mcp,approval,cas,rbac]
|
|
# @BRIEF Contract tests for MCP-owned human approval gates.
|
|
# @RELATION VERIFIES -> [Services.McpApprovals]
|
|
# @TEST_EDGE duplicate_request -> same pending gate
|
|
# @TEST_EDGE foreign_user_decision -> gate not found
|
|
# @TEST_EDGE concurrent_decision -> one CAS winner
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import secrets
|
|
import sys
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from src.core.auth.security import get_password_hash
|
|
from src.core.database import Base
|
|
from src.models.auth import McpToolInvocationRecord, Permission, Role, User
|
|
from src.models.scenario_approval import ActionApprovalGate
|
|
from src.services.mcp_approvals import (
|
|
claim_approved_mcp_dispatch,
|
|
claim_approved_mcp_dispatch_lease,
|
|
complete_mcp_dispatch,
|
|
decide_mcp_approval,
|
|
list_pending_mcp_approvals,
|
|
recover_stale_mcp_dispatches,
|
|
renew_mcp_dispatch_lease,
|
|
request_mcp_approval,
|
|
poll_approved_mcp_dispatches,
|
|
)
|
|
import src.services.mcp_approvals as approval_service
|
|
from src.services.mcp_provenance import validate_continuation_payload
|
|
|
|
|
|
# #region Test.McpApprovals.Fixture [C:2] [TYPE Function]
|
|
# @ingroup Test.McpApprovals
|
|
# @BRIEF Build two users with distinct live approval permissions.
|
|
def _fixture():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
db = sessionmaker(bind=engine)()
|
|
suffix = secrets.token_hex(4)
|
|
permission = Permission(resource="scenario", action="RUN_PROD")
|
|
role = Role(name=f"Approver-{suffix}", is_admin=False, permissions=[permission])
|
|
owner = User(username=f"owner-{suffix}", password_hash=get_password_hash("pw"), is_active=True, roles=[role])
|
|
foreign = User(username=f"foreign-{suffix}", password_hash=get_password_hash("pw"), is_active=True)
|
|
db.add_all([owner, foreign])
|
|
db.commit()
|
|
record = McpToolInvocationRecord(
|
|
request_id="req-1",
|
|
client_id="client-1",
|
|
subject=owner.username,
|
|
principal_type="user",
|
|
operation="tools/call",
|
|
tool_name="execute_migration",
|
|
arguments_sha256="b" * 64,
|
|
outcome="denied",
|
|
error_code="approval_required",
|
|
)
|
|
db.add(record)
|
|
db.commit()
|
|
return db, owner, foreign, record
|
|
|
|
|
|
# #endregion Test.McpApprovals.Fixture
|
|
|
|
|
|
# #region Test.McpApprovals.Lifecycle [C:5] [TYPE Function]
|
|
# @ingroup Test.McpApprovals
|
|
# @BRIEF Verify idempotency, human ownership, live permission, and one-shot CAS decision.
|
|
def test_mcp_approval_is_idempotent_and_one_shot() -> None:
|
|
db, owner, foreign, record = _fixture()
|
|
first = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="execute_migration",
|
|
request_hash="b" * 64,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
second = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="execute_migration",
|
|
request_hash="b" * 64,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
assert first.id == second.id
|
|
assert len(list_pending_mcp_approvals(db, owner)) == 1
|
|
|
|
try:
|
|
decide_mcp_approval(db, first.id, foreign, "approve")
|
|
except ValueError as exc:
|
|
assert str(exc) == "approval gate not found"
|
|
else:
|
|
raise AssertionError("foreign user decided an MCP gate")
|
|
|
|
result = decide_mcp_approval(db, first.id, owner, "approve")
|
|
assert result["status"] == "approved"
|
|
assert db.get(McpToolInvocationRecord, record.id).dispatch_status == "queued"
|
|
assert claim_approved_mcp_dispatch(db, record.id, "worker-1") is True
|
|
assert claim_approved_mcp_dispatch(db, record.id, "worker-2") is False
|
|
claimed = db.get(McpToolInvocationRecord, record.id)
|
|
assert claimed.dispatch_status == "claimed"
|
|
assert claimed.dispatch_worker_id == "worker-1"
|
|
try:
|
|
decide_mcp_approval(db, first.id, owner, "deny")
|
|
except ValueError as exc:
|
|
assert str(exc) == "approval gate already decided"
|
|
else:
|
|
raise AssertionError("MCP gate was decided twice")
|
|
db.close()
|
|
|
|
|
|
def test_approved_scenario_start_stays_queued_without_mcp_dispatcher() -> None:
|
|
db, owner, _, record = _fixture()
|
|
record.tool_name = "start_scenario_run"
|
|
db.commit()
|
|
gate = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="start_scenario_run",
|
|
request_hash="b" * 64,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
decide_mcp_approval(db, gate.id, owner, "approve")
|
|
|
|
result = poll_approved_mcp_dispatches(db, "worker-1")
|
|
|
|
assert result == [{"invocation_id": record.id, "status": "not_ready", "error": "dispatcher_not_registered"}]
|
|
assert db.get(McpToolInvocationRecord, record.id).dispatch_status == "queued"
|
|
db.close()
|
|
|
|
|
|
def test_gate_rejects_request_hash_drift() -> None:
|
|
db, owner, _, record = _fixture()
|
|
try:
|
|
request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="execute_migration",
|
|
request_hash="c" * 64,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
except ValueError as exc:
|
|
assert str(exc) == "payload_drift"
|
|
else:
|
|
raise AssertionError("approval accepted a changed payload hash")
|
|
db.close()
|
|
|
|
|
|
def test_gate_rejects_operation_binding_mismatch() -> None:
|
|
db, _, _, record = _fixture()
|
|
try:
|
|
request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="run_backup",
|
|
request_hash=record.arguments_sha256,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
except ValueError as exc:
|
|
assert str(exc) == "operation_binding_mismatch"
|
|
else:
|
|
raise AssertionError("approval accepted a different operation binding")
|
|
db.close()
|
|
|
|
|
|
def test_expired_lease_is_requeued_and_fences_old_worker() -> None:
|
|
db, owner, _, record = _fixture()
|
|
gate = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="execute_migration",
|
|
request_hash=record.arguments_sha256,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
decide_mcp_approval(db, gate.id, owner, "approve")
|
|
|
|
first_token = claim_approved_mcp_dispatch_lease(db, record.id, "worker-1")
|
|
assert first_token
|
|
claimed = db.get(McpToolInvocationRecord, record.id)
|
|
claimed.dispatch_lease_expires_at = datetime.now(UTC) - timedelta(seconds=1)
|
|
db.commit()
|
|
|
|
assert recover_stale_mcp_dispatches(db) == 1
|
|
assert db.get(McpToolInvocationRecord, record.id).dispatch_status == "queued"
|
|
|
|
second_token = claim_approved_mcp_dispatch_lease(db, record.id, "worker-2")
|
|
assert second_token and second_token != first_token
|
|
assert complete_mcp_dispatch(db, record.id, "worker-1", first_token, {"stale": True}) is False
|
|
assert complete_mcp_dispatch(db, record.id, "worker-2", second_token, {"ok": True}) is True
|
|
assert db.get(McpToolInvocationRecord, record.id).dispatch_status == "completed"
|
|
db.close()
|
|
|
|
|
|
def test_active_worker_can_renew_lease_but_expired_worker_cannot() -> None:
|
|
db, owner, _, record = _fixture()
|
|
gate = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="execute_migration",
|
|
request_hash=record.arguments_sha256,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
decide_mcp_approval(db, gate.id, owner, "approve")
|
|
token = claim_approved_mcp_dispatch_lease(db, record.id, "worker-1")
|
|
assert token
|
|
assert renew_mcp_dispatch_lease(db, record.id, "worker-1", token, lease_seconds=30) is True
|
|
claimed = db.get(McpToolInvocationRecord, record.id)
|
|
claimed.dispatch_lease_expires_at = datetime.now(UTC) - timedelta(seconds=1)
|
|
db.commit()
|
|
assert renew_mcp_dispatch_lease(db, record.id, "worker-1", token) is False
|
|
assert complete_mcp_dispatch(db, record.id, "worker-1", token, {"late": True}) is False
|
|
db.close()
|
|
|
|
|
|
def test_continuation_payload_rejects_secrets_and_bounds_size() -> None:
|
|
try:
|
|
validate_continuation_payload({"access_token": "secret"})
|
|
except ValueError as exc:
|
|
assert str(exc) == "continuation_payload_secret_field"
|
|
else:
|
|
raise AssertionError("secret continuation field was accepted")
|
|
|
|
try:
|
|
validate_continuation_payload({"value": "x" * (64 * 1024)})
|
|
except ValueError as exc:
|
|
assert str(exc) == "continuation_payload_too_large"
|
|
else:
|
|
raise AssertionError("oversized continuation payload was accepted")
|
|
|
|
|
|
def test_maintenance_dispatcher_receives_persisted_payload(monkeypatch) -> None:
|
|
db, owner, _, record = _fixture()
|
|
record.tool_name = "start_maintenance"
|
|
record.continuation_payload = {"source_env": "dev", "target_env": "preprod"}
|
|
db.commit()
|
|
gate = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="start_maintenance",
|
|
request_hash=record.arguments_sha256,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
decide_mcp_approval(db, gate.id, owner, "approve")
|
|
received = []
|
|
monkeypatch.setattr(approval_service, "enqueue_start_maintenance", lambda invocation: received.append(invocation) or {"ok": True})
|
|
result = poll_approved_mcp_dispatches(db, "worker-1")
|
|
|
|
assert result[0]["status"] == "completed"
|
|
assert received[0].continuation_payload == {"source_env": "dev", "target_env": "preprod"}
|
|
assert db.get(McpToolInvocationRecord, record.id).dispatch_status == "completed"
|
|
assert db.get(McpToolInvocationRecord, record.id).result_sha256
|
|
db.close()
|
|
|
|
|
|
def test_approved_retry_reuses_gate_and_dispatches_same_maintenance_payload(monkeypatch) -> None:
|
|
db, owner, _, record = _fixture()
|
|
record.tool_name = "start_maintenance"
|
|
record.continuation_payload = {"source_env": "dev", "target_env": "preprod"}
|
|
db.commit()
|
|
gate = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="start_maintenance",
|
|
request_hash=record.arguments_sha256,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
decide_mcp_approval(db, gate.id, owner, "approve", "approved for acceptance test")
|
|
retried = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="start_maintenance",
|
|
request_hash=record.arguments_sha256,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
assert retried.id == gate.id
|
|
assert db.query(ActionApprovalGate).filter(ActionApprovalGate.owner_id == record.id).count() == 1
|
|
|
|
received = []
|
|
monkeypatch.setattr(approval_service, "enqueue_start_maintenance", lambda invocation: received.append(invocation) or {"ok": True})
|
|
result = poll_approved_mcp_dispatches(db, "worker-acceptance")
|
|
|
|
assert result[0]["status"] == "completed"
|
|
assert received[0].continuation_payload == {"source_env": "dev", "target_env": "preprod"}
|
|
db.close()
|
|
|
|
|
|
def test_failed_maintenance_dispatch_is_terminal_and_does_not_store_result(monkeypatch) -> None:
|
|
db, owner, _, record = _fixture()
|
|
record.tool_name = "start_maintenance"
|
|
db.commit()
|
|
gate = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="start_maintenance",
|
|
request_hash=record.arguments_sha256,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
decide_mcp_approval(db, gate.id, owner, "approve")
|
|
|
|
monkeypatch.setattr(
|
|
approval_service,
|
|
"enqueue_start_maintenance",
|
|
lambda payload: (_ for _ in ()).throw(RuntimeError("provider unavailable")),
|
|
)
|
|
result = poll_approved_mcp_dispatches(db, "worker-1")
|
|
|
|
assert result[0]["status"] == "error"
|
|
persisted = db.get(McpToolInvocationRecord, record.id)
|
|
assert persisted.dispatch_status == "failed"
|
|
assert persisted.result_sha256 is None
|
|
assert persisted.error_code == "MCP_DISPATCH_RuntimeError"
|
|
db.close()
|
|
|
|
|
|
def test_maintenance_dispatcher_receives_invocation_record_not_client_payload(monkeypatch) -> None:
|
|
db, owner, _, record = _fixture()
|
|
record.tool_name = "start_maintenance"
|
|
record.continuation_payload = {"approved": True}
|
|
db.commit()
|
|
gate = request_mcp_approval(
|
|
db,
|
|
invocation_id=record.id,
|
|
operation="mcp_tool_call",
|
|
tool_name="start_maintenance",
|
|
request_hash=record.arguments_sha256,
|
|
required_permission=("scenario", "RUN_PROD"),
|
|
)
|
|
decide_mcp_approval(db, gate.id, owner, "approve")
|
|
received = []
|
|
monkeypatch.setattr(approval_service, "enqueue_start_maintenance", lambda invocation: received.append(invocation) or {"ok": True})
|
|
poll_approved_mcp_dispatches(db, "worker-1")
|
|
|
|
assert received[0].id == record.id
|
|
assert received[0].continuation_payload == {"approved": True}
|
|
db.close()
|
|
|
|
|
|
# #endregion Test.McpApprovals.Lifecycle
|
|
|
|
# #endregion Test.McpApprovals
|