loyaly-catalogue/backend/app/infrastructure/security.py

"""
Password hashing, access-token issuance/verification, and the Principal that
represents an authenticated caller.

Two kinds of credential reach this module:

  * Interactive users. ``POST /api/auth/login`` exchanges a username/password
    for a short-lived signed JWT. No password is ever stored - only a PBKDF2
    digest, read from the environment (``AUTH_ADMIN_PASSWORD_HASH`` /
    ``AUTH_USER_PASSWORD_HASH``). Generate those with
    ``python scripts/make_auth_secrets.py``.

  * Machine consumers. A static key sent as ``X-API-Key``, mapped to a role by
    ``API_KEYS``. These do not expire, so treat one as a long-lived secret and
    give each consumer its own so it can be revoked individually.

PBKDF2-HMAC-SHA256 is used rather than bcrypt or argon2 deliberately: it is in
the standard library, so the slim Python image needs no compiled dependency,
and at the iteration count below it meets OWASP's current guidance. The
encoded form carries its own iteration count, so raising the constant later
does not invalidate hashes already issued.
"""
from __future__ import annotations

import base64
import hashlib
import hmac
import logging
import secrets
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple

import jwt

from app.infrastructure.settings import (
    API_KEYS,
    AUTH_ADMIN_PASSWORD_HASH,
    AUTH_ADMIN_USERNAME,
    AUTH_ALLOW_ANY_LOGIN,
    AUTH_ENABLED,
    AUTH_SECRET_KEY,
    AUTH_TOKEN_TTL_MINUTES,
    config_source,
)

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Roles and permissions
# ---------------------------------------------------------------------------
# These mirror the permission strings the React UI already keys its navigation
# off, so the server now enforces the same vocabulary the client was only
# displaying. `admin` is a superuser: has_permission() grants it everything
# rather than requiring every new permission to be added to this list.
ROLE_PERMISSIONS: Dict[str, List[str]] = {
    "admin": [
        "view_catalog",
        "view_project_details",
        "upload_train_test",
        "allocate_discounts",
        "manage_analytics",
        "manage_nutrition",
    ],
    "user": [
        "add_product",
        "upload_batch_products",
        "update_db_and_json",
        "fetch_images",
        "upload_store_inventory",
        "view_store_analytics",
        "view_nutrition_insights",
        "optimize_profits",
    ],
    # An outside API client that may send spreadsheets for catalog ingestion and
    # do NOTHING else. One permission, deliberately.
    #
    # This role exists because API keys carry no per-key scoping:
    # principal_for_api_key() derives permissions entirely from the role, so
    # "upload-only" can only be expressed as a role. Reusing `user` would have
    # been less code and would also have handed an outside contributor
    # add_product, upload_batch_products and upload_store_inventory - real
    # write access to the catalog - to solve a problem that needed one verb.
    #
    # WHAT A LEAKED UPLOADER KEY COSTS. Real CPU: this permission starts the
    # 11-stage pipeline, which is the point of the endpoint. The bound is not
    # "this role cannot work" but "all ingestion, from every source, shares one
    # worker" - batch_worker runs a single batch at a time behind a queue of
    # BATCH_QUEUE_MAX, past which POST /api/uploads/catalog answers 429. So a
    # key can occupy the ingestion worker; it cannot multiply it, and it cannot
    # touch the request path the healthcheck reads.
    #
    # What it still cannot do: read the catalog, read another caller's
    # submissions (every read on that router is filtered by submitted_by), or
    # cancel, resume or delete anything.
    "uploader": [
        "upload_catalog",
    ],
}

VALID_ROLES = frozenset(ROLE_PERMISSIONS)

JWT_ALGORITHM = "HS256"
JWT_ISSUER = "brand-catalog-rag"

# OWASP's floor for PBKDF2-HMAC-SHA256 at time of writing.
_PBKDF2_ITERATIONS = 600_000
_PBKDF2_PREFIX = "pbkdf2_sha256"


@dataclass(frozen=True)
class Principal:
    """Whoever is making the current request, once their credential checks out."""

    username: str
    role: str
    permissions: List[str] = field(default_factory=list)
    # "user"      - logged in via /api/auth/login, carrying a JWT
    # "api_key"   - a machine consumer from API_KEYS
    # "anonymous" - AUTH_ENABLED=false; no credential was checked at all
    kind: str = "user"

    def has_permission(self, permission: str) -> bool:
        return self.role == "admin" or permission in self.permissions


class AuthError(Exception):
    """A credential was absent, malformed, expired, or simply wrong."""


