"""Adversarial HTTP boundary tests for the loopback HUX foundation service."""

from __future__ import annotations

import json
import socket
import sys
import threading
from http.client import HTTPConnection
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
    sys.path.insert(0, str(FOUNDATION))

from hux import contracts, errors, flags, identity  # noqa: E402
from hux import http as hux_http  # noqa: E402
from hux.http import DEFAULT_REQUEST_TIMEOUT_SECONDS, RateLimiter, Router, serve  # noqa: E402
from hux.server import build_router  # noqa: E402

ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
HEADERS = {
    "X-Hermes-Tenant-Identity": "slot-3",
    "X-Hux-Subject": "usr_0123456789abcdef",
    "X-Hux-Surface": "chat",
    "X-Hux-Relay-Key": "rk",
}


def _router(tmp_path: Path, **extra: str) -> Router:
    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk", **extra})


def _start(router: Router):
    server = serve(router, "127.0.0.1", 0)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


def test_each_trust_uses_its_own_0400_key_file(tmp_path):
    """Router, relay and worker credentials are distinct files; weak permissions fail closed."""
    files = {}
    for trust in ("router", "relay", "worker"):
        path = tmp_path / f"{trust}-key"
        path.write_text(f"{trust}-secret\n")
        path.chmod(0o400)
        files[trust] = path
    base = {key: value for key, value in HEADERS.items() if key != "X-Hux-Relay-Key"}
    for trust, surface in (("router", "chat"), ("relay", "telegram"), ("worker", "worker")):
        headers = {**base, "X-Hux-Trust": trust, "X-Hux-Surface": surface, "X-Hux-Relay-Key": f"{trust}-secret"}
        env = {f"HUX_{trust.upper()}_KEY_FILE": str(files[trust])}
        assert identity.resolve(headers, env).trust == trust
        wrong = files["router" if trust != "router" else "worker"]
        with pytest.raises(errors.Unauthorized):
            identity.resolve(headers, {f"HUX_{trust.upper()}_KEY_FILE": str(wrong)})
    files["router"].chmod(0o440)
    with pytest.raises(errors.Unauthorized):
        identity.resolve({**base, "X-Hux-Relay-Key": "router-secret"}, {"HUX_ROUTER_KEY_FILE": str(files["router"])})


def _trusted_headers(trust: str, subject: str = "usr_0123456789abcdef") -> dict[str, str]:
    """Build headers for one separately keyed authenticated hop."""
    surfaces = {"router": "chat", "relay": "telegram", "worker": "worker"}
    return {
        **HEADERS,
        "X-Hux-Subject": subject,
        "X-Hux-Trust": trust,
        "X-Hux-Surface": surfaces[trust],
        "X-Hux-Relay-Key": f"{trust}-key",
    }


def _binding_env(path: Path) -> dict[str, str]:
    """Return the three-key environment used by subject-binding tests."""
    return {
        "HUX_ROUTER_KEY": "router-key",
        "HUX_RELAY_KEY": "relay-key",
        "HUX_WORKER_KEY": "worker-key",
        "HUX_SUBJECT_BINDING_FILE": str(path),
    }


def test_edge_hop_binds_subject_before_worker_can_assert_it(tmp_path):
    """Only a keyed router/relay may initialize the file; all later callers must match it."""
    binding = tmp_path / "subject"
    environ = _binding_env(binding)
    with pytest.raises(errors.Unauthorized, match="not bound"):
        identity.resolve(_trusted_headers("worker"), environ)
    assert identity.resolve(_trusted_headers("router"), environ).subject == "usr_0123456789abcdef"
    assert binding.read_text().strip() == "usr_0123456789abcdef"
    assert binding.stat().st_mode & 0o777 == 0o440
    assert identity.resolve(_trusted_headers("worker"), environ).trust == "worker"
    other = "usr_fedcba9876543210"
    for trust in ("router", "relay", "worker"):
        with pytest.raises(errors.Unauthorized, match="does not match"):
            identity.resolve(_trusted_headers(trust, other), environ)


