refresh_token_required is unuseful

[JackyNiu]

this is my code copyed in document authx document
method: refresh_token_required raise Exception authx.exceptions.MissingTokenError: No token found in request from '[]'

from pydantic import BaseModel
from fastapi import FastAPI, Depends, HTTPException
from authx import AuthX, TokenPayload, AuthXConfig

auth_config = AuthXConfig()

auth_config.JWT_ALGORITHM = 'HS256'
auth_config.JWT_SECRET_KEY = 'SECRET_KEY'

auth_config.JWT_TOKEN_LOCATION = ['headers']

app = FastAPI()
security = AuthX(auth_config)


class LoginForm(BaseModel):
    username: str
    password: str


@app.post('/login')
def login(data: LoginForm):
    if data.username == "test" and data.password == "test":
        access_token = security.create_access_token(data.username)
        refresh_token = security.create_refresh_token(data.username)
        return {
            "access_token": access_token,
            "refresh_token": refresh_token
        }
    raise HTTPException(401, "Bad username/password")


@app.post('/refresh')
def refresh(
    refresh_payload: TokenPayload = Depends(security.refresh_token_required)
):
    """
    TODO refresh_token_required error
    """
    access_token = security.create_access_token(refresh_payload.sub)
    return {"access_token": access_token}


@app.get('/protected', dependencies=[Depends(security.access_token_required)])
def protected():
    return "You have access to this protected resource"


if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, port=8000)

[yokoberek]

Is there any solution? I am also having the same problem.

Edit:
For now, I resolved this by changing JSON_TOKEN_LOCATION from headers to json and sending the refresh token in the body instead of the request header.

[yezz123]

Hey @JackyNiu @yokoberek,

I will investigate this over the next few days to see how I can solve it. From what I understand, it might be related to passing the token in the request header. 🙏🏻

[HAWK-Soft]

Same for me. Thanks for the tip @yokoberek it works now

[Batres35]

Same for me. Someone found a solution for this?

What I did while the issue is being fixed is instead of creating a refresh_token directly when a user logs in, I create another access_token with an expiration time of 60 days and use it as a refresh_token. I know it's not the right way to do it, but it works.

[themutesinger]

Hi guys, I'm here with exactly the same problem.

[yezz123]

Hello Everyone! Sorry for the late reply and bunch of tries to fix this issue here:

from pydantic import BaseModel
from fastapi import FastAPI, Depends, HTTPException, Request
from authx import AuthX, TokenPayload, AuthXConfig

auth_config = AuthXConfig()

auth_config.JWT_ALGORITHM = 'HS256'
auth_config.JWT_SECRET_KEY = 'SECRET_KEY'

# Configure token locations for both access and refresh tokens
auth_config.JWT_TOKEN_LOCATION = ['headers', 'json']

# Make sure the header type is correctly set
auth_config.JWT_HEADER_TYPE = 'Bearer'

# Set the JSON keys for tokens
auth_config.JWT_REFRESH_JSON_KEY = 'refresh_token'
auth_config.JWT_JSON_KEY = 'access_token'  # Set the JSON key for access tokens

app = FastAPI()
security = AuthX(auth_config)


class LoginForm(BaseModel):
    username: str
    password: str


class RefreshForm(BaseModel):
    refresh_token: str


class ProtectedForm(BaseModel):
    access_token: str


@app.post('/login')
def login(data: LoginForm):
    if data.username == "test" and data.password == "test":
        access_token = security.create_access_token(data.username)
        refresh_token = security.create_refresh_token(data.username)
        return {
            "access_token": access_token,
            "refresh_token": refresh_token
        }
    raise HTTPException(401, "Bad username/password")


@app.post('/refresh')
async def refresh(request: Request, refresh_data: RefreshForm = None):
    """
    Refresh endpoint - creates a new access token using a refresh toke.n
    
    Can accept the refresh token either:
    1. In the Authorization header
    2. In the request body as JSON
    """
    try:
        # First try to get the refresh token from the Authorization header
        try:
            refresh_payload = await security.refresh_token_required(request)
        except Exception as header_error:
            if not refresh_data or not refresh_data.refresh_token:
                # If we don't have a token in either place, raise the original error
                raise header_error

            # Manually decode and verify the refresh token
            token = refresh_data.refresh_token
            refresh_payload = security.verify_token(
                token,
                verify_type=True,
                type="refresh"
            )
        # Create a new access token
        access_token = security.create_access_token(refresh_payload.sub)
        return {"access_token": access_token}
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e)) from e


@app.get('/protected')
async def protected(request: Request, protected_data: ProtectedForm = None):
    """
    Protected endpoint - requires a valid access token
    
    Can accept the access token either:
    1. In the Authorization header
    2. In the request body as JSON
    """
    try:
        # First try to get the access token from the Authorization header
        try:
            await security.access_token_required(request)
        except Exception as header_error:
            if not protected_data or not protected_data.access_token:
                # If we don't have a token in either place, raise the original error
                raise header_error

            # Manually decode and verify the access token
            token = protected_data.access_token
            security.verify_token(
                token,
                verify_type=True,
                type="access"
            )
        
        return "You have access to this protected resource"
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e)) from e


