Simone
45670394b9
Adds the four destructive routes (vm.remove, vm.cleanup, cache.purge, host.backup) plus a read-only /audit view, all behind require_user and a synchronizer-token CSRF dependency (web/security.py). Every mutating POST is a two-step confirm gate: a first POST always dry-runs the underlying operations/ call and renders a confirm partial; only a stage=execute POST performs the real action. Both stages write one audit entry via operations/audit.py. VM remove additionally validates that the target VMX resolves inside config.paths.build_vms before touching anything (400 otherwise), and vm.remove/vm.cleanup refuse to run (409) when a fresh vm-start.lock is present, guarding against racing a live job. SessionMiddleware is now installed unconditionally in create_app (was previously only added when OIDC auth was active) since CSRF needs request.session in both auth modes; the session secret stays keyring-sourced when OIDC is active and falls back to a per-process ephemeral secret in open mode. Adds python-multipart to the [web] extra (required by FastAPI/Starlette form parsing, including plain url-encoded forms, in the installed starlette version).
835 lines
28 KiB
Python
835 lines
28 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 = "/cache") -> 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
|
|
assert "would remove" 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]
|
|
|
|
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_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("/vms")
|
|
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_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("/cache").text)
|
|
csrf = _csrf_from(bodies[-1])
|
|
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_preview_missing_template_dir_is_error_partial(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, "template": "NoSuchTemplate"})
|
|
|
|
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_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)
|
|
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]
|