mirror
/
tiangolo.fastapi
mirror of https://github.com/tiangolo/fastapi
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

♻️ Refactor implementation of security schemes to allow subclassing to create the exception to be raised

pull/13786/head
Sebastián Ramírez 8 months ago
parent
37e47ab9ce
commit
dcd5b168df
4 changed files with 102 additions and 320 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 128
      fastapi/security/api_key.py
  2. 161
      fastapi/security/http.py
  3. 76
      fastapi/security/oauth2.py
  4. 55
      fastapi/security/open_id_connect_url.py

128
fastapi/security/api_key.py
View File

@ -5,8 +5,8 @@ from fastapi.openapi.models import APIKey, APIKeyIn
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
from typing_extensions import Annotated, Literal, deprecated
from starlette.status import HTTP_401_UNAUTHORIZED
from typing_extensions import Annotated
class APIKeyBase(SecurityBase):
@ -17,12 +17,8 @@ class APIKeyBase(SecurityBase):
description: Union[str, None],
scheme_name: Union[str, None],
auto_error: bool,
not_authenticated_status_code: Literal[401, 403],
):
self.parameter_location = location.value
self.parameter_name = name
self.auto_error = auto_error
self.not_authenticated_status_code = not_authenticated_status_code
self.model: APIKey = APIKey(
**{"in": location},
@ -31,31 +27,26 @@ class APIKeyBase(SecurityBase):
)
self.scheme_name = scheme_name or self.__class__.__name__
def format_www_authenticate_header_value(self) -> str:
"""
The WWW-Authenticate header is not standardized for API Key authentication.
It's considered good practice to include information about the authentication
challange.
This method follows one of the common templates.
If a different format is required, override this method in a subclass.
def make_not_authenticated_error(self) -> HTTPException:
"""
The WWW-Authenticate header is not standardized for API Key authentication but
the HTTP specification requires that an error of 401 "Unauthorized" must
include a WWW-Authenticate header.
return f'ApiKey in="{self.parameter_location}", name="{self.parameter_name}"'
Ref: https://datatracker.ietf.org/doc/html/rfc9110#name-401-unauthorized
def check_api_key(self, api_key: Optional[str]) -> Optional[str]:
if not api_key:
if self.auto_error:
if self.not_authenticated_status_code == HTTP_403_FORBIDDEN:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else: # By default use 401
www_authenticate = self.format_www_authenticate_header_value()
raise HTTPException(
For this, this method sends a custom challenge `APIKey`.
"""
return HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": www_authenticate},
headers={"WWW-Authenticate": "APIKey"},
)
def check_api_key(self, api_key: Optional[str]) -> Optional[str]:
if not api_key:
if self.auto_error:
raise self.make_not_authenticated_error()
return None
return api_key
@ -138,34 +129,6 @@ class APIKeyQuery(APIKeyBase):
"""
),
] = True,
not_authenticated_status_code: Annotated[
Literal[401, 403],
Doc(
"""
By default, if the query parameter is not provided and `auto_error` is
set to `True`, `APIKeyQuery` will automatically raise an
`HTTPException` with the status code `401`.
If your client relies on the old (incorrect) behavior and expects the
status code to be `403`, you can set `not_authenticated_status_code` to
`403` to achieve it.
Keep in mind that this parameter is temporary and will be removed in
the near future.
Consider updating your clients to align with the new behavior.
"""
),
deprecated(
"""
This parameter is temporary. It was introduced to give users time
to upgrade their clients to follow the new behavior and will eventually
be removed.
Use it as a short-term workaround, but consider updating your clients
to align with the new behavior.
"""
),
] = 401,
):
super().__init__(
location=APIKeyIn.query,
@ -173,7 +136,6 @@ class APIKeyQuery(APIKeyBase):
scheme_name=scheme_name,
description=description,
auto_error=auto_error,
not_authenticated_status_code=not_authenticated_status_code,
)
async def __call__(self, request: Request) -> Optional[str]:
@ -255,34 +217,6 @@ class APIKeyHeader(APIKeyBase):
"""
),
] = True,
not_authenticated_status_code: Annotated[
Literal[401, 403],
Doc(
"""
By default, if the header is not provided and `auto_error` is
set to `True`, `APIKeyHeader` will automatically raise an
`HTTPException` with the status code `401`.
If your client relies on the old (incorrect) behavior and expects the
status code to be `403`, you can set `not_authenticated_status_code` to
`403` to achieve it.
Keep in mind that this parameter is temporary and will be removed in
the near future.
Consider updating your clients to align with the new behavior.
"""
),
deprecated(
"""
This parameter is temporary. It was introduced to give users time
to upgrade their clients to follow the new behavior and will eventually
be removed.
Use it as a short-term workaround, but consider updating your clients
to align with the new behavior.
"""
),
] = 401,
):
super().__init__(
location=APIKeyIn.header,
@ -290,7 +224,6 @@ class APIKeyHeader(APIKeyBase):
scheme_name=scheme_name,
description=description,
auto_error=auto_error,
not_authenticated_status_code=not_authenticated_status_code,
)
async def __call__(self, request: Request) -> Optional[str]:
@ -372,34 +305,6 @@ class APIKeyCookie(APIKeyBase):
"""
),
] = True,
not_authenticated_status_code: Annotated[
Literal[401, 403],
Doc(
"""
By default, if the cookie is not provided and `auto_error` is
set to `True`, `APIKeyCookie` will automatically raise an
`HTTPException` with the status code `401`.
If your client relies on the old (incorrect) behavior and expects the
status code to be `403`, you can set `not_authenticated_status_code` to
`403` to achieve it.
Keep in mind that this parameter is temporary and will be removed in
the near future.
Consider updating your clients to align with the new behavior.
"""
),
deprecated(
"""
This parameter is temporary. It was introduced to give users time
to upgrade their clients to follow the new behavior and will eventually
be removed.
Use it as a short-term workaround, but consider updating your clients
to align with the new behavior.
"""
),
] = 401,
):
super().__init__(
location=APIKeyIn.cookie,
@ -407,7 +312,6 @@ class APIKeyCookie(APIKeyBase):
scheme_name=scheme_name,
description=description,
auto_error=auto_error,
not_authenticated_status_code=not_authenticated_status_code,
)
async def __call__(self, request: Request) -> Optional[str]:

