Codex
67 lines
4.1 KiB
Python
67 lines
4.1 KiB
Python
"""Network status must be scoped, bounded and honest about reachability."""
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch,Mock
|
|
import network_status as ns
|
|
from test_app import app
|
|
import test_households as fixture
|
|
|
|
|
|
class NetworkTests(unittest.TestCase):
|
|
setUp=fixture.HouseholdTests.setUp
|
|
stop_server=fixture.HouseholdTests.stop_server
|
|
request=fixture.HouseholdTests.request
|
|
|
|
def test_scoped_checks_cannot_probe_an_arbitrary_host(self):
|
|
device=app.INVENTORY.save(dict(name='网关',deviceType='路由器',managementHost='192.168.8.1'),self.owner)
|
|
endpoint='/api/devices/network?source=device&id='+device['id']
|
|
with patch.object(ns,'probe',return_value=dict(host='192.168.8.1',reachable=True,ports=[80])) as probe:
|
|
result=self.request(endpoint)[1]
|
|
self.assertTrue(result['targets'][0]['reachable'])
|
|
self.assertEqual(result['links'][0]['url'],'http://192.168.8.1/')
|
|
self.assertEqual(self.request(endpoint,self.other)[0],404)
|
|
self.assertEqual(self.request(endpoint,self.members[self.aid])[0],403)
|
|
self.assertEqual(self.request('/api/devices/network?source=device&host=example.com')[0],404)
|
|
again=self.request(endpoint)[1];self.assertTrue(again['cached']);self.assertEqual(probe.call_count,1)
|
|
|
|
def test_record_changes_invalidate_cache(self):
|
|
device=app.INVENTORY.save(dict(name='网关',deviceType='路由器',managementHost='192.168.8.1'),self.owner)
|
|
with patch.object(ns,'probe',side_effect=lambda host:dict(host=host,ports=[],reachable=False)) as probe:
|
|
query=dict(source='device',id=device['id'])
|
|
app.NETWORK.read(self.owner,query)
|
|
app.INVENTORY.save(dict(device,managementHost='192.168.8.2'),self.owner)
|
|
result=app.NETWORK.read(self.owner,query)
|
|
self.assertEqual(probe.call_count,2)
|
|
self.assertFalse(result['targets'][0]['reachable'])
|
|
self.assertFalse(result['links'][0]['verified'])
|
|
self.assertEqual(result['links'][0]['url'],'http://192.168.8.2/')
|
|
|
|
|
|
class NetworkUnitTests(unittest.TestCase):
|
|
def test_rejects_local_metadata_and_unsafe_links(self):
|
|
for host in ['127.0.0.1','169.254.169.254','::1','0.0.0.0','224.0.0.1']:
|
|
self.assertFalse(ns.safe_ip(host))
|
|
self.assertTrue(ns.safe_ip('192.168.8.1'));self.assertTrue(ns.safe_ip('100.64.1.2'))
|
|
for host in ['127.0.0.1','javascript:alert(1)','user:secret@example.com','example.com/path']:
|
|
self.assertEqual(ns.management_links([dict(host=host,kind='local',ports=[80])]),[])
|
|
with patch.object(ns.socket,'getaddrinfo',return_value=[(2,1,6,'',('127.0.0.1',0))]),patch.object(ns.socket,'create_connection') as connect:
|
|
self.assertFalse(ns.probe('malicious.example')['reachable']);connect.assert_not_called()
|
|
|
|
def test_peer_online_does_not_prove_subnet_route(self):
|
|
status={'Peer':{'abc':{'DNSName':'gateway.example.ts.net.','TailscaleIPs':['100.64.1.2'],'Online':True,'Active':False,'AllowedIPs':['100.64.1.2/32'],'PrimaryRoutes':None}}}
|
|
with patch.object(ns.subprocess,'run',return_value=Mock(stdout=json.dumps(status))):
|
|
peer=ns.tail_peer('100.64.1.2','192.168.8.0/24')
|
|
self.assertTrue(peer['online']);self.assertFalse(peer['subnetAvailable']);self.assertNotIn('Peer',peer)
|
|
status['Peer']['abc']['PrimaryRoutes']=['192.168.8.0/24']
|
|
with patch.object(ns.subprocess,'run',return_value=Mock(stdout=json.dumps(status))):
|
|
self.assertTrue(ns.tail_peer('gateway.example.ts.net','192.168.8.0/24')['subnetAvailable'])
|
|
with patch.object(ns.subprocess,'run',side_effect=FileNotFoundError):self.assertFalse(ns.tail_peer('100.64.1.2','')['available'])
|
|
|
|
def test_uses_detected_web_port_and_ipv6_brackets(self):
|
|
links=ns.management_links([dict(host='192.168.3.151',kind='local',ports=[8790]),dict(host='fd7a:115c:a1e0::1',kind='tailscale',ports=[443])])
|
|
self.assertEqual(links[0]['url'],'http://192.168.3.151:8790/')
|
|
self.assertEqual(links[1]['url'],'https://[fd7a:115c:a1e0::1]/')
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|