praxis/tests/test_auth.py

"""Auth unit tests (TASK-03-06) — mocked PgStore, no real Postgres.

Covers: password hash/verify/rehash, cookie config (secure flag, missing
secret), rate limiter threshold, current_operator dependency (401/503
cases, active/inactive), login/logout/me route handlers.
"""

from __future__ import annotations

import os
import types
from unittest.mock import AsyncMock, MagicMock

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from starlette.middleware.sessions import SessionMiddleware

from server.auth.cookies import get_session_middleware_kwargs
from server.auth.dependencies import current_operator
from server.auth.models import Operator
from server.auth.passwords import hash_password, needs_rehash, verify_password
from server.auth.rate_limit import limiter, rate_limit_login, reset_login_rate_limit
from server.auth.routes import router


@pytest.fixture(autouse=True)
def _reset_limiter():
    reset_login_rate_limit()
    yield
    reset_login_rate_limit()


# ── Passwords ───────────────────────────────────────────────────────────────


def test_password_hash_verify_roundtrip():
    h = hash_password("correct horse battery staple")
    assert h.startswith("$argon2id$")
    assert verify_password(h, "correct horse battery staple") is True


def test_password_verify_wrong_returns_false():
    h = hash_password("secret-1")
    assert verify_password(h, "secret-2") is False
    # no exception raised — uniform 401 path
    assert verify_password(h, "") is False


def test_needs_rehash_false_for_current_defaults():
    h = hash_password("x")
    assert needs_rehash(h) is False


def test_needs_rehash_true_for_weak_hash():
    # A hash produced with weaker params triggers rehash.
    from argon2 import PasswordHasher
    weak = PasswordHasher(time_cost=1, memory_cost=8, parallelism=1).hash("x")
    assert needs_rehash(weak) is True


# ── Cookie config ───────────────────────────────────────────────────────────


def test_cookie_kwargs_defaults(monkeypatch):
    monkeypatch.setenv("PRAXIS_COOKIE_SECRET", "x" * 48)
    monkeypatch.setenv("PRAXIS_COOKIE_SECURE", "true")
    kw = get_session_middleware_kwargs()
    assert kw["session_cookie"] == "praxis_op"
    assert kw["max_age"] == 28800
    # Starlette SessionMiddleware: https_only (not secure), same_site (not samesite),
    # httponly is always True (no kwarg). path is the cookie path.
    assert kw["https_only"] is True
    assert kw["same_site"] == "strict"
    assert kw["path"] == "/"


def test_cookie_secure_false(monkeypatch):
    monkeypatch.setenv("PRAXIS_COOKIE_SECRET", "x" * 48)
    monkeypatch.setenv("PRAXIS_COOKIE_SECURE", "false")
    kw = get_session_middleware_kwargs()
    assert kw["https_only"] is False


def test_cookie_secret_unset_generates_random(monkeypatch):
    monkeypatch.delenv("PRAXIS_COOKIE_SECRET", raising=False)
    kw = get_session_middleware_kwargs()
    assert kw["secret_key"]
    assert len(kw["secret_key"]) >= 32


def test_cookie_secret_short_logs_warning_accepted(monkeypatch, caplog):
    """TASK-12-02 (P1+ #3): a secret <32 bytes logs a WARNING but is accepted.

    A short non-empty secret (e.g., 'x') weakens the HMAC signature. The
    secret is still accepted (backward compat — pilot); post-pilot this
    should be a hard error. The WARNING is logged with remediation guidance.
    """
    from loguru import logger as _logger

    monkeypatch.setenv("PRAXIS_COOKIE_SECRET", "short-secret")  # 11 bytes < 32
    monkeypatch.setenv("PRAXIS_COOKIE_SECURE", "true")
    # Capture loguru warnings.
    msgs: list[str] = []
    sink_id = _logger.add(lambda m: msgs.append(str(m)), level="WARNING")
    try:
        kw = get_session_middleware_kwargs()
    finally:
        _logger.remove(sink_id)
    # The short secret is accepted (backward compat — no hard error in pilot).
    assert kw["secret_key"] == "short-secret"
    # A WARNING about the short secret was logged.
    assert any("<32 bytes" in m for m in msgs), \
        "short PRAXIS_COOKIE_SECRET should log a <32 bytes WARNING"


