Add email-based password reset flow

Users who forget their password can now request a reset link via email.
Tokens are SHA-256 hashed before storage, expire after 1 hour, and
previous unused tokens are invalidated on new requests. The forgot-password
endpoint always returns success to prevent email enumeration.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-13 21:42:45 -05:00
parent d874e0e2ae
commit d354da5b51
11 changed files with 661 additions and 1 deletions
+76 -1
View File
@@ -3,17 +3,26 @@ import asyncio
from quart import Blueprint, g, jsonify, request, session
from fabledassistant.auth import login_required, get_current_user_id
from fabledassistant.config import Config
from fabledassistant.services.auth import (
authenticate,
change_password,
create_password_reset_token,
create_user,
get_user_by_email,
get_user_by_id,
get_user_by_username,
get_user_count,
is_registration_open,
reset_password_with_token,
)
from fabledassistant.services.logging import log_audit
from fabledassistant.services.notifications import notify_security_event
from fabledassistant.services.notifications import (
notify_security_event,
send_password_reset_email,
send_password_reset_success_email,
)
from fabledassistant.services.email import is_smtp_configured
auth_bp = Blueprint("auth", __name__, url_prefix="/api/auth")
@@ -121,6 +130,72 @@ async def update_password():
return jsonify({"status": "ok"})
@auth_bp.route("/forgot-password", methods=["POST"])
async def forgot_password():
data = await request.get_json()
email = (data.get("email") or "").strip().lower()
if not email:
return jsonify({"error": "Email is required"}), 400
# Always return success to prevent email enumeration
user = await get_user_by_email(email)
if user and await is_smtp_configured():
raw_token = await create_password_reset_token(user.id)
reset_url = f"{Config.BASE_URL}/reset-password?token={raw_token}"
asyncio.create_task(send_password_reset_email(user.email, reset_url))
await log_audit(
"password_reset_requested",
user_id=user.id,
username=user.username,
ip_address=request.remote_addr,
)
return jsonify({"status": "ok"})
@auth_bp.route("/reset-password", methods=["POST"])
async def reset_password():
data = await request.get_json()
token = (data.get("token") or "").strip()
new_password = data.get("new_password") or ""
if not token:
return jsonify({"error": "Reset token is required"}), 400
if len(new_password) < 8:
return jsonify({"error": "Password must be at least 8 characters"}), 400
success = await reset_password_with_token(token, new_password)
if not success:
return jsonify({"error": "Invalid or expired reset link"}), 400
# Look up user by token to send notifications (token is now used, so look up by hash)
import hashlib
from fabledassistant.models import async_session
from fabledassistant.models.password_reset import PasswordResetToken
from sqlalchemy import select
token_hash = hashlib.sha256(token.encode()).hexdigest()
async with async_session() as session:
result = await session.execute(
select(PasswordResetToken).where(PasswordResetToken.token_hash == token_hash)
)
reset_token = result.scalars().first()
if reset_token:
user = await get_user_by_id(reset_token.user_id)
if user:
await log_audit(
"password_reset_completed",
user_id=user.id,
username=user.username,
ip_address=request.remote_addr,
)
if user.email:
asyncio.create_task(send_password_reset_success_email(user.email))
return jsonify({"status": "ok"})
@auth_bp.route("/status", methods=["GET"])
async def status():
count = await get_user_count()