"""Security hardening of the rate-limit layer: X-Forwarded-For handling (A1), per-endpoint auth throttles and
fail-closed behaviour (A5/A14), bounded key cache + cross-worker invalidation (A12/A20), key-lookup throttle."""
from __future__ import annotations

from dataclasses import replace

from starlette.datastructures import Headers

JSON = {"Content-Type": "application/json"}


def scope_with(client_host: str | None, **headers) -> tuple[dict, Headers]:
    raw = [(k.lower().encode(), v.encode()) for k, v in headers.items()]
    scope = {"type": "http", "headers": raw, "client": (client_host, 1234) if client_host else None}
    return scope, Headers(scope=scope)


def test_client_ip_takes_the_hop_appended_by_the_proxy(app):
    """A1: the FIRST hop is attacker-controlled; ngrok appends the real peer at the END."""
    from ratelimit import middleware as mw
    scope, h = scope_with("127.0.0.1", **{"X-Forwarded-For": "6.6.6.6, 198.51.100.23"})
    assert mw.client_ip(scope, h) == "198.51.100.23"
    scope, h = scope_with("10.0.0.2", **{"X-Forwarded-For": "198.51.100.23"})
    assert mw.client_ip(scope, h) == "198.51.100.23"
    # two trusted proxies → second from the end
    scope, h = scope_with("10.0.0.2", **{"X-Forwarded-For": "6.6.6.6, 198.51.100.23, 10.0.0.9"})
    assert mw.client_ip(scope, h) == "10.0.0.9"
    old = mw.settings
    mw.settings = replace(old, trusted_proxy_hops=2)
    try:
        assert mw.client_ip(scope, h) == "198.51.100.23"
        # hops=0 → the header is ignored entirely
        mw.settings = replace(old, trusted_proxy_hops=0)
        assert mw.client_ip(scope, h) == "10.0.0.2"
    finally:
        mw.settings = old
    # direct peer is a public address (no proxy in front) → X-Forwarded-For is not trusted
    scope, h = scope_with("8.8.8.8", **{"X-Forwarded-For": "6.6.6.6"})       # (203.0.113.x is "private" for ipaddress)
    assert mw.client_ip(scope, h) == "8.8.8.8"
    # X-Real-IP only when trusted
    scope, h = scope_with("127.0.0.1", **{"X-Real-IP": "198.51.100.77"})
    assert mw.client_ip(scope, h) == "198.51.100.77"
    scope, h = scope_with("8.8.8.8", **{"X-Real-IP": "198.51.100.77"})
    assert mw.client_ip(scope, h) == "8.8.8.8"
    scope, h = scope_with(None)
    assert mw.client_ip(scope, h) == "unknown"


def test_spoofed_first_hop_cannot_escape_the_keyless_quota(client):
    """A1 (live reproduction): varying the first hop while ngrok appends the same peer = one principal."""
    for i in range(30):
        r = client.get("/v1/status", headers={"X-Forwarded-For": f"1.2.3.{i}, 198.51.100.200"})
        assert r.status_code == 200, i
    r = client.get("/v1/status", headers={"X-Forwarded-For": "9.9.9.9, 198.51.100.200"})
    assert r.status_code == 429
    # …and the auth throttle cannot be bypassed either
    for i in range(20):
        r = client.post("/v1/auth/login", json={"email": "nobody@example.com", "password": "x"},
                        headers={**JSON, "X-Forwarded-For": f"1.2.3.{i}, 198.51.100.201"})
        assert r.status_code == 401, i
    r = client.post("/v1/auth/login", json={"email": "nobody@example.com", "password": "x"},
                    headers={**JSON, "X-Forwarded-For": "7.7.7.7, 198.51.100.201"})
    assert r.status_code == 429


def test_auth_throttle_is_per_endpoint(client):
    """A5/A14: signup 5/h, forgot 5/h, login 20/h, verify 30/h — separate buckets per IP."""
    h = {**JSON, "X-Forwarded-For": "198.51.100.10"}
    for i in range(5):
        assert client.post("/v1/auth/signup", json={"email": "bad", "name": "x", "password": "long-enough-password"}, headers=h).status_code == 400, i
    r = client.post("/v1/auth/signup", json={"email": "bad", "name": "x", "password": "long-enough-password"}, headers=h)
    assert r.status_code == 429 and r.json()["error"]["details"]["endpoint"] == "signup" and "Retry-After" in r.headers
    # the login bucket of the same IP is untouched
    assert client.post("/v1/auth/login", json={"email": "nobody@example.com", "password": "x"}, headers=h).status_code == 401
    for i in range(5):
        assert client.post("/v1/auth/forgot", json={"email": f"n{i}@example.com"}, headers=h).status_code == 202
    assert client.post("/v1/auth/forgot", json={"email": "n9@example.com"}, headers=h).status_code == 429
    for i in range(30):
        assert client.post("/v1/auth/verify", json={"token": "nope"}, headers=h).status_code == 400, i
    assert client.post("/v1/auth/verify", json={"token": "nope"}, headers=h).status_code == 429
    # logout is never throttled
    for _ in range(40):
        assert client.post("/v1/auth/logout", json={}, headers=h).status_code == 200


