"""Bounded read-only HTTPS study. Credentials remain in memory; default is offline."""
import argparse
import getpass
import ipaddress
import concurrent.futures
import ctypes
import datetime
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import platform
import shutil
import socket
import struct
import subprocess
import threading
import time

HERE = Path(__file__).resolve().parent
TARGETS = [
    {'id': 'D01', 'url': 'https://www.ripe.net/'},
    {'id': 'D02', 'url': 'https://whois.kr/eng/main.jsp'},
    {'id': 'D03', 'url': 'https://registro.br/'},
    {'id': 'D04', 'url': 'https://afrinic.net/'},
    {'id': 'D05', 'url': 'https://www.apnic.net/'},
]
LEVELS = [1, 2, 4]
ALIASES = ['T01', 'T02', 'T03']
ROUNDS = 2

def utc():
    return datetime.datetime.now(datetime.timezone.utc).isoformat()

def save(value):
    temp = HERE / 'measurements.writing'
    temp.write_text(json.dumps(value, indent=2), encoding='utf-8')
    os.replace(temp, HERE / 'measurements.json')

def quote(value):
    if any(ord(ch) < 32 for ch in value):
        raise ValueError('invalid-config-token')
    return '"' + value.replace('\\', '\\\\').replace('"', '\\"') + '"'

def config(url, row):
    host, port, user, password = row.split(':', 3)
    fields = ['http_code', 'http_connect', 'exitcode', 'time_connect', 'time_appconnect',
              'time_starttransfer', 'time_total', 'size_download', 'num_connects', 'ssl_verify_result']
    writeout = json.dumps({key: '%{' + key + '}' for key in fields}, separators=(',', ':'))
    values = {'url': url, 'proxy': 'http://' + host + ':' + port,
              'proxy-user': user + ':' + password, 'noproxy': '', 'max-redirs': '0',
              'connect-timeout': '6', 'max-time': '6', 'max-filesize': '1048576',
              'proto': '=https', 'proto-redir': '=https', 'output': 'NUL',
              'write-out': writeout, 'user-agent': 'BuyProxies-Reserved-HTTPS-Check/2.0'}
    return '\n'.join(key + ' = ' + quote(value) for key, value in values.items()) + '\nproxy-basic\nhead\nhttp1.1\nsilent\n'

def tcp_rows():
    # Windows IP Helper API. Private addresses are matched in memory only.
    size = ctypes.c_ulong(0)
    api = ctypes.windll.iphlpapi.GetExtendedTcpTable
    api(None, ctypes.byref(size), False, 2, 5, 0)
    buffer = ctypes.create_string_buffer(size.value)
    if api(buffer, ctypes.byref(size), False, 2, 5, 0):
        raise RuntimeError('tcp-counter-unavailable')
    raw = buffer.raw
    count = struct.unpack_from('<I', raw)[0]
    return [struct.unpack_from('<6I', raw, 4 + index * 24) for index in range(count)]

def clean(raw, exitcode):
    data = json.loads(raw)
    expected = {'http_code', 'http_connect', 'exitcode', 'time_connect', 'time_appconnect',
                'time_starttransfer', 'time_total', 'size_download', 'num_connects', 'ssl_verify_result'}
    if set(data) != expected or int(data['exitcode']) != exitcode:
        raise ValueError('invalid-numeric-response')
    result = {key: int(data[key]) for key in expected if not key.startswith('time_')}
    result.update({key + '_ms': round(float(data[key]) * 1000, 3)
                   for key in expected if key.startswith('time_')})
    if any(value < 0 for value in result.values()) or result['time_total_ms'] > 12000:
        raise ValueError('invalid-numeric-bounds')
    status, tunnel = result['http_code'], result['http_connect']
    if tunnel == 407:
        category = 'proxy-authentication-rejected'
    elif status in (401, 407):
        category = 'destination-authentication-rejected'
    elif status == 429 or tunnel == 429:
        category = 'rate-limited'
    elif status == 403 or tunnel == 403:
        category = 'blocked'
    elif exitcode == 28:
        category = 'timeout'
    elif exitcode == 60:
        category = 'tls-verification-error'
    elif exitcode == 7:
        category = 'connection-error'
    elif exitcode == 63:
        category = 'size-limit'
    elif exitcode:
        category = 'transport-error'
    elif 200 <= status < 300:
        category = 'http-success'
    elif 300 <= status < 400:
        category = 'redirect-not-followed'
    else:
        category = 'destination-http-error'
    result.update(category=category, success=exitcode == 0 and 200 <= status < 300,
                  transport_completed=exitcode == 0 and status > 0)
    return result