161
fastapi/security/http.py
View File

@ -10,8 +10,8 @@ from fastapi.security.base import SecurityBase
from fastapi.security.utils import get_authorization_scheme_param
from pydantic import BaseModel
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
from typing_extensions import Annotated, Literal, deprecated
from starlette.status import HTTP_401_UNAUTHORIZED
from typing_extensions import Annotated
class HTTPBasicCredentials(BaseModel):
@ -75,13 +75,29 @@ class HTTPBase(SecurityBase):
scheme_name: Optional[str] = None,
description: Optional[str] = None,
auto_error: bool = True,
not_authenticated_status_code: Literal[401, 403] = 401,
):
self.model = HTTPBaseModel(scheme=scheme, description=description)
self.model_scheme = scheme
self.model: HTTPBaseModel = HTTPBaseModel(
scheme=scheme, description=description
)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
self.not_authenticated_status_code = not_authenticated_status_code
def make_authenticate_headers(self) -> dict[str, str]:
return {"WWW-Authenticate": f"{self.model.scheme.title()}"}
def make_not_authenticated_error(self) -> HTTPException:
return HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers=self.make_authenticate_headers(),
)
def make_invalid_user_credentials_error(self) -> HTTPException:
return HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers=self.make_authenticate_headers(),
)
async def __call__(
self, request: Request
@ -90,16 +106,7 @@ class HTTPBase(SecurityBase):
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
if self.auto_error:
if self.not_authenticated_status_code == HTTP_403_FORBIDDEN:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": self.model_scheme},
)
raise self.make_not_authenticated_error()
else:
return None
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
@ -109,6 +116,8 @@ class HTTPBasic(HTTPBase):
"""
HTTP Basic authentication.
Ref: https://datatracker.ietf.org/doc/html/rfc7617
## Usage
Create an instance object and use that object as the dependency in `Depends()`.
@ -195,36 +204,28 @@ class HTTPBasic(HTTPBase):
self.realm = realm
self.auto_error = auto_error
def make_authenticate_headers(self) -> dict[str, str]:
if self.realm:
return {"WWW-Authenticate": f'Basic realm="{self.realm}"'}
return {"WWW-Authenticate": "Basic"}
async def __call__( # type: ignore
self, request: Request
) -> Optional[HTTPBasicCredentials]:
authorization = request.headers.get("Authorization")
scheme, param = get_authorization_scheme_param(authorization)
if self.realm:
unauthorized_headers = {"WWW-Authenticate": f'Basic realm="{self.realm}"'}
else:
unauthorized_headers = {"WWW-Authenticate": "Basic"}
if not authorization or scheme.lower() != "basic":
if self.auto_error:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers=unauthorized_headers,
)
raise self.make_not_authenticated_error()
else:
return None
invalid_user_credentials_exc = HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers=unauthorized_headers,
)
try:
data = b64decode(param).decode("ascii")
except (ValueError, UnicodeDecodeError, binascii.Error):
raise invalid_user_credentials_exc # noqa: B904
except (ValueError, UnicodeDecodeError, binascii.Error) as e:
raise self.make_invalid_user_credentials_error() from e
username, separator, password = data.partition(":")
if not separator:
raise invalid_user_credentials_exc
raise self.make_invalid_user_credentials_error()
return HTTPBasicCredentials(username=username, password=password)
@ -304,38 +305,10 @@ class HTTPBearer(HTTPBase):
"""
),
] = True,
not_authenticated_status_code: Annotated[
Literal[401, 403],
Doc(
"""
By default, if the HTTP Bearer token is not provided and `auto_error`
is set to `True`, `HTTPBearer` will automatically raise an
`HTTPException` with the status code `401`.
If your client relies on the old (incorrect) behavior and expects the
status code to be `403`, you can set `not_authenticated_status_code` to
`403` to achieve it.
Keep in mind that this parameter is temporary and will be removed in
the near future.
"""
),
deprecated(
"""
This parameter is temporary. It was introduced to give users time
to upgrade their clients to follow the new behavior and will eventually
be removed.
Use it as a short-term workaround, but consider updating your clients
to align with the new behavior.
"""
),
] = 401,
):
self.model = HTTPBearerModel(bearerFormat=bearerFormat, description=description)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
self.not_authenticated_status_code = not_authenticated_status_code
async def __call__(
self, request: Request
@ -344,33 +317,27 @@ class HTTPBearer(HTTPBase):
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
if self.auto_error:
self._raise_not_authenticated_error(error_message="Not authenticated")
raise self.make_not_authenticated_error()
else:
return None
if scheme.lower() != "bearer":
if self.auto_error:
self._raise_not_authenticated_error(
error_message="Invalid authentication credentials"
)
raise self.make_invalid_user_credentials_error()
else:
return None
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
def _raise_not_authenticated_error(self, error_message: str) -> None:
if self.not_authenticated_status_code == HTTP_403_FORBIDDEN:
raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail=error_message)
else:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail=error_message,
headers={"WWW-Authenticate": "Bearer"},
)
class HTTPDigest(HTTPBase):
"""
HTTP Digest authentication.
**Warning**: this is only a stub to connect the components with OpenAPI in FastAPI,
but it doesn't implement the full Digest scheme, you would need to to subclass it
and implement it in your code.
Ref: https://datatracker.ietf.org/doc/html/rfc7616
## Usage
Create an instance object and use that object as the dependency in `Depends()`.
@ -441,38 +408,10 @@ class HTTPDigest(HTTPBase):
"""
),
] = True,
not_authenticated_status_code: Annotated[
Literal[401, 403],
Doc(
"""
By default, if the HTTP Digest is not provided and `auto_error`
is set to `True`, `HTTPDigest` will automatically raise an
`HTTPException` with the status code `401`.
If your client relies on the old (incorrect) behavior and expects the
status code to be `403`, you can set `not_authenticated_status_code` to
`403` to achieve it.
Keep in mind that this parameter is temporary and will be removed in
the near future.
"""
),
deprecated(
"""
This parameter is temporary. It was introduced to give users time
to upgrade their clients to follow the new behavior and will eventually
be removed.
Use it as a short-term workaround, but consider updating your clients
to align with the new behavior.
"""
),
] = 401,
):
self.model = HTTPBaseModel(scheme="digest", description=description)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
self.not_authenticated_status_code = not_authenticated_status_code
async def __call__(
self, request: Request
@ -481,24 +420,12 @@ class HTTPDigest(HTTPBase):
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
if self.auto_error:
self._raise_not_authenticated_error(error_message="Not authenticated")
raise self.make_not_authenticated_error()
else:
return None
if scheme.lower() != "digest":
if self.auto_error:
self._raise_not_authenticated_error(
error_message="Invalid authentication credentials",
)
raise self.make_invalid_user_credentials_error()
else:
return None
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
def _raise_not_authenticated_error(self, error_message: str) -> None:
if self.not_authenticated_status_code == HTTP_403_FORBIDDEN:
raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail=error_message)
else:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail=error_message,
headers={"WWW-Authenticate": "Digest"},
)

