Add endpoint to authenticate user

.env.template

@@ -1,3 +1,4 @@
 ALLOW_ORIGINS=*
 DATABASE_URL=postgres+asyncpg://user:password@host/db_name
 TRUSTED_HOST=localhost
+SECRET_KEY=PRODUCTION-SECRET-KEY

app/configs.py

@@ -119,6 +119,7 @@ class Settings(BaseModel):
     project_root: Path = Path(__file__).parent.parent.resolve()
     static: Path = Path("static")
     trusted_host: str = get_env_var("TRUSTED_HOST", cast=str)
+    secret_key: str = get_env_var("SECRET_KEY", cast=str)
 
 
 settings = Settings()

app/main.py

@@ -13,6 +13,7 @@
 from app.routers.notifications import router as notifications_router
 from app.routers.root import router as root_router
 from app.routers.seasons import router as seasons_router
+from app.routers.tokens import router as tokens_router
 from app.routers.users import router as users_router
 
 mimetypes.add_type("image/webp", ".webp")
@@ -67,6 +68,11 @@
     tags=["users"],
     prefix="/api",
 )
+app.include_router(
+    tokens_router,
+    tags=["tokens"],
+    prefix="/api",
+)
 
 app.mount("/static", StaticFiles(directory="static"), name="static")
 

app/routers/tokens.py

@@ -0,0 +1,30 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.ext.asyncio.session import AsyncSession
+
+from app.repositories.sessions import get_async_session
+from app.services.security import OAuth2PasswordRequestJson, UnauthorizedResponse
+from app.services.tokens import (
+    Token,
+    process_token_auth_user,
+)
+
+router = APIRouter(prefix="/tokens")
+
+
+@router.post(
+    "/users/auth",
+    responses={
+        status.HTTP_401_UNAUTHORIZED: {
+            "model": UnauthorizedResponse,
+        },
+    },
+    response_model=Token,
+    name="user_token_auth",
+)
+async def token_auth_user(
+    form_data: Annotated[OAuth2PasswordRequestJson, Depends()],
+    session: AsyncSession = Depends(get_async_session),
+) -> Token:
+    return await process_token_auth_user(session, form_data)

app/services/security.py

@@ -0,0 +1,36 @@
+from fastapi.param_functions import Body
+from pydantic import BaseModel
+from typing_extensions import Annotated, Doc
+
+
+class UnauthorizedResponse(BaseModel):
+    detail: str
+
+
+class OAuth2PasswordRequestJson:
+    def __init__(
+        self,
+        *,
+        username: Annotated[
+            str,
+            Body(),
+            Doc(
+                """
+                `username` string. The OAuth2 spec requires the exact field name
+                `username`.
+                """
+            ),
+        ],
+        password: Annotated[
+            str,
+            Body(),
+            Doc(
+                """
+                `password` string. The OAuth2 spec requires the exact field name
+                `password".
+                """
+            ),
+        ],
+    ):
+        self.username = username
+        self.password = password

app/services/tokens.py

@@ -0,0 +1,98 @@
+from copy import deepcopy
+from datetime import datetime, timedelta
+from typing import List
+
+from fastapi import HTTPException, status
+from jose import exceptions, jwt
+from pydantic import BaseModel, ConfigDict, Field
+from sqlalchemy.ext.asyncio.session import AsyncSession
+
+from app.configs import settings
+from app.repositories.models import User as UserModel, UserDoesNotExist
+from app.services.hashers import hasher
+from app.services.security import OAuth2PasswordRequestJson
+
+DEFAULT_JWT_EXPIRATION_TIME: int = 15 * 60
+
+
+def generate_jwt_signature(
+    payload: dict,
+    /,
+    *,
+    expiration_time: int = DEFAULT_JWT_EXPIRATION_TIME,
+    algorithm: str = "HS256",
+) -> str:
+    cleaned_payload: dict = deepcopy(payload)
+
+    cleaned_payload.update(
+        {
+            "exp": datetime.now() + timedelta(seconds=expiration_time),
+        }
+    )
+
+    return jwt.encode(cleaned_payload, settings.secret_key, algorithm=algorithm)
+
+
+class SignatureErrorBase(Exception):
+    """Base JWT Error"""
+
+
+class FatalSignatureError(SignatureErrorBase):
+    """Fatal Signature Error"""
+
+
+class SignatureExpiredError(SignatureErrorBase):
+    """Signature Expired Error"""
+
+
+def decode_jwt_signature(
+    token: str,
+    /,
+    *,
+    algorithms: List[str] = None,
+):
+    if algorithms is None:
+        algorithms = ["HS256"]
+
+    try:
+        return jwt.decode(token, settings.secret_key, algorithms=algorithms)
+    except (exceptions.JWSError, exceptions.JWSSignatureError):
+        raise FatalSignatureError()
+    except exceptions.ExpiredSignatureError:
+        raise SignatureExpiredError()
+
+
+class Token(BaseModel):
+    access_token: str = Field(alias="accessToken")
+
+    model_config = ConfigDict(
+        from_attributes=True,
+        populate_by_name=True,
+    )
+
+
+class TokenData(BaseModel):
+    uuid: str
+
+
+async def process_token_auth_user(
+    session: AsyncSession,
+    data: OAuth2PasswordRequestJson,
+    /,
+) -> Token:
+    try:
+        user: UserModel = await UserModel.get(
+            session,
+            data.username,
+            field=UserModel.username,
+        )
+    except UserDoesNotExist:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+    if not hasher.verify(data.password, user.password):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+
+    return Token(
+        access_token=generate_jwt_signature(
+            TokenData(uuid=str(user.uuid)).model_dump(by_alias=True)
+        )
+    )

requirements.txt

@@ -6,10 +6,13 @@ Babel==2.13.1
 boto3==1.33.10
 botocore==1.33.10
 certifi==2023.11.17
+cffi==1.16.0
 cfgv==3.4.0
 click==8.1.7
+cryptography==42.0.0
 distlib==0.3.7
 dnspython==2.5.0
+ecdsa==0.18.0
 email-validator==2.1.0.post1
 fastapi==0.104.1
 fastapi-pagination==0.12.13
@@ -38,12 +41,16 @@ platformdirs==4.0.0
 pluggy==1.3.0
 pre-commit==3.5.0
 priority==2.0.0
+pyasn1==0.5.1
+pycparser==2.21
 pydantic==2.5.1
 pydantic_core==2.14.3
 pytest==7.4.3
 python-dateutil==2.8.2
+python-jose==3.3.0
 python-multipart==0.0.6
 PyYAML==6.0.1
+rsa==4.9
 s3transfer==0.8.2
 setuptools==69.0.2
 six==1.16.0