#!/usr/bin/env python3 """Read bounded deployment-host network evidence without syncing files or changing a network.""" import argparse import ipaddress import json from pathlib import Path import re import shlex from identity_diagnostics import ReaderFailure, bounded, retain_result from network_pools import FAILURES as INVENTORY_FAILURES from network_selected_receipt import validate_selected from network_provider import validate_provider SCHEMA = 'gowalk-cicd/backend-network-diagnostics.v2' MODULES = ('identity_diagnostics', 'network_runtime', 'network_pools', 'network_inventory', 'network_readback', 'network_selected', 'network_provider') FAILURES = INVENTORY_FAILURES | frozenset({'network_inventory_unverified', 'network_command_unavailable', ''}) def remote_source(app, host='', provider='none'): """Send only package-owned Python sources; no environment, source checkout or private inputs.""" sources = {name: Path(__file__).with_name(name + '.py').read_text() for name in MODULES} program = '''import json, sys, types sources = SOURCES for name, source in sources.items(): module = types.ModuleType(name) sys.modules[name] = module exec(compile(source, name + '.py', 'exec'), module.__dict__) from network_inventory import collect, InventoryFailure from network_runtime import run, NetworkFailure from network_selected import selected_validation from network_provider import observe result = {'schema': SCHEMA, 'ok': False, 'project': APP, 'failure': ''} try: inventory = collect(run, require_pools=False) own = [item for item in inventory['networks'] if item['Labels'].get('com.docker.compose.project') == APP] result.update({key: inventory[key] for key in ('docker_version', 'compose_version', 'python_version', 'pool_source', 'pool_failure', 'pools', 'observed_at')}) result.update(ok=True, complete=True, network_count=len(inventory['networks']), excluded_subnets=inventory['excluded'], host_identity=observe(inventory, HOST) if PROVIDER == 'digitalocean' else {'provider': 'none', 'status': 'not_requested'}, networks=[{'id': n['Id'], 'name': n['Name'], 'driver': n['Driver'], 'key': n['Labels'].get('com.docker.compose.network', ''), 'subnets': [p['Subnet'] for p in n['IPAM']['Config']], 'endpoints': len(n['Containers']), 'validation': selected_validation(inventory, n, APP)} for n in own]) except (InventoryFailure, NetworkFailure) as error: result = {'schema': SCHEMA, 'ok': False, 'project': APP, 'failure': error.code} result.update(getattr(error, 'details', {})) except Exception: result = {'schema': SCHEMA, 'ok': False, 'project': APP, 'failure': 'network_inventory_unverified'} print(json.dumps(result, separators=(',', ':'))) sys.exit(0 if result['ok'] else 1) ''' program = program.replace('SCHEMA', repr(SCHEMA)).replace('APP', repr(app)) return program.replace('HOST', repr(host)).replace('PROVIDER', repr(provider)).replace( 'SOURCES', repr(sources)).encode() def decode(status, raw, app): try: result = json.loads(raw) except (ValueError, TypeError): raise ReaderFailure('remote_invalid') from None allowed = {'schema', 'ok', 'project', 'failure', 'docker_version', 'compose_version', 'python_version', 'pool_source', 'host_identity', 'pool_failure', 'pools', 'observed_at', 'complete', 'network_count', 'excluded_subnets', 'networks'} if (not isinstance(result, dict) or set(result) - allowed or result.get('schema') != SCHEMA or result.get('project') != app or type(result.get('ok')) is not bool or status != (0 if result['ok'] else 1)): raise ReaderFailure('remote_invalid') if result.get('failure') not in FAILURES: raise ReaderFailure('remote_invalid') if result['ok']: validate_inventory(result) elif set(result) != {'schema', 'ok', 'project', 'failure'}: partial = {'schema', 'ok', 'project', 'failure', 'docker_version', 'compose_version', 'python_version', 'pools', 'pool_source', 'pool_failure', 'observed_at', 'complete'} if set(result) != partial or result['complete'] is not False: raise ReaderFailure('remote_invalid') validate_metadata(result) return result def validate_metadata(result): for key in ('docker_version', 'compose_version', 'python_version'): pattern = r'\d+\.\d+\.\d+(?:[-+][A-Za-z0-9._-]{1,64})?' if not isinstance(result.get(key), str) or not re.fullmatch(pattern, result[key]): raise ReaderFailure('remote_invalid') if result.get('pool_source') not in ('engine_info', 'versioned_moby_builtin', 'unverified'): raise ReaderFailure('remote_invalid') if (result.get('pool_failure') not in INVENTORY_FAILURES | {''} or not isinstance(result.get('observed_at'), str) or not re.fullmatch(r'[0-9T:+.\-Z]{20,40}', result['observed_at'])): raise ReaderFailure('remote_invalid') if not isinstance(result.get('pools'), list) or len(result['pools']) > 64: raise ReaderFailure('remote_invalid') try: for pool in result['pools']: if set(pool) != {'base', 'size'} or type(pool['size']) is not int or not 0 <= pool['size'] <= 32: raise ValueError() ipaddress.IPv4Network(pool['base'], strict=True) except (ValueError, TypeError, KeyError): raise ReaderFailure('remote_invalid') from None def validate_inventory(result): validate_metadata(result) if (result.get('complete') is not True or type(result.get('network_count')) is not int or not 0 <= result['network_count'] <= 1024 or result['failure']): raise ReaderFailure('remote_invalid') for field, limit in (('excluded_subnets', 16384), ('networks', 1024)): if not isinstance(result.get(field), list) or len(result[field]) > limit: raise ReaderFailure('remote_invalid') try: if result.get('host_identity') != {'provider': 'none', 'status': 'not_requested'}: validate_provider(result.get('host_identity')) for value in result['excluded_subnets']: ipaddress.IPv4Network(value, strict=True) for network in result['networks']: validate_network(network, result) except (ValueError, TypeError, KeyError): raise ReaderFailure('remote_invalid') from None def validate_network(network, inventory): fields = {'id', 'name', 'driver', 'key', 'subnets', 'endpoints', 'validation'} if (not isinstance(network, dict) or set(network) != fields or not isinstance(network['id'], str) or not re.fullmatch('[0-9a-f]{64}', network['id']) or not isinstance(network['name'], str) or not re.fullmatch('[A-Za-z0-9_.-]{1,256}', network['name']) or not isinstance(network['key'], str) or not re.fullmatch('[A-Za-z0-9_.-]{0,128}', network['key']) or network['driver'] not in ('bridge', 'host', 'null', 'overlay', 'macvlan', 'ipvlan') or type(network['endpoints']) is not int or not 0 <= network['endpoints'] <= 100000 or not isinstance(network['subnets'], list) or len(network['subnets']) > 64): raise ValueError() for value in network['subnets']: if not isinstance(value, str): raise ValueError() ipaddress.ip_network(value, strict=True) validate_selected(network['validation'], network, inventory) def diagnose(app, host, user, provider='none'): if (not re.fullmatch(r'[a-z0-9][a-z0-9-]{0,79}', app) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9.-]{0,252}', host) or not re.fullmatch(r'[a-z_][a-z0-9_-]{0,31}', user) or provider not in ('none', 'digitalocean')): raise ReaderFailure('invalid_destination') remote = shlex.join(['python3', '-']) argv = ['ssh', '-i', str(Path.home() / '.ssh/backend_deploy'), '-o', 'IdentitiesOnly=yes', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10', '-o', 'ServerAliveInterval=5', '-o', 'ServerAliveCountMax=2', user + '@' + host, remote] status, raw = bounded(argv, payload=remote_source(app, host, provider), seconds=60, with_status=True) return decode(status, raw, app) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--app', required=True) parser.add_argument('--host', required=True) parser.add_argument('--user', default='root') parser.add_argument('--provider', choices=('none', 'digitalocean'), default='none') parser.add_argument('--output', required=True) args = parser.parse_args() try: result = diagnose(args.app, args.host, args.user, args.provider) except Exception as error: code = error.code if isinstance(error, ReaderFailure) else 'network_diagnostic_unverified' result = {'schema': SCHEMA, 'ok': False, 'project': args.app, 'failure': code} try: retain_result(args.output, result) except ReaderFailure: return 1 print(json.dumps(result, separators=(',', ':'))) return 0 if result['ok'] else 1 if __name__ == '__main__': raise SystemExit(main())