Browse Source

feat(security, encoders): add cookie/SSO auth helpers and fix pydantic v2 encoder path

- Add HTTPCookieBearer for cookie-based JWT auth
- Add OAuth2AuthorizationCodeState for CSRF-safe third-party callbacks
- Fix OpenIdConnect.__call__ to return token instead of raw header value
- Short-circuit jsonable_encoder BaseModel branch via model_dump(mode="json")
pull/15318/head
ilike314272 3 months ago
parent
commit
a992ac3bcf
  1. 12
      fastapi/encoders.py
  2. 2
      fastapi/security/__init__.py
  3. 47
      fastapi/security/http.py
  4. 34
      fastapi/security/oauth2.py
  5. 9
      fastapi/security/open_id_connect_url.py

12
fastapi/encoders.py

@ -224,7 +224,7 @@ def jsonable_encoder(
if exclude is not None and not isinstance(exclude, (set, dict)): if exclude is not None and not isinstance(exclude, (set, dict)):
exclude = set(exclude) # type: ignore[assignment] # ty: ignore[unused-ignore-comment] exclude = set(exclude) # type: ignore[assignment] # ty: ignore[unused-ignore-comment]
if isinstance(obj, BaseModel): if isinstance(obj, BaseModel):
obj_dict = obj.model_dump( result = obj.model_dump(
mode="json", mode="json",
include=include, include=include,
exclude=exclude, exclude=exclude,
@ -233,12 +233,10 @@ def jsonable_encoder(
exclude_none=exclude_none, exclude_none=exclude_none,
exclude_defaults=exclude_defaults, exclude_defaults=exclude_defaults,
) )
return jsonable_encoder(
obj_dict, if sqlalchemy_safe:
exclude_none=exclude_none, result = {k: v for k, v in result.items() if not (isinstance(k, str) and k.startswith("_sa"))}
exclude_defaults=exclude_defaults, return result
sqlalchemy_safe=sqlalchemy_safe,
)
if dataclasses.is_dataclass(obj): if dataclasses.is_dataclass(obj):
assert not isinstance(obj, type) assert not isinstance(obj, type)
obj_dict = dataclasses.asdict(obj) obj_dict = dataclasses.asdict(obj)

2
fastapi/security/__init__.py

@ -5,9 +5,11 @@ from .http import HTTPAuthorizationCredentials as HTTPAuthorizationCredentials
from .http import HTTPBasic as HTTPBasic from .http import HTTPBasic as HTTPBasic
from .http import HTTPBasicCredentials as HTTPBasicCredentials from .http import HTTPBasicCredentials as HTTPBasicCredentials
from .http import HTTPBearer as HTTPBearer from .http import HTTPBearer as HTTPBearer
from .http import HTTPCookieBearer as HTTPCookieBearer
from .http import HTTPDigest as HTTPDigest from .http import HTTPDigest as HTTPDigest
from .oauth2 import OAuth2 as OAuth2 from .oauth2 import OAuth2 as OAuth2
from .oauth2 import OAuth2AuthorizationCodeBearer as OAuth2AuthorizationCodeBearer from .oauth2 import OAuth2AuthorizationCodeBearer as OAuth2AuthorizationCodeBearer
from .oauth2 import OAuth2AuthorizationCodeState as OAuth2AuthorizationCodeState
from .oauth2 import OAuth2PasswordBearer as OAuth2PasswordBearer from .oauth2 import OAuth2PasswordBearer as OAuth2PasswordBearer
from .oauth2 import OAuth2PasswordRequestForm as OAuth2PasswordRequestForm from .oauth2 import OAuth2PasswordRequestForm as OAuth2PasswordRequestForm
from .oauth2 import OAuth2PasswordRequestFormStrict as OAuth2PasswordRequestFormStrict from .oauth2 import OAuth2PasswordRequestFormStrict as OAuth2PasswordRequestFormStrict

47
fastapi/security/http.py

@ -315,6 +315,53 @@ class HTTPBearer(HTTPBase):
return None return None
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials) return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
class HTTPCookieBearer(SecurityBase):
"""
Bearer token authentication via a named cookie (e.g. `access_token`).
Reads the cookie value directly; your route handler is responsible for
JWT validation. Set `HttpOnly=True, Secure=True, SameSite="lax"` when
issuing the cookie this class does not enforce those attributes.
## Example
from fastapi import Depends, FastAPI
from fastapi.security import HTTPCookieBearer, HTTPAuthorizationCredentials
app = FastAPI()
security = HTTPCookieBearer(cookie_name="access_token")
@app.get("/me")
def read_me(creds: HTTPAuthorizationCredentials = Depends(security)):
return {"token": creds.credentials}
"""
def __init__(
self,
*,
cookie_name: str = "access_token",
scheme_name: str | None = None,
auto_error: bool = True,
):
self.cookie_name = cookie_name
self.model = HTTPBearerModel(description="Bearer token in a cookie")
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
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) -> HTTPAuthorizationCredentials | None:
token = request.cookies.get(self.cookie_name)
if not token:
if self.auto_error:
raise self.make_not_authenticated_error()
return None
return HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
class HTTPDigest(HTTPBase): class HTTPDigest(HTTPBase):
""" """

