feat: convert to self-hosted version

api/backend/auth/auth_router.py

@@ -0,0 +1,57 @@
+# STL
+from datetime import timedelta
+
+# PDM
+from fastapi import Depends, APIRouter, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+
+# LOCAL
+from api.backend.schemas import User, Token, UserCreate
+from api.backend.database import get_user_collection
+from api.backend.auth.auth_utils import (
+    ACCESS_TOKEN_EXPIRE_MINUTES,
+    get_current_user,
+    authenticate_user,
+    get_password_hash,
+    create_access_token,
+)
+
+auth_router = APIRouter()
+
+
+@auth_router.post("/api/auth/token", response_model=Token)
+async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
+    user = await authenticate_user(form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    expire_minutes = (
+        int(ACCESS_TOKEN_EXPIRE_MINUTES) if ACCESS_TOKEN_EXPIRE_MINUTES else 60
+    )
+
+    access_token_expires = timedelta(minutes=expire_minutes)
+    access_token = create_access_token(
+        data={"sub": user.email}, expires_delta=access_token_expires
+    )
+
+    return {"access_token": access_token, "token_type": "bearer"}
+
+
+@auth_router.post("/api/auth/signup", response_model=User)
+async def create_user(user: UserCreate):
+    users_collection = get_user_collection()
+    hashed_password = get_password_hash(user.password)
+    user_dict = user.model_dump()
+    user_dict["hashed_password"] = hashed_password
+    del user_dict["password"]
+    _ = await users_collection.insert_one(user_dict)
+    return user_dict
+
+
+@auth_router.get("/api/auth/users/me", response_model=User)
+async def read_users_me(current_user: User = Depends(get_current_user)):
+    return current_user

api/backend/auth/auth_utils.py

@@ -0,0 +1,101 @@
+# STL
+import os
+from typing import Any, Optional
+from datetime import datetime, timedelta
+
+# PDM
+from jose import JWTError, jwt
+from dotenv import load_dotenv
+from fastapi import Depends, HTTPException, status
+from passlib.context import CryptContext
+from fastapi.security import OAuth2PasswordBearer
+
+# LOCAL
+from api.backend.schemas import User, UserInDB, TokenData
+from api.backend.database import get_user_collection
+
+_ = load_dotenv()
+
+SECRET_KEY = os.getenv("SECRET_KEY") or ""
+ALGORITHM = os.getenv("ALGORITHM") or ""
+ACCESS_TOKEN_EXPIRE_MINUTES = os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES")
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
+
+
+def verify_password(plain_password: str, hashed_password: str):
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password: str):
+    return pwd_context.hash(password)
+
+
+async def get_user(email: str):
+    user_collection = get_user_collection()
+    user = await user_collection.find_one({"email": email})
+
+    if not user:
+        return
+
+    return UserInDB(**user)
+
+
+async def authenticate_user(email: str, password: str):
+    user = await get_user(email)
+
+    if not user:
+        return False
+
+    if not verify_password(password, user.hashed_password):
+        return False
+
+    return user
+
+
+def create_access_token(
+    data: dict[str, Any], expires_delta: Optional[timedelta] = None
+):
+    to_encode = data.copy()
+
+    if expires_delta:
+        expire = datetime.now() + expires_delta
+    else:
+        expire = datetime.now() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+async def get_current_user(token: str = Depends(oauth2_scheme)):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    try:
+        payload: Optional[dict[str, Any]] = jwt.decode(
+            token, SECRET_KEY, algorithms=[ALGORITHM]
+        )
+        if not payload:
+            raise credentials_exception
+
+        email = payload.get("sub")
+
+        if email is None:
+            raise credentials_exception
+
+        token_data = TokenData(email=email)
+
+    except JWTError:
+        raise credentials_exception
+
+    user = await get_user(email=token_data.email)
+
+    if user is None:
+        raise credentials_exception
+
+    return user