def test_cookie_secret_32_bytes_no_warning(monkeypatch, caplog):
    """TASK-12-02: a secret >=32 bytes logs no <32 bytes warning."""
    from loguru import logger as _logger

    monkeypatch.setenv("PRAXIS_COOKIE_SECRET", "x" * 32)  # exactly 32 bytes
    monkeypatch.setenv("PRAXIS_COOKIE_SECURE", "true")
    msgs: list[str] = []
    sink_id = _logger.add(lambda m: msgs.append(str(m)), level="WARNING")
    try:
        kw = get_session_middleware_kwargs()
    finally:
        _logger.remove(sink_id)
    assert kw["secret_key"] == "x" * 32
    # No <32 bytes warning (the secret is exactly 32 bytes).
    assert not any("<32 bytes" in m for m in msgs), \
        "32-byte secret should NOT log a <32 bytes warning"


def test_cookie_secret_long_no_warning(monkeypatch):
    """TASK-12-02: a secret >32 bytes logs no warning."""
    from loguru import logger as _logger

    monkeypatch.setenv("PRAXIS_COOKIE_SECRET", "x" * 64)  # 64 bytes
    monkeypatch.setenv("PRAXIS_COOKIE_SECURE", "true")
    msgs: list[str] = []
    sink_id = _logger.add(lambda m: msgs.append(str(m)), level="WARNING")
    try:
        kw = get_session_middleware_kwargs()
    finally:
        _logger.remove(sink_id)
    assert kw["secret_key"] == "x" * 64
    assert not any("<32 bytes" in m for m in msgs)


# ── current_operator dependency ─────────────────────────────────────────────


def _make_app_with_store(store) -> FastAPI:
    app = FastAPI()
    app.state.pg_store = store
    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef")
    app.include_router(router)
    return app


def _mock_store(operator_row=None):
    store = MagicMock()
    store.get_operator_by_id = AsyncMock(return_value=operator_row)
    return store


def test_current_operator_no_cookie_401():
    app = _make_app_with_store(_mock_store(operator_row=None))
    with TestClient(app) as client:
        r = client.get("/api/operator/me")
        assert r.status_code == 401


def test_current_operator_no_postgres_503():
    app = FastAPI()
    app.state.pg_store = None
    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef")
    app.include_router(router)
    with TestClient(app) as client:
        r = client.get("/api/operator/me")
        assert r.status_code == 503


def test_current_operator_inactive_401():
    op = {
        "id": "11111111-1111-1111-1111-111111111111",
        "username": "ghost",
        "display_name": "Ghost",
        "role": "operator",
        "is_active": False,
    }
    store = _mock_store(operator_row=op)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        # seed a session by hitting login would need a real store; instead
        # set the session directly via a cookie. Use TestClient's cookie jar.
        # Easiest: POST /login with a mocked store that returns the op.
        store.get_operator_by_username = AsyncMock(return_value=op)
        store.update_last_login = AsyncMock()
        store.pool = MagicMock()
        conn = MagicMock()
        conn.execute = AsyncMock()
        cm = MagicMock()
        cm.__aenter__ = AsyncMock(return_value=conn)
        cm.__aexit__ = AsyncMock(return_value=None)
        store.pool.acquire = MagicMock(return_value=cm)
        # hash the password so verify works
        op = dict(op)
        op["password_hash"] = hash_password("pw")
        store.get_operator_by_username = AsyncMock(return_value=op)
        r = client.post("/api/operator/login", json={"username": "ghost", "password": "pw"})
        # inactive operator → 401 even with correct password
        assert r.status_code == 401


def test_current_operator_valid_cookie_returns_operator():
    op = {
        "id": "22222222-2222-2222-2222-222222222222",
        "username": "alice",
        "display_name": "Alice",
        "role": "operator",
        "is_active": True,
        "password_hash": hash_password("pw"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    store.pool = MagicMock()
    conn = MagicMock()
    conn.execute = AsyncMock()
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "alice", "password": "pw"})
        assert r.status_code == 200
        assert r.json()["operator"]["username"] == "alice"
        # cookie is now set; /me should work
        r2 = client.get("/api/operator/me")
        assert r2.status_code == 200
        assert r2.json()["operator"]["username"] == "alice"


# ── Login route ─────────────────────────────────────────────────────────────


def test_login_wrong_password_401_no_cookie():
    op = {
        "id": "33333333-3333-3333-3333-333333333333",
        "username": "bob",
        "display_name": None,
        "role": "operator",
        "is_active": True,
        "password_hash": hash_password("correct"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "bob", "password": "wrong"})
        assert r.status_code == 401
        # no auth cookie set on failure
        cookies = client.cookies.get("praxis_op")
        assert not cookies


def test_login_unknown_user_401():
    store = _mock_store(operator_row=None)
    store.get_operator_by_username = AsyncMock(return_value=None)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "nobody", "password": "x"})
        assert r.status_code == 401


