How to implement refresh token

[wildsonc]

Is there any example of how to implement oauth2 with refresh token? I found mentions of refresh_token but I didn't find how to use it.

I'm trying to integrate with NextAuth.

[Alc-Alc]

Heya @wildsonc, can you add more information on what exactly you are trying achieve here? It is kinda ambiguous to me (unsure about others), maybe adding more details can help other members and maintainers to help you more 🙂

[wildsonc]

Using this OAuth2 example the refresh_token is returned but it is null, I would like to know the correct way to implement it, I did a draft, however, may not be the best/correct way to do this.

from datetime import timedelta
from os import environ
from typing import Any
from uuid import UUID

from jose import JWTError, jwt
from litestar import Litestar, Request, Response, post
from litestar.exceptions import HTTPException
from litestar.connection import ASGIConnection
from litestar.contrib.jwt import OAuth2Login, OAuth2PasswordBearerAuth, Token
from litestar.openapi.config import OpenAPIConfig
from pydantic import BaseModel


class User(BaseModel):
    id: UUID
    email: str

class RefreshToken(BaseModel):
    refresh_token: str


MOCK_DB: dict[str, User] = {}


async def retrieve_user_handler(token: "Token", connection: "ASGIConnection[Any, Any, Any, Any]") -> User | None:
    return MOCK_DB.get(token.sub)

oauth2_auth = OAuth2PasswordBearerAuth[User](
    retrieve_user_handler=retrieve_user_handler,
    token_secret=environ.get("JWT_SECRET", "abcd123"),
    token_url="/login",
    exclude=["/login", "refresh", "/schema"],
    default_token_expiration=timedelta(minutes=15),
)

@post("/login")
async def login_handler(request: "Request[Any, Any, Any]", data: "User") -> "Response[OAuth2Login]":
    MOCK_DB[str(data.id)] = data
    identifier = str(data.id)
    response = oauth2_auth.login(identifier=identifier)
    response.content['refresh_token'] = oauth2_auth.create_token(identifier, token_expiration=timedelta(hours=8))
    return response

@post("/refresh")
async def refresh_handler(request: "Request[Any, Any, Any]", data: "RefreshToken")-> "Response[OAuth2Login]":
    try:
        payload = jwt.decode(data.refresh_token, environ.get("JWT_SECRET", "abcd123"), algorithms=['HS256'])
        identifier = payload.get("sub")
        if identifier is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    response = oauth2_auth.login(identifier=identifier)
    del response.content['refresh_token']
    return response


openapi_config = OpenAPIConfig(
    title="My API",
    version="1.0.0",
)

app = Litestar(
    route_handlers=[login_handler, refresh_handler],
    on_app_init=[oauth2_auth.on_app_init],
    openapi_config=openapi_config,
    debug=True,
)

[scott2b]

I am confused about why you would pass OAuth2 tokens around inside of a JWT. My understanding is that most OAuth2 clients and client libraries are designed to work directly with access and refresh tokens according to the OAuth2 specification which has a token response that looks something like this:

{
  "access_token": "<access_token>",
  "token_type": "Bearer",
  "expires_in": 3600,
  "refresh_token": "<refresh_token>"
}

I would not expect an OAuth2 client to know what to do with the JWT. Is this a typical use case? Is there something I am missing?

[JacobCoffee]

@cofin

[pablo-tx]

I'm also looking into implementing the refresh token, similar to how it's done in the DRF simpleJWT lib