#!/usr/bin/env python3 """PeerCortex operator tools. Python 3.10+; audit is read-only, filters produce drafts. No router login, shell execution, RPKI publication or automatic script self-update. """ import argparse import datetime as dt import hashlib import ipaddress import json import pathlib import re import subprocess import sys import tempfile import urllib.error import urllib.parse import urllib.request VERSION = '0.8.3' DEFAULT_ASN = 199121 # SERVER_GENERATED_ASN MAX_BYTES = 12 * 1024 * 1024 BASE = 'https://peercortex.org' class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ValueError('Unexpected source redirect') def asn_value(value): if not re.fullmatch(r'(?:AS)?[1-9][0-9]{0,9}', str(value), re.I): raise argparse.ArgumentTypeError('Expected an ASN from 1 to 4294967295') number = int(re.sub('^AS', '', str(value), flags=re.I)) if number > 4294967295: raise argparse.ArgumentTypeError('ASN out of range') return number def fetch(url): parsed = urllib.parse.urlsplit(url) if parsed.scheme != 'https' or parsed.hostname not in ('peercortex.org', 'stat.ripe.net') or parsed.username or parsed.port: raise ValueError('Unexpected source URL') request = urllib.request.Request(url, headers={'User-Agent': 'PeerCortex-Operator/' + VERSION, 'Accept': 'application/json'}) with urllib.request.build_opener(NoRedirect()).open(request, timeout=20) as response: data = response.read(MAX_BYTES + 1) if len(data) > MAX_BYTES: raise ValueError('Source response too large') result = json.loads(data) return result, {'url': url, 'retrieved_at': dt.datetime.now(dt.timezone.utc).isoformat(), 'sha256': hashlib.sha256(data).hexdigest()} def source_time(value): if not isinstance(value, str): raise ValueError('Missing source time') # RIPE explicitly documents zone-less API timestamps as UTC. return dt.datetime.fromisoformat(value.replace('Z', '+00:00')).replace(tzinfo=dt.timezone.utc) if not re.search(r'Z$|[+-]\d\d:\d\d$', value) else dt.datetime.fromisoformat(value.replace('Z', '+00:00')) def source_prefixes(raw, asn, now): d = raw.get('data', {}) if raw.get('status') != 'ok' or asn_value(d.get('resource', '')) != asn: raise ValueError('RIPE response not bound to requested ASN') end = source_time(d.get('query_endtime')) start = source_time(d.get('query_starttime')) if end < start or end > now + dt.timedelta(minutes=5) or now - end >= dt.timedelta(hours=48): raise ValueError('Invalid or stale source interval') rows = d.get('prefixes') if not isinstance(rows, list) or len(rows) > 100000: raise ValueError('Invalid prefix list') prefixes = [str(ipaddress.ip_network(row['prefix'], strict=True)) for row in rows] response = source_time(raw.get('time')) if end > response or response > now + dt.timedelta(minutes=5): raise ValueError('Invalid source response time') for row in rows: intervals = row.get('timelines') if not isinstance(intervals, list) or not intervals: raise ValueError('Missing prefix observation interval') for interval in intervals: left, right = source_time(interval.get('starttime')), source_time(interval.get('endtime')) if left < start or right > end or left > right: raise ValueError('Invalid prefix observation interval') if len(set(prefixes)) != len(prefixes): raise ValueError('Duplicate source prefixes') return set(prefixes), {'start': start.isoformat(), 'end': end.isoformat(), 'min_peers_seeing': 10} def audit(asn, getter=fetch): now = dt.datetime.now(dt.timezone.utc) report = {'schema_version': 1, 'tool_version': VERSION, 'asn': asn, 'created_at': now.isoformat(), 'overall_validity_percent': None, 'scope': 'Bounded public-source audit. Re-fetching RIPE checks ingestion, not independent routing truth. PeeringDB is self-reported; no router enforcement is assessed.', 'checks': [], 'sources': []} def add(name, status, details): report['checks'].append({'check': name, 'status': status, 'details': details}) lookup = None try: lookup, evidence = getter(BASE + '/api/lookup?asn=' + str(asn)) report['sources'].append(evidence) if lookup.get('network', {}).get('asn') != asn or lookup.get('meta', {}).get('query') != 'AS' + str(asn): raise ValueError('PeerCortex response is not bound to this ASN') if source_time(lookup.get('meta', {}).get('expires_at')) <= now: raise ValueError('PeerCortex snapshot expired') add('PeerCortex contract', 'observed', {'source_evidence': lookup.get('meta', {}).get('source_evidence', {}), 'quality': lookup.get('data_quality')}) for key in ['prefixes','neighbours','network','ix_presence','facilities']: source = lookup.get('meta', {}).get('source_evidence', {}).get(key) or {} add('Source contract: '+key, 'reported' if source.get('status') == 'available' else 'unavailable', source) if lookup.get('data_quality', {}).get('status') == 'discrepancies': add('Source disagreement', 'difference', lookup['data_quality']) except (OSError, ValueError, KeyError, TypeError, AttributeError, OverflowError) as error: lookup = None add('PeerCortex contract', 'unavailable', str(error)) try: raw, evidence = getter('https://stat.ripe.net/data/announced-prefixes/data.json?resource=AS' + str(asn)) report['sources'].append(evidence) prefixes, interval = source_prefixes(raw, asn, now) if lookup is None or lookup.get('prefixes', {}).get('status') != 'available': add('RIPE ingestion comparison', 'unavailable', {'reason': 'No usable PeerCortex prefix evidence', 'source_count': len(prefixes), 'interval': interval}) else: published = {str(ipaddress.ip_network(p, strict=True)) for p in lookup['prefixes']['list']} source_interval = lookup['prefixes'].get('evidence', {}) comparable = source_time(source_interval.get('query_starttime')) == source_time(interval['start']) and source_time(source_interval.get('query_endtime')) == source_time(interval['end']) add('RIPE ingestion comparison', 'match' if comparable and prefixes == published else 'difference' if comparable else 'incomparable', {'same_interval': comparable, 'interval': interval, 'source_count': len(prefixes), 'peercortex_count': len(published), 'only_source': sorted(prefixes-published), 'only_peercortex': sorted(published-prefixes)}) except (OSError, ValueError, KeyError, TypeError, AttributeError, OverflowError) as error: add('RIPE ingestion comparison', 'unavailable', str(error)) if lookup: checks = lookup.get('data_quality', {}).get('cross_checks', {}) comparison = checks.get('rpki') or {} add('RPKI cross-check sample', ('difference' if comparison.get('agreement_pct') is not None and comparison['agreement_pct'] < 100 else 'reported') if comparison.get('comparable_size', 0) > 0 else 'unavailable', comparison) add('Unverified assertions', 'not_verified', 'Physical PoPs, commercial relationships, complete routing visibility and router-side ROV/ASPA enforcement require operator evidence.') try: aspa, evidence = getter(BASE + '/api/aspa?asn=' + str(asn)) report['sources'].append(evidence) # Keep the entire scoped observation, not an inferred provider proposal. bound = aspa.get('asn', aspa.get('customer_asn')) if bound != asn or aspa.get('evidence', {}).get('requested_resource') != 'AS' + str(asn) or not isinstance(aspa.get('aspa_object_exists'), bool): raise ValueError('ASPA response not bound to this ASN') if source_time(aspa.get('evidence', {}).get('feed_expires_at')) <= dt.datetime.now(dt.timezone.utc): raise ValueError('ASPA evidence expired') add('Published ASPA observation', 'reported', aspa) except (OSError, ValueError, KeyError, TypeError, AttributeError, OverflowError) as error: add('Published ASPA observation', 'unavailable', str(error)) if lookup and source_time(lookup['meta']['expires_at']) <= dt.datetime.now(dt.timezone.utc): add('Snapshot validity at completion', 'unavailable', 'Snapshot expired during the audit; repeat the query') report['status'] = 'differences' if any(c['status'] == 'difference' for c in report['checks']) else 'incomplete' if any(c['status'] in ('unavailable','incomparable') for c in report['checks']) else 'scoped_checks_completed' return report def filter_commands(asn, as_set, platform, sources): # bgpq4 arguments are data in an argv array. Never shell-interpolate public text. target = as_set or 'AS' + str(asn) if len(target) > 200 or not re.fullmatch(r'(?:AS[0-9]+|AS-[A-Z0-9_-]+)(?::(?:AS[0-9]+|AS-[A-Z0-9_-]+))*', target, re.I): raise ValueError('Invalid IRR AS-SET (use an ASN or explicitly confirmed AS-SET)') allowed = {'RIPE', 'ARIN', 'APNIC', 'AFRINIC', 'LACNIC', 'RADB', 'NTTCOM'} source_list = sources.upper().split(',') if not source_list or any(s not in allowed for s in source_list): raise ValueError('Unsupported IRR source') flag = {'bird':'-b', 'junos':'-J', 'iosxr':'-X', 'json':'-j'}[platform] return [['bgpq4', '-' + family, flag, '-S', ','.join(source_list), '-l', 'PC_AS' + str(asn) + '_V' + family, target] for family in ('4','6')] def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--version', action='version', version='PeerCortex Operator ' + VERSION) sub = parser.add_subparsers(dest='command', required=True) a = sub.add_parser('audit');a.add_argument('--asn', type=asn_value, default=DEFAULT_ASN);a.add_argument('--output');a.add_argument('--previous') f = sub.add_parser('filters');f.add_argument('--asn', type=asn_value, default=DEFAULT_ASN);f.add_argument('--as-set');f.add_argument('--platform', choices=['bird','junos','iosxr','json'], default='json');f.add_argument('--sources', default='RIPE,ARIN,APNIC,AFRINIC,LACNIC');f.add_argument('--run', action='store_true', help='Run local bgpq4 to stdout only; never applies configuration') args = parser.parse_args(argv) if args.command == 'filters': commands = filter_commands(args.asn,args.as_set,args.platform,args.sources) if not args.run: print(json.dumps({'version':VERSION,'status':'draft','commands':commands,'scope':'IRR-derived draft, not RPKI validation. Confirm AS-SET, IRR source selection, platform version and policy. No router changes.'},indent=2));return 0 # Buffer both families before emitting anything: a second-family failure # must not leave a seemingly successful partial configuration on stdout. drafts = [] for command in commands: with tempfile.TemporaryFile() as output, tempfile.TemporaryFile() as errors: subprocess.run(command, check=True, timeout=60, stdout=output, stderr=errors) if output.tell() > MAX_BYTES:raise ValueError('Filter draft exceeds size limit') output.seek(0); draft = output.read(MAX_BYTES + 1) if not draft.strip():raise ValueError('Empty filter output: refused') drafts.append(draft) if args.platform == 'json': print(json.dumps({'status':'draft','ipv4':json.loads(drafts[0]),'ipv6':json.loads(drafts[1])},indent=2)) else: for draft in drafts:sys.stdout.buffer.write(draft) return 0 report = audit(args.asn) if args.previous: with open(args.previous, 'rb') as stream: previous_bytes = stream.read(MAX_BYTES + 1) if len(previous_bytes) > MAX_BYTES:raise ValueError('Previous report too large') previous = json.loads(previous_bytes) if previous.get('asn') != args.asn:raise ValueError('Previous report belongs to another ASN') report['previous_comparison'] = {'created_at':previous.get('created_at'),'status_before':previous.get('status'),'status_now':report['status'],'checks_changed':previous.get('checks') != report['checks']} output = json.dumps(report,ensure_ascii=False,indent=2)+'\n' if args.output: with open(args.output,'x',encoding='utf-8') as stream:stream.write(output) else:print(output,end='') return 2 if report['status'] in ('incomplete','differences') else 0 if __name__ == '__main__': try:sys.exit(main()) except (ValueError,OSError,subprocess.SubprocessError) as error: print('PeerCortex: '+str(error),file=sys.stderr);sys.exit(2)