485 lines
27 KiB
Python
485 lines
27 KiB
Python
"""Acceptance tests exercise authorization and installation state through HTTP."""
|
|
|
|
import base64
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from copy import deepcopy
|
|
from dataclasses import replace
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import time
|
|
import tomllib
|
|
from uuid import uuid4
|
|
|
|
from cryptography.fernet import Fernet
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
from fastapi.testclient import TestClient
|
|
import pytest
|
|
|
|
from provisioner.app import create_app
|
|
from provisioner.cli import ServiceLock, backup, initialize, restore
|
|
from provisioner.config import Settings
|
|
from provisioner.db import Database
|
|
from provisioner.security import Security
|
|
|
|
|
|
PASSWORD = "Ais-test-admin-only-452!"
|
|
ROOT_HASH = "$6$testsalt$" + "A" * 86
|
|
HOST_UUID = "d2e59b03-13cf-4ac9-a390-78c55f6a36d3"
|
|
HOST_MAC = "02:00:00:00:00:01"
|
|
SOURCE = '#!/bin/bash\nset -euo pipefail\ncase "$1" in\ncheck|apply|verify) exit 0;;\n*) exit 64;;\nesac\n'
|
|
|
|
|
|
def login(client, username="admin", password=PASSWORD):
|
|
response = client.post("/auth/login", data={"username": username, "password": password}, follow_redirects=False)
|
|
assert response.status_code == 303, response.text
|
|
response = client.get("/api/v1/me")
|
|
assert response.status_code == 200, response.text
|
|
return {"X-CSRF-Token": response.json()["csrf_token"]}
|
|
|
|
|
|
def post(client, path, data, csrf):
|
|
response = client.post(path, json=data, headers=csrf)
|
|
assert response.status_code in {200, 201}, f"{path}: {response.status_code} {response.text}"
|
|
return response.json()
|
|
|
|
|
|
@pytest.fixture
|
|
def environment(tmp_path):
|
|
settings = Settings(
|
|
data_dir=tmp_path / "data", master_key_file=tmp_path / "keys" / "master.key",
|
|
public_url="https://testserver", secure_cookies=False, bootstrap_username="admin",
|
|
bootstrap_password=PASSWORD, testing=True, four_eyes=False,
|
|
)
|
|
app = create_app(settings)
|
|
with TestClient(app, base_url="https://testserver") as client:
|
|
csrf = login(client)
|
|
yield app, client, csrf
|
|
|
|
|
|
def prepare_host(environment, disk_setup=None, approve=True):
|
|
app, client, csrf = environment
|
|
secret = post(client, "/api/v1/secrets", {"name": "test-root", "value": ROOT_HASH}, csrf)
|
|
group = post(client, "/api/v1/groups", {"name": "test-lab", "site": "lab", "valid_hours": 1}, csrf)
|
|
iso = post(client, "/api/v1/iso-records", {
|
|
"name": "Simulated ISO; not a real hardware certification", "build": "9.1-1", "sha256": "1" * 64,
|
|
"assistant_version": "test-only", "fingerprint": "2" * 64, "group_id": group["id"],
|
|
"native_token_support": True, "test_status": "passed", "test_evidence": "Synthetic HTTP acceptance fixture",
|
|
}, csrf)
|
|
module = post(client, "/api/v1/modules", {"name": "test-verification", "source": SOURCE,
|
|
"target_builds": ["9.1-1"], "retry_safe": True}, csrf)
|
|
publication = {"test_evidence": "Synthetic HTTP acceptance fixture", "reason": "Testing publication"}
|
|
post(client, f"/api/v1/modules/{module['id']}/publish", publication, csrf)
|
|
profile_data = json.loads((Path(__file__).parents[1] / "docs" / "sample-profile.json").read_text())
|
|
profile_data["values"]["root_secret_id"] = secret["id"]
|
|
if disk_setup is not None:
|
|
profile_data["values"]["disk_setup"] = deepcopy(disk_setup)
|
|
installation = post(client, "/api/v1/profiles", profile_data, csrf)
|
|
post(client, f"/api/v1/profiles/{installation['id']}/publish", publication, csrf)
|
|
postinstall = post(client, "/api/v1/profiles", {"name": "test-postinstall", "kind": "postinstall",
|
|
"target_builds": ["9.1-1"], "steps": [{"id": "verify", "module_id": module["id"], "required": True}]}, csrf)
|
|
post(client, f"/api/v1/profiles/{postinstall['id']}/publish", publication, csrf)
|
|
identities = [{"kind": "uuid", "value": HOST_UUID}, {"kind": "serial", "value": "LAB-HOST-001"},
|
|
{"kind": "mac", "value": HOST_MAC}]
|
|
host = post(client, "/api/v1/hosts", {"fqdn": "pve01.lab.example.net", "site": "lab",
|
|
"management_ip": "192.0.2.10/24", "identities": identities,
|
|
"installation_profile_id": installation["id"], "postinstall_profile_id": postinstall["id"], "iso_id": iso["id"]}, csrf)
|
|
run = post(client, f"/api/v1/hosts/{host['id']}/approve-install", {
|
|
"expected_version": host["version"], "valid_minutes": 30, "confirmation": host["fqdn"],
|
|
"disks_confirmed": True, "reason": "Dedicated simulated test host"}, csrf) if approve else None
|
|
payload = {"$schema": {"version": "1.0"}, "product": {"product": "pve"},
|
|
"iso": {"release": "9.1", "build": "1"}, "dmi": {"system": {"uuid": HOST_UUID, "serial": "LAB-HOST-001"}},
|
|
"network-interfaces": [{"mac": HOST_MAC}]}
|
|
return {"app": app, "client": client, "csrf": csrf, "group": group, "iso": iso,
|
|
"module": module, "installation": installation, "profile_data": profile_data,
|
|
"host": host, "run": run, "payload": payload, "identities": identities}
|
|
|
|
|
|
@pytest.fixture
|
|
def prepared(environment):
|
|
return prepare_host(environment)
|
|
|
|
|
|
def answer(prepared, payload=None):
|
|
return prepared["client"].post("/installer/v1/answer", json=payload or prepared["payload"],
|
|
headers={"Authorization": f"Bearer {prepared['group']['token']}"})
|
|
|
|
|
|
def enroll(prepared):
|
|
response = answer(prepared)
|
|
assert response.status_code == 200, response.text
|
|
config = tomllib.loads(response.text)
|
|
bootstrap = prepared["client"].get(config["first-boot"]["url"])
|
|
assert bootstrap.status_code == 200
|
|
encoded = re.search(r"config = base64.b64decode\('([^']+)'\)", bootstrap.text).group(1)
|
|
runtime = json.loads(base64.b64decode(encoded))
|
|
key = Ed25519PrivateKey.generate()
|
|
public_key = base64.b64encode(key.public_key().public_bytes_raw()).decode()
|
|
registration = {"run_id": prepared["run"]["id"], "enrollment_secret": runtime["enrollment_secret"],
|
|
"public_key": public_key, "identities": prepared["identities"], "boot_id": "boot-test-1"}
|
|
response = prepared["client"].post("/agent/v1/enroll", json=registration)
|
|
assert response.status_code == 200, response.text
|
|
return key, public_key, registration
|
|
|
|
|
|
def signed(prepared, key, method, path, payload=None, headers=None):
|
|
body = b"" if payload is None else json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()
|
|
timestamp, nonce = str(int(time.time())), uuid4().hex
|
|
message = f"{method}\n{path}\n{timestamp}\n{nonce}\n{hashlib.sha256(body).hexdigest()}".encode()
|
|
request_headers = {"X-Run-ID": prepared["run"]["id"],
|
|
"X-Device-Key": base64.b64encode(key.public_key().public_bytes_raw()).decode(),
|
|
"X-Timestamp": timestamp, "X-Nonce": nonce, "X-Signature": base64.b64encode(key.sign(message)).decode(),
|
|
"Content-Type": "application/json"}
|
|
request_headers.update(headers or {})
|
|
return prepared["client"].request(method, path, content=body, headers=request_headers)
|
|
|
|
|
|
def test_authentication_csrf_and_reader_permissions(environment):
|
|
app, client, csrf = environment
|
|
anonymous = TestClient(app, base_url="https://testserver")
|
|
try:
|
|
assert anonymous.get("/api/v1/hosts").status_code == 401
|
|
finally:
|
|
anonymous.close()
|
|
assert client.post("/api/v1/groups", json={"name": "denied", "site": "lab"}).status_code == 403
|
|
post(client, "/api/v1/users", {"username": "reader", "password": PASSWORD, "role": "reader"}, csrf)
|
|
reader_csrf = login(client, "reader")
|
|
assert client.get("/api/v1/hosts").status_code == 200
|
|
assert client.post("/api/v1/groups", json={"name": "denied", "site": "lab"}, headers=reader_csrf).status_code == 403
|
|
assert client.get("/api/v1/secrets").status_code == 403
|
|
|
|
|
|
def test_unknown_conflicting_and_blocked_hosts_never_receive_answer(prepared):
|
|
unknown = deepcopy(prepared["payload"])
|
|
unknown["dmi"]["system"] = {"uuid": str(uuid4()), "serial": "UNKNOWN-HOST"}
|
|
unknown["network-interfaces"] = [{"mac": "02:00:00:00:ff:fe"}]
|
|
assert answer(prepared, unknown).status_code == 403
|
|
contradictory = deepcopy(prepared["payload"])
|
|
contradictory["dmi"]["system"]["uuid"] = str(uuid4())
|
|
assert answer(prepared, contradictory).status_code == 409
|
|
host = prepared["client"].get(f"/api/v1/hosts/{prepared['host']['id']}").json()
|
|
response = prepared["client"].patch(f"/api/v1/hosts/{host['id']}", json={"expected_version": host["version"], "blocked": True}, headers=prepared["csrf"])
|
|
assert response.status_code == 200, response.text
|
|
assert answer(prepared).status_code == 403
|
|
|
|
|
|
def test_expired_approval_and_wrong_group_cannot_install(prepared):
|
|
other = post(prepared["client"], "/api/v1/groups", {"name": "other-group", "site": "lab"}, prepared["csrf"])
|
|
denied = prepared["client"].post("/installer/v1/answer", json=prepared["payload"], headers={"Authorization": f"Bearer {other['token']}"})
|
|
assert denied.status_code == 403
|
|
with prepared["app"].state.db.connection(write=True) as connection:
|
|
connection.execute("UPDATE approvals SET expires_at=?", (time.time() - 1,))
|
|
assert answer(prepared).status_code == 410
|
|
|
|
|
|
def test_author_cannot_publish_own_version_when_four_eyes_enabled(environment):
|
|
app, client, csrf = environment
|
|
app.state.settings.four_eyes = True
|
|
draft = post(client, "/api/v1/profiles", {"name": "self-publish-denied", "kind": "installation"}, csrf)
|
|
response = client.post(f"/api/v1/profiles/{draft['id']}/publish", json={
|
|
"test_evidence": "A laboratory test report", "reason": "Self publication attempt"}, headers=csrf)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_concurrent_installer_retries_reserve_one_immutable_answer(prepared):
|
|
with ThreadPoolExecutor(max_workers=10) as executor:
|
|
responses = list(executor.map(lambda _: answer(prepared), range(10)))
|
|
assert {response.status_code for response in responses} == {200}, [r.text for r in responses]
|
|
assert len({response.text for response in responses}) == 1
|
|
assert all(response.headers["cache-control"] == "no-store" for response in responses)
|
|
native = tomllib.loads(responses[0].text)
|
|
assert native["global"]["fqdn"] == prepared["host"]["fqdn"]
|
|
assert native["disk-setup"]["filter"] == {"ID_SERIAL_SHORT": "LAB_SYSTEM_DISK_001"}
|
|
assert native["disk-setup"]["filter-match"] == "all"
|
|
assert "selection" not in native["disk-setup"]
|
|
assert "expected_count" not in native["disk-setup"]
|
|
assert prepared["run"]["snapshot"]["warnings"] == [
|
|
"Die Hardwarekennung dient der Zuordnung im kontrollierten Provisionierungsnetz."]
|
|
with prepared["app"].state.db.connection() as connection:
|
|
assert connection.execute("SELECT count(*) FROM runs").fetchone()[0] == 1
|
|
assert connection.execute("SELECT status FROM approvals").fetchone()[0] == "consumed"
|
|
|
|
|
|
@pytest.mark.parametrize("evidence", [None, "Single-target-disk laboratory inventory"])
|
|
def test_automatic_zfs_raid0_preview_approval_and_native_answer(environment, evidence):
|
|
disk_setup = {"filesystem": "zfs", "selection": "all", "zfs": {"raid": "raid0", "ashift": 12}}
|
|
if evidence is not None:
|
|
disk_setup["inventory_evidence"] = evidence
|
|
configured = prepare_host(environment, disk_setup, approve=False)
|
|
client, csrf, host = configured["client"], configured["csrf"], configured["host"]
|
|
preview = client.get(f"/api/v1/hosts/{host['id']}/preview")
|
|
assert preview.status_code == 200, preview.text
|
|
snapshot = preview.json()
|
|
assert snapshot["disks"] == disk_setup
|
|
assert snapshot["warnings"][-1] == (
|
|
"Automatische Datenträgerwahl: Alle vom Installer erkannten Zielplatten werden verwendet. "
|
|
"Dieses Profil ist für Server mit genau einer Zielplatte vorgesehen.")
|
|
approval = {"expected_version": host["version"], "valid_minutes": 30,
|
|
"confirmation": host["fqdn"], "disks_confirmed": False, "reason": "Single-target-disk simulation"}
|
|
endpoint = f"/api/v1/hosts/{host['id']}/approve-install"
|
|
assert client.post(endpoint, json=approval, headers=csrf).status_code == 422
|
|
configured["run"] = post(client, endpoint, {**approval, "disks_confirmed": True}, csrf)
|
|
assert configured["run"]["snapshot"] == snapshot
|
|
response = answer(configured)
|
|
assert response.status_code == 200, response.text
|
|
assert tomllib.loads(response.text)["disk-setup"] == {
|
|
"filesystem": "zfs", "zfs": {"raid": "raid0", "ashift": 12},
|
|
"filter": {"DEVTYPE": "disk"}, "filter-match": "all"}
|
|
assert answer(configured).text == response.text
|
|
|
|
|
|
@pytest.mark.parametrize("override", [
|
|
{"filter": {"ID_SERIAL_SHORT": "LAB_SYSTEM_DISK_001"}},
|
|
{"filter": {}},
|
|
{"filter_match": "all"},
|
|
{"expected_count": 1},
|
|
{"expected_serials": []},
|
|
{"filesystem": "ext4"},
|
|
{"filesystem": "xfs"},
|
|
{"zfs": {"raid": "raid1"}},
|
|
{"zfs": {"raid": "raid10"}},
|
|
{"zfs": {"raid": "raidz-1"}},
|
|
{"zfs": {}},
|
|
{"zfs": {"raid": "raid0", "ashift": True}},
|
|
{"zfs": {"raid": "raid0", "copies": 4}},
|
|
{"lvm": {}},
|
|
{"selection": "ALL"},
|
|
{"selection": None},
|
|
{"selection": True},
|
|
{"selection": ["all"]},
|
|
{"inventory_evidence": ""},
|
|
{"inventory_evidence": " "},
|
|
{"inventory_evidence": None},
|
|
{"disk_list": ["sda"]},
|
|
])
|
|
def test_automatic_disk_selection_rejects_ambiguous_or_unsupported_profiles(environment, override):
|
|
disk_setup = {"filesystem": "zfs", "selection": "all", "zfs": {"raid": "raid0"}, **override}
|
|
configured = prepare_host(environment, disk_setup, approve=False)
|
|
client, csrf, host = configured["client"], configured["csrf"], configured["host"]
|
|
assert client.get(f"/api/v1/hosts/{host['id']}/preview").status_code == 422
|
|
response = client.post(f"/api/v1/hosts/{host['id']}/approve-install", json={
|
|
"expected_version": host["version"], "valid_minutes": 30, "confirmation": host["fqdn"],
|
|
"disks_confirmed": True, "reason": "Invalid selection must not authorize an install"}, headers=csrf)
|
|
assert response.status_code == 422, response.text
|
|
with configured["app"].state.db.connection() as connection:
|
|
assert connection.execute("SELECT count(*) FROM approvals").fetchone()[0] == 0
|
|
assert connection.execute("SELECT count(*) FROM runs").fetchone()[0] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("disk_setup", [
|
|
{"filesystem": "zfs", "zfs": {"raid": "raid0"}},
|
|
{"filesystem": "zfs", "zfs": {"raid": "raid0"}, "filter": {"DEVTYPE": "disk"}},
|
|
{"filesystem": "zfs", "zfs": {"raid": "raid0"},
|
|
"filter": {"ID_SERIAL_SHORT": "LAB_SYSTEM_DISK_001"}, "expected_count": 1,
|
|
"expected_serials": ["LAB_SYSTEM_DISK_001"]},
|
|
{"filesystem": "zfs", "zfs": {"raid": "raid1"},
|
|
"filter": {"ID_SERIAL_SHORT": "LAB_SYSTEM_DISK_001"}, "expected_count": 1,
|
|
"expected_serials": ["LAB_SYSTEM_DISK_001"], "inventory_evidence": "Verified one-disk inventory"},
|
|
])
|
|
def test_filtered_disk_selection_keeps_existing_requirements(environment, disk_setup):
|
|
configured = prepare_host(environment, disk_setup, approve=False)
|
|
response = configured["client"].get(f"/api/v1/hosts/{configured['host']['id']}/preview")
|
|
assert response.status_code == 422, response.text
|
|
|
|
|
|
def test_new_profile_version_cannot_change_prepared_run(prepared):
|
|
original = prepared["run"]["snapshot"]
|
|
data = deepcopy(prepared["profile_data"])
|
|
data["values"]["network"]["dns"] = "192.0.2.54"
|
|
version2 = post(prepared["client"], "/api/v1/profiles", data, prepared["csrf"])
|
|
assert version2["version"] == prepared["installation"]["version"] + 1
|
|
assert version2["id"] != prepared["installation"]["id"]
|
|
current = prepared["client"].get(f"/api/v1/runs/{prepared['run']['id']}").json()
|
|
assert current["snapshot"] == original
|
|
response = answer(prepared)
|
|
assert response.status_code == 200, response.text
|
|
assert tomllib.loads(response.text)["network"]["dns"] == "192.0.2.53"
|
|
|
|
|
|
def test_enrollment_signature_sequence_and_verified_completion(prepared):
|
|
key, _, registration = enroll(prepared)
|
|
run_id = prepared["run"]["id"]
|
|
assert answer(prepared).status_code in {403, 410}
|
|
# Retrying a lost enrollment response must not generate a second device identity.
|
|
assert prepared["client"].post("/agent/v1/enroll", json=registration).status_code == 200
|
|
lease = signed(prepared, key, "POST", "/agent/v1/lease", {"run_id": run_id})
|
|
assert lease.status_code == 200 and lease.json()["action"] == "run"
|
|
manifest = signed(prepared, key, "GET", f"/agent/v1/runs/{run_id}/manifest")
|
|
assert manifest.status_code == 200, manifest.text
|
|
wrong_key = Ed25519PrivateKey.generate()
|
|
assert signed(prepared, wrong_key, "GET", f"/agent/v1/runs/{run_id}/manifest").status_code in {401, 403}
|
|
premature = signed(prepared, key, "POST", f"/agent/v1/runs/{run_id}/complete", {"verification": {"verify": {"passed": True}}})
|
|
assert premature.status_code == 409
|
|
def event(sequence, kind, **extra):
|
|
return {"sequence": sequence, "type": kind, "boot_id": "boot-test-1", "step_id": "verify",
|
|
"occurred_at": "2026-09-13T12:00:00Z", **extra}
|
|
out_of_order = {"events": [event(2, "step.started")]}
|
|
assert signed(prepared, key, "POST", f"/agent/v1/runs/{run_id}/events", out_of_order).status_code == 409
|
|
started = {"events": [event(1, "step.started")]}
|
|
assert signed(prepared, key, "POST", f"/agent/v1/runs/{run_id}/events", started).status_code == 200
|
|
assert signed(prepared, key, "POST", f"/agent/v1/runs/{run_id}/events", started).status_code == 200
|
|
false_success = {"events": [event(2, "step.succeeded", exit_code=0, verification={"passed": False})]}
|
|
assert signed(prepared, key, "POST", f"/agent/v1/runs/{run_id}/events", false_success).status_code == 409
|
|
succeeded = {"events": [event(2, "step.succeeded", exit_code=0, verification={"passed": True})]}
|
|
response = signed(prepared, key, "POST", f"/agent/v1/runs/{run_id}/events", succeeded)
|
|
assert response.status_code == 200, response.text
|
|
false_completion = signed(prepared, key, "POST", f"/agent/v1/runs/{run_id}/complete", {"verification": {"verify": {"passed": False}}})
|
|
assert false_completion.status_code == 409
|
|
completion = signed(prepared, key, "POST", f"/agent/v1/runs/{run_id}/complete", {"verification": {"verify": {"passed": True}}})
|
|
assert completion.status_code == 200, completion.text
|
|
status = prepared["client"].get(f"/api/v1/runs/{run_id}").json()["status"]
|
|
assert status == "succeeded"
|
|
assert answer(prepared).status_code in {403, 410}
|
|
assert signed(prepared, key, "GET", f"/agent/v1/runs/{run_id}/manifest").status_code in {403, 410}
|
|
|
|
|
|
def test_device_signature_replay_and_cross_run_access_are_denied(prepared):
|
|
key, _, _ = enroll(prepared)
|
|
run_id = prepared["run"]["id"]
|
|
assert signed(prepared, key, "POST", "/agent/v1/lease", {"run_id": run_id}).status_code == 200
|
|
original = signed(prepared, key, "GET", f"/agent/v1/runs/{run_id}/manifest")
|
|
assert original.status_code == 200
|
|
repeated = prepared["client"].request(original.request.method, original.request.url, content=original.request.content, headers=original.request.headers)
|
|
assert repeated.status_code == 409
|
|
foreign_path = "/agent/v1/runs/another-run/manifest"
|
|
tampered = prepared["client"].get(foreign_path, headers=original.request.headers)
|
|
assert tampered.status_code == 401
|
|
assert signed(prepared, key, "GET", foreign_path).status_code == 403
|
|
|
|
|
|
def test_tampered_module_is_not_delivered_to_runner(prepared):
|
|
key, _, _ = enroll(prepared)
|
|
run_id = prepared["run"]["id"]
|
|
assert signed(prepared, key, "POST", "/agent/v1/lease", {"run_id": run_id}).status_code == 200
|
|
checksum = prepared["module"]["digest"]
|
|
endpoint = f"/agent/v1/artifacts/{checksum}"
|
|
assert signed(prepared, key, "GET", endpoint).status_code == 200
|
|
artifact = prepared["app"].state.settings.data_dir / "artifacts" / checksum
|
|
artifact.write_bytes(b"#!/bin/bash\nexit 99\n")
|
|
response = signed(prepared, key, "GET", endpoint)
|
|
assert response.status_code == 503
|
|
assert "exit 99" not in response.text
|
|
|
|
|
|
def test_reconciliation_waits_for_issued_lease_and_requires_local_confirmation(prepared):
|
|
key, _, _ = enroll(prepared)
|
|
client, csrf = prepared["client"], prepared["csrf"]
|
|
run_id = prepared["run"]["id"]
|
|
lease = signed(prepared, key, "POST", "/agent/v1/lease", {"run_id": run_id})
|
|
assert lease.status_code == 200 and lease.json()["action"] == "run"
|
|
issued_until = lease.json()["expires_at"]
|
|
current = client.get(f"/api/v1/runs/{run_id}").json()
|
|
cancelled = post(client, f"/api/v1/runs/{run_id}/cancel", {
|
|
"expected_version": current["version"], "reason": "Stop before checking local host state"}, csrf)
|
|
stop = signed(prepared, key, "POST", "/agent/v1/lease", {"run_id": run_id})
|
|
assert stop.status_code == 200 and stop.json()["action"] == "stop"
|
|
with prepared["app"].state.db.connection(write=True) as connection:
|
|
assert connection.execute("SELECT lease_until FROM runs WHERE id=?", (run_id,)).fetchone()[0] == issued_until
|
|
connection.execute("UPDATE runs SET answer_until=? WHERE id=?", (time.time() - 1, run_id))
|
|
request = {"expected_version": cancelled["version"], "reason": "Installer and runner stopped locally and checked",
|
|
"confirmation": prepared["host"]["fqdn"], "execution_stopped": True}
|
|
endpoint = f"/api/v1/runs/{run_id}/reconcile"
|
|
assert client.post(endpoint, json=request, headers=csrf).status_code == 409
|
|
with prepared["app"].state.db.connection(write=True) as connection:
|
|
connection.execute("UPDATE runs SET lease_until=? WHERE id=?", (time.time() - 1, run_id))
|
|
assert client.post(endpoint, json={**request, "confirmation": "wrong.lab.example.net"}, headers=csrf).status_code == 422
|
|
assert client.post(endpoint, json={**request, "execution_stopped": False}, headers=csrf).status_code == 422
|
|
reconciled = post(client, endpoint, request, csrf)
|
|
assert reconciled["status"] == "cancelled"
|
|
with prepared["app"].state.db.connection() as connection:
|
|
row = connection.execute("SELECT * FROM runs WHERE id=?", (run_id,)).fetchone()
|
|
assert row["device_key"] is None and row["enrollment_hash"] is None
|
|
assert connection.execute("SELECT status FROM approvals WHERE id=?", (row["approval_id"],)).fetchone()[0] == "revoked"
|
|
assert connection.execute("SELECT 1 FROM audit WHERE action='run.reconciled' AND object_id=?", (run_id,)).fetchone()
|
|
assert signed(prepared, key, "POST", "/agent/v1/lease", {"run_id": run_id}).status_code == 401
|
|
host = client.get(f"/api/v1/hosts/{prepared['host']['id']}").json()
|
|
next_run = post(client, f"/api/v1/hosts/{host['id']}/approve-install", {
|
|
"expected_version": host["version"], "valid_minutes": 30, "confirmation": host["fqdn"],
|
|
"disks_confirmed": True, "reason": "Explicit new simulation after verified local stop"}, csrf)
|
|
assert next_run["id"] != run_id and next_run["status"] == "prepared"
|
|
|
|
|
|
def test_sensitive_values_are_encrypted_and_absent_from_management(prepared):
|
|
response = answer(prepared)
|
|
assert response.status_code == 200, response.text
|
|
bootstrap_url = tomllib.loads(response.text)["first-boot"]["url"]
|
|
for endpoint in ("/api/v1/hosts", "/api/v1/profiles", "/api/v1/runs", "/api/v1/audit", "/api/v1/groups"):
|
|
management = prepared["client"].get(endpoint)
|
|
assert management.status_code == 200, management.text
|
|
assert ROOT_HASH not in management.text
|
|
assert prepared["group"]["token"] not in management.text
|
|
assert bootstrap_url not in management.text
|
|
with prepared["app"].state.db.connection() as connection:
|
|
row = connection.execute("SELECT * FROM runs").fetchone()
|
|
assert ROOT_HASH not in row["answer_ciphertext"]
|
|
assert ROOT_HASH in prepared["app"].state.security.decrypt(row["answer_ciphertext"])
|
|
secret = connection.execute("SELECT ciphertext FROM secrets").fetchone()[0]
|
|
assert secret != ROOT_HASH
|
|
assert prepared["app"].state.security.decrypt(secret) == ROOT_HASH
|
|
|
|
|
|
def test_live_backup_offline_restore_revokes_all_active_credentials(prepared, tmp_path):
|
|
enroll(prepared)
|
|
settings = prepared["app"].state.settings
|
|
destination = tmp_path / "snapshot"
|
|
backup(settings, destination)
|
|
key_bytes = settings.master_key_file.read_bytes()
|
|
for path in destination.rglob("*"):
|
|
if path.is_file():
|
|
assert key_bytes not in path.read_bytes()
|
|
restored = replace(settings, data_dir=tmp_path / "restored")
|
|
restore(restored, destination)
|
|
db = Database(restored)
|
|
with db.connection() as connection:
|
|
assert connection.execute("SELECT count(*) FROM sessions").fetchone()[0] == 0
|
|
assert connection.execute("SELECT count(*) FROM groups WHERE revoked=0").fetchone()[0] == 0
|
|
row = connection.execute("SELECT * FROM runs").fetchone()
|
|
assert row["status"] == "needs_review"
|
|
assert all(row[name] is None for name in ("device_key", "bootstrap_hash", "enrollment_hash", "report_hash", "answer_ciphertext"))
|
|
assert Security(restored).decrypt(connection.execute("SELECT ciphertext FROM secrets").fetchone()[0]) == ROOT_HASH
|
|
assert list((restored.data_dir / "artifacts").iterdir())
|
|
with pytest.raises((RuntimeError, ValueError), match="in use|empty"):
|
|
restore(settings, destination)
|
|
restored_app = create_app(restored)
|
|
with TestClient(restored_app, base_url="https://testserver") as restored_client:
|
|
restored_csrf = login(restored_client)
|
|
current = restored_client.get(f"/api/v1/runs/{prepared['run']['id']}").json()
|
|
reconciled = post(restored_client, f"/api/v1/runs/{current['id']}/reconcile", {
|
|
"expected_version": current["version"], "confirmation": prepared["host"]["fqdn"],
|
|
"execution_stopped": True, "reason": "Physical host state checked after restore"}, restored_csrf)
|
|
assert reconciled["status"] == "cancelled"
|
|
|
|
|
|
def test_init_creates_only_password_hash_and_separate_key(tmp_path, monkeypatch):
|
|
settings = Settings(data_dir=tmp_path / "data", master_key_file=tmp_path / "keys" / "master.key")
|
|
monkeypatch.setattr("provisioner.cli.getpass.getpass", lambda _: PASSWORD)
|
|
initialize(settings, "admin")
|
|
with Database(settings).connection() as connection:
|
|
stored = connection.execute("SELECT password_hash FROM users").fetchone()[0]
|
|
assert PASSWORD not in stored
|
|
assert Security.verify_password(PASSWORD, stored)
|
|
assert settings.master_key_file.is_file()
|
|
with ServiceLock(settings.data_dir):
|
|
with pytest.raises(RuntimeError, match="in use"):
|
|
with ServiceLock(settings.data_dir):
|
|
pass
|
|
|
|
|
|
def test_restore_rejects_wrong_key_and_modified_backup(tmp_path, monkeypatch):
|
|
settings = Settings(data_dir=tmp_path / "original", master_key_file=tmp_path / "keys" / "master.key")
|
|
monkeypatch.setattr("provisioner.cli.getpass.getpass", lambda _: PASSWORD)
|
|
initialize(settings, "admin")
|
|
destination = tmp_path / "backup"
|
|
backup(settings, destination)
|
|
wrong_key = tmp_path / "keys" / "wrong.key"
|
|
wrong_key.write_bytes(Fernet.generate_key())
|
|
wrong_settings = replace(settings, data_dir=tmp_path / "wrong-restore", master_key_file=wrong_key)
|
|
with pytest.raises(ValueError, match="does not match"):
|
|
restore(wrong_settings, destination)
|
|
assert not wrong_settings.data_dir.exists()
|
|
config = destination / "settings.json"
|
|
config.write_text(config.read_text() + "\n", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="checksum mismatch"):
|
|
restore(replace(settings, data_dir=tmp_path / "corrupt-restore"), destination)
|