397 lines
16 KiB
Python
397 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
|
|
import httpx
|
|
|
|
from backend.bootstrap.application import create_application
|
|
from backend.bootstrap.settings import Settings
|
|
from backend.features.accounts.auth import CSRF_COOKIE, SESSION_COOKIE, SmartAccessPrincipal
|
|
from backend.features.accounts.service import add_months
|
|
from tests.support import run_scenario
|
|
|
|
ADMIN_PASSWORD = "Admin-pass-123!"
|
|
USER_PASSWORD = "User-pass-123!"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BrowserSession:
|
|
session: str
|
|
csrf: str
|
|
|
|
|
|
def current_session(client: httpx.AsyncClient) -> BrowserSession:
|
|
return BrowserSession(
|
|
session=client.cookies.get(SESSION_COOKIE),
|
|
csrf=client.cookies.get(CSRF_COOKIE),
|
|
)
|
|
|
|
|
|
def use_session(client: httpx.AsyncClient, session: BrowserSession) -> None:
|
|
client.cookies.clear()
|
|
client.cookies.set(SESSION_COOKIE, session.session)
|
|
client.cookies.set(CSRF_COOKIE, session.csrf)
|
|
|
|
|
|
def csrf_headers(session: BrowserSession) -> dict[str, str]:
|
|
return {"X-CSRF-Token": session.csrf}
|
|
|
|
|
|
async def register(
|
|
client: httpx.AsyncClient, username: str, password: str
|
|
) -> tuple[httpx.Response, BrowserSession]:
|
|
response = await client.post(
|
|
"/api/auth/register", json={"username": username, "password": password}
|
|
)
|
|
return response, current_session(client)
|
|
|
|
|
|
def test_first_account_is_admin_and_second_is_regular_user(tmp_path) -> None:
|
|
application = create_application(Settings.for_test(tmp_path))
|
|
|
|
async def scenario(client: httpx.AsyncClient) -> None:
|
|
admin_response, admin_session = await register(client, "leefer", ADMIN_PASSWORD)
|
|
assert admin_response.status_code == 201
|
|
assert admin_response.json()["account"] == {
|
|
"id": 1,
|
|
"username": "leefer",
|
|
"is_admin": True,
|
|
"membership_status": "not_open",
|
|
"membership_active": False,
|
|
"smart_access": True,
|
|
"badges": ["admin"],
|
|
}
|
|
|
|
client.cookies.clear()
|
|
user_response, user_session = await register(client, "小白用户", USER_PASSWORD)
|
|
assert user_response.status_code == 201
|
|
assert user_response.json()["account"]["is_admin"] is False
|
|
assert user_response.json()["account"]["smart_access"] is False
|
|
assert user_response.json()["account"]["badges"] == []
|
|
|
|
use_session(client, admin_session)
|
|
assert (await client.get("/api/auth/session")).json()["is_admin"] is True
|
|
use_session(client, user_session)
|
|
assert (await client.get("/api/auth/session")).json()["is_admin"] is False
|
|
|
|
with sqlite3.connect(application.state.settings.database_path) as connection:
|
|
stored = {row[0] for row in connection.execute("SELECT token_hash FROM sessions")}
|
|
assert admin_session.session not in stored
|
|
assert user_session.session not in stored
|
|
assert all(len(value) == 64 for value in stored)
|
|
|
|
run_scenario(application, scenario)
|
|
|
|
|
|
def test_csrf_password_change_and_other_session_revocation(tmp_path) -> None:
|
|
application = create_application(Settings.for_test(tmp_path))
|
|
|
|
async def scenario(client: httpx.AsyncClient) -> None:
|
|
_, first_session = await register(client, "secure-user", USER_PASSWORD)
|
|
second_login = await client.post(
|
|
"/api/auth/login",
|
|
json={"username": "secure-user", "password": USER_PASSWORD},
|
|
)
|
|
assert second_login.status_code == 200
|
|
second_session = current_session(client)
|
|
|
|
without_csrf = await client.patch(
|
|
"/api/account/password",
|
|
json={
|
|
"current_password": USER_PASSWORD,
|
|
"new_password": "Changed-pass-456!",
|
|
"confirmation": "Changed-pass-456!",
|
|
},
|
|
)
|
|
assert without_csrf.status_code == 403
|
|
assert without_csrf.json()["error"]["code"] == "csrf_failed"
|
|
|
|
wrong_current = await client.patch(
|
|
"/api/account/password",
|
|
headers=csrf_headers(second_session),
|
|
json={
|
|
"current_password": "Wrong-pass-999!",
|
|
"new_password": "Changed-pass-456!",
|
|
"confirmation": "Changed-pass-456!",
|
|
},
|
|
)
|
|
assert wrong_current.status_code == 400
|
|
assert wrong_current.json()["error"]["code"] == "invalid_current_password"
|
|
|
|
changed = await client.patch(
|
|
"/api/account/password",
|
|
headers=csrf_headers(second_session),
|
|
json={
|
|
"current_password": USER_PASSWORD,
|
|
"new_password": "Changed-pass-456!",
|
|
"confirmation": "Changed-pass-456!",
|
|
},
|
|
)
|
|
assert changed.status_code == 200
|
|
|
|
use_session(client, first_session)
|
|
assert (await client.get("/api/auth/session")).status_code == 401
|
|
client.cookies.clear()
|
|
old_login = await client.post(
|
|
"/api/auth/login",
|
|
json={"username": "secure-user", "password": USER_PASSWORD},
|
|
)
|
|
assert old_login.status_code == 401
|
|
assert old_login.json()["error"]["message"] == "账号或密码错误。"
|
|
new_login = await client.post(
|
|
"/api/auth/login",
|
|
json={"username": "secure-user", "password": "Changed-pass-456!"},
|
|
)
|
|
assert new_login.status_code == 200
|
|
|
|
run_scenario(application, scenario)
|
|
|
|
|
|
def test_birth_profile_is_encrypted_and_isolated_by_account(tmp_path) -> None:
|
|
application = create_application(Settings.for_test(tmp_path))
|
|
|
|
async def scenario(client: httpx.AsyncClient) -> None:
|
|
_, first_session = await register(client, "profile-a", USER_PASSWORD)
|
|
saved = await client.put(
|
|
"/api/account/profile",
|
|
headers=csrf_headers(first_session),
|
|
json={"birth_date": "1990-03-08", "birth_time": "08:30", "gender": "male"},
|
|
)
|
|
assert saved.status_code == 200
|
|
assert saved.json()["configured"] is True
|
|
assert "仅当前账号可见" in saved.json()["privacy_notice"]
|
|
|
|
client.cookies.clear()
|
|
_, second_session = await register(client, "profile-b", USER_PASSWORD)
|
|
assert (await client.get("/api/account/profile")).json()["configured"] is False
|
|
|
|
with sqlite3.connect(application.state.settings.database_path) as connection:
|
|
encrypted = connection.execute(
|
|
"SELECT encrypted_payload FROM birth_profiles WHERE user_id = 1"
|
|
).fetchone()[0]
|
|
assert "1990-03-08" not in encrypted
|
|
assert "08:30" not in encrypted
|
|
|
|
use_session(client, first_session)
|
|
own_profile = await client.get("/api/account/profile")
|
|
assert own_profile.json()["birth_date"] == "1990-03-08"
|
|
deleted = await client.delete("/api/account/profile", headers=csrf_headers(first_session))
|
|
assert deleted.status_code == 200
|
|
assert (await client.get("/api/account/profile")).json()["configured"] is False
|
|
|
|
use_session(client, second_session)
|
|
assert (await client.get("/api/account/profile")).json()["configured"] is False
|
|
|
|
run_scenario(application, scenario)
|
|
|
|
|
|
def test_incomplete_encrypted_profile_returns_controlled_error(tmp_path) -> None:
|
|
application = create_application(Settings.for_test(tmp_path))
|
|
|
|
async def scenario(client: httpx.AsyncClient) -> None:
|
|
response, _session = await register(client, "broken-profile", USER_PASSWORD)
|
|
user_id = response.json()["account"]["id"]
|
|
encrypted = application.state.container.accounts._cipher.encrypt('{"gender":"male"}')
|
|
with sqlite3.connect(application.state.settings.database_path) as connection:
|
|
connection.execute(
|
|
"INSERT INTO birth_profiles VALUES (?,?,datetime('now'),datetime('now'))",
|
|
(user_id, encrypted),
|
|
)
|
|
result = await client.get("/api/account/profile")
|
|
assert result.status_code == 503
|
|
assert result.json()["error"]["code"] == "profile_unavailable"
|
|
|
|
run_scenario(application, scenario)
|
|
|
|
|
|
def test_membership_and_admin_are_independent_dimensions(tmp_path) -> None:
|
|
application = create_application(Settings.for_test(tmp_path))
|
|
|
|
async def scenario(client: httpx.AsyncClient) -> None:
|
|
_, admin_session = await register(client, "admin-user", ADMIN_PASSWORD)
|
|
client.cookies.clear()
|
|
user_response, user_session = await register(client, "member-user", USER_PASSWORD)
|
|
user_id = user_response.json()["account"]["id"]
|
|
|
|
use_session(client, admin_session)
|
|
activated = await client.patch(
|
|
f"/api/admin/memberships/{user_id}",
|
|
headers=csrf_headers(admin_session),
|
|
json={"action": "activate", "duration": "1_month", "daily_limit": 80},
|
|
)
|
|
assert activated.status_code == 200
|
|
assert activated.json()["membership"]["active"] is True
|
|
assert activated.json()["membership"]["daily_limit"] == 80
|
|
|
|
admin_membership = await client.get("/api/account/membership")
|
|
assert admin_membership.json()["active"] is False
|
|
assert admin_membership.json()["smart_access"] is True
|
|
|
|
use_session(client, user_session)
|
|
user_identity = await client.get("/api/auth/session")
|
|
assert user_identity.json()["badges"] == ["member"]
|
|
assert user_identity.json()["smart_access"] is True
|
|
|
|
use_session(client, admin_session)
|
|
permanent = await client.patch(
|
|
"/api/admin/memberships/1",
|
|
headers=csrf_headers(admin_session),
|
|
json={"action": "activate", "duration": "permanent"},
|
|
)
|
|
assert permanent.status_code == 200
|
|
assert permanent.json()["membership"]["is_permanent"] is True
|
|
assert permanent.json()["membership"]["expires_at"] is None
|
|
assert permanent.json()["membership"]["remaining_days"] is None
|
|
identity = await client.get("/api/auth/session")
|
|
assert identity.json()["badges"] == ["admin", "member"]
|
|
|
|
run_scenario(application, scenario)
|
|
|
|
|
|
def test_non_admin_cannot_read_or_write_system_credentials(tmp_path) -> None:
|
|
application = create_application(Settings.for_test(tmp_path))
|
|
|
|
async def scenario(client: httpx.AsyncClient) -> None:
|
|
_, admin_session = await register(client, "system-admin", ADMIN_PASSWORD)
|
|
client.cookies.clear()
|
|
_, user_session = await register(client, "system-user", USER_PASSWORD)
|
|
|
|
denied_read = await client.get("/api/admin/system/credentials")
|
|
assert denied_read.status_code == 403
|
|
denied_write = await client.put(
|
|
"/api/admin/system/credentials/tushare_token",
|
|
headers=csrf_headers(user_session),
|
|
json={"value": "user-must-not-save-this"},
|
|
)
|
|
assert denied_write.status_code == 403
|
|
|
|
use_session(client, admin_session)
|
|
saved = await client.put(
|
|
"/api/admin/system/credentials/tushare_token",
|
|
headers=csrf_headers(admin_session),
|
|
json={"value": "real-test-credential-value"},
|
|
)
|
|
assert saved.status_code == 200
|
|
statuses = await client.get("/api/admin/system/credentials")
|
|
assert statuses.status_code == 200
|
|
configured = next(item for item in statuses.json() if item["name"] == "tushare_token")
|
|
assert configured["configured"] is True
|
|
assert "value" not in configured
|
|
assert "real-test-credential-value" not in statuses.text
|
|
|
|
with sqlite3.connect(application.state.settings.database_path) as connection:
|
|
encrypted = connection.execute(
|
|
"SELECT encrypted_value FROM system_credentials WHERE name = 'tushare_token'"
|
|
).fetchone()[0]
|
|
assert encrypted != "real-test-credential-value"
|
|
assert "real-test-credential-value" not in encrypted
|
|
|
|
run_scenario(application, scenario)
|
|
|
|
|
|
def test_session_cookies_duplicate_username_and_logout(tmp_path) -> None:
|
|
application = create_application(Settings.for_test(tmp_path))
|
|
|
|
async def scenario(client: httpx.AsyncClient) -> None:
|
|
response, session = await register(client, "CaseUser", USER_PASSWORD)
|
|
cookies = response.headers.get_list("set-cookie")
|
|
session_cookie = next(value for value in cookies if value.startswith(f"{SESSION_COOKIE}="))
|
|
csrf_cookie = next(value for value in cookies if value.startswith(f"{CSRF_COOKIE}="))
|
|
assert "HttpOnly" in session_cookie
|
|
assert "SameSite=lax" in session_cookie
|
|
assert "HttpOnly" not in csrf_cookie
|
|
assert "SameSite=lax" in csrf_cookie
|
|
|
|
client.cookies.clear()
|
|
duplicate = await client.post(
|
|
"/api/auth/register",
|
|
json={"username": "caseuser", "password": USER_PASSWORD},
|
|
)
|
|
assert duplicate.status_code == 409
|
|
assert duplicate.json()["error"]["code"] == "username_taken"
|
|
|
|
use_session(client, session)
|
|
logged_out = await client.post("/api/auth/logout", headers=csrf_headers(session))
|
|
assert logged_out.status_code == 200
|
|
assert (await client.get("/api/auth/session")).status_code == 401
|
|
|
|
run_scenario(application, scenario)
|
|
|
|
|
|
def test_active_membership_renews_from_existing_expiry_and_disable_keeps_it(tmp_path) -> None:
|
|
application = create_application(Settings.for_test(tmp_path))
|
|
|
|
async def scenario(client: httpx.AsyncClient) -> None:
|
|
_, admin_session = await register(client, "renew-admin", ADMIN_PASSWORD)
|
|
client.cookies.clear()
|
|
user_response, _ = await register(client, "renew-user", USER_PASSWORD)
|
|
user_id = user_response.json()["account"]["id"]
|
|
use_session(client, admin_session)
|
|
|
|
first = await client.patch(
|
|
f"/api/admin/memberships/{user_id}",
|
|
headers=csrf_headers(admin_session),
|
|
json={"action": "activate", "duration": "1_month"},
|
|
)
|
|
first_expiry = datetime.fromisoformat(first.json()["membership"]["expires_at"])
|
|
extended = await client.patch(
|
|
f"/api/admin/memberships/{user_id}",
|
|
headers=csrf_headers(admin_session),
|
|
json={"action": "activate", "duration": "3_months"},
|
|
)
|
|
extended_expiry = datetime.fromisoformat(extended.json()["membership"]["expires_at"])
|
|
assert extended_expiry == add_months(first_expiry, 3)
|
|
|
|
disabled = await client.patch(
|
|
f"/api/admin/memberships/{user_id}",
|
|
headers=csrf_headers(admin_session),
|
|
json={"action": "disable"},
|
|
)
|
|
assert disabled.json()["membership"]["status"] == "disabled"
|
|
assert disabled.json()["membership"]["active"] is False
|
|
assert (
|
|
datetime.fromisoformat(disabled.json()["membership"]["expires_at"]) == extended_expiry
|
|
)
|
|
|
|
run_scenario(application, scenario)
|
|
|
|
|
|
def test_smart_access_allows_admin_and_member_but_not_regular_user(tmp_path) -> None:
|
|
application = create_application(Settings.for_test(tmp_path))
|
|
|
|
@application.get("/api/test/smart-access")
|
|
def smart_access_probe(_principal: SmartAccessPrincipal) -> dict[str, bool]:
|
|
return {"allowed": True}
|
|
|
|
async def scenario(client: httpx.AsyncClient) -> None:
|
|
_, admin_session = await register(client, "smart-admin", ADMIN_PASSWORD)
|
|
client.cookies.clear()
|
|
member_response, member_session = await register(
|
|
client, "smart-member", USER_PASSWORD
|
|
)
|
|
member_id = member_response.json()["account"]["id"]
|
|
client.cookies.clear()
|
|
_, user_session = await register(client, "smart-user", USER_PASSWORD)
|
|
|
|
denied = await client.get("/api/test/smart-access")
|
|
assert denied.status_code == 403
|
|
assert denied.json()["error"]["code"] == "membership_required"
|
|
|
|
use_session(client, admin_session)
|
|
assert (await client.get("/api/test/smart-access")).status_code == 200
|
|
activated = await client.patch(
|
|
f"/api/admin/memberships/{member_id}",
|
|
headers=csrf_headers(admin_session),
|
|
json={"action": "activate", "duration": "1_month"},
|
|
)
|
|
assert activated.status_code == 200
|
|
|
|
use_session(client, member_session)
|
|
assert (await client.get("/api/test/smart-access")).status_code == 200
|
|
use_session(client, user_session)
|
|
assert (await client.get("/api/test/smart-access")).status_code == 403
|
|
|
|
run_scenario(application, scenario)
|