roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

security: replace WireGuard catch-all ACCEPT with DROP

Browse Source 
The PostUp rule appended `iptables -A FORWARD -i wg0 -j ACCEPT` which
allowed any WireGuard-connected client full internet access regardless of
per-peer rules, even when no peers were configured in wg0.conf.

Fix: change PostUp/PostDown to use DROP as the catch-all. Per-peer and
per-cell rules use -I (insert at top) so they take precedence; unknown
or unconfigured WG traffic hits the DROP at the bottom.

Also add reconcile_stale_peer_rules() called on startup to remove FORWARD
rules for peer IPs that no longer exist in the registry, preventing deleted
peers from retaining firewall access across container restarts.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dmitrii Iurco
2026-05-02 00:31:55 -04:00
parent 2b93c8aec5
commit 68c27b4521
6 changed files with 133 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_firewall_manager.py
+79
View File
@@ -715,5 +715,84 @@ class TestEnsureCellApiDnat(unittest.TestCase):
self.assertFalse(result)
# ---------------------------------------------------------------------------
# reconcile_stale_peer_rules
# ---------------------------------------------------------------------------
class TestReconcileStale(unittest.TestCase):
def _save_result(self, stdout_text):
r = MagicMock()
r.returncode = 0
r.stdout = stdout_text
return r
def test_returns_zero_when_no_rules(self):
with patch.object(firewall_manager, '_wg_exec', return_value=self._save_result('*filter\nCOMMIT\n')):
n = firewall_manager.reconcile_stale_peer_rules([])
self.assertEqual(n, 0)
def test_returns_zero_when_all_peers_known(self):
save_out = (
'*filter\n'
'-A FORWARD -s 10.0.0.2 -m comment --comment "pic-peer-10-0-0-2/32" -j ACCEPT\n'
'COMMIT\n'
)
peers = [{'ip': '10.0.0.2'}]
with patch.object(firewall_manager, '_wg_exec', return_value=self._save_result(save_out)):
n = firewall_manager.reconcile_stale_peer_rules(peers)
self.assertEqual(n, 0)
def test_clears_stale_peer(self):
save_out = (
'*filter\n'
'-A FORWARD -s 10.0.0.9 -m comment --comment "pic-peer-10-0-0-9/32" -j ACCEPT\n'
'COMMIT\n'
)
cleared = []
with patch.object(firewall_manager, '_wg_exec', return_value=self._save_result(save_out)):
with patch.object(firewall_manager, 'clear_peer_rules', side_effect=cleared.append) as mock_clear:
n = firewall_manager.reconcile_stale_peer_rules([])
self.assertEqual(n, 1)
mock_clear.assert_called_once_with('10.0.0.9')
def test_handles_cidr_peer_ip(self):
"""Peer IPs stored as 'x.x.x.x/32' should still match."""
save_out = (
'*filter\n'
'-A FORWARD -s 10.0.0.5 -m comment --comment "pic-peer-10-0-0-5/32" -j ACCEPT\n'
'COMMIT\n'
)
peers = [{'ip': '10.0.0.5/32'}]
with patch.object(firewall_manager, '_wg_exec', return_value=self._save_result(save_out)):
with patch.object(firewall_manager, 'clear_peer_rules') as mock_clear:
n = firewall_manager.reconcile_stale_peer_rules(peers)
self.assertEqual(n, 0)
mock_clear.assert_not_called()
def test_returns_zero_on_iptables_save_failure(self):
fail_r = MagicMock()
fail_r.returncode = 1
fail_r.stdout = ''
with patch.object(firewall_manager, '_wg_exec', return_value=fail_r):
n = firewall_manager.reconcile_stale_peer_rules([])
self.assertEqual(n, 0)
def test_multiple_stale_ips_all_cleared(self):
save_out = (
'*filter\n'
'-A FORWARD -s 10.0.0.7 -m comment --comment "pic-peer-10-0-0-7/32" -j DROP\n'
'-A FORWARD -s 10.0.0.8 -m comment --comment "pic-peer-10-0-0-8/32" -j ACCEPT\n'
'COMMIT\n'
)
cleared = []
with patch.object(firewall_manager, '_wg_exec', return_value=self._save_result(save_out)):
with patch.object(firewall_manager, 'clear_peer_rules', side_effect=cleared.append):
n = firewall_manager.reconcile_stale_peer_rules([])
self.assertEqual(n, 2)
self.assertIn('10.0.0.7', cleared)
self.assertIn('10.0.0.8', cleared)
if __name__ == '__main__':
unittest.main()
tests/test_wireguard_manager.py
+3 -2
View File
@@ -480,9 +480,10 @@ class TestWireGuardSysctlAndPortCheck(unittest.TestCase):
cfg = self.wg.generate_config()
self.assertIn('MASQUERADE', cfg)
def test_generate_config_has_forward_rule(self):
def test_generate_config_has_forward_drop_rule(self):
cfg = self.wg.generate_config()
self.assertIn('FORWARD -i %i -j ACCEPT', cfg)
self.assertIn('FORWARD -i %i -j DROP', cfg)
self.assertNotIn('FORWARD -i %i -j ACCEPT', cfg)
# ── check_port_open ───────────────────────────────────────────────────────
tests/test_wireguard_vpn_routing.py
+12 -6
View File
@@ -46,8 +46,13 @@ def _make_wg(tmp: str) -> WireGuardManager:
class TestInternetForwardingRules(unittest.TestCase):
"""
Verify that generate_config() emits the exact iptables rules required for
'internet through VPN': MASQUERADE on eth0 (outbound NAT) and FORWARD ACCEPT
on the wg0 interface. Missing either rule means VPN clients get no internet.
'internet through VPN': MASQUERADE on eth0 (outbound NAT) and a catch-all
FORWARD DROP on the wg0 interface.
The catch-all is DROP (not ACCEPT) so that only per-peer rules inserted at
chain position 1 via apply_peer_rules() can forward traffic. An ACCEPT
catch-all would allow any WireGuard-connected client full internet access
even if they have no entry in peers.json.
"""
def setUp(self):
@@ -61,10 +66,11 @@ class TestInternetForwardingRules(unittest.TestCase):
cfg = self.wg.generate_config()
self.assertIn('POSTROUTING -o eth0 -j MASQUERADE', cfg)
def test_postup_has_forward_accept_on_wg_interface(self):
"""FORWARD ACCEPT allows packets from the WireGuard interface through the kernel."""
def test_postup_has_forward_drop_on_wg_interface(self):
"""Catch-all DROP blocks unconfigured WG clients; per-peer rules inserted above it allow known peers."""
cfg = self.wg.generate_config()
self.assertIn('FORWARD -i %i -j ACCEPT', cfg)
self.assertIn('FORWARD -i %i -j DROP', cfg)
self.assertNotIn('FORWARD -i %i -j ACCEPT', cfg)
def test_postdown_removes_masquerade_rule(self):
"""PostDown must mirror PostUp so rules are cleaned up when the tunnel goes down."""
@@ -73,7 +79,7 @@ class TestInternetForwardingRules(unittest.TestCase):
def test_postdown_removes_forward_rule(self):
cfg = self.wg.generate_config()
self.assertIn('FORWARD -i %i -j ACCEPT', cfg.split('PostDown')[1])
self.assertIn('FORWARD -i %i -j DROP', cfg.split('PostDown')[1])
def test_postup_and_postdown_are_present(self):
"""Both PostUp and PostDown must exist — PostUp without PostDown leaks rules."""