47722dbb19
Verifies that on a successful change the digest rotates, the password_changed_at advances, this session keeps working with the re-stamped marker, and a parallel session forged from the pre-change marker is rejected by load_current_user. profile_password_change now writes a naive password_changed_at so the in-memory marker matches what SQLite returns on next read. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
from flask import Blueprint, Response, redirect, render_template, request, session
|
|
from sqlalchemy import select
|
|
|
|
from l4d2web.auth import (
|
|
current_user,
|
|
hash_password,
|
|
require_login,
|
|
validate_new_password,
|
|
verify_password,
|
|
)
|
|
from l4d2web.db import session_scope
|
|
from l4d2web.models import User, now_utc
|
|
from l4d2web.services.rate_limit import check_rate_limit
|
|
|
|
|
|
bp = Blueprint("profile", __name__)
|
|
|
|
|
|
_ERROR_MESSAGES = {
|
|
"fields_required": "Fill in all three fields.",
|
|
"mismatch": "New password and confirmation do not match.",
|
|
"too_short": "New password must be at least 8 characters.",
|
|
"empty": "New password must not be empty.",
|
|
"wrong_current": "Current password is incorrect.",
|
|
}
|
|
|
|
|
|
PROFILE_PW_RATE_LIMIT_WINDOW_SECONDS = 60.0
|
|
PROFILE_PW_RATE_LIMIT_MAX_ATTEMPTS = 10
|
|
PROFILE_PW_ATTEMPTS_BY_IP: dict[str, list[float]] = {}
|
|
|
|
|
|
def reset_profile_password_rate_limits() -> None:
|
|
PROFILE_PW_ATTEMPTS_BY_IP.clear()
|
|
|
|
|
|
def _redirect_with_error(error_key: str) -> Response:
|
|
return redirect(f"/profile?error={error_key}")
|
|
|
|
|
|
@bp.get("/profile")
|
|
@require_login
|
|
def profile_page() -> str:
|
|
error_key = request.args.get("error", "")
|
|
success = request.args.get("success") == "1"
|
|
return render_template(
|
|
"profile.html",
|
|
error_message=_ERROR_MESSAGES.get(error_key, ""),
|
|
success=success,
|
|
)
|
|
|
|
|
|
@bp.post("/profile/password")
|
|
@require_login
|
|
def profile_password_change() -> Response:
|
|
remote_addr = request.remote_addr or "unknown"
|
|
if check_rate_limit(
|
|
PROFILE_PW_ATTEMPTS_BY_IP,
|
|
remote_addr,
|
|
window=PROFILE_PW_RATE_LIMIT_WINDOW_SECONDS,
|
|
max_attempts=PROFILE_PW_RATE_LIMIT_MAX_ATTEMPTS,
|
|
):
|
|
return Response("too many attempts", status=429)
|
|
|
|
current_password = request.form.get("current_password", "")
|
|
new_password = request.form.get("new_password", "")
|
|
confirm_new_password = request.form.get("confirm_new_password", "")
|
|
|
|
if not current_password or not new_password or not confirm_new_password:
|
|
return _redirect_with_error("fields_required")
|
|
if new_password != confirm_new_password:
|
|
return _redirect_with_error("mismatch")
|
|
policy_error = validate_new_password(new_password)
|
|
if policy_error is not None:
|
|
error_key = "empty" if policy_error == "password must not be empty" else "too_short"
|
|
return _redirect_with_error(error_key)
|
|
|
|
actor = current_user()
|
|
assert actor is not None # @require_login
|
|
|
|
with session_scope() as db:
|
|
user = db.scalar(select(User).where(User.id == actor.id))
|
|
if user is None or not verify_password(current_password, user.password_digest):
|
|
return _redirect_with_error("wrong_current")
|
|
user.password_digest = hash_password(new_password)
|
|
# Strip tz so the marker matches what a subsequent DB read returns
|
|
# (SQLite DateTime columns don't preserve tzinfo).
|
|
user.password_changed_at = now_utc().replace(tzinfo=None)
|
|
new_marker = user.password_changed_at.isoformat()
|
|
|
|
session["pw_changed_at"] = new_marker
|
|
|
|
return redirect("/profile?success=1")
|