261 lines
10 KiB
Python
261 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
|
|
|
from backend.app.db.models import AuditEvent, GatewayUser, JobRun, Policy
|
|
from backend.app.services import (
|
|
action_execution,
|
|
deployment_broker,
|
|
policy_engine,
|
|
user_access,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def session(tmp_path):
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'acceptance.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as value:
|
|
yield value
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"action", ["runner.acceptance.preflight", "runner.acceptance.quiesce"]
|
|
)
|
|
@pytest.mark.parametrize(
|
|
"target,source,expected",
|
|
[
|
|
("tower/ci-acceptance-residue", "chatgpt-mcp", True),
|
|
("tower/ci-acceptance-residue", "manual", True),
|
|
("tower/ci-acceptance-residue", "voice", False),
|
|
("other-target", "chatgpt-mcp", False),
|
|
("*", "manual", False),
|
|
],
|
|
)
|
|
def test_finite_policy_never_inherits_wildcard_or_voice(
|
|
session, action, target, source, expected
|
|
):
|
|
session.add(
|
|
Policy(
|
|
id="wildcard",
|
|
action=action,
|
|
target="*",
|
|
manual="allowed",
|
|
voice="allowed",
|
|
risk="low",
|
|
)
|
|
)
|
|
session.commit()
|
|
result = policy_engine.evaluate_policy(
|
|
session, action=action, target=target, source=source
|
|
)
|
|
assert (result["decision"] != "blocked") is expected
|
|
assert result["policy_id"] is None
|
|
if expected and action.endswith(".quiesce"):
|
|
assert result["decision"] == "approval_required"
|
|
assert (
|
|
policy_engine.evaluate_policy(
|
|
session, action="container.stop", target=target, source=source
|
|
)["decision"]
|
|
== "blocked"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("role", ["operator", "read_only", "custom"])
|
|
@pytest.mark.parametrize(
|
|
"action", ["runner.acceptance.preflight", "runner.acceptance.quiesce"]
|
|
)
|
|
def test_human_nonadmin_cannot_call_new_executor(session, monkeypatch, role, action):
|
|
user = GatewayUser(
|
|
id="human",
|
|
provider="authentik",
|
|
issuer="https://issuer",
|
|
subject="jens",
|
|
role=role,
|
|
)
|
|
session.add(user)
|
|
session.commit()
|
|
actor = user_access.actor_identity.set("authentik:jens")
|
|
try:
|
|
monkeypatch.setattr(
|
|
deployment_broker.DeploymentBrokerClient,
|
|
"runner_acceptance",
|
|
lambda *a, **k: pytest.fail("must not call broker"),
|
|
)
|
|
result = action_execution.execute_governed_action(
|
|
session,
|
|
action=action,
|
|
target="tower/ci-acceptance-residue",
|
|
source="manual",
|
|
risk="high",
|
|
reason="inspect fixture",
|
|
original_policy_decision="allowed",
|
|
)
|
|
assert result["policyDecision"] == "blocked"
|
|
assert session.exec(select(JobRun)).all() == []
|
|
finally:
|
|
user_access.actor_identity.reset(actor)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"action,status,success",
|
|
[
|
|
("runner.acceptance.preflight", "ready", True),
|
|
("runner.acceptance.preflight", "blocked", False),
|
|
("runner.acceptance.quiesce", "quiesced", True),
|
|
("runner.acceptance.quiesce", "already_quiesced", True),
|
|
("runner.acceptance.quiesce", "partial", False),
|
|
("runner.acceptance.quiesce", "blocked", False),
|
|
("runner.acceptance.quiesce", "ready", False),
|
|
],
|
|
)
|
|
def test_typed_mcp_records_actual_controller_result(
|
|
session, monkeypatch, action, status, success
|
|
):
|
|
monkeypatch.setattr(
|
|
deployment_broker.DeploymentBrokerClient,
|
|
"runner_acceptance",
|
|
lambda *_, **kw: {
|
|
"status": status,
|
|
"revision": "a" * 40,
|
|
"blockers": ["terminal_origin_claim_missing"],
|
|
},
|
|
)
|
|
result = action_execution.execute_governed_action(
|
|
session,
|
|
action=action,
|
|
target="tower/ci-acceptance-residue",
|
|
source="chatgpt-mcp",
|
|
risk="high",
|
|
reason="inspect fixture",
|
|
original_policy_decision="allowed",
|
|
)
|
|
assert result["success"] is success
|
|
assert session.get(JobRun, result["jobId"]).status == (
|
|
"succeeded" if success else "failed"
|
|
)
|
|
assert (
|
|
session.get(AuditEvent, result["auditId"]).target
|
|
== "tower/ci-acceptance-residue"
|
|
)
|
|
assert result["blockers"] == ["terminal_origin_claim_missing"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"target,source,decision",
|
|
[
|
|
("other", "chatgpt-mcp", "allowed"),
|
|
("tower/ci-acceptance-residue", "voice", "allowed"),
|
|
("tower/ci-acceptance-residue", "chatgpt-mcp", "blocked"),
|
|
("tower/ci-acceptance-residue", "manual", "allowed"),
|
|
],
|
|
)
|
|
def test_unsupported_target_or_auth_refuses_before_job(
|
|
session, monkeypatch, target, source, decision
|
|
):
|
|
monkeypatch.setattr(
|
|
deployment_broker.DeploymentBrokerClient,
|
|
"runner_acceptance",
|
|
lambda *a, **k: pytest.fail("must not call broker"),
|
|
)
|
|
result = action_execution.execute_governed_action(
|
|
session,
|
|
action="runner.acceptance.quiesce",
|
|
target=target,
|
|
source=source,
|
|
risk="high",
|
|
reason="inspect fixture",
|
|
original_policy_decision=decision,
|
|
)
|
|
assert result["policyDecision"] == "blocked"
|
|
assert session.exec(select(JobRun)).all() == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"preflight,path",
|
|
[
|
|
(True, "/v1/appops/runner-acceptance/preflight"),
|
|
(False, "/v1/appops/runner-acceptance/quiesce"),
|
|
],
|
|
)
|
|
def test_broker_has_fixed_signed_path_and_bounded_body(monkeypatch, preflight, path):
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
deployment_broker.DeploymentBrokerClient,
|
|
"_request",
|
|
lambda _, *a, **kw: calls.append((a, kw)) or {},
|
|
)
|
|
deployment_broker.DeploymentBrokerClient().runner_acceptance(
|
|
reason="x" * 1100, preflight=preflight
|
|
)
|
|
assert calls[0][0] == ("POST", path)
|
|
assert calls[0][1]["payload"] == {"reason": "x" * 1000}
|
|
assert "secret" in calls[0][1]
|
|
|
|
|
|
def test_controller_canary_is_discarded_before_job_and_audit(session, monkeypatch):
|
|
monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_acceptance", lambda *a, **k: {
|
|
"status": "Bearer SECRET-CANARY", "reason": "Bearer SECRET-CANARY", "revision": "SECRET-CANARY",
|
|
"blockers": ["SECRET-CANARY"], "verifiedRunners": ["SECRET-CANARY"], "projectbrainDeliveryRequired": "SECRET-CANARY",
|
|
})
|
|
result = action_execution.execute_governed_action(session, action="runner.acceptance.quiesce", target="tower/ci-acceptance-residue",
|
|
source="chatgpt-mcp", risk="high", reason="inspect fixture",
|
|
original_policy_decision="allowed")
|
|
assert result["success"] is False
|
|
assert result["blockers"] == ["controller_response_untrusted"]
|
|
assert "SECRET-CANARY" not in str(result)
|
|
assert "SECRET-CANARY" not in str(session.get(JobRun, result["jobId"]))
|
|
assert "SECRET-CANARY" not in str(session.get(AuditEvent, result["auditId"]))
|
|
|
|
|
|
def test_acceptance_transport_loss_is_unknown_never_effect_free(monkeypatch):
|
|
monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "_request", lambda *a, **kw: {"status": "unreachable"})
|
|
result = deployment_broker.DeploymentBrokerClient().runner_acceptance(reason="preserve residue", preflight=False)
|
|
assert result["status"] == "unknown" and result["noMutation"] is False
|
|
assert result["effectsPerformed"] is None
|
|
|
|
|
|
def test_acceptance_unknown_transport_is_preserved_in_job_result_and_audit(session, monkeypatch):
|
|
monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_acceptance", lambda *a, **kw:
|
|
{"status": "unknown", "noMutation": False, "effectsPerformed": None,
|
|
"blockers": ["maintenance_outcome_unknown"]})
|
|
result = action_execution.execute_governed_action(session, action="runner.acceptance.quiesce",
|
|
target="tower/ci-acceptance-residue", source="chatgpt-mcp", risk="high", reason="preserve residue",
|
|
original_policy_decision="allowed")
|
|
assert result["controllerStatus"] == "unknown" and result["result"] == "unknown"
|
|
assert result["noMutation"] is False and result["effectsPerformed"] is None
|
|
|
|
|
|
@pytest.mark.parametrize("action", ["runner.acceptance.preflight", "runner.acceptance.quiesce"])
|
|
def test_actual_mcp_request_preserves_actor_guard_and_never_grants_automation(session, monkeypatch, action):
|
|
from backend.app.services import mcp_operations
|
|
session.add(GatewayUser(id="operator", provider="authentik", issuer="test", subject="operator", role="operator"))
|
|
session.commit()
|
|
token = user_access.actor_identity.set("authentik:operator")
|
|
monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_acceptance",
|
|
lambda *a, **k: pytest.fail("nonadmin must not reach broker"))
|
|
try:
|
|
result = mcp_operations.request_operation(session, action=action, target="tower/ci-acceptance-residue",
|
|
reason="preserve finite residue", idempotency_key="nonadmin-"+action)
|
|
finally:
|
|
user_access.actor_identity.reset(token)
|
|
assert result["policyDecision"] == "blocked"
|
|
assert session.exec(select(JobRun)).all() == []
|
|
assert policy_engine.evaluate_policy(session, action=action, target="tower/ci-acceptance-residue", source="automation")["decision"] == "blocked"
|
|
profile = action_execution.action_execution_profile(action, "allowed", "operator")
|
|
assert profile["executionMode"] == "governance_only" and profile["commandExecutionSupported"] is False
|
|
|
|
|
|
@pytest.mark.parametrize("action,target", [("runner.acceptance.quiesce", "other-target"),
|
|
("runner.acceptance.preflight", "*"),
|
|
("container.stop", "tower/ci-acceptance-residue")])
|
|
def test_actual_mcp_request_keeps_wrong_target_and_generic_stop_denied(session, monkeypatch, action, target):
|
|
from backend.app.services import mcp_operations
|
|
monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_acceptance", lambda *a, **k: pytest.fail("must remain denied"))
|
|
result = mcp_operations.request_operation(session, action=action, target=target,
|
|
reason="preserve finite residue", idempotency_key="deny-"+action)
|
|
assert result["policyDecision"] == "blocked" and result["executed"] is False
|
|
assert session.exec(select(JobRun)).all() == []
|
|
assert mcp_operations.mcp_policy_source(action, target) == "automation"
|