""" 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 import jwt from app.infrastructure.settings import ( API_KEYS, AUTH_ENABLED, AUTH_SECRET_KEY, AUTH_TOKEN_TTL_MINUTES, ) 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", ], } 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$$$``.""" 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 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 try: prefix, raw_iterations, raw_salt, raw_digest = encoded.split("$") if prefix != _PBKDF2_PREFIX: return False expected = base64.b64decode(raw_salt), base64.b64decode(raw_digest) salt, digest = expected iterations = int(raw_iterations) except (ValueError, TypeError): logger.error( "A configured password hash is malformed and cannot be used. Regenerate " "it with: python scripts/make_auth_secrets.py" ) return False 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." )