140 lines
7.1 KiB
Python
140 lines
7.1 KiB
Python
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_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()
|