Merge remote-tracking branch 'prgs/master' into feat/issue-651-usage-cost-analytics
# Conflicts: # docs/webui-authz-audit.md # webui/console_authz.py
This commit is contained in:
@@ -0,0 +1,502 @@
|
||||
"""Sanctioned restart / graceful reload control tests (#642).
|
||||
|
||||
Acceptance criteria under test:
|
||||
|
||||
1. The sanctioned restart path is implemented behind gates (capability,
|
||||
confirmation, operator authorization, host hook).
|
||||
2. Manual ``pkill`` stays forbidden and is classified as contamination.
|
||||
3. Post-restart mutations require clean health/session proof.
|
||||
4. Authorized restart preview, unauthorized deny, contamination classification.
|
||||
5. No entry point exposes a raw kill.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
import mcp_namespace_health
|
||||
import runtime_recovery_guard
|
||||
from task_capability_map import TASK_CAPABILITY_MAP
|
||||
from webui import console_audit, console_authz, gated_actions, sanctioned_restart
|
||||
|
||||
NAMESPACE = "gitea-author"
|
||||
|
||||
# An operator-authorized, hook-configured host. Passed explicitly so no test
|
||||
# depends on (or mutates) the real process environment.
|
||||
READY_ENV = {
|
||||
sanctioned_restart.RESTART_HOOK_ENV: "launchd:cc.prgs.gitea-author",
|
||||
runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV: "ops-ticket-4821",
|
||||
}
|
||||
|
||||
|
||||
def admin(subject: str = "[email protected]") -> console_authz.Principal:
|
||||
return console_authz.Principal(
|
||||
subject=subject,
|
||||
role=console_authz.ADMIN,
|
||||
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
|
||||
authenticated=True,
|
||||
)
|
||||
|
||||
|
||||
def viewer() -> console_authz.Principal:
|
||||
return console_authz.Principal(
|
||||
subject="[email protected]",
|
||||
role=console_authz.VIEWER,
|
||||
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
|
||||
authenticated=True,
|
||||
)
|
||||
|
||||
|
||||
class TestCapabilityWiring(unittest.TestCase):
|
||||
"""AC1: authority is declared, not invented by the console."""
|
||||
|
||||
def test_actions_resolve_through_the_capability_map(self):
|
||||
for action_id in (
|
||||
sanctioned_restart.ACTION_RESTART_NAMESPACE,
|
||||
sanctioned_restart.ACTION_RELOAD_NAMESPACE,
|
||||
):
|
||||
with self.subTest(action=action_id):
|
||||
action = console_authz.get_action(action_id)
|
||||
self.assertIsNotNone(action)
|
||||
self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
|
||||
self.assertEqual(
|
||||
action.mcp_permission,
|
||||
TASK_CAPABILITY_MAP[action.task_key]["permission"],
|
||||
)
|
||||
|
||||
def test_restart_permission_is_not_a_gitea_operation(self):
|
||||
"""No configured Gitea profile should satisfy a host restart."""
|
||||
permission = TASK_CAPABILITY_MAP["restart_namespace"]["permission"]
|
||||
self.assertFalse(permission.startswith("gitea."))
|
||||
|
||||
def test_restart_is_destructive_dual_control_break_glass(self):
|
||||
action = console_authz.get_action(
|
||||
sanctioned_restart.ACTION_RESTART_NAMESPACE
|
||||
)
|
||||
self.assertEqual(action.action_class, console_authz.CLASS_DESTRUCTIVE)
|
||||
self.assertEqual(action.minimum_role, console_authz.ADMIN)
|
||||
self.assertTrue(action.dual_control)
|
||||
self.assertTrue(action.break_glass)
|
||||
self.assertTrue(action.requires_confirmation)
|
||||
|
||||
def test_reload_is_privileged_but_not_destructive(self):
|
||||
action = console_authz.get_action(
|
||||
sanctioned_restart.ACTION_RELOAD_NAMESPACE
|
||||
)
|
||||
self.assertEqual(action.action_class, console_authz.CLASS_PRIVILEGED)
|
||||
self.assertTrue(action.requires_confirmation)
|
||||
|
||||
|
||||
class TestPreview(unittest.TestCase):
|
||||
"""AC4: an authorized preview renders the plan without executing it."""
|
||||
|
||||
def test_preview_lists_the_mutation_ledger(self):
|
||||
preview = sanctioned_restart.build_restart_preview(
|
||||
NAMESPACE, principal=admin(), env=READY_ENV
|
||||
)
|
||||
steps = [entry["step"] for entry in preview["mutation_ledger"]]
|
||||
self.assertEqual(
|
||||
steps, ["quiesce", "host_restart_hook", "health_recheck", "audit"]
|
||||
)
|
||||
self.assertTrue(preview["scope_valid"])
|
||||
self.assertTrue(preview["post_restart_verification_required"])
|
||||
|
||||
def test_reload_preview_drains_instead_of_restarting(self):
|
||||
preview = sanctioned_restart.build_restart_preview(
|
||||
NAMESPACE, sanctioned_restart.MODE_RELOAD,
|
||||
principal=admin(), env=READY_ENV,
|
||||
)
|
||||
steps = [entry["step"] for entry in preview["mutation_ledger"]]
|
||||
self.assertIn("host_graceful_reload", steps)
|
||||
self.assertNotIn("host_restart_hook", steps)
|
||||
|
||||
def test_preview_never_enables_execution(self):
|
||||
preview = sanctioned_restart.build_restart_preview(
|
||||
NAMESPACE, principal=admin(), env=READY_ENV
|
||||
)
|
||||
self.assertFalse(preview["execution_enabled"])
|
||||
self.assertFalse(preview["authorization"]["execution_enabled"])
|
||||
|
||||
def test_confirmation_phrase_binds_the_namespace(self):
|
||||
self.assertTrue(
|
||||
sanctioned_restart.confirmation_matches(
|
||||
NAMESPACE, sanctioned_restart.MODE_RESTART,
|
||||
"restart gitea-author",
|
||||
)
|
||||
)
|
||||
# A phrase typed for one namespace must not authorize another.
|
||||
self.assertFalse(
|
||||
sanctioned_restart.confirmation_matches(
|
||||
"gitea-merger", sanctioned_restart.MODE_RESTART,
|
||||
"restart gitea-author",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class TestGates(unittest.TestCase):
|
||||
"""AC1/AC4: every gate denies with a stable reason code."""
|
||||
|
||||
def _assess(self, **kwargs):
|
||||
params = {
|
||||
"principal": admin(),
|
||||
"confirmation": f"restart {NAMESPACE}",
|
||||
"env": READY_ENV,
|
||||
}
|
||||
params.update(kwargs)
|
||||
namespace = params.pop("namespace", NAMESPACE)
|
||||
mode = params.pop("mode", sanctioned_restart.MODE_RESTART)
|
||||
return sanctioned_restart.assess_restart_request(
|
||||
namespace, mode, **params
|
||||
)
|
||||
|
||||
def test_authorized_confirmed_request_passes_every_gate(self):
|
||||
result = self._assess()
|
||||
self.assertTrue(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.ALLOW_HOST_ACTION_REQUIRED
|
||||
)
|
||||
|
||||
def test_passing_every_gate_is_not_an_execution_grant(self):
|
||||
"""An allowed request still never lets the console touch the process."""
|
||||
result = self._assess()
|
||||
self.assertTrue(result["allowed"])
|
||||
self.assertFalse(result["execution_enabled"])
|
||||
self.assertFalse(result["console_executes"])
|
||||
|
||||
def test_unauthorized_principal_is_denied(self):
|
||||
result = self._assess(principal=viewer())
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
|
||||
)
|
||||
|
||||
def test_anonymous_principal_is_denied(self):
|
||||
result = self._assess(principal=None)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
|
||||
)
|
||||
|
||||
def test_missing_confirmation_is_denied(self):
|
||||
result = self._assess(confirmation=None)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.DENY_CONFIRMATION_MISSING
|
||||
)
|
||||
|
||||
def test_confirmation_for_another_namespace_is_denied(self):
|
||||
result = self._assess(confirmation="restart gitea-merger")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.DENY_CONFIRMATION_MISMATCH
|
||||
)
|
||||
|
||||
def test_missing_operator_authorization_is_denied(self):
|
||||
env = {sanctioned_restart.RESTART_HOOK_ENV: "launchd:cc.prgs.author"}
|
||||
result = self._assess(env=env)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"],
|
||||
sanctioned_restart.DENY_OPERATOR_AUTHORIZATION,
|
||||
)
|
||||
|
||||
def test_missing_host_hook_is_denied_without_kill_fallback(self):
|
||||
env = {
|
||||
runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV: "ops-ticket-1",
|
||||
}
|
||||
result = self._assess(env=env)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.DENY_HOOK_NOT_CONFIGURED
|
||||
)
|
||||
|
||||
def test_fleet_scope_is_refused(self):
|
||||
for scope in ("all", "*", "fleet"):
|
||||
with self.subTest(scope=scope):
|
||||
result = self._assess(
|
||||
namespace=scope, confirmation=f"restart {scope}"
|
||||
)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.DENY_FLEET_SCOPE
|
||||
)
|
||||
|
||||
def test_unknown_namespace_is_refused(self):
|
||||
result = self._assess(
|
||||
namespace="gitea-nope", confirmation="restart gitea-nope"
|
||||
)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.DENY_UNKNOWN_NAMESPACE
|
||||
)
|
||||
|
||||
def test_unknown_mode_is_refused(self):
|
||||
result = self._assess(mode="obliterate")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.DENY_UNKNOWN_MODE
|
||||
)
|
||||
|
||||
def test_live_contamination_marker_blocks_restart(self):
|
||||
marker = runtime_recovery_guard.build_contamination_record(
|
||||
reason_class=runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL,
|
||||
command_redacted="pkill -f mcp_server.py",
|
||||
)
|
||||
result = self._assess(contamination_marker=marker)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["reason_code"], sanctioned_restart.DENY_CONTAMINATED_RUNTIME
|
||||
)
|
||||
|
||||
def test_reconciler_cleared_marker_no_longer_blocks(self):
|
||||
marker = runtime_recovery_guard.build_contamination_record(
|
||||
reason_class=runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL,
|
||||
command_redacted="pkill -f mcp_server.py",
|
||||
)
|
||||
marker = dict(marker, cleared_by_reconciler=True)
|
||||
result = self._assess(contamination_marker=marker)
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
|
||||
class TestExecutionNeverKills(unittest.TestCase):
|
||||
"""AC5: no path exposes or runs a raw process kill."""
|
||||
|
||||
def test_authorized_execution_defers_to_the_host_supervisor(self):
|
||||
result = sanctioned_restart.execute_restart(
|
||||
NAMESPACE,
|
||||
principal=admin(),
|
||||
confirmation=f"restart {NAMESPACE}",
|
||||
env=READY_ENV,
|
||||
)
|
||||
self.assertTrue(result["allowed"])
|
||||
self.assertFalse(result["success"])
|
||||
self.assertFalse(result["process_kill_executed"])
|
||||
self.assertEqual(
|
||||
result["outcome"], sanctioned_restart.ALLOW_HOST_ACTION_REQUIRED
|
||||
)
|
||||
|
||||
def test_denied_execution_reports_the_refusing_gate(self):
|
||||
result = sanctioned_restart.execute_restart(
|
||||
NAMESPACE, principal=viewer(), confirmation=f"restart {NAMESPACE}",
|
||||
env=READY_ENV,
|
||||
)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(
|
||||
result["outcome"], sanctioned_restart.DENY_UNAUTHORIZED
|
||||
)
|
||||
self.assertFalse(result["process_kill_executed"])
|
||||
|
||||
def test_module_never_spawns_a_process(self):
|
||||
path = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
"webui", "sanctioned_restart.py",
|
||||
)
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
source = handle.read()
|
||||
for forbidden in (
|
||||
"import subprocess", "import signal", "os.kill", "os.system",
|
||||
"popen",
|
||||
):
|
||||
with self.subTest(forbidden=forbidden):
|
||||
self.assertNotIn(forbidden, source.lower())
|
||||
|
||||
def test_no_surface_returns_a_kill_command(self):
|
||||
payloads = [
|
||||
sanctioned_restart.build_restart_preview(
|
||||
NAMESPACE, principal=admin(), env=READY_ENV
|
||||
),
|
||||
sanctioned_restart.restart_policy(),
|
||||
sanctioned_restart.execute_restart(
|
||||
NAMESPACE, principal=admin(),
|
||||
confirmation=f"restart {NAMESPACE}", env=READY_ENV,
|
||||
),
|
||||
]
|
||||
for payload in payloads:
|
||||
rendered = json.dumps(payload, default=str).lower()
|
||||
self.assertNotIn("kill -9", rendered)
|
||||
self.assertNotIn("pkill -f", rendered)
|
||||
|
||||
def test_policy_declares_no_raw_kill_and_no_silent_restart(self):
|
||||
policy = sanctioned_restart.restart_policy()
|
||||
self.assertFalse(policy["raw_kill_exposed"])
|
||||
self.assertFalse(policy["console_executes_process_kill"])
|
||||
self.assertFalse(policy["fleet_scope_permitted"])
|
||||
self.assertFalse(policy["silent_auto_restart_permitted"])
|
||||
self.assertTrue(policy["audit_required"])
|
||||
|
||||
|
||||
class TestContaminationClassification(unittest.TestCase):
|
||||
"""AC2: manual pkill is contamination, and it blocks clean claims."""
|
||||
|
||||
def test_manual_daemon_pkill_is_contamination(self):
|
||||
result = sanctioned_restart.classify_restart_command(
|
||||
"pkill -f mcp_server.py"
|
||||
)
|
||||
self.assertTrue(result["contamination"])
|
||||
self.assertFalse(result["clean_claim_allowed"])
|
||||
self.assertIsNotNone(result["contamination_marker"])
|
||||
self.assertEqual(
|
||||
result["sanctioned_alternative"],
|
||||
sanctioned_restart.ACTION_RESTART_NAMESPACE,
|
||||
)
|
||||
|
||||
def test_broad_process_kill_is_contamination(self):
|
||||
result = sanctioned_restart.classify_restart_command("killall -9 Python")
|
||||
self.assertTrue(result["contamination"])
|
||||
self.assertFalse(result["clean_claim_allowed"])
|
||||
|
||||
def test_marker_names_the_sanctioned_alternative(self):
|
||||
result = sanctioned_restart.classify_restart_command(
|
||||
"pkill -f mcp_server.py"
|
||||
)
|
||||
marker = result["contamination_marker"]
|
||||
self.assertIn(
|
||||
sanctioned_restart.ACTION_RESTART_NAMESPACE, marker["detail"]
|
||||
)
|
||||
|
||||
def test_benign_command_is_not_contamination(self):
|
||||
result = sanctioned_restart.classify_restart_command("git status")
|
||||
self.assertFalse(result["contamination"])
|
||||
self.assertTrue(result["clean_claim_allowed"])
|
||||
|
||||
def test_no_command_is_not_contamination(self):
|
||||
result = sanctioned_restart.classify_restart_command(None)
|
||||
self.assertFalse(result["contamination"])
|
||||
self.assertTrue(result["clean_claim_allowed"])
|
||||
|
||||
|
||||
class TestPostRestartHealth(unittest.TestCase):
|
||||
"""AC3: a clean post-restart claim needs live client-namespace proof."""
|
||||
|
||||
def test_live_client_probe_clears_the_session(self):
|
||||
result = sanctioned_restart.verify_post_restart_health(
|
||||
NAMESPACE,
|
||||
probe_result={"success": True},
|
||||
probe_source=mcp_namespace_health.PROBE_SOURCE_CLIENT,
|
||||
registered_tools=["gitea_whoami"],
|
||||
required_tool="gitea_whoami",
|
||||
)
|
||||
self.assertEqual(result["status"], sanctioned_restart.HEALTH_CLEAN)
|
||||
self.assertTrue(result["clean_claim_allowed"])
|
||||
self.assertTrue(result["mutations_allowed"])
|
||||
|
||||
def test_offline_probe_does_not_clear_the_session(self):
|
||||
result = sanctioned_restart.verify_post_restart_health(
|
||||
NAMESPACE,
|
||||
probe_result={"success": True},
|
||||
probe_source=mcp_namespace_health.PROBE_SOURCE_OFFLINE,
|
||||
registered_tools=["gitea_whoami"],
|
||||
required_tool="gitea_whoami",
|
||||
)
|
||||
self.assertFalse(result["clean_claim_allowed"])
|
||||
self.assertFalse(result["mutations_allowed"])
|
||||
|
||||
def test_failed_probe_is_unhealthy(self):
|
||||
result = sanctioned_restart.verify_post_restart_health(
|
||||
NAMESPACE,
|
||||
probe_result={"success": False, "error": "client is closing: EOF"},
|
||||
probe_source=mcp_namespace_health.PROBE_SOURCE_CLIENT,
|
||||
registered_tools=["gitea_whoami"],
|
||||
required_tool="gitea_whoami",
|
||||
)
|
||||
self.assertEqual(result["status"], sanctioned_restart.HEALTH_UNHEALTHY)
|
||||
self.assertFalse(result["clean_claim_allowed"])
|
||||
|
||||
def test_static_registration_alone_never_clears_the_session(self):
|
||||
result = sanctioned_restart.verify_post_restart_health(
|
||||
NAMESPACE,
|
||||
registered_tools=["gitea_whoami"],
|
||||
required_tool="gitea_whoami",
|
||||
)
|
||||
self.assertFalse(result["clean_claim_allowed"])
|
||||
|
||||
|
||||
class TestAuditEmission(unittest.TestCase):
|
||||
"""Every restart attempt is audited with actor, target, and result."""
|
||||
|
||||
def _run(self, principal, sink):
|
||||
prior = os.environ.get(console_audit.AUDIT_LOG_ENV)
|
||||
os.environ[console_audit.AUDIT_LOG_ENV] = sink
|
||||
try:
|
||||
return sanctioned_restart.execute_restart(
|
||||
NAMESPACE,
|
||||
principal=principal,
|
||||
confirmation=f"restart {NAMESPACE}",
|
||||
env=READY_ENV,
|
||||
request_id="req-642",
|
||||
)
|
||||
finally:
|
||||
if prior is None:
|
||||
os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
|
||||
else:
|
||||
os.environ[console_audit.AUDIT_LOG_ENV] = prior
|
||||
|
||||
def test_allowed_attempt_is_written_with_actor_and_target(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
sink = os.path.join(tmp, "audit.jsonl")
|
||||
result = self._run(admin(), sink)
|
||||
self.assertTrue(result["audit"]["written"])
|
||||
with open(sink, encoding="utf-8") as handle:
|
||||
record = json.loads(handle.read().strip())
|
||||
self.assertEqual(
|
||||
record["action"], sanctioned_restart.ACTION_RESTART_NAMESPACE
|
||||
)
|
||||
self.assertEqual(record["target"]["namespace"], NAMESPACE)
|
||||
self.assertEqual(record["target"]["mode"], "restart")
|
||||
self.assertEqual(record["result"], console_audit.RESULT_ALLOWED)
|
||||
self.assertEqual(record["actor"]["subject"], "[email protected]")
|
||||
self.assertFalse(record["metadata"]["process_kill_executed"])
|
||||
|
||||
def test_denied_attempt_is_audited_too(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
sink = os.path.join(tmp, "audit.jsonl")
|
||||
self._run(viewer(), sink)
|
||||
with open(sink, encoding="utf-8") as handle:
|
||||
record = json.loads(handle.read().strip())
|
||||
self.assertEqual(record["result"], console_audit.RESULT_DENIED)
|
||||
self.assertEqual(
|
||||
record["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
|
||||
)
|
||||
|
||||
def test_restart_audit_uses_break_glass_retention(self):
|
||||
action = console_authz.get_action(
|
||||
sanctioned_restart.ACTION_RESTART_NAMESPACE
|
||||
)
|
||||
self.assertEqual(
|
||||
console_audit.retention_class_for(action),
|
||||
console_audit.RETENTION_BREAK_GLASS,
|
||||
)
|
||||
|
||||
|
||||
class TestRegistrySurface(unittest.TestCase):
|
||||
"""AC5: the console surfaces the control, still disabled, with no kill."""
|
||||
|
||||
def test_registry_exposes_both_actions_disabled(self):
|
||||
registry = gated_actions.load_action_registry()
|
||||
for action_id in (
|
||||
sanctioned_restart.ACTION_RESTART_NAMESPACE,
|
||||
sanctioned_restart.ACTION_RELOAD_NAMESPACE,
|
||||
):
|
||||
with self.subTest(action=action_id):
|
||||
action = registry.get(action_id)
|
||||
self.assertIsNotNone(action)
|
||||
self.assertFalse(action.enabled)
|
||||
|
||||
def test_registry_preview_names_the_namespace_target(self):
|
||||
preview = gated_actions.preview_action(
|
||||
sanctioned_restart.ACTION_RESTART_NAMESPACE, namespace=NAMESPACE
|
||||
)
|
||||
target = preview["mutation_ledger"][0]["target"]
|
||||
self.assertIn(NAMESPACE, target)
|
||||
self.assertFalse(preview["enabled"])
|
||||
|
||||
def test_registry_attempt_fails_closed(self):
|
||||
result = gated_actions.attempt_action(
|
||||
sanctioned_restart.ACTION_RESTART_NAMESPACE, namespace=NAMESPACE
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user