roof
67362349d1
3 new tests in TestSetPeerRouteVia: - 409 when remote_exit_relay_active=True (would create A→B→A cycle) - disable (via_cell=null) bypasses loop check — always allowed - no 409 when remote_exit_relay_active=False (safe to enable) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
203 lines
8.6 KiB
Python
203 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Tests for PUT /api/peers/<peer_name>/route-via.
|
|
|
|
Covers:
|
|
- missing via_cell field → 400
|
|
- peer not found → 404
|
|
- connected cell not found → 404
|
|
- enable route-via: calls wg methods and set_exit_relay_active(True)
|
|
- disable route-via: clears wg methods and set_exit_relay_active(False)
|
|
- switching from one exit cell to another removes old routing first
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
api_dir = Path(__file__).parent.parent / 'api'
|
|
sys.path.insert(0, str(api_dir))
|
|
|
|
from app import app
|
|
|
|
_PEER = {
|
|
'peer': 'alice',
|
|
'ip': '10.0.0.5',
|
|
'public_key': 'alicepubkey=',
|
|
'route_via': None,
|
|
'internet_access': True,
|
|
'service_access': [],
|
|
}
|
|
|
|
_CELL_LINK = {
|
|
'cell_name': 'exit-cell',
|
|
'public_key': 'exitcellpubkey=',
|
|
'vpn_subnet': '10.1.0.0/24',
|
|
'dns_ip': '10.1.0.1',
|
|
'remote_exit_relay_active': False,
|
|
}
|
|
|
|
|
|
class TestSetPeerRouteVia(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
app.config['TESTING'] = True
|
|
self.client = app.test_client()
|
|
|
|
def _put(self, peer_name, body):
|
|
return self.client.put(
|
|
f'/api/peers/{peer_name}/route-via',
|
|
data=json.dumps(body),
|
|
content_type='application/json',
|
|
)
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_missing_via_cell_returns_400(self, mock_reg, mock_wg, mock_clm):
|
|
r = self._put('alice', {})
|
|
self.assertEqual(r.status_code, 400)
|
|
data = json.loads(r.data)
|
|
self.assertIn('via_cell', data['error'])
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_peer_not_found_returns_404(self, mock_reg, mock_wg, mock_clm):
|
|
mock_reg.get_peer.return_value = None
|
|
r = self._put('nobody', {'via_cell': 'exit-cell'})
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_unknown_cell_returns_404(self, mock_reg, mock_wg, mock_clm):
|
|
mock_reg.get_peer.return_value = _PEER
|
|
mock_clm.list_connections.return_value = []
|
|
r = self._put('alice', {'via_cell': 'no-such-cell'})
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_enable_route_via_calls_wg_methods(self, mock_reg, mock_wg, mock_clm):
|
|
mock_reg.get_peer.return_value = _PEER
|
|
mock_reg.set_route_via.return_value = dict(_PEER, route_via='exit-cell')
|
|
mock_clm.list_connections.return_value = [_CELL_LINK]
|
|
r = self._put('alice', {'via_cell': 'exit-cell'})
|
|
self.assertEqual(r.status_code, 200)
|
|
mock_wg.update_cell_peer_allowed_ips.assert_called_once_with(
|
|
'exitcellpubkey=', '10.1.0.0/24', add_default_route=True)
|
|
mock_wg.apply_peer_route_via.assert_called_once_with(
|
|
'10.0.0.5', via_wg_ip='10.1.0.1')
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_enable_route_via_signals_exit_relay_active(self, mock_reg, mock_wg, mock_clm):
|
|
mock_reg.get_peer.return_value = _PEER
|
|
mock_reg.set_route_via.return_value = dict(_PEER, route_via='exit-cell')
|
|
mock_clm.list_connections.return_value = [_CELL_LINK]
|
|
self._put('alice', {'via_cell': 'exit-cell'})
|
|
mock_clm.set_exit_relay_active.assert_called_once_with('exit-cell', True)
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_disable_route_via_clears_old_routing(self, mock_reg, mock_wg, mock_clm):
|
|
peer_with_via = dict(_PEER, route_via='exit-cell')
|
|
mock_reg.get_peer.return_value = peer_with_via
|
|
mock_reg.set_route_via.return_value = dict(_PEER, route_via=None)
|
|
mock_clm.list_connections.return_value = [_CELL_LINK]
|
|
r = self._put('alice', {'via_cell': None})
|
|
self.assertEqual(r.status_code, 200)
|
|
mock_wg.update_cell_peer_allowed_ips.assert_called_once_with(
|
|
'exitcellpubkey=', '10.1.0.0/24', add_default_route=False)
|
|
mock_wg.remove_peer_route_via.assert_called_once_with('10.0.0.5')
|
|
mock_clm.set_exit_relay_active.assert_called_once_with('exit-cell', False)
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_enable_route_via_persists_field(self, mock_reg, mock_wg, mock_clm):
|
|
mock_reg.get_peer.return_value = _PEER
|
|
updated = dict(_PEER, route_via='exit-cell')
|
|
mock_reg.set_route_via.return_value = updated
|
|
mock_clm.list_connections.return_value = [_CELL_LINK]
|
|
r = self._put('alice', {'via_cell': 'exit-cell'})
|
|
self.assertEqual(r.status_code, 200)
|
|
mock_reg.set_route_via.assert_called_once_with('alice', 'exit-cell')
|
|
data = json.loads(r.data)
|
|
self.assertEqual(data['peer']['route_via'], 'exit-cell')
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_switch_exit_cell_removes_old_before_adding_new(self, mock_reg, mock_wg, mock_clm):
|
|
old_link = dict(_CELL_LINK, cell_name='old-exit', public_key='oldpubkey=',
|
|
vpn_subnet='10.2.0.0/24', dns_ip='10.2.0.1')
|
|
new_link = dict(_CELL_LINK, cell_name='new-exit', public_key='newpubkey=')
|
|
peer_with_old = dict(_PEER, route_via='old-exit')
|
|
mock_reg.get_peer.return_value = peer_with_old
|
|
mock_reg.set_route_via.return_value = dict(_PEER, route_via='new-exit')
|
|
mock_clm.list_connections.return_value = [old_link, new_link]
|
|
self._put('alice', {'via_cell': 'new-exit'})
|
|
calls = mock_wg.update_cell_peer_allowed_ips.call_args_list
|
|
# First call removes old (add_default_route=False), second adds new (True)
|
|
self.assertEqual(len(calls), 2)
|
|
self.assertFalse(calls[0].kwargs['add_default_route'])
|
|
self.assertTrue(calls[1].kwargs['add_default_route'])
|
|
# set_exit_relay_active called twice: False for old, True for new
|
|
relay_calls = mock_clm.set_exit_relay_active.call_args_list
|
|
self.assertEqual(len(relay_calls), 2)
|
|
self.assertEqual(relay_calls[0].args, ('old-exit', False))
|
|
self.assertEqual(relay_calls[1].args, ('new-exit', True))
|
|
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_loop_detection_returns_409(self, mock_reg, mock_wg, mock_clm):
|
|
"""If the target cell already routes peers through us (remote_exit_relay_active=True),
|
|
enabling route-via would create an A→B→A loop — expect 409."""
|
|
loop_link = dict(_CELL_LINK, remote_exit_relay_active=True)
|
|
mock_reg.get_peer.return_value = _PEER
|
|
mock_clm.list_connections.return_value = [loop_link]
|
|
r = self._put('alice', {'via_cell': 'exit-cell'})
|
|
self.assertEqual(r.status_code, 409)
|
|
data = json.loads(r.data)
|
|
self.assertIn('loop', data['error'].lower())
|
|
mock_wg.update_cell_peer_allowed_ips.assert_not_called()
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_loop_detection_skipped_when_disabling(self, mock_reg, mock_wg, mock_clm):
|
|
"""Disabling route-via (via_cell=null) must NOT be blocked by the loop check."""
|
|
loop_link = dict(_CELL_LINK, remote_exit_relay_active=True)
|
|
peer_with_via = dict(_PEER, route_via='exit-cell')
|
|
mock_reg.get_peer.return_value = peer_with_via
|
|
mock_reg.set_route_via.return_value = dict(_PEER, route_via=None)
|
|
mock_clm.list_connections.return_value = [loop_link]
|
|
r = self._put('alice', {'via_cell': None})
|
|
self.assertEqual(r.status_code, 200)
|
|
|
|
@patch('app.cell_link_manager')
|
|
@patch('app.wireguard_manager')
|
|
@patch('app.peer_registry')
|
|
def test_no_loop_when_remote_exit_relay_false(self, mock_reg, mock_wg, mock_clm):
|
|
"""remote_exit_relay_active=False means no loop — route-via should succeed."""
|
|
safe_link = dict(_CELL_LINK, remote_exit_relay_active=False)
|
|
mock_reg.get_peer.return_value = _PEER
|
|
mock_reg.set_route_via.return_value = dict(_PEER, route_via='exit-cell')
|
|
mock_clm.list_connections.return_value = [safe_link]
|
|
mock_clm.set_exit_relay_active.return_value = safe_link
|
|
r = self._put('alice', {'via_cell': 'exit-cell'})
|
|
self.assertEqual(r.status_code, 200)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|