IndominusByte / fastapi-jwt-auth

FastAPI extension that provides JWT Auth support (secure, easy to use, and lightweight)
http://indominusbyte.github.io/fastapi-jwt-auth/
MIT License
660 stars 153 forks source link

No authorize option in Swagger #34

Open himalacharya opened 3 years ago

himalacharya commented 3 years ago

By using this library, I donot see Authorize option in Swagger UI and have to use POSTMAN to test the app. It will be better if Authorize option is shown like this https://fastapi-mongo.herokuapp.com/docs image

IndominusByte commented 3 years ago

Nice idea to add this feature in fastapi-jwt-auth, I will add this in the next version thankyou 🙏

Heymann03 commented 3 years ago

@IndominusByte. Great work with your library. Started using it and then realized that there is no AUTHORIZE option like mentioned above. Can't we use the OAuthPasswordBearer along with this. I'm new to fastapi so i'm really not sure.

Also I want to change the expires_time for both the tokens. Where can I go about it?

IndominusByte commented 3 years ago

Glad to hear that ❤️ , unfortunately you can't do that along OAuthPasswordBearer, you can do that manually but it's to complex and I still testing only on jwt with a cookie, I will show you how to add them automatically and it's only working on jwt with cookie

import inspect, re
from fastapi import FastAPI
from fastapi.routing import APIRoute
from fastapi.openapi.utils import get_openapi

app = FastAPI()

@app.get("/items/")
async def read_items():
    return {"msg": "hello world"}

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title="Custom title",
        version="2.5.0",
        description="This is a very custom OpenAPI schema",
        routes=app.routes,
    )
    openapi_schema["info"]["x-logo"] = {
        "url": "https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png"
    }

    cookie_security_schemes = {
        "AuthJWTCookieAccess": {
            "type": "apiKey",
            "in": "header",
            "name": "X-CSRF-TOKEN"
        },
        "AuthJWTCookieRefresh": {
            "type": "apiKey",
            "in": "header",
            "name": "X-CSRF-TOKEN"
        }
    }
    refresh_token_cookie = {
        "name": "refresh_token_cookie",
        "in": "cookie",
        "required": False,
        "schema": {
            "title": "refresh_token_cookie",
            "type": "string"
        }
    }
    access_token_cookie = {
        "name": "access_token_cookie",
        "in": "cookie",
        "required": False,
        "schema": {
            "title": "access_token_cookie",
            "type": "string"
        }
    }

    if "components" in openapi_schema:
        openapi_schema["components"].update({"securitySchemes": cookie_security_schemes})
    else:
        openapi_schema["components"] = {"securitySchemes": cookie_security_schemes}

    api_router = [route for route in app.routes if isinstance(route, APIRoute)]

    for route in api_router:
        path = getattr(route, "path")
        endpoint = getattr(route,"endpoint")
        methods = [method.lower() for method in getattr(route, "methods")]

        for method in methods:
            # access_token
            if (
                re.search("jwt_required",inspect.getsource(endpoint)) or
                re.search("fresh_jwt_required",inspect.getsource(endpoint)) or
                re.search("jwt_optional",inspect.getsource(endpoint))
            ):
                try:
                    openapi_schema["paths"][path][method]['parameters'].append(access_token_cookie)
                except Exception:
                    openapi_schema["paths"][path][method].update({"parameters":[access_token_cookie]})

                # method GET doesn't need to pass X-CSRF-TOKEN
                if method != "get":
                    openapi_schema["paths"][path][method].update({
                        'security': [{"AuthJWTCookieAccess": []}]
                    })

            # refresh_token
            if re.search("jwt_refresh_token_required",inspect.getsource(endpoint)):
                try:
                    openapi_schema["paths"][path][method]['parameters'].append(refresh_token_cookie)
                except Exception:
                    openapi_schema["paths"][path][method].update({"parameters":[refresh_token_cookie]})

                # method GET doesn't need to pass X-CSRF-TOKEN
                if method != "get":
                    openapi_schema["paths"][path][method].update({
                        'security': [{"AuthJWTCookieRefresh": []}]
                    })

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

to change expires time you can check this example below, and for more detailed information you can go in there

from fastapi_jwt_auth import AuthJWT
from pydantic import BaseModel

class Settings(BaseModel):
    authjwt_secret_key: str = "secret"
    authjwt_access_token_expires: int = 300 # 5 minutes
    authjwt_refresh_token_expires: int = 300 # 5 minutes

