update emails and verifing

backend/database.py

@@ -5,9 +5,6 @@ DB_PATH = Path(__file__).parent.parent / "database" / "calendar.db"
 
 
 def get_connection():
-    """
-    Maakt een verbinding met de SQLite database.
-    """
     conn = sqlite3.connect(DB_PATH)
 
     conn.execute("PRAGMA foreign_keys = ON")
@@ -18,9 +15,6 @@ def get_connection():
 
 
 def execute(query, params=()):
-    """
-    Voor INSERT, UPDATE, DELETE.
-    """
     with get_connection() as conn:
         cursor = conn.cursor()
         cursor.execute(query, params)
@@ -30,9 +24,6 @@ def execute(query, params=()):
 
 
 def fetch_one(query, params=()):
-    """
-    Haalt één resultaat op.
-    """
     with get_connection() as conn:
         cursor = conn.cursor()
         cursor.execute(query, params)
@@ -52,9 +43,6 @@ def fetch_all(query, params=()):
 
 
 def test_connection():
-    """
-    Test of de database bereikbaar is.
-    """
     with get_connection() as conn:
         cursor = conn.cursor()
         cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")

backend/email_service.py

@@ -0,0 +1,272 @@
+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 / "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://jouwdomein.nl/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.
+"""
+    )
+
+    with smtplib.SMTP(smtp_host, smtp_port) as smtp:
+        smtp.starttls()
+        smtp.login(
+            smtp_username,
+            smtp_password,
+        )
+        smtp.send_message(message)

backend/main.py

@@ -1,15 +1,24 @@
-# GEEN SQL IN MAIN.PY, DIT IS DE API, HOU HET OVERZICHTELIJK!!
+# GEEN SQL IN MAIN.PY, DIT IS DE API/ERMISSIECHECK, HOU HET OVERZICHTELIJK!!
 
 from datetime import date
 from fastapi import FastAPI, Depends, HTTPException, Query
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import PlainTextResponse
-from pydantic import BaseModel
+from pydantic import BaseModel, EmailStr
 from typing import Literal
 import sqlite3
+from datetime import datetime, timezone
+from fastapi import Query
+
+from database import fetch_one, execute
+from email_service import hash_token
 
 from database import get_table_names
 
+from email_service import (
+    update_email
+)
+
 from auth import (
     create_user,
     verify_user,
@@ -59,6 +68,8 @@ from event_signups import (
     get_event_responses,
 )
 
+class EmailUpdate(BaseModel):
+    email: EmailStr
 
 class MemberRoleUpdate(BaseModel):
     role: Literal["member", "admin"]
@@ -976,6 +987,78 @@ def api_delete_event(
         "status": "deleted",
     }
 
+# email endpoints
+
+@app.put(
+    "/me/email",
+    summary="Update email address",
+    description="Changes the authenticated user's email address.",
+)
+def api_update_email(
+    data: EmailUpdate,
+    user_id: int = Depends(get_current_user),
+):
+    return update_email(
+        user_id,
+        data.email,
+    )
+
+@app.get("/verify-email")
+def verify_email(token: str = Query(...)):
+    token_hash = hash_token(token)
+
+    verification = fetch_one(
+        """
+        SELECT id, user_id, expires_at
+        FROM email_verifications
+        WHERE token_hash = ?
+        """,
+        (token_hash,)
+    )
+
+    if not verification:
+        raise HTTPException(
+            status_code=400,
+            detail="Invalid verification token"
+        )
+
+    expires_at = datetime.fromisoformat(verification["expires_at"])
+
+    if expires_at < datetime.now(timezone.utc):
+        execute(
+            """
+            DELETE FROM email_verifications
+            WHERE id = ?
+            """,
+            (verification["id"],)
+        )
+
+        raise HTTPException(
+            status_code=400,
+            detail="Verification token has expired"
+        )
+
+    execute(
+        """
+        UPDATE users
+        SET email_verified = 1
+        WHERE id = ?
+        """,
+        (verification["user_id"],)
+    )
+
+    execute(
+        """
+        DELETE FROM email_verifications
+        WHERE id = ?
+        """,
+        (verification["id"],)
+    )
+
+    return {
+        "status": "verified",
+        "message": "Email address successfully verified"
+    }
 
 # easter egg