def test_login_no_postgres_503():
    app = FastAPI()
    app.state.pg_store = None
    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef")
    app.include_router(router)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "a", "password": "b"})
        assert r.status_code == 503


# ── Logout ──────────────────────────────────────────────────────────────────


def test_logout_clears_session():
    op = {
        "id": "44444444-4444-4444-4444-444444444444",
        "username": "carol",
        "display_name": "Carol",
        "role": "operator",
        "is_active": True,
        "password_hash": hash_password("pw"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    store.pool = MagicMock()
    conn = MagicMock()
    conn.execute = AsyncMock()
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        client.post("/api/operator/login", json={"username": "carol", "password": "pw"})
        r = client.post("/api/operator/logout")
        assert r.status_code == 200
        assert r.json()["ok"] is True
        # /me now 401
        r2 = client.get("/api/operator/me")
        assert r2.status_code == 401


# ── Rehash on login ─────────────────────────────────────────────────────────


def test_login_rehash_when_needed():
    from argon2 import PasswordHasher
    weak_hasher = PasswordHasher(time_cost=1, memory_cost=8, parallelism=1)
    op = {
        "id": "55555555-5555-5555-5555-555555555555",
        "username": "dave",
        "display_name": "Dave",
        "role": "operator",
        "is_active": True,
        "password_hash": weak_hasher.hash("pw"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    store.pool = MagicMock()
    executed = []
    conn = MagicMock()
    async def _exec(*a, **kw):
        executed.append(a)
    conn.execute = _exec
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "dave", "password": "pw"})
        assert r.status_code == 200
        assert executed, "rehash UPDATE should have run"
        # the second arg to execute is the new hash; verify it's argon2id
        assert executed[0][1].startswith("$argon2id$")


# ── Rate limiter ────────────────────────────────────────────────────────────


def test_rate_limit_login_decorator():
    # The decorator factory returns a decorator; applying it should not raise.
    deco = rate_limit_login()
    assert callable(deco)


def test_limiter_is_in_memory():
    assert getattr(limiter, "_storage_uri", "memory://") == "memory://" or limiter._storage is not None


# ── TASK-12-04 (P1+ #1/#2/#5): argon2id offload + 429 mock test + audit log ──


def test_login_argon2id_offloaded_to_thread():
    """TASK-12-04 (P1+ #1): verify_password is offloaded to asyncio.to_thread.

    The login handler should call verify_password via asyncio.to_thread (not
    directly) so the ~100-300ms argon2id hashing does not block the event loop.
    We verify by patching asyncio.to_thread to record the call.
    """
    import asyncio as _asyncio

    op = {
        "id": "44444444-4444-4444-4444-444444444444",
        "username": "erin",
        "display_name": "Erin",
        "role": "operator",
        "is_active": True,
        "password_hash": hash_password("pw"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    store.pool = MagicMock()
    conn = MagicMock()
    conn.execute = AsyncMock()
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)

    to_thread_calls: list = []
    real_to_thread = _asyncio.to_thread

    async def _spy_to_thread(func, *args, **kwargs):
        to_thread_calls.append((func, args, kwargs))
        return await real_to_thread(func, *args, **kwargs)

    import server.auth.routes as _routes_mod
    orig = _routes_mod.asyncio.to_thread
    _routes_mod.asyncio.to_thread = _spy_to_thread
    try:
        app = _make_app_with_store(store)
        with TestClient(app) as client:
            r = client.post("/api/operator/login", json={"username": "erin", "password": "pw"})
            assert r.status_code == 200
    finally:
        _routes_mod.asyncio.to_thread = orig

    # verify_password should have been called via asyncio.to_thread.
    assert to_thread_calls, "login should offload verify_password to asyncio.to_thread"
    func = to_thread_calls[0][0]
    assert func.__name__ == "verify_password", (
        f"expected verify_password offloaded, got {func.__name__}"
    )


def test_login_rehash_offloaded_to_thread():
    """TASK-12-04 (P1+ #1): hash_password (rehash) is also offloaded to thread."""
    from argon2 import PasswordHasher

    weak_hasher = PasswordHasher(time_cost=1, memory_cost=8, parallelism=1)
    op = {
        "id": "55555555-5555-5555-5555-555555555555",
        "username": "frank",
        "display_name": "Frank",
        "role": "operator",
        "is_active": True,
        "password_hash": weak_hasher.hash("pw"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    store.pool = MagicMock()
    conn = MagicMock()
    conn.execute = AsyncMock()
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)

    import asyncio as _asyncio
    import server.auth.routes as _routes_mod

    to_thread_calls: list = []
    real_to_thread = _routes_mod.asyncio.to_thread

    async def _spy_to_thread(func, *args, **kwargs):
        to_thread_calls.append((func, args, kwargs))
        return await real_to_thread(func, *args, **kwargs)

    _routes_mod.asyncio.to_thread = _spy_to_thread
    try:
        app = _make_app_with_store(store)
        with TestClient(app) as client:
            r = client.post("/api/operator/login", json={"username": "frank", "password": "pw"})
            assert r.status_code == 200
    finally:
        _routes_mod.asyncio.to_thread = real_to_thread

    # Both verify_password + hash_password should be offloaded.
    func_names = [c[0].__name__ for c in to_thread_calls]
    assert "verify_password" in func_names
    assert "hash_password" in func_names, "rehash should offload hash_password to thread"


def test_login_rate_limit_429_after_5_attempts():
    """TASK-12-04 (P1+ #2): mock-based 429 test — 6th login attempt → 429.

    The full 6th-attempt→429 path is in the PG-requiring integration test; this
    adds a mock-based test for CI coverage without Postgres. slowapi's in-memory
    limiter tracks per-IP; 5/minute → 6th attempt gets 429.
    """
    from slowapi.errors import RateLimitExceeded
    from slowapi.middleware import SlowAPIMiddleware
    from slowapi import _rate_limit_exceeded_handler

    op = {
        "id": "66666666-6666-6666-6666-666666666666",
        "username": "grace",
        "display_name": "Grace",
        "role": "operator",
        "is_active": True,
        "password_hash": hash_password("pw"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    store.pool = MagicMock()
    conn = MagicMock()
    conn.execute = AsyncMock()
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)

    app = _make_app_with_store(store)
    app.state.limiter = limiter
    app.add_middleware(SlowAPIMiddleware)
    app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

    with TestClient(app) as client:
        # 5 attempts should succeed (or 401 for wrong password — both count).
        statuses: list[int] = []
        for _ in range(5):
            r = client.post(
                "/api/operator/login", json={"username": "grace", "password": "pw"}
            )
            statuses.append(r.status_code)
        # The 5 attempts should not be 429 (within the 5/minute limit).
        assert all(s != 429 for s in statuses), f"first 5 should not be 429: {statuses}"
        # 6th attempt → 429 (rate limit exceeded).
        r6 = client.post(
            "/api/operator/login", json={"username": "grace", "password": "pw"}
        )
        assert r6.status_code == 429, (
            f"6th login attempt should be rate-limited (429), got {r6.status_code}"
        )


def test_credential_revocation_logs_audit_event():
    """TASK-12-04 (P1+ #5): credential revocation logs operator + cred_id.

    The revoke_credential endpoint should log an application-level audit event
    (no audit_log table — the log is sufficient for pilot per D-056).
    """
    import logging as _logging

    from server.operator.credentials import router as creds_router

    op = {
        "id": "77777777-7777-7777-7777-777777777777",
        "username": "heidi",
        "display_name": "Heidi",
        "role": "operator",
    }
    store = MagicMock()
    store.get_credential = AsyncMock(return_value={"id": "cred-xyz", "status": "active"})
    store.set_credential_status = AsyncMock()

    app = FastAPI()
    app.state.pg_store = store
    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef")
    app.include_router(creds_router)
    # Stub auth.
    from server.auth.dependencies import current_operator
    from server.auth.models import Operator

    async def _stub_op():
        return Operator(id=op["id"], username=op["username"],
                       display_name=op["display_name"], role=op["role"])
    app.dependency_overrides[current_operator] = _stub_op

    # Capture the audit log.
    cred_log = _logging.getLogger("server.operator.credentials")
    records: list[_logging.LogRecord] = []
    handler = _logging.Handler()
    handler.emit = records.append  # type: ignore[method-assign]
    cred_log.addHandler(handler)
    cred_log.setLevel(_logging.INFO)
    try:
        with TestClient(app) as client:
            r = client.post("/api/operator/credentials/cred-xyz/revoke")
            assert r.status_code == 200
            assert r.json()["status"] == "revoked"
    finally:
        cred_log.removeHandler(handler)

    # The audit log should contain the operator id + cred_id.
    audit_msgs = [r.getMessage() for r in records if r.levelno >= _logging.INFO]
    assert any("credential revoked" in m for m in audit_msgs), \
        f"revocation should log 'credential revoked': {audit_msgs}"
    assert any("cred-xyz" in m for m in audit_msgs), \
        f"audit log should contain cred_id: {audit_msgs}"
    assert any(op["id"] in m for m in audit_msgs), \
        f"audit log should contain operator id: {audit_msgs}"