381 lines
7.9 KiB
Python
381 lines
7.9 KiB
Python
import hashlib
|
|
import secrets
|
|
import sqlite3
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import os
|
|
import smtplib
|
|
from email.message import EmailMessage
|
|
|
|
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()
|
|
|
|
|
|
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()
|
|
|
|
def send_verification_email(email: str, token: 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")
|
|
|
|
verification_url = (
|
|
f"https://ben.de-roo.org/api/verify-email?token={token}"
|
|
)
|
|
|
|
message = EmailMessage()
|
|
|
|
message["From"] = smtp_from
|
|
message["To"] = email
|
|
message["Subject"] = "Verify your Calendar email address"
|
|
|
|
message.set_content(
|
|
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.
|
|
"""
|
|
)
|
|
|
|
print(f"Sending verification 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("Verification email sent")
|
|
|
|
def send_password_reset_email(email: str, token: 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")
|
|
|
|
password_reset_url = (
|
|
f"https://ben.de-roo.org/api/update-password?token={token}"
|
|
)
|
|
|
|
message = EmailMessage()
|
|
|
|
message["From"] = smtp_from
|
|
message["To"] = email
|
|
message["Subject"] = "Reset your password"
|
|
|
|
message.set_content(
|
|
f"""Hello,
|
|
|
|
You requested to use this email address for reseting your Calendar password.
|
|
|
|
Reset your password using this link:
|
|
|
|
{password_reset_url}
|
|
|
|
This link expires after 24 hours.
|
|
|
|
If you did not request this, you can ignore this email.
|
|
"""
|
|
)
|
|
|
|
print(f"Sending password reset 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("Password reset email sent")
|
|
|
|
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() |