feat(#10): add authentication process

Author: ArielMAJ

api/database/models/users.py

@@ -1,6 +1,9 @@
+from typing import Union
+
 from api.database.models.model_base import ModelBase
 from sqlalchemy import String
 from sqlalchemy.orm import Mapped, mapped_column
+from typing_extensions import Self
 
 
 class User(ModelBase):
@@ -13,3 +16,7 @@ class User(ModelBase):
         String(255), nullable=False, unique=False, index=True
     )
     password: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+
+    @classmethod
+    async def get_by_email(cls, email: str) -> Union[Self, None]:
+        return await cls.get(cls.email == email)

api/entrypoints/auth.py

@@ -0,0 +1,25 @@
+from typing import Annotated, Union
+
+from api.database.models.users import User
+from api.exceptions.http_exceptions import CredentialsException
+from api.schemas.auth import Token
+from api.services.auth import AuthService
+from fastapi import APIRouter, Depends
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+
+router = APIRouter()
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
+
+
+@router.post("/token", response_model=Token)
+async def login_for_access_token(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+) -> Token:
+    user: Union[User, None] = await AuthService.authenticate_user(
+        form_data.username, form_data.password
+    )
+    if not user:
+        raise CredentialsException()
+    access_token, expires_at = AuthService.create_access_token(data={"sub": user.email})
+    return Token(access_token=access_token, token_type="bearer", expires_at=expires_at)

api/entrypoints/router.py

@@ -1,10 +1,13 @@
-from api.entrypoints import monitoring, random_response, root_response, user
+from api.entrypoints import auth, monitoring, random_response, root_response, user
 from fastapi.routing import APIRouter
 
+user.router.include_router(user.authenticated_router)
+
 router = APIRouter()
 router.include_router(monitoring.router, tags=["Monitoring"])
 router.include_router(
     random_response.router, prefix="/random_number", tags=["Random Number"]
 )
 router.include_router(root_response.router, prefix="", tags=["Root Response"])
 router.include_router(user.router, prefix="/user", tags=["User"])
+router.include_router(auth.router, prefix="/auth", tags=["Auth"])

api/entrypoints/user.py

@@ -1,13 +1,18 @@
-from typing import List
+from typing import Annotated, List
 
+from api.database.models.users import User
 from api.schemas.user import UserCreate, UserOut
+from api.services.auth import AuthService
 from api.services.user_service import UserService
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
 
 router = APIRouter()
+authenticated_router = APIRouter(
+    dependencies=[Depends(AuthService.get_current_active_user)]
+)
 
 
-@router.get("/", response_model=List[UserOut])
+@authenticated_router.get("/", response_model=List[UserOut])
 async def get_users() -> List[UserOut]:
     """
     Retrieve a list of all users.
@@ -18,7 +23,17 @@ async def get_users() -> List[UserOut]:
     return await UserService().get_all()
 
 
-@router.get("/{user_id}", response_model=UserOut)
+@authenticated_router.get("/me", response_model=UserOut)
+async def read_user_me(
+    current_user: Annotated[User, Depends(AuthService.get_current_active_user)]
+):
+    """
+    Retrieve the currently authenticated user by Bearer token.
+    """
+    return current_user
+
+
+@authenticated_router.get("/{user_id}", response_model=UserOut)
 async def get_user(user_id: int) -> UserOut:
     """
     Retrieve a user by ID.
@@ -46,7 +61,7 @@ async def create_user(user: UserCreate):
     return await UserService().create_user(user)
 
 
-@router.put("/{user_id}", response_model=None)
+@authenticated_router.put("/{user_id}", response_model=None)
 async def update_user(user_id: int, user: UserCreate) -> None:
     """
     Update a user by ID.
@@ -61,7 +76,7 @@ async def update_user(user_id: int, user: UserCreate) -> None:
     return await UserService().update_user(user_id, user)
 
 
-@router.delete("/{user_id}", response_model=None)
+@authenticated_router.delete("/{user_id}", response_model=None)
 async def delete_user(user_id: int) -> None:
     """
     Delete a user by ID.

