"""Admin session cookies, CSRF tokens, rate limiting and password checks."""
from __future__ import annotations

import hmac
import ipaddress
import secrets
import threading
import time
from collections import defaultdict, deque
from typing import Optional

from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer

SESSION_COOKIE = "curas_admin_session"


class SessionManager:
    def __init__(self, secret: str, max_age_seconds: int):
        self._s = URLSafeTimedSerializer(secret, salt="curas-admin-session")
        self.max_age = max_age_seconds

    def create(self) -> tuple[str, str]:
        """Return (cookie_value, csrf_token) for a fresh admin session."""
        csrf = secrets.token_urlsafe(32)
        return self._s.dumps({"csrf": csrf, "v": 1}), csrf

    def read(self, cookie_value: Optional[str]) -> Optional[dict]:
        if not cookie_value:
            return None
        try:
            data = self._s.loads(cookie_value, max_age=self.max_age)
        except (BadSignature, SignatureExpired):
            return None
        if not isinstance(data, dict) or "csrf" not in data:
            return None
        return data


class FormTokens:
    """Signed, time-limited tokens for the anonymous public form.

    The token is carried only in a hidden input (no cookie), so the form also
    works when the site is shown inside another site's frame, such as the
    Space page on huggingface.co. The form is anonymous, so there is no
    session for a cross-site request to ride on; the token's job is to
    accept posts only from a form this server rendered recently.
    """

    def __init__(self, secret: str, max_age_seconds: int = 6 * 3600):
        self._s = URLSafeTimedSerializer(secret, salt="curas-form-token")
        self.max_age = max_age_seconds

    def issue(self) -> str:
        return self._s.dumps({"n": secrets.token_urlsafe(16)})

    def verify(self, form_value: Optional[str]) -> bool:
        if not form_value:
            return False
        try:
            data = self._s.loads(str(form_value), max_age=self.max_age)
        except (BadSignature, SignatureExpired):
            return False
        return isinstance(data, dict) and "n" in data


class RateLimiter:
    """In-memory sliding-window limiter. Adequate for a single-instance Space."""

    def __init__(self):
        self._events: dict[tuple[str, str], deque] = defaultdict(deque)
        self._lock = threading.Lock()

    def hit(self, bucket: str, key: str, limit: int, window_seconds: int) -> bool:
        """Record an event; return True if still within the limit."""
        now = time.monotonic()
        with self._lock:
            q = self._events[(bucket, key)]
            while q and now - q[0] > window_seconds:
                q.popleft()
            if len(q) >= limit:
                return False
            q.append(now)
            if len(self._events) > 10000:
                # Drop stale buckets so memory cannot grow without bound.
                for k in [k for k, v in self._events.items() if not v or now - v[-1] > window_seconds]:
                    del self._events[k]
            return True

    def is_limited(self, bucket: str, key: str, limit: int, window_seconds: int) -> bool:
        now = time.monotonic()
        with self._lock:
            q = self._events.get((bucket, key))
            if not q:
                return False
            while q and now - q[0] > window_seconds:
                q.popleft()
            return len(q) >= limit

    def reset(self, bucket: str, key: str) -> None:
        with self._lock:
            self._events.pop((bucket, key), None)


def password_matches(expected: str, supplied: str) -> bool:
    if not expected or supplied is None:
        return False
    return hmac.compare_digest(expected.encode("utf-8"), supplied.encode("utf-8"))


def client_ip(headers, fallback: str) -> str:
    """Client address as seen by the proxy in front of the app.

    Proxies append the real peer address to whatever X-Forwarded-For the
    client sent (Hugging Face Spaces does this, one hop), so the LAST entry is
    the trustworthy one; the first entry can be chosen by the client.
    """
    xff = headers.get("x-forwarded-for")
    host = fallback or "unknown"
    if xff:
        hosts = [h.strip() for h in xff.split(",") if h.strip()]
        if hosts:
            host = hosts[-1]
    return rate_limit_key(host)


def rate_limit_key(host: str) -> str:
    """IPv6 clients are keyed by their /64 so one network cannot mint endless addresses."""
    try:
        ip = ipaddress.ip_address(host)
    except ValueError:
        return host
    if ip.version == 6:
        return str(ipaddress.ip_network((ip, 64), strict=False))
    return host