feat(auth): self-service password reset (forgot/reset endpoints)

Add an email-based "forgot password" flow:
- PasswordResetToken model (stores only a SHA-256 hash, single-use, TTL)
  and migration 27 for the password_reset_tokens table.
- POST /api/auth/forgot-password — generic response (no account
  enumeration), 60s throttle, emails a one-time reset link via the
  existing EmailChannel as a background task.
- GET /api/auth/reset-password/validate — token validity check for the UI.
- POST /api/auth/reset-password — validates the token, enforces the
  password policy, sets the new password, clears lock state, invalidates
  outstanding tokens and terminates all sessions.
- Configurable link lifetime via PASSWORD_RESET_TOKEN_EXPIRE_MINUTES (30m).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-26 03:24:19 +05:00
parent 65e14585c5
commit 8b93161434
6 changed files with 329 additions and 6 deletions

View File

@@ -1,5 +1,5 @@
from .base import *
from .auth import Role, user_servers, User, ActiveSession, LoginAttempt, BlockedIP, UserSSHKey
from .auth import Role, user_servers, User, ActiveSession, LoginAttempt, BlockedIP, UserSSHKey, PasswordResetToken
from .proxmox import OSTemplateGroup, OSTemplate, ProxmoxServer, VMInstance, VMSnapshotArchive
from .ipam import IPAMNetwork, IPAMPool, IPAMAllocation, IPAMHistory
from .tasks import TaskQueue, ProxmoxTask, DeployTask
@@ -10,7 +10,7 @@ from .catalog import ImageMirror
__all__ = [
'Base', 'utcnow',
'Role', 'user_servers', 'User', 'ActiveSession', 'LoginAttempt', 'BlockedIP', 'UserSSHKey',
'Role', 'user_servers', 'User', 'ActiveSession', 'LoginAttempt', 'BlockedIP', 'UserSSHKey', 'PasswordResetToken',
'OSTemplateGroup', 'OSTemplate', 'ProxmoxServer', 'VMInstance', 'VMSnapshotArchive',
'IPAMNetwork', 'IPAMPool', 'IPAMAllocation', 'IPAMHistory',
'TaskQueue', 'ProxmoxTask', 'DeployTask',

View File

@@ -116,6 +116,36 @@ class User(Base):
def __repr__(self):
return f"<User(id={self.id}, username='{self.username}', is_admin={self.is_admin})>"
class PasswordResetToken(Base):
"""One-time tokens for the self-service 'forgot password' flow.
Only a SHA-256 hash of the token is stored; the raw token lives solely in
the emailed link. A token is single-use (``used_at``) and time-limited
(``expires_at``).
"""
__tablename__ = "password_reset_tokens"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
token_hash = Column(String(64), unique=True, nullable=False, index=True)
expires_at = Column(DateTime(timezone=True), nullable=False)
used_at = Column(DateTime(timezone=True), nullable=True)
ip_address = Column(String(45), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
user = relationship("User")
def is_valid(self) -> bool:
"""Token is unused and not expired."""
if self.used_at is not None:
return False
if self.expires_at is None or make_tz_aware(self.expires_at) <= utcnow():
return False
return True
def __repr__(self):
return f"<PasswordResetToken(id={self.id}, user_id={self.user_id}, used={self.used_at is not None})>"
class ActiveSession(Base):
"""Active user sessions for single-session enforcement"""
__tablename__ = "active_sessions"