#!/usr/bin/env python3
"""Portable network probe. Python 3.10+, curl; optional ping. No uploads."""
from concurrent.futures import ThreadPoolExecutor
from functools import lru_cache
import argparse
import datetime as dt
import ipaddress
import json
import os
import re
import socket
import subprocess
import sys
import time
from urllib.parse import urlsplit, urlunsplit


def public_ip(value):
    ip = ipaddress.ip_address(value)
    if not ip.is_global or ip.is_multicast or ip.is_unspecified or ip.is_reserved:
        raise ValueError("Разрешены только публичные IP-адреса")
    if ip.version == 6 and (ip.ipv4_mapped or ip.sixtofour or ip.teredo or ip in ipaddress.ip_network('64:ff9b::/96')):
        raise ValueError("Переходные IPv6-адреса не поддерживаются")
    return str(ip)


def target_url(value):
    if not isinstance(value, str) or not value or len(value) > 2048:
        raise ValueError("Введите URL или IP, до 2048 символов")
    value = value.strip()
    if any(ord(c) < 33 for c in value) or '\\' in value:
        raise ValueError("Недопустимые символы в URL")
    if '://' not in value:
        try:
            ip = ipaddress.ip_address(value)
            value = f'[{ip}]' if ip.version == 6 else str(ip)
        except ValueError:
            pass
        value = 'https://' + value
    p = urlsplit(value)
    if p.scheme not in ('http', 'https') or not p.hostname or p.username is not None or p.password is not None:
        raise ValueError("Только HTTP/HTTPS без логина и пароля")
    port = p.port or (443 if p.scheme == 'https' else 80)
    if port not in (80, 443):
        raise ValueError("URL разрешён только на портах 80 и 443")
    host = p.hostname.encode('idna').decode('ascii').lower()
    if not re.fullmatch(r'[a-z0-9.:-]+', host) or '%' in host:
        raise ValueError("Некорректное имя хоста")
    authority = f'[{host}]' if ':' in host else host
    return urlunsplit((p.scheme, f'{authority}:{port}', p.path or '/', p.query, ''))


def command(args, timeout=10):
    env = {k: v for k, v in os.environ.items() if k.lower() not in ('http_proxy', 'https_proxy', 'all_proxy', 'no_proxy')}
    env['LC_ALL'] = 'C'
    try:
        p = subprocess.run(args, capture_output=True, text=True, timeout=timeout, env=env)
        return p.returncode, p.stdout, p.stderr[:1500]
    except (OSError, subprocess.TimeoutExpired) as e:
        return -1, '', str(e)[:500]


def resolve(host):
    started = time.monotonic()
    # Bound even a stalled system resolver, including on the portable client.
    script = 'import socket,json,sys; print(json.dumps(sorted(set(x[4][0] for x in socket.getaddrinfo(sys.argv[1],None,0,socket.SOCK_STREAM)))))'
    code, out, err = command([sys.executable, '-c', script, host], 8)
    if code:
        return {'ok': False, 'error': err, 'addresses': [], 'ms': round((time.monotonic()-started)*1000)}
    ips = json.loads(out)
    for ip in ips:
        public_ip(ip)  # Reject mixed public/private answers; pin every connection below.
    return {'ok': True, 'addresses': ips, 'ms': round((time.monotonic()-started)*1000)}