@AuthJWT.load_config
def get_config():
    return Settings()
Heymann03 commented 3 years ago

@IndominusByte thanks for the example. I tried and It throws me an error like below:

 re.search("jwt_required",inspect.getsource(endpoint)) or
NameError: name 're' is not defined

Thanks for the expires thing I read it. Should it always be in seconds?

IndominusByte commented 3 years ago

ahh I forgot to add this on an example, try to import this @Heymann03

import inspect, re
Heymann03 commented 3 years ago

Ah! got it.. it's working now. However I'm not able to access the logout endpoint. In the response I send an empty access token to delete it. I'm getting the following error in the response body:

{
  "detail": "Missing CSRF Token"
}
IndominusByte commented 3 years ago

You must send X-CSRF-TOKEN in the header, for detail information you can check this issue #29

Heymann03 commented 3 years ago

@IndominusByte! Should I use POSTMAN separately to send this X-CSRF-TOKEN in the header. Or is it something to do with curl commands.

Compared to the example provided from the docs of fastapi there is something called _authenticateuser and _get_currentuser. How can that be achieved? Because I'm trying to replace fake_db with backend connection and I'm stuck at this point

@app.post("/token", response_model=Token)
async def login_for_access_token(db: AsyncSession = Depends(database.get_db), user: User, client: Client, user_role: UserRole, Authorize: AuthJWT = Depends()):
    user = await authenticate_user(db, form_data.username, form_data.password)

any help would appreciate.

Heymann03 commented 3 years ago

@IndominusByte. Is current user authenticated before using the end point '/protected' ?

IndominusByte commented 3 years ago

@IndominusByte! Should I use POSTMAN separately to send this X-CSRF-TOKEN in the header. Or is it something to do with curl commands.

Compared to the example provided from the docs of fastapi there is something called _authenticateuser and _get_currentuser. How can that be achieved? Because I'm trying to replace fake_db with backend connection and I'm stuck at this point

@app.post("/token", response_model=Token)
async def login_for_access_token(db: AsyncSession = Depends(database.get_db), user: User, client: Client, user_role: UserRole, Authorize: AuthJWT = Depends()):
    user = await authenticate_user(db, form_data.username, form_data.password)

any help would appreciate.

For right now you can use extending openAPI and custom swager according to your needs, or if you use jwt via cookie you can copy my code above and seem like this. I just provide code to generate authorize option in swagger for only jwt cookie, i will add this feature to the next version so stay tune 😁

https://user-images.githubusercontent.com/33957651/105730063-e3526f00-5f68-11eb-9f70-4f9834872b14.mov

IndominusByte commented 3 years ago

@IndominusByte. Is current user authenticated before using the end point '/protected' ?

Yes user must authenticate before can access endpoint '/protection'

Heymann03 commented 3 years ago

@IndominusByte. Thanks for the screen recording. I will try to incorporate the authorise in later stage. Right now I just need to produce access token refresh_token, retrieve current_user. But I'm confused now refresh_token is to produce new_access_token or refresh the existing access_token by using a newly created token? This is my token_refresh endpoint:

@app.post('/token_refresh')
def refresh(Authorize: AuthJWT = Depends()):
    Authorize.jwt_refresh_token_required()

    current_user = Authorize.get_jwt_subject()
    new_access_token = Authorize.create_access_token(subject=current_user)
    response = JSONResponse(content={"access_token_cookie": new_access_token,
                                     "user_name": user.display_name,
                                     "client_name": client.display_name,
                                     "role_id": user_role.role_id,
                                     "user_options": {}})

    # Set the JWT and CSRF double submit cookies in the response
    Authorize.set_access_cookies(new_access_token, response)
    return response

I'm also trying to set up db connection and to access user from there. Which is why my request body is little messed up. I'm not sure how send only required parameters.

IndominusByte commented 3 years ago

refresh_token for refresh the existing access_token by using a newly created token. the refresh token can help reduce the damage that can be done if an access token is stolen. However, if an attacker gets a refresh token they can keep generating new access tokens and accessing protected endpoints as though he was that user. We can help combat this by using the fresh tokens pattern, for further information

Heymann03 commented 3 years ago

refresh_token for refresh the existing access_token by using a newly created token. the refresh token can help reduce the damage that can be done if an access token is stolen. However, if an attacker gets a refresh token they can keep generating new access tokens and accessing protected endpoints as though he was that user. We can help combat this by using the fresh tokens pattern, for further information