34
fastapi/security/oauth2.py

@ -9,6 +9,8 @@ from fastapi.security.base import SecurityBase
from fastapi.security.utils import get_authorization_scheme_param from fastapi.security.utils import get_authorization_scheme_param
from starlette.requests import Request from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED from starlette.status import HTTP_401_UNAUTHORIZED
from starlette.responses import RedirectResponse
import secrets
class OAuth2PasswordRequestForm: class OAuth2PasswordRequestForm:
@ -650,6 +652,38 @@ class OAuth2AuthorizationCodeBearer(OAuth2):
return param return param
class OAuth2AuthorizationCodeState:
"""
Thin wrapper around OAuth2AuthorizationCodeBearer that validates the
`state` parameter on the callback to prevent CSRF.
Store `pending_states` in a short-lived cache (Redis, TTLCache, etc.)
in production. This in-memory set is for single-process use only.
"""
def __init__(self, authorization_url: str, token_url: str, redirect_uri: str):
self.authorization_url = authorization_url
self.token_url = token_url
self.redirect_uri = redirect_uri
self._pending: set[str] = set()
def authorization_redirect(self, scope: str = "") -> RedirectResponse:
state = secrets.token_urlsafe(32)
self._pending.add(state)
params = (
f"response_type=code&client_id=__configure__"
f"&redirect_uri={self.redirect_uri}&scope={scope}&state={state}"
)
return RedirectResponse(f"{self.authorization_url}?{params}")
def validate_callback(self, code: str, state: str) -> str:
"""Returns `code` if state is valid; raises HTTPException otherwise."""
if state not in self._pending:
raise HTTPException(status_code=400, detail="Invalid OAuth2 state")
self._pending.discard(state)
return code
class SecurityScopes: class SecurityScopes:
""" """
This is a special class that you can define in a parameter in a dependency to This is a special class that you can define in a parameter in a dependency to

9
fastapi/security/open_id_connect_url.py

@ -85,10 +85,11 @@ class OpenIdConnect(SecurityBase):
) )
async def __call__(self, request: Request) -> str | None: async def __call__(self, request: Request) -> str | None:
from fastapi.security.utils import get_authorization_scheme_param
authorization = request.headers.get("Authorization") authorization = request.headers.get("Authorization")
if not authorization: scheme, param = get_authorization_scheme_param(authorization)
if not authorization or scheme.lower() != "bearer":
if self.auto_error: if self.auto_error:
raise self.make_not_authenticated_error() raise self.make_not_authenticated_error()
else: return None
return None return param # <-- returns just the token
return authorization

Loading…
Cancel
Save