fix(setup): confronto costante, blocco dopo l'init, status minimale (#9)

_check_password confrontava la password con !=, ora hmac.compare_digest.

Nuova guardia _require_not_initialised: se esiste già un admin,
reset-db, seed, manage-user, change-user-password e init-db rispondono
403 "setup already completed", a meno di settings.setup_allow_after_init
(nuovo, env SETUP_ALLOW_AFTER_INIT, default False). create-admin,
list-users e toggle-user-active restano fuori dalla guardia.

GET /api/setup/status senza password valida restituisce solo
{"initialised": bool}; l'elenco tabelle solo con la password.

Adeguati i test esistenti che seedavano un admin e poi richiamavano un
endpoint ora guardato (test_setup.py, test_station_seed.py: bypassano
la guardia con setup_allow_after_init quando serve testare altro).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-21 22:38:09 +02:00
parent 9afe1e712b
commit 78970c69b0
4 changed files with 237 additions and 6 deletions
+63 -3
View File
@@ -3,6 +3,7 @@
Protected by SETUP_PASSWORD env var. When empty/None, all endpoints return 404.
NOT protected by API Key auth (standalone access).
"""
import hmac
import random
import secrets
from datetime import datetime, timedelta
@@ -87,13 +88,57 @@ def _check_setup_enabled() -> None:
def _check_password(password: str) -> None:
"""Verify the provided password matches the setup password."""
_check_setup_enabled()
if password != settings.setup_password:
if not hmac.compare_digest(password, settings.setup_password):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid setup password",
)
def _password_matches(password: str | None) -> bool:
"""Constant-time password check that tolerates a missing password."""
if not password or not settings.setup_password:
return False
return hmac.compare_digest(password, settings.setup_password)
async def _admin_exists() -> bool:
"""Whether at least one admin user has been created.
False on a database with no `users` table yet, so init-db and the first
seed are never blocked by a query against a table that does not exist.
"""
async with engine.connect() as conn:
has_users = await conn.run_sync(
lambda sync_conn: sa_inspect(sync_conn).has_table("users")
)
if not has_users:
return False
async with async_session_factory() as session:
result = await session.execute(select(User).where(User.is_admin == True)) # noqa: E712
return result.scalars().first() is not None
async def _require_not_initialised() -> None:
"""Refuse privileged/destructive setup endpoints once an admin exists.
reset-db, seed, manage-user, change-user-password and init-db all reach
into a live database once setup is done: reset-db wipes it, seed adds
demo users and data, manage-user/change-user-password can take over an
account, init-db races Alembic. All of them are meant for the one-time
bring-up before there is an admin to log in as.
`settings.setup_allow_after_init` is the explicit override for scripted
re-provisioning of a demo/test deploy.
"""
if settings.setup_allow_after_init:
return
if await _admin_exists():
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="setup already completed",
)
def _generate_measurements_for_subtask(
subtask: RecipeSubtask,
version_id: int,
@@ -220,14 +265,24 @@ async def setup_page(request: Request):
@router.get("/status")
async def setup_status():
"""Return database table status (no password required, but setup must be enabled)."""
async def setup_status(password: str | None = None):
"""Database status. Without a valid password, only whether setup is done.
The table listing is only returned with the setup password: it is
internal detail, not something to hand to whoever happens to load the
page.
"""
_check_setup_enabled()
initialised = await _admin_exists()
if not _password_matches(password):
return {"initialised": initialised}
async with engine.connect() as conn:
table_names: list[str] = await conn.run_sync(
lambda sync_conn: sa_inspect(sync_conn).get_table_names()
)
return {
"initialised": initialised,
"tables_exist": len(table_names) > 0,
"table_count": len(table_names),
"tables": table_names,
@@ -238,6 +293,7 @@ async def setup_status():
async def init_db(body: PasswordBody):
"""Create all database tables."""
_check_password(body.password)
await _require_not_initialised()
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
return {"status": "ok", "message": "Database tables created successfully"}
@@ -274,6 +330,7 @@ async def create_admin(body: CreateAdminBody):
async def seed_demo_data(body: PasswordBody):
"""Seed database with demo users and a sample recipe."""
_check_password(body.password)
await _require_not_initialised()
async with async_session_factory() as session:
async with session.begin():
@@ -544,6 +601,7 @@ async def list_users(body: PasswordBody):
async def manage_user(body: ManageUserBody):
"""Create or update a user. user_id=null creates new, user_id=int updates."""
_check_password(body.password)
await _require_not_initialised()
invalid = set(body.roles) - VALID_ROLES
if invalid:
@@ -618,6 +676,7 @@ async def manage_user(body: ManageUserBody):
async def change_user_password(body: ChangeUserPasswordBody):
"""Change a user's password and invalidate their API key."""
_check_password(body.password)
await _require_not_initialised()
async with async_session_factory() as session:
async with session.begin():
@@ -664,6 +723,7 @@ async def toggle_user_active(body: ToggleUserActiveBody):
async def reset_db(body: ResetDbBody):
"""Drop all tables and recreate them. Requires confirm=true."""
_check_password(body.password)
await _require_not_initialised()
if not body.confirm:
raise HTTPException(
+4
View File
@@ -43,6 +43,10 @@ class Settings(BaseSettings):
# Setup page (empty = disabled)
setup_password: str | None = None
# By default, once an admin user exists, setup's privileged endpoints
# (reset-db, seed, manage-user, change-user-password, init-db) refuse to
# run again. Set true for scripted re-provisioning of a demo/test deploy.
setup_allow_after_init: bool = False
# AI / OpenRouter (for technical sheet parsing)
openrouter_api_key: str | None = None
+164 -3
View File
@@ -157,9 +157,12 @@ async def test_create_user(client: AsyncClient):
@pytest.mark.asyncio
async def test_create_user_duplicate_username(client: AsyncClient):
async def test_create_user_duplicate_username(client: AsyncClient, monkeypatch):
"""Creating a user with a duplicate username should return 409."""
await _seed(client)
# _seed already created an admin; bypass the "setup already completed"
# guard (#9) so this test can still exercise manage-user's own checks.
monkeypatch.setattr(settings, "setup_allow_after_init", True)
resp = await client.post("/api/setup/manage-user", json={
"password": SETUP_PWD,
@@ -188,9 +191,10 @@ async def test_create_user_without_password(client: AsyncClient):
@pytest.mark.asyncio
async def test_update_user(client: AsyncClient):
async def test_update_user(client: AsyncClient, monkeypatch):
"""Update an existing user via manage-user endpoint."""
await _seed(client)
monkeypatch.setattr(settings, "setup_allow_after_init", True)
# Get users to find the ID
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
@@ -232,9 +236,10 @@ async def test_create_user_invalid_role(client: AsyncClient):
@pytest.mark.asyncio
async def test_change_password(client: AsyncClient):
async def test_change_password(client: AsyncClient, monkeypatch):
"""Changing password should succeed and invalidate api_key."""
await _seed(client)
monkeypatch.setattr(settings, "setup_allow_after_init", True)
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
users = resp.json()["users"]
@@ -292,3 +297,159 @@ async def test_toggle_user_active(client: AsyncClient):
})
assert resp.status_code == 200
assert resp.json()["active"] is True
# ---------------------------------------------------------------------------
# Feature 3: _check_password is timing-safe
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_wrong_password_still_rejected_with_compare_digest(client: AsyncClient):
"""hmac.compare_digest must not change the pass/fail outcome."""
resp = await client.post("/api/setup/list-users", json={"password": "definitely-wrong"})
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_correct_password_still_accepted_with_compare_digest(client: AsyncClient):
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 200
# ---------------------------------------------------------------------------
# Feature 4: setup already completed - privileged endpoints blocked (#9)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_privileged_endpoints_blocked_once_admin_exists(client: AsyncClient, monkeypatch):
"""reset-db, seed, manage-user, change-user-password, init-db all 403
once an admin exists, unless setup_allow_after_init is set."""
# This test makes more than rate_limit_login (5/min) requests to
# /api/setup/* on purpose, to exercise every guarded endpoint - raise the
# limit so it is the #9 guard being tested, not the #8/#9(d) rate limit.
monkeypatch.setattr(settings, "rate_limit_login", 100)
data = await _seed(client) # creates the "admin" user (is_admin=True)
admin_id = None
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
for u in resp.json()["users"]:
if u["username"] == "admin":
admin_id = u["id"]
assert admin_id is not None
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 403
assert "already completed" in resp.json()["detail"]
resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD})
assert resp.status_code == 403
resp = await client.post("/api/setup/manage-user", json={
"password": SETUP_PWD,
"user_id": None,
"username": "blocked",
"user_password": "pass123",
"display_name": "Blocked",
"roles": [],
})
assert resp.status_code == 403
resp = await client.post("/api/setup/change-user-password", json={
"password": SETUP_PWD,
"user_id": admin_id,
"new_password": "whatever123",
})
assert resp.status_code == 403
resp = await client.post("/api/setup/reset-db", json={
"password": SETUP_PWD,
"confirm": True,
})
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_privileged_endpoints_allowed_before_any_admin(client: AsyncClient):
"""On a fresh setup (no admin yet), init-db and seed still work."""
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 200
resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD})
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_setup_allow_after_init_overrides_guard(client: AsyncClient, monkeypatch):
"""SETUP_ALLOW_AFTER_INIT=true restores the old (pre-#9) behaviour."""
await _seed(client)
monkeypatch.setattr(settings, "setup_allow_after_init", True)
resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD})
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_toggle_user_active_not_guarded(client: AsyncClient):
"""toggle-user-active is not in the guarded list, admin or not."""
await _seed(client)
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
maker = next(u for u in resp.json()["users"] if u["username"] == "maker1")
resp = await client.post("/api/setup/toggle-user-active", json={
"password": SETUP_PWD,
"user_id": maker["id"],
})
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_create_admin_not_guarded(client: AsyncClient):
"""create-admin is not in the guarded list either (see analysis of #9)."""
await _seed(client)
resp = await client.post("/api/setup/create-admin", json={
"password": SETUP_PWD,
"admin_username": "second_admin",
"admin_password": "pass12345",
"admin_email": "second@test.com",
"admin_display_name": "Second Admin",
})
assert resp.status_code == 200
# ---------------------------------------------------------------------------
# Feature 5: GET /api/setup/status without/with password (#9)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_status_without_password_is_minimal(client: AsyncClient):
resp = await client.get("/api/setup/status")
assert resp.status_code == 200
data = resp.json()
assert data == {"initialised": False}
@pytest.mark.asyncio
async def test_status_with_wrong_password_is_minimal(client: AsyncClient):
resp = await client.get("/api/setup/status", params={"password": "nope"})
assert resp.status_code == 200
assert resp.json() == {"initialised": False}
@pytest.mark.asyncio
async def test_status_with_valid_password_lists_tables(client: AsyncClient):
await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
resp = await client.get("/api/setup/status", params={"password": SETUP_PWD})
assert resp.status_code == 200
data = resp.json()
assert "tables" in data
assert "table_count" in data
assert data["tables_exist"] is True
@pytest.mark.asyncio
async def test_status_initialised_true_after_seed(client: AsyncClient):
await _seed(client)
resp = await client.get("/api/setup/status")
assert resp.status_code == 200
assert resp.json() == {"initialised": True}
+6
View File
@@ -83,11 +83,17 @@ async def test_setup_seed_creates_default_station_and_assigns_recipes(
async def test_setup_seed_station_idempotent(
client: AsyncClient,
db_session: AsyncSession,
monkeypatch,
):
"""Running seed twice must not duplicate ST-DEFAULT or its assignments."""
# First run — creates everything.
await _seed(client)
# Second run: the first seed already created an admin, so this exercises
# setup's own "setup already completed" guard (#9) unless bypassed - the
# idempotency this test is about is seed's, not the guard's.
monkeypatch.setattr(settings, "setup_allow_after_init", True)
# Second run — recipe DEMO-001 already exists → seed returns early.
# ST-DEFAULT must still exist (not re-created, not duplicated).
resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD})