473 lines
8.6 KiB
Python
473 lines
8.6 KiB
Python
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()
|
|
|
|
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"""\
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
</head>
|
|
|
|
<body>
|
|
|
|
<p>Hello,</p>
|
|
|
|
<p>
|
|
You requested to use this email address
|
|
for your Calendar account.
|
|
</p>
|
|
|
|
<p>
|
|
<a href="{verification_url}">
|
|
https://ben.de-roo.org/api/verify
|
|
</a>
|
|
</p>
|
|
|
|
<p>
|
|
This link expires after 24 hours.
|
|
</p>
|
|
|
|
<p>
|
|
If you did not request this,
|
|
you can ignore this email.
|
|
</p>
|
|
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
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"""\
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
</head>
|
|
|
|
<body>
|
|
|
|
<p>Hello,</p>
|
|
|
|
<p>
|
|
You requested to reset your Calendar password.
|
|
</p>
|
|
|
|
<p>
|
|
<a href="{password_reset_url}">
|
|
https://ben.de-roo.org/api/update-password_page
|
|
</a>
|
|
</p>
|
|
|
|
<p>
|
|
This link expires after 1 hour.
|
|
</p>
|
|
|
|
<p>
|
|
If you did not request this,
|
|
you can ignore this email.
|
|
</p>
|
|
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
send_email(
|
|
email=email,
|
|
subject="Reset your password",
|
|
plain_text=plain_text,
|
|
html=html,
|
|
)
|
|
|
|
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 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()
|