Extract auth cookie session auth service

Author: koldakov

futuramaapi/apps/fastapi.py

@@ -80,9 +80,9 @@ def _setup_middlewares(self) -> None:
             self.add_middleware(APIRequestsCounter)
 
     def _setup_routers(self) -> None:
-        from futuramaapi.routers import api_router, graphql_router, root_router, views_router  # noqa: PLC0415
+        from futuramaapi.routers import api_router, graphql_router, views_router  # noqa: PLC0415
 
-        for router in [api_router, graphql_router, root_router, views_router]:
+        for router in [api_router, graphql_router, views_router]:
             self.include_router(router)
 
     def _setup_static(self) -> None:

futuramaapi/routers/__init__.py

@@ -9,7 +9,6 @@
 from .rest.links import router as links_router
 from .rest.notifications import router as notification_router
 from .rest.randoms import router as randoms_router
-from .rest.root import router as root_router
 from .rest.seasons import router as seasons_router
 from .rest.tokens import router as tokens_router
 from .rest.users import router as users_router
@@ -19,7 +18,6 @@
 __all__ = [
     "api_router",
     "graphql_router",
-    "root_router",
     "views_router",
 ]
 

futuramaapi/routers/rest/root/__init__.py

@@ -1,16 +1,15 @@
 from typing import Annotated, Literal
 
-from fastapi import Depends, Form, HTTPException, Request, status
-from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from fastapi import Depends, Form, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
 from pydantic import SecretStr
 from sqlalchemy.ext.asyncio.session import AsyncSession
 
 from futuramaapi.mixins.pydantic import DecodedTokenError
-from futuramaapi.repositories.models import AuthSessionModel
 from futuramaapi.repositories.session import get_async_session
 from futuramaapi.routers.exceptions import ModelNotFoundError
 from futuramaapi.routers.rest.tokens.schemas import DecodedUserToken
-from futuramaapi.routers.rest.users.schemas import User, UserPasswordError, UserUpdateRequest
+from futuramaapi.routers.rest.users.schemas import User, UserUpdateRequest
 
 _oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/tokens/users/auth")
 
@@ -78,24 +77,3 @@ def password_from_form_data(
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Passwords mismatch") from None
 
     return UserUpdateRequest(password=password1.get_secret_value())
-
-
-async def cookie_user_from_form_data(
-    request: Request,
-    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
-    session: AsyncSession = Depends(get_async_session),  # noqa: B008
-) -> User | None:
-    try:
-        user: User = await User.auth(session, form_data.username, form_data.password)
-    except (ModelNotFoundError, UserPasswordError):
-        return None
-
-    auth_session: AuthSessionModel = AuthSessionModel()
-    auth_session.user_id = user.id
-    auth_session.ip_address = request.client.host
-
-    session.add(auth_session)
-    await session.commit()
-
-    user._cookie_session = auth_session.key
-    return user

futuramaapi/routers/rest/root/api.py

@@ -0,0 +1,78 @@
+from typing import ClassVar
+
+from fastapi import Request, status
+from fastapi.responses import RedirectResponse
+from pydantic import SecretStr
+from sqlalchemy import Result, Select, select
+from sqlalchemy.exc import NoResultFound
+
+from futuramaapi.repositories.models import AuthSessionModel, UserModel
+from futuramaapi.routers.services import BaseSessionService
+
+
+class AuthCookieSessionUserService(BaseSessionService[RedirectResponse]):
+    username: str
+    password: SecretStr
+
+    cookie_auth_key: ClassVar[str] = "Authorization"
+
+    @property
+    def request(self) -> Request:
+        if self.context is None:
+            raise AttributeError("Request is not defined.")
+
+        if "request" not in self.context:
+            raise AttributeError("Request is not defined.")
+
+        return self.context["request"]
+
+    @property
+    def __get_user_statement(self) -> Select[tuple[UserModel]]:
+        return select(UserModel).where(UserModel.username == self.username)
+
+    async def _get_user(self) -> UserModel:
+        result: Result[tuple[UserModel]] = await self.session.execute(self.__get_user_statement)
+        try:
+            return result.scalars().one()
+        except NoResultFound:
+            raise
+
+    async def _get_auth_session(self, user: UserModel, /) -> AuthSessionModel:
+        auth_session: AuthSessionModel = AuthSessionModel()
+        auth_session.user_id = user.id
+        auth_session.ip_address = self.request.client.host
+
+        self.session.add(auth_session)
+        await self.session.commit()
+
+        return auth_session
+
+    def _build_response(self, auth_session: AuthSessionModel, /) -> RedirectResponse:
+        response: RedirectResponse = RedirectResponse(
+            "/",
+            status_code=status.HTTP_302_FOUND,
+        )
+        response.set_cookie(
+            self.cookie_auth_key,
+            value=auth_session.key,
+            expires=AuthSessionModel.cookie_expiration_time,
+        )
+        return response
+
+    async def process(self, *args, **kwargs) -> RedirectResponse:
+        try:
+            user: UserModel = await self._get_user()
+        except NoResultFound:
+            return RedirectResponse(
+                url="/auth",
+                status_code=status.HTTP_302_FOUND,
+            )
+
+        if not self.hasher.verify(self.password.get_secret_value(), user.password):
+            return RedirectResponse(
+                url="/auth",
+                status_code=status.HTTP_302_FOUND,
+            )
+
+        auth_session: AuthSessionModel = await self._get_auth_session(user)
+        return self._build_response(auth_session)

futuramaapi/routers/rest/users/dependencies.py

@@ -1,8 +1,12 @@
-from fastapi import APIRouter, Request, Response, status
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Request, Response, status
 from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
 from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
+from fastapi.security import OAuth2PasswordRequestForm
 
 from futuramaapi.routers.services.about.get_about import GetAboutService
+from futuramaapi.routers.services.auth.auth_cookie_session_user import AuthCookieSessionUserService
 from futuramaapi.routers.services.auth.get_user_auth import GetUserAuthService
 from futuramaapi.routers.services.auth.logout_cookie_session_user import LogoutCookieSessionUserService
 from futuramaapi.routers.services.changelog.get_changelog import GetChangelogService
@@ -144,3 +148,25 @@ async def logout_cookie_session_user(
         },
     )
     return await service()
+
+
+@router.post(
+    "/auth",
+    include_in_schema=False,
+    name="auth_cookie_session_user",
+)
+async def auth_cookie_session_user(
+    request: Request,
+    form_data: Annotated[
+        OAuth2PasswordRequestForm,
+        Depends(),
+    ],
+) -> RedirectResponse:
+    service: AuthCookieSessionUserService = AuthCookieSessionUserService(
+        username=form_data.username,
+        password=form_data.password,
+        context={
+            "request": request,
+        },
+    )
+    return await service()