import hashlib import os import secrets import smtplib import sqlite3 from datetime import datetime, timedelta, timezone from email.message import EmailMessage from pathlib import Path from dotenv import load_dotenv load_dotenv() DB_NAME = Path(__file__).parent.parent / "database" / "calendar.db" def get_connection(): conn = sqlite3.connect(DB_NAME) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON;") return conn def hash_token(token: str) -> str: return hashlib.sha256( token.encode("utf-8") ).hexdigest() # ============================================================ # EMAIL # ============================================================ def send_email( email: str, subject: str, plain_text: str, html: str, ): smtp_host = os.getenv("SMTP_HOST") smtp_port = int(os.getenv("SMTP_PORT", "587")) smtp_username = os.getenv("SMTP_USERNAME") smtp_password = os.getenv("SMTP_PASSWORD") smtp_from = os.getenv("SMTP_FROM") if not all([ smtp_host, smtp_username, smtp_password, smtp_from, ]): raise RuntimeError("SMTP configuration is incomplete") message = EmailMessage() message["From"] = smtp_from message["To"] = email message["Subject"] = subject # Plain-text fallback message.set_content(plain_text) # HTML version message.add_alternative( html, subtype="html", ) print(f"Sending email to {email}") with smtplib.SMTP( smtp_host, smtp_port, ) as smtp: print("SMTP connected") smtp.starttls() print("TLS started") smtp.login( smtp_username, smtp_password, ) print("SMTP login successful") smtp.send_message(message) print("Email sent") def send_verification_email( email: str, token: str, ): verification_url = ( "https://ben.de-roo.org" f"/api/verify-email?token={token}" ) plain_text = f"""Hello, You requested to use this email address for your Calendar account. Verify your email address using this link: {verification_url} This link expires after 24 hours. If you did not request this, you can ignore this email. """ html = f"""\

Hello,

You requested to use this email address for your Calendar account.

This link expires after 24 hours.

If you did not request this, you can ignore this email.

""" send_email( email=email, subject="Verify your Calendar email address", plain_text=plain_text, html=html, ) def send_password_reset_email( email: str, token: str, ): password_reset_url = ( "https://ben.de-roo.org" f"/api/update-password_page?token={token}" ) plain_text = f"""Hello, You requested to reset your Calendar password. Reset your password using this link: {password_reset_url} This link expires after 1 hour. If you did not request this, you can ignore this email. """ html = f"""\

Hello,

You requested to reset your Calendar password.

Reset your password

This link expires after 1 hour.

If you did not request this, you can ignore this email.

""" send_email( email=email, subject="Reset your password", plain_text=plain_text, html=html, ) # ============================================================ # EMAIL ADDRESS VERIFICATION # ============================================================ def update_email( user_id: int, new_email: str, ): new_email = new_email.strip().lower() conn = get_connection() cur = conn.cursor() try: cur.execute( """ SELECT id FROM users WHERE id = ? """, (user_id,), ) if cur.fetchone() is None: return { "status": "failed", "reason": "user_not_found", } cur.execute( """ SELECT id FROM users WHERE email = ? AND id != ? """, ( new_email, user_id, ), ) if cur.fetchone() is not None: return { "status": "failed", "reason": "email_already_in_use", } cur.execute( """ UPDATE users SET email = ?, email_verified = 0 WHERE id = ? """, ( new_email, user_id, ), ) cur.execute( """ DELETE FROM email_verifications WHERE user_id = ? """, (user_id,), ) conn.commit() return { "status": "updated", "email": new_email, "email_verified": False, } except sqlite3.IntegrityError: conn.rollback() return { "status": "failed", "reason": "email_already_in_use", } finally: conn.close() def create_verification( user_id: int, ): token = secrets.token_urlsafe(32) token_hash = hash_token(token) expires_at = ( datetime.now(timezone.utc) + timedelta(hours=24) ).isoformat() conn = get_connection() cur = conn.cursor() try: cur.execute( """ DELETE FROM email_verifications WHERE user_id = ? """, (user_id,), ) cur.execute( """ INSERT INTO email_verifications ( user_id, token_hash, expires_at ) VALUES (?, ?, ?) """, ( user_id, token_hash, expires_at, ), ) conn.commit() return token finally: conn.close() def verify_email( token: str, ): token_hash = hash_token(token) conn = get_connection() cur = conn.cursor() try: cur.execute( """ SELECT id, user_id, expires_at FROM email_verifications WHERE token_hash = ? """, (token_hash,), ) verification = cur.fetchone() if verification is None: return { "status": "failed", "reason": "invalid_token", } expires_at = datetime.fromisoformat( verification["expires_at"] ) if datetime.now(timezone.utc) >= expires_at: cur.execute( """ DELETE FROM email_verifications WHERE id = ? """, (verification["id"],), ) conn.commit() return { "status": "failed", "reason": "token_expired", } cur.execute( """ UPDATE users SET email_verified = 1 WHERE id = ? """, (verification["user_id"],), ) cur.execute( """ DELETE FROM email_verifications WHERE id = ? """, (verification["id"],), ) conn.commit() return { "status": "verified", } finally: conn.close() # ============================================================ # PASSWORD RESET # ============================================================ def create_password_reset( user_id: int, ): token = secrets.token_urlsafe(32) token_hash = hash_token(token) expires_at = ( datetime.now(timezone.utc) + timedelta(hours=1) ).isoformat() conn = get_connection() cur = conn.cursor() try: cur.execute( """ DELETE FROM password_reset_tokens WHERE user_id = ? """, (user_id,), ) cur.execute( """ INSERT INTO password_reset_tokens ( user_id, token_hash, expires_at ) VALUES (?, ?, ?) """, ( user_id, token_hash, expires_at, ), ) conn.commit() return token finally: conn.close()