davideandClaude Sonnet 5.5
24e5fc3d8b
- login and password-change fields: at most 256 characters
- wrong current password: per-user brake (5 free, 2s doubling to 300s), 429 even for the
right password while locked; password_changed audited in the same transaction
- POST /bug-reports: 10 per IP per hour, consumed only by a stored report
- GET /admin/users: limit/after_id paging, the panel walks the pages
- 401/403/404/405/422/503 all answer {code, message, params}; the 422 omits the input
- static files revalidated on every use (Cache-Control: no-cache)
- error.* keys for the new codes in all seven languages
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
184 lines
8.3 KiB
Python
184 lines
8.3 KiB
Python
"""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")
|