border-red-post-nalati/backend/tests/test_auth.py

"""Authentication, authorization and CSRF boundaries over real PostgreSQL."""

import json
from datetime import timedelta

import pytest
from pydantic import ValidationError
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.core.config import Settings
from app.models.auth import AuthSession, User
from app.models.catalog import utc_now
from app.services.auth import COOKIE_NAME, check_password, token_hash

PASSWORD = "test-password-123"
ORIGIN = "http://localhost:5173"


def register(client, username="test_user", **extra):
    return client.post(
        "/api/auth/register", json={"username": username, "password": PASSWORD, **extra}
    )


def login(client, username="test_user", password=PASSWORD):
    response = client.post("/api/auth/login", json={"username": username, "password": password})
    if response.status_code == 200:
        client.headers["X-CSRF-Token"] = response.json()["csrf_token"]
    return response


def test_registration_hashes_password_and_cannot_select_privilege(
    anonymous_client, migrated_database
):
    c = anonymous_client
    for extra in [{"role": "admin"}, {"id": 1}, {"password_hash": "injected"}]:
        assert register(c, **extra).status_code == 422
    result = register(c, username=" Test_User ", email="user@example.com", nickname="占位用户")
    assert result.status_code == 201
    assert result.json()["username"] == "test_user" and result.json()["role"] == "user"
    assert c.get("/api/auth/me").status_code == 401
    assert register(c).status_code == 409
    with Session(migrated_database) as db:
        user = db.scalar(select(User))
        assert user.password_hash.startswith("$argon2id$")
        assert PASSWORD not in user.password_hash and check_password(user.password_hash, PASSWORD)
        assert db.scalar(select(func.count()).select_from(AuthSession)) == 0
    assert "password" not in result.text and "credential_version" not in result.text


@pytest.mark.parametrize(
    "data",
    [
        {"password": "short"},
        {"password": "x" * 129},
        {"username": "a"},
        {"username": "a b"},
        {"email": "invalid"},
        {"nickname": "x" * 81},
        {"nickname": "bad\u0000name"},
    ],
)
def test_validation_does_not_echo_credentials(anonymous_client, data):
    submitted = {"username": "test_user", "password": PASSWORD, **data}
    response = anonymous_client.post("/api/auth/register", json=submitted)
    assert response.status_code == 422
    assert '"input"' not in response.text
    assert json.dumps(submitted["password"]) not in response.text
    assert response.headers["cache-control"] == "no-store"


@pytest.mark.parametrize("origin", [None, "null", "https://evil.example", ORIGIN + ".evil.example"])
def test_login_and_registration_reject_untrusted_origin(
    anonymous_client, migrated_database, origin
):
    c = anonymous_client
    c.headers.pop("Origin", None)
    headers = {"Origin": origin} if origin else {}
    for path in ["register", "login"]:
        response = c.post(
            f"/api/auth/{path}",
            json={"username": "test_user", "password": PASSWORD},
            headers=headers,
        )
        assert response.status_code == 403
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(User)) == 0


def test_login_cookie_rotation_logout_and_opaque_storage(anonymous_client, migrated_database):
    c = anonymous_client
    register(c)
    response = login(c)
    assert response.status_code == 200
    assert all(
        value in response.headers["set-cookie"]
        for value in ["HttpOnly", "Secure", "SameSite=lax", "Path=/"]
    )
    first = c.cookies.get(COOKIE_NAME)
    csrf = response.json()["csrf_token"]
    assert c.get("/api/auth/me").json()["csrf_token"] == csrf
    with Session(migrated_database) as db:
        stored = db.scalar(select(AuthSession))
        assert stored.token_hash == token_hash(first) and first not in stored.token_hash
    assert login(c).status_code == 200
    second = c.cookies.get(COOKIE_NAME)
    assert first != second
    c.cookies.clear()
    c.cookies.set(COOKIE_NAME, first)
    assert c.get("/api/auth/me").status_code == 401
    c.cookies.clear()
    c.cookies.set(COOKIE_NAME, second)
    response = c.post("/api/auth/logout")
    assert response.status_code == 204 and "Max-Age=0" in response.headers["set-cookie"]
    c.cookies.clear()
    c.cookies.set(COOKIE_NAME, second)
    assert c.get("/api/auth/me").status_code == 401


def test_login_failure_is_generic_and_throttled_in_database(anonymous_client):
    c = anonymous_client
    register(c)
    absent = login(c, "absent_user", "wrong").json()
    wrong = login(c, password="wrong")
    assert wrong.status_code == 401 and wrong.json() == absent
    for _ in range(9):
        assert login(c, password="wrong").status_code == 401
    blocked = login(c)
    assert blocked.status_code == 429 and blocked.headers["retry-after"] == "300"


@pytest.mark.parametrize("username", ["bad\u0000name", "invalid name", "../user"])
def test_login_rejects_invalid_username_before_database_lookup(anonymous_client, username):
    assert login(anonymous_client, username=username).status_code == 422


def test_expired_forged_and_credential_version_sessions_are_rejected(
    anonymous_client, migrated_database
):
    c = anonymous_client
    register(c)
    login(c)
    with Session(migrated_database) as db:
        session = db.scalar(select(AuthSession))
        session.expires_at = utc_now() - timedelta(seconds=1)
        db.commit()
    assert c.get("/api/auth/me").status_code == 401
    login(c)
    with Session(migrated_database) as db:
        db.scalar(select(User)).credential_version += 1
        db.commit()
    assert c.get("/api/auth/me").status_code == 401
    c.cookies.clear()
    c.cookies.set(COOKIE_NAME, "x" * 43)
    assert c.get("/api/auth/me").status_code == 401
    c.cookies.clear()
    c.cookies.set(COOKIE_NAME, "invalid")
    assert c.get("/api/auth/me").status_code == 401


def test_csrf_is_bound_to_session_and_required_for_all_authenticated_writes(anonymous_client):
    c = anonymous_client
    register(c)
    login(c)
    first = c.headers["X-CSRF-Token"]
    login(c)
    for token in ["", "invalid", first]:
        assert (
            c.patch(
                "/api/users/me", json={"nickname": "占位"}, headers={"X-CSRF-Token": token}
            ).status_code
            == 403
        )
        assert c.post("/api/auth/logout", headers={"X-CSRF-Token": token}).status_code == 403
    assert (
        c.patch(
            "/api/users/me", json={"nickname": "占位"}, headers={"Origin": "https://evil.example"}
        ).status_code
        == 403
    )
    c.headers.pop("Origin")
    assert c.patch("/api/users/me", json={"nickname": "占位"}).status_code == 403
    c.headers["Origin"] = ORIGIN
    assert c.patch("/api/users/me", json={"nickname": " 占位 "}).json()["nickname"] == "占位"


def test_profile_updates_only_own_allowed_fields(anonymous_client, migrated_database):
    c = anonymous_client
    first = register(c).json()
    register(c, "other_user")
    login(c)
    for payload in [
        {"id": 2},
        {"role": "admin"},
        {"username": "other_user"},
        {"password": "changed"},
        {},
    ]:
        assert c.patch("/api/users/me", json=payload).status_code == 422
    assert (
        c.patch("/api/users/me", json={"email": "changed@example.com", "nickname": "示例"}).json()[
            "id"
        ]
        == first["id"]
    )
    assert c.patch("/api/users/me", json={"email": "", "nickname": None}).json()["email"] is None
    with Session(migrated_database) as db:
        assert db.scalar(select(User).where(User.username == "other_user")).email is None
        assert db.scalar(select(User).where(User.id == first["id"])).role == "user"


def test_password_change_revokes_all_sessions(anonymous_client):
    c = anonymous_client
    register(c)
    login(c)
    first = c.cookies.get(COOKIE_NAME)
    # Independent second browser: existing first session must also be invalidated.
    c.cookies.clear()
    login(c)
    assert (
        c.post(
            "/api/auth/change-password",
            json={"current_password": "wrong", "new_password": "new-password-123"},
        ).status_code
        == 400
    )
    response = c.post(
        "/api/auth/change-password",
        json={"current_password": PASSWORD, "new_password": "new-password-123"},
    )
    assert response.status_code == 204
    assert c.get("/api/auth/me").status_code == 401
    c.cookies.set(COOKIE_NAME, first)
    assert c.get("/api/auth/me").status_code == 401
    assert login(c).status_code == 401
    assert login(c, password="new-password-123").status_code == 200


def test_all_management_routes_require_admin_and_history_is_private(anonymous_client):
    c = anonymous_client
    paths = [
        ("GET", "/api/admin/users"),
        ("GET", "/api/knowledge"),
        ("GET", "/api/knowledge/page"),
        ("GET", "/api/knowledge/1"),
        ("POST", "/api/knowledge"),
        ("PATCH", "/api/knowledge/1"),
        ("DELETE", "/api/knowledge/1?expected_version=1"),
        ("POST", "/api/knowledge/1/review"),
        ("GET", "/api/knowledge/1/history"),
        ("GET", "/api/sources"),
        ("POST", "/api/sources"),
        ("GET", "/api/sources/1"),
        ("PATCH", "/api/sources/1"),
        ("DELETE", "/api/sources/1?expected_version=1"),
        ("GET", "/api/sources/1/history"),
        ("GET", "/api/documents"),
        ("POST", "/api/documents"),
        ("GET", "/api/documents/1"),
        ("PATCH", "/api/documents/1"),
        ("DELETE", "/api/documents/1?expected_version=1"),
        ("GET", "/api/documents/1/history"),
        ("GET", "/api/retrieval/runs"),
        ("GET", "/api/retrieval/runs/00000000-0000-0000-0000-000000000000"),
    ]
    for method, path in paths:
        assert c.request(method, path, json={}).status_code == 401, path
    register(c)
    login(c)
    for method, path in paths:
        assert c.request(method, path, json={}).status_code == 403, path
    assert c.post("/api/retrieval", json={"query": "占位"}).json()["status"] == "no_evidence"
    assert c.post("/api/chat", json={"question": "占位"}).json()["status"] == "stub"


def test_admin_access_live_role_and_no_csrf_bypass(client, migrated_database):
    c = client
    users = c.get("/api/admin/users").json()
    assert users[0]["role"] == "admin" and "password_hash" not in str(users)
    assert c.get("/api/admin/users?limit=101").status_code == 422
    assert (
        c.post(
            "/api/knowledge",
            json={"title": "示例", "content": "示例"},
            headers={"X-CSRF-Token": ""},
        ).status_code
        == 403
    )
    assert c.post("/api/knowledge", json={"title": "示例", "content": "示例"}).status_code == 201
    with Session(migrated_database) as db:
        db.scalar(select(User)).role = "user"
        db.commit()
    assert c.get("/api/admin/users").status_code == 403
    assert c.get("/api/knowledge").status_code == 403


def test_auth_settings_require_explicit_origins_and_secure_production():
    with pytest.raises(ValidationError):
        Settings(_env_file=None, app_env="production", auth_cookie_secure=False)
    for origin in [
        "*",
        "null",
        "https://example.com/path",
        "https://user@example.com",
        "https://*.example.com",
    ]:
        with pytest.raises(ValidationError):
            Settings(_env_file=None, auth_allowed_origins=origin)


def test_local_admin_provisioning_has_no_default_account(migrated_database, monkeypatch, capsys):
    from sqlalchemy.orm import sessionmaker

    from app import manage

    monkeypatch.setattr(manage, "SessionLocal", sessionmaker(bind=migrated_database))
    monkeypatch.setattr("sys.argv", ["manage", "create-admin", "--username", "local_admin"])
    monkeypatch.setattr(manage.getpass, "getpass", lambda _: PASSWORD)
    manage.main()
    with Session(migrated_database) as db:
        user = db.scalar(select(User))
        assert user.role == "admin" and check_password(user.password_hash, PASSWORD)
    with pytest.raises(SystemExit) as error:
        manage.main()
    assert error.value.code == 1
    assert PASSWORD not in capsys.readouterr().out


def test_client_budget_is_shared_across_names_and_registration(anonymous_client, migrated_database):
    c = anonymous_client
    for index in range(30):
        assert login(c, username=f"absent_{index}").status_code == 401
    assert register(c).status_code == 429
    # Expire the persisted budget instead of waiting in a test.
    from app.models.auth import AuthRateLimit

    with Session(migrated_database) as db:
        for bucket in db.scalars(select(AuthRateLimit)):
            bucket.window_start = utc_now() - timedelta(minutes=6)
        db.commit()
    assert register(c).status_code == 201