""" 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