This part I get it I read all the docs related JWT in cookie from your project. I do not want to use extending openAPI for authorization as of now. I just need to access refresh_token as of now. And I still get a missing csrf-token error. Are you there on gitter tiangolo/fastapi channel by any chance I can explain it there. I do not want to prolong it here. It's good to have complete solution for this issue once your done with your fix.

IndominusByte commented 3 years ago

yeah sure, you can search my name on gitter tiangolo/fastapi and chat with me privately 😁

saipavan9 commented 3 years ago

When can we expect this feature?? @IndominusByte . I really like the options given by this package but lack of this authorise option really hurts. Can we have some timeline for this feature??

IndominusByte commented 3 years ago

When can we expect this feature?? @IndominusByte . I really like the options given by this package but lack of this authorise option really hurts. Can we have some timeline for this feature??

I still have work to do, I'm sorry to late update this package, or what if I make an example for jwt in the header because jwt from cookie already exists on above? for right now you can manually add docs via extending OpenAPI while waiting for an update

pndiku commented 3 years ago

This took me many days of reading specs and comparing to other APIs.

If you're using Bearer Auth then it's simple to set up. You just need to specify the "securitySchemes" and "security" sections as below and you'll get an Authorize button.

Access your /login endpoint, get an access token then click the Authorize button and enter Bearer <access-token and you're good to go!

EDIT: Added a loop to only put the lock icon (for authorization) on routes requiring it

EDIT 2: Previous edit actually forced you to put manual operation_ids. This is cleaner.

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title = "My Auth API",
        version = "1.0",
        description = "An API with an Authorize Button",
        routes = app.routes,
    )

    openapi_schema["components"]["securitySchemes"] = {
        "Bearer Auth": {
            "type": "apiKey",
            "in": "header",
            "name": "Authorization",
            "description": "Enter: **'Bearer &lt;JWT&gt;'**, where JWT is the access token"
        }
    }

    # Get all routes where jwt_optional() or jwt_required
    api_router = [route for route in app.routes if isinstance(route, APIRoute)]

    for route in api_router:
        path = getattr(route, "path")
        endpoint = getattr(route,"endpoint")
        methods = [method.lower() for method in getattr(route, "methods")]

        for method in methods:
            # access_token
            if (
                re.search("jwt_required", inspect.getsource(endpoint)) or
                re.search("fresh_jwt_required", inspect.getsource(endpoint)) or
                re.search("jwt_optional", inspect.getsource(endpoint))
            ):
                openapi_schema["paths"][path][method]["security"] = [
                    {
                        "Bearer Auth": []
                    }
                ]

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi
ninjix commented 3 years ago

I was able to get the "pad lock" working on the Swagger page using the following code I based on the Bearer example provided by the https://github.com/testdrivenio/fastapi-jwt team using a JWTBearer subclassed from the HTTPBearer. See also: https://testdriven.io/blog/fastapi-jwt-auth/

# app/auth/bearer.py
# Based on example provide by fastapi-jwt project team
# https://github.com/testdrivenio/fastapi-jwt

from typing import Optional

from fastapi import Request, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi_jwt_auth import AuthJWT

class JWTBearer(HTTPBearer):
    def __init__(self, auto_error: bool = True):
        super(JWTBearer, self).__init__(auto_error=auto_error)

    async def __call__(self, request: Request):
        credentials: HTTPAuthorizationCredentials = await super(JWTBearer, self).__call__(request)
        if credentials:
            if not credentials.scheme == "Bearer":
                raise HTTPException(status_code=403, detail="Invalid authentication scheme.")
            if not self.verify_jwt(request, credentials.credentials):
                raise HTTPException(status_code=403, detail="Invalid token or expired token.")
            return credentials.credentials
        else:
            raise HTTPException(status_code=403, detail="Invalid authorization code.")

    # noinspection PyBroadException
    @staticmethod
    def verify_jwt(request: Request, token: Optional) -> bool:
        is_token_valid: bool = False
        auth_jwt = AuthJWT(request, token)
        try:
            payload = auth_jwt.get_raw_jwt()
        except Exception:
            payload = None
        if payload:
            is_token_valid = True
        return is_token_valid

Then on the JWT protected method use the dependencies attr on the decorator.

