mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-08-13 19:42:59 +00:00
feat: add security module
This commit is contained in:
@@ -0,0 +1,902 @@
|
||||
"""Tests for fastapi_toolsets.security."""
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI, Security
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from fastapi_toolsets.exceptions import UnauthorizedError, init_exceptions_handlers
|
||||
from fastapi_toolsets.security import (
|
||||
AuthSource,
|
||||
BearerTokenAuth,
|
||||
CookieAuth,
|
||||
MultiAuth,
|
||||
OAuth2Auth,
|
||||
OpenIDAuth,
|
||||
)
|
||||
|
||||
|
||||
def _app(*routes_setup_fns):
|
||||
"""Build a minimal FastAPI test app with exception handlers."""
|
||||
app = FastAPI()
|
||||
init_exceptions_handlers(app)
|
||||
for fn in routes_setup_fns:
|
||||
fn(app)
|
||||
return app
|
||||
|
||||
|
||||
VALID_TOKEN = "secret"
|
||||
VALID_COOKIE = "session123"
|
||||
|
||||
|
||||
async def simple_validator(credential: str) -> dict:
|
||||
if credential != VALID_TOKEN:
|
||||
raise UnauthorizedError()
|
||||
return {"user": "alice"}
|
||||
|
||||
|
||||
async def role_validator(credential: str, *, role: str) -> dict:
|
||||
if credential != VALID_TOKEN:
|
||||
raise UnauthorizedError()
|
||||
return {"user": "alice", "role": role}
|
||||
|
||||
|
||||
async def cookie_validator(value: str) -> dict:
|
||||
if value != VALID_COOKIE:
|
||||
raise UnauthorizedError()
|
||||
return {"session": value}
|
||||
|
||||
|
||||
class TestBearerTokenAuth:
|
||||
def test_valid_token_returns_identity(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(bearer)):
|
||||
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_header_returns_401(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(bearer)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_invalid_token_returns_401(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(bearer)):
|
||||
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):
|
||||
bearer = BearerTokenAuth(role_validator, role="admin")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(bearer)):
|
||||
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_prefix_matching_passes_full_token(self):
|
||||
"""Token with matching prefix: full token (with prefix) is passed to validator."""
|
||||
received: list[str] = []
|
||||
|
||||
async def capturing_validator(credential: str) -> dict:
|
||||
received.append(credential)
|
||||
return {"user": "alice"}
|
||||
|
||||
bearer = BearerTokenAuth(capturing_validator, prefix="user_")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(bearer)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": "Bearer user_abc123"})
|
||||
assert response.status_code == 200
|
||||
# Prefix is kept — validator receives the full token as stored in DB
|
||||
assert received == ["user_abc123"]
|
||||
|
||||
def test_prefix_mismatch_returns_401(self):
|
||||
bearer = BearerTokenAuth(simple_validator, prefix="user_")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(bearer)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": "Bearer org_abc123"})
|
||||
assert response.status_code == 401
|
||||
|
||||
# --- extract() ---
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_no_header(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
scope = {"type": "http", "method": "GET", "path": "/", "headers": []}
|
||||
request = Request(scope)
|
||||
assert await bearer.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_empty_token(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer ")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await bearer.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_no_prefix(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer mytoken")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await bearer.extract(request) == "mytoken"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_prefix_match(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
bearer = BearerTokenAuth(simple_validator, prefix="user_")
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer user_abc")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await bearer.extract(request) == "user_abc"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_prefix_no_match(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
bearer = BearerTokenAuth(simple_validator, prefix="user_")
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer org_abc")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await bearer.extract(request) is None
|
||||
|
||||
# --- generate_token() ---
|
||||
|
||||
def test_generate_token_no_prefix(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
token = bearer.generate_token()
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 0
|
||||
|
||||
def test_generate_token_with_prefix(self):
|
||||
bearer = BearerTokenAuth(simple_validator, prefix="user_")
|
||||
token = bearer.generate_token()
|
||||
assert token.startswith("user_")
|
||||
|
||||
def test_generate_token_uniqueness(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
assert bearer.generate_token() != bearer.generate_token()
|
||||
|
||||
def test_generate_token_is_valid_credential(self):
|
||||
"""A generated token (with prefix) is accepted by the same auth source."""
|
||||
stored: list[str] = []
|
||||
|
||||
async def storing_validator(credential: str) -> dict:
|
||||
stored.append(credential)
|
||||
return {"token": credential}
|
||||
|
||||
bearer = BearerTokenAuth(storing_validator, prefix="user_")
|
||||
token = bearer.generate_token()
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(bearer)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {token}"})
|
||||
assert response.status_code == 200
|
||||
assert stored == [token]
|
||||
|
||||
|
||||
class TestCookieAuth:
|
||||
def test_valid_cookie_returns_identity(self):
|
||||
cookie_auth = CookieAuth("session", cookie_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(cookie_auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", cookies={"session": VALID_COOKIE})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"session": VALID_COOKIE}
|
||||
|
||||
def test_missing_cookie_returns_401(self):
|
||||
cookie_auth = CookieAuth("session", cookie_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(cookie_auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_invalid_cookie_returns_401(self):
|
||||
cookie_auth = CookieAuth("session", cookie_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(cookie_auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", cookies={"session": "wrong"})
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_kwargs_forwarded_to_validator(self):
|
||||
async def session_validator(value: str, *, scope: str) -> dict:
|
||||
if value != VALID_COOKIE:
|
||||
raise UnauthorizedError()
|
||||
return {"session": value, "scope": scope}
|
||||
|
||||
cookie_auth = CookieAuth("session", session_validator, scope="read")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(cookie_auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", cookies={"session": VALID_COOKIE})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"session": VALID_COOKIE, "scope": "read"}
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_no_cookie(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
auth = CookieAuth("session", cookie_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_cookie_present(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
auth = CookieAuth("session", cookie_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"cookie", b"session=abc")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) == "abc"
|
||||
|
||||
|
||||
class TestOAuth2Auth:
|
||||
def test_valid_token_returns_identity(self):
|
||||
oauth = OAuth2Auth(token_url="/token", validator=simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oauth)):
|
||||
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):
|
||||
oauth = OAuth2Auth(token_url="/token", validator=simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oauth)):
|
||||
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
|
||||
|
||||
auth = OAuth2Auth("/token", 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_token(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
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
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_token(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
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"
|
||||
|
||||
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)
|
||||
|
||||
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"}
|
||||
|
||||
|
||||
class TestMultiAuth:
|
||||
def test_first_source_matches(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
cookie = CookieAuth("session", cookie_validator)
|
||||
multi = MultiAuth(bearer, cookie)
|
||||
|
||||
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"}
|
||||
|
||||
def test_second_source_matches_when_first_absent(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
cookie = CookieAuth("session", cookie_validator)
|
||||
multi = MultiAuth(bearer, cookie)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(multi)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
# No Authorization header — falls through to cookie
|
||||
response = client.get("/me", cookies={"session": VALID_COOKIE})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"session": VALID_COOKIE}
|
||||
|
||||
def test_no_source_matches_returns_401(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
cookie = CookieAuth("session", cookie_validator)
|
||||
multi = MultiAuth(bearer, cookie)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(multi)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_invalid_credential_does_not_fallthrough(self):
|
||||
"""If a credential is found but invalid, the next source is NOT tried."""
|
||||
second_called: list[bool] = []
|
||||
|
||||
async def tracking_validator(credential: str) -> dict:
|
||||
second_called.append(True)
|
||||
return {"from": "second"}
|
||||
|
||||
bearer = BearerTokenAuth(simple_validator) # raises on wrong token
|
||||
cookie = CookieAuth("session", tracking_validator)
|
||||
multi = MultiAuth(bearer, cookie)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(multi)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
# Bearer credential present but wrong — should NOT try cookie
|
||||
response = client.get(
|
||||
"/me",
|
||||
headers={"Authorization": "Bearer wrong"},
|
||||
cookies={"session": VALID_COOKIE},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
assert second_called == [] # cookie validator was never called
|
||||
|
||||
def test_prefix_routes_to_correct_source(self):
|
||||
"""Prefix-based dispatch: only the matching source's validator is called."""
|
||||
user_calls: list[str] = []
|
||||
org_calls: list[str] = []
|
||||
|
||||
async def user_validator(credential: str) -> dict:
|
||||
user_calls.append(credential)
|
||||
return {"type": "user", "id": credential}
|
||||
|
||||
async def org_validator(credential: str) -> dict:
|
||||
org_calls.append(credential)
|
||||
return {"type": "org", "id": credential}
|
||||
|
||||
user_bearer = BearerTokenAuth(user_validator, prefix="user_")
|
||||
org_bearer = BearerTokenAuth(org_validator, prefix="org_")
|
||||
multi = MultiAuth(user_bearer, org_bearer)
|
||||
|
||||
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": "Bearer user_alice"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"type": "user", "id": "user_alice"}
|
||||
assert user_calls == ["user_alice"]
|
||||
assert org_calls == []
|
||||
|
||||
user_calls.clear()
|
||||
|
||||
response = client.get("/me", headers={"Authorization": "Bearer org_acme"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"type": "org", "id": "org_acme"}
|
||||
assert user_calls == []
|
||||
assert org_calls == ["org_acme"]
|
||||
|
||||
def test_require_returns_new_multi_auth(self):
|
||||
from fastapi_toolsets.security.multi import MultiAuth as MultiAuthClass
|
||||
|
||||
bearer = BearerTokenAuth(role_validator)
|
||||
multi = MultiAuth(bearer)
|
||||
derived = multi.require(role="admin")
|
||||
assert isinstance(derived, MultiAuthClass)
|
||||
assert derived is not multi
|
||||
|
||||
def test_require_forwards_kwargs_to_sources(self):
|
||||
"""multi.require() propagates to all sources that support it."""
|
||||
bearer = BearerTokenAuth(role_validator)
|
||||
multi = MultiAuth(bearer)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(multi.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_skips_sources_without_require(self):
|
||||
"""Sources without require() are passed through unchanged."""
|
||||
header_auth = _HeaderAuth(secret="s3cr3t")
|
||||
multi = MultiAuth(header_auth)
|
||||
derived = multi.require(role="admin")
|
||||
assert derived._sources[0] is header_auth
|
||||
|
||||
def test_require_does_not_mutate_original(self):
|
||||
bearer = BearerTokenAuth(role_validator, role="user")
|
||||
multi = MultiAuth(bearer)
|
||||
multi.require(role="admin")
|
||||
assert bearer._kwargs == {"role": "user"}
|
||||
|
||||
def test_require_mixed_sources(self):
|
||||
"""require() applies to sources with require(), skips those without."""
|
||||
from typing import cast
|
||||
|
||||
bearer = BearerTokenAuth(role_validator)
|
||||
header_auth = _HeaderAuth(secret="s3cr3t")
|
||||
multi = MultiAuth(bearer, header_auth)
|
||||
derived = multi.require(role="admin")
|
||||
# bearer got require() applied, header_auth passed through
|
||||
assert cast(BearerTokenAuth, derived._sources[0])._kwargs == {"role": "admin"}
|
||||
assert derived._sources[1] is header_auth
|
||||
|
||||
|
||||
class TestRequire:
|
||||
def test_bearer_require_forwards_kwargs(self):
|
||||
"""require() creates a new instance that passes merged kwargs to validator."""
|
||||
bearer = BearerTokenAuth(role_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(bearer.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_bearer_require_overrides_existing_kwarg(self):
|
||||
"""require() kwargs override kwargs set at instantiation."""
|
||||
bearer = BearerTokenAuth(role_validator, role="user")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(bearer.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()["role"] == "admin"
|
||||
|
||||
def test_bearer_require_preserves_prefix(self):
|
||||
"""require() keeps the prefix of the original instance."""
|
||||
bearer = BearerTokenAuth(role_validator, prefix="user_")
|
||||
derived = bearer.require(role="admin")
|
||||
assert derived._prefix == "user_"
|
||||
|
||||
def test_bearer_require_does_not_mutate_original(self):
|
||||
"""require() returns a new instance — original kwargs are unchanged."""
|
||||
bearer = BearerTokenAuth(role_validator, role="user")
|
||||
bearer.require(role="admin")
|
||||
assert bearer._kwargs == {"role": "user"}
|
||||
|
||||
def test_cookie_require_forwards_kwargs(self):
|
||||
async def scoped_validator(value: str, *, scope: str) -> dict:
|
||||
if value != VALID_COOKIE:
|
||||
raise UnauthorizedError()
|
||||
return {"session": value, "scope": scope}
|
||||
|
||||
cookie = CookieAuth("session", scoped_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(cookie.require(scope="admin"))):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/admin", cookies={"session": VALID_COOKIE})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"session": VALID_COOKIE, "scope": "admin"}
|
||||
|
||||
def test_cookie_require_preserves_name(self):
|
||||
cookie = CookieAuth("session", cookie_validator)
|
||||
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}"
|
||||
|
||||
async def prefixed_role_validator(credential: str, *, role: str) -> dict:
|
||||
if credential != PREFIXED_TOKEN:
|
||||
raise UnauthorizedError()
|
||||
return {"user": "alice", "role": role}
|
||||
|
||||
bearer = BearerTokenAuth(prefixed_role_validator, prefix="user_")
|
||||
multi = MultiAuth(bearer.require(role="admin"))
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(multi)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get(
|
||||
"/admin", headers={"Authorization": f"Bearer {PREFIXED_TOKEN}"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
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"
|
||||
|
||||
|
||||
# Minimal concrete subclass used only in tests below.
|
||||
class _HeaderAuth(AuthSource):
|
||||
"""Reads a custom X-Token header — no FastAPI security scheme."""
|
||||
|
||||
def __init__(self, secret: str) -> None:
|
||||
super().__init__()
|
||||
self._secret = secret
|
||||
|
||||
async def extract(self, request) -> str | None:
|
||||
return request.headers.get("X-Token") or None
|
||||
|
||||
async def authenticate(self, credential: str) -> dict:
|
||||
if credential != self._secret:
|
||||
raise UnauthorizedError()
|
||||
return {"token": credential}
|
||||
|
||||
|
||||
class TestAuthSource:
|
||||
def test_cannot_instantiate_abstract_class(self):
|
||||
with pytest.raises(TypeError):
|
||||
AuthSource()
|
||||
|
||||
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
|
||||
)
|
||||
assert isinstance(bearer, AuthSource)
|
||||
assert isinstance(cookie, AuthSource)
|
||||
assert isinstance(oauth, AuthSource)
|
||||
assert isinstance(oidc, AuthSource)
|
||||
|
||||
def test_custom_source_standalone_valid(self):
|
||||
"""Default __call__ wires extract + authenticate via Request injection."""
|
||||
auth = _HeaderAuth(secret="s3cr3t")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"X-Token": "s3cr3t"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"token": "s3cr3t"}
|
||||
|
||||
def test_custom_source_standalone_missing_credential(self):
|
||||
auth = _HeaderAuth(secret="s3cr3t")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me") # no X-Token header
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_custom_source_standalone_invalid_credential(self):
|
||||
auth = _HeaderAuth(secret="s3cr3t")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"X-Token": "wrong"})
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_custom_source_in_multi_auth(self):
|
||||
"""Custom AuthSource works transparently inside MultiAuth."""
|
||||
header_auth = _HeaderAuth(secret="s3cr3t")
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
multi = MultiAuth(bearer, header_auth)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(multi)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
|
||||
# Bearer matches first
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
# No bearer → falls through to custom header source
|
||||
response = client.get("/me", headers={"X-Token": "s3cr3t"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"token": "s3cr3t"}
|
||||
Reference in New Issue
Block a user