"""B-94 (SEC-54, SEC-55, API-09, API-15, API-16, API-19, API-21, API-28, API-33, REG-12): credential length bounds, the password-change brake and audit, the bug-report quota, paged user list, structured errors everywhere and revalidated static files.""" import json import pytest from cryptography.fernet import Fernet from httpx import ASGITransport, AsyncClient from sqlalchemy import select from app.config import settings @pytest.fixture async def client(monkeypatch, tmp_path): monkeypatch.setattr(settings, "database_url", f"sqlite+aiosqlite:///{tmp_path}/test.db") monkeypatch.setattr(settings, "jwt_secret", "test-jwt-secret") monkeypatch.setattr(settings, "admin_token", "admin-secret") monkeypatch.setattr(settings, "xprv_encryption_key", Fernet.generate_key().decode()) monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc")) import app.db.models # noqa: F401 import app.wallet.hd as hd from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.db import base as db_base from app.db import session as db_session hd._account_key = None hd.generate_master_key() db_base.engine = create_async_engine(settings.database_url) db_base.AsyncSessionLocal = async_sessionmaker(db_base.engine, expire_on_commit=False) db_session.AsyncSessionLocal = db_base.AsyncSessionLocal async with db_base.engine.begin() as conn: await conn.run_sync(db_base.Base.metadata.create_all) from app.electrum.listener import ElectrumListener from app.main import app app.state.electrum_listener = ElectrumListener(lambda endpoint: None, db_base.AsyncSessionLocal) app.state.auth_rate_limiters = None if hasattr(app.state, "auth_rate_limiters"): del app.state.auth_rate_limiters # fresh throttle state per test async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: yield ac await db_base.engine.dispose() ADMIN = {"X-Admin-Token": "admin-secret"} async def _register(client, username="alice", password="original-password"): resp = await client.post("/auth/register", json={"username": username, "password": password}) assert resp.status_code == 201 return {"Authorization": f"Bearer {resp.json()['access_token']}"} def _code(resp): return resp.json()["detail"]["code"] # --- field lengths (API-19, REG-12) and the 422 shape (API-15) --------------------- async def test_login_fields_are_bounded_and_the_value_is_not_echoed(client): resp = await client.post("/auth/login", json={"username": "a", "password": "secret-" + "x" * 300}) assert resp.status_code == 422 assert _code(resp) == "invalid_request" errors = resp.json()["detail"]["params"]["errors"] assert errors[0]["loc"] == ["body", "password"] and set(errors[0]) == {"loc", "msg", "type"} assert "secret-" not in resp.text async def test_change_password_fields_are_bounded(client): headers = await _register(client) resp = await client.post( "/users/me/change-password", headers=headers, json={"current_password": "original-password", "new_password": "x" * 257}, ) assert resp.status_code == 422 and _code(resp) == "invalid_request" # --- the password-change brake (SEC-54) and its audit (AUD-42) --------------------- async def test_the_fifth_wrong_current_password_locks_even_the_right_one_out(client): headers = await _register(client) body = {"current_password": "wrong-password", "new_password": "brand-new-password"} for _ in range(5): assert (await client.post("/users/me/change-password", headers=headers, json=body)).status_code == 401 resp = await client.post("/users/me/change-password", headers=headers, json=body) assert resp.status_code == 429 and _code(resp) == "rate_limited" right = {"current_password": "original-password", "new_password": "brand-new-password"} locked = await client.post("/users/me/change-password", headers=headers, json=right) assert locked.status_code == 429 and locked.json()["detail"]["params"]["retry_after_seconds"] >= 1 async def test_a_successful_change_is_audited_and_clears_the_brake(client): from app.db import base as db_base from app.db.models import AuditLog headers = await _register(client) bad = {"current_password": "wrong-password", "new_password": "brand-new-password"} for _ in range(3): await client.post("/users/me/change-password", headers=headers, json=bad) ok = {"current_password": "original-password", "new_password": "brand-new-password"} assert (await client.post("/users/me/change-password", headers=headers, json=ok)).status_code == 200 async with db_base.AsyncSessionLocal() as session: rows = (await session.execute(select(AuditLog).where(AuditLog.event_type == "password_changed"))).scalars().all() assert len(rows) == 1 and json.loads(rows[0].payload_json) == {"user_id": rows[0].user_id} assert "password" not in rows[0].payload_json.replace("user_id", "") # --- bug reports (SEC-55) ---------------------------------------------------------- async def test_the_11th_bug_report_from_one_ip_waits_but_an_invalid_one_costs_nothing(client): for _ in range(3): assert (await client.post("/bug-reports", json={"description": " "})).status_code == 422 for i in range(10): assert (await client.post("/bug-reports", json={"description": f"bug {i}"})).status_code == 201 resp = await client.post("/bug-reports", json={"description": "one too many"}) assert resp.status_code == 429 and _code(resp) == "rate_limited" assert 3000 < resp.json()["detail"]["params"]["retry_after_seconds"] <= 3601 # --- paged user list (API-33) ------------------------------------------------------ async def test_admin_users_are_paged_by_id(client): for i in range(5): await _register(client, username=f"user{i}") first = (await client.get("/admin/users", headers=ADMIN, params={"limit": 2})).json() second = (await client.get("/admin/users", headers=ADMIN, params={"limit": 2, "after_id": first[-1]["id"]})).json() last = (await client.get("/admin/users", headers=ADMIN, params={"limit": 2, "after_id": second[-1]["id"]})).json() assert [u["username"] for u in first + second + last] == [f"user{i}" for i in range(5)] assert len(last) == 1 assert (await client.get("/admin/users", headers=ADMIN, params={"limit": 501})).status_code == 422 # --- structured errors everywhere (API-16) ----------------------------------------- async def test_admin_errors_are_structured(client): wrong = await client.get("/admin/users", headers={"X-Admin-Token": "nope"}) assert wrong.status_code == 403 and _code(wrong) == "invalid_admin_token" missing = await client.post("/admin/users/999/reset-password", headers=ADMIN) assert missing.status_code == 404 and _code(missing) == "user_not_found" report = await client.post("/admin/bug-reports/999/status", headers=ADMIN, json={"status": "read"}) assert report.status_code == 404 and _code(report) == "bug_report_not_found" async def test_framework_errors_are_structured(client): absent = await client.get("/users/me") assert absent.status_code == 401 and _code(absent) == "not_authenticated" assert absent.headers["www-authenticate"] == "Bearer" assert _code(await client.get("/no/such/path")) == "not_found" wrong_method = await client.put("/rounds/current") assert _code(wrong_method) in {"method_not_allowed", "not_found"} assert (await client.get("/users/me", headers={"Authorization": "Bearer junk"})).json()["detail"]["code"] == "session_expired" async def test_the_sse_cap_answers_in_the_structured_shape(client, monkeypatch): from app.api.routes import rounds from app.rounds.events import RoundEventCapacityError def full(_ip): raise RoundEventCapacityError() monkeypatch.setattr(rounds.broadcaster, "subscribe", full) resp = await client.get("/rounds/stream") assert resp.status_code == 503 and _code(resp) == "too_many_streams" # --- static files (API-09) --------------------------------------------------------- async def test_static_files_are_revalidated_on_every_use(client): resp = await client.get("/app.js") assert resp.status_code == 200 and resp.headers["cache-control"] == "no-cache" assert resp.headers.get("etag")