gateway_auth.py

"""Public-safe account orchestration and bounded login throttling."""

from __future__ import annotations

import threading
import time
from dataclasses import dataclass
from typing import Any, Callable

import auth_storage


INVALID_LOGIN_MESSAGE = "Invalid username or password."
THROTTLED_LOGIN_MESSAGE = "Too many login attempts. Try again later."
THROTTLED_SIGNUP_MESSAGE = "Too many account requests. Try again later."
_SIGNUP_THROTTLE_USERNAME = "__account_signup__"


@dataclass(frozen=True)
class AuthResult:
    ok: bool
    message: str
    session_token: str = ""
    username: str = ""
    account_status: str = ""


@dataclass
class _AttemptState:
    failures: list[float]
    locked_until: float = 0.0
    updated_at: float = 0.0


class LoginThrottle:
    """Process-local brute-force limiter keyed by account and requester."""

    def __init__(
        self,
        *,
        max_failures: int = 5,
        window_seconds: float = 300,
        lock_seconds: float = 600,
        max_entries: int = 10_000,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        self.max_failures = max(1, int(max_failures))
        self.window_seconds = max(1.0, float(window_seconds))
        self.lock_seconds = max(1.0, float(lock_seconds))
        self.max_entries = max(100, int(max_entries))
        self._clock = clock
        self._states: dict[tuple[str, str], _AttemptState] = {}
        self._lock = threading.RLock()

    @staticmethod
    def _key(username: str, requester: str) -> tuple[str, str]:
        return (
            auth_storage.normalize_username(username).lower(),
            str(requester or "unknown").strip() or "unknown",
        )

    def _prune_state(self, state: _AttemptState, now: float) -> None:
        cutoff = now - self.window_seconds
        state.failures[:] = [stamp for stamp in state.failures if stamp >= cutoff]
        if state.locked_until and now >= state.locked_until:
            state.locked_until = 0.0
            state.failures.clear()
        state.updated_at = now

    def _bound_entries(self) -> None:
        excess = len(self._states) - self.max_entries
        if excess <= 0:
            return
        oldest = sorted(self._states.items(), key=lambda item: item[1].updated_at)
        for key, _state in oldest[:excess]:
            self._states.pop(key, None)

    def allowed(self, username: str, requester: str) -> bool:
        now = self._clock()
        key = self._key(username, requester)
        with self._lock:
            state = self._states.get(key)
            if state is None:
                return True
            self._prune_state(state, now)
            if not state.failures and not state.locked_until:
                self._states.pop(key, None)
                return True
            return state.locked_until <= now

    def record_failure(self, username: str, requester: str) -> None:
        now = self._clock()
        key = self._key(username, requester)
        with self._lock:
            state = self._states.setdefault(key, _AttemptState([]))
            self._prune_state(state, now)
            state.failures.append(now)
            if len(state.failures) >= self.max_failures:
                state.locked_until = now + self.lock_seconds
            state.updated_at = now
            self._bound_entries()

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

    def failure_count(self, username: str, requester: str) -> int:
        now = self._clock()
        with self._lock:
            state = self._states.get(self._key(username, requester))
            if state is None:
                return 0
            self._prune_state(state, now)
            return len(state.failures)


DEFAULT_LOGIN_THROTTLE = LoginThrottle()
DEFAULT_SIGNUP_THROTTLE = LoginThrottle(
    max_failures=5,
    window_seconds=3600,
    lock_seconds=3600,
)


def requester_key(request: Any | None) -> str:
    """Use the server-observed peer address, never a spoofable forwarded header."""

    client = getattr(request, "client", None)
    host = str(getattr(client, "host", "") or "").strip()
    return host or "unknown"


def login(
    username: str,
    password: str,
    requester: str,
    *,
    throttle: LoginThrottle = DEFAULT_LOGIN_THROTTLE,
) -> AuthResult:
    normalized = auth_storage.normalize_username(username)
    if not throttle.allowed(normalized, requester):
        return AuthResult(False, THROTTLED_LOGIN_MESSAGE)
    status = auth_storage.account_authentication_status(normalized, password)
    if status == "invalid":
        throttle.record_failure(normalized, requester)
        return AuthResult(False, INVALID_LOGIN_MESSAGE)
    throttle.record_success(normalized, requester)
    if status == auth_storage.ACCOUNT_STATUS_PENDING:
        return AuthResult(
            False,
            "Administrators have not approved this account yet.",
            username=normalized,
            account_status=status,
        )
    if status == auth_storage.ACCOUNT_STATUS_DENIED:
        return AuthResult(
            False,
            "Your request for account creation was denied by administrators. "
            "Choose whether to re-send or delete the request below.",
            username=normalized,
            account_status=status,
        )
    return AuthResult(
        True,
        f"Signed in as {normalized}.",
        session_token=auth_storage.issue_session_token(normalized),
        username=normalized,
        account_status=auth_storage.ACCOUNT_STATUS_APPROVED,
    )


def request_signup(
    username: str,
    password: str,
    confirm_password: str,
    requester: str,
    *,
    throttle: LoginThrottle = DEFAULT_SIGNUP_THROTTLE,
) -> AuthResult:
    normalized = auth_storage.normalize_username(username)
    if not auth_storage.signups_allowed():
        return AuthResult(False, "Account creation is disabled for this deployment.")
    if not throttle.allowed(_SIGNUP_THROTTLE_USERNAME, requester):
        return AuthResult(False, THROTTLED_SIGNUP_MESSAGE)
    if password != confirm_password:
        return AuthResult(False, "Passwords do not match.")
    try:
        user = auth_storage.request_account_creation(normalized, password)
    except ValueError as exc:
        throttle.record_failure(_SIGNUP_THROTTLE_USERNAME, requester)
        return AuthResult(False, f"Could not submit account request: {exc}")
    except Exception:
        throttle.record_failure(_SIGNUP_THROTTLE_USERNAME, requester)
        return AuthResult(
            False,
            "Could not submit the account request. Try again later.",
        )
    throttle.record_failure(_SIGNUP_THROTTLE_USERNAME, requester)
    return AuthResult(
        False,
        f"Request for account creation for {user['username']} has been sent to administrators.",
        username=str(user["username"]),
        account_status=auth_storage.ACCOUNT_STATUS_PENDING,
    )


def handle_denied_request(
    username: str,
    password: str,
    choice: str,
) -> AuthResult:
    normalized = auth_storage.normalize_username(username)
    try:
        if choice == "Re-send request":
            auth_storage.resubmit_denied_account(normalized, password)
            return AuthResult(
                False,
                "Your account creation request has been re-sent to administrators.",
                username=normalized,
                account_status=auth_storage.ACCOUNT_STATUS_PENDING,
            )
        if choice == "Do not re-send; delete my account request":
            auth_storage.delete_denied_account(normalized, password)
            return AuthResult(
                False,
                "Your denied account request and all related database information "
                "have been deleted.",
            )
        return AuthResult(
            False,
            "Choose whether to re-send or delete the account request.",
            username=normalized,
            account_status=auth_storage.ACCOUNT_STATUS_DENIED,
        )
    except (PermissionError, ValueError):
        return AuthResult(
            False,
            "Could not process the denied account request. Re-enter the password "
            "used to sign up and try again.",
            username=normalized,
            account_status=auth_storage.ACCOUNT_STATUS_DENIED,
        )


def change_password(
    username: str,
    old_password: str,
    new_password: str,
    confirm_password: str,
) -> AuthResult:
    normalized = auth_storage.normalize_username(username)
    if not normalized or not old_password or not new_password or not confirm_password:
        return AuthResult(False, "Enter your username, current password, and the new password twice.")
    if new_password != confirm_password:
        return AuthResult(False, "New passwords do not match.")
    if old_password == new_password:
        return AuthResult(False, "Choose a new password that differs from your current password.")
    try:
        auth_storage.update_password(normalized, old_password, new_password)
    except PermissionError:
        return AuthResult(
            False,
            "Could not update password: Invalid username or current password.",
        )
    except ValueError as exc:
        return AuthResult(False, f"Could not update password: {exc}")
    return AuthResult(
        True,
        "Password updated. Sign in again with your new password.",
        username=normalized,
    )


__all__ = [
    "AuthResult",
    "LoginThrottle",
    "DEFAULT_LOGIN_THROTTLE",
    "DEFAULT_SIGNUP_THROTTLE",
    "requester_key",
    "login",
    "request_signup",
    "handle_denied_request",
    "change_password",
]