"""Tenant-bound SupraTix mobile login and refresh session for the workcell UI."""

from __future__ import annotations

import ipaddress
import json
import math
import os
import re
import socket
import stat
import threading
import time
from pathlib import Path
from typing import Any, Callable, Mapping, Optional
from urllib.parse import urlsplit, urlunsplit

from .http import TransportError, UrllibTransport, json_body

LOGIN_PATH = "/api/speaksphere/mobile/auth/login"
REFRESH_PATH = "/api/speaksphere/mobile/auth/refresh"
SESSION_SCHEMA_VERSION = 1
MAX_AUTH_RESPONSE_BYTES = 64 * 1024
MAX_TOKEN_LENGTH = 16 * 1024
MAX_IDENTIFIER_LENGTH = 254
MAX_PASSWORD_LENGTH = 128
ACCESS_REFRESH_LEEWAY_SECONDS = 60
REFRESH_EXPIRY_LEEWAY_SECONDS = 5
_SESSION_FILENAME = "session.json"
_TENANT_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}\Z")
_DEVICE_ID_RE = re.compile(r"[A-Za-z0-9_.:-]{1,100}\Z")


class TenantAuthError(RuntimeError):
    """A bounded, non-secret tenant authentication failure."""

    def __init__(self, code: str) -> None:
        self.code = str(code)
        super().__init__(self.code)


def _is_loopback_host(host: str) -> bool:
    return host in {"localhost", "127.0.0.1", "::1"}


def _resolved_addresses_are_public(
    host: str,
    port: int,
    resolver: Callable[..., Any],
) -> bool:
    try:
        records = resolver(host, port, type=socket.SOCK_STREAM)
    except (OSError, socket.gaierror) as exc:
        raise TenantAuthError("tenant_unavailable") from exc
    addresses = set()
    for record in records:
        try:
            addresses.add(ipaddress.ip_address(record[4][0]))
        except (IndexError, TypeError, ValueError):
            raise TenantAuthError("invalid_tenant_url")
    return bool(addresses) and all(address.is_global for address in addresses)


def normalize_tenant_url(
    raw_value: object,
    *,
    allow_insecure_localhost: bool = False,
    resolver: Callable[..., Any] = socket.getaddrinfo,
    verify_dns: bool = True,
) -> str:
    """Normalize one domain URL and reject credentials, paths and private SSRF."""

    value = str(raw_value or "").strip()
    if not value:
        raise TenantAuthError("invalid_tenant_url")
    if "://" not in value:
        value = "https://" + value
    try:
        parsed = urlsplit(value)
        port = parsed.port
    except ValueError as exc:
        raise TenantAuthError("invalid_tenant_url") from exc
    scheme = parsed.scheme.lower()
    host = (parsed.hostname or "").lower().rstrip(".")
    if (
        not host
        or len(host) > 253
        or parsed.username is not None
        or parsed.password is not None
        or parsed.query
        or parsed.fragment
        or parsed.path not in ("", "/")
    ):
        raise TenantAuthError("invalid_tenant_url")
    localhost = _is_loopback_host(host)
    if scheme != "https" and not (
        allow_insecure_localhost and scheme == "http" and localhost
    ):
        raise TenantAuthError("insecure_tenant_url")
    effective_port = port or (443 if scheme == "https" else 80)
    if scheme == "https" and effective_port != 443:
        raise TenantAuthError("invalid_tenant_url")
    if localhost:
        if not allow_insecure_localhost:
            raise TenantAuthError("invalid_tenant_url")
    elif verify_dns and not _resolved_addresses_are_public(
        host,
        effective_port,
        resolver,
    ):
        raise TenantAuthError("invalid_tenant_url")
    authority = host
    if ":" in host:
        authority = "[{}]".format(host)
    if port is not None and not (
        (scheme == "https" and port == 443) or (scheme == "http" and port == 80)
    ):
        authority = "{}:{}".format(authority, port)
    return urlunsplit((scheme, authority, "", "", ""))