# protect endpoint with function jwt_required(), which requires
# a valid access token in the request headers to access.
@app.get('/user', dependencies=[Depends(JWTBearer())])
def user(authorize: AuthJWT = Depends()):
    authorize.jwt_required()

    current_user = authorize.get_jwt_subject()
    return {"user": current_user}

Note that I used get_raw_jwt() method to collect the payload. Maybe there is a better more secure way. Thoughts?

pndiku commented 3 years ago

Hey @ninjix this looks good but requires adding a dependency on every route.

If building an API from scratch it would be worth a try for sure, but I had 5 existing APIs I wanted to do this with and rewriting the swagger yaml was the path of least resistance! :) YMMV

I was able to get the "pad lock" working on the Swagger page using the following code I based on the Bearer example provided by the https://github.com/testdrivenio/fastapi-jwt team using a JWTBearer subclassed from the HTTPBearer. See also: https://testdriven.io/blog/fastapi-jwt-auth/

hanzohan commented 3 years ago

This took me many days of reading specs and comparing to other APIs.

If you're using Bearer Auth then it's simple to set up. You just need to specify the "securitySchemes" and "security" sections as below and you'll get an Authorize button.

Access your /login endpoint, get an access token then click the Authorize button and enter Bearer <access-token and you're good to go!

EDIT: Added a loop to only put the lock icon (for authorization) on routes requiring it

EDIT 2: Previous edit actually forced you to put manual operation_ids. This is cleaner.

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title = "My Auth API",
        version = "1.0",
        description = "An API with an Authorize Button",
        routes = app.routes,
    )

    openapi_schema["components"]["securitySchemes"] = {
        "Bearer Auth": {
            "type": "apiKey",
            "in": "header",
            "name": "Authorization",
            "description": "Enter: **'Bearer &lt;JWT&gt;'**, where JWT is the access token"
        }
    }

    # Get all routes where jwt_optional() or jwt_required
    api_router = [route for route in app.routes if isinstance(route, APIRoute)]

    for route in api_router:
        path = getattr(route, "path")
        endpoint = getattr(route,"endpoint")
        methods = [method.lower() for method in getattr(route, "methods")]

        for method in methods:
            # access_token
            if (
                re.search("jwt_required", inspect.getsource(endpoint)) or
                re.search("fresh_jwt_required", inspect.getsource(endpoint)) or
                re.search("jwt_optional", inspect.getsource(endpoint))
            ):
                openapi_schema["paths"][path][method]["security"] = [
                    {
                        "Bearer Auth": []
                    }
                ]

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

Libraries to import for running code above.

import re
import inspect
from fastapi.routing import APIRoute
from fastapi.openapi.utils import get_openapi
khashashin commented 2 years ago

Any update on this?

s3rius commented 2 years ago

This took me many days of reading specs and comparing to other APIs. If you're using Bearer Auth then it's simple to set up. You just need to specify the "securitySchemes" and "security" sections as below and you'll get an Authorize button. Access your /login endpoint, get an access token then click the Authorize button and enter Bearer <access-token and you're good to go! EDIT: Added a loop to only put the lock icon (for authorization) on routes requiring it EDIT 2: Previous edit actually forced you to put manual operation_ids. This is cleaner.

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title = "My Auth API",
        version = "1.0",
        description = "An API with an Authorize Button",
        routes = app.routes,
    )

    openapi_schema["components"]["securitySchemes"] = {
        "Bearer Auth": {
            "type": "apiKey",
            "in": "header",
            "name": "Authorization",
            "description": "Enter: **'Bearer &lt;JWT&gt;'**, where JWT is the access token"
        }
    }

    # Get all routes where jwt_optional() or jwt_required
    api_router = [route for route in app.routes if isinstance(route, APIRoute)]

    for route in api_router:
        path = getattr(route, "path")
        endpoint = getattr(route,"endpoint")
        methods = [method.lower() for method in getattr(route, "methods")]

        for method in methods:
            # access_token
            if (
                re.search("jwt_required", inspect.getsource(endpoint)) or
                re.search("fresh_jwt_required", inspect.getsource(endpoint)) or
                re.search("jwt_optional", inspect.getsource(endpoint))
            ):
                openapi_schema["paths"][path][method]["security"] = [
                    {
                        "Bearer Auth": []
                    }
                ]

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

Libraries to import for running code above.

import re
import inspect
from fastapi.routing import APIRoute
from fastapi.openapi.utils import get_openapi

