"""Existing bridge evidence with full pools, foreign collisions and untrusted public receipts.""" import copy import json import subprocess import sys import unittest from identity_diagnostics import ReaderFailure from network_diagnostics import decode, remote_source from network_selected import KINDS, selected_validation from network_test_fixture import Host, network def fixture(): inventory = Host().inventory(None) selected = network('fixture', '10.253.55.0/28') inventory.update(complete=True, pool_source='engine_info', networks=inventory['networks'] + [selected]) bridge = 'br-' + selected['Id'][:12] records = [('10.253.55.0/28', 'route_connected'), ('10.253.55.1/32', 'route_connected_source'), ('10.253.55.1/32', 'route_local'), ('10.253.55.15/32', 'route_broadcast'), ('10.253.55.0/28', 'address_network'), ('10.253.55.1/32', 'address_local'), ('10.253.55.15/32', 'address_broadcast')] inventory['host_records'] = [{'subnet': subnet, 'kind': kind, 'device': bridge} for subnet, kind in records] inventory['host_excluded'] = sorted({row['subnet'] for row in inventory['host_records']}) inventory['excluded'] += inventory['host_excluded'] return inventory, selected class SelectedTests(unittest.TestCase): def test_self_routes_do_not_conflict_or_prove_an_outside_pool_reservation(self): inventory, selected = fixture() # No spare capacity in the only pool; that does not determine existing-network overlap. inventory['pools'] = [{'base': '172.28.0.0/24', 'size': 24}] result = selected_validation(inventory, selected, 'fixture') self.assertEqual((result['ownership'], result['overlap'], result['reason']), ('matched', 'clear', '')) self.assertEqual(result['self_record_count'], 7) self.assertEqual(result['pool_membership'], 'outside_verified_pools') inventory['pools'] = [{'base': '10.253.55.0/28', 'size': 28}] self.assertEqual(selected_validation(inventory, selected, 'fixture')['pool_membership'], 'inside_verified_pools') inventory['pool_failure'] = 'pool_provenance_unknown' self.assertEqual(selected_validation(inventory, selected, 'fixture')['overlap'], 'clear') self.assertEqual(selected_validation(inventory, selected, 'fixture')['pool_membership'], 'unverified') def test_identical_foreign_docker_prefix_is_preserved_after_self_exclusions(self): inventory, selected = fixture() inventory['networks'].append(network('foreign', '10.253.55.0/28')) result = selected_validation(inventory, selected, 'fixture') self.assertEqual(result['overlap'], 'conflict') self.assertEqual(result['docker_conflicts'], ['10.253.55.0/28']) self.assertNotIn('foreign', json.dumps(result)) def test_foreign_routes_and_dns_inside_self_subnet_are_never_excused(self): for kind, device in (('route_connected', 'vpn0'), ('route_other', 'tailscale0'), ('route_gateway', 'eth0'), ('resolver', ''), ('route_other', 'self')): inventory, selected = fixture() bridge = 'br-' + selected['Id'][:12] inventory['host_records'].append({'subnet': '10.253.55.1/32', 'kind': kind, 'device': bridge if device == 'self' else device}) with self.subTest(kind=kind, device=device): result = selected_validation(inventory, selected, 'fixture') self.assertEqual(result['overlap'], 'conflict') self.assertEqual(result['host_conflicts'][0]['kind'], kind) self.assertNotIn('device', result['host_conflicts'][0]) def test_missing_route_provenance_and_missing_bridge_facts_do_not_claim_clear(self): for change, reason in (('missing', 'selected_bridge_records_missing'), ('incomplete', 'host_records_incomplete')): inventory, selected = fixture() if change == 'missing': inventory['host_records'] = [] inventory['host_excluded'] = [] else: inventory['host_excluded'].append('10.108.0.0/20') result = selected_validation(inventory, selected, 'fixture') self.assertEqual((result['overlap'], result['reason']), ('unverified', reason)) inventory['networks'].append(network('foreign', '10.253.55.0/24')) self.assertEqual(selected_validation(inventory, selected, 'fixture')['overlap'], 'conflict') def test_ownership_and_unsupported_bridge_shapes_never_exempt_host_records(self): changes = [({'Name': 'custom'}, 'selected_ownership_unverified'), ({'Labels': {'com.docker.compose.project': 'fixture'}}, 'selected_ownership_unverified'), ({'Driver': 'overlay'}, 'selected_bridge_unsupported'), ({'EnableIPv6': True}, 'selected_bridge_unsupported'), ({'Options': {'com.docker.network.bridge.name': 'custom0'}}, 'selected_bridge_unsupported')] for settings, reason in changes: inventory, selected = fixture() selected.update(settings) with self.subTest(settings=settings): result = selected_validation(inventory, selected, 'fixture') self.assertEqual((result['overlap'], result['reason']), ('unverified', reason)) self.assertEqual(result['self_record_count'], 0) inventory, selected = fixture() inventory['networks'].append(copy.deepcopy(selected)) self.assertEqual(selected_validation(inventory, selected, 'fixture')['reason'], 'selected_ownership_unverified') inventory, selected = fixture() selected['IPAM']['Config'][0].update(IPRange='', AuxiliaryAddresses={}) self.assertEqual(selected_validation(inventory, selected, 'fixture')['overlap'], 'clear') selected['IPAM']['Config'][0]['IPRange'] = '10.253.55.8/29' self.assertEqual(selected_validation(inventory, selected, 'fixture')['reason'], 'selected_bridge_unsupported') def test_truncated_evidence_retains_conflict_without_unbounded_host_records(self): inventory, selected = fixture() for index in range(300): inventory['host_records'].append({'subnet': '10.253.55.1/32', 'kind': 'route_other', 'device': 'vpn%d' % index}) # Different device names collapse to the same safe public relation, not raw private inventory. result = selected_validation(inventory, selected, 'fixture') self.assertEqual(result['overlap'], 'conflict') self.assertEqual(len(result['host_conflicts']), 1) self.assertNotIn('vpn', json.dumps(result)) for index in range(16): for kind in KINDS: inventory['host_records'].append({'subnet': '10.253.55.%d/32' % index, 'kind': kind, 'device': 'vpn0'}) for kind in KINDS: inventory['host_records'].append({'subnet': '10.253.55.0/31', 'kind': kind, 'device': 'vpn0'}) inventory['host_records'].append({'subnet': '10.253.55.0/30', 'kind': kind, 'device': 'vpn0'}) inventory['host_excluded'] = sorted({row['subnet'] for row in inventory['host_records']}) result = selected_validation(inventory, selected, 'fixture') self.assertEqual((result['overlap'], result['reason']), ('conflict', 'overlap_evidence_limit')) self.assertEqual(len(result['host_conflicts']), 256) def test_streamed_v2_receipt_binds_selected_network_and_rejects_false_clear_or_secret_fields(self): inventory, selected = fixture() source = remote_source('fixture').decode().replace( 'inventory = collect(run, require_pools=False)', 'inventory = ' + repr(inventory)) process = subprocess.run([sys.executable, '-'], input=source.encode(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=5) self.assertEqual(process.returncode, 0, process.stderr.decode()) receipt = decode(0, process.stdout, 'fixture') self.assertEqual(receipt['networks'][0]['validation']['overlap'], 'clear') for changed in ({'bridge': 'br-' + 'f' * 12}, {'self_record_count': 0}, {'pool_membership': 'inside_verified_pools'}, {'gateway': 184352513}, {'secret': 'PRIVATE_SENTINEL'}, {'docker_conflicts': ['10.253.55.0/28']}, {'host_conflicts': [{'subnet': '10.253.55.1/32', 'kind': 'PRIVATE_SENTINEL', 'device_relation': 'foreign_device'}]}): bad = copy.deepcopy(receipt) bad['networks'][0]['validation'].update(changed) with self.subTest(changed=changed), self.assertRaises(ReaderFailure): decode(0, json.dumps(bad).encode(), 'fixture') for changed in ({'name': 'another_default'}, {'driver': 'overlay'}, {'subnets': [184352512]}): bad = copy.deepcopy(receipt) bad['networks'][0].update(changed) with self.subTest(changed=changed), self.assertRaises(ReaderFailure): decode(0, json.dumps(bad).encode(), 'fixture') if __name__ == '__main__': unittest.main()