428 lines
16 KiB
Python
428 lines
16 KiB
Python
"""
|
|
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."
|
|
)
|