auth: stamp password_changed_at marker in session on login

login_user now records the user's current password_changed_at on the
session. The next commit will use this marker to invalidate sessions
whose password has been rotated under them.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
mwiegand 2026-05-11 21:46:20 +02:00
parent 26a6a9d7b0
commit 84dc672180
No known key found for this signature in database
3 changed files with 18 additions and 2 deletions

View file

@ -48,8 +48,9 @@ def current_user() -> User | None:
return getattr(g, "user", None) return getattr(g, "user", None)
def login_user(user_id: int) -> None: def login_user(user_id: int, password_changed_at) -> None:
session["user_id"] = user_id session["user_id"] = user_id
session["pw_changed_at"] = password_changed_at.isoformat()
def logout_user() -> None: def logout_user() -> None:

View file

@ -49,7 +49,7 @@ def login() -> Response:
# Same generic response for missing user, wrong password, or # Same generic response for missing user, wrong password, or
# deactivated account — no timing oracle for deactivation status. # deactivated account — no timing oracle for deactivation status.
return Response("invalid credentials", status=401) return Response("invalid credentials", status=401)
login_user(user.id) login_user(user.id, user.password_changed_at)
LOGIN_ATTEMPTS_BY_IP.pop(remote_addr, None) LOGIN_ATTEMPTS_BY_IP.pop(remote_addr, None)
next_target = request.form.get("next", "") next_target = request.form.get("next", "")
return redirect(next_target if is_safe_next(next_target) else "/dashboard") return redirect(next_target if is_safe_next(next_target) else "/dashboard")

View file

@ -134,6 +134,21 @@ def test_login_sets_session(client) -> None:
assert sess.get("user_id") is not None assert sess.get("user_id") is not None
def test_login_stamps_password_changed_at_in_session(client) -> None:
with session_scope() as session:
session.add(User(username="alice", password_digest=hash_password("secret")))
response = client.post("/login", data={"username": "alice", "password": "secret"})
assert response.status_code == 302
with client.session_transaction() as sess:
marker = sess.get("pw_changed_at")
assert marker is not None
with session_scope() as session:
user = session.query(User).filter_by(username="alice").one()
assert marker == user.password_changed_at.isoformat()
def test_create_user_cli_uses_environment_password(tmp_path, monkeypatch) -> None: def test_create_user_cli_uses_environment_password(tmp_path, monkeypatch) -> None:
db_url = f"sqlite:///{tmp_path/'create_user.db'}" db_url = f"sqlite:///{tmp_path/'create_user.db'}"
monkeypatch.setenv("DATABASE_URL", db_url) monkeypatch.setenv("DATABASE_URL", db_url)