""" Tests for authentication and the endpoint guards. The behaviours asserted here are the ones the previous implementation got wrong, so each has a comment saying what it prevents rather than just what it checks. They need no database or Ollama: a request that is rejected at the guard never reaches a service. """ from __future__ import annotations import time import jwt import pytest from tests.conftest import TEST_ADMIN_PASSWORD, TEST_API_KEY, TEST_USER_PASSWORD # Every write/compute endpoint, with a request body valid enough that a 422 # would prove the guard let the request through to validation. WRITE_ENDPOINTS = [ ("/api/catalog/generate", {"json": {"brand": "X", "max_products": 1}}), ("/api/chat", {"json": {"query": "snacks"}}), ("/api/system/init", {}), ("/api/admin/store-intelligence/seed", {"json": {}}), ("/api/admin/store-intelligence/train", {"json": {}}), ("/api/admin/nutrition-intelligence/enrich", {"json": {}}), ("/api/admin/nutrition-intelligence/train", {"json": {}}), ("/api/admin/training/allocate-discounts", {"json": {}}), ("/api/admin/training/upload-dataset", {"files": {"file": ("a.csv", b"x")}}), ("/api/user/products/add", {"json": {}}), ("/api/user/products/batch-add", {"json": {}}), ("/api/user/products/upload-file", {"files": {"file": ("a.csv", b"x")}}), ("/api/upload/stores", {"files": {"file": ("a.csv", b"x")}}), ("/api/upload/stores/upload", {"files": {"file": ("a.csv", b"x")}}), ("/api/upload/analytics", {"files": {"file": ("a.csv", b"x")}}), ("/api/upload/analytics/upload", {"files": {"file": ("a.csv", b"x")}}), ("/api/upload/nutrition", {"files": {"file": ("a.csv", b"x")}}), ("/api/upload/nutrition/upload", {"files": {"file": ("a.csv", b"x")}}), ] ADMIN_ONLY_ENDPOINTS = [ ("/api/catalog/generate", {"brand": "X", "max_products": 1}), ("/api/system/init", None), ("/api/admin/store-intelligence/seed", {}), ("/api/admin/store-intelligence/train", {}), ("/api/admin/nutrition-intelligence/enrich", {}), ("/api/admin/nutrition-intelligence/train", {}), ] # --------------------------------------------------------------------------- # Guards # --------------------------------------------------------------------------- @pytest.mark.parametrize("path,kwargs", WRITE_ENDPOINTS) def test_write_endpoints_reject_anonymous_callers(client, path, kwargs): """ The whole point of this change. Every one of these was reachable by anyone who could resolve the hostname, including catalog generation, ML training, the upload endpoints, and unmetered Ollama access. """ resp = client.post(path, **kwargs) assert resp.status_code == 401, f"{path} answered {resp.status_code}, expected 401" @pytest.mark.parametrize("path,body", ADMIN_ONLY_ENDPOINTS) def test_admin_endpoints_reject_the_user_role(client, user_headers, path, body): """403, not 401: the caller is authenticated, just not allowed.""" resp = client.post(path, json=body, headers=user_headers) assert resp.status_code == 403, f"{path} answered {resp.status_code}, expected 403" def test_admin_passes_an_admin_only_endpoint(client, admin_headers): resp = client.post( "/api/catalog/generate", json={"brand": "TestBrand", "max_products": 1}, headers=admin_headers, ) assert resp.status_code == 202 def test_admin_is_a_superuser_over_user_role_permissions(client, admin_headers): """ `upload_store_inventory` is granted to the user role only, but admin must still pass - Principal.has_permission short-circuits on role == 'admin'. Not 403 is the assertion; the 400 that follows is the handler rejecting a bogus file, which proves the request got past the guard. """ resp = client.post( "/api/upload/stores", files={"file": ("a.txt", b"x")}, headers=admin_headers ) assert resp.status_code != 403 def test_user_reaches_its_own_permissioned_endpoint(client, user_headers): resp = client.post( "/api/upload/stores", files={"file": ("a.txt", b"x")}, headers=user_headers ) assert resp.status_code != 403 @pytest.mark.parametrize("path", ["/api/health", "/api/brands", "/api/auth/roles"]) def test_read_endpoints_stay_public(client, path): """Guarding writes must not have closed off the catalog the app browses.""" assert client.get(path).status_code == 200 # --------------------------------------------------------------------------- # Login # --------------------------------------------------------------------------- def test_login_succeeds_and_returns_a_token(client): resp = client.post( "/api/auth/login", json={"username": "admin", "password": TEST_ADMIN_PASSWORD} ) assert resp.status_code == 200 body = resp.json() assert body["token_type"] == "bearer" assert body["access_token"] assert body["expires_in"] > 0 assert body["user"]["role"] == "admin" def test_login_is_case_insensitive_on_username_only(client): """Usernames are normalised; passwords are not. The old version lowercased the password before comparing, which quietly shrank the keyspace.""" assert client.post( "/api/auth/login", json={"username": "ADMIN", "password": TEST_ADMIN_PASSWORD} ).status_code == 200 assert client.post( "/api/auth/login", json={"username": "admin", "password": TEST_ADMIN_PASSWORD.upper()} ).status_code == 401 def test_login_rejects_an_empty_password(client): """The old implementation treated an empty password as valid for any known username (`if pwd in passwords or pwd == ""`).""" resp = client.post("/api/auth/login", json={"username": "admin", "password": ""}) assert resp.status_code == 422 # min_length=1 on the schema def test_login_rejects_an_unknown_username(client): """The old fallback granted a profile to ANY username, and `admin` to any username that also asked for role='admin'.""" resp = client.post( "/api/auth/login", json={"username": "somebody-new", "password": "whatever"} ) assert resp.status_code == 401 def test_login_cannot_be_talked_into_a_role(client): """A `role` field in the body is not part of the schema and must not be honoured - the role comes from the account the password belongs to.""" resp = client.post( "/api/auth/login", json={"username": "user", "password": TEST_USER_PASSWORD, "role": "admin"}, ) assert resp.status_code == 200 assert resp.json()["user"]["role"] == "user" def test_failed_logins_are_throttled(client): """An exposed login endpoint must not be an unlimited password oracle.""" for _ in range(3): # AUTH_MAX_LOGIN_ATTEMPTS in conftest assert client.post( "/api/auth/login", json={"username": "admin", "password": "wrong"} ).status_code == 401 resp = client.post("/api/auth/login", json={"username": "admin", "password": "wrong"}) assert resp.status_code == 429 assert "Retry-After" in resp.headers # The lockout must also hold against the CORRECT password, or it is trivial # to sidestep by guessing until you land on it. assert client.post( "/api/auth/login", json={"username": "admin", "password": TEST_ADMIN_PASSWORD} ).status_code == 429 def test_roles_endpoint_no_longer_publishes_working_passwords(client): """It used to return demo_username/demo_password for both accounts.""" body = client.get("/api/auth/roles").json() assert "demo_password" not in client.get("/api/auth/roles").text assert {r["id"] for r in body["roles"]} == {"admin", "user"} # --------------------------------------------------------------------------- # Tokens # --------------------------------------------------------------------------- def test_me_returns_the_signed_in_profile(client, admin_headers): resp = client.get("/api/auth/me", headers=admin_headers) assert resp.status_code == 200 assert resp.json()["username"] == "admin" def test_me_requires_a_token(client): assert client.get("/api/auth/me").status_code == 401 def test_expired_token_is_rejected(client): from app.infrastructure.security import create_access_token token, _ = create_access_token("admin", "admin", [], ttl_minutes=-1) resp = client.get("/api/auth/me", headers={"Authorization": f"Bearer {token}"}) assert resp.status_code == 401 assert "expired" in resp.json()["detail"].lower() def test_unsigned_alg_none_token_is_rejected(client): """ The classic JWT bypass: present a token with `alg: none` and no signature. decode_access_token pins algorithms to ["HS256"] instead of trusting the header, which is what closes it. """ forged = jwt.encode( { "sub": "admin", "role": "admin", "perms": [], "iss": "brand-catalog-rag", "iat": int(time.time()), "exp": int(time.time()) + 3600, }, key="", algorithm="none", ) assert client.get( "/api/auth/me", headers={"Authorization": f"Bearer {forged}"} ).status_code == 401 def test_token_signed_with_the_wrong_key_is_rejected(client): forged = jwt.encode( { "sub": "admin", "role": "admin", "perms": [], "iss": "brand-catalog-rag", "iat": int(time.time()), "exp": int(time.time()) + 3600, }, # At least 32 bytes: PyJWT warns about shorter HMAC keys (RFC 7518 # ยง3.2), and a warning raised from a test asserting a rejection is # noise that hides real ones. key="a-wrong-key-that-is-long-enough-to-not-warn", algorithm="HS256", ) assert client.get( "/api/auth/me", headers={"Authorization": f"Bearer {forged}"} ).status_code == 401 def test_token_naming_an_unknown_role_is_rejected(client): """A validly signed token still cannot invent a role.""" from app.infrastructure.settings import AUTH_SECRET_KEY token = jwt.encode( { "sub": "admin", "role": "superuser", "perms": ["everything"], "iss": "brand-catalog-rag", "iat": int(time.time()), "exp": int(time.time()) + 3600, }, key=AUTH_SECRET_KEY, algorithm="HS256", ) assert client.get( "/api/auth/me", headers={"Authorization": f"Bearer {token}"} ).status_code == 401 def test_garbage_bearer_token_is_rejected(client): assert client.get( "/api/auth/me", headers={"Authorization": "Bearer not-even-a-jwt"} ).status_code == 401 # --------------------------------------------------------------------------- # API keys (machine consumers) # --------------------------------------------------------------------------- def test_valid_api_key_authenticates(client): resp = client.get("/api/auth/me", headers={"X-API-Key": TEST_API_KEY}) assert resp.status_code == 200 assert resp.json()["role"] == "user" def test_invalid_api_key_is_rejected(client): assert client.get( "/api/auth/me", headers={"X-API-Key": "wrong-key"} ).status_code == 401 def test_api_key_is_bound_to_its_configured_role(client): """The test key is a `user`, so admin-only endpoints must still refuse it.""" resp = client.post( "/api/catalog/generate", json={"brand": "X", "max_products": 1}, headers={"X-API-Key": TEST_API_KEY}, ) assert resp.status_code == 403 # --------------------------------------------------------------------------- # Password hashing # --------------------------------------------------------------------------- def test_password_round_trip(): from app.infrastructure.security import hash_password, verify_password encoded = hash_password("correct horse battery staple", iterations=1000) assert verify_password("correct horse battery staple", encoded) assert not verify_password("wrong", encoded) def test_hashes_are_salted(): """Two hashes of the same password must differ, or the digest leaks that two accounts share a password.""" from app.infrastructure.security import hash_password assert hash_password("same", iterations=1000) != hash_password("same", iterations=1000) @pytest.mark.parametrize("bad", ["", "not-a-hash", "pbkdf2_sha256$notanint$a$b", "a$b$c$d"]) def test_malformed_hash_fails_closed(bad): """A typo in AUTH_ADMIN_PASSWORD_HASH must fail the login, not 500 the endpoint and hand the caller a stack trace of the credential store.""" from app.infrastructure.security import verify_password assert verify_password("anything", bad) is False # --------------------------------------------------------------------------- # Why a sign-in failed # --------------------------------------------------------------------------- # The caller is told the same thing whatever went wrong - that is deliberate and # is pinned below. The operator is not: an account whose configured hash is # stale or corrupt needs a different repair from a mistyped password, and # collapsing the two is how a production sign-in outage stayed unexplained for a # day. These tests hold both halves at once: three reasons in the log, one # response on the wire. # # Throttle budget: conftest sets AUTH_MAX_LOGIN_ATTEMPTS=3 per (username, IP), # so each test below keeps `admin` to at most two attempts. Exceeding it turns a # 401 assertion into a 429 and reads like a code bug. import logging from app.api.routers import auth as auth_router from app.infrastructure.security import hash_is_wellformed _AUTH_LOGGER = "app.api.routers.auth" def test_an_unknown_username_is_logged_as_such(client, caplog): with caplog.at_level(logging.WARNING, logger=_AUTH_LOGGER): assert client.post( "/api/auth/login", json={"username": "nobody", "password": "whatever"} ).status_code == 401 assert "reason=unknown-username" in caplog.text # Names the setting to look at, since that is the actual repair. assert "AUTH_ADMIN_USERNAME" in caplog.text def test_a_wrong_password_is_logged_as_such(client, caplog): with caplog.at_level(logging.WARNING, logger=_AUTH_LOGGER): assert client.post( "/api/auth/login", json={"username": "admin", "password": "not-the-password"} ).status_code == 401 assert "reason=bad-password" in caplog.text def test_a_malformed_configured_hash_is_logged_as_an_error(client, caplog, monkeypatch): """Not a WARNING: no password can match an unparseable digest, so this is a broken deployment rather than a failed guess. `_accounts()` re-reads this module global on every call, which is what makes it patchable here.""" monkeypatch.setattr(auth_router, "AUTH_ADMIN_PASSWORD_HASH", "not-a-hash") with caplog.at_level(logging.WARNING, logger=_AUTH_LOGGER): assert client.post( "/api/auth/login", json={"username": "admin", "password": TEST_ADMIN_PASSWORD} ).status_code == 401 assert "reason=malformed-hash" in caplog.text assert any( r.levelno == logging.ERROR and "malformed-hash" in r.getMessage() for r in caplog.records ) def test_every_failure_reason_returns_an_identical_response(client, monkeypatch): """The log distinguishes them; the wire must not. If any of these three responses differed - by status, body, or headers - the endpoint would enumerate valid usernames and report its own misconfiguration to anyone.""" unknown = client.post( "/api/auth/login", json={"username": "nobody", "password": "x"} ) wrong = client.post( "/api/auth/login", json={"username": "admin", "password": "not-the-password"} ) monkeypatch.setattr(auth_router, "AUTH_ADMIN_PASSWORD_HASH", "not-a-hash") broken = client.post( "/api/auth/login", json={"username": "admin", "password": TEST_ADMIN_PASSWORD} ) responses = [unknown, wrong, broken] assert {r.status_code for r in responses} == {401} assert len({r.text for r in responses}) == 1 assert all(r.json() == {"detail": "Invalid username or password."} for r in responses) for r in responses: joined = r.text + " ".join(f"{k}:{v}" for k, v in r.headers.items()) for leak in ("unknown-username", "bad-password", "malformed-hash", "reason"): assert leak not in joined def test_the_failure_log_never_carries_the_hash_or_the_password(client, caplog): with caplog.at_level(logging.WARNING, logger=_AUTH_LOGGER): client.post( "/api/auth/login", json={"username": "admin", "password": "some-guessed-password"}, ) assert "some-guessed-password" not in caplog.text assert TEST_ADMIN_PASSWORD not in caplog.text assert "pbkdf2_sha256$" not in caplog.text def test_a_malformed_hash_still_costs_a_full_password_check(client, monkeypatch): """verify_password returns from an unparseable digest without doing any PBKDF2 work - measured at 0.16ms against 439ms for a real one. Left alone, an account with a corrupt hash would answer ~2700x faster than every other username and announce itself to anyone with a stopwatch, inverting the property _DUMMY_HASH exists to provide. So the work must still be paid.""" checked = [] real_verify = auth_router.verify_password def spy(password, encoded): checked.append(encoded) return real_verify(password, encoded) monkeypatch.setattr(auth_router, "AUTH_ADMIN_PASSWORD_HASH", "not-a-hash") monkeypatch.setattr(auth_router, "verify_password", spy) assert client.post( "/api/auth/login", json={"username": "admin", "password": TEST_ADMIN_PASSWORD} ).status_code == 401 assert len(checked) == 1, "exactly one verification per attempt" assert hash_is_wellformed(checked[0]), "the broken hash must not short-circuit it"