329 lines
12 KiB
Python
329 lines
12 KiB
Python
"""
|
|
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
|