add email support for reset password

This commit is contained in:
2025-12-23 01:04:48 +01:00
parent 9a2f47dd78
commit e0d0e8c3df
3 changed files with 128 additions and 30 deletions

View File

@@ -19,4 +19,13 @@ services:
- PGID=${PGID:-1000}
# Your Gotify URL for notifications. Should be set in a .env file.
- GOTIFY_URL=${GOTIFY_URL}
# E-Mail settings for password reset
- MAIL_USERNAME=${MAIL_USERNAME}
- MAIL_PASSWORD=${MAIL_PASSWORD}
- MAIL_FROM=${MAIL_FROM}
- MAIL_PORT=${MAIL_PORT:-587}
- MAIL_SERVER=${MAIL_SERVER}
- MAIL_STARTTLS=${MAIL_STARTTLS:-True}
- MAIL_SSL_TLS=${MAIL_SSL_TLS:-False}
- MAIL_SUPPRESS_SEND=${MAIL_SUPPRESS_SEND:-False}

94
main.py
View File

@@ -29,20 +29,26 @@ DB_FILE = "data/shoppinglist.db"
# --- Email Configuration ---
# For development, we'll use a console backend to print emails to stdout.
# In production, replace with actual SMTP settings from environment variables.
mail_starttls = os.getenv("MAIL_STARTTLS", "True").lower() in ('true', '1', 't')
mail_ssl_tls = os.getenv("MAIL_SSL_TLS", "False").lower() in ('true', '1', 't')
if mail_starttls and mail_ssl_tls:
raise ValueError("MAIL_STARTTLS and MAIL_SSL_TLS cannot both be True. Please check your .env file.")
conf = ConnectionConfig(
MAIL_USERNAME=os.getenv("MAIL_USERNAME", "your_email@example.com"),
MAIL_PASSWORD=os.getenv("MAIL_PASSWORD", "your_email_password"),
MAIL_FROM=os.getenv("MAIL_FROM", "your_email@example.com"),
MAIL_PORT=int(os.getenv("MAIL_PORT", 587)),
MAIL_SERVER=os.getenv("MAIL_SERVER", "smtp.example.com"),
MAIL_STARTTLS=bool(os.getenv("MAIL_STARTTLS", True)), # Use MAIL_STARTTLS
MAIL_SSL_TLS=bool(os.getenv("MAIL_SSL_TLS", False)), # Use MAIL_SSL_TLS
USE_CREDENTIALS=bool(os.getenv("USE_CREDENTIALS", True)),
VALIDATE_CERTS=bool(os.getenv("VALIDATE_CERTS", True)),
MAIL_STARTTLS=mail_starttls,
MAIL_SSL_TLS=mail_ssl_tls,
USE_CREDENTIALS=os.getenv("USE_CREDENTIALS", "True").lower() in ('true', '1', 't'),
VALIDATE_CERTS=os.getenv("VALIDATE_CERTS", "True").lower() in ('true', '1', 't'),
TEMPLATE_FOLDER='./templates/email',
)
conf.SUPPRESS_SEND = bool(os.getenv("MAIL_SUPPRESS_SEND", True))
conf.SUPPRESS_SEND = bool(os.getenv("MAIL_SUPPRESS_SEND", False))
fm = FastMail(conf)
@@ -70,6 +76,7 @@ class TokenData(BaseModel):
class User(BaseModel):
username: str
is_admin: bool
email: Optional[str] = None
class UserInDB(User):
@@ -635,6 +642,33 @@ async def change_password(
return {"status": "Password updated successfully"}
class EmailUpdateRequest(BaseModel):
email: str
@app.put("/api/users/me/email", status_code=status.HTTP_200_OK)
async def update_email(
request: EmailUpdateRequest,
current_user: UserInDB = Depends(get_current_active_user)
):
# Basic email validation
if "@" not in request.email and request.email != "":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid email format.",
)
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
cursor.execute(
"UPDATE users SET email = ? WHERE id = ?",
(request.email, current_user.id)
)
conn.commit()
conn.close()
return {"status": "Email updated successfully"}
class UserListUser(BaseModel):
id: int
username: str
@@ -652,7 +686,7 @@ async def get_users(current_user: UserInDB = Depends(get_current_active_user)):
conn = sqlite3.connect(DB_FILE)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT id, username, is_admin, is_locked FROM users")
cursor.execute("SELECT id, username, is_admin, is_locked, email FROM users")
users = cursor.fetchall()
conn.close()
return [dict(row) for row in users]
@@ -697,7 +731,7 @@ async def get_locked_users(current_user: UserInDB = Depends(get_current_active_u
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(
"SELECT id, username, is_admin, is_locked FROM users WHERE is_locked = 1")
"SELECT id, username, is_admin, is_locked, email FROM users WHERE is_locked = 1")
locked_users = cursor.fetchall()
conn.close()
return [dict(row) for row in locked_users]
@@ -721,28 +755,31 @@ async def unlock_user(user_id: int, current_user: UserInDB = Depends(get_current
import secrets
@app.post("/api/users/forgot-password")
async def forgot_password(request: ForgotPasswordRequest):
async def forgot_password(request: Request, forgot_req: ForgotPasswordRequest):
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
# Determine language from request header
accept_language = request.headers.get("accept-language", "en")
lang = "de" if "de" in accept_language.split(",")[0] else "en"
translations = get_all_translations(lang)
user_info = None
# Try to find user by email
cursor.execute("SELECT id, username, email FROM users WHERE email = ?", (request.email_or_username,))
cursor.execute("SELECT id, username, email FROM users WHERE email = ?", (forgot_req.email_or_username,))
user_info = cursor.fetchone()
# If not found by email, try by username
if not user_info:
cursor.execute("SELECT id, username, email FROM users WHERE username = ?", (request.email_or_username,))
cursor.execute("SELECT id, username, email FROM users WHERE username = ?", (forgot_req.email_or_username,))
user_info = cursor.fetchone()
if not user_info or not user_info[2]: # user_info[2] is the email
conn.close()
# To prevent user enumeration, always return a success message
return {"message": "If a matching user with an email address was found, a password reset email has been sent."}
return {"message": translations.get("forgot_password_email_sent", "If a matching user with an email address was found, a password reset email has been sent.")}
user_id, username, email = user_info
# Generate a secure token
token = secrets.token_urlsafe(32)
hashed_token = pwd_context.hash(token)
expires_at = datetime.utcnow() + timedelta(hours=1)
@@ -752,30 +789,31 @@ async def forgot_password(request: ForgotPasswordRequest):
conn.commit()
conn.close()
# Send email
reset_link = f"http://localhost:8000/reset-password?token={token}" # TODO: Replace with actual domain
reset_link = f"{request.url.scheme}://{request.url.hostname}:{request.url.port}/#reset-password?token={token}"
email_body = f"""
<p>{translations.get('forgot_password_email_body_hello', 'Hello {username},').format(username=username)}</p>
<p>{translations.get('forgot_password_email_body_request_text', 'You have requested a password reset for your account.')}</p>
<p>{translations.get('forgot_password_email_body_link_text', 'Please click on the following link to reset your password:')}</p>
<p><a href="{reset_link}">{reset_link}</a></p>
<p>{translations.get('forgot_password_email_body_expiry_text', 'This link is valid for 1 hour.')}</p>
<p>{translations.get('forgot_password_email_body_ignore_text', 'If you did not request a password reset, please ignore this email.')}</p>
"""
message = MessageSchema(
subject="Password Reset Request",
subject=translations.get('forgot_password_email_subject', 'Password Reset Request'),
recipients=[email],
body=f"""
<p>Hello {username},</p>
<p>You have requested a password reset for your account.</p>
<p>Please click on the following link to reset your password:</p>
<p><a href="{reset_link}">{reset_link}</a></p>
<p>This link is valid for 1 hour.</p>
<p>If you did not request a password reset, please ignore this email.</p>
""",
body=email_body,
subtype="html"
)
try:
await fm.send_message(message)
logging.info(f"Password reset email sent to {email}") # Changed this line
logging.info(f"Password reset email sent to {email}")
except Exception as e:
logging.error(f"Failed to send email: {e}") # Changed this line
# Log error, but still return success to frontend to avoid leaking information
logging.error(f"Failed to send email: {e}")
return {"message": "If a matching user with an email address was found, a password reset email has been sent."}
return {"message": translations.get("forgot_password_email_sent", "If a matching user with an email address was found, a password reset email has been sent.")}
@app.post("/api/users/reset-password")
async def reset_password(request: ResetPasswordRequest):

View File

@@ -88,7 +88,33 @@ def get_all_translations(lang: str) -> Dict[str, str]:
"restore_to_list": "Wiederherstellen nach: {listName}",
"use_trash_label": "Papierkorb für diese Liste verwenden",
"use_trash_all_label": "Papierkorb für alle Listen verwenden",
"notification_send_list_only_label": "Nur Listennamen mitsenden"
"notification_send_list_only_label": "Nur Listennamen mitsenden",
"forgot_password_email_subject": "Passwort zurücksetzen",
"forgot_password_email_body_hello": "Hallo {username},",
"forgot_password_email_body_request_text": "Sie haben eine Anfrage zum Zurücksetzen Ihres Passworts gestellt.",
"forgot_password_email_body_link_text": "Bitte klicken Sie auf den folgenden Link, um Ihr Passwort zurückzusetzen:",
"forgot_password_email_body_expiry_text": "Dieser Link ist 1 Stunde lang gültig.",
"forgot_password_email_body_ignore_text": "Wenn Sie keine Passwortzurücksetzung beantragt haben, ignorieren Sie diese E-Mail bitte.",
"settings_button": "Einstellungen",
"settings_modal_label": "Einstellungen",
"update_email_title": "E-Mail aktualisieren",
"user_email_label": "E-Mail-Adresse",
"update_email_button": "E-Mail aktualisieren",
"email_update_success": "E-Mail erfolgreich aktualisiert.",
"forgot_password_link": "Passwort vergessen?",
"forgot_password_title": "Passwort vergessen",
"email_username_label": "E-Mail oder Benutzername",
"send_reset_link_button": "Link zum Zurücksetzen senden",
"back_to_login_link": "Zurück zum Login",
"reset_password_title": "Passwort zurücksetzen",
"set_new_password_button": "Neues Passwort festlegen",
"forgot_password_email_sent": "Wenn ein passender Benutzer mit einer E-Mail-Adresse gefunden wurde, wurde eine E-Mail zum Zurücksetzen des Passworts gesendet.",
"unlock_user_button": "Entsperren",
"confirm_unlock_user": "Sind Sie sicher, dass Sie den Benutzer '{username}' entsperren möchten?",
"delete_user_button": "Löschen",
"confirm_delete_user": "Sind Sie sicher, dass Sie den Benutzer '{username}' löschen möchten?",
"generic_error": "Ein Fehler ist aufgetreten.",
"email_label": "E-Mail"
}
else: # Fallback to English
return {
@@ -153,5 +179,30 @@ def get_all_translations(lang: str) -> Dict[str, str]:
"restore_to_list": "Restore to: {listName}",
"use_trash_label": "Use trash for this list",
"use_trash_all_label": "Use trash for all lists",
"notification_send_list_only_label": "Send list name only"
"notification_send_list_only_label": "Send list name only",
"forgot_password_email_subject": "Password Reset Request",
"forgot_password_email_body_hello": "Hello {username},",
"forgot_password_email_body_request_text": "You have requested a password reset for your account.",
"forgot_password_email_body_link_text": "Please click on the following link to reset your password:",
"forgot_password_email_body_expiry_text": "This link is valid for 1 hour.",
"forgot_password_email_body_ignore_text": "If you did not request a password reset, please ignore this email.",
"settings_button": "Settings",
"settings_modal_label": "Settings",
"update_email_title": "Update Email",
"user_email_label": "Email Address",
"update_email_button": "Update Email",
"email_update_success": "Email updated successfully.",
"forgot_password_link": "Forgot password?",
"forgot_password_title": "Forgot Password",
"email_username_label": "Email or Username",
"send_reset_link_button": "Send Reset Link",
"back_to_login_link": "Back to Login",
"reset_password_title": "Reset Password",
"set_new_password_button": "Set New Password",
"forgot_password_email_sent": "If a matching user with an email address was found, a password reset email has been sent.",
"unlock_user_button": "Unlock",
"confirm_unlock_user": "Are you sure you want to unlock {username}?",
"delete_user_button": "Delete", "confirm_delete_user": "Are you sure you want to delete user {username}?",
"generic_error": "An error occurred.",
"email_label": "Email"
}