def _origin(value: str) -> tuple[str, str, int]:
    parsed = urlsplit(value)
    try:
        port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80)
    except ValueError as exc:
        raise TenantAuthError("invalid_endpoint_url") from exc
    return parsed.scheme.lower(), (parsed.hostname or "").lower(), port


def _bounded_token(value: object) -> str:
    token = str(value or "").strip()
    if (
        not token
        or len(token) > MAX_TOKEN_LENGTH
        or any(ord(character) < 33 or ord(character) > 126 for character in token)
    ):
        raise TenantAuthError("invalid_response")
    return token


def _positive_lifetime(value: object) -> int:
    if isinstance(value, bool):
        raise TenantAuthError("invalid_response")
    try:
        lifetime = int(value)
    except (TypeError, ValueError) as exc:
        raise TenantAuthError("invalid_response") from exc
    if not 1 <= lifetime <= 31_536_000:
        raise TenantAuthError("invalid_response")
    return lifetime


def _unwrap_envelope(value: object) -> Mapping[str, Any]:
    current = value
    for _depth in range(4):
        if not isinstance(current, Mapping):
            raise TenantAuthError("invalid_response")
        for key in ("data", "result", "payload"):
            nested = current.get(key)
            if isinstance(nested, Mapping):
                current = nested
                break
        else:
            return current
    raise TenantAuthError("invalid_response")


def _reject_duplicate_pairs(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError("duplicate JSON key")
        value[key] = item
    return value


def _reject_json_constant(_value):
    raise ValueError("non-finite JSON value")


def _session_values(
    payload: Mapping[str, Any],
    *,
    existing_refresh: str = "",
    existing_tenant: str = "",
) -> dict[str, Any]:
    session = payload.get("mobile_session") or payload.get("session")
    if not isinstance(session, Mapping):
        session = payload
    token_type = str(session.get("token_type") or "Bearer").strip()
    if token_type.lower() != "bearer":
        raise TenantAuthError("invalid_response")
    access_token = _bounded_token(session.get("access_token") or session.get("access"))
    refresh_token = _bounded_token(
        session.get("refresh_token") or session.get("refresh") or existing_refresh
    )
    tenant = str(session.get("tenant") or existing_tenant or "").strip()
    if not _TENANT_RE.fullmatch(tenant):
        raise TenantAuthError("invalid_response")
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "tenant": tenant,
        "access_lifetime": _positive_lifetime(session.get("access_expires_in")),
        "refresh_lifetime": _positive_lifetime(session.get("refresh_expires_in")),
    }


def _user_values(payload: Mapping[str, Any], existing: Mapping[str, Any]) -> dict:
    raw = payload.get("user")
    if not isinstance(raw, Mapping):
        raw = existing
    username = str(raw.get("username") or existing.get("username") or "").strip()
    display_name = str(
        raw.get("display_name") or existing.get("display_name") or username
    ).strip()
    if not username or len(username) > MAX_IDENTIFIER_LENGTH:
        raise TenantAuthError("invalid_response")
    return {
        "username": username,
        "display_name": display_name[:MAX_IDENTIFIER_LENGTH],
    }


