91 lines
5.5 KiB
Python
91 lines
5.5 KiB
Python
"""Explicit Linux/root test: native processes in two disposable network namespaces.
|
|
No containers, host routes, production ports, keys or existing WireGuard peers used.
|
|
"""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
ROOT=Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0,str(ROOT))
|
|
A='deck-wg-test-a';B='deck-wg-test-b'
|
|
def run(*args,data=None):
|
|
r=subprocess.run(args,input=data,text=True,capture_output=True,timeout=15)
|
|
if r.returncode:raise RuntimeError('Isolated integration command failed: '+args[0])
|
|
return r.stdout.strip()
|
|
def ns(name,*args,data=None):return run('ip','netns','exec',name,*args,data=data)
|
|
SCRIPT=r'''
|
|
import json,os,signal,socketserver,threading
|
|
from pathlib import Path
|
|
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
|
|
from network.native import Manager,RPC,Server
|
|
policy={'lan_address':'10.241.88.1','lan_interface':'testa','gui_port':18108,'ports':[18108,18120]}
|
|
m=Manager(os.environ['TEST_STATE'],policy)
|
|
class App(BaseHTTPRequestHandler):
|
|
def log_message(self,*args):pass
|
|
def do_GET(self):
|
|
body=b'{"synthetic":true}';self.send_response(200);self.send_header('Content-Length',str(len(body)));self.end_headers();self.wfile.write(body)
|
|
class Stream(socketserver.BaseRequestHandler):
|
|
def handle(self):self.request.sendall(b'synthetic-stream');self.request.close()
|
|
threading.Thread(target=ThreadingHTTPServer(('127.0.0.1',18108),App).serve_forever,daemon=True).start()
|
|
threading.Thread(target=socketserver.ThreadingTCPServer(('127.0.0.1',18120),Stream).serve_forever,daemon=True).start()
|
|
with Server(os.environ['TEST_SOCKET'],RPC) as server:
|
|
server.manager=m;server.client_uid=0
|
|
signal.signal(signal.SIGTERM,lambda *_:threading.Thread(target=server.shutdown,daemon=True).start())
|
|
try:server.serve_forever()
|
|
finally:
|
|
m.disconnect()
|
|
for s in m.listeners:s.shutdown();s.server_close()
|
|
'''
|
|
def main():
|
|
if os.geteuid()!=0:raise SystemExit('Explicit root test on Linux required.')
|
|
process=None
|
|
before=run('ip','-4','route','show')
|
|
with tempfile.TemporaryDirectory(prefix='deck-native-test-') as t:
|
|
try:
|
|
for n in (A,B):run('ip','netns','add',n)
|
|
run('ip','link','add','testa','type','veth','peer','name','testb')
|
|
run('ip','link','set','testa','netns',A);run('ip','link','set','testb','netns',B)
|
|
for n,dev,ip in ((A,'testa','10.241.88.1/24'),(B,'testb','10.241.88.2/24')):
|
|
ns(n,'ip','link','set','lo','up');ns(n,'ip','address','add',ip,'dev',dev);ns(n,'ip','link','set',dev,'up')
|
|
env=dict(os.environ,PYTHONPATH=str(ROOT),TEST_STATE=t,TEST_SOCKET=t+'/rpc.sock')
|
|
process=subprocess.Popen(['ip','netns','exec',A,'python3','-c',SCRIPT],env=env,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
|
|
for _ in range(30):
|
|
if Path(t+'/rpc.sock').exists():break
|
|
time.sleep(.1)
|
|
from network.rpc import request
|
|
def rpc(action,**values):return request(dict(action=action,**values),path=t+'/rpc.sock')
|
|
ka=ns(A,'wg','genkey');kb=ns(B,'wg','genkey');pa=ns(A,'wg','pubkey',data=ka+'\n');pb=ns(B,'wg','pubkey',data=kb+'\n')
|
|
rpc('import',config=f'[Interface]\nPrivateKey = {ka}\nAddress = 10.242.88.1/24, fd42:88::1/64\n[Peer]\nPublicKey = {pb}\nAllowedIPs = 10.242.88.2/32, fd42:88::2/128\nEndpoint = 10.241.88.2:51835\nPersistentKeepalive = 1\n')
|
|
peer=Path(t)/'peer.conf';peer.write_text(f'[Interface]\nPrivateKey = {kb}\nListenPort = 51835\n[Peer]\nPublicKey = {pa}\nAllowedIPs = 10.242.88.1/32, fd42:88::1/128\nEndpoint = 10.241.88.1:51836\nPersistentKeepalive = 1\n');peer.chmod(0o600)
|
|
ns(B,'ip','link','add','peerwg0','type','wireguard');ns(B,'wg','setconf','peerwg0',str(peer));peer.unlink()
|
|
ns(B,'ip','address','add','10.242.88.2/32','dev','peerwg0');ns(B,'ip','link','set','peerwg0','up');ns(B,'ip','route','add','10.242.88.1/32','dev','peerwg0')
|
|
ns(B,'ip','-6','address','add','fd42:88::2/128','dev','peerwg0');ns(B,'ip','-6','route','add','fd42:88::1/128','dev','peerwg0')
|
|
rpc('connect');ns(A,'wg','set','wgdeck0','listen-port','51836')
|
|
for _ in range(40):
|
|
if rpc('status')['state']=='connected':break
|
|
time.sleep(.25)
|
|
else:raise RuntimeError('Synthetic WireGuard handshake missing')
|
|
def http(host):
|
|
if ':' in host:host='['+host+']'
|
|
return int(ns(B,'python3','-c',"import urllib.request,urllib.error;\ntry: print(urllib.request.urlopen('http://"+host+":18108/',timeout=5).status)\nexcept urllib.error.HTTPError as e: print(e.code)"))
|
|
assert http('10.241.88.1')==200;assert http('10.242.88.1')==403
|
|
s=rpc('mode',mode='both');assert http('10.242.88.1')==200;assert http('fd42:88::1')==200
|
|
try:rpc('confirm',trial_id=s['pending']['id'],ingress='tunnel',proxy_token='fake')
|
|
except ValueError:pass
|
|
else:raise AssertionError('Forged ingress accepted')
|
|
token=(Path(t)/'proxy-token').read_text().strip();rpc('confirm',trial_id=s['pending']['id'],ingress='tunnel',proxy_token=token)
|
|
assert ns(B,'python3','-c',"import socket;s=socket.create_connection(('10.242.88.1',18120),5);print(s.recv(1024).decode())")=='synthetic-stream'
|
|
s=rpc('mode',mode='tunnel');assert http('10.241.88.1')==403;assert http('10.242.88.1')==200
|
|
rpc('cancel');assert http('10.241.88.1')==200
|
|
rpc('disconnect');assert rpc('status')['enabled'] is False
|
|
assert run('ip','-4','route','show')==before
|
|
print('PASS native dual-stack handshake/access, LAN/tunnel/both, forged confirmation rejection, streamed API bytes, cancellation, cleanup; host routes unchanged')
|
|
finally:
|
|
if process:process.terminate();process.wait(timeout=15)
|
|
for n in (A,B):subprocess.run(['ip','netns','del',n],capture_output=True)
|
|
if __name__=='__main__':main()
|