import os
import pydantic
import socket
+import typing
from . import apiv1
from . import backend
@router.post("/user")
async def auth_user(credentials: fastapi.security.OAuth2PasswordRequestForm =
- fastapi.Depends()) -> AuthResponse:
+ fastapi.Depends()) -> fastapi.responses.JSONResponse:
# Set keytab to use
os.environ["KRB5_KTNAME"] = KERBEROS_KEYTAB
type="refresh", expires_after=REFRESH_TOKEN_EXPIRY_TIME)
# Send the response
- return AuthResponse(access_token=access_token, refresh_token=refresh_token)
+ data = AuthResponse(access_token=access_token, refresh_token=refresh_token)
+
+ # Serialize the JSON response
+ response = fastapi.responses.JSONResponse(content=data.model_dump())
+
+ # Set the refresh token as a cookie too for persistent browser storage
+ response.set_cookie(
+ "refresh_token",
+ refresh_token,
+ httponly=True,
+ samesite="Strict",
+
+ # Only send the cookie when performing a refresh
+ path="/api/v1/auth/refresh",
+ # Forget the cookie when the token expires
+ max_age=REFRESH_TOKEN_EXPIRY_TIME,
+ )
+
+ return response
class RefreshRequest(pydantic.BaseModel):
refresh_token: str
@router.post("/refresh")
-async def auth_refresh(data: RefreshRequest):
+async def auth_refresh(request: fastapi.Request, data: typing.Optional[RefreshRequest]=None):
+ # Fetch the refresh token from either the body or the cookie
+ refresh_token = data.refresh_token if data else request.cookies.get("refresh_token")
+ if not refresh_token:
+ raise fastapi.HTTPException(401, "No refresh token provided")
+
# Fetch the principal from the given token
- principal = get_principal(data.refresh_token)
+ principal = get_principal(refresh_token)
# XXX Check if the principal actually still exists
type="access", expires_after=ACCESS_TOKEN_EXPIRY_TIME)
# Send the response
- return AuthResponse(access_token=access_token, refresh_token=data.refresh_token)
+ return AuthResponse(access_token=access_token, refresh_token=refresh_token)
# Add everything to the app
apiv1.include_router(router)