def run(executable, row, target, alias, level, round_id, state):
    environment = os.environ.copy()
    environment.pop('SSLKEYLOGFILE', None)
    started, monotonic_start = utc(), time.perf_counter_ns()
    process = subprocess.Popen([executable, '--disable', '--config', '-'],
                               stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               env=environment, creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
    with state['lock']:
        state['pids'].add(process.pid)
        state['process_peak'] = max(state['process_peak'], len(state['pids']))
    try:
        stdout, stderr = process.communicate(config(target['url'], row).encode(), timeout=12)
        del stderr
        result = clean(stdout, process.returncode)
        del stdout
    finally:
        with state['lock']:
            state['pids'].discard(process.pid)
    result.update(proxy_alias=alias, target_id=target['id'], concurrency_limit=level,
                  round=round_id, started_utc=started, finished_utc=utc(),
                  process_start_monotonic_ns=monotonic_start, process_end_monotonic_ns=time.perf_counter_ns())
    return result

def observe(state, endpoints, stop):
    while not stop.is_set():
        with state['lock']:
            pids = state['pids'].copy()
        try:
            rows = tcp_rows()
            count = sum(1 for status, local_address, local_port, remote_address, remote_port, pid in rows
                        if status == 5 and pid in pids and (remote_address, remote_port) in endpoints)
            with state['lock']:
                state['tcp_peak'] = max(state['tcp_peak'], count)
                state['tcp_samples'] += 1
        except Exception:
            state['tcp_errors'] += 1
        stop.wait(0.02)

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--execute', action='store_true')
    args = parser.parse_args()
    plan = {'aliases': ALIASES, 'levels': LEVELS, 'rounds': ROUNDS, 'targets': TARGETS,
            'planned_requests': len(ALIASES) * len(LEVELS) * ROUNDS * len(TARGETS),
            'maximum_global_active_requests': max(LEVELS), 'method': 'HEAD',
            'timeout_seconds': 6, 'redirects': 0, 'retries': 0, 'tls_verification': True,
            'maximum_file_bytes': 1048576, 'payload': 'headers only; body discarded',
            'cell_semantics': 'one proxy at a time; each round has five targets; global limit equals level',
            'inter_round_gap_seconds': 2, 'inter_cell_gap_seconds': 10}
    if not args.execute:
        print(json.dumps({'state': 'OFFLINE_PLAN', 'credentials_read': False, **plan}))
        return
    if (HERE / 'measurements.json').exists():
        raise ValueError('existing-attempt-no-replay')
    rows = []
    seen = set()
    for alias in ALIASES:
        print('Use only your own explicitly authorized reserved test proxy for ' + alias)
        host = getpass.getpass('Public IPv4 proxy address (hidden): ')
        port = getpass.getpass('Proxy port (hidden): ')
        user = getpass.getpass('Proxy username (hidden): ')
        password = getpass.getpass('Proxy password (hidden): ')
        if (not ipaddress.IPv4Address(host).is_global or not 1 <= int(port) <= 65535
                or (host, port) in seen or not user or not password or ':' in user
                or any(ord(ch) < 33 for ch in host + port + user + password)):
            raise ValueError('invalid-authorized-test-proxy')
        seen.add((host, port))
        rows.append(':'.join((host, port, user, password)))
        del host, port, user, password
    executable = shutil.which('curl.exe')
    identity = subprocess.run([executable, '--disable', '--version'], capture_output=True, check=True).stdout.decode().splitlines()[0]
    record = {'schema': 1, 'state': 'RUNNING', 'started_utc': utc(), 'plan': plan,
              'origin': 'local Windows computer', 'client': {'python': platform.python_version(), 'curl': identity},
              'runner_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
              'credential_source': 'operator-provided authorized test access, hidden prompts and in-memory only',
              'credentials_persisted': False, 'raw_response_persisted': False,
              'proxy_locations': 'not independently verified', 'cells': [], 'observations': []}
    save(record)
    for alias, row in zip(ALIASES, rows):
        host, port, user, password = row.split(':', 3)
        endpoints = {(struct.unpack('<I', socket.inet_aton(host))[0], socket.htons(int(port)))}
        del host, port, user, password
        for level in LEVELS:
            cell = {'proxy_alias': alias, 'concurrency_limit': level, 'started_utc': utc(), 'rounds': []}
            for round_id in range(1, ROUNDS + 1):
                state = {'lock': threading.Lock(), 'pids': set(), 'process_peak': 0,
                         'tcp_peak': 0, 'tcp_samples': 0, 'tcp_errors': 0}
                stop = threading.Event()
                watcher = threading.Thread(target=observe, args=(state, endpoints, stop))
                watcher.start()
                try:
                    with concurrent.futures.ThreadPoolExecutor(max_workers=level) as pool:
                        futures = [pool.submit(run, executable, row, target, alias, level, round_id, state) for target in TARGETS]
                        outcomes = [future.result() for future in futures]
                finally:
                    stop.set()
                    watcher.join()
                record['observations'].extend(outcomes)
                cell['rounds'].append({'round': round_id, 'peak_active_curl_processes': state['process_peak'],
                                       'observed_peak_established_proxy_tcp_connections': state['tcp_peak'],
                                       'tcp_observer_samples': state['tcp_samples'], 'tcp_observer_errors': state['tcp_errors']})
                save(record)
                if any(item['category'] in ('rate-limited', 'proxy-authentication-rejected', 'destination-authentication-rejected') for item in outcomes):
                    record.update(state='STOPPED_REQUIRES_REVIEW', reason='authentication-or-rate-limit-no-retry', finished_utc=utc())
                    save(record)
                    print(json.dumps({'state': record['state'], 'observations': len(record['observations'])}))
                    return
                if round_id < ROUNDS:
                    time.sleep(2)
            cell['finished_utc'] = utc()
            record['cells'].append(cell)
            save(record)
            print(json.dumps({'cell': alias, 'level': level, 'observations': len(record['observations']),
                              'tcp_peaks': [item['observed_peak_established_proxy_tcp_connections'] for item in cell['rounds']]}), flush=True)
            if not (alias == ALIASES[-1] and level == LEVELS[-1]):
                time.sleep(10)
    record.update(state='COMPLETE', finished_utc=utc())
    save(record)
    print(json.dumps({'state': 'COMPLETE', 'observations': len(record['observations'])}))

if __name__ == '__main__':
    try:
        main()
    except Exception as error:
        print(json.dumps({'state': 'SAFE_FAILURE', 'exception_type': type(error).__name__, 'details_withheld': True}))
        raise SystemExit(1)