def test_auth_fails_closed_when_redis_is_down_but_data_fails_open(client, monkeypatch):
    """A5: no Redis → login/signup/forgot answer 429 (Retry-After 30); data endpoints keep working."""
    import redis as redis_lib

    from ratelimit import redis_limiter as rl

    class Broken:
        def __getattr__(self, name):
            def _raise(*a, **k):
                raise redis_lib.exceptions.ConnectionError("down")
            return _raise

    monkeypatch.setattr(rl, "_client", Broken())
    monkeypatch.setattr(rl, "_sha", None)
    monkeypatch.setattr(rl, "_down_until", 0.0)
    h = {**JSON, "X-Forwarded-For": "198.51.100.11"}
    for path, body in (("/v1/auth/login", {"email": "a@example.com", "password": "x"}),
                       ("/v1/auth/signup", {"email": "a@example.com", "name": "x", "password": "long-enough-password"}),
                       ("/v1/auth/forgot", {"email": "a@example.com"})):
        r = client.post(path, json=body, headers=h)
        assert r.status_code == 429 and r.headers["Retry-After"] == "30", path
    assert client.post("/v1/auth/logout", json={}, headers=h).status_code == 200
    assert client.post("/v1/auth/verify", json={"token": "nope"}, headers=h).status_code == 400   # not in the fail-closed set
    r = client.get("/v1/status", headers={"X-Forwarded-For": "198.51.100.11"})
    assert r.status_code == 200 and "X-RateLimit-Limit-Requests" not in r.headers


def test_key_cache_is_bounded_and_invalidated_across_workers(client, make_user, monkeypatch):
    """A12/A20: LRU 10 000 with TTL; a Redis generation bump makes every worker miss."""
    from ratelimit import middleware as mw
    from ratelimit import redis_limiter as rl
    lru = mw._LRU(maxsize=3, ttl_s=60)
    for i in range(5):
        lru.put(f"k{i}", i, generation=1)
    assert len(lru) == 3 and lru.get("k0", 1) == (False, None) and lru.get("k4", 1) == (True, 4)
    assert lru.get("k4", 2) == (False, None)            # generation changed → miss
    lru.put("neg", None, generation=None)
    assert lru.get("neg", 7) == (True, None)             # negatives are cached, None generation = not tracked
    assert mw._key_cache.maxsize == 10_000 and mw._key_cache.ttl_s == 60

    _, key, _, _ = make_user()
    calls = {"n": 0}
    real = mw._lookup_key

    def counting(h, ip_hash=None):
        calls["n"] += 1
        return real(h, ip_hash)

    monkeypatch.setattr(mw, "_lookup_key", counting)
    for _ in range(3):
        assert client.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).status_code == 200
    assert calls["n"] == 1                               # cached
    rl.bump_keys_version()                               # what another worker does after a revoke / tier change
    assert client.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).status_code == 200
    assert calls["n"] == 2
    # unknown keys are cached negatively: 3 requests, 1 lookup
    unknown = "hfmd_live_" + "z" * 32
    for _ in range(3):
        assert client.get("/v1/status", headers={"Authorization": f"Bearer {unknown}"}).status_code == 401
    assert calls["n"] == 3


def test_unknown_key_lookups_are_throttled_before_sqlite(client, monkeypatch):
    """A12: brute-forcing keys hits a per-IP throttle (120/min) before the database."""
    from ratelimit import middleware as mw
    from ratelimit.tiers import KEY_LOOKUPS_PER_MINUTE_PER_IP
    calls = {"n": 0}
    real = mw._lookup_key

    def counting(h, ip_hash=None):
        calls["n"] += 1
        return real(h, ip_hash)

    monkeypatch.setattr(mw, "_lookup_key", counting)
    h = {"X-Forwarded-For": "198.51.100.12"}
    for i in range(KEY_LOOKUPS_PER_MINUTE_PER_IP):
        r = client.get("/v1/status", headers={**h, "Authorization": f"Bearer hfmd_live_{i:032d}"})
        assert r.status_code == 401, i
    r = client.get("/v1/status", headers={**h, "Authorization": "Bearer hfmd_live_" + "q" * 32})
    assert r.status_code == 429 and "Retry-After" in r.headers
    assert calls["n"] == KEY_LOOKUPS_PER_MINUTE_PER_IP   # the 121st never reached SQLite
    # another IP is unaffected
    assert client.get("/v1/status", headers={"X-Forwarded-For": "198.51.100.13", "Authorization": "Bearer hfmd_live_" + "q" * 32}).status_code == 401


def test_new_error_codes_are_documented(client):
    spec = client.get("/openapi.json").json()
    codes = spec["components"]["schemas"]["Error"]["properties"]["error"]["properties"]["code"]["enum"]
    assert {"ACCOUNT_LOCKED", "SESSION_REQUIRED", "LAST_ADMIN"} <= set(codes)
    login = spec["paths"]["/v1/auth/login"]["post"]
    assert "423" in login["responses"] and "ACCOUNT_LOCKED" in login["responses"]["423"]["description"]
    assert "SESSION_REQUIRED" in spec["paths"]["/v1/me/keys"]["post"]["responses"]["403"]["description"]
    assert spec["paths"]["/v1/auth/verify"]["post"]["summary"] and spec["paths"]["/v1/auth/verify"]["get"]["responses"]["302"]
    for p in ("/v1/me/password", "/v1/me/email", "/v1/me/sessions/revoke-all", "/v1/me/limits", "/v1/me/usage.csv", "/v1/admin/users/{user_id}"):
        assert p in spec["paths"], p
    assert "delete" in spec["paths"]["/v1/me"] and "delete" in spec["paths"]["/v1/admin/users/{user_id}"]