loyaly-catalogue/backend/app/api/routers/auth.py

"""
Authentication router - issues and inspects access tokens.

This replaces an earlier version that returned a role profile without issuing
anything, accepted an empty password, and granted `admin` to any username that
asked for the role. It decided which buttons the UI drew; it protected nothing.
Now the token this returns is the credential every write endpoint checks (see
app/api/deps.py), so the rules hold for curl and partner scripts too, not just
for the React app.

Accounts come from the environment - two of them, admin and user, configured as
PBKDF2 digests. That is deliberately not a user database: this project has no
user table, no registration flow and no password reset, and inventing one here
would be a bigger change than the problem calls for. Machine consumers get
API_KEYS instead. If per-user accounts become a real requirement, this module
is the seam to replace.

For local work there is AUTH_ALLOW_ANY_LOGIN, which skips the password check
here and nowhere else - the token still gets signed and every guard downstream
still checks it. It is off by default and logs a warning at startup when on.
"""
from __future__ import annotations

import logging
import threading
import time
from typing import Dict, List, Tuple

from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field

from app.api.deps import get_principal
from app.infrastructure.security import (
    ROLE_PERMISSIONS,
    Principal,
    create_access_token,
    hash_is_wellformed,
    password_hash_fingerprint,
    verify_password,
)
from app.infrastructure.settings import (
    AUTH_ADMIN_PASSWORD_HASH,
    AUTH_ADMIN_USERNAME,
    AUTH_ALLOW_ANY_LOGIN,
    AUTH_ENABLED,
    AUTH_LOCKOUT_SECONDS,
    AUTH_MAX_LOGIN_ATTEMPTS,
    AUTH_USER_PASSWORD_HASH,
    AUTH_USER_USERNAME,
    config_source,
)

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/auth", tags=["auth"])

if AUTH_ENABLED and AUTH_ALLOW_ANY_LOGIN:
    logger.warning(
        "AUTH_ALLOW_ANY_LOGIN=true: /api/auth/login accepts ANY password, so anyone "
        "who can reach this port can sign in as admin. Local development only - "
        "set it to false in backend/.env before exposing this server."
    )


class LoginRequest(BaseModel):
    username: str = Field(min_length=1, max_length=150)
    password: str = Field(min_length=1, max_length=1024)


class UserProfile(BaseModel):
    username: str
    role: str
    display_name: str
    email: str
    permissions: List[str] = Field(default_factory=list)


class LoginResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"
    expires_in: int = Field(description="Token lifetime in seconds")
    user: UserProfile


# A syntactically valid hash of an unguessable value. Never matches any real
# password; it exists only so the unknown-username path in login() does the
# same PBKDF2 work as the known one, keeping the two indistinguishable by timing.
_DUMMY_HASH = (
    "pbkdf2_sha256$600000$YWJjZGVmZ2hpamtsbW5vcA==$"
    "S1cVFrGD4pDkGqSjbEbaVSTONzGhCT9BOaWPQ2vwvvA="
)


def _accounts() -> Dict[str, dict]:
    """
    The configured accounts, read per call so a settings reload is picked up.

    Usernames are compared case-insensitively (matching what the login form
    sends), but the password is not touched - the previous version lowercased
    it before comparing, which silently shrank the effective keyspace.

    An account with a blank password hash is omitted entirely rather than
    included with an unmatchable digest. Both spellings deny the login, but
    only omission keeps it out of the account table, so nothing downstream can
    treat it as a real account. This is how the optional `user` account is
    switched off: leave AUTH_USER_PASSWORD_HASH unset and only `admin` exists.
    """
    accounts = {
        AUTH_ADMIN_USERNAME.lower(): {
            "password_hash": AUTH_ADMIN_PASSWORD_HASH,
            "role": "admin",
            "display_name": "System Administrator",
            "email": "admin@nutritionintel.com",
        },
    }

    if AUTH_USER_PASSWORD_HASH:
        accounts[AUTH_USER_USERNAME.lower()] = {
            "password_hash": AUTH_USER_PASSWORD_HASH,
            "role": "user",
            "display_name": "Product & Store Manager",
            "email": "user@nutritionintel.com",
        }

    return accounts