I updated the function to be cleaner, more general and optimized it a little bit. Also this version adds response examples when auth failed.

import inspect
from typing import Any, Callable, Dict, Optional

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from fastapi.routing import APIRoute
from fastapi_jwt_auth import AuthJWT

def custom_openapi(  # noqa: C901, WPS210, WPS231
    app: FastAPI,
    response_schema: Optional[Dict[str, Any]] = None,
) -> Callable[[], Dict[str, Any]]:
    """
    Generate custom openAPI schema.

    This function generates openapi schema based on
    routes from FastAPI application.

    :param app: current FastAPI application.
    :param response_schema: json schema of the response.
    :returns: actual schema generator.
    """

    def openapi_generator() -> Dict[str, Any]:  # noqa: WPS210, WPS231, WPS430
        # If api schema already exists we do nothing.
        if app.openapi_schema:
            return app.openapi_schema

        # Generate base API schema based on
        # current application properties.
        openapi_schema = get_openapi(
            title=app.title,
            version=app.version,
            description=app.description,
            routes=app.routes,
            license_info=app.license_info,
            contact=app.contact,
            terms_of_service=app.terms_of_service,
            tags=app.openapi_tags,
            servers=app.servers,
            openapi_version=app.openapi_version,
        )

        # Header type from configuration.
        header_type = AuthJWT._header_type  # noqa: WPS437
        scheme_name = f"{header_type} auth"

        if "components" not in openapi_schema:
            openapi_schema["components"] = {}

        # Add security scheme configuration
        # only if if jwt must be in headers.
        # Because cokies schema doesn't work on
        # swagger page.
        if "headers" in AuthJWT._token_location:  # noqa: WPS437
            openapi_schema["components"]["securitySchemes"] = {
                scheme_name: {
                    "type": "apiKey",
                    "in": "header",
                    "name": AuthJWT._header_name,  # noqa: WPS437
                    "description": (
                        f"Enter: **'{header_type} &lt;JWT&gt;'**, "
                        "where JWT is the access token"
                    ),
                    "bearerFormat": header_type,
                },
            }

        # Get all routes where jwt_optional() or jwt_required
        api_router = [route for route in app.routes if isinstance(route, APIRoute)]

        for route in api_router:
            # Getting python sources of a handler function.
            sources = inspect.getsource(route.endpoint)
            # Handler requires JWT auth if any of these words
            # exist in handler's source code.
            jwt_flags = ["jwt_required", "fresh_jwt_required", "jwt_optional"]
            for method in route.methods:
                # If handler has jwt_flag inside its source code
                # we add security schema in swagger spec.
                if any([flag in sources for flag in jwt_flags]):
                    method_info = openapi_schema["paths"][route.path][method.lower()]
                    method_info["security"] = [
                        {scheme_name: []},
                    ]
                    method_info["responses"][401] = {  # noqa: WPS432
                        "description": "Authorization failed",
                    }
                    if response_schema is not None:
                        method_info["responses"][401][  # noqa: WPS220, WPS432
                            "content"
                        ] = {
                            "application/json": {"schema": response_schema},
                        }

        # Applying generated openapi schema to our app.
        app.openapi_schema = openapi_schema
        return app.openapi_schema

    return openapi_generator

Now you can use it like this:

app.openapi = custom_openapi(app)  # type: ignore

Or if you want to show response schema in your swagger you can try this:

app.openapi = custom_openapi(  # type: ignore
    app,
    response_schema={
        "type": "object",
        "properties": {
            "detail": {
                "type": "string",
            },
        },
    },
)

This will generate nice looking response example in docs.

Eretick commented 2 years ago
@app.get('/user', dependencies=[Depends(JWTBearer())])

Based on your reply, handled it for myself by setting up HTTPBearer directly from from fastapi.security, without custom JWTBearer implementation, so it looks like

@user_router.get('/profile', dependencies=[Depends(HTTPBearer())])
def profile(Authorize: AuthJWT = Depends()) -> dict:
    Authorize.jwt_required()
    current_user = Authorize.get_jwt_subject()
    return {"user": current_user}

After I added this dependencie to the first route, the Authorize button appeared. And I don't see a problem in adding dependencies to every route you want to secure, because 1) It feels logical. I don't want make all routes protected and I want to control them intuitively. 2) Rewriting all the scheme feels more... voluminous (?) work.

