Backend fixed and features

packages/sshecret-admin/src/sshecret_admin/api/endpoints/auth.py

@@ -2,39 +2,86 @@
 
 # pyright: reportUnusedFunction=false
 import logging
-from typing import Annotated
-from fastapi import APIRouter, Depends, HTTPException, status
+from typing import Annotated, Literal
+from fastapi import APIRouter, Depends, Form, HTTPException, status
 from fastapi.security import OAuth2PasswordRequestForm
+from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from sshecret_admin.auth import Token, authenticate_user_async, create_access_token
+from sshecret_admin.auth import (
+    Token,
+    authenticate_user_async,
+    create_access_token,
+    create_refresh_token,
+    decode_token,
+)
 from sshecret_admin.core.dependencies import AdminDependencies
 
 LOG = logging.getLogger(__name__)
 
 
+class RefreshTokenForm(BaseModel):
+    """The refresh token form data."""
+
+    grant_type: Literal["refresh_token"]
+    refresh_token: str
+
+
 def create_router(dependencies: AdminDependencies) -> APIRouter:
     """Create auth router."""
     app = APIRouter()
 
     @app.post("/token")
     async def login_for_access_token(
-
         session: Annotated[AsyncSession, Depends(dependencies.get_async_session)],
         form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
     ) -> Token:
         """Login user and generate token."""
-        user = await authenticate_user_async(session, form_data.username, form_data.password)
+        user = await authenticate_user_async(
+            session, 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"},
             )
+
+        token_data: dict[str, str] = {"sub": user.username}
         access_token = create_access_token(
             dependencies.settings,
-            data={"sub": user.username},
+            data=token_data,
+        )
+        refresh_token = create_refresh_token(dependencies.settings, data=token_data)
+
+        return Token(
+            access_token=access_token, refresh_token=refresh_token, token_type="bearer"
+        )
+
+    @app.post("/refresh")
+    async def refresh_token(
+        form_data: Annotated[RefreshTokenForm, Form()],
+    ) -> Token:
+        """Refresh access token."""
+        LOG.info("Refresh token data: %r", form_data)
+        claims = decode_token(dependencies.settings, form_data.refresh_token)
+        if not claims:
+            LOG.info("Could not decode claims")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Invalid refresh token",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+
+        token_data: dict[str, str] = {"sub": claims.sub}
+        access_token = create_access_token(
+            dependencies.settings,
+            data=token_data,
+        )
+        refresh_token = create_refresh_token(dependencies.settings, data=token_data)
+
+        return Token(
+            access_token=access_token, refresh_token=refresh_token, token_type="bearer"
         )
-        return Token(access_token=access_token, token_type="bearer")
 
     return app