api/exceptions/http_exceptions.py

@@ -8,3 +8,12 @@ def __init__(self, model: ModelBase):
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"{model.__name__} not found",
         )
+
+
+class CredentialsException(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not validate credentials",
+            headers={"WWW-Authenticate": "Bearer"},
+        )

api/schemas/auth.py

@@ -0,0 +1,18 @@
+from datetime import datetime
+from typing import Union
+
+from pydantic import BaseModel, field_serializer
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+    expires_at: datetime
+
+    @field_serializer("expires_at")
+    def serialize_expires_at(self, value: datetime):
+        return value.strftime("%d-%m-%Y %H:%M:%S")
+
+
+class TokenData(BaseModel):
+    email: Union[str, None] = None

api/schemas/base_db_schema.py

@@ -11,7 +11,7 @@ class BaseDBSchema(BaseModel):
     deleted_at: Optional[datetime]
 
     class Config:
-        orm_mode = True
+        from_attributes = True
 
     @field_serializer("created_at", "updated_at", "deleted_at")
     def serialize_dt(self, dt: Optional[datetime]):

api/schemas/user.py

@@ -1,13 +1,22 @@
 from api.schemas.base_db_schema import BaseDBSchema
-from pydantic import BaseModel
+from api.services.auth import AuthService
+from pydantic import BaseModel, field_serializer
 
 
 class UserCreate(BaseModel):
     name: str
     email: str
     password: str
 
+    @field_serializer("password")
+    def hash_password(self, password: str) -> str:
+        return AuthService.get_password_hash(password)
+
 
 class UserOut(BaseDBSchema):
     name: str
     email: str
+
+
+class UserInDB(UserOut):
+    hashed_password: str

api/services/auth.py

@@ -0,0 +1,72 @@
+from datetime import datetime, timedelta, timezone
+from typing import Annotated, Union
+
+import jwt
+from api.database.models.users import User
+from api.exceptions.http_exceptions import CredentialsException
+from api.schemas.auth import TokenData
+from fastapi import Depends
+from fastapi.security import OAuth2PasswordBearer
+from passlib.context import CryptContext
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
+
+
+SECRET_KEY = "ef555b4c8637c33623fe8e91ba7256725e7e2a1bcc75fe84acb189bcaa6c8693"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+
+class AuthService:
+    @staticmethod
+    def verify_password(plain_password, hashed_password):
+        return pwd_context.verify(plain_password, hashed_password)
+
+    @staticmethod
+    def get_password_hash(password):
+        return pwd_context.hash(password)
+
+    @staticmethod
+    async def authenticate_user(username: str, password: str):
+        user: Union[User, None] = await User.get_by_email(username)
+        if not user:
+            return False
+        if not AuthService.verify_password(password, user.password):
+            return False
+        return user
+
+    @staticmethod
+    def create_access_token(
+        data: dict, expires_delta_in_minutes: Union[int, None] = None
+    ):
+        to_encode = data.copy()
+        expires_at = datetime.now(timezone.utc) + timedelta(
+            minutes=expires_delta_in_minutes or ACCESS_TOKEN_EXPIRE_MINUTES
+        )
+        to_encode.update({"exp": expires_at})
+        encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+        return encoded_jwt, expires_at
+
+    # @staticmethod
+    async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
+        try:
+            payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+            username: Union[str, None] = payload.get("sub")
+            if username is None:
+                raise CredentialsException()
+            token_data = TokenData(email=username)
+        except jwt.InvalidTokenError:
+            raise CredentialsException()
+        user: Union[User, None] = await User.get_by_email(token_data.email)
+        if user is None:
+            raise CredentialsException()
+        return user
+
+    @staticmethod
+    async def get_current_active_user(
+        current_user: Annotated[User, Depends(get_current_user)],
+    ):
+        if current_user.deleted_at:
+            raise CredentialsException()
+        return current_user