200 lines
7.9 KiB
Python
200 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import subprocess
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from scripts import provision_device
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def restore_windows_test_file_permissions(tmp_path):
|
|
yield
|
|
if os.name == "nt":
|
|
for path in tmp_path.rglob("*"):
|
|
if path.is_file():
|
|
path.chmod(0o600)
|
|
|
|
|
|
def account_config(username: str = "example") -> dict:
|
|
return {"account": {"username": username, "password": "example-password"}}
|
|
|
|
|
|
class FakePwd:
|
|
def __init__(self, values: dict[str, SimpleNamespace]) -> None:
|
|
self.values = values
|
|
|
|
def getpwnam(self, username: str) -> SimpleNamespace:
|
|
try:
|
|
return self.values[username]
|
|
except KeyError as exc:
|
|
raise KeyError(username) from exc
|
|
|
|
|
|
def write_vendor_sudoers(root: Path) -> Path:
|
|
path = root / "etc" / "sudoers"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
"Defaults env_reset\n"
|
|
"@includedir /etc/sudoers.d\n"
|
|
"Defaults rootpw\n"
|
|
"pi ALL=(ALL) NOPASSWD:ALL\n"
|
|
"Defaults !env_reset\n",
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
def test_new_account_gets_password_protected_full_sudo_and_locks_pi(tmp_path, monkeypatch):
|
|
main_sudoers = write_vendor_sudoers(tmp_path)
|
|
pi_home = tmp_path / "home" / "pi"
|
|
pi_home.mkdir(parents=True)
|
|
(pi_home / ".profile").write_text("profile\n", encoding="utf-8")
|
|
(pi_home / ".ssh").mkdir()
|
|
monkeypatch.setattr(provision_device, "pwd", FakePwd({"pi": SimpleNamespace(pw_dir=str(pi_home))}))
|
|
monkeypatch.setattr(
|
|
provision_device.subprocess,
|
|
"check_output",
|
|
lambda args, text: "pi sudo adm\n" if args == ["id", "-nG", "pi"] else "",
|
|
)
|
|
calls: list[tuple[list[str], str | None]] = []
|
|
|
|
def fake_run(args: list[str], *, input_text: str | None = None) -> None:
|
|
calls.append((args, input_text))
|
|
if args[0] == "useradd":
|
|
(tmp_path / "home" / "example").mkdir(parents=True)
|
|
|
|
monkeypatch.setattr(provision_device, "run", fake_run)
|
|
provision_device.configure_account(account_config(), root=tmp_path)
|
|
|
|
assert (["useradd", "--create-home", "--shell", "/bin/bash", "--groups", "sudo,adm", "example"], None) in calls
|
|
assert (["chpasswd"], "example:example-password\n") in calls
|
|
assert (["usermod", "--append", "--groups", "sudo", "example"], None) in calls
|
|
assert (["passwd", "--lock", "pi"], None) in calls
|
|
assert any(args[:2] == ["visudo", "-cf"] for args, _ in calls)
|
|
sudoers = tmp_path / "etc" / "sudoers.d" / "90-matrix-screen-controller-account"
|
|
assert sudoers.read_text(encoding="utf-8") == "example ALL=(ALL:ALL) ALL\n"
|
|
assert main_sudoers.read_text(encoding="utf-8") == (
|
|
"Defaults env_reset\n@includedir /etc/sudoers.d\nDefaults !env_reset\n"
|
|
)
|
|
if os.name != "nt":
|
|
assert stat.S_IMODE(sudoers.stat().st_mode) == 0o440
|
|
assert (tmp_path / "home" / "example" / ".profile").is_file()
|
|
assert not (tmp_path / "home" / "example" / ".ssh").exists()
|
|
|
|
|
|
def test_existing_pi_account_is_updated_without_locking_it(tmp_path, monkeypatch):
|
|
main_sudoers = write_vendor_sudoers(tmp_path)
|
|
monkeypatch.setattr(
|
|
provision_device,
|
|
"pwd",
|
|
FakePwd({"pi": SimpleNamespace(pw_dir=str(tmp_path / "home" / "pi"))}),
|
|
)
|
|
calls: list[tuple[list[str], str | None]] = []
|
|
monkeypatch.setattr(
|
|
provision_device,
|
|
"run",
|
|
lambda args, input_text=None: calls.append((args, input_text)),
|
|
)
|
|
provision_device.configure_account(account_config("pi"), root=tmp_path)
|
|
|
|
assert not any(args[0] == "useradd" for args, _ in calls)
|
|
assert (["usermod", "--append", "--groups", "sudo", "pi"], None) in calls
|
|
assert not any(args == ["passwd", "--lock", "pi"] for args, _ in calls)
|
|
assert (tmp_path / "etc" / "sudoers.d" / "90-matrix-screen-controller-account").read_text(
|
|
encoding="utf-8"
|
|
) == "pi ALL=(ALL:ALL) ALL\n"
|
|
assert "rootpw" not in main_sudoers.read_text(encoding="utf-8")
|
|
assert "NOPASSWD" not in main_sudoers.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_invalid_sudoers_is_not_installed(tmp_path, monkeypatch):
|
|
main_sudoers = write_vendor_sudoers(tmp_path)
|
|
original = main_sudoers.read_text(encoding="utf-8")
|
|
monkeypatch.setattr(
|
|
provision_device,
|
|
"pwd",
|
|
FakePwd({"example": SimpleNamespace(pw_dir=str(tmp_path / "home" / "example"))}),
|
|
)
|
|
|
|
def fake_run(args: list[str], *, input_text: str | None = None) -> None:
|
|
if args[0] == "visudo":
|
|
raise subprocess.CalledProcessError(1, args)
|
|
|
|
monkeypatch.setattr(provision_device, "run", fake_run)
|
|
with pytest.raises(subprocess.CalledProcessError):
|
|
provision_device.configure_account(account_config(), root=tmp_path)
|
|
assert not (tmp_path / "etc" / "sudoers.d" / "90-matrix-screen-controller-account").exists()
|
|
assert main_sudoers.read_text(encoding="utf-8") == original
|
|
|
|
|
|
def test_identity_reset_updates_hostname_mapping_and_unique_identity_files(tmp_path, monkeypatch):
|
|
hosts = tmp_path / "etc" / "hosts"
|
|
hosts.parent.mkdir(parents=True)
|
|
hosts.write_text("127.0.0.1\tlocalhost\n127.0.1.1\twalnutpi\n", encoding="utf-8")
|
|
machine_id = tmp_path / "etc" / "machine-id"
|
|
machine_id.write_text("old-machine-id\n", encoding="ascii")
|
|
ssh = tmp_path / "etc" / "ssh"
|
|
ssh.mkdir()
|
|
(ssh / "ssh_host_ed25519_key").write_text("old-key", encoding="ascii")
|
|
dbus = tmp_path / "var" / "lib" / "dbus"
|
|
dbus.mkdir(parents=True)
|
|
(dbus / "machine-id").write_text("old-machine-id\n", encoding="ascii")
|
|
calls: list[list[str]] = []
|
|
monkeypatch.setattr(provision_device, "run", lambda args, input_text=None: calls.append(args))
|
|
|
|
provision_device.reset_identity(root=tmp_path)
|
|
|
|
assert hosts.read_text(encoding="utf-8") == "127.0.0.1\tlocalhost\n127.0.1.1\tmatrix-screen\n"
|
|
assert machine_id.read_text(encoding="ascii") == ""
|
|
assert not (ssh / "ssh_host_ed25519_key").exists()
|
|
assert not (dbus / "machine-id").exists()
|
|
assert ["hostnamectl", "set-hostname", "matrix-screen"] in calls
|
|
|
|
|
|
def _effective_ssh(**overrides: str) -> str:
|
|
values = {**provision_device.SSH_EXPECTED, **overrides}
|
|
return "\n".join(f"{key} {value}" for key, value in values.items()) + "\n"
|
|
|
|
|
|
def test_ssh_policy_is_validated_then_enabled_without_starting(tmp_path, monkeypatch):
|
|
calls: list[list[str]] = []
|
|
monkeypatch.setattr(provision_device, "run", lambda args, input_text=None: calls.append(args))
|
|
monkeypatch.setattr(
|
|
provision_device.subprocess,
|
|
"check_output",
|
|
lambda args, text: _effective_ssh(),
|
|
)
|
|
provision_device.configure_ssh(root=tmp_path)
|
|
|
|
installed = tmp_path / "etc" / "ssh" / "sshd_config.d" / "10-walnutpi-screen-hardening.conf"
|
|
assert installed.read_text(encoding="utf-8") == provision_device.SSH_POLICY_SOURCE.read_text(encoding="utf-8")
|
|
runtime = tmp_path / "run" / "sshd"
|
|
assert runtime.is_dir()
|
|
if os.name != "nt":
|
|
assert stat.S_IMODE(installed.stat().st_mode) == 0o644
|
|
assert stat.S_IMODE(runtime.stat().st_mode) == 0o755
|
|
assert calls == [
|
|
["/usr/sbin/sshd", "-t"],
|
|
["systemctl", "unmask", "ssh.service"],
|
|
["systemctl", "enable", "ssh.service"],
|
|
]
|
|
assert all("--now" not in call and "start" not in call for call in calls)
|
|
|
|
|
|
def test_ssh_effective_policy_mismatch_fails_before_service_enable(tmp_path, monkeypatch):
|
|
calls: list[list[str]] = []
|
|
monkeypatch.setattr(provision_device, "run", lambda args, input_text=None: calls.append(args))
|
|
monkeypatch.setattr(
|
|
provision_device.subprocess,
|
|
"check_output",
|
|
lambda args, text: _effective_ssh(passwordauthentication="no"),
|
|
)
|
|
with pytest.raises(RuntimeError, match="effective SSH policy is invalid"):
|
|
provision_device.configure_ssh(root=tmp_path)
|
|
assert calls == [["/usr/sbin/sshd", "-t"]]
|