Add subnet conflict validation for wireguard.address and ip_range changes

When a cell is connected to others, changing the local WireGuard address
or Docker ip_range to a subnet that overlaps a connected cell's vpn_subnet
would break routing. Both now return 409 with the conflicting cell name.

- wireguard.address: derive network from new address, check all connected
  cells' vpn_subnet for overlap (after existing format validation)
- ip_range: check all connected cells' vpn_subnet for overlap (after
  existing RFC-1918 validation)

Tests: 4 cases each (overlap → 409, no overlap → ok, no cells → ok,
format error still fires first → 400).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-04 10:00:58 -04:00
parent c658d2b16c
commit 8ee1d88e37
2 changed files with 106 additions and 1 deletions
+80
View File
@@ -209,5 +209,85 @@ class TestDomainConflictValidation(unittest.TestCase):
self.assertNotEqual(r.status_code, 409)
# ---------------------------------------------------------------------------
# WireGuard address subnet conflict validation
# ---------------------------------------------------------------------------
class TestWireGuardAddressConflictValidation(unittest.TestCase):
"""Changing wireguard.address to a subnet overlapping a connected cell → 409."""
def setUp(self):
self.client = _make_client()
self._connected = [{'cell_name': 'remote', 'domain': 'remote.cell',
'vpn_subnet': '10.5.0.0/24', 'dns_ip': '10.5.0.1'}]
def test_wg_address_overlapping_connected_cell_returns_409(self):
with patch('app.cell_link_manager') as mock_clm:
mock_clm.list_connections.return_value = self._connected
r = _put(self.client, {'wireguard': {'address': '10.5.0.1/24'}})
self.assertEqual(r.status_code, 409)
data = json.loads(r.data)
self.assertIn('remote', data['error'])
def test_wg_address_non_overlapping_connected_cell_accepted(self):
with patch('app.cell_link_manager') as mock_clm:
mock_clm.list_connections.return_value = self._connected
r = _put(self.client, {'wireguard': {'address': '10.6.0.1/24'}})
self.assertNotEqual(r.status_code, 409)
def test_wg_address_no_connected_cells_accepted(self):
with patch('app.cell_link_manager') as mock_clm:
mock_clm.list_connections.return_value = []
r = _put(self.client, {'wireguard': {'address': '10.5.0.1/24'}})
self.assertNotEqual(r.status_code, 409)
def test_wg_address_missing_prefix_still_returns_400(self):
"""Format check fires before the conflict check."""
with patch('app.cell_link_manager') as mock_clm:
mock_clm.list_connections.return_value = self._connected
r = _put(self.client, {'wireguard': {'address': '10.5.0.1'}})
self.assertEqual(r.status_code, 400)
# ---------------------------------------------------------------------------
# ip_range subnet conflict validation
# ---------------------------------------------------------------------------
class TestIpRangeConflictValidation(unittest.TestCase):
"""Changing ip_range to one overlapping a connected cell's vpn_subnet → 409."""
def setUp(self):
self.client = _make_client()
self._connected = [{'cell_name': 'remote', 'domain': 'remote.cell',
'vpn_subnet': '10.5.0.0/24', 'dns_ip': '10.5.0.1'}]
def test_ip_range_overlapping_connected_cell_returns_409(self):
with patch('app.cell_link_manager') as mock_clm:
mock_clm.list_connections.return_value = self._connected
r = _put(self.client, {'ip_range': '10.5.0.0/16'})
self.assertEqual(r.status_code, 409)
data = json.loads(r.data)
self.assertIn('remote', data['error'])
def test_ip_range_non_overlapping_connected_cell_accepted(self):
with patch('app.cell_link_manager') as mock_clm:
mock_clm.list_connections.return_value = self._connected
r = _put(self.client, {'ip_range': '10.6.0.0/16'})
self.assertNotEqual(r.status_code, 409)
def test_ip_range_no_connected_cells_accepted(self):
with patch('app.cell_link_manager') as mock_clm:
mock_clm.list_connections.return_value = []
r = _put(self.client, {'ip_range': '10.5.0.0/16'})
self.assertNotEqual(r.status_code, 409)
def test_ip_range_non_rfc1918_still_returns_400(self):
"""RFC-1918 check fires before the conflict check."""
with patch('app.cell_link_manager') as mock_clm:
mock_clm.list_connections.return_value = self._connected
r = _put(self.client, {'ip_range': '8.8.8.0/24'})
self.assertEqual(r.status_code, 400)
if __name__ == '__main__':
unittest.main()