calendar/backend/auth.py

from database import execute, fetch_one, fetch_all
import bcrypt
from jose import jwt
from datetime import datetime, timedelta, timezone
from dotenv import load_dotenv
import os
from fastapi import Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from email_service import (
    create_verification,
    send_password_reset_email,
    create_password_reset,
    hash_token,
)

import secrets

security = HTTPBearer()
load_dotenv()

SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = "HS256"
CHEESE_IN_DUTCH = "kaas"


def create_token(user_id):
    payload = {
        "user_id": user_id,
        "exp": datetime.utcnow() + timedelta(hours=24)
    }

    token = jwt.encode(
        payload,
        SECRET_KEY,
        algorithm=ALGORITHM
    )

    return token


def create_user(username, password, email):

    password_hash = bcrypt.hashpw(
        password.encode(),
        bcrypt.gensalt()
    ).decode()

    user_id = execute(
        """
        INSERT INTO users (
            username,
            password_hash,
            email,
            email_verified
        )
        VALUES (?, ?, ?, 0)
        """,
        (
            username,
            password_hash,
            email,
        )
    )

    token = create_verification(user_id)

    send_verification_email(
        email,
        token,
    )

    return user_id

def verify_user(username, password):

    user = fetch_one(
        "SELECT * FROM users WHERE username = ?",
        (username,)
    )

    if not user:
        return None

    if bcrypt.checkpw(
        password.encode(),
        user["password_hash"].encode()
    ):
        return user["id"]

    return None


def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    token = credentials.credentials

    payload = jwt.decode(
        token,
        SECRET_KEY,
        algorithms=[ALGORITHM]
    )

    return payload["user_id"]


def get_user(user_id):
    return fetch_one(
        """
        SELECT id, username, email, email_verified
        FROM users
        WHERE id = ?
        """,
        (user_id,)
    )

def change_password(user_id: int, old_password: str, new_password: str):
    user = fetch_one(
        """
        SELECT password_hash
        FROM users
        WHERE id = ?
        """,
        (user_id,),
    )

    if user is None:
        return {
            "status": "failed",
            "reason": "user_not_found",
        }

    if not bcrypt.checkpw(
        old_password.encode(),
        user["password_hash"].encode(),
    ):
        return {
            "status": "failed",
            "reason": "incorrect_password",
        }

    password_hash = bcrypt.hashpw(
        new_password.encode(),
        bcrypt.gensalt(),
    ).decode()

    execute(
        """
        UPDATE users
        SET password_hash = ?
        WHERE id = ?
        """,
        (
            password_hash,
            user_id,
        ),
    )

    return {
        "status": "updated",
    }

def delete_me(user_id, password):
    user = fetch_one(
        """
        SELECT password_hash
        FROM users
        WHERE id = ?
        """,
        (user_id,)
    )

    if user is None:
        return {
            "status": "failed",
            "reason": "user_not_found",
        }

    if not bcrypt.checkpw(
        password.encode(),
        user["password_hash"].encode(),
    ):
        return {
            "status": "failed",
            "reason": "incorrect_password",
        }

    owned_rooms = fetch_all(
        """
        SELECT id
        FROM rooms
        WHERE owner_id = ?
        """,
        (user_id,)
    )

    for room in owned_rooms:
        room_id = room["id"]

        new_owner = fetch_one(
            """
            SELECT user_id
            FROM room_members
            WHERE room_id = ?
            AND user_id != ?
            ORDER BY rowid
            LIMIT 1
            """,
            (room_id, user_id)
        )

        if new_owner:
            new_owner_id = new_owner["user_id"]

            execute(
                """
                UPDATE rooms
                SET owner_id = ?
                WHERE id = ?
                """,
                (new_owner_id, room_id)
            )

            execute(
                """
                UPDATE room_members
                SET role = 'owner'
                WHERE room_id = ?
                AND user_id = ?
                """,
                (room_id, new_owner_id)
            )

    execute(
        """
        DELETE FROM users
        WHERE id = ?
        """,
        (user_id,)
    )

    return {
        "status": "deleted"
    }

def reset_password(email: str):
    email = email.strip().lower()

    user = fetch_one(
        """
        SELECT id
        FROM users
        WHERE email = ?
        """,
        (email,),
    )

    if user is None:
        return {
            "status": "sent",
        }

    token = create_password_reset(user["id"])

    send_password_reset_email(
        email,
        token,
    )

    return {
        "status": "sent",
    }

def reset_password_with_token(token: str, new_password: str):
    token_hash = hash_token(token)

    reset = fetch_one(
        """
        SELECT id, user_id, expires_at
        FROM password_reset_tokens
        WHERE token_hash = ?
        """,
        (token_hash,),
    )

    if reset is None:
        return {
            "status": "failed",
            "reason": "invalid_token",
        }

    expires_at = datetime.fromisoformat(reset["expires_at"])

    if datetime.now(timezone.utc) >= expires_at:
        execute(
            """
            DELETE FROM password_reset_tokens
            WHERE id = ?
            """,
            (reset["id"],),
        )

        return {
            "status": "failed",
            "reason": "token_expired",
        }

    password_hash = bcrypt.hashpw(
        new_password.encode(),
        bcrypt.gensalt(),
    ).decode()

    execute(
        """
        UPDATE users
        SET password_hash = ?
        WHERE id = ?
        """,
        (
            password_hash,
            reset["user_id"],
        ),
    )

    execute(
        """
        DELETE FROM password_reset_tokens
        WHERE id = ?
        """,
        (reset["id"],),
    )

    return {
        "status": "updated",
    }