Add OAuth2 refresh token Form dependency in security/oauth2.py

[perkfly]

First check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.
• After submitting this, I commit to:
  • Read open issues with questions until I find 2 issues where I can help someone and add a comment to help there.
  • Or, I already hit the "watch" button in this repository to receive notifications and I commit to help at least 2 people that ask questions in the future.
  • Implement a Pull Request for a confirmed bug.

Description

security/oauth2.py already contains OAuth2PasswordRequestForm, why don't we add OAuth2RerefreshRequestForm to FastAPI as well? It is also well defined in the OAuth RFC. To implement a complete OAuth2 with FastAPI, token refresh is needed. Also, the docs can be updated with a full example.

As the RFC states, the refresh request has to be exactly like this:

POST /token HTTP/1.1
     Host: server.example.com
     Authorization: Basic czZCaGRSa3F0MzpnWDFmQmF0M2JW
     Content-Type: application/x-www-form-urlencoded

     grant_type=refresh_token&refresh_token=tGzv3JOkF0XG5Qx2TlKWIA

The solution you would like

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2RefreshRequestForm

app = FastAPI()


@app.route("/refresh")
def refresh_token(form: OAuth2RefreshRequestForm = Depends()):
    # check if the refresh token is valid
    return {
       "access_token":"2YotnFZFEjr1zCsicMWpAA",
       "token_type":"example",
       "expires_in":3600
    }

A possible implementation would be:

class OAuth2RefreshRequestForm:

    def __init__(
        self,
        grant_type: str = Form(None, regex="refresh_token"),
        refresh_token: str = Form(...)
    ):
        self.grant_type = grant_type
        self.refresh_token = refresh_token

It is indeed simple, but having it in builtin still saves users some time to look up in the RFC. Also, the tutorial in docs can be more complete.

Additional context

RFC6749 section 6 on refresh token grant

[jw-00000]

Note that according to the spec, the refresh request should also be to the same /token endpoint as the original access token (and not to /refresh as in the code example).

[Jan-Jasek]

I can take a shot at the implementation

[mwvaughn]

@Jan-Jasek - do you still have plans to work on this feature?

[alexandar1000]

Hi guys thanks for the work. Is this functionality still in the works? Also, @Jan-Jasek do you need help with it?

[ccueto36]

Any update on this feature? This would be a huge improvement to the current implementation.

[kbernst30]

Is this still being looked at by anybody? @Jan-Jasek ? I'm willing to take a stab at implementing this if nobody else is.

[Jan-Jasek]

Is this still being looked at by anybody? @Jan-Jasek ? I'm willing to take a stab at implementing this if nobody else is.

@kbernst30 I tried the implementation, but did not like the result, because OAuth2 spec has a single token endpoint, that expects different set of parameters based on response_type field and I was not sure how to expose the conditional pydantic models that would propagate to swagger nicely.

So feel free to take the stab.

[benlau6]

I have integrated the refresh token workflow into the same endpoint. We could just simply change fastapi.security.OAuth2PasswordRequestForm by altering grant_type and adding refresh_token attribute as follows.

class OAuth2PasswordAndRefreshRequestForm:
    """Modified from fastapi.security.OAuth2PasswordRequestForm"""

    def __init__(
        self,
        grant_type: str = Form(default=None, regex="password|refresh_token"),
        username: str = Form(default=""),
        password: str = Form(default=""),
        refresh_token: str = Form(default=""),
        scope: str = Form(default=""),
        client_id: str | None = Form(default=None),
        client_secret: str | None = Form(default=None),
    ):
        self.grant_type = grant_type
        self.username = username
        self.password = password
        self.refresh_token = refresh_token
        self.scopes = scope.split()
        self.client_id = client_id
        self.client_secret = client_secret

And then alter the schemas.Token to return

class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str

In the login endpoint, implement a refresh token authentication function (which could make use of the access token authentication logic), and make a conditional statement on grant_type.

@router.post("/token", response_model=schemas.Token)
async def login_for_tokens(
    response: Response,
    session: AsyncSession = Depends(get_session),
    form_data: OAuth2PasswordAndRefreshRequestForm = Depends(),
):
    if form_data.grant_type == "refresh_token":
        user = authenticate_user_from_token(
            session=session,
            token=form_data.refresh_token,
            secret_key="REFRESH_SECRET_KEY",
        )
    else:
        user = authenticate_user(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"},
        )
    return create_tokens_from_user(user=user, response=response)

Finally, implementing the create_tokens_from_user function to create access token and refresh token to return

def create_tokens_from_user(user: models.User, response: Response) -> dict[str, str]:
    jwt_data = {"sub": str(user.id)}

    access_token_expires = timedelta(minutes="15")
    access_token = create_access_token(
        data=jwt_data,
        expires_delta=access_token_expires,
    )

    refresh_token_expires = timedelta(minutes="60")
    refresh_token = create_refresh_token(
        data=jwt_data,
        expires_delta=refresh_token_expires,
    )
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
    }

All the details not mentioned above are well explained in the FastAPI documentation.

[yavuzakyazici]

I was trying this solution but ...

   user = authenticate_user_from_token()

functions implementationis not shown here. Anybody?

[yavuzakyazici]

OK, I figured it out and got it working for my project..

Thanks to benlau6

I modified the OAuth2PasswordAndRefreshRequestForm class but extended it from OAuth2PasswordRequestForm to make it a bit more understandable I guess

class OAuth2PasswordAndRefreshRequestForm(OAuth2PasswordRequestForm):
    """Modified from fastapi.security.OAuth2PasswordRequestForm"""

    def __init__(
        self,
        grant_type: str = Form(default=None, regex="password|refresh_token"),
        username: str = Form(default=""),
        password: str = Form(default=""),
        refresh_token: str = Form(default=""),
        scope: str = Form(default=""),
        client_id: str | None = Form(default=None),
        client_secret: str | None = Form(default=None),
    ):
        super().__init__(
            grant_type=grant_type,
            username=username,
            password=password,
            client_id=client_id,
            client_secret=client_secret,
        )
        self.scopes = scope.split()
        self.refresh_token = refresh_token

I also had not understood what was this function doing since name confused me.

user = authenticate_user_from_token()

So I changed it to this

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)], db: Session = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("username")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = get_user(username, db)
    if user is None:
        raise credentials_exception
    elif user.activated == False:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User account is deactivated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

Originally at FastAPI docs the function above was get_current_active_user() and had a separate function for get_current_user().
It seemed more practical to combine the two with one elif ( see above )
And create_tokens_from_user() function is pretty much the same except, I am not using response: Response variable for login or create_tokens_from_user()..
This way login works

[Actticus]

Any update here?

[sanzoghenzo]

Sorry to necropost this, but I'm running into the same issue here.

One question: after adding the above code to an app, is it possible to tell swagger UI to use the refresh token to keep the user authenticated?

[N1ntek]

any updates?