loyaly-catalogue/backend/tests/test_auth.py

"""
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/elec/admin/runs", {"json": {"category": "mobiles"}}),
    ("/api/elec/admin/review/1", {"json": {"approve": True}}),
    ("/api/elec/admin/sites/croma.com/probe", {}),
]

ADMIN_ONLY_ENDPOINTS = [
    ("/api/elec/admin/runs", {"category": "mobiles"}),
    ("/api/elec/admin/review/1", {"approve": True}),
    ("/api/elec/admin/sites/croma.com/probe", None),
]


# ---------------------------------------------------------------------------
# Guards
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("path,kwargs", WRITE_ENDPOINTS)
def test_write_endpoints_reject_anonymous_callers(client, path, kwargs):
    """Starting a crawl or probing a site must never be open to anyone who can
    reach the port."""
    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):
    """400 (a brand outside the allow-list) proves the guard let admin through."""
    resp = client.post("/api/elec/admin/runs", json={"category": "mobiles", "brands": ["nokia"]},
                       headers=admin_headers)
    assert resp.status_code == 400


@pytest.mark.parametrize("path", ["/api/health", "/api/auth/roles", "/openapi.json"])
def test_read_endpoints_stay_public(client, path):
    """Guarding writes must not have closed off what 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/elec/admin/runs",
        json={"category": "mobiles"},
        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"