"""HTTP-Proxy (HTTP/1.1, CONNECT-Tunnel für HTTPS und beliebige TCP-Protokolle)."""
import asyncio
import base64
import html
import time
from urllib.parse import urlsplit

from .auth import users
from .core import Blocked, Conn, history, open_upstream, register, relay

MAX_HEAD = 64 * 1024
HOP_BY_HOP = {"connection", "keep-alive", "proxy-authorization", "proxy-connection", "proxy-authenticate",
              "te", "trailer", "upgrade", "expect"}


class BadRequest(Exception):
    pass


def parse_head(head: bytes) -> tuple[str, list[tuple[str, str]]]:
    lines = head.decode("latin-1").split("\r\n")
    first = lines[0]
    headers = []
    for line in lines[1:]:
        if not line:
            continue
        if ":" not in line:
            raise BadRequest("Ungültige Kopfzeile")
        k, v = line.split(":", 1)
        headers.append((k.strip(), v.strip()))
    return first, headers


def get_header(headers, name: str) -> str | None:
    name = name.lower()
    for k, v in headers:
        if k.lower() == name:
            return v
    return None


def split_hostport(value: str, default_port: int) -> tuple[str, int]:
    value = value.strip()
    if value.startswith("["):
        end = value.index("]")
        host, rest = value[1:end], value[end + 1:]
        port = int(rest[1:]) if rest.startswith(":") else default_port
        return host, port
    if value.count(":") == 1:
        host, port = value.rsplit(":", 1)
        return host, int(port)
    return value, default_port


def response(code: int, reason: str, body: str = "", extra: list[str] | None = None, close: bool = False) -> bytes:
    data = body.encode()
    lines = [f"HTTP/1.1 {code} {reason}", "Content-Type: text/html; charset=utf-8",
             f"Content-Length: {len(data)}", "Connection: close" if close else "Connection: keep-alive",
             *(extra or [])]
    return ("\r\n".join(lines) + "\r\n\r\n").encode() + data


def error_page(title: str, text: str) -> str:
    return (f"<!doctype html><html lang=de><meta charset=utf-8><title>{html.escape(title)}</title>"
            "<body style='font-family:system-ui,sans-serif;background:#f4f5f7;color:#1d2330;display:flex;"
            "align-items:center;justify-content:center;min-height:90vh'>"
            "<div style='background:#fff;border-radius:12px;padding:32px 40px;max-width:560px;"
            "box-shadow:0 4px 24px rgba(0,0,0,.08)'>"
            f"<h1 style='margin:0 0 12px;font-size:22px'>{html.escape(title)}</h1>"
            f"<p style='margin:0;color:#4a5263;line-height:1.5'>{html.escape(text)}</p></div></body></html>")


# --------------------------------------------------------------------------- Body-Übertragung
async def copy_exact(reader, writer, n: int, counter) -> None:
    while n > 0:
        data = await reader.read(min(n, 65536))
        if not data:
            raise ConnectionError("Verbindung vorzeitig beendet")
        n -= len(data)
        if writer:
            writer.write(data)
            await writer.drain()
        counter(len(data))


async def copy_chunked(reader, writer, counter) -> None:
    while True:
        line = await reader.readuntil(b"\r\n")
        size = int(line.split(b";", 1)[0].strip() or b"0", 16)
        if writer:
            writer.write(line)
        counter(len(line))
        if size == 0:
            # Trailer bis zur Leerzeile
            while True:
                t = await reader.readuntil(b"\r\n")
                if writer:
                    writer.write(t)
                counter(len(t))
                if t == b"\r\n":
                    break
            if writer:
                await writer.drain()
            return
        await copy_exact(reader, writer, size + 2, counter)


async def copy_until_eof(reader, writer, counter) -> None:
    while True:
        data = await reader.read(65536)
        if not data:
            return
        writer.write(data)
        counter(len(data))
        await writer.drain()


async def transfer_body(headers, reader, writer, counter) -> None:
    te = (get_header(headers, "transfer-encoding") or "").lower()
    if "chunked" in te:
        await copy_chunked(reader, writer, counter)
        return
    cl = get_header(headers, "content-length")
    if cl:
        await copy_exact(reader, writer, int(cl), counter)


def body_is_discardable(headers) -> bool:
    if "chunked" in (get_header(headers, "transfer-encoding") or "").lower():
        return False
    cl = get_header(headers, "content-length")
    return not cl or int(cl) <= 1_000_000


# --------------------------------------------------------------------------- Handler
async def authenticate(headers, client_ip: str):
    """(user, anonym erlaubt?, Zugangsdaten angegeben?)"""
    from .policy import policy
    value = get_header(headers, "proxy-authorization")
    if value and value.lower().startswith("basic "):
        try:
            raw = base64.b64decode(value[6:].strip()).decode("utf-8", "replace")
            username, _, password = raw.partition(":")
        except (ValueError, UnicodeError):
            return None, False, True
        user = await users.authenticate(username, password, client_ip)
        return user, False, True
    if policy.is_anonymous_allowed(client_ip):
        return None, True, False
    return None, False, False


async def handle_http(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, prefix: bytes, client_ip: str):
    from .policy import policy
    realm = policy.settings.get("auth_realm", "Proxy")
    while True:
        try:
            head = prefix + await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), timeout=300)
        except (asyncio.IncompleteReadError, asyncio.TimeoutError, ConnectionError):
            return
        except asyncio.LimitOverrunError:
            writer.write(response(431, "Request Header Fields Too Large", close=True))
            return
        prefix = b""
        try:
            first, headers = parse_head(head[:-4])
            method, target, version = first.split(" ", 2)
        except (BadRequest, ValueError):
            writer.write(response(400, "Bad Request", error_page("Ungültige Anfrage", "Anfrage nicht lesbar."),
                                  close=True))
            return
        client_close = (get_header(headers, "proxy-connection") or get_header(headers, "connection") or "").lower() \
            == "close" or version == "HTTP/1.0"

        user, anonymous, gave_credentials = await authenticate(headers, client_ip)
        if not user and not anonymous:
            if gave_credentials:
                history.add(username=None, client_ip=client_ip, proto="http", method=method, url=target[:2000],
                            status="auth_failed", detail="Anmeldung fehlgeschlagen")
            if not body_is_discardable(headers):
                client_close = True
            else:
                await transfer_body(headers, reader, None, lambda n: None)
            writer.write(response(407, "Proxy Authentication Required",
                                  error_page("Anmeldung erforderlich", "Bitte am Proxy anmelden."),
                                  [f'Proxy-Authenticate: Basic realm="{realm}"'], close=client_close))
            await writer.drain()
            if client_close:
                return
            continue

        username = user["username"] if user else "anonym"
        if method.upper() == "CONNECT":
            keep = await handle_connect(reader, writer, user, username, client_ip, target)
        else:
            keep = await handle_plain(reader, writer, user, username, client_ip, method, target, version, headers)
        if not keep or client_close:
            return


async def handle_connect(reader, writer, user, username, client_ip, target) -> bool:
    try:
        host, port = split_hostport(target, 443)
    except (ValueError, IndexError):
        writer.write(response(400, "Bad Request", close=True))
        return False
    conn = Conn(username, client_ip, "https" if port == 443 else "connect", host, port, "CONNECT", f"{host}:{port}")
    try:
        r_reader, r_writer = await open_upstream(user, host, port)
    except Blocked as b:
        conn.finish("blocked", b.decision.reason)
        writer.write(response(403, "Forbidden", error_page("Zugriff gesperrt",
                                                           f"{host}:{port} ist durch die Proxy-Richtlinie gesperrt.")))
        await writer.drain()
        return True
    except OSError as e:
        conn.finish("error", str(e))
        writer.write(response(502, "Bad Gateway", error_page("Ziel nicht erreichbar", f"{host}:{port} – {e}"),
                              close=True))
        await writer.drain()
        return False
    writer.write(b"HTTP/1.1 200 Connection established\r\n\r\n")
    register(conn)
    try:
        await writer.drain()
        await relay(conn, reader, writer, r_reader, r_writer)
    finally:
        conn.finish("ok")
    return False


