feat: add cookie sessions and backend role authorization

2026-10-16 07:51:59 +08:00
parent 411eadb431
commit 3f25ffab83
28 changed files with 1186 additions and 31 deletions
@@ -0,0 +1,342 @@
+"""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