fix: cross-cell routing for split-tunnel peers

Three related fixes for split-tunnel peers that need to reach connected cells:

1. apply_peer_rules/apply_all_peer_rules now accept wg_subnet (actual local VPN
   subnet) and cell_subnets (connected cells' vpn_subnets) parameters instead of
   hardcoding 10.0.0.0/24. All callers (startup, add_peer, update_peer,
   apply-enforcement endpoint) pass the real values.

2. Explicit ACCEPT rules are inserted in FORWARD for each connected cell's
   subnet so split-tunnel peers (internet_access=False) can still reach
   connected cells via the wg0→wg0 path.

3. apply_ip_range in network_manager now loads cell_links.json and passes it
   to generate_corefile(), fixing a race where the bootstrap DNS thread could
   overwrite the Corefile and wipe cross-cell DNS forwarding zones on startup.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-04 14:36:28 -04:00
parent 8ee1d88e37
commit c2d215ee2e
6 changed files with 171 additions and 19 deletions
+93
View File
@@ -308,6 +308,87 @@ class TestApplyPeerRules(unittest.TestCase):
for c in vpn_rules:
self.assertIn('DROP', c)
def test_custom_wg_subnet_replaces_default(self):
"""wg_subnet parameter is used instead of hardcoded 10.0.0.0/24."""
calls_made = []
def fake_wg_exec(args):
calls_made.append(args)
m = MagicMock()
m.returncode = 0
m.stdout = ''
return m
with patch.object(firewall_manager, '_wg_exec', side_effect=fake_wg_exec):
firewall_manager.apply_peer_rules(
'10.0.2.5',
{'internet_access': True, 'peer_access': True,
'service_access': list(firewall_manager.SERVICE_IPS.keys())},
wg_subnet='10.0.2.0/24',
)
iptables_calls = [c for c in calls_made if 'iptables' in c]
subnets_in_rules = [token for c in iptables_calls for token in c
if '/' in token and token.startswith('10.')]
self.assertIn('10.0.2.0/24', subnets_in_rules,
"Custom wg_subnet should appear in peer-to-peer FORWARD rule")
self.assertNotIn('10.0.0.0/24', subnets_in_rules,
"Default hardcoded subnet must not appear when custom wg_subnet given")
def test_cell_subnets_get_explicit_accept_rules(self):
"""Each cell subnet gets an explicit ACCEPT rule for cross-cell routing."""
calls_made = []
def fake_wg_exec(args):
calls_made.append(args)
m = MagicMock()
m.returncode = 0
m.stdout = ''
return m
with patch.object(firewall_manager, '_wg_exec', side_effect=fake_wg_exec):
firewall_manager.apply_peer_rules(
'10.0.2.5',
{'internet_access': False, 'peer_access': True,
'service_access': list(firewall_manager.SERVICE_IPS.keys())},
wg_subnet='10.0.2.0/24',
cell_subnets=['10.0.0.0/24', '10.0.1.0/24'],
)
iptables_calls = [c for c in calls_made if 'iptables' in c]
for cell_net in ('10.0.0.0/24', '10.0.1.0/24'):
cell_rules = [c for c in iptables_calls if '-d' in c and cell_net in c]
self.assertTrue(cell_rules,
f"Expected FORWARD rule for cell subnet {cell_net}")
for c in cell_rules:
self.assertIn('ACCEPT', c,
f"Cell subnet rule for {cell_net} must be ACCEPT")
def test_no_cell_subnets_no_extra_rules(self):
"""When cell_subnets is empty/None, no extra FORWARD rules are added."""
calls_made = []
def fake_wg_exec(args):
calls_made.append(args)
m = MagicMock()
m.returncode = 0
m.stdout = ''
return m
with patch.object(firewall_manager, '_wg_exec', side_effect=fake_wg_exec):
firewall_manager.apply_peer_rules(
'10.0.2.5',
{'internet_access': True, 'peer_access': True,
'service_access': list(firewall_manager.SERVICE_IPS.keys())},
wg_subnet='10.0.2.0/24',
cell_subnets=[],
)
iptables_calls = [c for c in calls_made if 'iptables' in c and '-I' in c]
# Only 2 rules expected: the catch-all ACCEPT + the peer-to-peer ACCEPT
self.assertEqual(len(iptables_calls), 2,
f"Expected exactly 2 INSERT rules, got: {iptables_calls}")
# ---------------------------------------------------------------------------
# apply_all_peer_rules
@@ -330,6 +411,18 @@ class TestApplyAllPeerRules(unittest.TestCase):
firewall_manager.apply_all_peer_rules(peers)
self.assertEqual(mock_apply.call_count, 1)
def test_wg_subnet_and_cell_subnets_forwarded(self):
"""apply_all_peer_rules passes wg_subnet and cell_subnets to each apply_peer_rules call."""
peers = [_make_peer('10.0.2.2')]
cell_subnets = ['10.0.0.0/24', '10.0.1.0/24']
with patch.object(firewall_manager, 'ensure_caddy_virtual_ips', return_value=True), \
patch.object(firewall_manager, 'apply_peer_rules', return_value=True) as mock_apply:
firewall_manager.apply_all_peer_rules(peers, wg_subnet='10.0.2.0/24',
cell_subnets=cell_subnets)
call_kwargs = mock_apply.call_args_list[0].kwargs
self.assertEqual(call_kwargs.get('wg_subnet'), '10.0.2.0/24')
self.assertEqual(call_kwargs.get('cell_subnets'), cell_subnets)
# ---------------------------------------------------------------------------
# clear_peer_rules