mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-08-13 19:42:59 +00:00
wip4
This commit is contained in:
@@ -1,13 +1,12 @@
|
||||
"""Authentication helpers for FastAPI using Security()."""
|
||||
|
||||
from .abc import AuthSource
|
||||
from .sources import BearerTokenAuth, CookieAuth, MultiAuth, OAuth2Auth, OpenIDAuth
|
||||
from .sources import APIKeyHeaderAuth, BearerTokenAuth, CookieAuth, MultiAuth
|
||||
|
||||
__all__ = [
|
||||
"APIKeyHeaderAuth",
|
||||
"AuthSource",
|
||||
"BearerTokenAuth",
|
||||
"CookieAuth",
|
||||
"MultiAuth",
|
||||
"OAuth2Auth",
|
||||
"OpenIDAuth",
|
||||
]
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
"""Built-in authentication source implementations."""
|
||||
|
||||
from .header import APIKeyHeaderAuth
|
||||
from .bearer import BearerTokenAuth
|
||||
from .cookie import CookieAuth
|
||||
from .multi import MultiAuth
|
||||
from .oauth2 import OAuth2Auth
|
||||
from .openid import OpenIDAuth
|
||||
|
||||
__all__ = ["BearerTokenAuth", "CookieAuth", "MultiAuth", "OAuth2Auth", "OpenIDAuth"]
|
||||
__all__ = ["APIKeyHeaderAuth", "BearerTokenAuth", "CookieAuth", "MultiAuth"]
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
"""API key header authentication source."""
|
||||
|
||||
import inspect
|
||||
from typing import Annotated, Any, Callable
|
||||
|
||||
from fastapi import Depends, Request
|
||||
from fastapi.security import APIKeyHeader, SecurityScopes
|
||||
|
||||
from fastapi_toolsets.exceptions import UnauthorizedError
|
||||
|
||||
from ..abc import AuthSource, _call_validator
|
||||
|
||||
|
||||
class APIKeyHeaderAuth(AuthSource):
|
||||
"""API key header authentication source.
|
||||
|
||||
Wraps :class:`fastapi.security.APIKeyHeader` for OpenAPI documentation.
|
||||
The validator is called as ``await validator(api_key, **kwargs)``
|
||||
where ``kwargs`` are the extra keyword arguments provided at instantiation.
|
||||
|
||||
Args:
|
||||
name: HTTP header name that carries the API key (e.g. ``"X-API-Key"``).
|
||||
validator: Sync or async callable that receives the API key and any
|
||||
extra keyword arguments, and returns the authenticated identity.
|
||||
Should raise :class:`~fastapi_toolsets.exceptions.UnauthorizedError`
|
||||
on failure.
|
||||
**kwargs: Extra keyword arguments forwarded to the validator on every
|
||||
call (e.g. ``role=Role.ADMIN``).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
validator: Callable[..., Any],
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self._name = name
|
||||
self._validator = validator
|
||||
self._kwargs = kwargs
|
||||
self._scheme = APIKeyHeader(name=name, auto_error=False)
|
||||
|
||||
_scheme = self._scheme
|
||||
_validator = validator
|
||||
_kwargs = kwargs
|
||||
|
||||
async def _call(
|
||||
security_scopes: SecurityScopes, # noqa: ARG001
|
||||
api_key: Annotated[str | None, Depends(_scheme)] = None,
|
||||
) -> Any:
|
||||
if api_key is None:
|
||||
raise UnauthorizedError()
|
||||
return await _call_validator(_validator, api_key, **_kwargs)
|
||||
|
||||
self._call_fn = _call
|
||||
self.__signature__ = inspect.signature(_call)
|
||||
|
||||
async def extract(self, request: Request) -> str | None:
|
||||
"""Extract the API key from the configured header."""
|
||||
return request.headers.get(self._name) or None
|
||||
|
||||
async def authenticate(self, credential: str) -> Any:
|
||||
"""Validate a credential and return the identity."""
|
||||
return await _call_validator(self._validator, credential, **self._kwargs)
|
||||
|
||||
def require(self, **kwargs: Any) -> "APIKeyHeaderAuth":
|
||||
"""Return a new instance with additional (or overriding) validator kwargs."""
|
||||
return APIKeyHeaderAuth(
|
||||
self._name,
|
||||
self._validator,
|
||||
**{**self._kwargs, **kwargs},
|
||||
)
|
||||
@@ -60,6 +60,7 @@ class MultiAuth:
|
||||
async def _call(
|
||||
request: Request,
|
||||
security_scopes: SecurityScopes, # noqa: ARG001
|
||||
**kwargs: Any, # noqa: ARG001 — absorbs scheme values injected by FastAPI
|
||||
) -> Any:
|
||||
for source in _sources:
|
||||
credential = await source.extract(request)
|
||||
@@ -68,7 +69,29 @@ class MultiAuth:
|
||||
raise UnauthorizedError()
|
||||
|
||||
self._call_fn = _call
|
||||
self.__signature__ = inspect.signature(_call)
|
||||
|
||||
# Build a merged signature that includes the security-scheme Depends()
|
||||
# parameters from every source so FastAPI registers them in OpenAPI docs.
|
||||
seen: set[str] = {"request", "security_scopes"}
|
||||
merged: list[inspect.Parameter] = [
|
||||
inspect.Parameter(
|
||||
"request",
|
||||
inspect.Parameter.POSITIONAL_OR_KEYWORD,
|
||||
annotation=Request,
|
||||
),
|
||||
inspect.Parameter(
|
||||
"security_scopes",
|
||||
inspect.Parameter.POSITIONAL_OR_KEYWORD,
|
||||
annotation=SecurityScopes,
|
||||
),
|
||||
]
|
||||
for i, source in enumerate(sources):
|
||||
for name, param in inspect.signature(source).parameters.items():
|
||||
if name in seen:
|
||||
continue
|
||||
merged.append(param.replace(name=f"_s{i}_{name}"))
|
||||
seen.add(name)
|
||||
self.__signature__ = inspect.Signature(merged, return_annotation=Any)
|
||||
|
||||
async def __call__(self, **kwargs: Any) -> Any:
|
||||
return await self._call_fn(**kwargs)
|
||||
|
||||
@@ -1,72 +0,0 @@
|
||||
"""OAuth2 password-bearer authentication source."""
|
||||
|
||||
import inspect
|
||||
from typing import Annotated, Any, Callable
|
||||
|
||||
from fastapi import Depends, Request
|
||||
from fastapi.security import OAuth2PasswordBearer, SecurityScopes
|
||||
|
||||
from fastapi_toolsets.exceptions import UnauthorizedError
|
||||
|
||||
from ..abc import AuthSource, _call_validator
|
||||
|
||||
|
||||
class OAuth2Auth(AuthSource):
|
||||
"""OAuth2 password-bearer authentication source.
|
||||
|
||||
Wraps :class:`fastapi.security.OAuth2PasswordBearer` for OpenAPI
|
||||
documentation.
|
||||
|
||||
Args:
|
||||
token_url: URL of the token endpoint (used in OpenAPI docs).
|
||||
validator: Sync or async callable that receives the token and any extra
|
||||
keyword arguments, and returns the authenticated identity.
|
||||
**kwargs: Extra keyword arguments forwarded to the validator on every
|
||||
call.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
token_url: str,
|
||||
validator: Callable[..., Any],
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self._token_url = token_url
|
||||
self._validator = validator
|
||||
self._kwargs = kwargs
|
||||
self._scheme = OAuth2PasswordBearer(tokenUrl=token_url, auto_error=False)
|
||||
|
||||
_scheme = self._scheme
|
||||
_validator = validator
|
||||
_kwargs = kwargs
|
||||
|
||||
async def _call(
|
||||
security_scopes: SecurityScopes, # noqa: ARG001
|
||||
token: Annotated[str | None, Depends(_scheme)] = None,
|
||||
) -> Any:
|
||||
if token is None:
|
||||
raise UnauthorizedError()
|
||||
return await _call_validator(_validator, token, **_kwargs)
|
||||
|
||||
self._call_fn = _call
|
||||
self.__signature__ = inspect.signature(_call)
|
||||
|
||||
async def extract(self, request: Request) -> str | None:
|
||||
"""Extract the bearer token from the Authorization header."""
|
||||
auth = request.headers.get("Authorization", "")
|
||||
if not auth.startswith("Bearer "):
|
||||
return None
|
||||
token = auth[7:]
|
||||
return token or None
|
||||
|
||||
async def authenticate(self, credential: str) -> Any:
|
||||
"""Validate a credential and return the identity."""
|
||||
return await _call_validator(self._validator, credential, **self._kwargs)
|
||||
|
||||
def require(self, **kwargs: Any) -> "OAuth2Auth":
|
||||
"""Return a new instance with additional (or overriding) validator kwargs."""
|
||||
return OAuth2Auth(
|
||||
self._token_url,
|
||||
self._validator,
|
||||
**{**self._kwargs, **kwargs},
|
||||
)
|
||||
@@ -1,84 +0,0 @@
|
||||
"""OpenID Connect authentication source."""
|
||||
|
||||
import inspect
|
||||
from typing import Annotated, Any, Callable
|
||||
|
||||
from fastapi import Depends, Request
|
||||
from fastapi.security import OpenIdConnect, SecurityScopes
|
||||
|
||||
from fastapi_toolsets.exceptions import UnauthorizedError
|
||||
|
||||
from ..abc import AuthSource, _call_validator
|
||||
|
||||
|
||||
class OpenIDAuth(AuthSource):
|
||||
"""OpenID Connect authentication source.
|
||||
|
||||
Wraps :class:`fastapi.security.OpenIdConnect` for OpenAPI documentation.
|
||||
Token extraction reads the ``Authorization: Bearer <token>`` header;
|
||||
validation is fully delegated to the user-supplied validator (use any
|
||||
OIDC / JWT library such as ``authlib``, ``python-jose``, or ``PyJWT``).
|
||||
|
||||
Args:
|
||||
openid_connect_url: URL of the OIDC discovery document
|
||||
(``/.well-known/openid-configuration``). Used only for OpenAPI
|
||||
documentation — no requests are made to this URL by this class.
|
||||
validator: Sync or async callable that receives the raw bearer token
|
||||
and any extra keyword arguments, and returns the authenticated
|
||||
identity. Should raise
|
||||
:class:`~fastapi_toolsets.exceptions.UnauthorizedError` on failure.
|
||||
**kwargs: Extra keyword arguments forwarded to the validator on every
|
||||
call (e.g. ``audience="my-app"``).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
openid_connect_url: str,
|
||||
validator: Callable[..., Any],
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self._openid_connect_url = openid_connect_url
|
||||
self._validator = validator
|
||||
self._kwargs = kwargs
|
||||
self._scheme = OpenIdConnect(
|
||||
openIdConnectUrl=openid_connect_url, auto_error=False
|
||||
)
|
||||
|
||||
_scheme = self._scheme
|
||||
_validator = validator
|
||||
_kwargs = kwargs
|
||||
|
||||
async def _call(
|
||||
security_scopes: SecurityScopes, # noqa: ARG001
|
||||
authorization: Annotated[str | None, Depends(_scheme)] = None,
|
||||
) -> Any:
|
||||
if authorization is None:
|
||||
raise UnauthorizedError()
|
||||
if not authorization.startswith("Bearer "):
|
||||
raise UnauthorizedError()
|
||||
token = authorization[7:]
|
||||
if not token:
|
||||
raise UnauthorizedError()
|
||||
return await _call_validator(_validator, token, **_kwargs)
|
||||
|
||||
self._call_fn = _call
|
||||
self.__signature__ = inspect.signature(_call)
|
||||
|
||||
async def extract(self, request: Request) -> str | None:
|
||||
"""Extract the bearer token from the Authorization header."""
|
||||
auth = request.headers.get("Authorization", "")
|
||||
if not auth.startswith("Bearer "):
|
||||
return None
|
||||
return auth[7:] or None
|
||||
|
||||
async def authenticate(self, credential: str) -> Any:
|
||||
"""Validate a credential and return the identity."""
|
||||
return await _call_validator(self._validator, credential, **self._kwargs)
|
||||
|
||||
def require(self, **kwargs: Any) -> "OpenIDAuth":
|
||||
"""Return a new instance with additional (or overriding) validator kwargs."""
|
||||
return OpenIDAuth(
|
||||
self._openid_connect_url,
|
||||
self._validator,
|
||||
**{**self._kwargs, **kwargs},
|
||||
)
|
||||
+100
-236
@@ -6,12 +6,11 @@ from fastapi.testclient import TestClient
|
||||
|
||||
from fastapi_toolsets.exceptions import UnauthorizedError, init_exceptions_handlers
|
||||
from fastapi_toolsets.security import (
|
||||
APIKeyHeaderAuth,
|
||||
AuthSource,
|
||||
BearerTokenAuth,
|
||||
CookieAuth,
|
||||
MultiAuth,
|
||||
OAuth2Auth,
|
||||
OpenIDAuth,
|
||||
)
|
||||
|
||||
|
||||
@@ -316,73 +315,85 @@ class TestCookieAuth:
|
||||
assert await auth.extract(request) == "abc"
|
||||
|
||||
|
||||
class TestOAuth2Auth:
|
||||
def test_valid_token_returns_identity(self):
|
||||
oauth = OAuth2Auth(token_url="/token", validator=simple_validator)
|
||||
class TestAPIKeyHeaderAuth:
|
||||
def test_valid_key_returns_identity(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oauth)):
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
response = client.get("/me", headers={"X-API-Key": VALID_TOKEN})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
def test_missing_token_returns_401(self):
|
||||
oauth = OAuth2Auth(token_url="/token", validator=simple_validator)
|
||||
def test_missing_header_returns_401(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oauth)):
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me")
|
||||
assert response.status_code == 401
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_no_header(self):
|
||||
from starlette.requests import Request
|
||||
def test_invalid_key_returns_401(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
|
||||
auth = OAuth2Auth("/token", simple_validator)
|
||||
scope = {"type": "http", "method": "GET", "path": "/", "headers": []}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) is None
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_empty_token(self):
|
||||
from starlette.requests import Request
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"X-API-Key": "wrong"})
|
||||
assert response.status_code == 401
|
||||
|
||||
auth = OAuth2Auth("/token", simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer ")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) is None
|
||||
def test_kwargs_forwarded_to_validator(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", role_validator, role="admin")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_token(self):
|
||||
from starlette.requests import Request
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
auth = OAuth2Auth("/token", simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer mytoken")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) == "mytoken"
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"X-API-Key": VALID_TOKEN})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_require_forwards_kwargs(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", role_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(auth.require(role="admin"))):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/admin", headers={"X-API-Key": VALID_TOKEN})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_require_preserves_name(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
derived = auth.require(role="admin")
|
||||
assert derived._name == "X-API-Key"
|
||||
|
||||
def test_require_does_not_mutate_original(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", role_validator, role="user")
|
||||
auth.require(role="admin")
|
||||
assert auth._kwargs == {"role": "user"}
|
||||
|
||||
def test_in_multi_auth(self):
|
||||
"""OAuth2Auth.authenticate() is exercised when used inside MultiAuth."""
|
||||
oauth = OAuth2Auth(token_url="/token", validator=simple_validator)
|
||||
multi = MultiAuth(oauth)
|
||||
"""APIKeyHeaderAuth.authenticate() is exercised inside MultiAuth."""
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
api_key = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
multi = MultiAuth(bearer, api_key)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
@@ -390,10 +401,52 @@ class TestOAuth2Auth:
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
# No bearer → falls through to API key header
|
||||
response = client.get("/me", headers={"X-API-Key": VALID_TOKEN})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
def test_is_auth_source(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
assert isinstance(auth, AuthSource)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_no_header(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
scope = {"type": "http", "method": "GET", "path": "/", "headers": []}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_empty_header(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"x-api-key", b"")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_key_present(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"x-api-key", b"mykey")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) == "mykey"
|
||||
|
||||
|
||||
class TestMultiAuth:
|
||||
def test_first_source_matches(self):
|
||||
@@ -626,26 +679,6 @@ class TestRequire:
|
||||
derived = cookie.require(scope="admin")
|
||||
assert derived._name == "session"
|
||||
|
||||
def test_oauth2_require_forwards_kwargs(self):
|
||||
oauth = OAuth2Auth("/token", role_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(oauth.require(role="admin"))):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get(
|
||||
"/admin", headers={"Authorization": f"Bearer {VALID_TOKEN}"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_oauth2_require_preserves_token_url(self):
|
||||
oauth = OAuth2Auth("/token", simple_validator)
|
||||
derived = oauth.require(role="admin")
|
||||
assert derived._token_url == "/token"
|
||||
|
||||
def test_bearer_require_in_multi_auth(self):
|
||||
"""require() instances work seamlessly inside MultiAuth."""
|
||||
PREFIXED_TOKEN = f"user_{VALID_TOKEN}"
|
||||
@@ -671,141 +704,6 @@ class TestRequire:
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
|
||||
class TestOpenIDAuth:
|
||||
DISCOVERY_URL = "https://accounts.example.com/.well-known/openid-configuration"
|
||||
|
||||
def test_valid_token_returns_identity(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
def test_missing_token_returns_401(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_invalid_token_returns_401(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": "Bearer wrong"})
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_kwargs_forwarded_to_validator(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, role_validator, role="admin")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_require_forwards_kwargs(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, role_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(oidc.require(role="admin"))):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get(
|
||||
"/admin", headers={"Authorization": f"Bearer {VALID_TOKEN}"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_require_preserves_discovery_url(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
derived = oidc.require(role="admin")
|
||||
assert derived._openid_connect_url == self.DISCOVERY_URL
|
||||
|
||||
def test_require_does_not_mutate_original(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, role_validator, role="user")
|
||||
oidc.require(role="admin")
|
||||
assert oidc._kwargs == {"role": "user"}
|
||||
|
||||
def test_is_auth_source(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
assert isinstance(oidc, AuthSource)
|
||||
|
||||
def test_in_multi_auth(self):
|
||||
"""OpenIDAuth works seamlessly inside MultiAuth."""
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
multi = MultiAuth(oidc)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(multi)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_no_header(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
scope = {"type": "http", "method": "GET", "path": "/", "headers": []}
|
||||
request = Request(scope)
|
||||
assert await oidc.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_empty_token(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer ")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await oidc.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_token(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer mytoken")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await oidc.extract(request) == "mytoken"
|
||||
|
||||
|
||||
class TestSyncValidators:
|
||||
"""Sync (non-async) validators — covers the sync path in _call_validator."""
|
||||
|
||||
@@ -849,36 +747,6 @@ class TestSyncValidators:
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
|
||||
class TestOpenIDAuthEdgeCases:
|
||||
DISCOVERY_URL = "https://accounts.example.com/.well-known/openid-configuration"
|
||||
|
||||
def test_non_bearer_authorization_returns_401(self):
|
||||
"""Authorization header present but not Bearer scheme."""
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": "Basic dXNlcjpwYXNz"})
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_empty_bearer_token_returns_401(self):
|
||||
"""Authorization: Bearer with no token after the scheme prefix."""
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": "Bearer "})
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestCookieAuthSigned:
|
||||
"""CookieAuth with HMAC-SHA256 signed cookies (secret_key path)."""
|
||||
|
||||
@@ -1029,14 +897,10 @@ class TestAuthSource:
|
||||
def test_builtin_classes_are_auth_sources(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
cookie = CookieAuth("session", cookie_validator)
|
||||
oauth = OAuth2Auth("/token", simple_validator)
|
||||
oidc = OpenIDAuth(
|
||||
"https://example.com/.well-known/openid-configuration", simple_validator
|
||||
)
|
||||
api_key = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
assert isinstance(bearer, AuthSource)
|
||||
assert isinstance(cookie, AuthSource)
|
||||
assert isinstance(oauth, AuthSource)
|
||||
assert isinstance(oidc, AuthSource)
|
||||
assert isinstance(api_key, AuthSource)
|
||||
|
||||
def test_custom_source_standalone_valid(self):
|
||||
"""Default __call__ wires extract + authenticate via Request injection."""
|
||||
|
||||
Reference in New Issue
Block a user