76
fastapi/security/oauth2.py
View File

@ -8,10 +8,10 @@ from fastapi.param_functions import Form
from fastapi.security.base import SecurityBase
from fastapi.security.utils import get_authorization_scheme_param
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
from starlette.status import HTTP_401_UNAUTHORIZED
# TODO: import from typing when deprecating Python 3.9
from typing_extensions import Annotated, Literal, deprecated
from typing_extensions import Annotated
class OAuth2PasswordRequestForm:
@ -370,56 +370,40 @@ class OAuth2(SecurityBase):
"""
),
] = True,
not_authenticated_status_code: Annotated[
Literal[401, 403],
Doc(
"""
By default, if no HTTP Authorization header provided and `auto_error`
is set to `True`, it will automatically raise an`HTTPException` with
the status code `401`.
If your client relies on the old (incorrect) behavior and expects the
status code to be `403`, you can set `not_authenticated_status_code` to
`403` to achieve it.
Keep in mind that this parameter is temporary and will be removed in
the near future.
"""
),
deprecated(
"""
This parameter is temporary. It was introduced to give users time
to upgrade their clients to follow the new behavior and will eventually
be removed.
Use it as a short-term workaround, but consider updating your clients
to align with the new behavior.
"""
),
] = 401,
):
self.model = OAuth2Model(
flows=cast(OAuthFlowsModel, flows), description=description
)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
self.not_authenticated_status_code = not_authenticated_status_code
async def __call__(self, request: Request) -> Optional[str]:
authorization = request.headers.get("Authorization")
if not authorization:
if self.auto_error:
if self.not_authenticated_status_code == HTTP_403_FORBIDDEN:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
raise HTTPException(
def make_not_authenticated_error(self) -> HTTPException:
"""
The OAuth 2 specification doesn't define the challenge that should be used,
because a `Bearer` token is not really the only option to authenticate.
But declaring any other authentication challenge would be application-specific
as it's not defined in the specification.
For practical reasons, this method uses the `Bearer` challenge by default, as
it's probably the most common one.
If you are implementing an OAuth2 authentication scheme other than the provided
ones in FastAPI (based on bearer tokens), you might want to override this.
Ref: https://datatracker.ietf.org/doc/html/rfc6749
"""
return HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
async def __call__(self, request: Request) -> Optional[str]:
authorization = request.headers.get("Authorization")
if not authorization:
if self.auto_error:
raise self.make_not_authenticated_error()
else:
return None
return authorization
@ -527,11 +511,7 @@ class OAuth2PasswordBearer(OAuth2):
scheme, param = get_authorization_scheme_param(authorization)
if not authorization or scheme.lower() != "bearer":
if self.auto_error:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
raise self.make_not_authenticated_error()
else:
return None
return param
@ -637,11 +617,7 @@ class OAuth2AuthorizationCodeBearer(OAuth2):
scheme, param = get_authorization_scheme_param(authorization)
if not authorization or scheme.lower() != "bearer":
if self.auto_error:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
raise self.make_not_authenticated_error()
else:
return None # pragma: nocover
return param

55
fastapi/security/open_id_connect_url.py
View File

@ -5,14 +5,19 @@ 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, HTTP_403_FORBIDDEN
from typing_extensions import Annotated, Literal, deprecated
from starlette.status import HTTP_401_UNAUTHORIZED
from typing_extensions import Annotated
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 to subclass it and implement it in your
code.
"""
def __init__(
@ -66,55 +71,25 @@ class OpenIdConnect(SecurityBase):
"""
),
] = True,
not_authenticated_status_code: Annotated[
Literal[401, 403],
Doc(
"""
By default, if no HTTP Authorization header provided and `auto_error`
is set to `True`, it will automatically raise an`HTTPException` with
the status code `401`.
If your client relies on the old (incorrect) behavior and expects the
status code to be `403`, you can set `not_authenticated_status_code` to
`403` to achieve it.
Keep in mind that this parameter is temporary and will be removed in
the near future.
"""
),
deprecated(
"""
This parameter is temporary. It was introduced to give users time
to upgrade their clients to follow the new behavior and will eventually
be removed.
Use it as a short-term workaround, but consider updating your clients
to align with the new behavior.
"""
),
] = 401,
):
self.model = OpenIdConnectModel(
openIdConnectUrl=openIdConnectUrl, description=description
)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
self.not_authenticated_status_code = not_authenticated_status_code
async def __call__(self, request: Request) -> Optional[str]:
authorization = request.headers.get("Authorization")
if not authorization:
if self.auto_error:
if self.not_authenticated_status_code == HTTP_403_FORBIDDEN:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
raise HTTPException(
def make_not_authenticated_error(self) -> HTTPException:
return HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
async def __call__(self, request: Request) -> Optional[str]:
authorization = request.headers.get("Authorization")
if not authorization:
if self.auto_error:
raise self.make_not_authenticated_error()
else:
return None
return authorization

Write Preview
Loading…
Cancel
Save