from pathlib import Path import base64 import hashlib import json import os import secrets import threading import unittest import tempfile import secrets import time 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.state=tempfile.TemporaryDirectory() self.server=Server(0,state_dir=self.state.name) self.password=secrets.token_urlsafe(32) self.api_token=secrets.token_urlsafe(32) record=self.server.credentials.setup(self.password,self.api_token) self.server.sessions['test-session']=dict(expires=time.monotonic()+300,revision=record['password']['hash']) self.cookie='deck_session=test-session' 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.endpoint.close();self.server.server_close();self.thread.join();self.state.cleanup() 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','Cookie':self.cookie,**(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): password=self.password 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_native_status_with_nullable_error_is_success(self): from network.client import NetworkClient with tempfile.TemporaryDirectory() as t: token=Path(t)/'proxy-token';token.write_text('synthetic-proxy-token') with patch.dict(os.environ, {'DECK_NETWORK_MODE':'native','DECK_PROXY_TOKEN_FILE':str(token)}), patch('network.client.request',return_value={'installed':True,'error':None,'state':'disabled'}) as transport: client=NetworkClient() self.assertTrue(client.call('status')['installed']) self.assertEqual(transport.call_args.kwargs['path'],'/run/athena-deck-network/control.sock') def test_native_installer_bundle_is_complete(self): from deploy.install import FILES,ROOT for name in ('network/native.py','deploy/setup_network_helper.py','network/config.py','network/client.py'): self.assertIn(name,FILES);self.assertTrue((ROOT/name).is_file()) def test_no_legacy_container_install(self): from network.client import NetworkClient client=NetworkClient() with self.assertRaises(ValueError):client.install('short')