def test_relay_can_initialize_binding_and_invalid_files_fail_closed(tmp_path):
    """Relay is an authoritative edge; malformed, linked, weak, and unavailable files never authenticate."""
    relay_binding = tmp_path / "relay-subject"
    assert identity.resolve(_trusted_headers("relay"), _binding_env(relay_binding)).trust == "relay"
    for name, payload, mode in (
        ("empty", b"", 0o440),
        ("oversized", b"x" * (identity.MAX_SUBJECT_BYTES + 1), 0o440),
        ("unicode", b"\xff", 0o440),
        ("malformed", b"brad@example.test", 0o440),
        ("weak", b"usr_0123456789abcdef", 0o444),
    ):
        binding = tmp_path / name
        binding.write_bytes(payload)
        binding.chmod(mode)
        with pytest.raises(errors.Unauthorized):
            identity.resolve(_trusted_headers("router"), _binding_env(binding))
    link = tmp_path / "linked"
    link.symlink_to(relay_binding)
    with pytest.raises(errors.Unauthorized):
        identity.resolve(_trusted_headers("router"), _binding_env(link))
    with pytest.raises(errors.Unauthorized, match="unavailable"):
        identity.resolve(_trusted_headers("router"), _binding_env(tmp_path / "missing" / "subject"))


def test_concurrent_edge_binding_has_exactly_one_subject(tmp_path):
    """First-writer publication is complete and immutable even when two edge hops race."""
    binding = tmp_path / "subject"
    environ = _binding_env(binding)
    barrier = threading.Barrier(2)
    outcomes: list[tuple[str, bool]] = []

    def bind(trust: str, subject: str) -> None:
        barrier.wait()
        try:
            identity.resolve(_trusted_headers(trust, subject), environ)
        except errors.Unauthorized:
            outcomes.append((subject, False))
        else:
            outcomes.append((subject, True))

    attempts = (
        threading.Thread(target=bind, args=("router", "usr_0123456789abcdef")),
        threading.Thread(target=bind, args=("relay", "usr_fedcba9876543210")),
    )
    for attempt in attempts:
        attempt.start()
    for attempt in attempts:
        attempt.join()
    bound = binding.read_text().strip()
    assert sorted(ok for _, ok in outcomes) == [False, True]
    assert bound in {subject for subject, ok in outcomes if ok}


def test_capability_flags_are_bound_to_the_routes_really_registered():
    """A configured card stays off when a declared route is absent; undeclared routes fail startup."""
    all_routes = {card: set(routes) for card, routes in flags.CARD_ROUTES.items() if routes}
    bound = flags.Flags({"HUX_FLAGS": ALL_ON})
    missing = {card: set(routes) for card, routes in all_routes.items()}
    missing["HUX-01"].remove("/hux/v1/conversations/{id}/events/stream")
    bound.bind_routes(missing)
    assert not bound.enabled("HUX-01") and bound.enabled("HUX-11")
    with pytest.raises(ValueError, match="unknown cards"):
        bound.bind_routes({"HUX-99": {"/hux/v1/nope"}})
    with pytest.raises(ValueError, match="undeclared routes"):
        bound.bind_routes({"HUX-11": {"/hux/v1/nope"}})


def _raw_request(address: tuple[str, int], request: bytes) -> bytes:
    """Send one HTTP/1.0 request and return its complete close-delimited response."""
    with socket.create_connection(address, timeout=1) as client:
        client.sendall(request)
        chunks = []
        while chunk := client.recv(4096):
            chunks.append(chunk)
    return b"".join(chunks)


def test_declared_body_is_rejected_before_socket_read(tmp_path):
    """An oversized Content-Length receives 413 without waiting for the claimed body."""
    server = _start(_router(tmp_path))
    try:
        conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=1)
        conn.putrequest("POST", "/hux/v1/capabilities")
        conn.putheader("Content-Length", str(1024 * 1024 + 1))
        conn.endheaders()
        reply = conn.getresponse()
        body = json.loads(reply.read())
        assert (reply.status, body["code"]) == (413, "too_large")
        assert reply.getheader("Cache-Control") == "no-store"
    finally:
        server.shutdown()
        server.server_close()


def test_incomplete_body_times_out_and_transfer_encoding_is_rejected(tmp_path):
    """Accepted sockets have a deadline, and unsupported framing fails closed."""
    server = _start(_router(tmp_path, HUX_REQUEST_TIMEOUT_SECONDS="0.1"))
    try:
        reply = _raw_request(
            server.server_address,
            b"POST /hux/v1/capabilities HTTP/1.0\r\nContent-Length: 2\r\n\r\n{",
        )
        assert b" 400 " in reply and b"request body is incomplete or timed out" in reply
        conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=1)
        conn.putrequest("POST", "/hux/v1/capabilities")
        conn.putheader("Transfer-Encoding", "chunked")
        conn.endheaders()
        assert conn.getresponse().status == 400
    finally:
        server.shutdown()
        server.server_close()