# ---------------------------------------------------------------------------
# Failed-login throttle
# ---------------------------------------------------------------------------
# In-process and per-worker: with several uvicorn workers a determined attacker
# gets AUTH_MAX_LOGIN_ATTEMPTS per worker, not overall. That is a real limit,
# not a rounding error - but it still turns an unbounded password oracle into a
# rate-limited one without adding Redis to the deployment. Move this to a shared
# store if you ever run many workers.
_failures: Dict[Tuple[str, str], Tuple[int, float]] = {}
_failures_lock = threading.Lock()


def _throttle_key(username: str, request: Request) -> Tuple[str, str]:
    # request.client.host is the real client IP because uvicorn runs with
    # --proxy-headers behind nginx/Caddy (see backend/Dockerfile); without that
    # every request would appear to come from the proxy and share one bucket.
    client = request.client.host if request.client else "unknown"
    return (username, client)


def _check_not_locked(key: Tuple[str, str]) -> None:
    with _failures_lock:
        entry = _failures.get(key)
        if entry is None:
            return
        count, first_seen = entry
        if time.time() - first_seen > AUTH_LOCKOUT_SECONDS:
            del _failures[key]
            return
        if count >= AUTH_MAX_LOGIN_ATTEMPTS:
            retry_after = int(AUTH_LOCKOUT_SECONDS - (time.time() - first_seen))
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail=f"Too many failed sign-in attempts. Try again in {retry_after}s.",
                headers={"Retry-After": str(max(retry_after, 1))},
            )


def _record_failure(key: Tuple[str, str]) -> None:
    now = time.time()
    with _failures_lock:
        count, first_seen = _failures.get(key, (0, now))
        if now - first_seen > AUTH_LOCKOUT_SECONDS:
            count, first_seen = 0, now
        _failures[key] = (count + 1, first_seen)


def _clear_failures(key: Tuple[str, str]) -> None:
    with _failures_lock:
        _failures.pop(key, None)


# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------
@router.post("/login", response_model=LoginResponse)
def login(payload: LoginRequest, request: Request) -> LoginResponse:
    """Exchange a username and password for an access token."""
    if not AUTH_ENABLED:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=(
                "Authentication is disabled on this server (AUTH_ENABLED=false), so no "
                "token can be issued. Every endpoint is open; sign-in is not required."
            ),
        )

    username = payload.username.strip().lower()
    key = _throttle_key(username, request)

    if AUTH_ALLOW_ANY_LOGIN:
        # Dev bypass: any password gets in. The username still picks the
        # account, so `admin` lands on the admin pages and `user` on the user
        # ones; anything else is an unconfigured name and gets the lower of the
        # two roles rather than silently minting an admin. Throttling is skipped
        # because there is no longer a password to guess.
        account = _accounts().get(username) or {
            "role": "user",
            "display_name": payload.username.strip() or username,
            "email": f"{username}@nutritionintel.com",
        }
        logger.warning(
            "AUTH_ALLOW_ANY_LOGIN: signing in %r as %s without checking the password",
            username,
            account["role"],
        )
    else:
        _check_not_locked(key)

        account = _accounts().get(username)

        # Verify against a dummy hash when the username is unknown so a bad
        # username and a bad password take the same time. Otherwise the response
        # latency alone enumerates valid usernames.
        stored_hash = account["password_hash"] if account else _DUMMY_HASH

        # A hash that does not parse can never match, and verify_password bails
        # out of one before doing any PBKDF2 work - measured here, 0.16ms against
        # 439ms for a real digest. That inverts the very property _DUMMY_HASH
        # exists to protect: an account whose configured hash is corrupt would
        # answer ~2700x faster than every other username, announcing which
        # account is broken to anyone with a stopwatch. So spend the same work
        # regardless; the result is a rejection either way.
        hash_usable = hash_is_wellformed(stored_hash)
        password_ok = verify_password(
            payload.password, stored_hash if hash_usable else _DUMMY_HASH
        )

        if account is None or not password_ok:
            _record_failure(key)
            # The reason goes to the LOG, never to the caller - the response
            # below is byte-identical whichever of these it was, so nothing here
            # can be used to enumerate usernames. It is computed after both the
            # lookup and the PBKDF2 call above, so it adds no timing signal
            # either. Without it, a deployment whose configured hash or admin
            # username has drifted is indistinguishable from someone simply
            # typing the wrong password, and this is exactly how a production
            # sign-in outage stayed unexplained: the log said "Failed sign-in
            # for 'admin'" and nothing more.
            if account is None:
                logger.warning(
                    "Failed sign-in for %r from %s: reason=unknown-username. "
                    "Configured accounts: %s (AUTH_ADMIN_USERNAME source=%s).",
                    username,
                    key[1],
                    ", ".join(sorted(_accounts())),
                    config_source("AUTH_ADMIN_USERNAME"),
                )
            elif not hash_usable:
                # ERROR, not WARNING: this is a broken deployment, not a bad
                # guess. No password can ever match, so every sign-in to this
                # account will 401 until the hash itself is replaced.
                logger.error(
                    "Failed sign-in for %r from %s: reason=malformed-hash. The configured "
                    "password hash does not parse as pbkdf2_sha256$<iterations>$<b64 salt>$"
                    "<b64 digest> (fingerprint=%s, source=%s). Nobody can sign in to this "
                    "account until it is regenerated with scripts/make_auth_secrets.py.",
                    username,
                    key[1],
                    password_hash_fingerprint(stored_hash) or "(empty)",
                    config_source("AUTH_ADMIN_PASSWORD_HASH"),
                )
            else:
                logger.warning(
                    "Failed sign-in for %r from %s: reason=bad-password. The account exists "
                    "and its hash parses (fingerprint=%s, source=%s); the password did not "
                    "match. If this IS the password you deployed, then the running config "
                    "carries a different hash than the file you are reading - compare that "
                    "fingerprint against: python scripts/make_auth_secrets.py "
                    "--fingerprint .env.production",
                    username,
                    key[1],
                    password_hash_fingerprint(stored_hash),
                    config_source("AUTH_ADMIN_PASSWORD_HASH"),
                )
            # One message for every failure mode, for the same reason.
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid username or password.",
            )

    _clear_failures(key)
    role = account["role"]
    permissions = ROLE_PERMISSIONS.get(role, [])
    token, expires_in = create_access_token(username, role, permissions)
    logger.info("Issued token for %r (role=%s)", username, role)

    return LoginResponse(
        access_token=token,
        expires_in=expires_in,
        user=UserProfile(
            username=username,
            role=role,
            display_name=account["display_name"],
            email=account["email"],
            permissions=permissions,
        ),
    )


@router.get("/me", response_model=UserProfile)
def me(principal: Principal = Depends(get_principal)) -> UserProfile:
    """
    Who the presented credential belongs to. 401 if it is missing or expired.

    The frontend calls this on boot to check a restored session before showing
    the app, so an expired token lands on the login page rather than on a
    dashboard whose every request then fails.
    """
    account = _accounts().get(principal.username, {})
    return UserProfile(
        username=principal.username,
        role=principal.role,
        display_name=account.get("display_name", principal.username.title()),
        email=account.get("email", f"{principal.username}@nutritionintel.com"),
        permissions=principal.permissions,
    )


@router.get("/roles")
def list_roles() -> dict:
    """
    The available roles and what each may do.

    Note there are no demo credentials here any more. The passwords are set per
    deployment via AUTH_ADMIN_PASSWORD_HASH / AUTH_USER_PASSWORD_HASH; this
    endpoint used to publish working ones to anyone who asked.
    """
    return {
        "roles": [
            {
                "id": "admin",
                "name": "Admin",
                "description": (
                    "Full access: catalog brand cards, project details, Excel/CSV "
                    "train/test uploads, discount allocation, analytics and nutrition. "
                    "Implicitly holds every permission."
                ),
                "permissions": ROLE_PERMISSIONS["admin"],
            },
            {
                "id": "user",
                "name": "User",
                "description": (
                    "Combined user and store role: single or batch product uploads with "
                    "image and DB/JSON sync, store inventory, profit analytics, nutrition."
                ),
                "permissions": ROLE_PERMISSIONS["user"],
            },
        ]
    }