feat: Implement session secret rotation support
Adds support for gradual session secret rotation without forcing logout: - Add BANGUI_SESSION_SECRET_PREVIOUS config field for rotation window - Implement unwrap_session_token_with_rotation() to accept tokens signed with either current or previous secret - Update validate_session() to transparently accept old tokens during rotation - Update logout() to accept tokens from both secrets - Add comprehensive logging for rotation events and metrics - Add 8 new tests covering all rotation scenarios - Update documentation with step-by-step rotation strategy - Update .env.example with previous secret field Key features: - No forced logout: old tokens continue working during rotation window - Transparent validation: old tokens are automatically logged for monitoring - Production-safe: can rotate secrets without service interruption - Metrics-ready: logs track token rotation for observability Rotation workflow: 1. Generate new secret and set BANGUI_SESSION_SECRET 2. Set BANGUI_SESSION_SECRET_PREVIOUS to old secret 3. Wait for old tokens to expire (≥ session_duration_minutes) 4. Unset BANGUI_SESSION_SECRET_PREVIOUS to complete rotation Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -239,3 +239,146 @@ class TestLogout:
|
||||
|
||||
stored = await session_repo.get_session(db, raw_token)
|
||||
assert stored is None
|
||||
|
||||
|
||||
class TestSecretRotation:
|
||||
"""Tests for session secret rotation support."""
|
||||
|
||||
async def test_unwrap_with_rotation_accepts_current_secret(
|
||||
self, db: aiosqlite.Connection
|
||||
) -> None:
|
||||
"""Tokens signed with current secret are validated immediately."""
|
||||
signed_token, _ = await auth_service.login(
|
||||
db,
|
||||
password="correctpassword1",
|
||||
session_duration_minutes=60,
|
||||
session_secret="current-secret",
|
||||
)
|
||||
raw_token, was_re_signed = auth_service.unwrap_session_token_with_rotation(
|
||||
signed_token, "current-secret", None
|
||||
)
|
||||
assert raw_token == auth_service.unwrap_session_token(signed_token, "current-secret")
|
||||
assert was_re_signed is False
|
||||
|
||||
async def test_unwrap_with_rotation_accepts_previous_secret(
|
||||
self, db: aiosqlite.Connection
|
||||
) -> None:
|
||||
"""Tokens signed with previous secret are accepted during rotation."""
|
||||
signed_token, _ = await auth_service.login(
|
||||
db,
|
||||
password="correctpassword1",
|
||||
session_duration_minutes=60,
|
||||
session_secret="old-secret",
|
||||
)
|
||||
raw_token, was_re_signed = auth_service.unwrap_session_token_with_rotation(
|
||||
signed_token, "new-secret", "old-secret"
|
||||
)
|
||||
assert raw_token == auth_service.unwrap_session_token(signed_token, "old-secret")
|
||||
assert was_re_signed is True
|
||||
|
||||
async def test_unwrap_with_rotation_rejects_unknown_secret(
|
||||
self, db: aiosqlite.Connection
|
||||
) -> None:
|
||||
"""Tokens signed with unknown secrets are rejected."""
|
||||
signed_token, _ = await auth_service.login(
|
||||
db,
|
||||
password="correctpassword1",
|
||||
session_duration_minutes=60,
|
||||
session_secret="secret-a",
|
||||
)
|
||||
with pytest.raises(ValueError, match="Invalid session token"):
|
||||
auth_service.unwrap_session_token_with_rotation(
|
||||
signed_token, "secret-b", "secret-c"
|
||||
)
|
||||
|
||||
async def test_unwrap_with_rotation_prefers_current_secret(
|
||||
self, db: aiosqlite.Connection
|
||||
) -> None:
|
||||
"""If both secrets are valid, current secret is preferred."""
|
||||
token_hex = "deadbeef" * 8
|
||||
current_sig = auth_service._session_token_signature(token_hex, "same-secret")
|
||||
signed_token = f"{token_hex}.{current_sig}"
|
||||
raw_token, was_re_signed = auth_service.unwrap_session_token_with_rotation(
|
||||
signed_token, "same-secret", "same-secret"
|
||||
)
|
||||
assert raw_token == token_hex
|
||||
assert was_re_signed is False
|
||||
|
||||
async def test_validate_session_re_signs_token_with_previous_secret(
|
||||
self, db: aiosqlite.Connection
|
||||
) -> None:
|
||||
"""During rotation, tokens signed with previous secret are re-signed."""
|
||||
signed_token, _ = await auth_service.login(
|
||||
db,
|
||||
password="correctpassword1",
|
||||
session_duration_minutes=60,
|
||||
session_secret="old-secret",
|
||||
)
|
||||
raw_token = auth_service.unwrap_session_token(signed_token, "old-secret")
|
||||
session = await auth_service.validate_session(
|
||||
db,
|
||||
signed_token,
|
||||
session_secret="new-secret",
|
||||
session_secret_previous="old-secret",
|
||||
)
|
||||
assert session.token == raw_token
|
||||
|
||||
async def test_validate_session_logs_rotation_event(
|
||||
self, db: aiosqlite.Connection
|
||||
) -> None:
|
||||
"""Validation processes token rotation during validation."""
|
||||
signed_token, _ = await auth_service.login(
|
||||
db,
|
||||
password="correctpassword1",
|
||||
session_duration_minutes=60,
|
||||
session_secret="old-secret",
|
||||
)
|
||||
session = await auth_service.validate_session(
|
||||
db,
|
||||
signed_token,
|
||||
session_secret="new-secret",
|
||||
session_secret_previous="old-secret",
|
||||
)
|
||||
assert session is not None
|
||||
assert session.token
|
||||
|
||||
async def test_logout_accepts_previous_secret(
|
||||
self, db: aiosqlite.Connection
|
||||
) -> None:
|
||||
"""logout() accepts tokens signed with the previous secret."""
|
||||
from app.repositories import session_repo
|
||||
|
||||
signed_token, _ = await auth_service.login(
|
||||
db,
|
||||
password="correctpassword1",
|
||||
session_duration_minutes=60,
|
||||
session_secret="old-secret",
|
||||
)
|
||||
raw_token = auth_service.unwrap_session_token(signed_token, "old-secret")
|
||||
await auth_service.logout(
|
||||
db,
|
||||
signed_token,
|
||||
session_secret="new-secret",
|
||||
session_secret_previous="old-secret",
|
||||
)
|
||||
stored = await session_repo.get_session(db, raw_token)
|
||||
assert stored is None
|
||||
|
||||
async def test_no_re_sign_without_previous_secret(
|
||||
self, db: aiosqlite.Connection
|
||||
) -> None:
|
||||
"""If no previous secret is configured, old tokens are rejected."""
|
||||
signed_token, _ = await auth_service.login(
|
||||
db,
|
||||
password="correctpassword1",
|
||||
session_duration_minutes=60,
|
||||
session_secret="old-secret",
|
||||
)
|
||||
with pytest.raises(ValueError, match="invalid"):
|
||||
await auth_service.validate_session(
|
||||
db,
|
||||
signed_token,
|
||||
session_secret="new-secret",
|
||||
session_secret_previous=None,
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user