fix(auth): api_key condivisa fra postazioni, timing costante al login (#25)

login_user riusa l'api_key esistente invece di rigenerarla ad ogni
login: due postazioni collegate con lo stesso utente restavano
altrimenti disconnesse a vicenda. logout_user diventa no-op sulla key
a meno di all_sessions=True (nuovo parametro, passato da
POST /api/auth/logout?all_sessions=true nel router).

authenticate_user esegue comunque bcrypt.checkpw contro un hash fittizio
quando lo username non esiste, per non rivelare col tempo di risposta
se uno username è registrato.

PUT /api/users/{id}/password azzera l'api_key dell'utente dopo il
cambio, come già fa setup.py:634.

Non introdotto in questa issue: hashing o scadenza della api_key
(annotato come MANCA nel rapporto del lotto).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-21 22:40:05 +02:00
parent 302903a8a7
commit ac76c0a395
5 changed files with 207 additions and 8 deletions
+6 -2
View File
@@ -34,11 +34,15 @@ async def login(data: LoginRequest, db: AsyncSession = Depends(get_db)):
@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT) @router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
async def logout( async def logout(
all_sessions: bool = False,
user: User = Depends(get_current_user), user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
): ):
"""Logout - invalidate API key.""" """Logout. The API key is shared across stations logged in as the same
await logout_user(db, user) user, so by default this does not invalidate it - only this station's
session (its Flask session, on the client side) ends. Pass
?all_sessions=true to invalidate the key everywhere."""
await logout_user(db, user, all_sessions=all_sessions)
@router.get("/me", response_model=UserResponse) @router.get("/me", response_model=UserResponse)
+6 -1
View File
@@ -100,12 +100,17 @@ async def change_user_password(
admin: User = Depends(require_admin_user), admin: User = Depends(require_admin_user),
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
): ):
"""Change user password (admin only).""" """Change user password (admin only). Invalidates the user's API key,
the same as setup.py's change-user-password: a password change should
force everyone logged in as that user to log back in.
"""
result = await db.execute(select(User).where(User.id == user_id)) result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none() user = result.scalar_one_or_none()
if user is None: if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
user.password_hash = hash_password(data.password) user.password_hash = hash_password(data.password)
user.api_key = None
await db.flush() await db.flush()
return {"message": f"Password changed for user {user.username}"} return {"message": f"Password changed for user {user.username}"}
+30 -5
View File
@@ -27,6 +27,13 @@ def generate_api_key() -> str:
return secrets.token_urlsafe(48) # 64 chars base64 return secrets.token_urlsafe(48) # 64 chars base64
# A precomputed, unused bcrypt hash. When the username does not match any
# user, authenticate_user still runs bcrypt against this instead of
# returning immediately, so "unknown username" costs the same as "wrong
# password" and a caller cannot use response time to enumerate usernames.
_DUMMY_PASSWORD_HASH = "$2b$12$BLtwrR2COyd5W/UWfxp7BOc7OHua9jazo6zZdWOwRD9PV.pyaqKZC"
async def authenticate_user( async def authenticate_user(
db: AsyncSession, username: str, password: str db: AsyncSession, username: str, password: str
) -> User | None: ) -> User | None:
@@ -35,7 +42,10 @@ async def authenticate_user(
select(User).where(User.username == username, User.active == True) select(User).where(User.username == username, User.active == True)
) )
user = result.scalar_one_or_none() user = result.scalar_one_or_none()
if user is None or not verify_password(password, user.password_hash): if user is None:
verify_password(password, _DUMMY_PASSWORD_HASH)
return None
if not verify_password(password, user.password_hash):
return None return None
return user return user
@@ -66,8 +76,14 @@ async def authorise_supervisor(db: AsyncSession, username: str, password: str) -
async def login_user(db: AsyncSession, user: User) -> str: async def login_user(db: AsyncSession, user: User) -> str:
"""Generate API key and update last_login for user.""" """Generate (or reuse) the API key and update last_login for user.
api_key = generate_api_key()
Reuses an existing key rather than rotating it on every login: the same
account logged in from two stations at once (a common shift-account
pattern here) must keep both sessions working. Rotating the key would
silently log the other station out.
"""
api_key = user.api_key or generate_api_key()
await db.execute( await db.execute(
update(User) update(User)
.where(User.id == user.id) .where(User.id == user.id)
@@ -77,8 +93,17 @@ async def login_user(db: AsyncSession, user: User) -> str:
return api_key return api_key
async def logout_user(db: AsyncSession, user: User) -> None: async def logout_user(
"""Invalidate user's API key.""" db: AsyncSession, user: User, all_sessions: bool = False
) -> None:
"""Log out. By default a no-op on the shared API key.
Clearing the key on every logout would kick out any other station still
using the same shift account. `all_sessions=True` is the explicit
override for a supervisor who wants every session logged out.
"""
if not all_sessions:
return
await db.execute( await db.execute(
update(User).where(User.id == user.id).values(api_key=None) update(User).where(User.id == user.id).values(api_key=None)
) )
+91
View File
@@ -78,6 +78,97 @@ class TestLogin:
assert len(api_key) >= 32 assert len(api_key) >= 32
class TestLoginKeyReuse:
"""login_user reuses an existing api_key instead of rotating it (#25)."""
async def test_two_logins_same_user_share_api_key(
self, client: AsyncClient, db_session
):
"""Two stations logging in as the same shift account both keep working."""
await _create_user(db_session, username="twostations", password="TwoStations1")
resp1 = await client.post(
"/api/auth/login",
json={"username": "twostations", "password": "TwoStations1"},
)
resp2 = await client.post(
"/api/auth/login",
json={"username": "twostations", "password": "TwoStations1"},
)
assert resp1.status_code == 200
assert resp2.status_code == 200
assert resp1.json()["api_key"] == resp2.json()["api_key"]
async def test_login_generates_key_when_user_has_none(
self, client: AsyncClient, db_session
):
"""A user created without an api_key (e.g. via create_user) still gets one."""
from src.backend.services.auth_service import create_user
await create_user(
db_session, username="brandnew", password="BrandNew1",
display_name="Brand New",
)
resp = await client.post(
"/api/auth/login",
json={"username": "brandnew", "password": "BrandNew1"},
)
assert resp.status_code == 200
assert resp.json()["api_key"]
class TestLogout:
"""POST /api/auth/logout is a no-op on the shared api_key by default (#25)."""
async def test_logout_default_keeps_api_key_valid(
self, client: AsyncClient, db_session
):
user = await _create_user(
db_session, username="shared_logout", password="SharedPass1"
)
headers = auth_headers(user)
resp = await client.post("/api/auth/logout", headers=headers)
assert resp.status_code == 204
# The key is still valid: another station on the same account keeps working.
resp = await client.get("/api/auth/me", headers=headers)
assert resp.status_code == 200
async def test_logout_all_sessions_invalidates_api_key(
self, client: AsyncClient, db_session
):
user = await _create_user(
db_session, username="shared_logout_all", password="SharedPass1"
)
headers = auth_headers(user)
resp = await client.post(
"/api/auth/logout", params={"all_sessions": "true"}, headers=headers
)
assert resp.status_code == 204
resp = await client.get("/api/auth/me", headers=headers)
assert resp.status_code == 401
class TestAuthenticateUserTiming:
"""authenticate_user pays the bcrypt cost even for an unknown username (#25)."""
async def test_unknown_username_still_returns_none(self, db_session):
from src.backend.services.auth_service import authenticate_user
result = await authenticate_user(db_session, "no-such-user-at-all", "whatever")
assert result is None
def test_dummy_hash_is_a_valid_bcrypt_hash(self):
from src.backend.services.auth_service import (
_DUMMY_PASSWORD_HASH,
verify_password,
)
assert verify_password(
"tiemeasureflow-dummy-hash-for-timing", _DUMMY_PASSWORD_HASH
)
class TestHealthAndAuth: class TestHealthAndAuth:
"""Health check and auth requirement tests.""" """Health check and auth requirement tests."""
@@ -0,0 +1,74 @@
"""Tests for PUT /api/users/{id}/password (admin only) - issue #25(c).
Kept in its own file rather than test_users.py: that file covers the rest of
the users router, worked on in a different lot in the same session.
"""
from httpx import AsyncClient
from src.backend.models.orm.user import User
from src.backend.tests.conftest import auth_headers
class TestChangeUserPassword:
async def test_admin_changes_password_invalidates_api_key(
self, client: AsyncClient, admin_user: User, maker_user: User
):
"""Changing a user's password invalidates their existing api_key,
the same as setup.py's change-user-password (setup.py:634).
"""
old_key = maker_user.api_key
old_username = maker_user.username
assert old_key is not None
resp = await client.put(
f"/api/users/{maker_user.id}/password",
headers=auth_headers(admin_user),
json={"password": "NewSecurePass1"},
)
assert resp.status_code == 200
assert "changed" in resp.json()["message"].lower()
# The old key no longer authenticates. Read it before the request
# above: the client and app share one session (see conftest), so
# `maker_user.api_key` in this process reflects the mutation too.
resp2 = await client.get("/api/auth/me", headers={"X-API-Key": old_key})
assert resp2.status_code == 401
# The new password works and issues a fresh key.
resp3 = await client.post(
"/api/auth/login",
json={"username": old_username, "password": "NewSecurePass1"},
)
assert resp3.status_code == 200
assert resp3.json()["api_key"] != old_key
async def test_non_admin_forbidden(
self, client: AsyncClient, maker_user: User
):
resp = await client.put(
f"/api/users/{maker_user.id}/password",
headers=auth_headers(maker_user),
json={"password": "WhateverPass1"},
)
assert resp.status_code == 403
async def test_user_not_found(
self, client: AsyncClient, admin_user: User
):
resp = await client.put(
"/api/users/999999/password",
headers=auth_headers(admin_user),
json={"password": "WhateverPass1"},
)
assert resp.status_code == 404
async def test_password_too_short_rejected(
self, client: AsyncClient, admin_user: User, maker_user: User
):
resp = await client.put(
f"/api/users/{maker_user.id}/password",
headers=auth_headers(admin_user),
json={"password": "short"},
)
assert resp.status_code == 422