"""Security and transport contract for the OBS loopback relay."""
from __future__ import annotations

import json
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from types import SimpleNamespace
from urllib.error import HTTPError
from urllib.request import Request, urlopen

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "deploy"))
import private_feed_relay as relay  # noqa: E402


class _Tokens:
    def __init__(self):
        self.calls: list[bool] = []

    def get(self, *, force: bool = False) -> str:
        self.calls.append(force)
        return "private-id-token"


class _UpstreamHandler(BaseHTTPRequestHandler):
    seen: list[tuple[str, str | None]] = []

    def do_GET(self):  # noqa: N802
        self.seen.append((self.path, self.headers.get("Authorization")))
        if self.path == "/events":
            body = b"event: init\ndata: {\"private\":true}\n\n"
            content_type = "text/event-stream"
        else:
            body = b"private producer"
            content_type = "text/plain"
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):  # noqa: N802
        length = int(self.headers["Content-Length"])
        body = self.rfile.read(length)
        self.seen.append((self.path, self.headers.get("Authorization")))
        assert self.headers["Content-Type"] == "application/json"
        assert json.loads(body) == {"fps": 30}
        self.send_response(204)
        self.end_headers()

    def log_message(self, _fmt, *_args):
        pass


@pytest.fixture
def private_relay():
    upstream = ThreadingHTTPServer(("127.0.0.1", 0), _UpstreamHandler)
    _UpstreamHandler.seen = []
    upstream_thread = threading.Thread(target=upstream.serve_forever, daemon=True)
    upstream_thread.start()
    tokens = _Tokens()
    proxy = relay.PrivateFeedRelay(
        0, f"http://127.0.0.1:{upstream.server_address[1]}", tokens
    )
    proxy_thread = threading.Thread(target=proxy.serve_forever, daemon=True)
    proxy_thread.start()
    try:
        yield proxy, tokens
    finally:
        proxy.shutdown()
        proxy.server_close()
        upstream.shutdown()
        upstream.server_close()


def test_relay_binds_only_loopback_and_authenticates_upstream(private_relay):
    proxy, tokens = private_relay
    assert proxy.server_address[0] == "127.0.0.1"
    base = f"http://127.0.0.1:{proxy.server_address[1]}"

    with urlopen(f"{base}/events", timeout=5) as response:
        assert response.read().startswith(b"event: init")
        assert response.headers["Cache-Control"] == "no-store"
        assert response.headers.get("Access-Control-Allow-Origin") is None

    assert _UpstreamHandler.seen == [("/events", "Bearer private-id-token")]
    assert tokens.calls == [False]


def test_relay_logs_never_include_query_credentials(private_relay, capsys):
    proxy, _tokens = private_relay
    base = f"http://127.0.0.1:{proxy.server_address[1]}"
    with urlopen(f"{base}/asset?token=must-not-be-logged", timeout=5) as response:
        assert response.status == 200
    log = capsys.readouterr().out
    assert "GET /asset" in log
    assert "must-not-be-logged" not in log


def test_relay_rejects_cross_site_and_mutating_requests(private_relay):
    proxy, _tokens = private_relay
    base = f"http://127.0.0.1:{proxy.server_address[1]}"

    with pytest.raises(HTTPError) as cross_site:
        urlopen(Request(f"{base}/events", headers={"Sec-Fetch-Site": "cross-site"}), timeout=5)
    assert cross_site.value.code == 403

    with pytest.raises(HTTPError) as post:
        urlopen(Request(f"{base}/events", data=b"x", method="POST"), timeout=5)
    assert post.value.code == 405
    assert _UpstreamHandler.seen == []


def test_relay_allows_only_the_bounded_renderer_health_post(private_relay):
    proxy, _tokens = private_relay
    base = f"http://127.0.0.1:{proxy.server_address[1]}"
    request = Request(
        f"{base}/client-stats",
        data=b'{"fps":30}',
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=5) as response:
        assert response.status == 204
    assert _UpstreamHandler.seen == [("/client-stats", "Bearer private-id-token")]

    oversized = Request(
        f"{base}/client-stats",
        data=b"x" * 16385,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with pytest.raises(HTTPError) as rejected:
        urlopen(oversized, timeout=5)
    assert rejected.value.code == 413


def test_identity_token_is_cached_but_can_be_forced_to_refresh():
    now = 1_800_000_000
    payload = json.dumps({"exp": now + 3600}).encode()
    encoded = relay.base64.urlsafe_b64encode(payload).rstrip(b"=").decode()
    token = f"header.{encoded}.signature"
    calls = []

    def runner(command, **_kwargs):
        calls.append(command)
        return SimpleNamespace(returncode=0, stdout=token, stderr="")

    provider = relay.GcloudIdentityToken("gcloud.cmd", runner=runner, clock=lambda: now)
    assert provider.get() == token
    assert provider.get() == token
    assert len(calls) == 1
    assert provider.get(force=True) == token
    assert len(calls) == 2


def test_discovery_refuses_a_non_tls_producer_origin():
    def runner(_command, **_kwargs):
        return SimpleNamespace(returncode=0, stdout="http://producer.example", stderr="")

    with pytest.raises(relay.RelayError, match="HTTPS"):
        relay.discover_upstream(
            service="producer", project="project", region="region", runner=runner
        )