import base64 import hashlib import json import os import secrets import threading import unittest import urllib.request import urllib.error from unittest.mock import patch from network.config import parse_config, ConfigError, wireguard_text from network.policy import Policy from server import Server def configuration(extra=''): # Synthetic random fixture, never real credentials. key=lambda:base64.b64encode(secrets.token_bytes(32)).decode() return f'[Interface]\nPrivateKey = {key()}\nAddress = 10.240.77.1/24\n{extra}\n[Peer]\nPublicKey = {key()}\nEndpoint = vpn.example.test:51820\nAllowedIPs = 0.0.0.0/0\n' class ParserTests(unittest.TestCase): def test_valid_sanitized_config(self): config=parse_config(configuration('DNS = 192.168.1.1')) self.assertEqual(config['address'],'10.240.77.1') self.assertEqual(config['allowed_ips'],['0.0.0.0/0']) self.assertEqual(config['keepalive'],25) self.assertTrue(config['warnings']) rendered=wireguard_text(config) for name in ('DNS','Address','Table','PostUp'):self.assertNotIn(name,rendered) def test_reject_hooks_without_echoing_secrets(self): for hook in ('PostUp','PostDown','PreUp','PreDown','Table','SaveConfig'): with self.assertRaises(ConfigError) as e:parse_config(configuration(hook+' = secret-canary')) self.assertNotIn('secret-canary',str(e.exception)) def test_duplicate_invalid_and_oversize(self): samples=[configuration('PrivateKey = invalid'),configuration().replace('10.240.77.1/24','::1/128'),configuration()+'[Peer]\n',configuration().replace('vpn.example.test:51820','$(whoami):22'),'x'*16385] for text in samples: with self.assertRaises(ConfigError):parse_config(text) class PolicyTests(unittest.TestCase): def setUp(self): self.now=0 self.saved=[] self.policy=Policy(clock=lambda:self.now,persist=self.saved.append) def test_disconnect_never_opens_lan(self): p=self.policy p.propose('tunnel',True) trial=p.status()['pending']['id'] self.assertFalse(p.allowed('lan')) with self.assertRaises(ValueError):p.confirm(trial,'management') with self.assertRaises(ValueError):p.confirm(trial,'lan') p.confirm(trial,'tunnel') self.now=999 self.assertFalse(p.allowed('lan')) self.assertEqual(self.saved,['tunnel']) def test_trial_timeout_and_reboot(self): p=self.policy p.propose('tunnel',True) self.now=121 self.assertTrue(p.allowed('lan')) self.assertEqual(self.saved,[]) self.assertIsNone(p.status()['pending']) self.assertEqual(Policy(mode='lan').mode,'lan') def test_no_tunnel_without_handshake(self): for mode in ('tunnel','both'): with self.assertRaises(ValueError):self.policy.propose(mode,False) def test_lan_confirmation_and_cancel(self): p=self.policy p.propose('both',True) p.confirm(p.pending['id'],'tunnel') p.propose('lan',False) with self.assertRaises(ValueError):p.confirm(p.pending['id'],'tunnel') p.cancel() self.assertEqual(p.mode,'both') class APITests(unittest.TestCase): def setUp(self): self.server=Server(0) self.thread=threading.Thread(target=self.server.serve_forever) self.thread.start() self.base=f'http://127.0.0.1:{self.server.server_port}' def tearDown(self): self.server.shutdown();self.server.demo.stop();self.server.server_close();self.thread.join() def post(self,path,body,headers=None): req=urllib.request.Request(self.base+'/api/v1/'+path,data=json.dumps(body).encode(),headers={'Content-Type':'application/json','X-Athena-Deck':'1',**(headers or {})}) return urllib.request.urlopen(req) def test_import_rejects_before_transport(self): with patch.object(self.server.network,'call') as transport: with self.assertRaises(urllib.error.HTTPError) as error:self.post('network/import',{'config':configuration('PostUp = no')}) self.assertEqual(error.exception.code,400);error.exception.close() transport.assert_not_called() def test_cannot_forge_confirm_ingress(self): with patch.object(self.server.network,'call',return_value={}) as transport: with self.post('network/confirm',{'trial_id':'test'}, {'X-Deck-Ingress':'tunnel'}) as result:self.assertEqual(result.status,200) self.assertEqual(transport.call_args.args[2],'management') with self.assertRaises(urllib.error.HTTPError) as error:self.post('network/confirm',{'trial_id':'test','ingress':'tunnel'}) self.assertEqual(error.exception.code,400);error.exception.close() def test_authentication(self): salt=secrets.token_bytes(16) password='test-only-'+secrets.token_hex(20) self.server.auth={'salt':salt.hex(),'hash':hashlib.pbkdf2_hmac('sha256',password.encode(),salt,600000).hex()} with self.assertRaises(urllib.error.HTTPError) as error:urllib.request.urlopen(self.base+'/api/v1/status') self.assertEqual(error.exception.code,401);error.exception.close() with self.post('login',{'password':password}) as response:cookie=response.headers['Set-Cookie'].split(';')[0] req=urllib.request.Request(self.base+'/api/v1/status',headers={'Cookie':cookie}) with urllib.request.urlopen(req) as response:self.assertEqual(response.status,200) with self.assertRaises(urllib.error.HTTPError) as error:self.post('login',{'password':'wrong'}) self.assertEqual(error.exception.code,401);error.exception.close() if __name__=='__main__':unittest.main() class TransportTests(unittest.TestCase): def test_status_with_nullable_error_is_success(self): import subprocess from network.client import NetworkClient with patch.dict(os.environ, {'DECK_NETWORK_SOCKET':'0'}): client=NetworkClient() result={'installed':True,'error':None,'state':'disabled'} with patch('network.client.subprocess.run',return_value=subprocess.CompletedProcess([],0,json.dumps(result),'')): self.assertTrue(client.call('status')['installed']) def test_installer_bundle_is_complete(self): from network.install_remote import FILES from network.client import ROOT for name in FILES: self.assertTrue((ROOT/name).is_file(),name) self.assertNotIn('test_network.py',FILES) def test_install_password_required_before_thread_or_ssh(self): from network.client import NetworkClient with patch.dict(os.environ, {'DECK_NETWORK_SOCKET':'0'}): client=NetworkClient() with patch('network.client.subprocess.run') as remote: with self.assertRaises(ValueError):client.install('short') remote.assert_not_called()