async def handle_plain(reader, writer, user, username, client_ip, method, target, version, headers) -> bool:
    url = urlsplit(target)
    if not url.scheme:
        body = error_page("Proxy-Server", "Dies ist ein Proxy-Server. Bitte im Browser als HTTP-Proxy eintragen.")
        writer.write(response(400, "Bad Request", body, close=True))
        return False
    if url.scheme.lower() != "http":
        writer.write(response(501, "Not Implemented",
                              error_page("Nicht unterstützt", f"Protokoll {url.scheme} wird nicht unterstützt."),
                              close=True))
        return False
    host = url.hostname or ""
    port = url.port or 80
    conn = Conn(username, client_ip, "http", host, port, method, target[:2000])
    try:
        r_reader, r_writer = await open_upstream(user, host, port)
    except Blocked as b:
        conn.finish("blocked", b.decision.reason)
        keep = body_is_discardable(headers)
        if keep:
            await transfer_body(headers, reader, None, lambda n: None)
        writer.write(response(403, "Forbidden", error_page("Zugriff gesperrt",
                                                           f"{host} ist durch die Proxy-Richtlinie gesperrt."),
                              close=not keep))
        await writer.drain()
        return keep
    except OSError as e:
        conn.finish("error", str(e))
        writer.write(response(502, "Bad Gateway", error_page("Ziel nicht erreichbar", f"{host}:{port} – {e}"),
                              close=True))
        await writer.drain()
        return False

    register(conn)
    status = "ok"
    detail = ""

    def up(n):
        conn.bytes_up += n

    def down(n):
        conn.bytes_down += n

    conn.closers += [writer.close, r_writer.close]
    try:
        # Hop-by-Hop-Header entfernen (inkl. der in "Connection" genannten)
        conn_tokens = {t.strip().lower() for t in (get_header(headers, "connection") or "").split(",")}
        out_headers = [(k, v) for k, v in headers if k.lower() not in HOP_BY_HOP and k.lower() not in conn_tokens]
        if not get_header(out_headers, "host"):
            out_headers.insert(0, ("Host", url.netloc.rsplit("@", 1)[-1]))
        path = url.path or "/"
        if url.query:
            path += "?" + url.query
        head = f"{method} {path} HTTP/1.1\r\n" + "".join(f"{k}: {v}\r\n" for k, v in out_headers) + \
               "Connection: close\r\n\r\n"
        r_writer.write(head.encode("latin-1"))
        if (get_header(headers, "expect") or "").lower() == "100-continue":
            writer.write(b"HTTP/1.1 100 Continue\r\n\r\n")
        await transfer_body(headers, reader, r_writer, up)
        await r_writer.drain()

        # Antwort lesen (1xx durchreichen)
        while True:
            rhead = await asyncio.wait_for(r_reader.readuntil(b"\r\n\r\n"), timeout=120)
            rfirst, rheaders = parse_head(rhead[:-4])
            parts = rfirst.split(" ", 2)
            code = int(parts[1])
            if 100 <= code < 200 and code != 101:
                writer.write(rhead)
                continue
            break
        detail = str(code)
        kept = [(k, v) for k, v in rheaders if k.lower() not in ("connection", "keep-alive", "proxy-connection")]
        writer.write((rfirst + "\r\n" + "".join(f"{k}: {v}\r\n" for k, v in kept) + "Connection: close\r\n\r\n")
                     .encode("latin-1"))
        down(len(rhead))
        no_body = method.upper() == "HEAD" or code in (204, 304)
        if not no_body:
            te = (get_header(rheaders, "transfer-encoding") or "").lower()
            if "chunked" in te or get_header(rheaders, "content-length"):
                await transfer_body(rheaders, r_reader, writer, down)
            else:
                await copy_until_eof(r_reader, writer, down)
        await writer.drain()
    except asyncio.CancelledError:
        status, detail = "error", "Verbindung beendet"
        raise
    except (OSError, asyncio.IncompleteReadError, asyncio.LimitOverrunError, asyncio.TimeoutError,
            ValueError, BadRequest) as e:
        status, detail = "error", (str(e) or e.__class__.__name__)[:300]
        if conn.bytes_down == 0:
            try:
                writer.write(response(502, "Bad Gateway", error_page("Fehler", detail), close=True))
            except Exception:
                pass
    finally:
        try:
            r_writer.close()
        except Exception:
            pass
        conn.finish(status, detail)
    return False