462 lines
17 KiB
Python
462 lines
17 KiB
Python
|
"""Tests for the first-run bootstrap installer (system and per-user)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import contextlib
|
||
|
|
import json
|
||
|
|
import re
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest import mock
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from volumen import installer
|
||
|
|
from volumen.cli import main
|
||
|
|
from volumen.config import Config
|
||
|
|
|
||
|
|
SYSTEM_DIR = Path("/etc/volumen")
|
||
|
|
SYSTEM_USERS_FILE = Path("/var/lib/volumen/users.toml")
|
||
|
|
|
||
|
|
|
||
|
|
# --- 1. system_install produces the expected config substitutions -----------
|
||
|
|
|
||
|
|
|
||
|
|
class TestSystemInstall:
|
||
|
|
def test_writes_config_with_expected_substitutions(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
content_dir = tmp_path / "posts"
|
||
|
|
users_file = tmp_path / "users.toml"
|
||
|
|
monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
|
||
|
|
result = installer.system_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=content_dir,
|
||
|
|
users_file=users_file,
|
||
|
|
service_user="volumen",
|
||
|
|
admin_password="TopSecret!",
|
||
|
|
)
|
||
|
|
assert result.config_path == config_path
|
||
|
|
text = config_path.read_text()
|
||
|
|
assert re.search(r'^host = "::1"', text, re.MULTILINE) is not None
|
||
|
|
assert re.search(r'^env = "production"', text, re.MULTILINE) is not None
|
||
|
|
assert re.search(r"^trust_proxy = true", text, re.MULTILINE) is not None
|
||
|
|
assert re.search(r"^cookie_secure = true", text, re.MULTILINE) is not None
|
||
|
|
assert f'content_dir = "{content_dir}"' in text
|
||
|
|
assert f'users_file = "{users_file}"' in text
|
||
|
|
assert content_dir.is_dir()
|
||
|
|
assert (content_dir / "media").is_dir()
|
||
|
|
|
||
|
|
def test_writes_valid_scrypt_password_hash(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
|
||
|
|
result = installer.system_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=tmp_path / "posts",
|
||
|
|
users_file=tmp_path / "users.toml",
|
||
|
|
admin_password="TopSecret!",
|
||
|
|
)
|
||
|
|
text = config_path.read_text()
|
||
|
|
m = re.search(r'^password_hash = "(.+)"', text, re.MULTILINE)
|
||
|
|
assert m is not None
|
||
|
|
encoded = m.group(1)
|
||
|
|
assert encoded == result.password_hash
|
||
|
|
assert encoded.startswith("scrypt$16384$8$1$")
|
||
|
|
# Verify against password.
|
||
|
|
from volumen.password import verify_password
|
||
|
|
|
||
|
|
assert verify_password("TopSecret!", encoded) is True
|
||
|
|
|
||
|
|
def test_writes_long_hex_session_key(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
|
||
|
|
result = installer.system_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=tmp_path / "posts",
|
||
|
|
users_file=tmp_path / "users.toml",
|
||
|
|
admin_password="whatever",
|
||
|
|
)
|
||
|
|
text = config_path.read_text()
|
||
|
|
m = re.search(r'^session_key = "([0-9a-f]+)"', text, re.MULTILINE)
|
||
|
|
assert m is not None
|
||
|
|
assert len(m.group(1)) >= 128 # 64 bytes hex-encoded
|
||
|
|
assert m.group(1) == result.session_key
|
||
|
|
|
||
|
|
def test_config_round_trips_through_config_load(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
posts = tmp_path / "posts"
|
||
|
|
users_file = tmp_path / "users.toml"
|
||
|
|
monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
|
||
|
|
installer.system_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=posts,
|
||
|
|
users_file=users_file,
|
||
|
|
admin_password="TopSecret!",
|
||
|
|
)
|
||
|
|
loaded = Config.load(path=str(config_path))
|
||
|
|
assert loaded.content_dir == str(posts)
|
||
|
|
assert loaded.users_file == str(users_file)
|
||
|
|
assert loaded.host == "::1"
|
||
|
|
assert loaded.env == "production"
|
||
|
|
assert loaded.trust_proxy is True
|
||
|
|
assert loaded.cookie_secure is True
|
||
|
|
assert loaded.admin["password_hash"].startswith("scrypt$")
|
||
|
|
assert len(loaded.admin["session_key"]) >= 128
|
||
|
|
|
||
|
|
def test_existing_config_is_not_overwritten_without_force(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
original = 'host = "1.2.3.4"\nport = 1234\n'
|
||
|
|
config_path.write_text(original)
|
||
|
|
monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
|
||
|
|
result = installer.system_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=tmp_path / "posts",
|
||
|
|
users_file=tmp_path / "users.toml",
|
||
|
|
admin_password="TopSecret!",
|
||
|
|
)
|
||
|
|
# File unchanged.
|
||
|
|
assert config_path.read_text() == original
|
||
|
|
# Result carries a sentinel message and no fresh password_hash.
|
||
|
|
assert any("leaving untouched" in m for m in result.messages)
|
||
|
|
assert result.password_hash == ""
|
||
|
|
|
||
|
|
def test_force_makes_backup_with_original_content(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
original = 'host = "1.2.3.4"\nport = 1234\n'
|
||
|
|
config_path.write_text(original)
|
||
|
|
monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
|
||
|
|
installer.system_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=tmp_path / "posts",
|
||
|
|
users_file=tmp_path / "users.toml",
|
||
|
|
admin_password="TopSecret!",
|
||
|
|
force=True,
|
||
|
|
)
|
||
|
|
backups = list(tmp_path.glob("config.toml.bak.*.toml"))
|
||
|
|
assert len(backups) == 1
|
||
|
|
assert backups[0].read_text() == original
|
||
|
|
# New content must round-trip.
|
||
|
|
loaded = Config.load(path=str(config_path))
|
||
|
|
assert loaded.host == "::1"
|
||
|
|
|
||
|
|
def test_ensure_user_only_runs_when_user_missing(
|
||
|
|
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||
|
|
) -> None:
|
||
|
|
getpwnam_log: list[str] = []
|
||
|
|
run_log: list[list[str]] = []
|
||
|
|
|
||
|
|
def fake_getpwnam(name: str) -> object:
|
||
|
|
getpwnam_log.append(name)
|
||
|
|
raise KeyError(name)
|
||
|
|
|
||
|
|
def fake_run(cmd: list[str], *args: object, **kwargs: object) -> object:
|
||
|
|
run_log.append(list(cmd))
|
||
|
|
return object()
|
||
|
|
|
||
|
|
# Redirect the home-dir creation away from /var/lib so the test
|
||
|
|
# can run unprivileged without hitting real root.
|
||
|
|
fake_home = tmp_path / "fakehome"
|
||
|
|
monkeypatch.setattr(installer, "DEFAULT_SERVICE_HOME", fake_home)
|
||
|
|
with monkeypatch.context() as m:
|
||
|
|
m.setattr(installer.pwd, "getpwnam", fake_getpwnam)
|
||
|
|
m.setattr(installer, "_is_root", lambda: True)
|
||
|
|
m.setattr(installer.subprocess, "run", fake_run)
|
||
|
|
created = installer._ensure_user("volumen")
|
||
|
|
assert created is True
|
||
|
|
assert getpwnam_log == ["volumen"]
|
||
|
|
# useradd should fire exactly once.
|
||
|
|
assert any(cmd[:1] == ["useradd"] for cmd in run_log)
|
||
|
|
# Home dir was created at our redirected location.
|
||
|
|
assert fake_home.is_dir()
|
||
|
|
|
||
|
|
# If the user exists, no useradd is run.
|
||
|
|
getpwnam_log.clear()
|
||
|
|
run_log.clear()
|
||
|
|
|
||
|
|
class _Pwnam:
|
||
|
|
pw_uid = 1234
|
||
|
|
|
||
|
|
with monkeypatch.context() as m:
|
||
|
|
m.setattr(installer.pwd, "getpwnam", lambda name: _Pwnam())
|
||
|
|
m.setattr(installer.subprocess, "run", fake_run)
|
||
|
|
created = installer._ensure_user("volumen")
|
||
|
|
assert created is False
|
||
|
|
assert run_log == []
|
||
|
|
|
||
|
|
# If not root, even a missing user doesn't trigger useradd.
|
||
|
|
getpwnam_log.clear()
|
||
|
|
run_log.clear()
|
||
|
|
with monkeypatch.context() as m:
|
||
|
|
m.setattr(installer.pwd, "getpwnam", fake_getpwnam)
|
||
|
|
m.setattr(installer, "_is_root", lambda: False)
|
||
|
|
m.setattr(installer.subprocess, "run", fake_run)
|
||
|
|
created = installer._ensure_user("volumen")
|
||
|
|
assert created is False
|
||
|
|
assert run_log == []
|
||
|
|
|
||
|
|
def test_chown_chmod_skipped_silently_when_not_root(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
posts = tmp_path / "posts"
|
||
|
|
users_file = tmp_path / "users.toml"
|
||
|
|
monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
|
||
|
|
# Pretend we're not root — the chown/chmod helpers must no-op rather
|
||
|
|
# than raise, and no subprocess is spawned for useradd/systemctl.
|
||
|
|
monkeypatch.setattr(installer, "_is_root", lambda: False)
|
||
|
|
run_log: list[list[str]] = []
|
||
|
|
monkeypatch.setattr(
|
||
|
|
installer.subprocess,
|
||
|
|
"run",
|
||
|
|
lambda cmd, *a, **kw: run_log.append(list(cmd)) or object(),
|
||
|
|
)
|
||
|
|
result = installer.system_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=posts,
|
||
|
|
users_file=users_file,
|
||
|
|
admin_password="TopSecret!",
|
||
|
|
)
|
||
|
|
assert config_path.is_file()
|
||
|
|
assert posts.is_dir()
|
||
|
|
# No subprocess calls touched the system at all.
|
||
|
|
assert run_log == []
|
||
|
|
# Chmod was applied as a no-op (mode unchanged).
|
||
|
|
assert oct(config_path.stat().st_mode & 0o777) != oct(0o600) # default
|
||
|
|
# Result is a valid InstallResult.
|
||
|
|
assert result.config_path == config_path
|
||
|
|
|
||
|
|
def test_install_service_writes_systemd_unit_and_calls_systemctl(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
content_dir = tmp_path / "posts"
|
||
|
|
users_file = tmp_path / "users.toml"
|
||
|
|
systemd_unit = tmp_path / "volumen.service"
|
||
|
|
cmd_log: list[list[str]] = []
|
||
|
|
|
||
|
|
class _Pwnam:
|
||
|
|
pw_uid = 1234
|
||
|
|
|
||
|
|
class _Grnam:
|
||
|
|
gr_gid = 1234
|
||
|
|
|
||
|
|
def fake_getpwnam(name: str) -> object:
|
||
|
|
return _Pwnam()
|
||
|
|
|
||
|
|
def fake_getgrnam(name: str) -> object:
|
||
|
|
return _Grnam()
|
||
|
|
|
||
|
|
def fake_which(name: str) -> str | None:
|
||
|
|
return "/usr/local/bin/volumen" if name == "volumen" else None
|
||
|
|
|
||
|
|
def fake_run(cmd: list[str], *args: object, **kwargs: object) -> object:
|
||
|
|
cmd_log.append(list(cmd))
|
||
|
|
return object()
|
||
|
|
|
||
|
|
monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
|
||
|
|
monkeypatch.setattr(installer, "_is_root", lambda: True)
|
||
|
|
monkeypatch.setattr(installer.pwd, "getpwnam", fake_getpwnam)
|
||
|
|
monkeypatch.setattr(installer.grp, "getgrnam", fake_getgrnam)
|
||
|
|
monkeypatch.setattr(installer.shutil, "which", fake_which)
|
||
|
|
monkeypatch.setattr(installer.subprocess, "run", fake_run)
|
||
|
|
monkeypatch.setattr(installer, "SYSTEMD_UNIT_PATH", systemd_unit)
|
||
|
|
result = installer.system_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=content_dir,
|
||
|
|
users_file=users_file,
|
||
|
|
admin_password="TopSecret!",
|
||
|
|
install_service=True,
|
||
|
|
enable_service=True,
|
||
|
|
)
|
||
|
|
assert systemd_unit.is_file()
|
||
|
|
unit_text = systemd_unit.read_text()
|
||
|
|
assert "/usr/local/bin/volumen" in unit_text
|
||
|
|
assert "serve --config" in unit_text
|
||
|
|
assert f"--config {config_path}" in unit_text
|
||
|
|
assert f"--content {content_dir}" in unit_text
|
||
|
|
assert "User=volumen" in unit_text
|
||
|
|
assert "NoNewPrivileges=true" in unit_text
|
||
|
|
assert "ProtectSystem=strict" in unit_text
|
||
|
|
assert result.systemd_unit_path == systemd_unit
|
||
|
|
assert ["systemctl", "daemon-reload"] in cmd_log
|
||
|
|
assert ["systemctl", "enable", "--now", "volumen"] in cmd_log
|
||
|
|
|
||
|
|
|
||
|
|
# --- 2. user_install ---------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestUserInstall:
|
||
|
|
def test_writes_config_under_user_path(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
content_dir = tmp_path / "posts"
|
||
|
|
users_file = tmp_path / "users.toml"
|
||
|
|
monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
|
||
|
|
result = installer.user_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=content_dir,
|
||
|
|
users_file=users_file,
|
||
|
|
admin_password="TopSecret!",
|
||
|
|
)
|
||
|
|
assert result.service_user is None
|
||
|
|
assert result.systemd_unit_path is None
|
||
|
|
text = config_path.read_text()
|
||
|
|
assert re.search(r'^env = "development"', text, re.MULTILINE) is not None
|
||
|
|
assert re.search(r"^cookie_secure = false", text, re.MULTILINE) is not None
|
||
|
|
assert config_path.exists()
|
||
|
|
assert content_dir.is_dir()
|
||
|
|
|
||
|
|
def test_rejects_install_service(self, tmp_path: Path) -> None:
|
||
|
|
with pytest.raises(installer.InstallerError):
|
||
|
|
installer.user_install(
|
||
|
|
config_path=tmp_path / "config.toml",
|
||
|
|
content_dir=tmp_path / "posts",
|
||
|
|
users_file=tmp_path / "users.toml",
|
||
|
|
admin_password="x",
|
||
|
|
install_service=True,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# --- 3. _mutate_config_text --------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestMutateConfigText:
|
||
|
|
def test_substitutes_six_keys_without_disturbing_comments(self) -> None:
|
||
|
|
text = (
|
||
|
|
'host = "::"\n'
|
||
|
|
'env = "development"\n'
|
||
|
|
"trust_proxy = false\n"
|
||
|
|
"cookie_secure = false\n"
|
||
|
|
'content_dir = "/old"\n'
|
||
|
|
'users_file = "/old/users.toml"\n'
|
||
|
|
'password_hash = ""\n'
|
||
|
|
'session_key = ""\n'
|
||
|
|
)
|
||
|
|
out = installer._mutate_config_text(
|
||
|
|
text,
|
||
|
|
host="::1",
|
||
|
|
env="production",
|
||
|
|
trust_proxy=True,
|
||
|
|
cookie_secure=True,
|
||
|
|
content_dir="/new",
|
||
|
|
users_file="/new/users.toml",
|
||
|
|
)
|
||
|
|
assert 'host = "::1"' in out
|
||
|
|
assert 'env = "production"' in out
|
||
|
|
assert "trust_proxy = true" in out
|
||
|
|
assert "cookie_secure = true" in out
|
||
|
|
assert 'content_dir = "/new"' in out
|
||
|
|
assert 'users_file = "/new/users.toml"' in out
|
||
|
|
|
||
|
|
def test_injects_password_hash_and_session_key(self) -> None:
|
||
|
|
text = 'host = "::"\npassword_hash = ""\nsession_key = ""\n'
|
||
|
|
out = installer._mutate_config_text(
|
||
|
|
text,
|
||
|
|
password_hash="scrypt$abc",
|
||
|
|
session_key="deadbeef" * 16,
|
||
|
|
)
|
||
|
|
assert 'password_hash = "scrypt$abc"' in out
|
||
|
|
assert 'session_key = "deadbeef"' * 4 not in out # sentinel
|
||
|
|
assert 'session_key = "' in out
|
||
|
|
|
||
|
|
|
||
|
|
# --- 4. CLI status -----------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestStatus:
|
||
|
|
def test_status_json_reports_all_fields(
|
||
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||
|
|
) -> None:
|
||
|
|
from volumen import installer as inst_mod
|
||
|
|
|
||
|
|
# Pre-create a config so the inspector finds something.
|
||
|
|
posts = tmp_path / "posts"
|
||
|
|
users_file = tmp_path / "users.toml"
|
||
|
|
posts.mkdir()
|
||
|
|
users_file.touch()
|
||
|
|
config_path = tmp_path / "config.toml"
|
||
|
|
monkeypatch.setattr(inst_mod, "_ensure_user", lambda name: False)
|
||
|
|
installer.system_install(
|
||
|
|
config_path=config_path,
|
||
|
|
content_dir=posts,
|
||
|
|
users_file=users_file,
|
||
|
|
admin_password="TopSecret!",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr(inst_mod, "SYSTEMD_UNIT_PATH", tmp_path / "volumen.service")
|
||
|
|
# Force systemctl to a known "unknown" branch.
|
||
|
|
monkeypatch.setattr(inst_mod.shutil, "which", lambda name: None)
|
||
|
|
# Inject a users.toml entry for "admin".
|
||
|
|
users_file.write_text(
|
||
|
|
'[[users]]\nusername = "admin"\npassword_hash = "scrypt$x"\nrole = "admin"\n'
|
||
|
|
)
|
||
|
|
with (
|
||
|
|
mock.patch.object(
|
||
|
|
sys,
|
||
|
|
"argv",
|
||
|
|
[
|
||
|
|
"volumen",
|
||
|
|
"status",
|
||
|
|
"--config",
|
||
|
|
str(config_path),
|
||
|
|
"--data",
|
||
|
|
str(posts),
|
||
|
|
"--users-file",
|
||
|
|
str(users_file),
|
||
|
|
"--json",
|
||
|
|
],
|
||
|
|
),
|
||
|
|
contextlib.suppress(SystemExit),
|
||
|
|
):
|
||
|
|
main()
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
assert captured.out.strip().startswith("{")
|
||
|
|
info = json.loads(captured.out)
|
||
|
|
for key in (
|
||
|
|
"config_path",
|
||
|
|
"data_dir",
|
||
|
|
"users_file",
|
||
|
|
"config_exists",
|
||
|
|
"data_dir_exists",
|
||
|
|
"users_file_exists",
|
||
|
|
"password_hash_set",
|
||
|
|
"session_key_ok",
|
||
|
|
"systemd_unit_installed",
|
||
|
|
"service_active",
|
||
|
|
"admin_user_present",
|
||
|
|
"issues",
|
||
|
|
):
|
||
|
|
assert key in info
|
||
|
|
assert info["config_exists"] is True
|
||
|
|
assert info["password_hash_set"] is True
|
||
|
|
assert info["session_key_ok"] is True
|
||
|
|
assert info["admin_user_present"] is True
|
||
|
|
|
||
|
|
|
||
|
|
class TestCLISystemdLocalRefuse:
|
||
|
|
def test_local_with_systemd_exits_nonzero(self, capsys: pytest.CaptureFixture[str]) -> None:
|
||
|
|
exit_code = 0
|
||
|
|
with mock.patch.object(
|
||
|
|
sys,
|
||
|
|
"argv",
|
||
|
|
["volumen", "init", "--local", "--systemd", "--admin-password", "x"],
|
||
|
|
):
|
||
|
|
try:
|
||
|
|
main()
|
||
|
|
except SystemExit as exc:
|
||
|
|
exit_code = exc.code if isinstance(exc.code, int) else 1
|
||
|
|
assert exit_code == 2
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
assert "--systemd" in captured.err
|