403 lines
25 KiB
Python
403 lines
25 KiB
Python
"""Real socket transport regressions; policy/namespace admission is separate."""
|
|
import selectors
|
|
import socket
|
|
import sys
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from c01_network_owner import Bridge
|
|
import c01_network_owner as network
|
|
from c01_endpoints import validate_route
|
|
import c01_supervisor as supervisor
|
|
import c01_network_client as client
|
|
import os
|
|
import json
|
|
import array
|
|
|
|
|
|
class AppointmentTests(unittest.TestCase):
|
|
def test_required_generation_loss_refuses_without_waiting_for_socket_pid_exit(self):
|
|
left,right=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET)
|
|
lease=client.NetworkLease.__new__(client.NetworkLease)
|
|
lease.channel=left;lease.pidfd=os.pidfd_open(os.getpid());lease.owner=network.process(os.getpid())
|
|
lease.policy=SimpleNamespace(routes={'required':{}},check=lambda:None,close=lambda:None)
|
|
lease.protocol=4;lease.required_targets={};lease.required_sockets={};lease.listeners=[]
|
|
lease.required_generations={'required':'a'*64};lease.required_since={'required':0}
|
|
lease.expiry=0;lease.start_deadline=time.clock_gettime_ns(time.CLOCK_BOOTTIME)+3_000_000_000
|
|
try:
|
|
right.send(network.canonical({'alive_until_boottime_ns':time.clock_gettime_ns(time.CLOCK_BOOTTIME)+3_000_000_000,
|
|
'generations':{'required':None}}))
|
|
with self.assertRaisesRegex(ValueError,'generation lost'):lease.check()
|
|
finally:lease.close();right.close()
|
|
def test_dual_stack_inventory_larger_than_scm_limit_is_complete_and_ordered(self):
|
|
left,right=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET)
|
|
listeners=[];sources=[]
|
|
request={'method':'lease','policy_sha256':'a'*64,'nonce':'b'*64}
|
|
try:
|
|
for _ in range(270):sources.append(socket.socket())
|
|
for offset in range(0,len(sources),64):
|
|
rights=array.array('i',[s.fileno() for s in sources[offset:offset+64]])
|
|
right.sendmsg([network.canonical({**request,'offset':offset,'total':len(sources)})],
|
|
[(socket.SOL_SOCKET,socket.SCM_RIGHTS,rights)])
|
|
client.receive_listeners(left,request,len(sources),4,listeners)
|
|
self.assertEqual([os.fstat(fd).st_ino for fd in listeners],
|
|
[os.fstat(s.fileno()).st_ino for s in sources])
|
|
self.assertTrue(all(not os.get_inheritable(fd) for fd in listeners))
|
|
finally:
|
|
for fd in listeners:os.close(fd)
|
|
for s in sources:s.close()
|
|
left.close();right.close()
|
|
|
|
def test_listener_chunk_cannot_reorder_or_change_total(self):
|
|
for change in ({'offset':64},{'total':2}):
|
|
left,right=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET)
|
|
received=[];request={'method':'lease','policy_sha256':'a'*64,'nonce':'b'*64}
|
|
try:
|
|
right.sendmsg([network.canonical({**request,'offset':0,'total':1,**change})],
|
|
[(socket.SOL_SOCKET,socket.SCM_RIGHTS,array.array('i',[right.fileno()]))])
|
|
with self.assertRaisesRegex(ValueError,'binding/rights'):
|
|
client.receive_listeners(left,request,1,4,received)
|
|
finally:
|
|
for fd in received:os.close(fd)
|
|
left.close();right.close()
|
|
|
|
def test_managed_registration_discovers_exact_listener_and_preserves_generation(self):
|
|
with socket.socket() as listener:
|
|
listener.bind(('127.0.0.2',0));listener.listen(1)
|
|
route=dict(target='managed',protocol='tcp',target_address='127.0.0.2',target_port=listener.getsockname()[1],port=61001)
|
|
lease=client.NetworkLease.__new__(client.NetworkLease)
|
|
lease.policy=SimpleNamespace(routes={'gate/fixture':route})
|
|
registered=[]
|
|
def request(body):
|
|
if body['method']=='inspect':return {'producer':None,'generation':'a'*64}
|
|
registered.append(body)
|
|
return {'registered':body['name'],'policy_sha256':'b'*64,'generation':'c'*64}
|
|
lease.request=request
|
|
with patch.object(client.LIBC,'syscall',side_effect=lambda syscall,pidfd,fd,flags:os.dup(fd)):
|
|
lease.register_managed_listeners(os.getpid(),['gate/fixture'])
|
|
self.assertEqual(registered,[{'method':'register','name':'gate/fixture','pid':os.getpid(),
|
|
'fd':listener.fileno(),'previous':'a'*64}])
|
|
|
|
def test_absent_gate_is_not_registered_to_an_unrelated_port(self):
|
|
lease=client.NetworkLease.__new__(client.NetworkLease)
|
|
lease.policy=SimpleNamespace(routes={'gate/fixture':dict(target='managed',protocol='tcp',target_address='127.0.0.2',target_port=1,port=61001)})
|
|
calls=[]
|
|
def request(body):
|
|
calls.append(body)
|
|
return {'producer':None,'generation':None}
|
|
lease.request=request
|
|
with patch.object(client.LIBC,'syscall',side_effect=lambda syscall,pidfd,fd,flags:os.dup(fd)):
|
|
lease.register_managed_listeners(os.getpid(),['gate/fixture'])
|
|
self.assertEqual(calls,[{'method':'inspect','name':'gate/fixture'}])
|
|
|
|
def test_gate_relay_cannot_bypass_authentication_or_change_producer(self):
|
|
group='/system.slice/archipelago.service/c01-application'
|
|
runtime={'manager_uid':61001}
|
|
manifests={'a'*64:dict(app_id='fixture',loopback_ports=[['tcp',7778]],target_cgroup=group)}
|
|
route=dict(name='gate/fixture/tcp/61001/host',source='host',target='managed',protocol='tcp',
|
|
port=61001,target_port=7778,app_id='fixture',projection_sha256='a'*64,
|
|
target_uid=61001,target_cgroup=group,source_addresses=['127.0.0.1'],target_address='127.0.0.2')
|
|
self.assertEqual(validate_route(route,manifests,runtime,address_version=4),route)
|
|
ingress={**route,'name':'gate/fixture/tcp/7778/host','port':7778,
|
|
'source_addresses':['192.168.63.100','fd79:1aa:b9e9:4c9f:1f80:5376:9385:1824']}
|
|
self.assertEqual(validate_route(ingress,manifests,runtime,address_version=4),ingress)
|
|
for change in ({'target':'runtime'},{'target_address':'127.0.0.1'},
|
|
{'target_port':7779},{'target_uid':1000},{'projection_sha256':'b'*64},
|
|
{'source_addresses':['0.0.0.0']},{'target_cgroup':'/system.slice/other.service'}):
|
|
with self.subTest(change=change),self.assertRaises(ValueError):
|
|
validate_route({**route,**change},manifests,runtime,address_version=4)
|
|
|
|
def test_native_peer_interface_is_fixed(self):
|
|
for name,address,port in (('peer','127.0.0.1',5679),):
|
|
route=dict(name=f'native/{name}/tcp/5679/host',source='host',target='managed',protocol='tcp',
|
|
port=5679,target_port=port,app_id=None,projection_sha256=None,target_uid=61001,
|
|
target_cgroup='/system.slice/archipelago.service/c01-application',
|
|
source_addresses=['fd00::1'],target_address=address)
|
|
self.assertEqual(validate_route(route,{}, {'manager_uid':61001},address_version=4),route)
|
|
for change in ({'target_port':5678},{'target_uid':1000},{'target':'runtime'},
|
|
{'target_address':'192.168.1.1'},{'protocol':'udp'}):
|
|
with self.subTest(name=name,change=change),self.assertRaises(ValueError):
|
|
validate_route({**route,**change},{},{'manager_uid':61001},address_version=4)
|
|
|
|
def test_8443_preserves_fips_transport_not_fabricated_https(self):
|
|
route=dict(name='fips-tcp6',source='host',target='host',protocol='tcp',port=8443,target_port=8443,
|
|
app_id=None,projection_sha256=None,target_uid=0,target_cgroup='/system.slice/archipelago-fips.service',
|
|
source_addresses=['::'],target_address='127.0.0.1')
|
|
self.assertEqual(validate_route(route,{},address_version=4),route)
|
|
for change in ({'target':'managed'},{'target_uid':61001},{'target_port':5678},
|
|
{'target_cgroup':'/system.slice/nginx.service'},{'source_addresses':['0.0.0.0']},
|
|
{'name':'native/https/tcp/8443/host'}):
|
|
with self.subTest(change=change),self.assertRaises(ValueError):validate_route({**route,**change},{},address_version=4)
|
|
|
|
def test_website_publication_keeps_fips_and_tor_exposures_separate(self):
|
|
for kind,port,addresses in (('website-fips',32000,['fd00::1']),('website-tor',32100,['127.0.0.1'])):
|
|
route=dict(name=f'native/{kind}/tcp/{port}/host',source='host',target='managed',protocol='tcp',
|
|
port=port,target_port=port,app_id=None,projection_sha256=None,target_uid=61001,
|
|
target_cgroup='/system.slice/archipelago.service/c01-application',source_addresses=addresses,target_address='127.0.0.1')
|
|
self.assertEqual(validate_route(route,{}, {'manager_uid':61001},address_version=4),route)
|
|
with self.assertRaises(ValueError):validate_route({**route,'source_addresses':['192.168.1.1']},{},{'manager_uid':61001},address_version=4)
|
|
|
|
def test_kiosk_web_routes_only_reach_fixed_host_nginx(self):
|
|
runtime = {'runtime_uid': 1000, 'user_cgroup': '/user.slice/user-1000.slice/user@1000.service'}
|
|
for name, port in (('kiosk-http', 80), ('kiosk-https', 443)):
|
|
route = dict(name=name, source='runtime', target='host', protocol='tcp', port=port,
|
|
app_id=None, projection_sha256=None, target_uid=0, target_cgroup='/system.slice/nginx.service',
|
|
source_addresses=['127.0.0.1'], target_address='127.0.0.1')
|
|
self.assertEqual(validate_route(route, {}, runtime, address_version=3), route)
|
|
with self.assertRaises(ValueError):validate_route(route, {}, address_version=3)
|
|
for change in ({'port':22}, {'target_uid':1000}, {'protocol':'udp'},
|
|
{'target_cgroup':'/system.slice/ssh.service'}, {'source':'managed'},
|
|
{'source_addresses':['0.0.0.0']}, {'target_address':'127.0.0.2'},
|
|
{'app_id':'nginx'}, {'projection_sha256':'a'*64}):
|
|
with self.subTest(name=name, change=change), self.assertRaises(ValueError):
|
|
validate_route({**route, **change}, {}, runtime, address_version=3)
|
|
|
|
def test_kiosk_cdp_uses_existing_exact_runtime_projection(self):
|
|
runtime = {'runtime_uid':1000,'user_cgroup':'/user.slice/user-1000.slice/user@1000.service'}
|
|
group = runtime['user_cgroup'] + '/app.slice/archipelago-kiosk-browser.service'
|
|
route = dict(name='app/archipelago-kiosk/tcp/9222/managed', source='managed', target='runtime',
|
|
protocol='tcp', port=9222, app_id='archipelago-kiosk', projection_sha256='a'*64,
|
|
target_uid=1000, target_cgroup=group)
|
|
manifests = {'a'*64:dict(app_id='archipelago-kiosk',loopback_ports=[['tcp',9222]],target_cgroup=group)}
|
|
self.assertEqual(validate_route(route, manifests, runtime), route)
|
|
with self.assertRaises(ValueError):validate_route(route, {}, runtime)
|
|
with self.assertRaises(ValueError):validate_route({**route,'target_cgroup':'/system.slice/archipelago-kiosk.service'}, manifests, runtime)
|
|
|
|
def test_network_lease_rejects_buffered_expired_progress(self):
|
|
left,right=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET)
|
|
lease=client.NetworkLease.__new__(client.NetworkLease)
|
|
lease.channel=left;lease.pidfd=os.pidfd_open(os.getpid());lease.owner=network.process(os.getpid())
|
|
lease.policy=SimpleNamespace(check=lambda:None,close=lambda:None)
|
|
lease.required_targets={};lease.listeners=[];lease.expiry=0
|
|
lease.start_deadline=time.clock_gettime_ns(time.CLOCK_BOOTTIME)+3_000_000_000
|
|
try:
|
|
right.send(json.dumps({'alive_until_boottime_ns':time.clock_gettime_ns(time.CLOCK_BOOTTIME)+2_000_000_000}).encode())
|
|
lease.check()
|
|
lease.expiry=0;lease.start_deadline=0
|
|
right.send(json.dumps({'alive_until_boottime_ns':time.clock_gettime_ns(time.CLOCK_BOOTTIME)-1}).encode())
|
|
with self.assertRaisesRegex(ValueError,'progress expired'):lease.check()
|
|
finally:lease.close();right.close()
|
|
|
|
def test_runtime_route_uses_appointed_uid_and_exact_manifest_producer(self):
|
|
runtime = {'runtime_uid': 21002, 'user_cgroup': '/user.slice/user-21002.slice/user@21002.service'}
|
|
group = runtime['user_cgroup'] + '/app.slice/ollama.service'
|
|
route = dict(name='ollama', source='managed', target='runtime', protocol='tcp', port=11434,
|
|
app_id='ollama', projection_sha256='a'*64, target_uid=21002, target_cgroup=group)
|
|
manifests = {'a'*64: dict(app_id='ollama', loopback_ports=[['tcp',11434]], target_cgroup=group)}
|
|
self.assertEqual(validate_route(route, manifests, runtime), route)
|
|
for changes in ({'target_uid':1000}, {'port':11435}, {'app_id':'other'},
|
|
{'target_cgroup':runtime['user_cgroup']+'/app.slice/other.service'}):
|
|
with self.assertRaises(ValueError):validate_route({**route, **changes}, manifests, runtime)
|
|
with self.assertRaises(ValueError):validate_route(route, manifests)
|
|
|
|
def test_same_cgroup_descendant_is_not_root_mainpid(self):
|
|
actor = dict(group='/system.slice/archipelago.service/supervisor', uids=[0]*4, gids=[0]*4)
|
|
with patch.object(network, 'process', return_value=actor), patch.object(network.subprocess, 'run',
|
|
return_value=SimpleNamespace(returncode=0, stdout='123\n')):
|
|
network.require_main(123, 'archipelago.service', actor['group'])
|
|
with self.assertRaises(PermissionError):network.require_main(124, 'archipelago.service', actor['group'])
|
|
actor['gids']=[1000]*4
|
|
with self.assertRaises(PermissionError):network.require_main(123, 'archipelago.service', actor['group'])
|
|
|
|
def test_reviewed_legacy_loopback_addresses_do_not_create_arbitrary_host_routes(self):
|
|
runtime={'runtime_uid':1000,'user_cgroup':'/user.slice/user-1000.slice/user@1000.service'}
|
|
group=runtime['user_cgroup']+'/app.slice/fixture.service'
|
|
route=dict(name='app/fixture/tcp/7778/host',source='host',target='runtime',protocol='tcp',port=7778,
|
|
app_id='fixture',projection_sha256='a'*64,target_uid=1000,target_cgroup=group,
|
|
source_addresses=['127.0.0.1','127.0.0.2'],target_address='127.0.0.2')
|
|
manifests={'a'*64:dict(app_id='fixture',loopback_ports=[['tcp',7778]],target_cgroup=group)}
|
|
self.assertEqual(validate_route(route,manifests,runtime,address_version=3),route)
|
|
with self.assertRaises(ValueError):validate_route(route,manifests,runtime)
|
|
for change in ({'source_addresses':['0.0.0.0']},{'source_addresses':['127.0.0.2']*2},
|
|
{'target_address':'192.168.63.1'},{'target_address':'127.0.0.3'},
|
|
{'target_uid':0},{'target_cgroup':runtime['user_cgroup']+'/app.slice/other.service'}):
|
|
with self.subTest(change=change),self.assertRaises(ValueError):
|
|
validate_route({**route,**change},manifests,runtime,address_version=3)
|
|
|
|
def test_network_loss_stops_roles_but_keeps_lease_until_reaped(self):
|
|
events=[]
|
|
owner=supervisor.Supervisor.__new__(supervisor.Supervisor)
|
|
children=SimpleNamespace(stopping=False, results={}, children={'application':object()}, reserved={},
|
|
poll=lambda: [], close=lambda:events.append('reaped'))
|
|
def stop():
|
|
children.stopping=True
|
|
events.append('stop')
|
|
children.request_stop=stop
|
|
owner.children=children;owner.watch=None;owner.failure=None;owner.completed_one_shots=set()
|
|
owner.power_lease=None;owner.power_channel=None;owner.prior_power_signal=supervisor.signal.SIG_DFL
|
|
owner.network_settings_watch=None;owner.tor_watch=None
|
|
owner.runtime_lease=None;owner.access_lease=None
|
|
def lost():raise ValueError('owner died')
|
|
owner.network_lease=SimpleNamespace(check=lost,close=lambda:events.append('lease closed'))
|
|
self.assertFalse(owner.tick())
|
|
self.assertEqual(owner.failure, 'network lease lost')
|
|
self.assertNotIn('lease closed',events)
|
|
children.children={}
|
|
owner.prior_wakeup=-1;owner.wake_read=10;owner.wake_write=11;owner.notifier=None
|
|
owner.lock=SimpleNamespace(close=lambda:events.append('lock closed'))
|
|
with patch.object(supervisor.signal,'set_wakeup_fd'), patch.object(supervisor.signal,'signal'), patch.object(supervisor.os,'close'):
|
|
owner.retire()
|
|
self.assertLess(events.index('reaped'),events.index('lease closed'))
|
|
self.assertLess(events.index('lease closed'),events.index('lock closed'))
|
|
|
|
|
|
class RegistrationTests(unittest.TestCase):
|
|
def test_live_process_cannot_keep_closed_producer_socket_admitted(self):
|
|
original=socket.socket();original.bind(('127.0.0.1',0));original.listen(1)
|
|
target=network.Target.__new__(network.Target)
|
|
target.guard=None
|
|
target.number=original.fileno();target.listener=socket.socket(fileno=os.dup(original.fileno()))
|
|
target.pidfd=os.pidfd_open(os.getpid());target.group=None
|
|
target.identity={'pid':os.getpid()};target.route={'target_cgroup':'/fixture'};target.actual_group='/fixture';target.group_id=(1,2)
|
|
try:
|
|
with patch.object(network,'process',return_value=target.identity),patch.object(network.os,'stat',return_value=SimpleNamespace(st_dev=1,st_ino=2)):
|
|
target.check()
|
|
original.close()
|
|
with self.assertRaises((FileNotFoundError,ValueError)):target.check()
|
|
finally:original.close();target.close()
|
|
|
|
def test_manager_retirement_does_not_replace_external_service_producers(self):
|
|
bridge=Bridge(SimpleNamespace(routes={},check=lambda:None))
|
|
retired=[]
|
|
try:
|
|
for name,realm in (('management','managed'),('tor-socks','host'),('lnd-rest','runtime')):
|
|
bridge.targets[name]=SimpleNamespace(route={'target':realm},close=lambda name=name:retired.append(name))
|
|
before=bridge.targets['tor-socks']
|
|
bridge.retire_management_lease()
|
|
self.assertEqual(retired,['management'])
|
|
self.assertIs(bridge.targets['tor-socks'],before)
|
|
self.assertIn('lnd-rest',bridge.targets)
|
|
finally:bridge.close()
|
|
|
|
def test_dead_producer_requires_explicit_generation_for_replacement(self):
|
|
policy = SimpleNamespace(routes={}, check=lambda: None, value={'version':1})
|
|
bridge = Bridge(policy)
|
|
policy.routes['fixture'] = {'target': 'runtime'}
|
|
def target(token, pid):
|
|
return SimpleNamespace(token=token, identity={'pid': pid},
|
|
check=lambda: None, close=lambda: None)
|
|
first = target('a' * 64, 100)
|
|
second = target('b' * 64, 101)
|
|
try:
|
|
with patch.object(network, 'Target', return_value=first):
|
|
self.assertEqual(bridge.register('fixture', 100, 3), 'a' * 64)
|
|
self.assertEqual(bridge.registration_status('fixture')['producer'], {'pid': 100})
|
|
with patch.object(network, 'Target', return_value=second):
|
|
with self.assertRaisesRegex(ValueError, 'live endpoint'):
|
|
bridge.register('fixture', 101, 3, 'a' * 64)
|
|
bridge.retire('fixture')
|
|
self.assertEqual(bridge.registration_status('fixture'),
|
|
{'name': 'fixture', 'generation': 'a' * 64, 'producer': None})
|
|
with self.assertRaisesRegex(ValueError, 'generation changed'):
|
|
bridge.register('fixture', 101, 3)
|
|
self.assertEqual(bridge.register('fixture', 101, 3, 'a' * 64), 'b' * 64)
|
|
with self.assertRaisesRegex(ValueError, 'generation changed'):
|
|
bridge.register('fixture', 101, 3, 'a' * 64)
|
|
finally:
|
|
bridge.close()
|
|
|
|
|
|
class StreamTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = Bridge(SimpleNamespace(routes={}, check=lambda: None))
|
|
self.a, self.client = socket.socketpair()
|
|
self.b, self.server = socket.socketpair()
|
|
for channel in (self.a, self.b, self.client, self.server):
|
|
channel.setblocking(False)
|
|
self.bridge.targets['fixture'] = SimpleNamespace(check=lambda: None, close=lambda: None)
|
|
for channel, peer in ((self.a, self.b), (self.b, self.a)):
|
|
self.bridge.streams[channel] = dict(peer=peer, name='fixture', buffer=bytearray(),
|
|
connecting=False, eof=False, deadline=time.monotonic()+120)
|
|
self.bridge.selector.register(channel, selectors.EVENT_READ, ('stream', 'fixture'))
|
|
|
|
def tearDown(self):
|
|
self.bridge.close()
|
|
self.client.close()
|
|
self.server.close()
|
|
|
|
def test_one_way_progress_renews_both_stream_lifetimes(self):
|
|
self.bridge.streams[self.a]['deadline'] = time.monotonic()-1
|
|
self.server.send(b'active media')
|
|
self.bridge.poll(0)
|
|
self.assertEqual(len(self.bridge.streams), 2)
|
|
for record in self.bridge.streams.values():
|
|
self.assertGreater(record['deadline'], time.monotonic())
|
|
self.bridge.poll(0)
|
|
self.assertEqual(self.client.recv(100), b'active media')
|
|
|
|
def test_dual_half_close_drains_buffered_bytes_before_retirement(self):
|
|
payload = b'retained final bytes' * 2048
|
|
self.bridge.streams[self.a]['buffer'].extend(payload)
|
|
self.bridge.streams[self.a]['eof'] = True
|
|
self.bridge.streams[self.b]['eof'] = True
|
|
self.bridge.selector.modify(self.a, selectors.EVENT_WRITE, ('stream', 'fixture'))
|
|
received = bytearray()
|
|
deadline = time.monotonic()+3
|
|
while time.monotonic() < deadline:
|
|
self.bridge.poll(0)
|
|
try:
|
|
block = self.client.recv(65536)
|
|
if not block:
|
|
break
|
|
received.extend(block)
|
|
except BlockingIOError:
|
|
pass
|
|
self.assertEqual(bytes(received), payload)
|
|
self.assertEqual(self.bridge.streams, {})
|
|
|
|
def test_idle_streams_are_retired(self):
|
|
self.bridge.streams[self.a]['deadline'] = time.monotonic()-1
|
|
self.bridge.poll(0)
|
|
self.assertEqual(self.bridge.streams, {})
|
|
|
|
|
|
class AddressTransportTests(unittest.TestCase):
|
|
def test_tcp_forwarding_reaches_the_preserved_second_loopback_address(self):
|
|
self.forwarding('127.0.0.2',socket.AF_INET)
|
|
|
|
def test_tcp_forwarding_reaches_ipv6_on_distinct_target_port(self):
|
|
self.forwarding('::1',socket.AF_INET6,translated=True)
|
|
|
|
def test_ipv6_ingress_reaches_the_authenticated_second_loopback_target(self):
|
|
self.forwarding('127.0.0.2',socket.AF_INET,translated=True,source_address='::1')
|
|
|
|
def forwarding(self,address,family,translated=False,source_address='127.0.0.1'):
|
|
with socket.socket(family,socket.SOCK_STREAM) as server:
|
|
server.bind((address,0));server.listen(1);server.setblocking(False)
|
|
port=server.getsockname()[1]
|
|
route=dict(source='managed',target='runtime',protocol='tcp',port=0 if translated else port,
|
|
target_port=port,source_addresses=[source_address],target_address=address)
|
|
policy=SimpleNamespace(routes={'fixture':route},check=lambda:None,
|
|
socket=lambda role,kind=socket.SOCK_STREAM,family=socket.AF_INET:socket.socket(family,kind))
|
|
bridge=Bridge(policy);remote=None
|
|
source_port=next(iter(bridge.listeners)).getsockname()[1]
|
|
bridge.targets['fixture']=SimpleNamespace(route=route,check=lambda:None,close=lambda:None)
|
|
try:
|
|
with socket.create_connection((source_address,source_port)) as caller:
|
|
caller.setblocking(False);caller.sendall(b'preserved-loopback')
|
|
result=b'';end=time.monotonic()+3
|
|
while time.monotonic()<end and result!=b'ack:preserved-loopback':
|
|
bridge.poll(.001)
|
|
if remote is None:
|
|
try:remote,_=server.accept();remote.setblocking(False)
|
|
except BlockingIOError:pass
|
|
if remote is not None:
|
|
try:
|
|
received=remote.recv(1024)
|
|
if received:remote.sendall(b'ack:'+received)
|
|
except BlockingIOError:pass
|
|
try:result+=caller.recv(1024)
|
|
except BlockingIOError:pass
|
|
self.assertEqual(result,b'ack:preserved-loopback')
|
|
finally:
|
|
bridge.close()
|
|
if remote is not None:remote.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|