# ---------------------------------------------------------------------------
# Password hashing
# ---------------------------------------------------------------------------
def hash_password(password: str, *, iterations: int = _PBKDF2_ITERATIONS) -> str:
    """Return an encoded digest: ``pbkdf2_sha256$<iterations>$<salt>$<hash>``."""
    salt = secrets.token_bytes(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
    return "$".join(
        (
            _PBKDF2_PREFIX,
            str(iterations),
            base64.b64encode(salt).decode("ascii"),
            base64.b64encode(digest).decode("ascii"),
        )
    )


def _parse_encoded_hash(encoded: str) -> Optional[Tuple[bytes, bytes, int]]:
    """
    Split an encoded digest into ``(salt, digest, iterations)``, or None if it
    is not one.

    One parser, three callers. `verify_password` needs the parts, while
    `hash_is_wellformed` and `describe_password_hash` need only the verdict -
    and a login failing because the *configured* hash is corrupt is a different
    incident from a wrong password, so the two must agree on what "corrupt"
    means. Two copies of this parse would eventually disagree.

    Values arrive here straight from the environment, so a hash pasted into a
    deployment platform's form field as "pbkdf2_sha256$..." is unwrapped rather
    than rejected: the surrounding quotes are almost never intended as part of
    the secret, and the failure they cause otherwise is a silent 401.
    """
    if not encoded:
        return None
    encoded = encoded.strip().strip("'\"")
    try:
        prefix, raw_iterations, raw_salt, raw_digest = encoded.split("$")
        if prefix != _PBKDF2_PREFIX:
            return None
        # validate=True so junk is rejected rather than silently discarded:
        # b64decode's default drops non-alphabet characters, which would let a
        # subtly corrupted hash decode to the wrong bytes and fail as a "wrong
        # password" instead of as the configuration error it is.
        # binascii.Error subclasses ValueError, so it is caught below.
        salt = base64.b64decode(raw_salt, validate=True)
        digest = base64.b64decode(raw_digest, validate=True)
        iterations = int(raw_iterations)
    except (ValueError, TypeError):
        return None
    # A structurally valid string that decodes to nothing is still unusable,
    # and PBKDF2 rejects a non-positive iteration count by raising.
    if not salt or not digest or iterations < 1:
        return None
    return salt, digest, iterations


def hash_is_wellformed(encoded: str) -> bool:
    """Whether a configured digest can be checked against at all.

    Distinct from "does the password match": this asks whether the credential
    *store* is usable, which is a deployment fault rather than a sign-in one.
    """
    return _parse_encoded_hash(encoded) is not None


def password_hash_fingerprint(encoded: str) -> str:
    """
    A short, non-reversible identifier for a configured digest.

    Safe to log and to publish: it is a truncated SHA-256 of the *encoded
    digest*, and that digest already embeds a 16-byte random salt, so this says
    which credential is loaded without saying anything about the password
    behind it. It exists so a running deployment can be compared against the
    config it was supposed to have been built from - the failure this project
    actually hit - without moving a secret in order to do the comparison.
    """
    if not encoded:
        return ""
    return hashlib.sha256(encoded.strip().strip("'\"").encode("utf-8")).hexdigest()[:12]


def api_key_fingerprint(name: str, secret: str) -> str:
    """
    A short, non-reversible identifier for a configured API key.

    Same purpose as password_hash_fingerprint - say *which* credential is loaded
    without moving the credential - but the safety argument is different and
    worth stating. That function digests an encoded hash which already embeds a
    16-byte random salt. An API key has no salt, so the name is mixed in here to
    keep two consumers that were mistakenly issued the same secret from
    fingerprinting identically, and settings._parse_api_keys enforces a minimum
    secret length so the digest cannot be walked back with a wordlist.
    """
    if not secret:
        return ""
    cleaned = secret.strip().strip("'\"")
    material = f"{name}:{cleaned}"
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:12]


def describe_api_keys() -> List[Dict[str, object]]:
    """Every configured key as {name, role, fingerprint}, sorted by name.

    Sorted so two deployments' /api/health output can be diffed line for line;
    API_KEYS is keyed by secret, whose iteration order says nothing useful.
    """
    return sorted(
        (
            {"name": name, "role": role, "fingerprint": api_key_fingerprint(name, secret)}
            for secret, (name, role) in API_KEYS.items()
        ),
        key=lambda entry: entry["name"],
    )


def describe_password_hash(encoded: str) -> Dict[str, object]:
    """A loggable/publishable summary of a configured digest. Never its bytes."""
    parsed = _parse_encoded_hash(encoded)
    return {
        "valid": parsed is not None,
        "algorithm": _PBKDF2_PREFIX if parsed is not None else None,
        "iterations": parsed[2] if parsed is not None else None,
        "fingerprint": password_hash_fingerprint(encoded),
    }


