lukas.pupkalipinski/BanGUI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use session_secret for signed auth session tokens

Browse Source
This commit is contained in:
Lukas
2026-04-09 21:30:08 +02:00
parent 6eab47f7ba
commit 208f98dc97
8 changed files with 136 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Docs/Tasks.md
View File

@@ -20,6 +20,7 @@ Reference: `Docs/Refactoring.md` for full analysis of each issue.
### 2. Remove or use `session_secret`
- Where found: `backend/app/config.py`
- Goal: either eliminate the unused `BANGUI_SESSION_SECRET` requirement or use it for session token generation / signing so the setting has purpose.
- Status: completed
- Possible traps and issues:
- Keeping it required without use is misleading and burdens deployments.
- Introducing a new crypto dependency for session tokens must preserve backward compatibility with existing sessions.

4
backend/app/dependencies.py
View File

@@ -206,6 +206,7 @@ async def get_pending_recovery(request: Request) -> PendingRecovery | None:
async def require_auth(
request: Request,
db: Annotated[aiosqlite.Connection, Depends(get_db)],
settings: Annotated[Settings, Depends(get_settings)],
) -> Session:
"""Validate the session token and return the active session.
@@ -220,6 +221,7 @@ async def require_auth(
Args:
request: The incoming FastAPI request.
db: Injected aiosqlite connection.
settings: Application settings used for signed session token validation.
Returns:
The active :class:`~app.models.auth.Session`.
@@ -253,7 +255,7 @@ async def require_auth(
_session_cache.pop(token, None)
try:
session = await auth_service.validate_session(db, token)
session = await auth_service.validate_session(db, token, settings.session_secret)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,

16
backend/app/routers/auth.py
View File

@@ -63,15 +63,19 @@ async def login(
detail=str(exc),
) from exc
signed_token = auth_service.sign_session_token(
session.token,
settings.session_secret,
)
response.set_cookie(
key=_COOKIE_NAME,
value=session.token,
value=signed_token,
httponly=True,
samesite="lax",
secure=False, # Set to True in production behind HTTPS
max_age=settings.session_duration_minutes * 60,
)
return LoginResponse(token=session.token, expires_at=session.expires_at)
return LoginResponse(token=signed_token, expires_at=session.expires_at)
@router.post(
@@ -83,6 +87,7 @@ async def logout(
request: Request,
response: Response,
db: DbDep,
settings: SettingsDep,
) -> LogoutResponse:
"""Invalidate the active session.
@@ -94,14 +99,17 @@ async def logout(
request: FastAPI request (used to extract the token).
response: FastAPI response (used to clear the cookie).
db: Injected aiosqlite connection.
settings: Application settings (used to unwrap signed tokens).
Returns:
:class:`~app.models.auth.LogoutResponse`.
"""
token = _extract_token(request)
if token:
await auth_service.logout(db, token)
invalidate_session_cache(token)
raw_token = await auth_service.logout(db, token, settings.session_secret)
if raw_token:
invalidate_session_cache(raw_token)
invalidate_session_cache(token)
response.delete_cookie(key=_COOKIE_NAME)
return LogoutResponse()

64
backend/app/services/auth_service.py
View File

@@ -8,6 +8,8 @@ survive server restarts.
from __future__ import annotations
import asyncio
import hashlib
import hmac
import secrets
from typing import TYPE_CHECKING
@@ -20,12 +22,39 @@ if TYPE_CHECKING:
from app.models.auth import Session
from app.repositories import session_repo
from app.utils.constants import SESSION_TOKEN_BYTES, SESSION_TOKEN_SIGNATURE_SEPARATOR
from app.utils.setup_utils import get_password_hash
from app.utils.time_utils import add_minutes, utc_now
log: structlog.stdlib.BoundLogger = structlog.get_logger()
def _session_token_signature(token: str, secret: str) -> str:
"""Return the HMAC-SHA256 signature for a session token."""
return hmac.new(secret.encode(), token.encode(), hashlib.sha256).hexdigest()
def sign_session_token(token: str, secret: str) -> str:
"""Return a signed session token string for the client."""
return f"{token}{SESSION_TOKEN_SIGNATURE_SEPARATOR}{_session_token_signature(token, secret)}"
def unwrap_session_token(token: str, secret: str) -> str:
"""Verify and return the raw token from a signed session token.
If the token has no signature component, it is returned unchanged. This
preserves compatibility with existing raw session tokens stored in the DB.
"""
if SESSION_TOKEN_SIGNATURE_SEPARATOR not in token:
return token
raw_token, signature = token.rsplit(SESSION_TOKEN_SIGNATURE_SEPARATOR, 1)
expected_signature = _session_token_signature(raw_token, secret)
if not hmac.compare_digest(expected_signature, signature):
raise ValueError("Invalid session token.")
return raw_token
async def _check_password(plain: str, hashed: str) -> bool:
"""Return ``True`` if *plain* matches the bcrypt *hashed* password.
@@ -74,7 +103,7 @@ async def login(
log.warning("bangui_login_wrong_password")
raise ValueError("Incorrect password.")
token = secrets.token_hex(32)
token = secrets.token_hex(SESSION_TOKEN_BYTES)
now = utc_now()
created_iso = now.isoformat()
expires_iso = add_minutes(now, session_duration_minutes).isoformat()
@@ -86,19 +115,30 @@ async def login(
return session
async def validate_session(db: aiosqlite.Connection, token: str) -> Session:
async def validate_session(
db: aiosqlite.Connection,
token: str,
session_secret: str | None = None,
) -> Session:
"""Return the session for *token* if it is valid and not expired.
Args:
db: Active aiosqlite connection.
token: The opaque session token from the client.
session_secret: Secret used to verify signed tokens.
Returns:
The :class:`~app.models.auth.Session` if it is valid.
Raises:
ValueError: If the token is not found or has expired.
ValueError: If the token is not found, invalid, or has expired.
"""
if session_secret is not None:
try:
token = unwrap_session_token(token, session_secret)
except ValueError as exc:
raise ValueError("Session token is invalid.") from exc
session = await session_repo.get_session(db, token)
if session is None:
raise ValueError("Session not found.")
@@ -111,12 +151,28 @@ async def validate_session(db: aiosqlite.Connection, token: str) -> Session:
return session
async def logout(db: aiosqlite.Connection, token: str) -> None:
async def logout(
db: aiosqlite.Connection,
token: str,
session_secret: str | None = None,
) -> str | None:
"""Invalidate the session identified by *token*.
Args:
db: Active aiosqlite connection.
token: The session token to revoke.
session_secret: Secret used to verify signed tokens.
Returns:
The raw session token that was revoked, or ``None`` if the token was invalid.
"""
if session_secret is not None:
try:
token = unwrap_session_token(token, session_secret)
except ValueError:
log.warning("bangui_logout_invalid_token", token_prefix=token[:8])
return None
await session_repo.delete_session(db, token)
log.info("bangui_logout", token_prefix=token[:8])
return token

21
backend/app/services/setup_service.py
View File

@@ -100,8 +100,25 @@ async def run_setup(
await _ensure_database_initialized(database_path)
# Mark setup as complete — must be last so a partial failure leaves
# setup_completed unset and does not lock out the user.
runtime_db: aiosqlite.Connection | None = None
try:
runtime_db = await open_db(database_path)
await settings_repo.set_setting(runtime_db, _KEY_PASSWORD_HASH, hashed)
await settings_repo.set_setting(runtime_db, _KEY_DATABASE_PATH, database_path)
await settings_repo.set_setting(runtime_db, _KEY_FAIL2BAN_SOCKET, fail2ban_socket)
await settings_repo.set_setting(runtime_db, _KEY_TIMEZONE, timezone)
await settings_repo.set_setting(
runtime_db, _KEY_SESSION_DURATION, str(session_duration_minutes)
)
await settings_repo.set_setting(runtime_db, _KEY_MAP_COLOR_THRESHOLD_HIGH, "100")
await settings_repo.set_setting(runtime_db, _KEY_MAP_COLOR_THRESHOLD_MEDIUM, "50")
await settings_repo.set_setting(runtime_db, _KEY_MAP_COLOR_THRESHOLD_LOW, "20")
await settings_repo.set_setting(runtime_db, _KEY_SETUP_DONE, "1")
finally:
if runtime_db is not None:
await runtime_db.close()
# Mark setup as complete in the bootstrap configuration as the final step.
await settings_repo.set_setting(db, _KEY_SETUP_DONE, "1")
log.info("bangui_setup_completed")

5
backend/app/utils/constants.py
View File

@@ -30,9 +30,12 @@ DEFAULT_DATABASE_PATH: Final[str] = "bangui.db"
DEFAULT_SESSION_DURATION_MINUTES: Final[int] = 60
"""Default session lifetime in minutes."""
SESSION_TOKEN_BYTES: Final[int] = 64
SESSION_TOKEN_BYTES: Final[int] = 32
"""Number of random bytes used when generating a session token."""
SESSION_TOKEN_SIGNATURE_SEPARATOR: Final[str] = "."
"""Separator used to append a signature to a signed session token."""
# ---------------------------------------------------------------------------
# Time-range presets (used by dashboard and history endpoints)
# ---------------------------------------------------------------------------

2
backend/tests/test_routers/test_auth.py
View File

@@ -54,6 +54,7 @@ class TestLogin:
body = response.json()
assert "token" in body
assert len(body["token"]) > 0
assert "." in body["token"]
assert "expires_at" in body
async def test_login_sets_cookie(self, client: AsyncClient) -> None:
@@ -64,6 +65,7 @@ class TestLogin:
)
assert response.status_code == 200
assert "bangui_session" in response.cookies
assert "." in response.cookies["bangui_session"]
async def test_login_fails_with_wrong_password(
self, client: AsyncClient

35
backend/tests/test_services/test_auth_service.py
View File

@@ -119,6 +119,30 @@ class TestValidateSession:
validated = await auth_service.validate_session(db, session.token)
assert validated.token == session.token
async def test_validate_accepts_signed_token(
self, db: aiosqlite.Connection
) -> None:
"""validate_session() accepts a token signed with the configured secret."""
session = await auth_service.login(db, password="correctpassword1", session_duration_minutes=60)
signed_token = auth_service.sign_session_token(session.token, "test-secret")
validated = await auth_service.validate_session(
db, signed_token, session_secret="test-secret"
)
assert validated.token == session.token
async def test_validate_rejects_tampered_signed_token(
self, db: aiosqlite.Connection
) -> None:
"""validate_session() rejects signed tokens with an invalid signature."""
session = await auth_service.login(db, password="correctpassword1", session_duration_minutes=60)
signed_token = auth_service.sign_session_token(session.token, "test-secret")
tampered_token = signed_token[:-1] + ("0" if signed_token[-1] != "0" else "1")
with pytest.raises(ValueError, match="invalid"):
await auth_service.validate_session(
db, tampered_token, session_secret="test-secret"
)
async def test_validate_raises_for_unknown_token(
self, db: aiosqlite.Connection
) -> None:
@@ -157,3 +181,14 @@ class TestLogout:
await auth_service.logout(db, session.token)
stored = await session_repo.get_session(db, session.token)
assert stored is None
async def test_logout_accepts_signed_token(self, db: aiosqlite.Connection) -> None:
"""logout() accepts a signed token and revokes the underlying raw session."""
from app.repositories import session_repo
session = await auth_service.login(db, password="correctpassword1", session_duration_minutes=60)
signed_token = auth_service.sign_session_token(session.token, "test-secret")
await auth_service.logout(db, signed_token, session_secret="test-secret")
stored = await session_repo.get_session(db, session.token)
assert stored is None