343 lines
13 KiB
Python
343 lines
13 KiB
Python
"""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
|