mirror of
https://github.com/Joulenap/joulenap.git
synced 2026-08-11 13:21:43 +02:00
Catubba
151b2a53f6
A backup server serving two datastores is two devices, which the duplicate guard deliberately allows -- but both wanted a token called joulenap, so setting up the second one deleted and recreated the first one's token. The first device was left holding a dead secret, and re-entering the new one did not repair it: deleting a token also drops its ACL entries, and provisioning re-grants only /datastore/<its own datastore>, so the other datastore stayed locked out until root ran acl update by hand. The product supported a configuration its own wizard could not provision. Tokens on a backup server are now named joulenap-<datastore>, sanitised to the character set PBS accepts and falling back to the bare prefix if nothing survives. The two never meet, and each keeps the narrow per-datastore grant rather than widening to /datastore. A Proxmox host is a single device and cannot collide with itself, so its token stays plain joulenap. The name is derived rather than exposed: a field would only invite tokens Joulenap later fails to find. Tokens already in use are untouched. The conflict dialog no longer claims the name is "joulenap", since on a backup server it is not. A device card also stops reporting "Connected - API OK" for what is a one second TCP connect to the API port. The authenticated call behind it is made and its failure discarded, so a server whose credential had been revoked advertised itself as healthy indefinitely, with cached usage figures beside it to match. The label now reads "Reachable", which is what is actually checked; the Test button, which surfaces the same call's error, owns the API verdict. Changing the underlying field was rejected: it is a published contract, both in the dashboard payload and as joulenap_pbs_online, documented as answering on the API port. Documented in the architecture, the wizard guide and the example config, including that replacing a token clears its permissions -- so a hand-made setup where one token served several datastores needs re-granting.
281 lines
9.6 KiB
Python
281 lines
9.6 KiB
Python
"""Setup-wizard endpoints: auth guard, request wiring, error mapping, real keygen."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.connectors.errors import ApiError, TokenExistsError
|
|
from app.connectors.provision import pbs_token_name
|
|
from app.core import wizard
|
|
from app.main import create_app
|
|
|
|
|
|
@pytest.fixture
|
|
def client(temp_config, temp_db):
|
|
app = create_app()
|
|
with TestClient(app) as c:
|
|
c.post("/api/auth/setup", json={"username": "admin", "password": "secret12"})
|
|
yield c
|
|
|
|
|
|
def test_wizard_requires_auth(temp_config, temp_db):
|
|
with TestClient(create_app()) as c:
|
|
r = c.post("/api/wizard/pve/connect", json={"host": "pve.local"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_pve_connect_passes_through(client, monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_connect(**kwargs):
|
|
captured.update(kwargs)
|
|
return {"connected": True, "nodes": [], "storages": [], "token": None}
|
|
|
|
monkeypatch.setattr(wizard, "pve_connect", fake_connect)
|
|
r = client.post(
|
|
"/api/wizard/pve/connect",
|
|
json={"host": "pve.local", "mode": "root", "username": "root@pam", "password": "pw"},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["connected"] is True
|
|
assert captured["host"] == "pve.local" and captured["mode"] == "root"
|
|
assert captured["username"] == "root@pam"
|
|
|
|
|
|
def test_connector_error_maps_to_502(client, monkeypatch):
|
|
def boom(**_kwargs):
|
|
raise ApiError("connection refused")
|
|
|
|
monkeypatch.setattr(wizard, "pve_connect", boom)
|
|
r = client.post("/api/wizard/pve/connect", json={"host": "pve.local"})
|
|
assert r.status_code == 502
|
|
|
|
|
|
def test_a_taken_token_name_is_409_not_502(client, monkeypatch):
|
|
"""Nothing failed upstream — we declined to replace a token the user has not agreed to
|
|
lose. The wizard keys on the status to offer "replace it" instead of a connection error."""
|
|
|
|
def taken(**_kwargs):
|
|
raise TokenExistsError("An API token named 'joulenap' already exists for root@pam.")
|
|
|
|
monkeypatch.setattr(wizard, "pbs_provision", taken)
|
|
r = client.post(
|
|
"/api/wizard/pbs/provision",
|
|
json={"host": "pbs.local", "password": "pw", "datastore": "backup"},
|
|
)
|
|
assert r.status_code == 409
|
|
assert "already exists" in r.json()["detail"]
|
|
|
|
|
|
def test_replace_token_is_forwarded(client, monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_provision(**kwargs):
|
|
captured.update(kwargs)
|
|
return {"id": "root@pam!joulenap", "secret": "s"}
|
|
|
|
monkeypatch.setattr(wizard, "pbs_provision", fake_provision)
|
|
r = client.post(
|
|
"/api/wizard/pbs/provision",
|
|
json={
|
|
"host": "pbs.local",
|
|
"password": "pw",
|
|
"datastore": "backup",
|
|
"replace_token": True,
|
|
},
|
|
)
|
|
assert r.status_code == 200
|
|
assert captured["replace_token"] is True
|
|
|
|
|
|
def test_the_pbs_token_name_is_derived_from_the_datastore(client, monkeypatch):
|
|
"""Two devices on one backup server must not contend for a token name: naming both
|
|
``joulenap`` meant provisioning the second deleted the first one's token."""
|
|
captured = {}
|
|
|
|
def fake_provision(**kwargs):
|
|
captured.update(kwargs)
|
|
return {"id": "root@pam!joulenap-lab", "secret": "s"}
|
|
|
|
monkeypatch.setattr(wizard, "pbs_provision", fake_provision)
|
|
r = client.post(
|
|
"/api/wizard/pbs/provision",
|
|
json={"host": "pbs.local", "password": "pw", "datastore": "lab"},
|
|
)
|
|
assert r.status_code == 200
|
|
assert captured["token_name"] == "joulenap-lab"
|
|
|
|
|
|
def test_an_explicit_token_name_still_wins(client, monkeypatch):
|
|
"""The field stays honoured for anyone driving the API directly."""
|
|
captured = {}
|
|
|
|
def fake_provision(**kwargs):
|
|
captured.update(kwargs)
|
|
return {"id": "root@pam!mine", "secret": "s"}
|
|
|
|
monkeypatch.setattr(wizard, "pbs_provision", fake_provision)
|
|
r = client.post(
|
|
"/api/wizard/pbs/provision",
|
|
json={
|
|
"host": "pbs.local",
|
|
"password": "pw",
|
|
"datastore": "lab",
|
|
"token_name": "mine",
|
|
},
|
|
)
|
|
assert r.status_code == 200
|
|
assert captured["token_name"] == "mine"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("datastore", "expected"),
|
|
[
|
|
("lab", "joulenap-lab"),
|
|
("backup_2", "joulenap-backup_2"),
|
|
("dot.name", "joulenap-dot.name"),
|
|
(" spaced ", "joulenap-spaced"),
|
|
("a b/c", "joulenap-a-b-c"),
|
|
# Nothing survives sanitising, so fall back rather than send PBS a name it rejects.
|
|
("///", "joulenap"),
|
|
("", "joulenap"),
|
|
],
|
|
)
|
|
def test_token_name_sanitising(datastore, expected):
|
|
assert pbs_token_name(datastore) == expected
|
|
|
|
|
|
def test_pbs_check_passes_through(client, monkeypatch):
|
|
monkeypatch.setattr(
|
|
wizard, "pbs_check", lambda **_k: {"reachable": True, "fingerprint": "AA:BB"}
|
|
)
|
|
r = client.post("/api/wizard/pbs/check", json={"host": "pbs.local"})
|
|
assert r.json() == {"reachable": True, "fingerprint": "AA:BB"}
|
|
|
|
|
|
def test_pbs_grant_sync_passes_through(client, monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_grant(**kwargs):
|
|
captured.update(kwargs)
|
|
return {"token_id": kwargs["token_id"], "roles": ["RemoteAdmin"]}
|
|
|
|
monkeypatch.setattr(wizard, "pbs_grant_sync", fake_grant)
|
|
r = client.post(
|
|
"/api/wizard/pbs/grant-sync",
|
|
json={
|
|
"host": "pbs.local",
|
|
"password": "pw",
|
|
"api_token_id": "root@pam!joulenap",
|
|
"fingerprint": "AA:BB",
|
|
},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["roles"] == ["RemoteAdmin"]
|
|
# the endpoint renames api_token_id -> token_id and defaults the root realm
|
|
assert captured["token_id"] == "root@pam!joulenap"
|
|
assert captured["username"] == "root@pam" and captured["password"] == "pw"
|
|
assert "pw" not in r.text
|
|
|
|
|
|
def test_pbs_grant_sync_needs_a_password_and_a_token(client):
|
|
r = client.post("/api/wizard/pbs/grant-sync", json={"host": "pbs.local", "password": "pw"})
|
|
assert r.status_code == 422
|
|
|
|
|
|
def test_detect_mac_passes_through(client, monkeypatch):
|
|
monkeypatch.setattr(wizard, "wol_detect_mac", lambda **_k: {"mac": "00:11:22:33:44:55"})
|
|
r = client.post("/api/wizard/wol/detect-mac", json={"host": "pbs.local"})
|
|
assert r.json()["mac"] == "00:11:22:33:44:55"
|
|
|
|
|
|
def test_ssh_keygen_generates_real_key(client):
|
|
r = client.post("/api/wizard/ssh/keygen")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["public_key"].startswith("ssh-ed25519 ")
|
|
assert body["key_path"].endswith("id_ed25519")
|
|
assert body["created"] is True
|
|
# The restricted line to paste/install locks the key to poweroff only.
|
|
assert body["authorized_keys_line"].startswith('command="systemctl poweroff",')
|
|
assert body["authorized_keys_line"].endswith(body["public_key"])
|
|
|
|
|
|
def test_ssh_keygen_reuses_an_existing_key(client):
|
|
"""Adding a second PBS must not invalidate the first one's power-off: every device's
|
|
ssh_key_path points at this same file, so a regenerated key would leave the first PBS
|
|
trusting a public half no private key matches any more."""
|
|
first = client.post("/api/wizard/ssh/keygen").json()
|
|
key_file = Path(first["key_path"])
|
|
written = key_file.read_bytes()
|
|
|
|
second = client.post("/api/wizard/ssh/keygen").json()
|
|
|
|
assert second["public_key"] == first["public_key"]
|
|
assert second["created"] is False
|
|
assert key_file.read_bytes() == written # not rewritten, byte for byte
|
|
|
|
|
|
def test_ssh_keygen_replaces_an_unreadable_key(client):
|
|
"""A file that isn't a usable private key can't be what any PBS trusts either, so there
|
|
is nothing to preserve — generate over it rather than failing the whole flow."""
|
|
key_file = Path(client.post("/api/wizard/ssh/keygen").json()["key_path"])
|
|
key_file.write_text("not a private key")
|
|
|
|
body = client.post("/api/wizard/ssh/keygen").json()
|
|
|
|
assert body["created"] is True
|
|
assert body["public_key"].startswith("ssh-ed25519 ")
|
|
|
|
|
|
def test_ssh_install_passes_through(client, monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_install(**kwargs):
|
|
captured.update(kwargs)
|
|
return {"installed": True}
|
|
|
|
monkeypatch.setattr(wizard, "ssh_install", fake_install)
|
|
r = client.post(
|
|
"/api/wizard/ssh/install",
|
|
json={"host": "pbs.local", "password": "pw", "public_key": "ssh-ed25519 AAAA"},
|
|
)
|
|
assert r.json() == {"installed": True}
|
|
assert captured["host"] == "pbs.local" and captured["user"] == "root"
|
|
|
|
|
|
def test_wol_test_sends_a_packet_for_a_mac_not_yet_saved(client, monkeypatch):
|
|
# The wizard tests a MAC it has just detected, before there is a device to hang it on —
|
|
# so this takes the MAC in the body rather than reading one out of the config.
|
|
sent: list[tuple] = []
|
|
monkeypatch.setattr(
|
|
"app.core.wizard.send_magic_packet",
|
|
lambda mac, broadcast=None, source_ip=None: sent.append((mac, broadcast)),
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.connectors.net.wol_target", lambda host, iface: ("192.0.2.255", "192.0.2.5")
|
|
)
|
|
|
|
resp = client.post(
|
|
"/api/wizard/wol/test", json={"mac": "00:11:22:33:44:55", "host": "192.0.2.20"}
|
|
)
|
|
|
|
assert resp.status_code == 200 and resp.json()["sent"] is True
|
|
assert sent == [("00:11:22:33:44:55", "192.0.2.255")]
|
|
|
|
|
|
def test_wol_test_falls_back_to_the_global_broadcast_without_a_host(client, monkeypatch):
|
|
sent: list[tuple] = []
|
|
monkeypatch.setattr(
|
|
"app.core.wizard.send_magic_packet",
|
|
lambda mac, broadcast=None, source_ip=None: sent.append((mac, broadcast)),
|
|
)
|
|
|
|
resp = client.post("/api/wizard/wol/test", json={"mac": "00:11:22:33:44:55"})
|
|
|
|
assert resp.status_code == 200
|
|
assert sent == [("00:11:22:33:44:55", "255.255.255.255")]
|