Simone
e56eae95c7
Adds operations/services.py: a fixed ManagedService allow-list (act-runner, act-runner-rust, vmware-host, ci-web) with find_service/service_states/control_service. control_service builds an argument list only (never shell=True), validates the action against ALLOWED_ACTIONS (ValueError->400), refuses stop on restart-only services (ci-web, allow_stop=False) without building a command, and defers ci-web's own restart via 'sudo -n systemd-run --on-active=2 systemctl restart ci-web.service' so the HTTP response returns first.
Route POST /services/{service_id}/{action} in web/routes.py mirrors vms_remove: require_user + require_csrf + two-step confirm gate + one audit entry per stage; unknown id / bad action / disallowed-stop short-circuit to 400 before any subprocess or audit; the vmware-host confirm step warns with the active-VM count. New /services read partial with state-aware Start/Stop/Restart buttons, polled every 15s.
deploy/sudoers.d/ci-web grants ci-runner exact-command NOPASSWD for exactly these nine systemctl lines plus the systemd-run deferral — no wildcards, no bare systemctl, no ci-web stop; setup-web-linux.sh installs it mode 0440 and validates with 'visudo -cf', rolling back on failure. Config gains [web].managed_services override; docs/WEB-UI.md documents the panel + security model.
Gates: 1044 tests pass (46 new security/unit tests), coverage 94.20%, ruff + mypy --strict clean. Not yet deployed: sudoers install + prod-venv reinstall + ci-web restart are a human privilege-boundary step.
370 lines
13 KiB
Python
370 lines
13 KiB
Python
"""Tests for the pure ``operations.services`` layer (Stage D of the web-UI
|
|
redesign) -- SECURITY-CRITICAL: this module builds the exact argv that ends
|
|
up behind ``sudo`` on the live host.
|
|
|
|
No real ``systemctl``/``sudo``/``systemd-run`` subprocess is ever invoked --
|
|
every ``subprocess.run`` call is monkeypatched at the name this module
|
|
imported it under (mirrors ``test_operations_host.py``'s convention).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
import ci_orchestrator.operations.host as host_module
|
|
import ci_orchestrator.operations.services as services_module
|
|
from ci_orchestrator.operations.services import (
|
|
ALLOWED_ACTIONS,
|
|
MANAGED_SERVICES,
|
|
ManagedService,
|
|
ServiceState,
|
|
control_service,
|
|
find_service,
|
|
service_states,
|
|
)
|
|
|
|
|
|
class _RunResult:
|
|
def __init__(self, stdout: str = "", stderr: str = "", returncode: int = 0) -> None:
|
|
self.stdout = stdout
|
|
self.stderr = stderr
|
|
self.returncode = returncode
|
|
|
|
|
|
def _spy(real: Any) -> Any:
|
|
"""Wrap ``real`` so ``.calls`` records every ``(args, kwargs)`` call, still delegating."""
|
|
|
|
def wrapper(*args: Any, **kwargs: Any) -> Any:
|
|
wrapper.calls.append((args, kwargs)) # type: ignore[attr-defined]
|
|
return real(*args, **kwargs)
|
|
|
|
wrapper.calls = [] # type: ignore[attr-defined]
|
|
return wrapper
|
|
|
|
|
|
# ── MANAGED_SERVICES / find_service ─────────────────────────────────────────
|
|
|
|
|
|
def test_managed_services_shape() -> None:
|
|
ids = {s.id for s in MANAGED_SERVICES}
|
|
assert ids == {"act-runner", "act-runner-rust", "vmware-host", "ci-web"}
|
|
by_id = {s.id: s for s in MANAGED_SERVICES}
|
|
assert by_id["ci-web"].allow_stop is False
|
|
assert by_id["ci-web"].unit == "ci-web.service"
|
|
assert by_id["act-runner"].allow_stop is True
|
|
assert by_id["vmware-host"].allow_stop is True
|
|
|
|
|
|
def test_find_service_returns_allowlist_member() -> None:
|
|
svc = find_service("act-runner")
|
|
assert svc is not None
|
|
assert svc.unit == "act-runner.service"
|
|
|
|
|
|
def test_find_service_unknown_id_returns_none() -> None:
|
|
assert find_service("does-not-exist") is None
|
|
|
|
|
|
def test_find_service_uses_custom_services_tuple() -> None:
|
|
custom = (ManagedService("x", "x.service", "X", True),)
|
|
assert find_service("x", custom) is not None
|
|
# Not present in the default list, only in the custom one.
|
|
assert find_service("x", MANAGED_SERVICES) is None
|
|
|
|
|
|
# ── service_states ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_service_states_reflects_active_and_inactive(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(host_module.shutil, "which", lambda _n: "/usr/bin/systemctl")
|
|
|
|
def _run(cmd: list[str], **_kw: Any) -> _RunResult:
|
|
unit = cmd[-1]
|
|
if cmd[1] == "is-active":
|
|
return _RunResult(stdout="active\n" if unit == "act-runner.service" else "inactive\n")
|
|
return _RunResult(stdout="running\n" if unit == "act-runner.service" else "dead\n")
|
|
|
|
monkeypatch.setattr(host_module.subprocess, "run", _run)
|
|
|
|
services = (
|
|
ManagedService("act-runner", "act-runner.service", "Runner", True),
|
|
ManagedService("ci-web", "ci-web.service", "Web UI", False),
|
|
)
|
|
states = service_states(services)
|
|
|
|
assert states == [
|
|
ServiceState(service=services[0], active=True, sub_state="running"),
|
|
ServiceState(service=services[1], active=False, sub_state="dead"),
|
|
]
|
|
|
|
|
|
def test_service_states_degrades_to_unknown_when_systemctl_missing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(host_module.shutil, "which", lambda _n: None)
|
|
|
|
states = service_states(MANAGED_SERVICES)
|
|
|
|
assert len(states) == len(MANAGED_SERVICES)
|
|
assert all(not s.active and s.sub_state == "unknown" for s in states)
|
|
|
|
|
|
def test_service_states_degrades_to_unknown_on_subprocess_timeout(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(host_module.shutil, "which", lambda _n: "/usr/bin/systemctl")
|
|
|
|
def _boom(*_a: Any, **_kw: Any) -> Any:
|
|
raise subprocess.TimeoutExpired(cmd="systemctl", timeout=10.0)
|
|
|
|
monkeypatch.setattr(host_module.subprocess, "run", _boom)
|
|
|
|
states = service_states(MANAGED_SERVICES)
|
|
|
|
assert all(not s.active and s.sub_state == "unknown" for s in states)
|
|
|
|
|
|
def test_service_states_never_uses_shell_true(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(host_module.shutil, "which", lambda _n: "/usr/bin/systemctl")
|
|
seen_kwargs: list[dict[str, Any]] = []
|
|
|
|
def _run(cmd: list[str], **kw: Any) -> _RunResult:
|
|
seen_kwargs.append(kw)
|
|
return _RunResult(stdout="active\n")
|
|
|
|
monkeypatch.setattr(host_module.subprocess, "run", _run)
|
|
service_states(MANAGED_SERVICES)
|
|
|
|
assert seen_kwargs # sanity: the fake was actually invoked
|
|
assert all(not kw.get("shell") for kw in seen_kwargs)
|
|
|
|
|
|
# ── control_service: validation ──────────────────────────────────────────────
|
|
|
|
|
|
def test_control_service_unsupported_action_raises_value_error(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
spy = _spy(subprocess.run)
|
|
monkeypatch.setattr(services_module.subprocess, "run", spy)
|
|
service = find_service("act-runner")
|
|
assert service is not None
|
|
|
|
with pytest.raises(ValueError):
|
|
control_service(service, "reboot")
|
|
|
|
assert spy.calls == [] # type: ignore[attr-defined]
|
|
|
|
|
|
def test_control_service_stop_refused_for_disallowed_service_builds_no_command(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
spy = _spy(subprocess.run)
|
|
monkeypatch.setattr(services_module.subprocess, "run", spy)
|
|
ci_web = find_service("ci-web")
|
|
assert ci_web is not None
|
|
|
|
result = control_service(ci_web, "stop")
|
|
|
|
assert result.service_id == "ci-web"
|
|
assert result.action == "stop"
|
|
assert result.ok is False
|
|
assert result.command == []
|
|
assert "restart-only" in result.detail or "disabled" in result.detail
|
|
assert spy.calls == [] # type: ignore[attr-defined]
|
|
|
|
|
|
def test_control_service_stop_refused_even_in_dry_run(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""The stop-refusal check must win over dry_run -- no command is built either way."""
|
|
spy = _spy(subprocess.run)
|
|
monkeypatch.setattr(services_module.subprocess, "run", spy)
|
|
ci_web = find_service("ci-web")
|
|
assert ci_web is not None
|
|
|
|
result = control_service(ci_web, "stop", dry_run=True)
|
|
|
|
assert result.ok is False
|
|
assert result.command == []
|
|
assert spy.calls == [] # type: ignore[attr-defined]
|
|
|
|
|
|
# ── control_service: dry_run argv, incl. the ci-web restart special case ───
|
|
|
|
|
|
@pytest.mark.parametrize("action", ALLOWED_ACTIONS)
|
|
def test_control_service_dry_run_argv_for_every_normal_service(action: str) -> None:
|
|
for service in MANAGED_SERVICES:
|
|
if service.id == "ci-web":
|
|
continue # covered separately below (special-cased argv)
|
|
if action == "stop" and not service.allow_stop:
|
|
continue # covered by the refusal tests above
|
|
result = control_service(service, action, dry_run=True)
|
|
|
|
assert result.ok is True
|
|
assert result.command == ["sudo", "-n", "systemctl", action, service.unit]
|
|
assert result.detail.startswith("would run:")
|
|
|
|
|
|
def test_control_service_dry_run_ci_web_start_and_stop_use_plain_systemctl() -> None:
|
|
ci_web = find_service("ci-web")
|
|
assert ci_web is not None
|
|
|
|
start_result = control_service(ci_web, "start", dry_run=True)
|
|
assert start_result.command == ["sudo", "-n", "systemctl", "start", "ci-web.service"]
|
|
|
|
|
|
def test_control_service_dry_run_ci_web_restart_uses_systemd_run_deferred() -> None:
|
|
ci_web = find_service("ci-web")
|
|
assert ci_web is not None
|
|
|
|
result = control_service(ci_web, "restart", dry_run=True)
|
|
|
|
assert result.ok is True
|
|
assert result.command == [
|
|
"sudo",
|
|
"-n",
|
|
"systemd-run",
|
|
"--on-active=2",
|
|
"systemctl",
|
|
"restart",
|
|
"ci-web.service",
|
|
]
|
|
assert "systemd-run" in result.detail
|
|
assert "--on-active=2" in result.detail
|
|
|
|
|
|
def test_control_service_dry_run_never_executes_anything(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
spy = _spy(subprocess.run)
|
|
monkeypatch.setattr(services_module.subprocess, "run", spy)
|
|
service = find_service("act-runner")
|
|
assert service is not None
|
|
|
|
control_service(service, "restart", dry_run=True)
|
|
|
|
assert spy.calls == [] # type: ignore[attr-defined]
|
|
|
|
|
|
# ── control_service: real execution ──────────────────────────────────────────
|
|
|
|
|
|
def test_control_service_real_run_success_builds_arg_list(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
seen: list[tuple[Any, ...]] = []
|
|
|
|
def _run(cmd: list[str], **kw: Any) -> _RunResult:
|
|
seen.append(tuple(cmd))
|
|
assert not kw.get("shell")
|
|
return _RunResult(returncode=0)
|
|
|
|
monkeypatch.setattr(services_module.subprocess, "run", _run)
|
|
service = find_service("act-runner")
|
|
assert service is not None
|
|
|
|
result = control_service(service, "restart", dry_run=False)
|
|
|
|
assert result.ok is True
|
|
assert seen == [("sudo", "-n", "systemctl", "restart", "act-runner.service")]
|
|
assert seen[0][0:3] == ("sudo", "-n", "systemctl")
|
|
assert result.command == ["sudo", "-n", "systemctl", "restart", "act-runner.service"]
|
|
|
|
|
|
def test_control_service_ci_web_restart_real_run_uses_systemd_run(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
seen: list[list[str]] = []
|
|
|
|
def _run(cmd: list[str], **_kw: Any) -> _RunResult:
|
|
seen.append(cmd)
|
|
return _RunResult(returncode=0)
|
|
|
|
monkeypatch.setattr(services_module.subprocess, "run", _run)
|
|
ci_web = find_service("ci-web")
|
|
assert ci_web is not None
|
|
|
|
result = control_service(ci_web, "restart", dry_run=False)
|
|
|
|
assert result.ok is True
|
|
assert "systemd-run" in seen[0]
|
|
assert "--on-active=2" in seen[0]
|
|
assert "scheduled" in result.detail
|
|
|
|
|
|
def test_control_service_real_run_nonzero_exit_is_not_ok(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def _run(cmd: list[str], **_kw: Any) -> _RunResult:
|
|
return _RunResult(returncode=1, stderr="Unit act-runner.service not found.\n")
|
|
|
|
monkeypatch.setattr(services_module.subprocess, "run", _run)
|
|
service = find_service("act-runner")
|
|
assert service is not None
|
|
|
|
result = control_service(service, "start", dry_run=False)
|
|
|
|
assert result.ok is False
|
|
assert "not found" in result.detail
|
|
assert "1" in result.detail
|
|
|
|
|
|
def test_control_service_real_run_nonzero_exit_no_stderr_still_reports_exit_code(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
def _run(cmd: list[str], **_kw: Any) -> _RunResult:
|
|
return _RunResult(returncode=1, stderr="")
|
|
|
|
monkeypatch.setattr(services_module.subprocess, "run", _run)
|
|
service = find_service("act-runner")
|
|
assert service is not None
|
|
|
|
result = control_service(service, "start", dry_run=False)
|
|
|
|
assert result.ok is False
|
|
assert "exit 1" in result.detail
|
|
|
|
|
|
def test_control_service_timeout_degrades_to_error_never_raises(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
def _boom(*_a: Any, **_kw: Any) -> Any:
|
|
raise subprocess.TimeoutExpired(cmd="sudo", timeout=15.0)
|
|
|
|
monkeypatch.setattr(services_module.subprocess, "run", _boom)
|
|
service = find_service("act-runner")
|
|
assert service is not None
|
|
|
|
result = control_service(service, "restart", dry_run=False)
|
|
|
|
assert result.ok is False
|
|
assert "sudo" in result.detail or "restart" in result.detail.lower()
|
|
|
|
|
|
def test_control_service_oserror_degrades_to_error_never_raises(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
def _boom(*_a: Any, **_kw: Any) -> Any:
|
|
raise OSError("sudo: command not found")
|
|
|
|
monkeypatch.setattr(services_module.subprocess, "run", _boom)
|
|
service = find_service("act-runner")
|
|
assert service is not None
|
|
|
|
result = control_service(service, "restart", dry_run=False)
|
|
|
|
assert result.ok is False
|
|
assert "sudo: command not found" in result.detail
|
|
|
|
|
|
def test_control_service_stderr_tail_is_truncated(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
huge_stderr = "x" * 5000
|
|
|
|
def _run(cmd: list[str], **_kw: Any) -> _RunResult:
|
|
return _RunResult(returncode=1, stderr=huge_stderr)
|
|
|
|
monkeypatch.setattr(services_module.subprocess, "run", _run)
|
|
service = find_service("act-runner")
|
|
assert service is not None
|
|
|
|
result = control_service(service, "start", dry_run=False)
|
|
|
|
assert len(result.detail) < len(huge_stderr) + 100
|