halimov-oa commented 2 years ago
@app.get('/user', dependencies=[Depends(JWTBearer())])

Based on your reply, handled it for myself by setting up HTTPBearer directly from from fastapi.security, without custom JWTBearer implementation, so it looks like

@user_router.get('/profile', dependencies=[Depends(HTTPBearer())])
def profile(Authorize: AuthJWT = Depends()) -> dict:
    Authorize.jwt_required()
    current_user = Authorize.get_jwt_subject()
    return {"user": current_user}

After I added this dependencie to the first route, the Authorize button appeared. And I don't see a problem in adding dependencies to every route you want to secure, because

  1. It feels logical. I don't want make all routes protected and I want to control them intuitively.
  2. Rewriting all the scheme feels more... voluminous (?) work.

It really works very nice, at least I didn't notice something has broken down

pndiku commented 2 years ago

After I added this dependencie to the first route, the Authorize button appeared. And I don't see a problem in adding dependencies to every route you want to secure, because

  1. It feels logical. I don't want make all routes protected and I want to control them intuitively.
  2. Rewriting all the scheme feels more... voluminous (?) work.

Like everything in OSS there are many ways to skin a cat.

  1. I prefer rewriting the scheme because I do it once.

  2. Also, we're not dealing with protecting the routes. Authorize.jwt_required() already does that. The problem being solved is for the swagger documentation to have the padlock. And the scheme rewrite sorts that out.

But hey, to each his own. As long as we all reach the end destination, it's all good!

Eretick commented 2 years ago

2. Authorize.jwt_required() already does that.

Yes, you are right, my fault.

Also, for thoose who would like my option: The better variant would be putting the dependencies=[Depends(HTTPBearer())] part to APIRouter() instead of every route in it.

Eretick commented 2 years ago
  1. Also, we're not dealing with protecting the routes. Authorize.jwt_required() already does that.

Wait a second, are you sure? I just added dependencies to APIRouter and didn't used jwt_required() inside any of my routes. But without sending a token it still returns "Not authenticated".

My route is looks like the following:

image_router = APIRouter(tags=["image"], dependencies=[Depends(HTTPBearer())])

@image_router.get("/{image_id}", response_model=ImageResponse)
def get_image(image_id: int, sharing_value: str = None):
    image_db: Screenshot = db.query(Screenshot).filter(Screenshot.id == image_id).first()
    .... more logic here...

And it works for all routes in this router without the jwt_required and even passing an Authorize class to the function.

pndiku commented 2 years ago

The very first page of the documentation says that's how to protect an endpoint: https://indominusbyte.github.io/fastapi-jwt-auth/usage/basic/

HOWEVER I think this "issue" is now being dragged into different directions.

The original bug report is about adding an authorize option in the swagger file, not about different methods of protecting endpoints.

If your method works, or mine, or someone else, then it's all good. :-)

ret7020 commented 1 year ago

I resolved this issue, by tuning openapi scheme. Code snippet:

# Add JWT for fastapi_jwt_auth in swagger
def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title=settings.project_name,
        version=settings.version,
        description="Freelance API",
        routes=app.routes,
    )
    openapi_schema["components"]["securitySchemes"] = {"bearerAuth": {
        "type": "http",
        "scheme": "bearer",
        "bearerFormat": "JWT"
    }}

    openapi_schema["security"] = [{"bearerAuth": []}]

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

This code will implement following structure inside openapi.json. And after that swagger can send auth header for each endpoint. image So, I think this issue can be closed. But maybe we need to add a function, named like setup_swagger, that will execute code above?

lpdswing commented 1 year ago

The better solution

reusable_oauth2 = OAuth2PasswordBearer(tokenUrl="auth/login")
csrf_header = APIKeyHeader(name="X-CSRF-TOKEN", scheme_name="X-CSRF-TOKEN", description="Fill in the csrf access "
                                                                                        "token when authorization is "
                                                                                        "required, and use the csrf "
                                                                                        "refresh token when the token "
                                                                                        "needs to be refreshed")

async def get_auth_jwt_access(auth: AuthJWT = Depends(),
                              csrf_token: str = Security(csrf_header)):
    print(csrf_token)
    auth.jwt_required()
    return auth

async def get_auth_jwt_refresh(auth: AuthJWT = Depends(), csrf_token: str = Security(csrf_header)):
    print(csrf_token)
    auth.jwt_refresh_token_required()
    return auth
image