"""Read-only fake-host tests for complete observations and proven pool boundaries.""" import copy import json import os import sys import unittest from unittest.mock import patch from network_inventory import HOST_PROBE, INFO_FORMAT, InventoryFailure, collect, network_record from network_pools import (BUILTIN_VERSIONS, address_exclusions, decode, effective_pools, resolver_exclusions, route_exclusions, validate_rules) NETWORK_ID = 'a' * 64 NETWORK = {'Id': NETWORK_ID, 'Name': 'other_default', 'Driver': 'bridge', 'Scope': 'local', 'Internal': False, 'Attachable': False, 'EnableIPv6': False, 'Containers': {}, 'IPAM': {'Driver': 'default', 'Config': [{'Subnet': '172.18.0.0/16', 'Gateway': '172.18.0.1'}], 'Options': None}, 'Options': {}, 'Labels': {'com.docker.compose.project': 'other'}} RULES = [{'priority': priority, 'src': 'all', 'table': table} for priority, table in ((0, 'local'), (32766, 'main'), (32767, 'default'))] ROUTES = [{'dst': 'default', 'gateway': '10.0.0.1', 'dev': 'eth0'}, {'dst': '10.0.0.0/24', 'dev': 'eth0'}, {'dst': '192.168.88.0/24', 'table': 200, 'dev': 'vpn0'}] ADDRESSES = [{'ifname': 'eth0', 'addr_info': [{'family': 'inet', 'local': '10.0.0.2', 'prefixlen': 24}]}] class FakeHost: def __init__(self): self.calls, self.churn, self.listings = [], False, 0 self.move_route = False self.rows = [copy.deepcopy(NETWORK)] self.answers = { ('docker', 'context', 'inspect', '--format', '{{json .Endpoints.docker.Host}}'): 'unix:///var/run/docker.sock', (sys.executable, '-c', HOST_PROBE, '/var/run/docker.sock'): {'local_rootful': True, 'same_namespace': True}, ('docker', 'info', '--format', INFO_FORMAT): {'version': '28.2.2', 'os': 'linux', 'security': ['name=seccomp,profile=builtin'], 'pools': None}, ('docker', 'compose', 'version', '--short'): b'2.39.4\n', ('ip', '-j', '-4', 'route', 'show', 'table', 'all'): copy.deepcopy(ROUTES), ('ip', '-j', '-4', 'rule', 'show'): copy.deepcopy(RULES), ('ip', '-j', '-4', 'address', 'show'): copy.deepcopy(ADDRESSES), ('cat', '/etc/resolv.conf'): b'nameserver 172.21.22.23\n', } def __call__(self, argv, seconds=10): self.calls.append(tuple(argv)) if argv[:3] == ['docker', 'network', 'ls']: self.listings += 1 ids = [row['Id'] for row in self.rows] if self.churn and self.listings > 1: ids.append('b' * 64) return ('\n'.join(ids) + '\n').encode() if argv[:3] == ['docker', 'network', 'inspect']: return json.dumps([row for row in self.rows if row['Id'] in argv[3:]]).encode() value = self.answers[tuple(argv)] if self.move_route and argv == ['ip', '-j', '-4', 'route', 'show', 'table', 'all'] and self.listings: value = copy.deepcopy(value) value[-1]['dev'] = 'vpn1' if isinstance(value, Exception): raise value return value if isinstance(value, bytes) else json.dumps(value).encode() class InventoryTests(unittest.TestCase): def test_complete_private_inventory_excludes_other_projects_and_all_tables(self): host = FakeHost() with patch.dict(os.environ, {}, clear=True): result = collect(host) self.assertTrue(result['complete']) self.assertEqual(result['docker_version'], '28.2.2') self.assertEqual(result['python_version'], '.'.join(str(part) for part in sys.version_info[:3])) self.assertEqual(result['pool_source'], 'versioned_moby_builtin') self.assertIn('172.18.0.0/16', result['excluded']) self.assertNotIn('172.18.0.0/16', result['host_excluded']) self.assertIn('192.168.88.0/24', result['host_excluded']) self.assertIn('172.21.22.23/32', result['host_excluded']) self.assertIn('10.0.0.1/32', result['host_excluded']) self.assertEqual(result['networks'][0]['Id'], NETWORK_ID) self.assertFalse(any('create' in call or 'rm' in call or 'prune' in call for call in host.calls)) def test_pool_provenance_exact_versions_custom_pools_and_diagnostic_uncertainty(self): self.assertIn('20.10.24', BUILTIN_VERSIONS) self.assertIn('28.5.2', BUILTIN_VERSIONS) pools, source = effective_pools([{'Base': '10.23.0.0/16', 'Size': 24}], '99.1.1') self.assertEqual((pools, source), ([{'base': '10.23.0.0/16', 'size': 24}], 'engine_info')) for bad in (None, []): with self.assertRaisesRegex(InventoryFailure, 'pool_provenance_unknown'): effective_pools(bad, '28.5.2-custom') for bad in ([{'Base': '10.1.2.3/16', 'Size': 24}], [{'Base': '8.0.0.0/8', 'Size': 24}], [{'Base': '10.0.0.0/8', 'Size': True}], [{'base': '10.0.0.0/8', 'size': 24}]): with self.subTest(bad=bad), self.assertRaises(InventoryFailure): effective_pools(bad, '28.5.2') host = FakeHost() host.answers[('docker', 'info', '--format', INFO_FORMAT)]['version'] = '99.1.1' with patch.dict(os.environ, {}, clear=True): result = collect(host, require_pools=False) self.assertEqual(result['docker_version'], '99.1.1') self.assertEqual(result['pool_failure'], 'pool_provenance_unknown') self.assertEqual(result['pools'], []) def test_refuses_churn_incomplete_inspection_bad_json_and_command_failure(self): host = FakeHost() host.churn = True with patch.dict(os.environ, {}, clear=True), self.assertRaisesRegex( InventoryFailure, 'network_inventory_changed'): collect(host) for key, value in ((('ip', '-j', '-4', 'route', 'show', 'table', 'all'), []), (('ip', '-j', '-4', 'rule', 'show'), RuntimeError('private command detail')), (('docker', 'info', '--format', INFO_FORMAT), b'{"version":1,"version":2}')): host = FakeHost() host.answers[key] = value with self.subTest(key=key), patch.dict(os.environ, {}, clear=True), self.assertRaises(InventoryFailure): collect(host) for value in (b'\xff', b'{"v":NaN}', b'{"v":1,"v":2}', b'{} private credential text'): with self.subTest(value=value), self.assertRaises(InventoryFailure): decode(value) for field in ('IPAM', 'Id', 'Containers', 'Attachable'): row = copy.deepcopy(NETWORK) row.pop(field) with self.subTest(field=field), self.assertRaises(InventoryFailure): network_record(row) def test_host_and_engine_scope_require_positive_proof(self): for env in ({'DOCKER_HOST': 'tcp://elsewhere:2375'}, {'DOCKER_CONTEXT': 'desktop-linux'}, {'DOCKER_CONFIG': '/private/config'}, {'DOCKER_API_VERSION': '1.41'}): with self.subTest(env=env), patch.dict(os.environ, env, clear=True), self.assertRaisesRegex( InventoryFailure, 'docker_endpoint_unsupported'): collect(FakeHost()) for proof in ({'local_rootful': False, 'same_namespace': True}, {'local_rootful': True, 'same_namespace': False}, {'local_rootful': True}, {'local_rootful': 1, 'same_namespace': 1}): host = FakeHost() host.answers[(sys.executable, '-c', HOST_PROBE, '/var/run/docker.sock')] = proof with self.subTest(proof=proof), patch.dict(os.environ, {}, clear=True), self.assertRaisesRegex( InventoryFailure, 'host_namespace_unverified'): collect(host) host = FakeHost() host.answers[('docker', 'info', '--format', INFO_FORMAT)]['security'] = ['name=rootless'] with patch.dict(os.environ, {}, clear=True), self.assertRaisesRegex( InventoryFailure, 'docker_engine_unsupported'): collect(host) def test_routes_peers_rules_and_malformed_prefixes(self): rows = copy.deepcopy(ROUTES) rows.append({'dst': '172.24.0.0/24', 'nexthops': [{'gateway': '10.0.0.3', 'dev': 'eth0', 'weight': 1}]}) self.assertIn('10.0.0.3/32', route_exclusions(rows)) rows = copy.deepcopy(ADDRESSES) rows[0]['addr_info'][0]['peer'] = '172.25.0.7/32' self.assertIn('172.25.0.7/32', address_exclusions(rows)) for bad in ([{'dst': 'bad-network'}], [{'dst': '10.0.0.0/8', 'nhid': 1}], [{'gateway': '10.0.0.1'}], [{'dst': '10.0.0.0/8', 'via': 'malformed'}], [{'dst': 'default', 'table': 52, 'dev': 'tailscale0'}]): with self.subTest(bad=bad), self.assertRaises(InventoryFailure): route_exclusions(bad) for bad in ([], RULES + [{'priority': 1, 'table': 100, 'src': 'all'}], [{'priority': row['priority'], 'table': row['table'], 'src': '10.0.0.0/8'} for row in RULES]): with self.subTest(bad=bad), self.assertRaisesRegex(InventoryFailure, 'host_rules_unsupported'): validate_rules(bad) marked = {'src': 'all', 'fwmark': '0x80000', 'fwmask': '0xff0000'} tailscale = RULES + [dict(marked, priority=5210, table='main'), dict(marked, priority=5230, table='default'), dict(marked, priority=5250, action='unreachable'), {'priority': 5270, 'src': 'all', 'table': 52}] validate_rules(tailscale) tailscale[-1]['table'] = 100 with self.assertRaises(InventoryFailure): validate_rules(tailscale) host = FakeHost() host.answers[('ip', '-j', '-4', 'rule', 'show')] = tailscale with patch.dict(os.environ, {}, clear=True), self.assertRaises(InventoryFailure) as failure: collect(host, require_pools=False) self.assertEqual(failure.exception.details['docker_version'], '28.2.2') self.assertEqual(failure.exception.details['complete'], False) with self.assertRaises(InventoryFailure): address_exclusions([{'addr_info': [{'family': 'inet', 'local': 'bad', 'prefixlen': 24}]}]) def test_systemd_resolver_upstream_and_unknown_local_stubs(self): host = FakeHost() host.answers[('cat', '/etc/resolv.conf')] = b'nameserver 127.0.0.53\n' host.answers[('resolvectl', 'dns')] = b'Global:\nLink 2 (eth0): 172.22.99.1 1.1.1.1\n' with patch.dict(os.environ, {}, clear=True): result = collect(host) self.assertIn('172.22.99.1/32', result['excluded']) for raw in (b'nameserver 127.0.0.1\n', b'nameserver garbage\n', b'nameserver 1.1.1.1 extra\n', b''): with self.subTest(raw=raw), self.assertRaises(InventoryFailure): resolver_exclusions(raw) with self.assertRaises(InventoryFailure): resolver_exclusions(b'Global: 127.0.0.53\n', resolved=True) self.assertEqual(resolver_exclusions(b'nameserver 2001:4860:4860::8888\n'), (set(), False)) def test_bounds_batches_ipv6_and_full_allocated_subnets(self): host = FakeHost() host.rows = [dict(copy.deepcopy(NETWORK), Id=('%064x' % number)) for number in range(40)] with patch.dict(os.environ, {}, clear=True): result = collect(host) calls = [call for call in host.calls if call[:3] == ('docker', 'network', 'inspect')] self.assertEqual([len(call) - 3 for call in calls], [32, 8]) self.assertEqual(len(result['networks']), 40) self.assertIn('172.18.0.0/16', result['excluded']) row = copy.deepcopy(NETWORK) row['IPAM']['Config'].append({'Subnet': 'fd00:1::/64', 'Gateway': 'fd00:1::1'}) self.assertEqual(network_record(row)[1], {'172.18.0.0/16'}) host.answers[('cat', '/etc/resolv.conf')] = b'x' * (1024 * 1024 + 1) with patch.dict(os.environ, {}, clear=True), self.assertRaisesRegex( InventoryFailure, 'inventory_output_limit'): collect(host) def test_host_provenance_distinguishes_owned_bridge_foreign_routes_dns_and_churn(self): bridge, subnet, gateway = 'br-' + NETWORK_ID[:12], '172.19.1.0/24', '172.19.1.1/32' rows = [{'dst': subnet, 'dev': bridge, 'protocol': 'kernel', 'scope': 'link', 'prefsrc': '172.19.1.1'}, {'dst': '172.19.1.1', 'dev': bridge, 'type': 'local', 'protocol': 'kernel', 'scope': 'host', 'table': 'local'}, {'dst': '172.19.1.255', 'dev': bridge, 'type': 'broadcast', 'protocol': 'kernel', 'scope': 'link', 'table': 'local'}, {'dst': subnet, 'dev': 'vpn0', 'protocol': 'static', 'scope': 'link'}, {'dst': 'default', 'gateway': '172.19.1.1', 'dev': 'eth0'}] records = [] route_exclusions(rows, records) self.assertIn({'subnet': subnet, 'kind': 'route_connected', 'device': bridge}, records) self.assertIn({'subnet': gateway, 'kind': 'route_connected_source', 'device': bridge}, records) self.assertIn({'subnet': gateway, 'kind': 'route_local', 'device': bridge}, records) self.assertIn({'subnet': '172.19.1.255/32', 'kind': 'route_broadcast', 'device': bridge}, records) self.assertIn({'subnet': subnet, 'kind': 'route_other', 'device': 'vpn0'}, records) self.assertIn({'subnet': gateway, 'kind': 'route_gateway', 'device': 'eth0'}, records) address_exclusions([{'ifname': bridge, 'addr_info': [ {'family': 'inet', 'local': '172.19.1.1', 'prefixlen': 24, 'broadcast': '172.19.1.255'}]}], records) self.assertIn({'subnet': gateway, 'kind': 'address_local', 'device': bridge}, records) resolver_exclusions(b'nameserver 172.19.1.1\n', records=records) self.assertIn({'subnet': gateway, 'kind': 'resolver', 'device': ''}, records) for device in ('bad/device', 'br-' + 'a' * 16, None): with self.subTest(device=device), self.assertRaises(InventoryFailure): route_exclusions([dict(rows[0], dev=device)], []) host = FakeHost() host.move_route = True with patch.dict(os.environ, {}, clear=True), self.assertRaisesRegex( InventoryFailure, 'host_inventory_changed') as error: collect(host) self.assertIn('python_version', error.exception.details) self.assertNotIn('host_records', error.exception.details) if __name__ == '__main__': unittest.main()