def tcp(ip, port):
    started = time.monotonic()
    try:
        with socket.socket(socket.AF_INET6 if ':' in ip else socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.settimeout(4)
            sock.connect((ip, port))
        return {'ok': True, 'ms': round((time.monotonic()-started)*1000)}
    except OSError as e:
        return {'ok': False, 'ms': round((time.monotonic()-started)*1000), 'error': str(e)}


def http(url, ip, mode='auto'):
    p = urlsplit(url)
    address = f'[{ip}]' if ':' in ip else ip
    args = ['curl', '-q', '--silent', '--show-error', '--noproxy', '*', '--proto', '=http,https',
            '--connect-timeout', '4', '--max-time', '7', '--max-filesize', '1048576', '--limit-rate', '128K',
            '--output', os.devnull, '--write-out', '%{json}', '--user-agent', 'Aizametki-Network-Tests/1.0',
            '--resolve', f'{p.hostname}:{p.port}:{address}', '-6' if ':' in ip else '-4']
    if mode in ('1.2', '1.3'):
        args += [f'--tlsv{mode}', '--tls-max', mode, '--http1.1']
    elif mode == 'h1':
        args += ['--http1.1']
    # No redirects, credentials, cookies, proxy or insecure certificate mode.
    code, out, err = command(args + ['--url', url], 9)
    try:
        raw = json.loads(out)
    except ValueError:
        raw = {}
    status = int(raw.get('http_code', 0))
    connected = float(raw.get('time_connect', 0)) > 0 or status > 0
    tls_ok = float(raw.get('time_appconnect', 0)) > 0 or status > 0
    return {'ok': code == 0, 'response_received': status > 0, 'status': status,
            'tcp': connected, 'tls': tls_ok if p.scheme == 'https' else None,
            'exit_code': code, 'error': err if code else None,
            'http_version': raw.get('http_version'), 'remote_ip': raw.get('remote_ip'),
            'certificate_verify': raw.get('ssl_verify_result'),
            'timings_ms': {k: round(float(raw.get(k, 0))*1000, 2) for k in
                           ('time_connect', 'time_appconnect', 'time_starttransfer', 'time_total')}}


def ping(ip):
    # Linux container and macOS portable client; no raw socket privileges required on Linux.
    flag = ['ping6'] if sys.platform == 'darwin' and ':' in ip else ['ping'] + (['-6'] if ':' in ip else [])
    code, out, err = command(flag + ['-n', '-c', '3', ip], 6)
    return {'ok': code == 0, 'output': (out + err)[-2500:],
            'note': 'Нет ответа ICMP не означает, что TCP/HTTPS недоступен.'}


@lru_cache(maxsize=64)
def ripe_data(endpoint, resource):
    url = f'https://stat.ripe.net/data/{endpoint}/data.json?resource={resource}'
    if endpoint == 'allocation-history':
        url += '&starttime=1980-01-01T00:00:00'
    if endpoint == 'country-resource-list':
        url += '&v4_format=prefix'
    code, out, _ = command(['curl', '-q', '--silent', '--show-error', '--fail', '--noproxy', '*',
                           '--max-time', '6', '--max-filesize', '2097152', '--url', url], 8)
    try:
        payload = json.loads(out)
        if code or payload.get('status') != 'ok' or not isinstance(payload.get('data'), dict):
            raise ValueError()
        return payload['data']
    except (ValueError, TypeError):
        return {'error': 'Источник временно недоступен'}


def ru_membership(ip, data):
    result = {'member': None, 'matches': [], 'date': data.get('query_time')}
    try:
        address = ipaddress.ip_address(ip)
        entries = data['resources'][f'ipv{address.version}']
        if not isinstance(entries, list) or not entries:
            raise ValueError()
        for entry in entries:
            if '-' in entry:
                start, end = [ipaddress.ip_address(x.strip()) for x in entry.split('-')]
                match = start.version == address.version and int(start) <= int(address) <= int(end)
            else:
                match = address in ipaddress.ip_network(entry)
            if match:
                result['matches'].append(entry)
        result['member'] = bool(result['matches'])
    except (KeyError, ValueError, TypeError):
        result.update(member=None, matches=[], error='Список RU не удалось проверить')
    return result


def ripe(ip):
    public_ip(ip)
    endpoints = ['network-info', 'whois', 'prefix-overview', 'country-resource-list', 'rir-stats-country',
                 'iana-registry-info', 'allocation-history', 'transfer-history']
    with ThreadPoolExecutor(max_workers=5) as pool:
        network, whois, overview, country, delegation, iana, history, transfers = list(pool.map(
            lambda name: ripe_data(name, 'ru' if name == 'country-resource-list' else ip), endpoints))
    registrations = []
    # Keep each database object separate: parent allocations and assignments can differ.
    for record in whois.get('records', []):
        fields = {}
        for item in record:
            if item.get('key') in ('inetnum', 'inet6num', 'netname', 'country', 'org', 'org-name', 'descr', 'status', 'source', 'last-modified'):
                fields.setdefault(item['key'], []).append(item.get('value', ''))
        if 'inetnum' in fields or 'inet6num' in fields:
            registrations.append(fields)
    result = {'source': 'https://stat.ripe.net/data/network-info/data.json?resource=' + ip,
              'prefix': network.get('prefix'), 'asns': network.get('asns'),
              'error': network.get('error'), 'holders': overview.get('asns', []),
              'registrations': registrations, 'authorities': whois.get('authorities', []),
              'iana': iana, 'allocation_history': history, 'transfers': transfers,
              'delegation': delegation, 'registration_error': whois.get('error'), 'ru_resources': ru_membership(ip, country),
              'checked_at': dt.datetime.now(dt.timezone.utc).isoformat()}
    try:
        subnet = ipaddress.ip_network(result['prefix'])
        result['range'] = {'first': str(subnet.network_address), 'last': str(subnet.broadcast_address),
                           'addresses': str(subnet.num_addresses)}
    except (ValueError, TypeError):
        pass
    return result


def validate(config):
    if not isinstance(config, dict):
        raise ValueError('Ожидается объект настроек')
    repeat = config.get('repeat', 10)
    if type(repeat) is not int or not 1 <= repeat <= 25:
        raise ValueError('Повторов: от 1 до 25')
    families = config.get('families', [4, 6])
    if not isinstance(families, list) or not families or any(type(f) is not int or f not in (4, 6) for f in families):
        raise ValueError('Выберите IPv4 и/или IPv6')
    controls = config.get('controls', [])
    if not isinstance(controls, list) or len(controls) > 2:
        raise ValueError('Не более двух контрольных URL')
    return {'url': target_url(config.get('url')), 'repeat': repeat,
            'families': sorted(set(families)), 'controls': [target_url(x) for x in controls],
            'ssh': config.get('ssh') is True, 'label': str(config.get('label', 'Локальный клиент'))[:100]}


def run(config, emit=lambda _: None):
    config = validate(config)
    report = {'schema': 1, 'started_at': dt.datetime.now(dt.timezone.utc).isoformat(),
              'vantage': config['label'], 'config': config, 'targets': [],
              'limitations': ['Измерения относятся только к указанной точке проверки.',
                  'Прокси из переменных окружения отключены; системный VPN влияет на маршрут.',
                  'Страна регистрации и ASN не доказывают фильтрацию. Причина требует сопоставления дампов.',
                  'TCP, TLS-режимы и HTTP выполняются в отдельных соединениях.',
                  'Проверяется до двух IP каждого семейства; редиректы не выполняются.']}
    for url in [config['url']] + config['controls']:
        emit({'message': f'DNS: {url}'})
        p = urlsplit(url)
        target = {'url': url, 'dns': resolve(p.hostname), 'endpoints': []}
        report['targets'].append(target)
        emit({'report': report})
        for family in config['families']:
            ips = [ip for ip in target['dns']['addresses'] if ipaddress.ip_address(ip).version == family]
            for ip in ips[:2]:
                entry = {'ip': ip, 'family': family, 'ping': ping(ip), 'ripe': ripe(ip), 'samples': []}
                target['endpoints'].append(entry)
                for n in range(config['repeat']):
                    emit({'message': f'{p.hostname} · IPv{family} {ip} · {n+1}/{config["repeat"]}'})
                    sample = {'at': dt.datetime.now(dt.timezone.utc).isoformat(),
                              'tcp': {str(port): tcp(ip, port) for port in ([80, 443, 22] if config['ssh'] else [80, 443])},
                              'http': http(url, ip)}
                    if p.scheme == 'https':
                        sample['tls12'] = http(url, ip, '1.2')
                        sample['tls13'] = http(url, ip, '1.3')
                        sample['http1'] = http(url, ip, 'h1')
                    entry['samples'].append(sample)
                    emit({'report': report})
                    if n + 1 < config['repeat']:
                        time.sleep(.3)
    report['finished_at'] = dt.datetime.now(dt.timezone.utc).isoformat()
    return report


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('url', nargs='?')
    parser.add_argument('--repeat', type=int, default=10)
    parser.add_argument('--label', default='Домашняя сеть / без VPN')
    parser.add_argument('--control', action='append', default=[])
    parser.add_argument('--output', default='network-report.json')
    parser.add_argument('--worker', action='store_true', help=argparse.SUPPRESS)
    args = parser.parse_args()
    if args.worker:
        cfg = json.load(sys.stdin)
        try:
            result = run(cfg, lambda event: print(json.dumps(event, ensure_ascii=False), flush=True))
            print(json.dumps({'report': result, 'done': True}, ensure_ascii=False), flush=True)
        except Exception as e:
            print(json.dumps({'error': str(e)}, ensure_ascii=False), flush=True)
            sys.exit(1)
    else:
        cfg = {'url': args.url, 'repeat': args.repeat, 'label': args.label, 'controls': args.control}
        result = run(cfg, lambda event: print(event['message'], file=sys.stderr) if 'message' in event else None)
        # Preserve existing reports.
        with open(args.output, 'x') as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
        print(f'Отчёт: {args.output}')