if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, port=8000)

this is the best way now you can guys use to fix this issues until i will release a new release

[C3EQUALZz]

@yezz123 Hello! Can you give new example how to verify token? Method "verify_token" doesn't have param "type" now, how I see. We must write like this?

@router.post(
    "/refresh/",
    status_code=status.HTTP_200_OK,
    response_model_exclude_none=True,
    summary="Endpoint for user refreshing",
)
async def refresh(
        security: FromDishka[AuthX],
        token: str = Depends(oauth2_scheme),
) -> TokenResponse:
    try:
        refresh_token_payload = security.verify_token(token=RequestToken(token=token, location="headers"))
        return TokenResponse(access_token=security.create_access_token(refresh_token_payload.sub))
    except AuthXException as e:
        logger.error(e)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))

    except ApplicationException as e:
        logger.error(e)
        raise HTTPException(status_code=e.status, detail=e.message)

UPD: This code works, this is example reference for updating tokens:

@router.post(
    "/refresh/",
    status_code=status.HTTP_200_OK,
    response_model_exclude_none=True,
    summary="Endpoint for user refreshing",
)
async def refresh(
        security: FromDishka[AuthX],
        token: str = Depends(oauth2_scheme),
) -> TokenResponse:
    try:
        refresh_token_payload = security.verify_token(token=RequestToken(
            token=token,
            location="headers",
            type="refresh"
        ))
        return TokenResponse(access_token=security.create_access_token(refresh_token_payload.sub))
    except AuthXException as e:
        logger.error(e)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))

    except ApplicationException as e:
        logger.error(e)
        raise HTTPException(status_code=e.status, detail=e.message)

[eighthcolor22]

@C3EQUALZz Thank you. It works.

[jimbob88]

Just my 2 cents, I believe the issue occurs here:

authx/authx/main.py

Lines 271 to 273 in 51fab94

	if locations is None:
	default_locations = set(self.config.JWT_TOKEN_LOCATION)
	locations = list(default_locations.intersection(["cookies", "json"]) if refresh else default_locations)

The intersection prevents one from supplying the refresh token in the header because it overrides the config. Hence the error:

No token found in request from '[]'

When you have the following authx config:

config = AuthXConfig(
    JWT_ALGORITHM="HS256",
    JWT_SECRET_KEY=settings.jwt_secret,
    JWT_TOKEN_LOCATION=["headers"],
)

auth = AuthX[User](config=config)

When this section of the code is executed:

>>> list(set(["headers"]).intersection(["cookies", "json"]))
[]

This intersection exists because the header verifier does not support using a refresh token bearer:

authx/authx/core.py

Lines 14 to 28 in 51fab94

	async def _get_token_from_headers(
	request: Request, config: AuthXConfig, refresh: bool = False, **kwargs: Any
	) -> RequestToken:
	"""Get access token from headers."""
	# Get Header
	auth_header: Optional[str] = request.headers.get(config.JWT_HEADER_NAME)
	if auth_header is None:
	raise MissingTokenError(f"Missing '{config.JWT_HEADER_TYPE}' in '{config.JWT_HEADER_NAME}' header.")
	if config.JWT_HEADER_TYPE:
	token = auth_header.replace(f"{config.JWT_HEADER_TYPE} ", "")
	else:
	token = auth_header
	return RequestToken(token=token, csrf=None, location="headers")

RFC 6749 states that refresh tokens should be stored in the POST body, I quote at length:

6. Refreshing an Access Token

If the authorization server issued a refresh token to the client, the
client makes a refresh request to the token endpoint by adding the
following parameters using the "application/x-www-form-urlencoded"
format per Appendix B with a character encoding of UTF-8 in the HTTP
request entity-body:

grant_type
REQUIRED. Value MUST be set to "refresh_token".

refresh_token
REQUIRED. The refresh token issued to the client.

scope
OPTIONAL. The scope of the access request as described by
Section 3.3. The requested scope MUST NOT include any scope
not originally granted by the resource owner, and if omitted is
treated as equal to the scope originally granted by the
resource owner.

Because refresh tokens are typically long-lasting credentials used to
request additional access tokens, the refresh token is bound to the
client to which it was issued. If the client type is confidential or
the client was issued client credentials (or assigned other
authentication requirements), the client MUST authenticate with the
authorization server as described in Section 3.2.1.

For example, the client makes the following HTTP request using
transport-layer security (with extra line breaks for display purposes
only):

 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

My solution was the same as others, I have moved from using headers to JSON POST body for refreshes (although, I believe this is still against the RFC, and a POST Form should be used with application/x-www-form-urlencoded).

[natalejoeraco1]

Same for me. Thanks for the tip @yokoberek it works now