def auth_config_summary() -> Dict[str, object]:
    """
    The effective authentication configuration, in a form safe to both log and
    publish. Contains no password and no hash - only the fingerprint.

    This is deliberately one function with two callers (the startup log in
    app/main.py and GET /api/health), because its entire purpose is letting two
    *deployments* be compared, and that only works if both report the same
    fields computed the same way.

    `*_source` is the field that earns this its keep. A value of "process-env"
    means the container's own environment supplied it and the .env file baked
    into the image was ignored - which is invisible from anywhere else, and is
    precisely how a corrected credential can keep failing after a redeploy.

    The same argument is why the API keys are summarised here. backend/Dockerfile
    copies .env.production in at BUILD time, so a key added to that file and then
    merely restarted is not present in the running process - and from outside,
    an undeployed key is indistinguishable from a wrong one, because both are
    just a 401. Publishing the names and fingerprints answers "is my key on this
    deployment?" without anyone having to send the secret to find out.
    """
    described = describe_password_hash(AUTH_ADMIN_PASSWORD_HASH)
    return {
        "enabled": AUTH_ENABLED,
        "allow_any_login": AUTH_ALLOW_ANY_LOGIN,
        "admin_username": AUTH_ADMIN_USERNAME,
        "password_hash_valid": bool(described["valid"]),
        "password_hash_iterations": described["iterations"],
        "password_hash_fingerprint": described["fingerprint"],
        "admin_username_source": config_source("AUTH_ADMIN_USERNAME"),
        "password_hash_source": config_source("AUTH_ADMIN_PASSWORD_HASH"),
        "api_keys_count": len(API_KEYS),
        "api_keys": describe_api_keys(),
        "api_keys_source": config_source("API_KEYS"),
    }


def verify_password(password: str, encoded: str) -> bool:
    """
    Check a password against an encoded digest.

    Returns False rather than raising on a malformed digest: a typo in
    AUTH_ADMIN_PASSWORD_HASH must fail the login, not 500 the endpoint and
    hand the caller a stack trace describing the credential store.
    """
    if not encoded:
        return False
    parsed = _parse_encoded_hash(encoded)
    if parsed is None:
        logger.error(
            "A configured password hash is malformed and cannot be used. Regenerate "
            "it with: python scripts/make_auth_secrets.py"
        )
        return False

    salt, digest, iterations = parsed
    candidate = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
    return hmac.compare_digest(candidate, digest)


# ---------------------------------------------------------------------------
# Access tokens
# ---------------------------------------------------------------------------
def create_access_token(
    username: str,
    role: str,
    permissions: List[str],
    *,
    ttl_minutes: Optional[int] = None,
) -> tuple[str, int]:
    """Issue a signed JWT. Returns ``(token, expires_in_seconds)``."""
    ttl = (ttl_minutes if ttl_minutes is not None else AUTH_TOKEN_TTL_MINUTES) * 60
    now = int(time.time())
    payload = {
        "sub": username,
        "role": role,
        "perms": permissions,
        "iss": JWT_ISSUER,
        "iat": now,
        "exp": now + ttl,
    }
    return jwt.encode(payload, AUTH_SECRET_KEY, algorithm=JWT_ALGORITHM), ttl


def decode_access_token(token: str) -> Principal:
    """
    Verify a JWT and return the Principal it names.

    The algorithm is pinned to a single-item allow-list rather than read from
    the token header. That is what closes the two classic JWT bypasses: a token
    presenting ``alg: none``, and one presenting ``alg: HS256`` against a key
    the server intended to use asymmetrically.
    """
    try:
        payload = jwt.decode(
            token,
            AUTH_SECRET_KEY,
            algorithms=[JWT_ALGORITHM],
            issuer=JWT_ISSUER,
            options={"require": ["exp", "iat", "sub"]},
        )
    except jwt.ExpiredSignatureError as exc:
        raise AuthError("Token has expired. Sign in again.") from exc
    except jwt.InvalidTokenError as exc:
        raise AuthError("Invalid authentication token.") from exc

    role = payload.get("role")
    if role not in VALID_ROLES:
        raise AuthError("Token names an unknown role.")

    perms = payload.get("perms")
    return Principal(
        username=str(payload["sub"]),
        role=role,
        # Fall back to the role's current grants if the token predates a
        # permission change, rather than trusting an arbitrary claim shape.
        permissions=list(perms) if isinstance(perms, list) else ROLE_PERMISSIONS.get(role, []),
        kind="user",
    )


# ---------------------------------------------------------------------------
# API keys (machine consumers)
# ---------------------------------------------------------------------------
def principal_for_api_key(presented: str) -> Principal:
    """
    Resolve an ``X-API-Key`` value to a Principal.

    Every configured key is compared even after a match, using compare_digest,
    so the time taken does not reveal how far down the list a near-miss got.
    """
    matched: Optional[tuple[str, str]] = None
    for secret, (name, role) in API_KEYS.items():
        if hmac.compare_digest(presented, secret):
            matched = (name, role)
    if matched is None:
        raise AuthError("Invalid API key.")

    name, role = matched
    return Principal(
        username=name,
        role=role,
        permissions=ROLE_PERMISSIONS.get(role, []),
        kind="api_key",
    )


def anonymous_principal() -> Principal:
    """
    The stand-in used when ``AUTH_ENABLED=false``.

    It is deliberately an admin: disabling auth is meant to make local
    development frictionless, and a half-privileged anonymous caller would
    produce confusing 403s instead. Nothing calls this when auth is on.
    """
    return Principal(
        username="anonymous",
        role="admin",
        permissions=ROLE_PERMISSIONS["admin"],
        kind="anonymous",
    )


if not AUTH_ENABLED:
    logger.warning(
        "AUTH_ENABLED=false: every endpoint is unauthenticated, including catalog "
        "generation, ML training, and the upload endpoints. This is for local "
        "development only - never run it on a host reachable from the internet."
    )