"""Наблюдаемый CONNECT-маршрут. Не читает HTTP тела, cookies, SDP и звук.

Локальный relay добавляет Basic только к CONNECT до нашего upstream proxy.
Шифрованный поток проходит байт-в-байт; TLS завершает существующий MITM,
не этот relay. Журнал не доказывает, какой OAuth выбрал удалённый gateway.
"""
import asyncio
import base64
from dataclasses import dataclass, field
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import socket
import ssl
import sys
import tempfile
import time
from urllib.parse import unquote, urlsplit

import aiohttp

ALLOWED = frozenset({"chatgpt.com:443", "api.openai.com:443"})


class RouteError(RuntimeError):
    """В сообщении нет входных URL с паролем и чужих тел ошибок."""


@dataclass(repr=False, frozen=True)
class Proxy:
    host: str
    port: int
    tls: bool
    authorization: str = field(repr=False)

    @classmethod
    def from_env(cls, name="HTTPS_PROXY"):
        # Явно выбранная переменная: NO_PROXY и другие fallback не применяются.
        raw = os.environ.get(name)
        if not raw:
            raise RouteError("proxy: выбранная переменная окружения пуста")
        try:
            u = urlsplit(raw)
            if u.scheme not in {"http", "https"} or not u.hostname or u.path not in {"", "/"} or u.query or u.fragment:
                raise ValueError()
            port = u.port or (443 if u.scheme == "https" else 80)
            if not 1 <= port <= 65535 or any(c in u.hostname for c in "\r\n \t"):
                raise ValueError()
            authorization = ""
            if u.username is not None:
                username, password = unquote(u.username), unquote(u.password or "")
                if ":" in username or any(c in username + password for c in "\r\n\0"):
                    raise ValueError()
                authorization = "Basic " + base64.b64encode((username + ":" + password).encode()).decode("ascii")
            return cls(u.hostname, port, u.scheme == "https", authorization)
        except ValueError:
            raise RouteError("proxy: некорректный HTTP(S) proxy; значение скрыто") from None

    def public(self):
        return {"host": self.host, "port": self.port, "tls_to_proxy": self.tls,
                "basic_present": bool(self.authorization)}


def endpoint(url):
    # call ID, conversation ID, message ID и неизвестные query-values скрыты.
    u = urlsplit(str(url))
    path = u.path
    if path.startswith("/v1/live/"):
        path = "/v1/live/<call_id>"
    return {"scheme": u.scheme, "host": u.hostname, "path": path,
            "query_keys": sorted({p.split("=", 1)[0] for p in u.query.split("&") if p})}


def tls_context(ca_file):
    # Системные CA остаются доверенными; добавляем именно выбранный public CA.
    context = ssl.create_default_context()
    context.load_verify_locations(cafile=str(ca_file.expanduser()))
    return context


class Journal:
    def __init__(self):
        self.events = []
        self.udp = {}
        self.active = True

    def add(self, kind, **fields):
        self.events.append({"at": datetime.now(timezone.utc).isoformat(), "event": kind, **fields})

    def install_udp_audit(self):
        # CPython audit socket.sendto содержит socket/address, но НЕ payload.
        # Это попытки syscall, не подтверждения доставки. RTP stats идут отдельно.
        def observe(event, args):
            if not self.active or event != "socket.sendto":
                return
            try:
                sock, address = args
                if sock.type & socket.SOCK_DGRAM != socket.SOCK_DGRAM or not isinstance(address, tuple):
                    return
                key = (str(address[0]), int(address[1]))
                if key not in self.udp and len(self.udp) >= 128:
                    return
                self.udp[key] = self.udp.get(key, 0) + 1
            except (ValueError, TypeError, AttributeError, IndexError):
                pass
        sys.addaudithook(observe)

    def udp_snapshot(self):
        return [{"host": h, "port": p, "sendto_attempts": n} for (h, p), n in sorted(self.udp.items())]

    def save(self, path, result):
        self.active = False
        data = {"schema_version": 1, "result": result, "udp_sendto": self.udp_snapshot(), "events": self.events,
                "limits": ["CONNECT proves client-to-proxy route, not upstream OAuth identity",
                           "UDP audit counts attempts; selected ICE + RTP stats corroborate media",
                           "No proxy server logs or Internet packet capture are included"]}
        fd, temporary = tempfile.mkstemp(prefix=".proxy-routes-", dir=path.parent)
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as target:
                json.dump(data, target, ensure_ascii=False, indent=2)
                target.write("\n")
                target.flush()
                os.fsync(target.fileno())
            os.link(temporary, path)  # Только новый файл; atomic/no overwrite/0600.
        finally:
            os.unlink(temporary)


