60 lines
3.3 KiB
Python
60 lines
3.3 KiB
Python
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from network.config import parse_config,ConfigError
|
|
from network.native import Manager
|
|
|
|
CONF='''[Interface]
|
|
PrivateKey = AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=
|
|
Address = 192.0.2.10/24, fd00::10/64
|
|
DNS = 192.0.2.1
|
|
DNS = example.test
|
|
[Peer]
|
|
PublicKey = AgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgI=
|
|
AllowedIPs = 0.0.0.0/0, ::/0
|
|
Endpoint = vpn.example.test:51820
|
|
PersistentKeepalive = 25
|
|
'''
|
|
class NativeTests(unittest.TestCase):
|
|
def manager(self,root):
|
|
with patch.object(Manager,'reconcile'):return Manager(root,dict(lan_address='192.0.2.11',gui_port=8108,ports=[8108,8120]))
|
|
def test_dual_stack_repeated_dns(self):
|
|
c=parse_config(CONF,native=True);self.assertEqual(len(c['addresses']),2);self.assertEqual(c['allowed_ips'],['0.0.0.0/0','::/0']);self.assertTrue(c['warnings'])
|
|
def test_hooks_rejected(self):
|
|
with self.assertRaises(ConfigError):parse_config(CONF.replace('DNS = example.test','PostUp = echo unsafe'),native=True)
|
|
def test_no_key_in_status_and_backup_complete(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m=self.manager(t);m.dispatch(dict(action='import',config=CONF));s=m.status()
|
|
self.assertNotIn('PrivateKey',json.dumps(s));self.assertNotIn('AQEBAQ',json.dumps(s));self.assertEqual(m.dispatch(dict(action='backup-export'))['config'],CONF)
|
|
self.assertEqual((Path(t)/'config.json').stat().st_mode&0o777,0o600)
|
|
def test_confirm_requires_native_ingress_token(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m=self.manager(t)
|
|
with patch.object(m,'connected',return_value=(True,1)):
|
|
s=m.dispatch(dict(action='mode',mode='both'));trial=s['pending']['id']
|
|
with self.assertRaises(ValueError):m.dispatch(dict(action='confirm',trial_id=trial,ingress='tunnel',proxy_token='fake'))
|
|
m.dispatch(dict(action='confirm',trial_id=trial,ingress='tunnel',proxy_token=m.token));self.assertEqual(m.policy.mode,'both')
|
|
def test_duplicate_host_ip_aborts_without_mutation(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m=self.manager(t);m.dispatch(dict(action='import',config=CONF))
|
|
with patch('network.native.run',return_value=json.dumps([{'addr_info':[{'local':'192.0.2.10'}]}])) as command:
|
|
with self.assertRaises(ValueError):m.connect()
|
|
self.assertEqual(command.call_count,1);self.assertFalse(m.enabled)
|
|
def test_source_routes_only_and_owned_cleanup(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
m=self.manager(t);m.dispatch(dict(action='import',config=CONF));calls=[]
|
|
def command(args,check=True):calls.append(args);return '[]' if '-j' in args else ''
|
|
with patch('network.native.run',side_effect=command),patch.object(m,'reconcile'):
|
|
m.connect();self.assertTrue(m.enabled)
|
|
self.assertFalse(any('main' in a or 'default' in a for a in calls))
|
|
up=next(i for i,a in enumerate(calls) if a[:3]==['ip','link','set'])
|
|
route=next(i for i,a in enumerate(calls) if a[:4]==['ip','-4','route','add'])
|
|
self.assertLess(up,route)
|
|
self.assertTrue(any(a[:4]==['ip','-4','rule','add'] and 'from' in a for a in calls))
|
|
self.assertTrue(any(a[:4]==['ip','-6','rule','add'] for a in calls))
|
|
m.disconnect();self.assertFalse(m.enabled)
|
|
self.assertFalse(any('iptables' in a or 'nft' in a for a in calls))
|
|
if __name__=='__main__':unittest.main()
|