mteehan
75b86fd60d
acme:
- acme.sh chmods its tree to owner-only (700/600) every run, which
broke the two-user model: a tree left owner-only by one user made
every acme.sh call of the other exit 2
- normalize_acme_home() reopens group access (sudo chmod g+rwX,
files only — setgid dirs trip RestrictSUIDSGID); _run_acme_preflight
is the choke point before every daemon acme.sh call + startup
- acme service now runs as the daemon user; --log persists the raw CA
transcript; SYS_LOG=6 journals manual issue/renew runs
- timer daily-only: two runs/day landed inside ZeroSSL's 24h
validation backoff (Retry-After: 86400) — a permanent renewal lockout
- _collect_acme no longer raises on cert-list failure; reports
status.error (AcmeState.status) so the certs page can surface it
firewall: re-stamp the applied baseline on live zone mutations
(interfaces/services/rich-rules/masquerade/forward-ports) so cancel-all
reverts to post-mutation state, not a stale install-era snapshot;
set_masquerade syncs the declarative config for existing zones;
add_forward_port records toaddr only with toport
status: apply-all accepts {"force": true} (forwarded to the firewall
apply only); ApplyConfirm force checkbox; applyResultToasts() — the
errors map wins over the 200; ActionButton checks errors before the
success toast; dashboard uses ApplyConfirm
system_import: drift re-imports carry the existing apply-meta; first
import stamps the adopted content as applied (it is the running state)
— no phantom pending changes
nginx: get_config only re-saves when migration actually changed the
config (no more owner/mtime churn on every read)
install: repair mis-owned top-level system dirs (tmpfiles
unsafe-path-transition), warn with a full-repair command for deeper
mis-ownership
daemon/server: loop.get_exception_handler() (aiohttp API fix)
tests: 888 pytest + 24 node passing; ruff clean
783 lines
31 KiB
Python
783 lines
31 KiB
Python
"""Tests for lib/system_import module."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from lib import system_import
|
|
from lib.common import _APPLY_HASH_KEY, _LAST_APPLIED_CONFIG_KEY, config_hash, save_json
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_project(tmp_path):
|
|
"""Patch all module-level path constants to tmp_path subdirs."""
|
|
originals = {
|
|
"PROJECT_DIR": system_import.PROJECT_DIR,
|
|
"DNSMASQ_CONF": system_import.DNSMASQ_CONF,
|
|
"WG_CONF": system_import.WG_CONF,
|
|
"NETWORKD_DIR": system_import.NETWORKD_DIR,
|
|
"NGINX_SITES_DIR": system_import.NGINX_SITES_DIR,
|
|
}
|
|
system_import.PROJECT_DIR = tmp_path
|
|
system_import.DNSMASQ_CONF = tmp_path / "etc" / "dnsmasq.d" / "vacuum-wall.conf"
|
|
system_import.WG_CONF = tmp_path / "etc" / "wireguard" / "wg0.conf"
|
|
system_import.NETWORKD_DIR = tmp_path / "etc" / "systemd" / "network"
|
|
system_import.NGINX_SITES_DIR = tmp_path / "data" / "nginx" / "sites-enabled"
|
|
yield tmp_path
|
|
system_import.PROJECT_DIR = originals["PROJECT_DIR"]
|
|
system_import.DNSMASQ_CONF = originals["DNSMASQ_CONF"]
|
|
system_import.WG_CONF = originals["WG_CONF"]
|
|
system_import.NETWORKD_DIR = originals["NETWORKD_DIR"]
|
|
system_import.NGINX_SITES_DIR = originals["NGINX_SITES_DIR"]
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
# Dnsmasq
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestImportDnsmasq:
|
|
def _write_conf(self, tmp_path, content: str) -> Path:
|
|
p = tmp_path / "etc" / "dnsmasq.d"
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
(p / "vacuum-wall.conf").write_text(content)
|
|
return p / "vacuum-wall.conf"
|
|
|
|
def _read_json(self, tmp_path) -> dict:
|
|
p = tmp_path / "config" / "dnsmasq" / "config.json"
|
|
return json.loads(p.read_text()) if p.exists() else {}
|
|
|
|
def test_no_conf_file(self, temp_project):
|
|
assert not system_import.import_dnsmasq()
|
|
|
|
def test_no_markers(self, temp_project, tmp_path):
|
|
self._write_conf(tmp_path, "# some random config\nserver=1.1.1.1\n")
|
|
assert not system_import.import_dnsmasq()
|
|
|
|
def test_empty_managed_block(self, temp_project, tmp_path):
|
|
self._write_conf(tmp_path, f"{system_import.DNSTART}\n{system_import.DNEND}")
|
|
assert system_import.import_dnsmasq()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg["dns"]["upstreams"] == []
|
|
assert cfg["dhcp"]["ranges"] == []
|
|
|
|
def test_upstreams_only(self, temp_project, tmp_path):
|
|
conf = (
|
|
f"{system_import.DNSTART}\n"
|
|
"server=8.8.8.8\n"
|
|
"server=1.1.1.1\n"
|
|
f"{system_import.DNEND}"
|
|
)
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_dnsmasq()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg["dns"]["upstreams"] == ["8.8.8.8", "1.1.1.1"]
|
|
|
|
def test_full_config(self, temp_project, tmp_path):
|
|
conf = (
|
|
f"{system_import.DNSTART}\n"
|
|
"server=8.8.8.8\n"
|
|
"server=1.1.1.1\n"
|
|
"domain=lan\n"
|
|
"expand-hosts\n"
|
|
"dhcp-range=set:eth1,192.168.2.100,192.168.2.200,12h\n"
|
|
"dhcp-option=tag:eth1,3,192.168.2.1\n"
|
|
"dhcp-option=tag:eth1,6,192.168.2.1\n"
|
|
"dhcp-host=aa:bb:cc:dd:ee:ff,192.168.2.50,printer\n"
|
|
"addr/nas.lan/192.168.2.10\n"
|
|
f"{system_import.DNEND}"
|
|
)
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_dnsmasq()
|
|
cfg = self._read_json(tmp_path)
|
|
|
|
assert cfg["dns"]["upstreams"] == ["8.8.8.8", "1.1.1.1"]
|
|
assert cfg["dns"]["domain"] == "lan"
|
|
assert len(cfg["dhcp"]["ranges"]) == 1
|
|
rng = cfg["dhcp"]["ranges"][0]
|
|
assert rng["interface"] == "eth1"
|
|
assert rng["start"] == "192.168.2.100"
|
|
assert rng["end"] == "192.168.2.200"
|
|
assert rng["lease_time"] == "12h"
|
|
assert rng["gateway"] == "192.168.2.1"
|
|
assert rng["dns"] == "192.168.2.1"
|
|
assert len(cfg["dhcp"]["static_leases"]) == 1
|
|
lease = cfg["dhcp"]["static_leases"][0]
|
|
assert lease["mac"] == "aa:bb:cc:dd:ee:ff"
|
|
assert lease["ip"] == "192.168.2.50"
|
|
assert lease["hostname"] == "printer"
|
|
assert len(cfg["dns"]["custom_records"]) == 1
|
|
assert cfg["dns"]["custom_records"][0] == {
|
|
"name": "nas.lan",
|
|
"address": "192.168.2.10",
|
|
}
|
|
|
|
def test_no_resolv_resets_upstreams(self, temp_project, tmp_path):
|
|
conf = f"{system_import.DNSTART}\nno-resolv\n{system_import.DNEND}"
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_dnsmasq()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg["dns"]["upstreams"] == []
|
|
|
|
def test_idempotent(self, temp_project, tmp_path):
|
|
conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}"
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_dnsmasq()
|
|
assert not system_import.import_dnsmasq()
|
|
|
|
def test_parse_error_returns_false(self, temp_project, tmp_path):
|
|
# Conf with markers — parses fine, so this tests the exception handler
|
|
# by mocking _parse_dnsmasq_block to raise
|
|
conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}"
|
|
self._write_conf(tmp_path, conf)
|
|
with patch(
|
|
"lib.system_import._parse_dnsmasq_block", side_effect=ValueError("bad")
|
|
):
|
|
assert not system_import.import_dnsmasq()
|
|
|
|
def test_preserves_apply_meta_on_drift(self, temp_project, tmp_path):
|
|
# Existing config differs from the live conf and carries apply
|
|
# bookkeeping — the rewrite must keep the baseline so pending
|
|
# detection and cancel-all survive daemon restarts.
|
|
conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}"
|
|
self._write_conf(tmp_path, conf)
|
|
cfg_path = tmp_path / "config" / "dnsmasq"
|
|
cfg_path.mkdir(parents=True, exist_ok=True)
|
|
baseline = {"dns": {"upstreams": ["1.1.1.1"]}}
|
|
save_json(
|
|
cfg_path / "config.json",
|
|
{
|
|
**baseline,
|
|
_APPLY_HASH_KEY: "old-hash",
|
|
_LAST_APPLIED_CONFIG_KEY: baseline,
|
|
},
|
|
)
|
|
assert system_import.import_dnsmasq()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg[_APPLY_HASH_KEY] == "old-hash"
|
|
assert cfg[_LAST_APPLIED_CONFIG_KEY] == baseline
|
|
assert cfg["dns"]["upstreams"] == ["8.8.8.8"]
|
|
|
|
def test_stamps_applied_on_first_import(self, temp_project, tmp_path):
|
|
# No config file yet: the imported content is the running state,
|
|
# so it must be stamped as applied (no phantom pending changes).
|
|
conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}"
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_dnsmasq()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg[_APPLY_HASH_KEY] == config_hash(cfg)
|
|
assert cfg[_LAST_APPLIED_CONFIG_KEY]["dns"]["upstreams"] == ["8.8.8.8"]
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
# WireGuard
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestImportWireguard:
|
|
def _write_conf(self, tmp_path, content: str) -> Path:
|
|
p = tmp_path / "etc" / "wireguard"
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
(p / "wg0.conf").write_text(content)
|
|
return p / "wg0.conf"
|
|
|
|
def _read_json(self, tmp_path) -> dict:
|
|
p = tmp_path / "config" / "wireguard" / "config.json"
|
|
return json.loads(p.read_text()) if p.exists() else {}
|
|
|
|
def test_no_conf_file(self, temp_project):
|
|
assert not system_import.import_wireguard()
|
|
|
|
def test_basic_interface(self, temp_project, tmp_path):
|
|
conf = (
|
|
"[Interface]\n"
|
|
" PrivateKey = abc123\n"
|
|
" Address = 10.137.0.1/24\n"
|
|
" ListenPort = 51820\n"
|
|
)
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_wireguard()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg["interface"]["private_key"] == "abc123"
|
|
assert cfg["interface"]["addresses"] == ["10.137.0.1/24"]
|
|
assert cfg["interface"]["listen_port"] == 51820
|
|
assert cfg["interface"]["name"] == "wg0"
|
|
assert cfg["peers"] == {}
|
|
|
|
def test_full_with_peers(self, temp_project, tmp_path):
|
|
conf = (
|
|
"[Interface]\n"
|
|
" PrivateKey = srv-priv\n"
|
|
" Address = 10.137.0.1/24\n"
|
|
" ListenPort = 51820\n"
|
|
" PostUp = iptables -I FORWARD -i wg0 -j ACCEPT\n"
|
|
" PostDown = iptables -D FORWARD -i wg0 -j ACCEPT\n"
|
|
"\n"
|
|
"[Peer] # alice\n"
|
|
" PublicKey = alice-pub\n"
|
|
" Endpoint = 203.0.113.1:51820\n"
|
|
" AllowedIPs = 0.0.0.0/0\n"
|
|
" PersistentKeepalive = 25\n"
|
|
"\n"
|
|
"[Peer] # bob\n"
|
|
" PublicKey = bob-pub\n"
|
|
" AllowedIPs = 10.0.0.0/8,172.16.0.0/12\n"
|
|
)
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_wireguard()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg["interface"]["post_up"] == "iptables -I FORWARD -i wg0 -j ACCEPT"
|
|
assert cfg["interface"]["post_down"] == "iptables -D FORWARD -i wg0 -j ACCEPT"
|
|
assert "alice" in cfg["peers"]
|
|
assert cfg["peers"]["alice"]["public_key"] == "alice-pub"
|
|
assert cfg["peers"]["alice"]["endpoint"] == "203.0.113.1:51820"
|
|
assert cfg["peers"]["alice"]["allowed_ips"] == ["0.0.0.0/0"]
|
|
assert cfg["peers"]["alice"]["persistent_keepalive"] == 25
|
|
assert "bob" in cfg["peers"]
|
|
assert cfg["peers"]["bob"]["allowed_ips"] == ["10.0.0.0/8", "172.16.0.0/12"]
|
|
|
|
def test_peer_without_name_uses_pubkey(self, temp_project, tmp_path):
|
|
conf = (
|
|
"[Interface]\n"
|
|
" PrivateKey = srv-priv\n"
|
|
" Address = 10.137.0.1/24\n"
|
|
" ListenPort = 51820\n"
|
|
"\n"
|
|
"[Peer]\n"
|
|
" PublicKey = anon-pub\n"
|
|
" AllowedIPs = 0.0.0.0/0\n"
|
|
)
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_wireguard()
|
|
cfg = self._read_json(tmp_path)
|
|
assert "anon-pub" in cfg["peers"]
|
|
|
|
def test_idempotent(self, temp_project, tmp_path):
|
|
conf = (
|
|
"[Interface]\n"
|
|
" PrivateKey = abc123\n"
|
|
" Address = 10.137.0.1/24\n"
|
|
" ListenPort = 51820\n"
|
|
)
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_wireguard()
|
|
assert not system_import.import_wireguard()
|
|
|
|
def test_preserves_apply_meta_on_drift(self, temp_project, tmp_path):
|
|
conf = (
|
|
"[Interface]\n"
|
|
" PrivateKey = abc123\n"
|
|
" Address = 10.137.0.1/24\n"
|
|
" ListenPort = 51820\n"
|
|
)
|
|
self._write_conf(tmp_path, conf)
|
|
cfg_path = tmp_path / "config" / "wireguard"
|
|
cfg_path.mkdir(parents=True, exist_ok=True)
|
|
baseline = {"interface": {"listen_port": 51821}, "peers": {}}
|
|
save_json(
|
|
cfg_path / "config.json",
|
|
{
|
|
**baseline,
|
|
_APPLY_HASH_KEY: "old-hash",
|
|
_LAST_APPLIED_CONFIG_KEY: baseline,
|
|
},
|
|
)
|
|
assert system_import.import_wireguard()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg[_APPLY_HASH_KEY] == "old-hash"
|
|
assert cfg[_LAST_APPLIED_CONFIG_KEY] == baseline
|
|
assert cfg["interface"]["listen_port"] == 51820
|
|
|
|
def test_stamps_applied_on_first_import(self, temp_project, tmp_path):
|
|
conf = (
|
|
"[Interface]\n"
|
|
" PrivateKey = abc123\n"
|
|
" Address = 10.137.0.1/24\n"
|
|
" ListenPort = 51820\n"
|
|
)
|
|
self._write_conf(tmp_path, conf)
|
|
assert system_import.import_wireguard()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg[_APPLY_HASH_KEY] == config_hash(cfg)
|
|
assert cfg[_LAST_APPLIED_CONFIG_KEY]["interface"]["private_key"] == "abc123"
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
# Networkd
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestImportNetworkd:
|
|
def _write_network(self, tmp_path, name: str, content: str) -> Path:
|
|
p = tmp_path / "etc" / "systemd" / "network"
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
file_path = p / f"99-{name}.network"
|
|
file_path.write_text(content)
|
|
return file_path
|
|
|
|
def _read_json(self, tmp_path) -> dict:
|
|
p = tmp_path / "config" / "network" / "config.json"
|
|
return json.loads(p.read_text()) if p.exists() else {}
|
|
|
|
def test_no_network_dir(self, temp_project):
|
|
assert not system_import.import_networkd()
|
|
|
|
def test_no_files(self, temp_project, tmp_path):
|
|
(tmp_path / "etc" / "systemd" / "network").mkdir(parents=True, exist_ok=True)
|
|
assert not system_import.import_networkd()
|
|
|
|
def test_basic_interface(self, temp_project, tmp_path):
|
|
conf = (
|
|
"[Match]\n"
|
|
"Name=eth0\n"
|
|
"\n"
|
|
"[Network]\n"
|
|
"DHCP=no\n"
|
|
"Addresses=192.168.1.1/24\n"
|
|
"Gateway=192.168.1.254\n"
|
|
"DNS=8.8.8.8\n"
|
|
"DNS=1.1.1.1\n"
|
|
)
|
|
self._write_network(tmp_path, "eth0", conf)
|
|
assert system_import.import_networkd()
|
|
cfg = self._read_json(tmp_path)
|
|
eth0 = cfg["interfaces"]["eth0"]
|
|
assert eth0["dhcp"] == "no"
|
|
assert eth0["gateway"] == "192.168.1.254"
|
|
assert eth0["dns"] == ["8.8.8.8", "1.1.1.1"]
|
|
|
|
def test_multiple_interfaces(self, temp_project, tmp_path):
|
|
self._write_network(
|
|
tmp_path, "eth0", "[Match]\nName=eth0\n\n[Network]\nDHCP=yes\n"
|
|
)
|
|
self._write_network(
|
|
tmp_path, "eth1", "[Match]\nName=eth1\n\n[Network]\nDHCP=no\n"
|
|
)
|
|
assert system_import.import_networkd()
|
|
cfg = self._read_json(tmp_path)
|
|
assert "eth0" in cfg["interfaces"]
|
|
assert "eth1" in cfg["interfaces"]
|
|
assert cfg["interfaces"]["eth0"]["dhcp"] == "yes"
|
|
assert cfg["interfaces"]["eth1"]["dhcp"] == "no"
|
|
|
|
def test_preserves_existing_interfaces(self, temp_project, tmp_path):
|
|
# Pre-existing JSON has eth2 with no .network file
|
|
cfg_path = tmp_path / "config" / "network"
|
|
cfg_path.mkdir(parents=True, exist_ok=True)
|
|
save_json(cfg_path / "config.json", {"interfaces": {"eth2": {"dhcp": "no"}}})
|
|
self._write_network(
|
|
tmp_path, "eth0", "[Match]\nName=eth0\n\n[Network]\nDHCP=yes\n"
|
|
)
|
|
assert system_import.import_networkd()
|
|
cfg = self._read_json(tmp_path)
|
|
assert "eth0" in cfg["interfaces"]
|
|
assert "eth2" in cfg["interfaces"]
|
|
|
|
def test_idempotent(self, temp_project, tmp_path):
|
|
self._write_network(
|
|
tmp_path, "eth0", "[Match]\nName=eth0\n\n[Network]\nDHCP=yes\n"
|
|
)
|
|
assert system_import.import_networkd()
|
|
assert not system_import.import_networkd()
|
|
|
|
def test_address_section_parsed(self, temp_project, tmp_path):
|
|
conf = (
|
|
"[Match]\n"
|
|
"Name=eth0\n"
|
|
"\n"
|
|
"[Network]\n"
|
|
"DHCP=no\n"
|
|
"\n"
|
|
"[Address]\n"
|
|
"Address=192.168.1.1/24\n"
|
|
)
|
|
self._write_network(tmp_path, "eth0", conf)
|
|
assert system_import.import_networkd()
|
|
cfg = self._read_json(tmp_path)
|
|
eth0 = cfg["interfaces"]["eth0"]
|
|
assert "192.168.1.1/24" in eth0.get("addresses", [])
|
|
|
|
def test_route_section_parsed(self, temp_project, tmp_path):
|
|
conf = (
|
|
"[Match]\n"
|
|
"Name=eth0\n"
|
|
"\n"
|
|
"[Network]\n"
|
|
"DHCP=no\n"
|
|
"\n"
|
|
"[Route]\n"
|
|
"Destination=10.0.0.0/8\n"
|
|
"Gateway=192.168.1.254\n"
|
|
"Metric=100\n"
|
|
)
|
|
self._write_network(tmp_path, "eth0", conf)
|
|
assert system_import.import_networkd()
|
|
cfg = self._read_json(tmp_path)
|
|
eth0 = cfg["interfaces"]["eth0"]
|
|
assert len(eth0.get("routes", [])) == 1
|
|
route = eth0["routes"][0]
|
|
assert route["destination"] == "10.0.0.0/8"
|
|
assert route["gateway"] == "192.168.1.254"
|
|
assert route["metric"] == 100
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
# Nginx
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestImportNginx:
|
|
def _write_site(self, tmp_path, domain: str, content: str) -> Path:
|
|
p = tmp_path / "data" / "nginx" / "sites-enabled"
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
file_path = p / f"{domain}.conf"
|
|
file_path.write_text(content)
|
|
return file_path
|
|
|
|
def _read_json(self, tmp_path) -> dict:
|
|
p = tmp_path / "config" / "nginx" / "config.json"
|
|
return json.loads(p.read_text()) if p.exists() else {}
|
|
|
|
def test_no_sites_dir(self, temp_project):
|
|
assert not system_import.import_nginx()
|
|
|
|
def test_no_files(self, temp_project, tmp_path):
|
|
(tmp_path / "data" / "nginx" / "sites-enabled").mkdir(
|
|
parents=True, exist_ok=True
|
|
)
|
|
assert not system_import.import_nginx()
|
|
|
|
def test_acme_challenge_skipped(self, temp_project, tmp_path):
|
|
(tmp_path / "data" / "nginx" / "sites-enabled").mkdir(
|
|
parents=True, exist_ok=True
|
|
)
|
|
(
|
|
tmp_path / "data" / "nginx" / "sites-enabled" / "_acme-challenge.conf"
|
|
).write_text("# stuff\n")
|
|
assert not system_import.import_nginx()
|
|
|
|
def test_unrecognized_file_skipped(self, temp_project, tmp_path):
|
|
self._write_site(tmp_path, "my-site", "# some random nginx config\nserver {}\n")
|
|
assert not system_import.import_nginx()
|
|
|
|
def test_basic_site(self, temp_project, tmp_path):
|
|
conf = (
|
|
"# Auto-generated by Vacuum Wall — do not edit manually\n"
|
|
"# Domain: example.com\n"
|
|
"\n"
|
|
"server {\n"
|
|
" listen 80;\n"
|
|
" listen [::]:80;\n"
|
|
" server_name example.com;\n"
|
|
" return 301 https://$host$request_uri;\n"
|
|
"}\n"
|
|
"\n"
|
|
"server {\n"
|
|
" listen 443 ssl;\n"
|
|
" listen [::]:443 ssl;\n"
|
|
" server_name example.com;\n"
|
|
"\n"
|
|
" ssl_certificate /home/wall/vacuum-wall/data/acme/example.com/fullchain.cer;\n"
|
|
" ssl_certificate_key /home/wall/vacuum-wall/data/acme/example.com/example.com.key;\n"
|
|
"\n"
|
|
" # / -> 192.168.2.50:8080\n"
|
|
" location / {\n"
|
|
" auth_basic off;\n"
|
|
" proxy_pass http://192.168.2.50:8080;\n"
|
|
" }\n"
|
|
"}\n"
|
|
)
|
|
self._write_site(tmp_path, "example.com", conf)
|
|
assert system_import.import_nginx()
|
|
cfg = self._read_json(tmp_path)
|
|
assert "example.com" in cfg["domains"]
|
|
dom = cfg["domains"]["example.com"]
|
|
assert dom["force_ssl"] is True
|
|
assert dom["cert"] == "acme"
|
|
assert "/" in dom["paths"]
|
|
assert dom["paths"]["/"]["backend"]["host"] == "192.168.2.50"
|
|
assert dom["paths"]["/"]["backend"]["port"] == 8080
|
|
|
|
def test_websocket_path(self, temp_project, tmp_path):
|
|
conf = (
|
|
"# Auto-generated by Vacuum Wall — do not edit manually\n"
|
|
"# Domain: example.com\n"
|
|
"\n"
|
|
"server {\n"
|
|
" listen 443 ssl;\n"
|
|
" server_name example.com;\n"
|
|
"\n"
|
|
" ssl_certificate /data/certs/example.com.crt;\n"
|
|
" ssl_certificate_key /data/certs/example.com.key;\n"
|
|
"\n"
|
|
" # / -> 127.0.0.1:9090\n"
|
|
" location / {\n"
|
|
" auth_basic off;\n"
|
|
" proxy_pass http://127.0.0.1:9090;\n"
|
|
" }\n"
|
|
"\n"
|
|
" # /ws -> 127.0.0.1:9091 (WebSocket)\n"
|
|
" location /ws {\n"
|
|
" auth_basic off;\n"
|
|
" proxy_pass http://127.0.0.1:9091;\n"
|
|
" }\n"
|
|
"}\n"
|
|
)
|
|
self._write_site(tmp_path, "example.com", conf)
|
|
assert system_import.import_nginx()
|
|
cfg = self._read_json(tmp_path)
|
|
dom = cfg["domains"]["example.com"]
|
|
assert dom["cert"] == "selfsigned"
|
|
assert dom["paths"]["/ws"]["is_websocket"] is True
|
|
assert dom["paths"]["/ws"]["backend"]["port"] == 9091
|
|
|
|
def test_idempotent(self, temp_project, tmp_path):
|
|
conf = (
|
|
"# Auto-generated by Vacuum Wall — do not edit manually\n"
|
|
"server {\n"
|
|
" listen 443 ssl;\n"
|
|
" server_name example.com;\n"
|
|
" ssl_certificate /data/acme/example.com/fullchain.cer;\n"
|
|
" ssl_certificate_key /data/acme/example.com/example.com.key;\n"
|
|
" # / -> 127.0.0.1:9090\n"
|
|
" location / {\n"
|
|
" auth_basic off;\n"
|
|
" proxy_pass http://127.0.0.1:9090;\n"
|
|
" }\n"
|
|
"}\n"
|
|
)
|
|
self._write_site(tmp_path, "example.com", conf)
|
|
assert system_import.import_nginx()
|
|
assert not system_import.import_nginx()
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
# Firewall
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
FIREWALL_ZONES_OUTPUT = (
|
|
"public (active)\n"
|
|
" target: default\n"
|
|
" interfaces: eth0 eth1\n"
|
|
" sources: \n"
|
|
" services: dhcpv6-cidr dns mdns ssh\n"
|
|
" ports: \n"
|
|
" protocols: \n"
|
|
" forward-ports: \n"
|
|
" source-ports: \n"
|
|
" icmp-blocks: \n"
|
|
" rich rules: \n"
|
|
"\n"
|
|
"internal (active)\n"
|
|
" target: DEFAULT\n"
|
|
" interfaces: eth2\n"
|
|
" sources: \n"
|
|
" services: dhcpv6-cidr dns mdns samba-client ssh\n"
|
|
" ports: \n"
|
|
" protocols: \n"
|
|
" forward-ports: \n"
|
|
" source-ports: \n"
|
|
" icmp-blocks: \n"
|
|
" rich rules: \n"
|
|
"\n"
|
|
"dmz (active)\n"
|
|
" target: DROP\n"
|
|
" interfaces: \n"
|
|
" sources: \n"
|
|
" services: dns\n"
|
|
" ports: \n"
|
|
" protocols: \n"
|
|
" forward-ports: \n"
|
|
" source-ports: \n"
|
|
" icmp-blocks: \n"
|
|
" rich rules: \n"
|
|
)
|
|
|
|
|
|
class TestImportFirewall:
|
|
def _read_json(self, tmp_path) -> dict:
|
|
p = tmp_path / "config" / "firewall" / "config.json"
|
|
return json.loads(p.read_text()) if p.exists() else {}
|
|
|
|
def test_no_config_file_and_firewalld_down(self, temp_project, tmp_path):
|
|
with patch(
|
|
"lib.system_import.run", side_effect=RuntimeError("firewalld not running")
|
|
):
|
|
assert not system_import.import_firewall()
|
|
|
|
def test_existing_config_not_overwritten(self, temp_project, tmp_path):
|
|
cfg_path = tmp_path / "config" / "firewall"
|
|
cfg_path.mkdir(parents=True, exist_ok=True)
|
|
save_json(
|
|
cfg_path / "config.json", {"zones": {"public": {"interfaces": ["eth0"]}}}
|
|
)
|
|
assert not system_import.import_firewall()
|
|
|
|
def test_import_zones(self, temp_project, tmp_path):
|
|
with patch("lib.system_import.run", return_value=FIREWALL_ZONES_OUTPUT):
|
|
assert system_import.import_firewall()
|
|
cfg = self._read_json(tmp_path)
|
|
assert "zones" in cfg
|
|
assert "public" in cfg["zones"]
|
|
assert "internal" in cfg["zones"]
|
|
# Live target normalizes to "default" -> the target key is omitted
|
|
# (key-absence is the canonical "unmanaged" notation, WI-2).
|
|
assert "target" not in cfg["zones"]["public"]
|
|
assert cfg["zones"]["public"]["interfaces"] == ["eth0", "eth1"]
|
|
assert cfg["zones"]["public"]["services"] == [
|
|
"dhcpv6-cidr",
|
|
"dns",
|
|
"mdns",
|
|
"ssh",
|
|
]
|
|
assert "target" not in cfg["zones"]["internal"]
|
|
assert cfg["zones"]["internal"]["interfaces"] == ["eth2"]
|
|
|
|
def test_import_keeps_nondefault_target(self, temp_project, tmp_path):
|
|
# A zone with interfaces and a non-default live target keeps its
|
|
# explicit target key (ACCEPT/DROP/REJECT remain fully managed).
|
|
output = (
|
|
"trusted (active)\n"
|
|
" target: ACCEPT\n"
|
|
" interfaces: eth3\n"
|
|
" sources: \n"
|
|
" services: \n"
|
|
" ports: \n"
|
|
" protocols: \n"
|
|
" forward-ports: \n"
|
|
" source-ports: \n"
|
|
" icmp-blocks: \n"
|
|
" rich rules: \n"
|
|
)
|
|
with patch("lib.system_import.run", return_value=output):
|
|
assert system_import.import_firewall()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg["zones"]["trusted"]["target"] == "ACCEPT"
|
|
assert cfg["zones"]["trusted"]["interfaces"] == ["eth3"]
|
|
|
|
def test_empty_interface_zones_skipped(self, temp_project, tmp_path):
|
|
with patch("lib.system_import.run", return_value=FIREWALL_ZONES_OUTPUT):
|
|
assert system_import.import_firewall()
|
|
cfg = self._read_json(tmp_path)
|
|
assert "dmz" not in cfg["zones"]
|
|
|
|
def test_import_stamps_applied(self, temp_project, tmp_path):
|
|
# Fresh import adopts the live firewalld state, which is by
|
|
# definition the applied state — the file must carry a baseline.
|
|
with patch("lib.system_import.run", return_value=FIREWALL_ZONES_OUTPUT):
|
|
assert system_import.import_firewall()
|
|
cfg = self._read_json(tmp_path)
|
|
assert cfg[_APPLY_HASH_KEY] == config_hash(cfg)
|
|
assert cfg[_LAST_APPLIED_CONFIG_KEY]["zones"]["public"]["interfaces"] == [
|
|
"eth0",
|
|
"eth1",
|
|
]
|
|
|
|
def test_parse_error_returns_false(self, temp_project, tmp_path):
|
|
with patch("lib.system_import.run", return_value="garbage with no valid zones"):
|
|
assert not system_import.import_firewall()
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
# import_all
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestImportAll:
|
|
def test_all_missing(self, temp_project, tmp_path):
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
mock_run = MagicMock(side_effect=RuntimeError("command not found"))
|
|
with patch.object(system_import, "run", mock_run):
|
|
result = system_import.import_all()
|
|
assert result == []
|
|
|
|
def test_returns_updated_subsystems(self, temp_project, tmp_path):
|
|
# Create dnsmasq conf
|
|
etc = tmp_path / "etc" / "dnsmasq.d"
|
|
etc.mkdir(parents=True, exist_ok=True)
|
|
conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}"
|
|
(etc / "vacuum-wall.conf").write_text(conf)
|
|
|
|
# Create wireguard conf
|
|
wg_etc = tmp_path / "etc" / "wireguard"
|
|
wg_etc.mkdir(parents=True, exist_ok=True)
|
|
(wg_etc / "wg0.conf").write_text(
|
|
"[Interface]\n PrivateKey = abc\n Address = 10.137.0.1/24\n ListenPort = 51820\n"
|
|
)
|
|
|
|
result = system_import.import_all()
|
|
assert "dnsmasq" in result
|
|
assert "wireguard" in result
|
|
assert "network" not in result
|
|
assert "nginx" not in result
|
|
|
|
def test_parse_error_does_not_crash(self, temp_project, tmp_path):
|
|
# Create a dnsmasq conf that will parse fine
|
|
etc = tmp_path / "etc" / "dnsmasq.d"
|
|
etc.mkdir(parents=True, exist_ok=True)
|
|
conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}"
|
|
(etc / "vacuum-wall.conf").write_text(conf)
|
|
|
|
# Make wireguard import fail
|
|
wg_etc = tmp_path / "etc" / "wireguard"
|
|
wg_etc.mkdir(parents=True, exist_ok=True)
|
|
(wg_etc / "wg0.conf").write_text("[Interface]\n")
|
|
|
|
# This should not raise, just log warning
|
|
result = system_import.import_all()
|
|
assert "dnsmasq" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
# _cfgs_equal
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestCfgsEqual:
|
|
def test_equal(self):
|
|
assert system_import._cfgs_equal({"a": 1}, {"a": 1})
|
|
|
|
def test_not_equal(self):
|
|
assert not system_import._cfgs_equal({"a": 1}, {"a": 2})
|
|
|
|
def test_ignores_applied_hash(self):
|
|
a = {"a": 1, "_last_applied_hash": "abc"}
|
|
b = {"a": 1, "_last_applied_hash": "xyz"}
|
|
assert system_import._cfgs_equal(a, b)
|
|
|
|
def test_nested(self):
|
|
a = {"dhcp": {"ranges": [{"start": "1.2.3.4"}]}}
|
|
b = {"dhcp": {"ranges": [{"start": "1.2.3.4"}]}}
|
|
assert system_import._cfgs_equal(a, b)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
# Template marker consistency
|
|
# ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestTemplateMarker:
|
|
"""Verify nginx templates contain the expected auto-generated marker."""
|
|
|
|
def test_acme_challenge_has_marker(self):
|
|
content = (
|
|
Path(__file__).resolve().parent.parent
|
|
/ "system"
|
|
/ "nginx"
|
|
/ "acme-challenge.conf"
|
|
).read_text()
|
|
assert "# Auto-generated by Vacuum Wall" in content
|
|
|
|
def test_server_block_has_marker(self):
|
|
content = (
|
|
Path(__file__).resolve().parent.parent
|
|
/ "system"
|
|
/ "nginx"
|
|
/ "server_block.conf"
|
|
).read_text()
|
|
assert "# Auto-generated by Vacuum Wall" in content
|