wyndham-ARR/arr_web/auth.py

"""Dependency-free login credentials and bounded attempt tracking."""

from __future__ import annotations

import math
import os
import secrets
import threading
import time
from dataclasses import dataclass, field
from typing import Callable, Dict, Mapping, Optional


USERNAME_ENV = "ARR_WEB_USERNAME"
PASSWORD_ENV = "ARR_WEB_PASSWORD"
MIN_PASSWORD_LENGTH = 12
MAX_USERNAME_LENGTH = 128
MAX_PASSWORD_LENGTH = 1024


@dataclass(frozen=True)
class LoginCredentials:
    """One environment-owned operator identity for the current portal deployment."""

    username: str
    password: str

    def __post_init__(self) -> None:
        if (
            not self.username
            or self.username != self.username.strip()
            or len(self.username) > MAX_USERNAME_LENGTH
        ):
            raise ValueError("ARR Web login username is invalid")
        if not MIN_PASSWORD_LENGTH <= len(self.password) <= MAX_PASSWORD_LENGTH:
            raise ValueError(
                f"ARR Web login password must be {MIN_PASSWORD_LENGTH}–"
                f"{MAX_PASSWORD_LENGTH} characters"
            )

    @classmethod
    def from_environment(
        cls,
        environ: Optional[Mapping[str, str]] = None,
    ) -> "LoginCredentials":
        source = os.environ if environ is None else environ
        username = source.get(USERNAME_ENV, "")
        password = source.get(PASSWORD_ENV, "")
        if not username or not password:
            raise ValueError(
                f"ARR Web login requires {USERNAME_ENV} and {PASSWORD_ENV}"
            )
        return cls(username=username, password=password)

    def verify(self, username: object, password: object) -> bool:
        if not isinstance(username, str) or not isinstance(password, str):
            return False
        if (
            len(username) > MAX_USERNAME_LENGTH
            or len(password) > MAX_PASSWORD_LENGTH
        ):
            return False
        username_matches = secrets.compare_digest(
            username.encode("utf-8"),
            self.username.encode("utf-8"),
        )
        password_matches = secrets.compare_digest(
            password.encode("utf-8"),
            self.password.encode("utf-8"),
        )
        return username_matches and password_matches


@dataclass
class _AttemptState:
    failures: list[float] = field(default_factory=list)
    blocked_until: float = 0.0


class LoginAttemptLedger:
    """Small per-client sliding-window limiter for the single-process Web runtime."""

    def __init__(
        self,
        *,
        max_failures: int = 5,
        window_seconds: int = 5 * 60,
        block_seconds: int = 60,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        if max_failures < 1 or window_seconds < 1 or block_seconds < 1:
            raise ValueError("login attempt limits must be positive")
        self._max_failures = max_failures
        self._window = float(window_seconds)
        self._block = float(block_seconds)
        self._clock = clock
        self._lock = threading.Lock()
        self._states: Dict[str, _AttemptState] = {}

    def retry_after(self, client_id: str) -> int:
        now = self._clock()
        key = self._key(client_id)
        with self._lock:
            state = self._states.get(key)
            if state is None:
                return 0
            self._prune(state, now)
            if state.blocked_until > now:
                return max(1, math.ceil(state.blocked_until - now))
            if not state.failures:
                self._states.pop(key, None)
            return 0

    def record_failure(self, client_id: str) -> int:
        now = self._clock()
        key = self._key(client_id)
        with self._lock:
            state = self._states.setdefault(key, _AttemptState())
            self._prune(state, now)
            if state.blocked_until > now:
                return max(1, math.ceil(state.blocked_until - now))
            state.failures.append(now)
            if len(state.failures) >= self._max_failures:
                state.failures.clear()
                state.blocked_until = now + self._block
                return max(1, math.ceil(self._block))
            self._trim(now)
            return 0

    def record_success(self, client_id: str) -> None:
        with self._lock:
            self._states.pop(self._key(client_id), None)

    def _prune(self, state: _AttemptState, now: float) -> None:
        state.failures[:] = [
            failure for failure in state.failures if failure > now - self._window
        ]
        if state.blocked_until <= now:
            state.blocked_until = 0.0

    def _trim(self, now: float) -> None:
        if len(self._states) <= 4096:
            return
        stale = []
        for key, state in self._states.items():
            self._prune(state, now)
            if not state.failures and state.blocked_until <= now:
                stale.append(key)
        for key in stale:
            self._states.pop(key, None)
        if len(self._states) > 4096:
            for key in list(self._states)[: len(self._states) - 4096]:
                self._states.pop(key, None)

    @staticmethod
    def _key(client_id: str) -> str:
        return (client_id or "unknown")[:128]