class ConnectRelay:
    def __init__(self, proxy, context, journal, timeout=30, allowed=ALLOWED):
        self.proxy, self.context, self.journal = proxy, context, journal
        self.timeout, self.allowed = timeout, allowed
        self.server = None
        self.tasks, self.writers = set(), set()
        self.number = 0

    async def start(self):
        self.server = await asyncio.start_server(self.handle, "127.0.0.1", 0, limit=65536)
        self.url = "http://127.0.0.1:" + str(self.server.sockets[0].getsockname()[1])
        self.journal.add("relay_started", bind="127.0.0.1", upstream=self.proxy.public())
        return self

    async def close(self):
        if self.server:
            self.server.close()
            await self.server.wait_closed()
        for writer in list(self.writers):
            writer.close()
        for task in list(self.tasks):
            task.cancel()
        await asyncio.gather(*list(self.tasks), return_exceptions=True)
        self.journal.add("relay_closed", remaining_handlers=len(self.tasks))

    async def handle(self, reader, writer):
        task = asyncio.current_task()
        self.tasks.add(task)
        self.writers.add(writer)
        upstream = None
        pumps = []
        self.number += 1
        number = self.number
        target = "unknown"
        counts = {"to_proxy": 0, "from_proxy": 0}
        started = time.monotonic()
        try:
            head = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), self.timeout)
            first = head.split(b"\r\n", 1)[0].decode("ascii")
            method, target, version = first.split(" ")
            if method != "CONNECT" or version != "HTTP/1.1" or target not in self.allowed:
                raise RouteError("relay: CONNECT target запрещён")
            # Ничего из исходных заголовков (включая Authorization) не копируем.
            # До proxy Basic незашифрован при scheme=http — как в acodex.
            upstream_reader, upstream = await asyncio.wait_for(asyncio.open_connection(
                self.proxy.host, self.proxy.port,
                ssl=self.context if self.proxy.tls else None,
                server_hostname=self.proxy.host if self.proxy.tls else None,
                limit=65536), self.timeout)
            self.writers.add(upstream)
            peer = upstream.get_extra_info("peername")
            self.journal.add("proxy_tcp_connected", connection=number, target=target,
                             peer_host=peer[0], peer_port=peer[1], **self.proxy.public())
            request = f"CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n"
            if self.proxy.authorization:
                request += "Proxy-Authorization: "[скрыто]"\r\n"
            upstream.write((request + "\r\n").encode("ascii"))
            await asyncio.wait_for(upstream.drain(), self.timeout)
            response = await asyncio.wait_for(upstream_reader.readuntil(b"\r\n\r\n"), self.timeout)
            status = int(response.split(b"\r\n", 1)[0].split(b" ")[1])
            self.journal.add("proxy_connect_response", connection=number, target=target, status=status)
            if status != 200:
                writer.write(f"HTTP/1.1 {status if status in {403,407,502,503,504} else 502} Proxy refusal\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".encode())
                await writer.drain()
                return
            writer.write(b"HTTP/1.1 200 Connection Established\r\n\r\n")
            await writer.drain()
            async def pump(source, destination, key):
                while chunk := await source.read(65536):
                    counts[key] += len(chunk)
                    destination.write(chunk)
                    await destination.drain()
            pumps = [asyncio.create_task(pump(reader, upstream, "to_proxy")),
                     asyncio.create_task(pump(upstream_reader, writer, "from_proxy"))]
            await asyncio.wait(pumps, return_when=asyncio.FIRST_COMPLETED)
        except asyncio.CancelledError:
            raise
        except Exception as error:
            # str(error) может содержать credential-bearing URI: не выводим его.
            self.journal.add("relay_error", connection=number, error_type=type(error).__name__)
            try:
                writer.write(b"HTTP/1.1 502 Relay failure\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
                await writer.drain()
            except Exception:
                pass
        finally:
            for pump in pumps:
                pump.cancel()
            await asyncio.gather(*pumps, return_exceptions=True)
            for sock in (upstream, writer):
                if sock:
                    sock.close()
                    try:
                        await asyncio.wait_for(sock.wait_closed(), 2)
                    except Exception:
                        pass
                    self.writers.discard(sock)
            self.journal.add("proxy_tunnel_closed", connection=number, **counts,
                             duration_seconds=round(time.monotonic() - started, 3))
            self.tasks.discard(task)


class RoutedClient:
    def __init__(self, network, headers, redirect_trace):
        self.network = network
        trace = aiohttp.TraceConfig()
        async def start(session, context, params):
            network.journal.add("client_request", method=params.method, **endpoint(params.url),
                                route=self.route(params.url))
        async def end(session, context, params):
            network.journal.add("client_response", method=params.method, **endpoint(params.url),
                                route=self.route(params.url), status=params.response.status)
        async def failure(session, context, params):
            network.journal.add("client_request_failed", method=params.method, **endpoint(params.url),
                                route=self.route(params.url), error_type=type(params.exception).__name__)
        trace.on_request_start.append(start)
        trace.on_request_end.append(end)
        trace.on_request_exception.append(failure)
        self.session = aiohttp.ClientSession(headers=headers, trust_env=False,
            cookie_jar=aiohttp.DummyCookieJar(),
            connector=aiohttp.TCPConnector(ssl=network.context),
            timeout=aiohttp.ClientTimeout(total=network.timeout), trace_configs=[trace, redirect_trace])

    def route(self, url):
        return "direct_explicit" if str(url).startswith("wss://") and self.network.sideband_route == "direct" else "proxy"

    def proxy_for(self, url):
        if self.route(url) == "direct_explicit":
            return None
        return self.network.relay.url

    def post(self, url, **kwargs):
        return self.session.post(url, proxy=self.proxy_for(url), **kwargs)

    async def ws_connect(self, url, **kwargs):
        return await self.session.ws_connect(url, proxy=self.proxy_for(url), **kwargs)

    async def close(self):
        await self.session.close()


class Network:
    def __init__(self, proxy, context, journal, timeout, sideband_route="proxy"):
        self.context, self.journal, self.timeout = context, journal, timeout
        self.sideband_route = sideband_route
        self.relay = ConnectRelay(proxy, context, journal, timeout)

    def client(self, headers, redirect_trace):
        return RoutedClient(self, headers, redirect_trace)

    def media_snapshot(self, pc):
        # Частный introspection API зафиксирован на aiortc 1.15.0/aioice.
        # При drift не выдумываем адрес: журнал остаётся unobserved.
        peers = []
        try:
            transports = {t.receiver.transport.transport for t in pc.getTransceivers()}
            for ice in transports:
                for pair in ice._connection._nominated.values():
                    peers.append({"local_host": pair.local_addr[0], "local_port": pair.local_addr[1],
                                  "remote_host": pair.remote_addr[0], "remote_port": pair.remote_addr[1],
                                  "local_candidate_type": pair.local_candidate.type,
                                  "remote_candidate_type": pair.remote_candidate.type,
                                  "transport": pair.remote_candidate.transport,
                                  "route": "outside_http_connect_proxy"})
        except (AttributeError, TypeError):
            pass
        self.journal.add("selected_ice", observed=bool(peers), peers=peers)
        return peers