You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

97 lines
3.7 KiB

from typing import Annotated
from annotated_doc import Doc
from fastapi.openapi.models import OpenIdConnect as OpenIdConnectModel
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED
DOCS = {
"url": Doc("""
The OpenID Connect URL.
"""),
"scheme": Doc("""
Security scheme name.
It will be included in the generated OpenAPI (e.g. visible at `/docs`).
"""),
"desc": Doc("""
Security scheme description.
It will be included in the generated OpenAPI (e.g. visible at `/docs`).
"""),
"auto": Doc("""
By default, if no HTTP Authorization header is provided, required for
OpenID Connect authentication, it will automatically cancel the request
and send the client an error.
If `auto_error` is set to `False`, when the HTTP Authorization header
is not available, instead of erroring out, the dependency result will
be `None`.
This is useful when you want to have optional authentication.
It is also useful when you want to have authentication that can be
provided in one of multiple optional ways (for example, with OpenID
Connect or in a cookie).
""")
}
class OpenIdConnect(SecurityBase):
"""
OpenID Connect authentication class. An instance of it would be used as a
dependency.
**Warning**: this is only a stub to connect the components with OpenAPI in FastAPI,
but it doesn't implement the full OpenIdConnect scheme, for example, it doesn't use
the OpenIDConnect URL. You would need to subclass it and implement it in your
code.
"""
def __init__(
self,
*,
openIdConnectUrl: Annotated[str, DOCS["url"]],
scheme_name: Annotated[str | None, DOCS["scheme"]] = None,
description: Annotated[str | None, DOCS["desc"]] = None,
auto_error: Annotated[bool, DOCS["auto"]] = True,
):
self.model = OpenIdConnectModel(openIdConnectUrl=openIdConnectUrl, description=description)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
def make_not_authenticated_error(self, detail: str = "Not authenticated") -> HTTPException:
return HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail=detail,
headers={"WWW-Authenticate": "Bearer"},
)
async def __call__(self, request: Request) -> str | None:
authorization = request.headers.get("Authorization")
# Case 1: Header is entirely missing
if not authorization:
if self.auto_error:
raise self.make_not_authenticated_error("Missing 'Authorization' header in request.")
return None
# Case 2: Header exists but uses the wrong protocol/scheme (e.g., Basic, APIKey, or raw token)
if not authorization.lower().startswith("bearer "):
if self.auto_error:
raise self.make_not_authenticated_error(
"Invalid authentication credentials. Expected a 'Bearer <token>' prefix."
)
return None
# Case 3: 'Bearer ' prefix is there, but no actual token follows it
token = authorization[7:].strip()
if not token:
if self.auto_error:
raise self.make_not_authenticated_error(
"Authentication token value is empty or missing after 'Bearer' prefix."
)
return None
return authorization