def test_malformed_duplicate_and_absurd_content_lengths_fail_closed(tmp_path):
    """Ambiguous or computationally large length headers never reach ``int`` or a body allocation."""
    server = _start(_router(tmp_path))
    try:
        malformed = _raw_request(
            server.server_address,
            b"POST /hux/v1/capabilities HTTP/1.0\r\nContent-Length: nope\r\n\r\n",
        )
        duplicate = _raw_request(
            server.server_address,
            b"POST /hux/v1/capabilities HTTP/1.0\r\nContent-Length: 0\r\nContent-Length: 0\r\n\r\n",
        )
        absurd = _raw_request(
            server.server_address,
            b"POST /hux/v1/capabilities HTTP/1.0\r\nContent-Length: 99999999999\r\n\r\n",
        )
        assert b" 400 " in malformed and b" 400 " in duplicate
        assert b" 413 " in absurd
    finally:
        server.shutdown()
        server.server_close()


def test_rate_limit_is_per_subject_and_method_class(tmp_path):
    """Read and write buckets are distinct and return a bounded Retry-After."""
    router = _router(tmp_path, HUX_READS_PER_MINUTE="1", HUX_WRITES_PER_MINUTE="1")
    first = router.dispatch("GET", "/hux/v1/capabilities", HEADERS, b"")
    limited = router.dispatch("GET", "/hux/v1/capabilities", HEADERS, b"")
    write = router.dispatch("POST", "/hux/v1/capabilities", HEADERS, b"{}")
    assert first.status == 200
    assert (limited.status, limited.body["code"]) == (429, "rate_limited")
    assert 1 <= int(limited.headers["Retry-After"]) <= 60
    assert write.status == 405, "method routing precedes the write rate bucket"
    live = _start(_router(tmp_path / "live", HUX_READS_PER_MINUTE="1"))
    try:
        conn = HTTPConnection("127.0.0.1", live.server_address[1], timeout=1)
        conn.request("GET", "/hux/v1/capabilities", headers=HEADERS)
        conn.getresponse().read()
        conn.request("GET", "/hux/v1/capabilities", headers=HEADERS)
        reply = conn.getresponse()
        reply.read()
        assert reply.status == 429 and reply.getheader("Retry-After")
        assert reply.getheader("Cache-Control") == "no-store"
    finally:
        live.shutdown()
        live.server_close()


def test_limiter_expires_windows_and_invalid_settings_use_safe_defaults(tmp_path, monkeypatch):
    """Expired subjects are evicted and malformed operator settings cannot disable bounds."""
    moments = iter((0.0, 0.0, 61.0))
    limiter = RateLimiter(1, 1, clock=lambda: next(moments))
    assert limiter.check("a", "GET") is None
    assert limiter.check("a", "GET") == 60
    assert limiter.check("a", "GET") is None
    monkeypatch.setattr(hux_http, "MAX_RATE_BUCKETS", 1)
    bounded = RateLimiter(2, 2, clock=lambda: 0.0)
    assert bounded.check("a", "GET") is None
    assert bounded.check("b", "GET") == 60
    router = _router(
        tmp_path,
        HUX_READS_PER_MINUTE="nan",
        HUX_WRITES_PER_MINUTE="0",
        HUX_REQUEST_TIMEOUT_SECONDS="forever",
    )
    assert router.rate_limiter.limits == {"read": 300, "write": 30}
    assert router.request_timeout == DEFAULT_REQUEST_TIMEOUT_SECONDS


def test_health_and_json_are_no_store_and_health_refuses_bodies(tmp_path):
    """Sensitive JSON never enters browser caches; health cannot be used for body smuggling."""
    server = _start(_router(tmp_path))
    try:
        conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=1)
        conn.request("GET", "/healthz")
        reply = conn.getresponse()
        reply.read()
        assert reply.getheader("Cache-Control") == "no-store"
        assert reply.getheader("X-Content-Type-Options") == "nosniff"
        conn.request("GET", "/healthz", body=b"x", headers={"Content-Length": "1"})
        assert conn.getresponse().status == 413
    finally:
        server.shutdown()
        server.server_close()


def test_unknown_routes_are_limited_and_query_fields_are_bounded(tmp_path):
    """Authenticated misses consume their bucket and oversized query maps fail as typed 400s."""
    router = _router(tmp_path, HUX_READS_PER_MINUTE="1")
    missing = router.dispatch("GET", "/hux/v1/nope", HEADERS, b"")
    limited = router.dispatch("GET", "/hux/v1/nope", HEADERS, b"")
    assert missing.status == 404
    assert (limited.status, limited.body["code"]) == (429, "rate_limited")
    roomy = _router(tmp_path / "roomy")
    query = "&".join(f"x{index}=1" for index in range(hux_http.MAX_QUERY_FIELDS + 1))
    response = roomy.dispatch("GET", f"/hux/v1/capabilities?{query}", HEADERS, b"")
    assert (response.status, response.body["code"]) == (400, "invalid")