Simone
af7dee60e5
The dashboard could only show what was powered on this instant, so it could
not answer "did the last build pass?" — and no local execution history exists
to answer it from (`report job` reads a PowerShell-era JSONL path that nothing
writes on this host, and the recent artifact dirs are empty).
Recent runs now come from Gitea's Actions API (`operations/executions.py`):
`/api/v1/admin/actions/runs` for every repo on the instance, falling back to
`/api/v1/user/actions/runs` when the token is not admin-scoped. One list call
per refresh, cached 10s, so the card's 5s htmx poll costs Gitea at most one
call per window. Live VMs are grouped under the run that spawned them — the
job id is `{run_id}-{run_attempt}[-suffix]` and the clone name carries it — so
a matrix run collects several VMs on one row. Every failure mode (no token,
401, unreachable Gitea, malformed payload) degrades to a notice plus the plain
VM rows, each keeping its Remove button, so the card is never less useful than
the table it replaces. Orphans stay where they were: the sweep in Maintenance,
the count in the status-header tile.
Auth reuses the access token from the operator's own OIDC login, which
`auth.py` previously discarded. SECURITY: that token is kept server-side only
(`app.state.gitea_tokens`, dropped on logout) and never written into the
session — `SessionMiddleware` signs the session cookie but does not encrypt
it, and this service is deployed over plain HTTP on the LAN, so a bearer token
in there would be readable in the browser and on the wire. An expired token is
refreshed once; `[web] gitea_token_cred` names an optional keyring PAT for the
cases a login token cannot cover (service restart, non-admin operator).
`GET /vms` and `partials/vms.html` are gone; `POST /vms/{id}/remove` and
`POST /vms/cleanup` are untouched.
Design: docs/superpowers/specs/2026-07-25-executions-card-design.md
958 lines
33 KiB
Python
958 lines
33 KiB
Python
"""Tests for the W6 mutating web routes (VM remove/cleanup, cache purge, host
|
|
backup, audit) -- SECURITY-CRITICAL: these routes destroy VMs, purge caches,
|
|
and back up templates.
|
|
|
|
No real ``vmrun``/``7z``/``systemctl``/disk writes outside ``tmp_path``: VM
|
|
backends are fakes/mocks, and ``operations/`` functions are spied on (not
|
|
replaced) so real side effects still land only inside ``tmp_path``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import ci_orchestrator.operations.host as host_module
|
|
import ci_orchestrator.web.routes as routes_module
|
|
from ci_orchestrator.backends.errors import BackendError
|
|
from ci_orchestrator.backends.protocol import VmHandle
|
|
from ci_orchestrator.config import (
|
|
BackendConfig,
|
|
CargoCacheConfig,
|
|
Config,
|
|
Paths,
|
|
WebConfig,
|
|
WebOidcConfig,
|
|
)
|
|
from ci_orchestrator.credentials import Credential, KeyringCredentialStore
|
|
from ci_orchestrator.operations.audit import AuditEntry, append_audit, read_audit
|
|
from ci_orchestrator.operations.vms import RemovalResult
|
|
from ci_orchestrator.web.app import create_app, get_backend
|
|
|
|
_SESSION_CRED_NAME = "TestWebSessionSecret"
|
|
_OIDC_CRED_NAME = "TestWebOidcClientSecret"
|
|
_FAKE_SESSION_SECRET = "sess-secret-do-not-leak-abc123"
|
|
_FAKE_OIDC_CLIENT_SECRET = "oidc-client-secret-do-not-leak-xyz987"
|
|
|
|
|
|
# ── shared fixtures / fakes ─────────────────────────────────────────────────
|
|
|
|
|
|
def _install_fake_keyring(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
secrets = {
|
|
_SESSION_CRED_NAME: _FAKE_SESSION_SECRET,
|
|
_OIDC_CRED_NAME: _FAKE_OIDC_CLIENT_SECRET,
|
|
}
|
|
|
|
def fake_get(self: KeyringCredentialStore, target: str) -> Credential:
|
|
if target not in secrets:
|
|
raise KeyError(target)
|
|
return Credential(username=target, password=secrets[target])
|
|
|
|
monkeypatch.setattr(KeyringCredentialStore, "get", fake_get)
|
|
|
|
|
|
def _make_config(root: Path, *, active: bool = False, allowed_users: list[str] | None = None) -> Config:
|
|
oidc = None
|
|
if active:
|
|
oidc = WebOidcConfig(
|
|
issuer="https://gitea.example.lan",
|
|
client_id="ci-web",
|
|
redirect_url="http://127.0.0.1:8800/auth/callback",
|
|
client_secret_cred=_OIDC_CRED_NAME,
|
|
scopes=["openid", "profile"],
|
|
)
|
|
return Config(
|
|
paths=Paths(
|
|
root=root,
|
|
templates=root / "templates",
|
|
build_vms=root / "build-vms",
|
|
artifacts=root / "artifacts",
|
|
keys=root / "keys",
|
|
),
|
|
backend=BackendConfig(),
|
|
cargo_cache=CargoCacheConfig(
|
|
enabled=True, root=root / "cache" / "cargo", retention_days=30
|
|
),
|
|
web=WebConfig(
|
|
enabled=active,
|
|
allowed_users=allowed_users if allowed_users is not None else [],
|
|
session_secret_cred=_SESSION_CRED_NAME,
|
|
audit_log=root / "web-audit.jsonl",
|
|
oidc=oidc,
|
|
),
|
|
)
|
|
|
|
|
|
class _FakeBackend:
|
|
"""Stub ``VmBackend`` -- proves no real ``vmrun`` is ever invoked."""
|
|
|
|
def __init__(self, handles: list[VmHandle], ip: str | None = "192.168.79.201") -> None:
|
|
self._handles = handles
|
|
self._ip = ip
|
|
self.stop_calls: list[VmHandle] = []
|
|
self.delete_calls: list[VmHandle] = []
|
|
|
|
def list_running(self) -> list[VmHandle]:
|
|
return self._handles
|
|
|
|
def get_ip(self, handle: VmHandle, timeout: float = 0.0) -> str | None:
|
|
return self._ip
|
|
|
|
def stop(self, handle: VmHandle, hard: bool = False) -> None:
|
|
self.stop_calls.append(handle)
|
|
|
|
def delete(self, handle: VmHandle) -> None:
|
|
self.delete_calls.append(handle)
|
|
|
|
|
|
def _make_vmx(root: Path, name: str = "job1") -> Path:
|
|
vm_dir = root / "build-vms" / name
|
|
vm_dir.mkdir(parents=True, exist_ok=True)
|
|
vmx = vm_dir / f"{name}.vmx"
|
|
vmx.write_text('guestOS = "ubuntu-64"\n', encoding="utf-8")
|
|
return vmx
|
|
|
|
|
|
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
|
|
|
|
|
|
def _csrf_from(text: str) -> str:
|
|
match = re.search(r'name="csrf" value="([^"]+)"', text)
|
|
assert match is not None, f"no csrf field found in: {text[:300]!r}"
|
|
return match.group(1)
|
|
|
|
|
|
def _get_csrf(client: TestClient, route: str = "/") -> str:
|
|
resp = client.get(route)
|
|
assert resp.status_code == 200
|
|
return _csrf_from(resp.text)
|
|
|
|
|
|
def _last_audit_entry(audit_log: Path) -> AuditEntry:
|
|
entries = read_audit(audit_log, last=100)
|
|
assert entries, "expected at least one audit entry"
|
|
return entries[-1]
|
|
|
|
|
|
# ── 1. CSRF enforcement on every POST ───────────────────────────────────────
|
|
|
|
|
|
def test_post_without_csrf_field_is_403(tmp_path: Path) -> None:
|
|
app = create_app(_make_config(tmp_path))
|
|
client = TestClient(app)
|
|
client.get("/cache") # establish a session
|
|
|
|
resp = client.post("/cache/purge", data={})
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_post_with_wrong_csrf_is_403(tmp_path: Path) -> None:
|
|
app = create_app(_make_config(tmp_path))
|
|
client = TestClient(app)
|
|
client.get("/cache") # establish a session with a real token
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": "totally-not-the-real-token"})
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_wrong_csrf_never_runs_the_operation(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from ci_orchestrator.operations.cache import purge_cargo_cache as real_purge
|
|
|
|
spy = _spy(real_purge)
|
|
monkeypatch.setattr(routes_module, "purge_cargo_cache", spy)
|
|
app = create_app(_make_config(tmp_path))
|
|
client = TestClient(app)
|
|
client.get("/cache")
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": "bogus", "stage": "execute"})
|
|
|
|
assert resp.status_code == 403
|
|
assert spy.calls == [] # type: ignore[attr-defined]
|
|
|
|
|
|
# ── 2. two-step confirm gate: preview always runs dry_run=True ─────────────
|
|
|
|
|
|
def test_cache_purge_preview_returns_confirm_partial_and_runs_dry_run(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from ci_orchestrator.operations.cache import purge_cargo_cache as real_purge
|
|
|
|
spy = _spy(real_purge)
|
|
monkeypatch.setattr(routes_module, "purge_cargo_cache", spy)
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": csrf})
|
|
|
|
assert resp.status_code == 200
|
|
assert "Confirm" in resp.text
|
|
# Age-based purge confirm spells out the max-age dependency (#4).
|
|
assert "older than 30 day(s)" in resp.text
|
|
assert len(spy.calls) == 1 # type: ignore[attr-defined]
|
|
assert spy.calls[0][1]["dry_run"] is True # type: ignore[attr-defined]
|
|
assert spy.calls[0][1]["force"] is False # type: ignore[attr-defined]
|
|
|
|
entry = _last_audit_entry(config.web.audit_log)
|
|
assert entry.result == "dry-run"
|
|
assert entry.action == "cache.purge"
|
|
|
|
|
|
def test_cache_purge_preview_does_not_delete_anything(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
cache_root = tmp_path / "cache" / "cargo"
|
|
cache_root.mkdir(parents=True)
|
|
entry_path = cache_root / "v1-deadbeef.tar.zst"
|
|
entry_path.write_bytes(b"x" * 10)
|
|
# Backdate it so it would be a purge candidate on a real run.
|
|
old = datetime.now(tz=UTC) - timedelta(days=999)
|
|
import os
|
|
|
|
os.utime(entry_path, (old.timestamp(), old.timestamp()))
|
|
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": csrf})
|
|
|
|
assert resp.status_code == 200
|
|
assert entry_path.exists()
|
|
|
|
|
|
# ── 3. execute stage: real dry_run=False, audit "ok", result partial ───────
|
|
|
|
|
|
def test_cache_purge_execute_runs_dry_run_false_and_audits_ok(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from ci_orchestrator.operations.cache import purge_cargo_cache as real_purge
|
|
|
|
spy = _spy(real_purge)
|
|
monkeypatch.setattr(routes_module, "purge_cargo_cache", spy)
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 200
|
|
assert len(spy.calls) == 1 # type: ignore[attr-defined]
|
|
assert spy.calls[0][1]["dry_run"] is False # type: ignore[attr-defined]
|
|
|
|
entry = _last_audit_entry(config.web.audit_log)
|
|
assert entry.result == "ok"
|
|
assert entry.action == "cache.purge"
|
|
|
|
|
|
def test_cache_purge_execute_actually_removes_expired_entries(tmp_path: Path) -> None:
|
|
cache_root = tmp_path / "cache" / "cargo"
|
|
cache_root.mkdir(parents=True)
|
|
entry_path = cache_root / "v1-deadbeef.tar.zst"
|
|
entry_path.write_bytes(b"x" * 10)
|
|
old = datetime.now(tz=UTC) - timedelta(days=999)
|
|
import os
|
|
|
|
os.utime(entry_path, (old.timestamp(), old.timestamp()))
|
|
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 200
|
|
assert not entry_path.exists()
|
|
|
|
|
|
def test_cache_purge_execute_nothing_removed_is_info_not_silent(tmp_path: Path) -> None:
|
|
# A fresh entry within retention -> age-based purge removes nothing. The
|
|
# result must say *why* (neutral "info", not a green "ok" that reads as a
|
|
# lie) and still ask panels to refresh via HX-Trigger (#3/#4).
|
|
cache_root = tmp_path / "cache" / "cargo"
|
|
cache_root.mkdir(parents=True)
|
|
(cache_root / "v1-fresh.tar.zst").write_bytes(b"x" * 10) # mtime = now
|
|
|
|
config = _make_config(tmp_path)
|
|
client = TestClient(create_app(config))
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 200
|
|
assert "Nothing removed" in resp.text
|
|
assert "max age" in resp.text
|
|
assert "result-info" in resp.text # neutral badge, not result-ok
|
|
assert resp.headers["HX-Trigger"] == "ci:refresh"
|
|
|
|
|
|
def test_cache_purge_max_age_override_from_form(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from ci_orchestrator.operations.cache import purge_cargo_cache as real_purge
|
|
|
|
spy = _spy(real_purge)
|
|
monkeypatch.setattr(routes_module, "purge_cargo_cache", spy)
|
|
config = _make_config(tmp_path)
|
|
client = TestClient(create_app(config))
|
|
csrf = _get_csrf(client)
|
|
|
|
# An explicit max age overrides the configured retention (30) for the
|
|
# age-based purge, and the confirm spells the chosen age out.
|
|
resp = client.post("/cache/purge", data={"csrf": csrf, "max_age_days": "7"})
|
|
|
|
assert resp.status_code == 200
|
|
assert "older than 7 day(s)" in resp.text
|
|
assert spy.calls[0][1]["retention_days"] == 7 # type: ignore[attr-defined]
|
|
assert spy.calls[0][1]["force"] is False # type: ignore[attr-defined]
|
|
|
|
|
|
def test_cache_empty_all_preview_forces_and_confirms(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from ci_orchestrator.operations.cache import purge_cargo_cache as real_purge
|
|
|
|
spy = _spy(real_purge)
|
|
monkeypatch.setattr(routes_module, "purge_cargo_cache", spy)
|
|
config = _make_config(tmp_path)
|
|
client = TestClient(create_app(config))
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": csrf, "mode": "all"})
|
|
|
|
assert resp.status_code == 200
|
|
assert "Confirm" in resp.text
|
|
assert "Empty cargo cache" in resp.text
|
|
assert "entire cache" in resp.text
|
|
# The confirm form must carry mode=all through to the execute POST.
|
|
assert 'value="all"' in resp.text
|
|
assert spy.calls[0][1]["force"] is True # type: ignore[attr-defined]
|
|
assert spy.calls[0][1]["dry_run"] is True # type: ignore[attr-defined]
|
|
|
|
|
|
def test_cache_empty_all_execute_removes_regardless_of_age(tmp_path: Path) -> None:
|
|
cache_root = tmp_path / "cache" / "cargo"
|
|
cache_root.mkdir(parents=True)
|
|
fresh = cache_root / "v1-fresh.tar.zst"
|
|
fresh.write_bytes(b"x" * 10) # brand-new, well within the 30-day retention
|
|
|
|
config = _make_config(tmp_path)
|
|
client = TestClient(create_app(config))
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": csrf, "mode": "all", "stage": "execute"})
|
|
|
|
assert resp.status_code == 200
|
|
assert not fresh.exists()
|
|
assert "Freed" in resp.text
|
|
assert resp.headers["HX-Trigger"] == "ci:refresh"
|
|
|
|
|
|
def test_vm_remove_preview_then_execute_full_cycle(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from ci_orchestrator.operations.vms import remove_vm as real_remove_vm
|
|
|
|
spy = _spy(real_remove_vm)
|
|
monkeypatch.setattr(routes_module, "remove_vm", spy)
|
|
|
|
vmx = _make_vmx(tmp_path)
|
|
config = _make_config(tmp_path)
|
|
fake_backend = _FakeBackend([VmHandle(identifier=str(vmx), name="job1")])
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: fake_backend
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
assert resp.status_code == 200
|
|
csrf = _csrf_from(resp.text)
|
|
vmx_id = routes_module.encode_vmx_id(vmx)
|
|
|
|
preview = client.post(f"/vms/{vmx_id}/remove", data={"csrf": csrf})
|
|
assert preview.status_code == 200
|
|
assert "Confirm" in preview.text
|
|
assert "would remove" in preview.text
|
|
assert spy.calls[-1][1]["dry_run"] is True # type: ignore[attr-defined]
|
|
assert fake_backend.stop_calls == []
|
|
assert fake_backend.delete_calls == []
|
|
assert _last_audit_entry(config.web.audit_log).result == "dry-run"
|
|
|
|
execute = client.post(f"/vms/{vmx_id}/remove", data={"csrf": csrf, "stage": "execute"})
|
|
assert execute.status_code == 200
|
|
assert spy.calls[-1][1]["dry_run"] is False # type: ignore[attr-defined]
|
|
assert len(fake_backend.stop_calls) == 1
|
|
assert len(fake_backend.delete_calls) == 1
|
|
assert not vmx.parent.exists()
|
|
|
|
entry = _last_audit_entry(config.web.audit_log)
|
|
assert entry.result == "ok"
|
|
assert entry.action == "vm.remove"
|
|
|
|
|
|
# ── 4. path containment ─────────────────────────────────────────────────────
|
|
|
|
|
|
def test_vm_remove_outside_build_vms_is_400_and_does_not_delete(tmp_path: Path) -> None:
|
|
outside_dir = tmp_path / "not-build-vms" / "evil"
|
|
outside_dir.mkdir(parents=True)
|
|
outside_vmx = outside_dir / "evil.vmx"
|
|
outside_vmx.write_text("keep me", encoding="utf-8")
|
|
|
|
config = _make_config(tmp_path)
|
|
fake_backend = _FakeBackend([])
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: fake_backend
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
vmx_id = routes_module.encode_vmx_id(outside_vmx)
|
|
|
|
resp = client.post(f"/vms/{vmx_id}/remove", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 400
|
|
assert outside_vmx.exists()
|
|
assert fake_backend.stop_calls == []
|
|
assert fake_backend.delete_calls == []
|
|
|
|
entries = read_audit(config.web.audit_log, last=100)
|
|
assert not any(e.result == "ok" for e in entries)
|
|
|
|
|
|
def test_vm_remove_vmx_directly_in_build_vms_is_400_not_wiping_base(tmp_path: Path) -> None:
|
|
"""A vmx whose parent IS build_vms would rmtree the whole clone base -- reject.
|
|
|
|
remove_vm() deletes vmx.parent; for ``build_vms/x.vmx`` that parent is
|
|
build_vms itself. The containment check must reject it so a crafted
|
|
vmx_id can't wipe every clone at once.
|
|
"""
|
|
config = _make_config(tmp_path)
|
|
build_vms = config.paths.build_vms
|
|
build_vms.mkdir(parents=True, exist_ok=True)
|
|
survivor = build_vms / "Clone_keep"
|
|
survivor.mkdir()
|
|
(survivor / "Clone_keep.vmx").write_text("keep", encoding="utf-8")
|
|
|
|
danger_vmx = build_vms / "x.vmx" # parent == build_vms itself
|
|
danger_vmx.write_text("x", encoding="utf-8")
|
|
|
|
fake_backend = _FakeBackend([])
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: fake_backend
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
vmx_id = routes_module.encode_vmx_id(danger_vmx)
|
|
|
|
resp = client.post(f"/vms/{vmx_id}/remove", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 400
|
|
assert build_vms.exists() # clone base NOT wiped
|
|
assert survivor.exists() # sibling clone survived
|
|
assert fake_backend.delete_calls == []
|
|
|
|
|
|
def test_vm_remove_malformed_vmx_id_is_400(tmp_path: Path) -> None:
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/vms/not-valid-base64!!!/remove", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# ── 5. unauthenticated POST -> redirect to /login, no execution ────────────
|
|
|
|
|
|
def test_unauthenticated_post_redirects_to_login_and_never_runs(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_install_fake_keyring(monkeypatch)
|
|
from ci_orchestrator.operations.cache import purge_cargo_cache as real_purge
|
|
|
|
spy = _spy(real_purge)
|
|
monkeypatch.setattr(routes_module, "purge_cargo_cache", spy)
|
|
config = _make_config(tmp_path, active=True, allowed_users=["simone"])
|
|
app = create_app(config)
|
|
client = TestClient(app, follow_redirects=False)
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": "whatever", "stage": "execute"})
|
|
|
|
assert resp.status_code in (302, 303)
|
|
assert "/login" in resp.headers["location"]
|
|
assert spy.calls == [] # type: ignore[attr-defined]
|
|
assert not config.web.audit_log.exists()
|
|
|
|
|
|
# ── 6. job-race guard ───────────────────────────────────────────────────────
|
|
|
|
|
|
def _write_fresh_lock(config: Config) -> Path:
|
|
lock_path = config.paths.build_vms.parent / "vm-start.lock"
|
|
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
|
lock_path.write_text("job in progress", encoding="utf-8")
|
|
return lock_path
|
|
|
|
|
|
def test_vm_cleanup_execute_refused_409_when_lock_is_fresh(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from ci_orchestrator.operations.vms import cleanup_orphans as real_cleanup
|
|
|
|
spy = _spy(real_cleanup)
|
|
monkeypatch.setattr(routes_module, "cleanup_orphans", spy)
|
|
config = _make_config(tmp_path)
|
|
config.paths.build_vms.mkdir(parents=True, exist_ok=True)
|
|
_write_fresh_lock(config)
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/vms/cleanup", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 409
|
|
assert "job may be running" in resp.text.lower()
|
|
assert spy.calls == [] # type: ignore[attr-defined]
|
|
|
|
|
|
def test_vm_remove_execute_refused_409_when_lock_is_fresh(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from ci_orchestrator.operations.vms import remove_vm as real_remove_vm
|
|
|
|
spy = _spy(real_remove_vm)
|
|
monkeypatch.setattr(routes_module, "remove_vm", spy)
|
|
vmx = _make_vmx(tmp_path)
|
|
config = _make_config(tmp_path)
|
|
_write_fresh_lock(config)
|
|
fake_backend = _FakeBackend([VmHandle(identifier=str(vmx), name="job1")])
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: fake_backend
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
vmx_id = routes_module.encode_vmx_id(vmx)
|
|
|
|
resp = client.post(f"/vms/{vmx_id}/remove", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 409
|
|
assert spy.calls == [] # type: ignore[attr-defined]
|
|
assert fake_backend.stop_calls == []
|
|
assert vmx.exists()
|
|
|
|
|
|
def test_vm_cleanup_preview_ignores_fresh_lock(tmp_path: Path) -> None:
|
|
"""A preview (dry-run) is always safe -- the lock guard only gates execute."""
|
|
config = _make_config(tmp_path)
|
|
config.paths.build_vms.mkdir(parents=True, exist_ok=True)
|
|
_write_fresh_lock(config)
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/vms/cleanup", data={"csrf": csrf})
|
|
|
|
assert resp.status_code == 200
|
|
assert "Confirm" in resp.text
|
|
|
|
|
|
def test_vm_cleanup_execute_proceeds_when_lock_is_stale(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from ci_orchestrator.operations.vms import cleanup_orphans as real_cleanup
|
|
|
|
spy = _spy(real_cleanup)
|
|
monkeypatch.setattr(routes_module, "cleanup_orphans", spy)
|
|
config = _make_config(tmp_path)
|
|
config.paths.build_vms.mkdir(parents=True, exist_ok=True)
|
|
lock_path = _write_fresh_lock(config)
|
|
stale = datetime.now(tz=UTC) - timedelta(minutes=90)
|
|
import os
|
|
|
|
os.utime(lock_path, (stale.timestamp(), stale.timestamp()))
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/vms/cleanup", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 200
|
|
assert len(spy.calls) == 1 # type: ignore[attr-defined]
|
|
assert _last_audit_entry(config.web.audit_log).result == "ok"
|
|
|
|
|
|
# ── 7. GET /audit renders written entries ───────────────────────────────────
|
|
|
|
|
|
def test_audit_view_renders_written_entries(tmp_path: Path) -> None:
|
|
config = _make_config(tmp_path)
|
|
append_audit(
|
|
config.web.audit_log,
|
|
AuditEntry(
|
|
ts="2026-07-23T10:00:00+00:00",
|
|
user="anonymous",
|
|
action="cache.purge",
|
|
target=str(config.cargo_cache.root),
|
|
result="ok",
|
|
detail="freed 123 bytes across 4 entrie(s)",
|
|
),
|
|
)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/audit")
|
|
|
|
assert resp.status_code == 200
|
|
assert "Audit log" in resp.text
|
|
assert "cache.purge" in resp.text
|
|
assert "freed 123 bytes across 4 entrie(s)" in resp.text
|
|
|
|
|
|
def test_audit_view_empty_log_still_200(tmp_path: Path) -> None:
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/audit")
|
|
|
|
assert resp.status_code == 200
|
|
assert "Audit log" in resp.text
|
|
|
|
|
|
def test_audit_view_requires_auth_when_active(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_install_fake_keyring(monkeypatch)
|
|
config = _make_config(tmp_path, active=True, allowed_users=["simone"])
|
|
app = create_app(config)
|
|
client = TestClient(app, follow_redirects=False)
|
|
|
|
resp = client.get("/audit")
|
|
|
|
assert resp.status_code in (302, 303)
|
|
assert "/login" in resp.headers["location"]
|
|
|
|
|
|
# ── 8. no secret value leaks into any response ──────────────────────────────
|
|
|
|
|
|
def test_no_secret_leaks_into_any_action_response(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_install_fake_keyring(monkeypatch)
|
|
config = _make_config(tmp_path, active=True, allowed_users=["simone"])
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
class _StubOAuthClient:
|
|
async def authorize_redirect(self, request: Any, redirect_uri: str | None = None, **kw: Any) -> Any:
|
|
return RedirectResponse(f"https://gitea.example.lan/authorize?r={redirect_uri}", 302)
|
|
|
|
async def authorize_access_token(self, request: Any, **kw: Any) -> dict[str, Any]:
|
|
return {"userinfo": {"preferred_username": "simone"}}
|
|
|
|
async def userinfo(self, **kw: Any) -> dict[str, Any]:
|
|
return {"preferred_username": "simone"}
|
|
|
|
app.state.oauth.gitea = _StubOAuthClient()
|
|
client = TestClient(app, follow_redirects=False)
|
|
client.get("/auth/callback") # logs in
|
|
|
|
bodies: list[str] = []
|
|
bodies.append(client.get("/").text)
|
|
csrf = _csrf_from(bodies[-1])
|
|
bodies.append(client.get("/cache").text)
|
|
bodies.append(client.post("/cache/purge", data={"csrf": csrf}).text)
|
|
bodies.append(client.post("/cache/purge", data={"csrf": csrf, "stage": "execute"}).text)
|
|
bodies.append(client.get("/audit").text)
|
|
|
|
for body in bodies:
|
|
assert _FAKE_SESSION_SECRET not in body
|
|
assert _FAKE_OIDC_CLIENT_SECRET not in body
|
|
|
|
|
|
# ── extra coverage: helper edge cases + host.backup + degraded paths ───────
|
|
|
|
|
|
def test_is_contained_returns_false_on_resolve_oserror(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def _boom(self: Path, strict: bool = False) -> Path:
|
|
raise OSError("boom")
|
|
|
|
monkeypatch.setattr(Path, "resolve", _boom)
|
|
|
|
assert routes_module._is_contained(Path("/a/b"), Path("/a")) is False
|
|
|
|
|
|
def test_lock_is_fresh_missing_file_returns_false(tmp_path: Path) -> None:
|
|
assert routes_module._lock_is_fresh(tmp_path / "missing.lock") is False
|
|
|
|
|
|
def test_vm_remove_vmx_id_with_bad_base64_padding_is_400(tmp_path: Path) -> None:
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/vms/a/remove", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_vm_cleanup_preview_with_non_numeric_max_age_hours_defaults(tmp_path: Path) -> None:
|
|
config = _make_config(tmp_path)
|
|
config.paths.build_vms.mkdir(parents=True, exist_ok=True)
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/vms/cleanup", data={"csrf": csrf, "max_age_hours": "not-a-number"})
|
|
|
|
assert resp.status_code == 200
|
|
assert "older than 4h" in resp.text
|
|
|
|
|
|
def test_vm_remove_backend_unavailable_renders_error_partial_and_audits_error(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
vmx = _make_vmx(tmp_path)
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: None
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
vmx_id = routes_module.encode_vmx_id(vmx)
|
|
|
|
resp = client.post(f"/vms/{vmx_id}/remove", data={"csrf": csrf})
|
|
|
|
assert resp.status_code == 200
|
|
assert "unavailable" in resp.text.lower()
|
|
entry = _last_audit_entry(config.web.audit_log)
|
|
assert entry.result == "error"
|
|
|
|
|
|
def test_vm_remove_execute_error_result_audits_error(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = _make_vmx(tmp_path)
|
|
config = _make_config(tmp_path)
|
|
|
|
def _failing_remove(backend: Any, target_vmx: Path, dry_run: bool = False) -> RemovalResult:
|
|
return RemovalResult(
|
|
target=str(target_vmx), stopped=True, deleted=False, dir_removed=False, error="delete failed"
|
|
)
|
|
|
|
monkeypatch.setattr(routes_module, "remove_vm", _failing_remove)
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
vmx_id = routes_module.encode_vmx_id(vmx)
|
|
|
|
resp = client.post(f"/vms/{vmx_id}/remove", data={"csrf": csrf, "stage": "execute"})
|
|
|
|
assert resp.status_code == 200
|
|
assert "error" in resp.text.lower()
|
|
entry = _last_audit_entry(config.web.audit_log)
|
|
assert entry.result == "error"
|
|
assert "error removing" in entry.detail
|
|
|
|
|
|
def test_vm_remove_swallows_real_backend_error_never_500s(tmp_path: Path) -> None:
|
|
"""End-to-end (not mocked) proof that a real BackendError never 500s."""
|
|
|
|
class _FailingBackend:
|
|
def list_running(self) -> list[VmHandle]:
|
|
return []
|
|
|
|
def get_ip(self, handle: VmHandle, timeout: float = 0.0) -> str | None:
|
|
return None
|
|
|
|
def stop(self, handle: VmHandle, hard: bool = False) -> None:
|
|
raise BackendError("stop failed")
|
|
|
|
def delete(self, handle: VmHandle) -> None:
|
|
raise BackendError("delete failed")
|
|
|
|
vmx = _make_vmx(tmp_path)
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FailingBackend()
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
vmx_id = routes_module.encode_vmx_id(vmx)
|
|
|
|
resp = client.post(f"/vms/{vmx_id}/remove", data={"csrf": csrf}) # preview only, no retries
|
|
|
|
assert resp.status_code == 200
|
|
|
|
|
|
def test_cache_purge_exception_never_500s(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def _boom(*args: Any, **kwargs: Any) -> Any:
|
|
raise RuntimeError("disk exploded")
|
|
|
|
monkeypatch.setattr(routes_module, "purge_cargo_cache", _boom)
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/cache/purge", data={"csrf": csrf})
|
|
|
|
assert resp.status_code == 200
|
|
assert "error" in resp.text.lower()
|
|
entry = _last_audit_entry(config.web.audit_log)
|
|
assert entry.result == "error"
|
|
|
|
|
|
def test_host_backup_missing_template_field_is_400(tmp_path: Path) -> None:
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/host/backup", data={"csrf": csrf})
|
|
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_host_backup_unknown_template_name_is_400_not_audited(tmp_path: Path) -> None:
|
|
"""B6: a template name absent from ``list_templates()`` is rejected before
|
|
``backup_template`` is ever called -- a pure validation refusal, so (like
|
|
the VM-remove path-containment check) it is never audited.
|
|
"""
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post("/host/backup", data={"csrf": csrf, "template": "NoSuchTemplate"})
|
|
|
|
assert resp.status_code == 400
|
|
assert not config.web.audit_log.exists()
|
|
|
|
|
|
def test_host_backup_preview_then_execute_full_cycle(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
host_module, "_compress_7z", lambda source, archive: archive.write_bytes(b"fake-7z")
|
|
)
|
|
template_dir = tmp_path / "templates" / "LinuxBuild2404"
|
|
template_dir.mkdir(parents=True)
|
|
(template_dir / "LinuxBuild2404.vmx").write_text("x", encoding="utf-8")
|
|
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
preview = client.post(
|
|
"/host/backup", data={"csrf": csrf, "template": "LinuxBuild2404"}
|
|
)
|
|
assert preview.status_code == 200
|
|
assert "Confirm" in preview.text
|
|
assert "would archive" in preview.text
|
|
assert _last_audit_entry(config.web.audit_log).result == "dry-run"
|
|
|
|
execute = client.post(
|
|
"/host/backup", data={"csrf": csrf, "stage": "execute", "template": "LinuxBuild2404"}
|
|
)
|
|
assert execute.status_code == 200
|
|
entry = _last_audit_entry(config.web.audit_log)
|
|
assert entry.result == "ok"
|
|
assert entry.action == "host.backup"
|
|
|
|
|
|
def test_host_backup_execute_exception_never_500s(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
def _boom(*args: Any, **kwargs: Any) -> Any:
|
|
raise RuntimeError("disk exploded")
|
|
|
|
monkeypatch.setattr(routes_module, "backup_template", _boom)
|
|
# Must be a name list_templates() actually discovers, or the B6
|
|
# known-template check 400s before backup_template is ever reached.
|
|
(tmp_path / "templates" / "X").mkdir(parents=True)
|
|
config = _make_config(tmp_path)
|
|
app = create_app(config)
|
|
client = TestClient(app)
|
|
csrf = _get_csrf(client)
|
|
|
|
resp = client.post(
|
|
"/host/backup", data={"csrf": csrf, "stage": "execute", "template": "X"}
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert "error" in resp.text.lower()
|
|
entry = _last_audit_entry(config.web.audit_log)
|
|
assert entry.result == "error"
|
|
|
|
|
|
# ── web/security.py: direct unit coverage of the CSRF primitives ──────────
|
|
|
|
|
|
def test_issue_csrf_is_idempotent_within_a_session() -> None:
|
|
from ci_orchestrator.web.security import issue_csrf
|
|
|
|
fake_request = type("_Req", (), {"session": {}})()
|
|
|
|
first = issue_csrf(fake_request) # type: ignore[arg-type]
|
|
second = issue_csrf(fake_request) # type: ignore[arg-type]
|
|
|
|
assert first == second
|
|
assert len(first) > 20
|
|
|
|
|
|
def test_validate_csrf_false_when_no_token_issued_yet() -> None:
|
|
from ci_orchestrator.web.security import validate_csrf
|
|
|
|
fake_request = type("_Req", (), {"session": {}})()
|
|
|
|
assert validate_csrf(fake_request, "anything") is False # type: ignore[arg-type]
|
|
|
|
|
|
def test_validate_csrf_false_when_nothing_submitted() -> None:
|
|
from ci_orchestrator.web.security import validate_csrf
|
|
|
|
fake_request = type("_Req", (), {"session": {"csrf": "real-token"}})()
|
|
|
|
assert validate_csrf(fake_request, None) is False # type: ignore[arg-type]
|
|
|
|
|
|
def test_validate_csrf_true_for_matching_token() -> None:
|
|
from ci_orchestrator.web.security import validate_csrf
|
|
|
|
fake_request = type("_Req", (), {"session": {"csrf": "real-token"}})()
|
|
|
|
assert validate_csrf(fake_request, "real-token") is True # type: ignore[arg-type]
|