class TenantMobileAuthClient:
    """Own one tenant-bound mobile JWT session without retaining passwords."""

    def __init__(
        self,
        *,
        state_dir: Path,
        transport: Optional[UrllibTransport] = None,
        timeout: float = 30.0,
        allow_insecure_localhost: bool = False,
        resolver: Callable[..., Any] = socket.getaddrinfo,
        wall_clock: Callable[[], float] = time.time,
        device_registration: Optional[Mapping[str, str]] = None,
    ) -> None:
        self.state_dir = Path(state_dir)
        self.transport = transport or UrllibTransport()
        self.timeout = min(max(float(timeout), 1.0), 60.0)
        self.allow_insecure_localhost = bool(allow_insecure_localhost)
        self.resolver = resolver
        self.wall_clock = wall_clock
        self.device_registration = self._validated_device_registration(
            device_registration
        )
        self._lock = threading.RLock()
        self._session: dict[str, Any] = {}
        self._access_token = ""
        self._access_expires_at = 0.0
        self._session_persisted = False
        self._last_error = ""
        self._restore()

    def login(
        self, tenant_url: object, username: object, password: object
    ) -> Mapping[str, Any]:
        """Authenticate using the same allauth-compatible mobile API as SupraTracker."""

        normalized = normalize_tenant_url(
            tenant_url,
            allow_insecure_localhost=self.allow_insecure_localhost,
            resolver=self.resolver,
        )
        identifier = str(username or "").strip()
        secret = str(password or "")
        if (
            not identifier
            or len(identifier) > MAX_IDENTIFIER_LENGTH
            or not secret
            or len(secret) > MAX_PASSWORD_LENGTH
        ):
            raise TenantAuthError("missing_credentials")
        with self._lock:
            existing_url = str(self._session.get("base_url") or "")
            if existing_url and existing_url != normalized:
                raise TenantAuthError("tenant_change_requires_logout")
            request_payload = {
                "login": identifier,
                "username": identifier,
                "password": secret,
            }
            if self.device_registration:
                request_payload["device"] = dict(self.device_registration)
            payload = self._request(
                normalized + LOGIN_PATH,
                request_payload,
                login=True,
            )
            session_values = _session_values(payload)
            user = _user_values(payload, {})
            now = self.wall_clock()
            self._session = {
                "schema_version": SESSION_SCHEMA_VERSION,
                "base_url": normalized,
                "tenant": session_values["tenant"],
                "refresh_token": session_values["refresh_token"],
                "refresh_expires_at": now + session_values["refresh_lifetime"],
                "user": user,
                "saved_at": now,
            }
            self._access_token = session_values["access_token"]
            self._access_expires_at = now + session_values["access_lifetime"]
            self._last_error = ""
            try:
                self._persist_locked()
                self._session_persisted = True
            except OSError:
                self._session_persisted = False
            result = dict(self.status())
            registration = payload.get("device_registration")
            if self.device_registration:
                if not isinstance(registration, Mapping):
                    self._clear_locked(remove_file=True)
                    raise TenantAuthError("invalid_response")
                expected_token = self.device_registration["token"]
                if (
                    str(registration.get("uid_device") or "")
                    != self.device_registration["uid_device"]
                    or str(registration.get("token") or "") != expected_token
                    or type(registration.get("created")) is not bool
                    or type(registration.get("email_sent")) is not bool
                    or not isinstance(registration.get("email_recipient"), str)
                    or len(registration.get("email_recipient")) > 254
                ):
                    self._clear_locked(remove_file=True)
                    raise TenantAuthError("invalid_response")
                result["device_registration"] = {
                    "uid_device": self.device_registration["uid_device"],
                    "token": expected_token,
                    "created": registration["created"],
                    "email_sent": registration["email_sent"],
                    "email_recipient": registration["email_recipient"],
                }
            return result

    @staticmethod
    def _validated_device_registration(value):
        if value is None:
            return {}
        if not isinstance(value, Mapping):
            raise ValueError("Device registration must be a mapping")
        allowed = {
            "uid_device",
            "manufacturer_id",
            "token",
            "name",
            "device_type",
            "location",
            "model",
        }
        if set(value) - allowed:
            raise ValueError("Device registration contains unknown fields")
        required = ("uid_device", "manufacturer_id", "token")
        normalized = {key: str(item or "").strip() for key, item in value.items()}
        if (
            any(not _DEVICE_ID_RE.fullmatch(normalized.get(key, "")) for key in required[:2])
            or not 1 <= len(normalized.get("token", "")) <= 128
            or any(ord(character) < 33 or ord(character) > 126 for character in normalized.get("token", ""))
            or any(len(normalized.get(key, "")) > 100 for key in allowed - {"token"})
        ):
            raise ValueError("Device registration is invalid")
        return normalized

    def logout(self) -> Mapping[str, Any]:
        """Forget both in-memory and persisted refresh credentials."""

        with self._lock:
            # Never report a successful logout while a refresh credential is
            # still durable and could be restored after a process restart.
            path = self.state_dir / _SESSION_FILENAME
            try:
                if path.is_symlink():
                    path.unlink()
                else:
                    path.unlink(missing_ok=True)
            except OSError as exc:
                raise TenantAuthError("tenant_unavailable") from exc
            self._clear_locked(remove_file=False)
            return self.status()

    def access_token_for_url(self, target_url: str, force_refresh: bool = False) -> str:
        """Return a current token only for the session's exact HTTPS origin."""

        with self._lock:
            if not self._session:
                return ""
            base_url = str(self._session.get("base_url") or "")
            parsed_target = urlsplit(str(target_url or ""))
            target_origin = urlunsplit(
                (parsed_target.scheme, parsed_target.netloc, "", "", "")
            )
            normalized_target = normalize_tenant_url(
                target_origin,
                allow_insecure_localhost=self.allow_insecure_localhost,
                resolver=self.resolver,
            )
            normalized_base = normalize_tenant_url(
                base_url,
                allow_insecure_localhost=self.allow_insecure_localhost,
                resolver=self.resolver,
            )
            if normalized_target != normalized_base or _origin(target_url) != _origin(
                base_url
            ):
                raise TenantAuthError("tenant_endpoint_mismatch")
            now = self.wall_clock()
            if (
                not force_refresh
                and self._access_token
                and self._access_expires_at - now > ACCESS_REFRESH_LEEWAY_SECONDS
            ):
                return self._access_token
            return self._refresh_locked()

    def endpoint_url(self, configured_url: str) -> str:
        """Bind one commissioned endpoint path to the authenticated tenant."""

        parsed = urlsplit(str(configured_url or ""))
        try:
            port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80)
        except ValueError as exc:
            raise TenantAuthError("invalid_endpoint_url") from exc
        if (
            parsed.scheme.lower() != "https"
            or not parsed.hostname
            or port != 443
            or parsed.username is not None
            or parsed.password is not None
            or parsed.fragment
        ):
            raise TenantAuthError("invalid_endpoint_url")
        with self._lock:
            base_url = str(self._session.get("base_url") or "")
            if not base_url:
                return str(configured_url)
            base = urlsplit(base_url)
            return urlunsplit(
                (
                    base.scheme,
                    base.netloc,
                    parsed.path or "/",
                    parsed.query,
                    "",
                )
            )

    def status(self) -> Mapping[str, Any]:
        """Return one bounded status document containing no token or password."""

        with self._lock:
            now = self.wall_clock()
            user = self._session.get("user")
            if not isinstance(user, Mapping):
                user = {}
            if not self._session:
                state = "logged_out"
            elif float(self._session.get("refresh_expires_at") or 0.0) <= (
                now + REFRESH_EXPIRY_LEEWAY_SECONDS
            ):
                state = "expired"
            elif self._access_token and self._access_expires_at > now:
                state = "authenticated"
            else:
                state = "refresh_required"
            return {
                "schema_version": 1,
                "configured": True,
                "state": state,
                "tenant_url": str(self._session.get("base_url") or "")[:512],
                "tenant": str(self._session.get("tenant") or "")[:128],
                "username": str(user.get("username") or "")[:MAX_IDENTIFIER_LENGTH],
                "display_name": str(user.get("display_name") or "")[
                    :MAX_IDENTIFIER_LENGTH
                ],
                "access_expires_in": max(0, int(self._access_expires_at - now)),
                "refresh_expires_in": max(
                    0,
                    int(float(self._session.get("refresh_expires_at") or 0.0) - now),
                ),
                "session_persisted": bool(self._session_persisted),
                "last_error": self._last_error,
            }

    def _refresh_locked(self) -> str:
        now = self.wall_clock()
        refresh_expires_at = float(self._session.get("refresh_expires_at") or 0.0)
        if refresh_expires_at - now <= REFRESH_EXPIRY_LEEWAY_SECONDS:
            self._last_error = "session_expired"
            self._clear_locked(remove_file=True)
            raise TenantAuthError("session_expired")
        existing_refresh = _bounded_token(self._session.get("refresh_token"))
        base_url = str(self._session.get("base_url") or "")
        try:
            payload = self._request(
                base_url + REFRESH_PATH,
                {"refresh": existing_refresh},
                login=False,
            )
            values = _session_values(
                payload,
                existing_refresh=existing_refresh,
                existing_tenant=str(self._session.get("tenant") or ""),
            )
            user = _user_values(payload, self._session.get("user") or {})
            existing_user = self._session.get("user")
            if not isinstance(existing_user, Mapping):
                existing_user = {}
            if values["tenant"] != self._session.get("tenant") or user[
                "username"
            ] != existing_user.get("username"):
                raise TenantAuthError("invalid_response")
        except TenantAuthError as exc:
            self._last_error = exc.code
            if exc.code == "session_expired":
                self._clear_locked(remove_file=True)
            raise
        self._session.update(
            {
                "tenant": values["tenant"],
                "refresh_token": values["refresh_token"],
                "refresh_expires_at": now + values["refresh_lifetime"],
                "user": user,
                "saved_at": now,
            }
        )
        self._access_token = values["access_token"]
        self._access_expires_at = now + values["access_lifetime"]
        self._last_error = ""
        try:
            self._persist_locked()
            self._session_persisted = True
        except OSError:
            self._session_persisted = False
        return self._access_token

    def _request(
        self,
        url: str,
        payload: Mapping[str, Any],
        *,
        login: bool,
    ) -> Mapping[str, Any]:
        try:
            response = self.transport.request(
                "POST",
                url,
                headers={
                    "Accept": "application/json",
                    "Content-Type": "application/json; charset=utf-8",
                    "X-Requested-With": "XMLHttpRequest",
                    "User-Agent": "LabCobot-TenantAuth/1",
                },
                body=json_body(payload),
                timeout=self.timeout,
                max_response_bytes=MAX_AUTH_RESPONSE_BYTES,
                allow_redirects=False,
            )
        except TransportError as exc:
            raise TenantAuthError("tenant_unavailable") from exc
        if response.status_code == 401:
            raise TenantAuthError("invalid_credentials" if login else "session_expired")
        if (
            response.status_code in (408, 425, 429)
            or 500 <= response.status_code <= 599
        ):
            raise TenantAuthError("tenant_unavailable")
        if not 200 <= response.status_code <= 299:
            raise TenantAuthError("login_rejected" if login else "refresh_rejected")
        content_type = next(
            (
                str(value).lower()
                for key, value in response.headers.items()
                if str(key).lower() == "content-type"
            ),
            "",
        )
        if not content_type.startswith("application/json"):
            raise TenantAuthError("invalid_response")
        try:
            decoded = json.loads(
                response.body.decode("utf-8"),
                object_pairs_hook=_reject_duplicate_pairs,
                parse_constant=_reject_json_constant,
            )
        except (UnicodeError, ValueError, json.JSONDecodeError) as exc:
            raise TenantAuthError("invalid_response") from exc
        return _unwrap_envelope(decoded)

    def _restore(self) -> None:
        with self._lock:
            path = self.state_dir / _SESSION_FILENAME
            try:
                descriptor = os.open(
                    path,
                    os.O_RDONLY
                    | getattr(os, "O_CLOEXEC", 0)
                    | getattr(os, "O_NOFOLLOW", 0),
                )
                try:
                    metadata = os.fstat(descriptor)
                    if (
                        not stat.S_ISREG(metadata.st_mode)
                        or stat.S_IMODE(metadata.st_mode) & 0o077
                        or metadata.st_size > MAX_AUTH_RESPONSE_BYTES
                    ):
                        raise OSError("unsafe tenant session file")
                    raw = os.read(descriptor, MAX_AUTH_RESPONSE_BYTES + 1)
                finally:
                    os.close(descriptor)
                value = json.loads(
                    raw.decode("utf-8"),
                    object_pairs_hook=_reject_duplicate_pairs,
                    parse_constant=_reject_json_constant,
                )
                if (
                    not isinstance(value, dict)
                    or set(value)
                    != {
                        "schema_version",
                        "base_url",
                        "tenant",
                        "refresh_token",
                        "refresh_expires_at",
                        "user",
                        "saved_at",
                    }
                    or value.get("schema_version") != SESSION_SCHEMA_VERSION
                ):
                    raise ValueError("invalid tenant session")
                base_url = normalize_tenant_url(
                    value.get("base_url"),
                    allow_insecure_localhost=self.allow_insecure_localhost,
                    resolver=self.resolver,
                    # A temporary DNS outage during boot must not erase a
                    # valid durable session. Public-address validation occurs
                    # on the original interactive login.
                    verify_dns=False,
                )
                _bounded_token(value.get("refresh_token"))
                if not _TENANT_RE.fullmatch(str(value.get("tenant") or "")):
                    raise ValueError("invalid tenant session")
                value["user"] = _user_values({"user": value.get("user")}, {})
                expires_at = float(value.get("refresh_expires_at") or 0.0)
                saved_at = float(value.get("saved_at") or 0.0)
                if (
                    not math.isfinite(expires_at)
                    or not math.isfinite(saved_at)
                    or saved_at <= 0.0
                    or expires_at <= self.wall_clock() + REFRESH_EXPIRY_LEEWAY_SECONDS
                ):
                    raise ValueError("expired tenant session")
                value["base_url"] = base_url
                self._session = value
                self._session_persisted = True
            except (
                OSError,
                TenantAuthError,
                TypeError,
                ValueError,
                UnicodeError,
                json.JSONDecodeError,
            ):
                self._session = {}
                self._session_persisted = False

    def _persist_locked(self) -> None:
        self.state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
        directory_stat = self.state_dir.lstat()
        if (
            not stat.S_ISDIR(directory_stat.st_mode)
            or directory_stat.st_uid != os.geteuid()
            or stat.S_IMODE(directory_stat.st_mode) & 0o077
        ):
            raise OSError("tenant auth state directory is unsafe")
        payload = json.dumps(
            self._session,
            ensure_ascii=True,
            allow_nan=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("ascii")
        if len(payload) > MAX_AUTH_RESPONSE_BYTES:
            raise OSError("tenant session is too large")
        temporary = self.state_dir / ".session.tmp"
        flags = (
            os.O_WRONLY
            | os.O_CREAT
            | os.O_EXCL
            | getattr(os, "O_CLOEXEC", 0)
            | getattr(os, "O_NOFOLLOW", 0)
        )
        try:
            descriptor = os.open(temporary, flags, 0o600)
        except FileExistsError:
            temporary.unlink()
            descriptor = os.open(temporary, flags, 0o600)
        try:
            view = memoryview(payload)
            while view:
                written = os.write(descriptor, view)
                if written <= 0:
                    raise OSError("short tenant session write")
                view = view[written:]
            os.fsync(descriptor)
        finally:
            os.close(descriptor)
        os.replace(temporary, self.state_dir / _SESSION_FILENAME)
        directory_fd = os.open(
            self.state_dir,
            os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_DIRECTORY", 0),
        )
        try:
            os.fsync(directory_fd)
        finally:
            os.close(directory_fd)

    def _clear_locked(self, *, remove_file: bool) -> None:
        self._access_token = ""
        self._access_expires_at = 0.0
        self._session = {}
        self._session_persisted = False
        self._last_error = ""
        if not remove_file:
            return
        path = self.state_dir / _SESSION_FILENAME
        try:
            if path.is_symlink():
                path.unlink()
            else:
                path.unlink(missing_ok=True)
        except OSError:
            pass
