mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: introduce two-user daemon architecture with socket-based communication

Browse Source 
- Add daemon/ module with aiohttp server, sync client, and handler registry
- Add daemon/handlers/ for privileged operations (acme, dnsmasq, firewall, logs, nginx, wireguard)
- Add system/acme-deploy.py, vacuum-walld sudoers and systemd service
- Update API routes to use daemon client instead of lib/ directly
- Update lib/, tests/, and webui/ for new architecture
- Update docs and deployment scripts
This commit is contained in:
Mike Teehan
2026-05-27 23:38:23 +00:00
parent 5ac69dfa7e
commit 200e078bc5
39 changed files with 4671 additions and 1810 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_api.py
+380 -204
View File
@@ -1,30 +1,56 @@
"""
API integration tests — all blueprints tested via a single Flask app fixture.
Mocks daemon.client in each blueprint's module namespace to avoid needing
a running daemon.
"""
from unittest.mock import patch
from unittest.mock import patch as _patch
import pytest
from webui.api.certs import bp as certs_bp
from webui.api.dhcp import bp as dhcp_bp
from webui.api.firewall import bp
from webui.api.proxy import bp as proxy_bp
from webui.api.wireguard import bp as wg_bp
def _fw(func, **kw):
"""Patch daemon.client.{func} in the firewall blueprint namespace."""
return _patch(f"webui.api.firewall.{func}", **kw)
def _dh(func, **kw):
"""Patch daemon.client.{func} in the dhcp blueprint namespace."""
return _patch(f"webui.api.dhcp.{func}", **kw)
def _px(func, **kw):
"""Patch daemon.client.{func} in the proxy blueprint namespace."""
return _patch(f"webui.api.proxy.{func}", **kw)
def _ce(func, **kw):
"""Patch daemon.client.{func} in the certs blueprint namespace."""
return _patch(f"webui.api.certs.{func}", **kw)
def _wg(func, **kw):
"""Patch daemon.client.{func} in the wireguard blueprint namespace."""
return _patch(f"webui.api.wireguard.{func}", **kw)
@pytest.fixture
def client():
from flask import Flask
app = Flask(__name__)
from webui.api.certs import bp as certs_bp
from webui.api.dhcp import bp as dhcp_bp
from webui.api.firewall import bp as firewall_bp
from webui.api.proxy import bp as proxy_bp
from webui.api.wireguard import bp as wg_bp
app.register_blueprint(bp, url_prefix="/api/firewall")
app = Flask(__name__)
app.register_blueprint(firewall_bp, url_prefix="/api/firewall")
app.register_blueprint(dhcp_bp, url_prefix="/api/dhcp")
app.register_blueprint(proxy_bp, url_prefix="/api/proxy")
app.register_blueprint(certs_bp, url_prefix="/api/certs")
app.register_blueprint(wg_bp, url_prefix="/api/wireguard")
return app.test_client()
@@ -34,20 +60,21 @@ def client():
class TestFirewallListZones:
@patch("webui.api.firewall.get_active_zones")
@patch("webui.api.firewall.get_available_zones")
def test_success(self, mock_available, mock_active, client):
mock_active.return_value = {"public": ["eth0"]}
mock_available.return_value = ["public", "internal"]
@_fw("get")
def test_success(self, mock_get, client):
mock_get.return_value = {
"active": {"public": ["eth0"]},
"available": ["public", "internal"],
}
resp = client.get("/api/firewall/zones")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
assert "public" in data["data"]["active"]
@patch("webui.api.firewall.get_active_zones")
def test_runtime_error(self, mock_active, client):
mock_active.side_effect = RuntimeError("no sudo")
@_fw("get")
def test_runtime_error(self, mock_get, client):
mock_get.side_effect = RuntimeError("no sudo")
resp = client.get("/api/firewall/zones")
assert resp.status_code == 500
data = resp.get_json()
@@ -55,11 +82,9 @@ class TestFirewallListZones:
class TestFirewallZoneDetails:
@patch("webui.api.firewall.get_zone_info")
@patch("webui.api.firewall.get_available_zones")
def test_success(self, mock_available, mock_info, client):
mock_available.return_value = ["public", "internal"]
mock_info.return_value = {"name": "public", "services": ["ssh"]}
@_fw("get")
def test_success(self, mock_get, client):
mock_get.return_value = {"name": "public", "services": ["ssh"]}
resp = client.get("/api/firewall/zones/public")
assert resp.status_code == 200
data = resp.get_json()
@@ -67,11 +92,11 @@ class TestFirewallZoneDetails:
class TestFirewallCreateZone:
@patch("webui.api.firewall.create_zone")
@patch("webui.api.firewall.get_available_zones")
def test_success(self, mock_zones, mock_create, client):
mock_zones.return_value = ["public", "internal"]
mock_create.return_value = None
@_fw("post")
@_fw("get")
def test_success(self, mock_get, mock_post, client):
mock_get.return_value = {"active": {}, "available": ["public", "internal"]}
mock_post.return_value = {"zone": "dmz"}
resp = client.post(
"/api/firewall/zones",
json={"name": "dmz", "target": "default"},
@@ -91,25 +116,29 @@ class TestFirewallCreateZone:
class TestFirewallDeleteZone:
@patch("webui.api.firewall.delete_zone")
@patch("webui.api.firewall.get_available_zones")
def test_success(self, mock_zones, mock_delete, client):
mock_zones.return_value = ["public", "dmz"]
mock_delete.return_value = None
@_fw("delete")
def test_success(self, mock_delete, client):
mock_delete.return_value = {"zone": "dmz"}
resp = client.delete("/api/firewall/zones/dmz")
assert resp.status_code == 200
@patch("webui.api.firewall.get_available_zones")
def test_not_found(self, mock_zones, client):
mock_zones.return_value = ["public"]
@_fw("delete")
def test_not_found(self, mock_delete, client):
from daemon.client import NotFound
mock_delete.side_effect = NotFound("Zone does not exist")
resp = client.delete("/api/firewall/zones/dmz")
assert resp.status_code == 404
class TestFirewallRichRules:
@patch("webui.api.firewall.add_rich_rule")
def test_add(self, mock_add, client):
mock_add.return_value = {"id": "abc123", "rule": "rule accept"}
@_fw("post")
def test_add(self, mock_post, client):
mock_post.return_value = {
"id": "abc123",
"rule": "rule accept",
"zone": "public",
}
resp = client.post(
"/api/firewall/rich-rules",
json={
@@ -126,55 +155,74 @@ class TestFirewallRichRules:
resp = client.post("/api/firewall/rich-rules", json={})
assert resp.status_code == 400
@patch("webui.api.firewall.get_rich_rules")
@patch("webui.api.firewall.get_config")
def test_list(self, mock_cfg, mock_list, client):
mock_list.return_value = ["rule1"]
mock_cfg.return_value = {
"zones": {"public": {"rich_rules": [{"id": "a1", "rule": "rule1"}]}}
}
@_fw("get")
def test_list(self, mock_get, client):
mock_get.return_value = [{"id": "a1", "rule": "rule1"}]
resp = client.get("/api/firewall/rich-rules/public")
assert resp.status_code == 200
data = resp.get_json()
assert isinstance(data["data"], list)
@patch("webui.api.firewall.remove_rich_rule_by_id")
def test_remove_by_id(self, mock_remove, client):
mock_remove.return_value = None
@_fw("delete")
def test_remove_by_id(self, mock_delete, client):
mock_delete.return_value = {"zone": "public", "id": "abc123"}
resp = client.delete("/api/firewall/rich-rules/public/abc123")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
@patch("webui.api.firewall.remove_rich_rule_by_id")
def test_remove_not_found(self, mock_remove, client):
mock_remove.side_effect = ValueError("not found")
@_fw("delete")
def test_remove_not_found(self, mock_delete, client):
from daemon.client import NotFound
mock_delete.side_effect = NotFound("not found")
resp = client.delete("/api/firewall/rich-rules/public/abc123")
assert resp.status_code == 404
class TestFirewallServices:
@patch("webui.api.firewall.get_services")
def test_list(self, mock_services, client):
mock_services.return_value = ["ssh", "http", "dns"]
@_fw("get")
def test_list(self, mock_get, client):
mock_get.return_value = ["ssh", "http", "dns"]
resp = client.get("/api/firewall/services")
assert resp.status_code == 200
assert resp.get_json()["data"] == ["ssh", "http", "dns"]
class TestFirewallInterfaces:
@patch("webui.api.firewall.get_interfaces")
def test_list(self, mock_ifaces, client):
mock_ifaces.return_value = ["eth0", "eth1"]
@_fw("get")
def test_list(self, mock_get, client):
mock_get.return_value = [
{
"name": "eth0",
"mac": "aa:bb:cc:dd:ee:00",
"state": "UP",
"mtu": 1500,
"ips": ["192.168.1.1/24"],
"ipv6": [],
"zone": "internal",
},
{
"name": "eth1",
"mac": "aa:bb:cc:dd:ee:01",
"state": "UP",
"mtu": 1500,
"ips": ["10.0.0.1/24"],
"ipv6": [],
"zone": "public",
},
]
resp = client.get("/api/firewall/interfaces")
assert resp.status_code == 200
assert resp.get_json()["data"] == ["eth0", "eth1"]
data = resp.get_json()["data"]
assert len(data) == 2
assert data[0]["name"] == "eth0"
class TestFirewallMasquerade:
@patch("webui.api.firewall.set_masquerade")
def test_enable(self, mock_set, client):
mock_set.return_value = None
@_fw("post")
def test_enable(self, mock_post, client):
mock_post.return_value = {"zone": "internal", "masquerade": True}
resp = client.post(
"/api/firewall/masquerade",
json={"zone": "internal", "enable": True},
@@ -187,9 +235,14 @@ class TestFirewallMasquerade:
class TestFirewallForwardPort:
@patch("webui.api.firewall.add_forward_port")
def test_add(self, mock_add, client):
mock_add.return_value = {"id": "fp1", "port": 443, "proto": "tcp"}
@_fw("post")
def test_add(self, mock_post, client):
mock_post.return_value = {
"id": "fp1",
"port": 443,
"proto": "tcp",
"zone": "public",
}
resp = client.post(
"/api/firewall/forward-port",
json={"zone": "public", "port": 443, "proto": "tcp"},
@@ -205,20 +258,52 @@ class TestFirewallForwardPort:
)
assert resp.status_code == 400
@patch("webui.api.firewall.remove_forward_port_by_id")
def test_remove_by_id(self, mock_remove, client):
mock_remove.return_value = None
@_fw("delete")
def test_remove_by_id(self, mock_delete, client):
mock_delete.return_value = {"zone": "public", "port": 443, "proto": "tcp"}
resp = client.delete("/api/firewall/forward-port/public/443/tcp")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
@patch("webui.api.firewall.remove_forward_port_by_id")
def test_remove_not_found(self, mock_remove, client):
mock_remove.side_effect = ValueError("not found")
@_fw("delete")
def test_remove_not_found(self, mock_delete, client):
from daemon.client import NotFound
mock_delete.side_effect = NotFound("not found")
resp = client.delete("/api/firewall/forward-port/public/999/tcp")
assert resp.status_code == 404
def test_add_value_error(self, client):
resp = client.post(
"/api/firewall/forward-port",
json={"zone": "public", "port": "not_a_number", "proto": "tcp"},
)
assert resp.status_code == 400
class TestFirewallConfigApply:
@_fw("post")
def test_config_apply_success(self, mock_post, client):
mock_post.return_value = {
"applied_zones": ["public"],
"backup": "/tmp/rules.json",
}
resp = client.post("/api/firewall/config/apply")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
assert "public" in data["data"]["applied_zones"]
@_fw("post")
def test_config_apply_error(self, mock_post, client):
mock_post.side_effect = RuntimeError("apply failed")
resp = client.post("/api/firewall/config/apply")
assert resp.status_code == 500
data = resp.get_json()
assert data["ok"] is False
assert data["error"] == "apply failed"
# ============================================================================
# DHCP
@@ -226,14 +311,15 @@ class TestFirewallForwardPort:
class TestDhcpConfig:
@patch("webui.api.dhcp.get_config")
@_dh("get")
def test_get(self, mock_get, client):
mock_get.return_value = {"dhcp": {}, "dns": {}}
resp = client.get("/api/dhcp/config")
assert resp.status_code == 200
assert resp.get_json()["ok"] is True
def test_post_invalid_body(self, client):
@_dh("post")
def test_post_invalid_body(self, mock_post, client):
resp = client.post(
"/api/dhcp/config", data="not json", content_type="text/plain"
)
@@ -242,9 +328,9 @@ class TestDhcpConfig:
class TestDhcpApply:
@patch("webui.api.dhcp.apply_config")
def test_apply(self, mock_apply, client):
mock_apply.return_value = None
@_dh("post")
def test_apply(self, mock_post, client):
mock_post.return_value = {"applied": True}
resp = client.post("/api/dhcp/apply")
assert resp.status_code == 200
data = resp.get_json()
@@ -252,9 +338,9 @@ class TestDhcpApply:
class TestDhcpStatus:
@patch("webui.api.dhcp.dnsmasq_status")
def test_success(self, mock_status, client):
mock_status.return_value = {"service_active": True}
@_dh("get")
def test_success(self, mock_get, client):
mock_get.return_value = {"service_active": True}
resp = client.get("/api/dhcp/status")
assert resp.status_code == 200
data = resp.get_json()
@@ -262,9 +348,13 @@ class TestDhcpStatus:
class TestDhcpRanges:
@patch("webui.api.dhcp.set_dhcp_range")
def test_add_range(self, mock_set, client):
mock_set.return_value = None
@_dh("post")
def test_add_range(self, mock_post, client):
mock_post.return_value = {
"interface": "eth0",
"start": "192.168.1.100",
"end": "192.168.1.200",
}
resp = client.post(
"/api/dhcp/ranges",
json={
@@ -282,9 +372,13 @@ class TestDhcpRanges:
resp = client.post("/api/dhcp/ranges", json={"start": "192.168.1.100"})
assert resp.status_code == 400
@patch("webui.api.dhcp.remove_dhcp_range")
def test_remove_range(self, mock_remove, client):
mock_remove.return_value = None
@_dh("delete")
def test_remove_range(self, mock_delete, client):
mock_delete.return_value = {
"interface": "eth0",
"start": "192.168.1.100",
"end": "192.168.1.200",
}
resp = client.delete(
"/api/dhcp/ranges",
json={
@@ -303,9 +397,9 @@ class TestDhcpRanges:
class TestDhcpStaticLease:
@patch("webui.api.dhcp.add_static_lease")
def test_add(self, mock_add, client):
mock_add.return_value = None
@_dh("post")
def test_add(self, mock_post, client):
mock_post.return_value = {"mac": "AA:BB:CC", "ip": "10.0.0.5"}
resp = client.post(
"/api/dhcp/static-lease",
json={"mac": "AA:BB:CC", "ip": "10.0.0.5"},
@@ -316,27 +410,17 @@ class TestDhcpStaticLease:
resp = client.post("/api/dhcp/static-lease", json={"ip": "10.0.0.5"})
assert resp.status_code == 400
@patch("webui.api.dhcp.remove_static_lease")
@patch("webui.api.dhcp.get_config")
def test_remove(self, mock_get, mock_remove, client):
mock_get.return_value = {
"dhcp": {"static_leases": [{"mac": "AA:BB:CC", "ip": "10.0.0.5"}]}
}
mock_remove.return_value = None
@_dh("delete")
def test_remove(self, mock_delete, client):
mock_delete.return_value = {"mac": "AA:BB:CC"}
resp = client.delete("/api/dhcp/static-lease/AA:BB:CC")
assert resp.status_code == 200
@patch("webui.api.dhcp.get_config")
def test_remove_not_found(self, mock_get, client):
mock_get.return_value = {"dhcp": {"static_leases": []}}
resp = client.delete("/api/dhcp/static-lease/AA:BB:CC")
assert resp.status_code == 404
class TestDhcpDnsRecord:
@patch("webui.api.dhcp.add_dns_record")
def test_add(self, mock_add, client):
mock_add.return_value = None
@_dh("post")
def test_add(self, mock_post, client):
mock_post.return_value = {"name": "host.local", "address": "10.0.0.10"}
resp = client.post(
"/api/dhcp/dns-record",
json={"name": "host.local", "address": "10.0.0.10"},
@@ -347,22 +431,12 @@ class TestDhcpDnsRecord:
resp = client.post("/api/dhcp/dns-record", json={})
assert resp.status_code == 400
@patch("webui.api.dhcp.remove_dns_record")
@patch("webui.api.dhcp.get_config")
def test_remove(self, mock_get, mock_remove, client):
mock_get.return_value = {
"dns": {"custom_records": [{"name": "host.local", "address": "10.0.0.10"}]}
}
mock_remove.return_value = None
@_dh("delete")
def test_remove(self, mock_delete, client):
mock_delete.return_value = {"name": "host.local"}
resp = client.delete("/api/dhcp/dns-record/host.local")
assert resp.status_code == 200
@patch("webui.api.dhcp.get_config")
def test_remove_not_found(self, mock_get, client):
mock_get.return_value = {"dns": {"custom_records": []}}
resp = client.delete("/api/dhcp/dns-record/host.local")
assert resp.status_code == 404
# ============================================================================
# Proxy
@@ -370,15 +444,15 @@ class TestDhcpDnsRecord:
class TestProxyDomains:
@patch("webui.api.proxy.get_domains")
@_px("get")
def test_list(self, mock_get, client):
mock_get.return_value = []
resp = client.get("/api/proxy/domains")
assert resp.status_code == 200
@patch("webui.api.proxy.add_domain")
def test_add(self, mock_add, client):
mock_add.return_value = None
@_px("post")
def test_add(self, mock_post, client):
mock_post.return_value = {"domain": "ex.com"}
resp = client.post(
"/api/proxy/domains",
json={"domain": "ex.com", "backend_host": "10.0.0.1", "backend_port": 80},
@@ -391,25 +465,25 @@ class TestProxyDomains:
class TestProxyApply:
@patch("webui.api.proxy.apply")
def test_apply(self, mock_apply, client):
mock_apply.return_value = None
@_px("post")
def test_apply(self, mock_post, client):
mock_post.return_value = {"applied": True}
resp = client.post("/api/proxy/apply")
assert resp.status_code == 200
class TestProxyTest:
@patch("webui.api.proxy.test_config")
def test_valid(self, mock_test, client):
mock_test.return_value = (True, "syntax ok")
@_px("post")
def test_valid(self, mock_post, client):
mock_post.return_value = {"valid": True, "output": "syntax ok"}
resp = client.post("/api/proxy/test")
assert resp.status_code == 200
data = resp.get_json()
assert data["data"]["valid"] is True
@patch("webui.api.proxy.test_config")
def test_invalid(self, mock_test, client):
mock_test.return_value = (False, "error msg")
@_px("post")
def test_invalid(self, mock_post, client):
mock_post.return_value = {"valid": False, "output": "error msg"}
resp = client.post("/api/proxy/test")
assert resp.status_code == 400
data = resp.get_json()
@@ -423,15 +497,17 @@ class TestProxyTest:
class TestCertsList:
@patch("webui.api.certs.list_certs")
def test_list(self, mock_list, client):
mock_list.return_value = []
@_ce("get")
def test_list(self, mock_get, client):
mock_get.return_value = []
resp = client.get("/api/certs/list")
assert resp.status_code == 200
@patch("webui.api.certs.get_cert_info")
def test_details_not_found(self, mock_info, client):
mock_info.side_effect = ValueError("not found")
@_ce("get")
def test_details_not_found(self, mock_get, client):
from daemon.client import NotFound
mock_get.side_effect = NotFound("not found")
resp = client.get("/api/certs/example.com")
assert resp.status_code == 404
@@ -454,10 +530,10 @@ class TestCertsEmail:
class TestWireguardConfig:
@patch("webui.api.wireguard.get_config")
@_wg("get")
def test_get(self, mock_get, client):
mock_get.return_value = {
"interface": {"name": "wg0", "private_key": "secret"},
"interface": {"name": "wg0"},
"peers": {},
}
resp = client.get("/api/wireguard/config")
@@ -465,55 +541,32 @@ class TestWireguardConfig:
assert data["ok"] is True
assert "private_key" not in data["data"]["interface"]
@patch("webui.api.wireguard.save_config")
def test_post(self, mock_save, client):
mock_save.return_value = None
@_wg("post")
def test_post(self, mock_post, client):
mock_post.return_value = {"config_saved": True}
resp = client.post("/api/wireguard/config", json={"peers": {}})
assert resp.status_code == 200
@patch("webui.api.wireguard.save_config")
@patch("webui.api.wireguard.get_config")
def test_post_strips_private_key(self, mock_get, mock_save, client):
mock_get.return_value = {
"interface": {"name": "wg0", "private_key": "existing"},
"peers": {},
}
mock_save.return_value = None
resp = client.post(
"/api/wireguard/config",
json={"interface": {"name": "wg0", "private_key": "secret"}, "peers": {}},
)
assert resp.status_code == 200
saved = mock_save.call_args[0][0]
assert saved["interface"]["private_key"] == "existing"
@patch("webui.api.wireguard.save_config")
@patch("webui.api.wireguard.get_config")
def test_patch_strips_private_key(self, mock_get, mock_save, client):
mock_get.return_value = {
"interface": {"name": "wg0", "private_key": "existing"},
"peers": {},
}
mock_save.return_value = None
@_wg("patch")
def test_patch(self, mock_patch, client):
mock_patch.return_value = {"config_saved": True}
resp = client.patch(
"/api/wireguard/config",
json={"interface": {"name": "wg1", "private_key": "injected"}},
json={"interface": {"name": "wg0"}},
)
assert resp.status_code == 200
saved = mock_save.call_args[0][0]
assert saved.get("interface", {}).get("private_key") == "existing"
class TestWireguardPeers:
@patch("webui.api.wireguard.get_peers")
@_wg("get")
def test_list(self, mock_get, client):
mock_get.return_value = []
resp = client.get("/api/wireguard/peers")
assert resp.status_code == 200
@patch("webui.api.wireguard.add_peer")
def test_add(self, mock_add, client):
mock_add.return_value = {
@_wg("post")
def test_add(self, mock_post, client):
mock_post.return_value = {
"name": "client1",
"public_key": "pub",
}
@@ -529,25 +582,17 @@ class TestWireguardPeers:
resp = client.post("/api/wireguard/peers", json={})
assert resp.status_code == 400
@patch("webui.api.wireguard.remove_peer")
@patch("webui.api.wireguard.get_config")
def test_remove_by_name(self, mock_get, mock_remove, client):
mock_get.return_value = {"peers": {"client1": {}}}
mock_remove.return_value = None
@_wg("delete")
def test_remove_by_name(self, mock_delete, client):
mock_delete.return_value = {"name": "client1"}
resp = client.delete("/api/wireguard/peers/client1")
assert resp.status_code == 200
@patch("webui.api.wireguard.get_config")
def test_remove_not_found(self, mock_get, client):
mock_get.return_value = {"peers": {}}
resp = client.delete("/api/wireguard/peers/unknown")
assert resp.status_code == 404
class TestWireguardInitialize:
@patch("webui.api.wireguard.initialize")
def test_initialize(self, mock_init, client):
mock_init.return_value = None
@_wg("post")
def test_initialize(self, mock_post, client):
mock_post.return_value = {"initialized": True}
resp = client.post("/api/wireguard/initialize")
data = resp.get_json()
assert data["ok"] is True
@@ -560,39 +605,39 @@ class TestWireguardGenerateClient:
class TestWireguardStatus:
@patch("webui.api.wireguard.status")
def test_get(self, mock_status, client):
mock_status.return_value = {"up": True, "interface": {}, "peers": []}
@_wg("get")
def test_get(self, mock_get, client):
mock_get.return_value = {"up": True, "interface": {}, "peers": []}
resp = client.get("/api/wireguard/status")
assert resp.status_code == 200
class TestWireguardUp:
@patch("webui.api.wireguard.apply")
def test_up_starts_tunnel(self, mock_apply, client):
mock_apply.return_value = None
@_wg("post")
def test_up_starts_tunnel(self, mock_post, client):
mock_post.return_value = {"applied": True}
resp = client.post("/api/wireguard/up")
assert resp.status_code == 200
@patch("webui.api.wireguard.apply")
def test_up_error(self, mock_apply, client):
mock_apply.side_effect = RuntimeError("interface down")
@_wg("post")
def test_up_error(self, mock_post, client):
mock_post.side_effect = RuntimeError("interface down")
resp = client.post("/api/wireguard/up")
assert resp.status_code == 500
class TestWireguardDown:
@patch("webui.api.wireguard.down")
def test_down_stops_tunnel(self, mock_down, client):
mock_down.return_value = None
@_wg("post")
def test_down_stops_tunnel(self, mock_post, client):
mock_post.return_value = {"down": True}
resp = client.post("/api/wireguard/down")
assert resp.status_code == 200
class TestWireguardApply:
@patch("webui.api.wireguard.apply")
def test_apply(self, mock_apply, client):
mock_apply.return_value = None
@_wg("post")
def test_apply(self, mock_post, client):
mock_post.return_value = {"applied": True}
resp = client.post("/api/wireguard/apply")
assert resp.status_code == 200
@@ -603,12 +648,143 @@ class TestWireguardApply:
class TestResponseHelpers:
@patch("webui.api.firewall.get_active_zones")
@patch("webui.api.firewall.get_available_zones")
def test_error_response_format(self, mock_a, mock_b, client):
mock_a.side_effect = RuntimeError("fail")
@_fw("get")
def test_error_response_format(self, mock_get, client):
mock_get.side_effect = RuntimeError("fail")
resp = client.get("/api/firewall/zones")
data = resp.get_json()
assert "error" in data
assert "ok" in data
assert data["ok"] is False
# ============================================================================
# Firewall Config CRUD (daemon-backed)
# ============================================================================
class TestFirewallConfig:
@_fw("get")
def test_config_get(self, mock_get, client):
mock_get.return_value = {"zones": {"public": {"interfaces": ["eth0"]}}}
resp = client.get("/api/firewall/config")
assert resp.status_code == 200
@_fw("post")
@_fw("get")
def test_config_save(self, mock_get, mock_post, client):
mock_post.return_value = {"config_saved": True}
mock_get.return_value = {
"pending": [],
"needs_apply": False,
"unmanaged_zones": {},
}
resp = client.post(
"/api/firewall/config", json={"zones": {"public": {"interfaces": ["eth0"]}}}
)
assert resp.status_code == 200
def test_config_save_missing_zones(self, client):
resp = client.post("/api/firewall/config", json={})
assert resp.status_code == 400
@_fw("patch")
@_fw("get")
def test_config_patch(self, mock_get, mock_patch, client):
mock_patch.return_value = {"config_saved": True}
mock_get.return_value = {
"pending": [],
"needs_apply": False,
"unmanaged_zones": {},
}
resp = client.patch("/api/firewall/config", json={"zones": {}})
assert resp.status_code == 200
@_fw("post")
def test_config_apply(self, mock_post, client):
mock_post.return_value = {
"applied_zones": ["public"],
"backup": "/tmp/rules.json",
}
resp = client.post("/api/firewall/config/apply")
assert resp.status_code == 200
@_fw("get")
def test_config_pending(self, mock_get, client):
mock_get.return_value = {"pending": [], "needs_apply": False}
resp = client.get("/api/firewall/config/pending")
assert resp.status_code == 200
# ============================================================================
# Proxy config
# ============================================================================
class TestProxyConfig:
@_px("get")
def test_get(self, mock_get, client):
mock_get.return_value = {"domains": {}, "ssl": {}}
resp = client.get("/api/proxy/config")
assert resp.status_code == 200
@_px("post")
def test_post(self, mock_post, client):
mock_post.return_value = {"config_saved": True}
resp = client.post("/api/proxy/config", json={"domains": {}})
assert resp.status_code == 200
@_px("patch")
def test_patch(self, mock_patch, client):
mock_patch.return_value = {"config_saved": True}
resp = client.patch("/api/proxy/config", json={})
assert resp.status_code == 200
# ============================================================================
# DHCP config
# ============================================================================
class TestDhcpConfigCrud:
@_dh("patch")
def test_patch(self, mock_patch, client):
mock_patch.return_value = {"config_saved": True}
resp = client.patch("/api/dhcp/config", json={"dhcp": {}})
assert resp.status_code == 200
# ============================================================================
# Proxy management
# ============================================================================
class TestProxyManagement:
@_px("post")
def test_set_management(self, mock_post, client):
mock_post.return_value = {"domain": "vacuum-wall.local"}
resp = client.post(
"/api/proxy/management",
json={
"domain": "vacuum-wall.local",
"flask_host": "127.0.0.1",
"flask_port": 9090,
},
)
assert resp.status_code == 200
# ============================================================================
# Proxy domain update
# ============================================================================
class TestProxyDomainUpdate:
@_px("post")
def test_update(self, mock_post, client):
mock_post.return_value = {"domain": "ex.com"}
resp = client.put(
"/api/proxy/domains/ex.com",
json={"backend_host": "10.0.0.2"},
)
assert resp.status_code == 200
tests/test_dnsmasq.py
+100
View File
@@ -167,3 +167,103 @@ class TestUpstreamsAndDomain:
dnsmasq.set_domain(None)
cfg = dnsmasq.get_config()
assert cfg["dns"]["domain"] is None
# ── Daemon handler tests (NotFoundError on missing resources) ──
class TestDaemonRemoveStaticLease:
@patch("daemon.handlers.dnsmasq._get_config")
@patch("daemon.handlers.dnsmasq._save_config")
def test_raises_not_found_when_missing(self, mock_save, mock_get):
from daemon.handlers import dnsmasq as daemon_dnsmasq
mock_get.return_value = {"dhcp": {"static_leases": []}, "dns": {}}
with pytest.raises(Exception, match="not found"):
daemon_dnsmasq.remove_static_lease(None, {"mac": "FF:FF:FF"})
@patch("daemon.handlers.dnsmasq._get_config")
@patch("daemon.handlers.dnsmasq._save_config")
def test_succeeds_when_exists(self, mock_save, mock_get):
from daemon.handlers import dnsmasq as daemon_dnsmasq
mock_get.return_value = {
"dhcp": {"static_leases": [{"mac": "aa:bb:cc", "ip": "10.0.0.1"}]},
"dns": {},
}
result = daemon_dnsmasq.remove_static_lease(None, {"mac": "AA:BB:CC"})
assert result["mac"] == "AA:BB:CC"
class TestDaemonRemoveDnsRecord:
@patch("daemon.handlers.dnsmasq._get_config")
@patch("daemon.handlers.dnsmasq._save_config")
def test_raises_not_found_when_missing(self, mock_save, mock_get):
from daemon.handlers import dnsmasq as daemon_dnsmasq
mock_get.return_value = {
"dhcp": {},
"dns": {"custom_records": []},
}
with pytest.raises(Exception, match="not found"):
daemon_dnsmasq.remove_dns_record(None, {"name": "nonexistent"})
@patch("daemon.handlers.dnsmasq._get_config")
@patch("daemon.handlers.dnsmasq._save_config")
def test_succeeds_when_exists(self, mock_save, mock_get):
from daemon.handlers import dnsmasq as daemon_dnsmasq
mock_get.return_value = {
"dhcp": {},
"dns": {"custom_records": [{"name": "host", "address": "10.0.0.1"}]},
}
result = daemon_dnsmasq.remove_dns_record(None, {"name": "host"})
assert result["name"] == "host"
class TestDaemonRemoveDhcpRange:
@patch("daemon.handlers.dnsmasq._get_config")
@patch("daemon.handlers.dnsmasq._save_config")
def test_raises_not_found_when_missing(self, mock_save, mock_get):
from daemon.handlers import dnsmasq as daemon_dnsmasq
mock_get.return_value = {
"dhcp": {"ranges": []},
"dns": {},
}
with pytest.raises(Exception, match="not found"):
daemon_dnsmasq.remove_dhcp_range(
None,
{
"interface": "eth0",
"start": "10.0.0.100",
"end": "10.0.0.200",
},
)
@patch("daemon.handlers.dnsmasq._get_config")
@patch("daemon.handlers.dnsmasq._save_config")
def test_succeeds_when_exists(self, mock_save, mock_get):
from daemon.handlers import dnsmasq as daemon_dnsmasq
mock_get.return_value = {
"dhcp": {
"ranges": [
{
"interface": "eth0",
"start": "10.0.0.100",
"end": "10.0.0.200",
}
]
},
"dns": {},
}
result = daemon_dnsmasq.remove_dhcp_range(
None,
{
"interface": "eth0",
"start": "10.0.0.100",
"end": "10.0.0.200",
},
)
assert result["interface"] == "eth0"
tests/test_firewall.py
+494 -236
View File
@@ -1,17 +1,26 @@
from datetime import datetime
from unittest.mock import patch
"""Tests for lib/firewall.py (pure logic) and daemon/handlers/firewall.py (privilege boundary)."""
from unittest.mock import MagicMock, patch
import pytest
from daemon.handlers import firewall as daemonfirewall
from daemon.server import NotFoundError
from lib import firewall
# ---------------------------------------------------------------------------
# lib/firewall.py — pure parsing (no sudo)
# ---------------------------------------------------------------------------
class TestParseForwardPorts:
def test_single_entry(self):
def test_lib_single_entry(self):
result = firewall._parse_forward_ports("port=443/proto=tcp")
assert len(result) == 1
assert result[0]["port"] == 443
assert result[0]["proto"] == "tcp"
def test_multiple_entries(self):
def test_lib_multiple_entries(self):
result = firewall._parse_forward_ports(
"port=443/proto=tcp port=80/proto=tcp/toaddr=10.0.0.1/toport=8080"
)
@@ -21,144 +30,59 @@ class TestParseForwardPorts:
assert result[1]["toaddr"] == "10.0.0.1"
assert result[1]["toport"] == 8080
def test_empty_string(self):
def test_lib_empty_string(self):
assert firewall._parse_forward_ports("") == []
def test_daemon_no_redundant_import(self):
assert not hasattr(daemonfirewall, "_parse_forward_ports")
class TestGetActiveZones:
@patch("lib.firewall.run")
def test_parses_active_zones(self, mock_run):
mock_run.return_value = "public\n eth0\ninternal\n eth1\n eth2"
result = firewall.get_active_zones()
class TestParseActiveZones:
def test_lib_parses_zones(self):
result = firewall._parse_active_zones(
"public\n eth0\ninternal\n eth1\n eth2"
)
assert result == {
"public": ["eth0"],
"internal": ["eth1", "eth2"],
}
@patch("lib.firewall.run")
def test_empty_output(self, mock_run):
mock_run.return_value = ""
result = firewall.get_active_zones()
assert result == {}
def test_lib_empty_output(self):
assert firewall._parse_active_zones("") == {}
@patch("lib.firewall.run")
def test_zone_with_no_interfaces(self, mock_run):
mock_run.return_value = "dmz"
result = firewall.get_active_zones()
assert result == {"dmz": []}
def test_lib_zone_no_interfaces(self):
assert firewall._parse_active_zones("dmz") == {"dmz": []}
def test_daemon_import_same(self):
assert daemonfirewall._parse_active_zones is firewall._parse_active_zones
class TestGetZoneInfo:
@patch("lib.firewall.run")
def test_parses_zone_info(self, mock_run):
mock_run.return_value = (
"target: default\n"
"interfaces: eth0\n"
"sources: \n"
"services: ssh dhcp\n"
"ports: 8080/tcp\n"
"protocols: \n"
"forward-ports: \n"
"masquerade: yes\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
class TestParseZoneOutput:
def test_lib_parses_zone(self):
result = firewall._parse_zone_output(
"public",
(
"target: default\n"
"interfaces: eth0\n"
"services: ssh dhcp\n"
"masquerade: yes\n"
),
)
result = firewall.get_zone_info("public")
assert result["name"] == "public"
assert result["services"] == ["ssh", "dhcp"]
assert result["ports"] == ["8080/tcp"]
assert result["masquerade"] is True
assert result["interfaces"] == ["eth0"]
assert result["sources"] == []
assert result["rich-rules"] == []
def test_daemon_import_same(self):
assert daemonfirewall._parse_zone_output is firewall._parse_zone_output
class TestGetInterfaces:
@patch("lib.firewall.run")
def test_parses_interfaces(self, mock_run):
mock_run.return_value = (
class TestParseInterfaces:
def test_lib_parses_interfaces(self):
result = firewall._parse_interfaces(
"1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536\n"
"2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500\n"
"3: eth1: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500\n"
)
result = firewall.get_interfaces()
assert result == ["lo", "eth0", "eth1"]
class TestGetRichRules:
@patch("lib.firewall.run")
def test_single_rule(self, mock_run):
mock_run.return_value = (
'rule family="ipv4" port protocol="tcp" port="443" accept;'
)
result = firewall.get_rich_rules("public")
assert len(result) == 1
@patch("lib.firewall.run")
def test_empty_rules(self, mock_run):
mock_run.return_value = ""
result = firewall.get_rich_rules("public")
assert result == []
@patch("lib.firewall.run")
def test_multiline_rule(self, mock_run):
mock_run.return_value = (
'rule family="ipv4"\n source address="10.0.0.0/24"\n reject;'
)
result = firewall.get_rich_rules("public")
assert len(result) == 1
assert "10.0.0.0/24" in result[0]
class TestNowIso:
def test_returns_iso_string(self):
result = firewall._now_iso()
datetime.fromisoformat(result)
assert "+" in result
class TestAddForwardPort:
@patch("lib.firewall.run")
def test_forward_port_basic(self, mock_run):
mock_run.return_value = ""
firewall.add_forward_port("public", 443, "tcp", toaddr="10.0.0.5", toport=8080)
calls = [c[0][0] for c in mock_run.call_args_list]
assert any("--add-forward-port=" in str(c) for c in calls)
@patch("lib.firewall.run")
def test_forward_port_port_only(self, mock_run):
mock_run.return_value = ""
firewall.add_forward_port("public", 80, "tcp", toport=8080)
class TestGetState:
@patch("lib.firewall.get_available_zones")
@patch("lib.firewall.get_zone_info")
@patch("lib.firewall.get_active_zones")
@patch("lib.firewall.get_interfaces")
@patch("lib.firewall.get_services")
@patch("lib.firewall.get_rich_rules")
def test_returns_full_state(
self,
mock_rich,
mock_services,
mock_ifaces,
mock_active,
mock_zone_info,
mock_available,
):
mock_available.return_value = ["public", "internal"]
mock_active.return_value = {"public": ["eth0"]}
mock_ifaces.return_value = ["eth0", "eth1"]
mock_services.return_value = ["ssh", "http"]
mock_zone_info.return_value = {"name": "public", "services": []}
mock_rich.return_value = []
result = firewall.get_state()
assert "zones" in result
assert "active_zones" in result
assert "timestamp" in result
assert result == ["lo", "eth0"]
class TestNormalizeTarget:
@@ -192,6 +116,11 @@ class TestLiveTargetToConfig:
assert firewall._live_target_to_config("") == "DEFAULT"
# ---------------------------------------------------------------------------
# lib/firewall.py — config helpers (no sudo)
# ---------------------------------------------------------------------------
class TestEnsureConfigFile:
def test_creates_file_if_missing(self, tmp_path):
cfg_dir = tmp_path / "config" / "firewall"
@@ -248,26 +177,299 @@ class TestConfigSet:
assert content["zones"]["test"]["interfaces"] == ["eth0"]
class TestConfigApply:
class TestConfigPending:
@patch("lib.firewall.get_config")
@patch("lib.firewall.save_backup")
@patch("lib.firewall.get_available_zones")
@patch("lib.firewall.create_zone")
@patch("lib.firewall.set_zone_services")
@patch("lib.firewall.set_zone_interfaces")
@patch("lib.firewall.set_masquerade")
@patch("lib.firewall._reload")
def test_applies_existing_zone(
self,
mock_reload,
mock_set_mq,
mock_set_ifaces,
mock_set_svcs,
mock_create,
mock_available,
mock_backup,
mock_cfg,
):
def test_detects_interface_drift(self, mock_cfg):
mock_cfg.return_value = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http"],
"masquerade": False,
},
},
}
state = {
"zones": {
"public": {
"interfaces": ["eth1"],
"services": ["http"],
"masquerade": False,
},
},
}
result = firewall.config_pending(state)
assert result["needs_apply"] is True
assert any(c["type"] == "interfaces" for c in result["pending"])
@patch("lib.firewall.get_config")
def test_in_sync(self, mock_cfg):
mock_cfg.return_value = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http"],
"masquerade": False,
},
},
}
state = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http"],
"masquerade": False,
},
},
}
result = firewall.config_pending(state)
assert result["needs_apply"] is False
@patch("lib.firewall.get_config")
def test_detects_services_drift(self, mock_cfg):
mock_cfg.return_value = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http", "ssh"],
"masquerade": False,
},
},
}
state = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http"],
"masquerade": False,
},
},
}
result = firewall.config_pending(state)
assert any(c["type"] == "services" for c in result["pending"])
@patch("lib.firewall.get_config")
def test_detects_unmanaged_zones(self, mock_cfg):
mock_cfg.return_value = {"zones": {}}
state = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": [],
"masquerade": False,
},
},
}
result = firewall.config_pending(state)
assert "public" in result["unmanaged_zones"]
# ---------------------------------------------------------------------------
# lib/firewall.py — parse zone output (used by both lib and daemon)
# ---------------------------------------------------------------------------
class TestGetZoneInfo:
def test_parses_zone_info(self):
result = firewall._parse_zone_output(
"public",
(
"target: default\n"
"interfaces: eth0\n"
"sources: \n"
"services: ssh dhcp\n"
"ports: 8080/tcp\n"
"protocols: \n"
"forward-ports: \n"
"masquerade: yes\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
),
)
assert result["name"] == "public"
assert result["services"] == ["ssh", "dhcp"]
assert result["ports"] == ["8080/tcp"]
assert result["masquerade"] is True
assert result["interfaces"] == ["eth0"]
assert result["sources"] == []
assert result["rich-rules"] == []
# ---------------------------------------------------------------------------
# lib/firewall.py — no sudo functions
# ---------------------------------------------------------------------------
class TestLibNoSudo:
def test_no_run_import(self):
import inspect
source = inspect.getsource(firewall)
assert "sudo=True" not in source, "lib/firewall.py must not call sudo"
# ---------------------------------------------------------------------------
# daemon/handlers/firewall.py — privileged operations
# ---------------------------------------------------------------------------
def _mock_run_factory(*outputs):
"""Create a mock run() that cycles through outputs on successive calls."""
idx = [0]
def side_effect(*args, **kwargs):
result = outputs[idx[0] % len(outputs)]
idx[0] += 1
if result is RuntimeError:
raise RuntimeError("command failed")
return result
return side_effect
class TestDaemonParseForwardPorts:
def test_handler_uses_get_forward_ports(self):
assert callable(daemonfirewall._get_forward_ports)
class TestDaemonParseActiveZones:
def test_parses_active_zones(self):
result = daemonfirewall._parse_active_zones(
"public\n eth0\ninternal\n eth1\n eth2"
)
assert result == {
"public": ["eth0"],
"internal": ["eth1", "eth2"],
}
def test_empty_output(self):
assert daemonfirewall._parse_active_zones("") == {}
def test_zone_with_no_interfaces(self):
assert daemonfirewall._parse_active_zones("dmz") == {"dmz": []}
class TestDaemonParseZoneOutput:
def test_parses_zone_info(self):
result = daemonfirewall._parse_zone_output(
"public",
(
"target: default\n"
"interfaces: eth0\n"
"sources: \n"
"services: ssh dhcp\n"
"ports: 8080/tcp\n"
"protocols: \n"
"forward-ports: \n"
"masquerade: yes\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
),
)
assert result["name"] == "public"
assert result["services"] == ["ssh", "dhcp"]
assert result["ports"] == ["8080/tcp"]
assert result["masquerade"] is True
assert result["interfaces"] == ["eth0"]
class TestDaemonGetInterfaces:
@patch("daemon.handlers.firewall.run")
def test_parses_interfaces(self, mock_run):
link_out = (
"1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536\n"
"2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500\n"
"3: eth1: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500\n"
)
mock_run.return_value = link_out
result = daemonfirewall.get_interfaces(None, None)
assert [i["name"] for i in result] == ["lo", "eth0", "eth1"]
class TestDaemonGetRichRules:
@patch("daemon.handlers.firewall.run")
def test_single_rule(self, mock_run):
mock_run.return_value = (
'rule family="ipv4" port protocol="tcp" port="443" accept;'
)
cfg_mock = MagicMock(return_value={"zones": {"public": {"rich_rules": []}}})
with patch.object(daemonfirewall, "_get_config", cfg_mock):
result = daemonfirewall.list_rich_rules(None, {"zone": "public"})
assert len(result) == 1
@patch("daemon.handlers.firewall.run")
def test_empty_rules(self, mock_run):
mock_run.return_value = ""
cfg_mock = MagicMock(return_value={"zones": {"public": {"rich_rules": []}}})
with patch.object(daemonfirewall, "_get_config", cfg_mock):
result = daemonfirewall.list_rich_rules(None, {"zone": "public"})
assert result == []
@patch("daemon.handlers.firewall.run")
def test_multiline_rule(self, mock_run):
mock_run.return_value = (
'rule family="ipv4"\n source address="10.0.0.0/24"\n reject;'
)
cfg_mock = MagicMock(return_value={"zones": {"public": {"rich_rules": []}}})
with patch.object(daemonfirewall, "_get_config", cfg_mock):
result = daemonfirewall.list_rich_rules(None, {"zone": "public"})
assert len(result) == 1
assert "10.0.0.0/24" in result[0]["rule"]
class TestDaemonGetState:
@patch("daemon.handlers.firewall.run")
def test_returns_full_state(self, mock_run):
def run_side_effect(args, **kwargs):
if "--get-zones" in args:
return "public\ninternal"
if "--get-active-zones" in args:
return "public\n eth0\ninternal\n eth1"
if "--get-services" in args:
return "ssh http dns"
if "ip" in args[0]:
if "link" in args:
return "1: lo: <LOOPBACK,UP> mtu 65536\n2: eth0: <UP> mtu 1500 link/ether aa:bb:cc\n"
if "addr" in args:
return "2: eth0 inet 192.168.1.1/24 brd 192.168.1.255 scope global eth0\n"
if "--list-all" in args:
return (
"target: default\n"
"interfaces: eth0\n"
"sources: \n"
"services: \n"
"ports: \n"
"protocols: \n"
"forward-ports: \n"
"masquerade: no\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
)
return ""
mock_run.side_effect = run_side_effect
result = daemonfirewall._get_state()
assert "zones" in result
assert "active_zones" in result
assert "timestamp" in result
assert "interfaces" in result
assert len(result["interfaces"]) >= 2
assert "public" in result["zones"]
class TestDaemonConfigApply:
@patch("daemon.handlers.firewall._save_backup")
@patch("daemon.handlers.firewall._get_state")
@patch("daemon.handlers.firewall._get_lib_config")
@patch("daemon.handlers.firewall.run")
def test_applies_existing_zone(self, mock_run, mock_cfg, mock_state, mock_backup):
mock_cfg.return_value = {
"zones": {
"public": {
@@ -278,34 +480,35 @@ class TestConfigApply:
},
},
}
mock_available.return_value = ["public", "internal"]
mock_run.return_value = (
"public\ninternal\ntarget: default\n"
"interfaces: \n"
"sources: \n"
"services: \n"
"ports: \n"
"protocols: \n"
"forward-ports: \n"
"masquerade: no\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
)
mock_state.return_value = {"zones": {"public": {}}}
mock_backup.return_value = "/tmp/rules.json"
result = firewall.config_apply()
result = daemonfirewall._config_apply()
assert result["applied_zones"] == ["public"]
assert result["backup"] == "/tmp/rules.json"
mock_set_ifaces.assert_called_once_with("public", ["eth0"])
mock_set_svcs.assert_called_once_with("public", ["http", "https"])
mock_set_mq.assert_called_once_with("public", True)
calls = [str(c) for c in mock_run.call_args_list]
assert any("--add-service=" in c for c in calls)
assert any("--add-interface=" in c for c in calls)
@patch("lib.firewall.get_config")
@patch("lib.firewall.save_backup")
@patch("lib.firewall.get_available_zones")
@patch("lib.firewall.create_zone")
@patch("lib.firewall.set_zone_services")
@patch("lib.firewall.set_zone_interfaces")
@patch("lib.firewall.set_masquerade")
@patch("lib.firewall._reload")
def test_creates_new_zone(
self,
mock_reload,
mock_set_mq,
mock_set_ifaces,
mock_set_svcs,
mock_create,
mock_available,
mock_backup,
mock_cfg,
):
@patch("daemon.handlers.firewall._save_backup")
@patch("daemon.handlers.firewall._get_state")
@patch("daemon.handlers.firewall._get_lib_config")
@patch("daemon.handlers.firewall.run")
def test_creates_new_zone(self, mock_run, mock_cfg, mock_state, mock_backup):
mock_cfg.return_value = {
"zones": {
"custom": {
@@ -316,18 +519,44 @@ class TestConfigApply:
},
},
}
mock_available.return_value = ["public", "internal"]
mock_run.return_value = (
"public\ninternal\ntarget: default\n"
"interfaces: \n"
"sources: \n"
"services: \n"
"ports: \n"
"protocols: \n"
"forward-ports: \n"
"masquerade: no\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
)
mock_state.return_value = {"zones": {"custom": {}}}
mock_backup.return_value = "/tmp/rules.json"
result = firewall.config_apply()
result = daemonfirewall._config_apply()
assert result["applied_zones"] == ["custom"]
mock_create.assert_called_once_with("custom", "ACCEPT")
mock_set_ifaces.assert_called_once_with("custom", ["eth2"])
@patch("daemon.handlers.firewall._save_backup")
@patch("daemon.handlers.firewall._get_state")
@patch("daemon.handlers.firewall._get_lib_config")
@patch("daemon.handlers.firewall.run")
def test_empty_config_no_ops(self, mock_run, mock_cfg, mock_state, mock_backup):
mock_cfg.return_value = {"zones": {}}
mock_run.return_value = ""
mock_state.return_value = {"zones": {}}
mock_backup.return_value = "/tmp/rules.json"
result = daemonfirewall._config_apply()
assert result["applied_zones"] == []
class TestConfigPending:
class TestDaemonConfigPending:
@patch("daemon.handlers.firewall._get_state")
@patch("lib.firewall.get_config")
@patch("lib.firewall.get_state")
def test_detects_interface_drift(self, mock_state, mock_cfg):
def test_detects_interface_drift(self, mock_cfg, mock_state):
mock_cfg.return_value = {
"zones": {
"public": {
@@ -346,13 +575,12 @@ class TestConfigPending:
},
},
}
result = firewall.config_pending()
result = daemonfirewall.config_pending(None, None)
assert result["needs_apply"] is True
assert any(c["type"] == "interfaces" for c in result["pending"])
@patch("daemon.handlers.firewall._get_state")
@patch("lib.firewall.get_config")
@patch("lib.firewall.get_state")
def test_in_sync(self, mock_state, mock_cfg):
def test_in_sync(self, mock_cfg, mock_state):
mock_cfg.return_value = {
"zones": {
"public": {
@@ -371,74 +599,104 @@ class TestConfigPending:
},
},
}
result = firewall.config_pending()
result = daemonfirewall.config_pending(None, None)
assert result["needs_apply"] is False
@patch("lib.firewall.get_config")
@patch("lib.firewall.get_state")
def test_detects_services_drift(self, mock_state, mock_cfg):
# ---------------------------------------------------------------------------
# Zone validation in add_rich_rule, remove_rich_rule, remove_forward_port
# ---------------------------------------------------------------------------
class TestDaemonZoneValidation:
@patch("daemon.handlers.firewall.run")
def test_add_rich_rule_invalid_zone(self, mock_run):
mock_run.return_value = "public\ninternal"
with (
patch.object(daemonfirewall, "_get_config", return_value={"zones": {}}),
patch.object(daemonfirewall, "_save_config"),
pytest.raises(NotFoundError),
):
daemonfirewall.add_rich_rule(
None,
{
"zone": "nonexistent",
"rule": "rule accept",
},
)
@patch("daemon.handlers.firewall.run")
def test_remove_rich_rule_invalid_zone(self, mock_run):
mock_run.return_value = "public\ninternal"
with (
patch.object(daemonfirewall, "_get_config", return_value={"zones": {}}),
pytest.raises(NotFoundError),
):
daemonfirewall.remove_rich_rule(
None, {"zone": "nonexistent", "id": "abc123"}
)
@patch("daemon.handlers.firewall.run")
def test_remove_forward_port_invalid_zone(self, mock_run):
mock_run.return_value = "public\ninternal"
with (
patch.object(daemonfirewall, "_get_config", return_value={"zones": {}}),
pytest.raises(NotFoundError),
):
daemonfirewall.remove_forward_port(
None,
{
"zone": "nonexistent",
"port": 443,
"proto": "tcp",
},
)
# ---------------------------------------------------------------------------
# Forward port removal during config_apply
# ---------------------------------------------------------------------------
class TestDaemonConfigApplyForwardPorts:
@patch("daemon.handlers.firewall._save_backup")
@patch("daemon.handlers.firewall._get_state")
@patch("daemon.handlers.firewall._get_lib_config")
@patch("daemon.handlers.firewall.run")
def test_removes_stale_forward_ports(
self, mock_run, mock_cfg, mock_state, mock_backup
):
mock_cfg.return_value = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http", "ssh"],
"masquerade": False,
},
},
}
mock_state.return_value = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http"],
"masquerade": False,
},
},
}
result = firewall.config_pending()
assert any(c["type"] == "services" for c in result["pending"])
@patch("lib.firewall.get_config")
@patch("lib.firewall.get_state")
def test_detects_unmanaged_zones(self, mock_state, mock_cfg):
mock_cfg.return_value = {"zones": {}}
mock_state.return_value = {
"zones": {
"public": {
"interfaces": ["eth0"],
"interfaces": [],
"services": [],
"masquerade": False,
"forward_ports": [
{"id": "fp_new", "port": 8443, "proto": "tcp"},
],
},
},
}
result = firewall.config_pending()
assert "public" in result["unmanaged_zones"]
class TestConfigEmptyZones:
@patch("lib.firewall.get_config")
@patch("lib.firewall.save_backup")
@patch("lib.firewall.get_available_zones")
@patch("lib.firewall.create_zone")
@patch("lib.firewall.set_zone_services")
@patch("lib.firewall.set_zone_interfaces")
@patch("lib.firewall.set_masquerade")
@patch("lib.firewall._reload")
def test_empty_config_no_ops(
self,
mock_reload,
mock_set_mq,
mock_set_ifaces,
mock_set_svcs,
mock_create,
mock_available,
mock_backup,
mock_cfg,
):
mock_cfg.return_value = {"zones": {}}
mock_available.return_value = []
mock_run.return_value = (
"public\ntarget: default\n"
"interfaces: \n"
"sources: \n"
"services: \n"
"ports: \n"
"protocols: \n"
"forward-ports: port=443/proto=tcp\n"
"masquerade: no\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
)
mock_state.return_value = {"zones": {"public": {}}}
mock_backup.return_value = "/tmp/rules.json"
result = firewall.config_apply()
assert result["applied_zones"] == []
mock_create.assert_not_called()
mock_set_ifaces.assert_not_called()
result = daemonfirewall._config_apply()
assert result["applied_zones"] == ["public"]
calls = [str(c) for c in mock_run.call_args_list]
assert any("--remove-forward-port=" in c for c in calls)
assert any("--add-forward-port=" in c for c in calls)
tests/test_server.py
+3 -22
View File
@@ -90,27 +90,8 @@ class TestSafelyHelper:
class TestPageRoutes:
@patch("webui.server.get_active_zones")
@patch("webui.server.get_interfaces")
@patch("webui.server.dnsmasq_status")
@patch("webui.server.get_domains")
@patch("webui.server.list_certs")
@patch("webui.server.wg_status")
def test_dashboard_no_crash(
self,
mock_wg,
mock_certs,
mock_domains,
mock_dnsmasq,
mock_ifaces,
mock_zones,
client,
):
mock_zones.return_value = {}
mock_ifaces.return_value = []
mock_dnsmasq.return_value = {}
mock_domains.return_value = []
mock_certs.return_value = []
mock_wg.return_value = {}
@patch("webui.server.get")
def test_dashboard_no_crash(self, mock_get, client):
mock_get.return_value = {}
resp = client.get("/")
assert resp.status_code == 200