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. Protected by SETUP_PASSWORD env var. When empty/None, all endpoints return 404.
NOT protected by API Key auth (standalone access). NOT protected by API Key auth (standalone access).
""" """
import hmac
import random import random
import secrets import secrets
from datetime import datetime, timedelta from datetime import datetime, timedelta
@@ -87,13 +88,57 @@ def _check_setup_enabled() -> None:
def _check_password(password: str) -> None: def _check_password(password: str) -> None:
"""Verify the provided password matches the setup password.""" """Verify the provided password matches the setup password."""
_check_setup_enabled() _check_setup_enabled()
if password != settings.setup_password: if not hmac.compare_digest(password, settings.setup_password):
raise HTTPException( raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid setup password", 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( def _generate_measurements_for_subtask(
subtask: RecipeSubtask, subtask: RecipeSubtask,
version_id: int, version_id: int,
@@ -220,14 +265,24 @@ async def setup_page(request: Request):
@router.get("/status") @router.get("/status")
async def setup_status(): async def setup_status(password: str | None = None):
"""Return database table status (no password required, but setup must be enabled).""" """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() _check_setup_enabled()
initialised = await _admin_exists()
if not _password_matches(password):
return {"initialised": initialised}
async with engine.connect() as conn: async with engine.connect() as conn:
table_names: list[str] = await conn.run_sync( table_names: list[str] = await conn.run_sync(
lambda sync_conn: sa_inspect(sync_conn).get_table_names() lambda sync_conn: sa_inspect(sync_conn).get_table_names()
) )
return { return {
"initialised": initialised,
"tables_exist": len(table_names) > 0, "tables_exist": len(table_names) > 0,
"table_count": len(table_names), "table_count": len(table_names),
"tables": table_names, "tables": table_names,
@@ -238,6 +293,7 @@ async def setup_status():
async def init_db(body: PasswordBody): async def init_db(body: PasswordBody):
"""Create all database tables.""" """Create all database tables."""
_check_password(body.password) _check_password(body.password)
await _require_not_initialised()
async with engine.begin() as conn: async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) await conn.run_sync(Base.metadata.create_all)
return {"status": "ok", "message": "Database tables created successfully"} 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): async def seed_demo_data(body: PasswordBody):
"""Seed database with demo users and a sample recipe.""" """Seed database with demo users and a sample recipe."""
_check_password(body.password) _check_password(body.password)
await _require_not_initialised()
async with async_session_factory() as session: async with async_session_factory() as session:
async with session.begin(): async with session.begin():
@@ -544,6 +601,7 @@ async def list_users(body: PasswordBody):
async def manage_user(body: ManageUserBody): async def manage_user(body: ManageUserBody):
"""Create or update a user. user_id=null creates new, user_id=int updates.""" """Create or update a user. user_id=null creates new, user_id=int updates."""
_check_password(body.password) _check_password(body.password)
await _require_not_initialised()
invalid = set(body.roles) - VALID_ROLES invalid = set(body.roles) - VALID_ROLES
if invalid: if invalid:
@@ -618,6 +676,7 @@ async def manage_user(body: ManageUserBody):
async def change_user_password(body: ChangeUserPasswordBody): async def change_user_password(body: ChangeUserPasswordBody):
"""Change a user's password and invalidate their API key.""" """Change a user's password and invalidate their API key."""
_check_password(body.password) _check_password(body.password)
await _require_not_initialised()
async with async_session_factory() as session: async with async_session_factory() as session:
async with session.begin(): async with session.begin():
@@ -664,6 +723,7 @@ async def toggle_user_active(body: ToggleUserActiveBody):
async def reset_db(body: ResetDbBody): async def reset_db(body: ResetDbBody):
"""Drop all tables and recreate them. Requires confirm=true.""" """Drop all tables and recreate them. Requires confirm=true."""
_check_password(body.password) _check_password(body.password)
await _require_not_initialised()
if not body.confirm: if not body.confirm:
raise HTTPException( raise HTTPException(
+4
View File
@@ -43,6 +43,10 @@ class Settings(BaseSettings):
# Setup page (empty = disabled) # Setup page (empty = disabled)
setup_password: str | None = None 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) # AI / OpenRouter (for technical sheet parsing)
openrouter_api_key: str | None = None openrouter_api_key: str | None = None
+164 -3
View File
@@ -157,9 +157,12 @@ async def test_create_user(client: AsyncClient):
@pytest.mark.asyncio @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.""" """Creating a user with a duplicate username should return 409."""
await _seed(client) 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={ resp = await client.post("/api/setup/manage-user", json={
"password": SETUP_PWD, "password": SETUP_PWD,
@@ -188,9 +191,10 @@ async def test_create_user_without_password(client: AsyncClient):
@pytest.mark.asyncio @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.""" """Update an existing user via manage-user endpoint."""
await _seed(client) await _seed(client)
monkeypatch.setattr(settings, "setup_allow_after_init", True)
# Get users to find the ID # Get users to find the ID
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD}) 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 @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.""" """Changing password should succeed and invalidate api_key."""
await _seed(client) await _seed(client)
monkeypatch.setattr(settings, "setup_allow_after_init", True)
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD}) resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
users = resp.json()["users"] users = resp.json()["users"]
@@ -292,3 +297,159 @@ async def test_toggle_user_active(client: AsyncClient):
}) })
assert resp.status_code == 200 assert resp.status_code == 200
assert resp.json()["active"] is True 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( async def test_setup_seed_station_idempotent(
client: AsyncClient, client: AsyncClient,
db_session: AsyncSession, db_session: AsyncSession,
monkeypatch,
): ):
"""Running seed twice must not duplicate ST-DEFAULT or its assignments.""" """Running seed twice must not duplicate ST-DEFAULT or its assignments."""
# First run — creates everything. # First run — creates everything.
await _seed(client) 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. # Second run — recipe DEMO-001 already exists → seed returns early.
# ST-DEFAULT must still exist (not re-created, not duplicated). # ST-DEFAULT must still exist (not re-created, not duplicated).
resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD}) resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD})