import warnings
from collections.abc import Callable, Iterable
from typing import Any
from uuid import UUID
from fastapi.params import Security
from .errors import AudienceWarning
from .jwt import DEFAULT_ISSUERS, DEFAULT_PUBLIC_KEYS, KeysType
from .roles import Role, deprecated_role_reason
from .token_scheme import TokenScheme
EMPTY = frozenset(())
TEST_CTX, LIVE_CTX = "test", "live"
KNOWN_SECURITY_CONTEXTS = (TEST_CTX, LIVE_CTX)
def check_audience_values(audience: frozenset[str]):
security_contexts = set()
for aud in audience:
ctx, _, _ = aud.partition(":")
security_contexts.add(ctx)
if ctx not in KNOWN_SECURITY_CONTEXTS:
msg = (
f"Audience value '{aud}' has missing or unknown security context. "
"By convention audience values should be prefixed with 'test:' or 'live:', "
"e.g. AUDIENCE='live:myapi'. "
"We suggest defaulting to live: and overriding in test environments."
)
warnings.warn(msg, stacklevel=3, category=AudienceWarning)
if TEST_CTX in security_contexts and LIVE_CTX in security_contexts:
msg = (
f"audience={sorted(audience)} includes both test and live security context. "
"Unless corrected, this configuration might lead to your live "
"system accepting bogus test tokens."
)
warnings.warn(msg, stacklevel=3, category=AudienceWarning)
[docs]
class SecurityRequires(Security):
def __init__(
self,
dependency: Callable[..., Any] | None,
*,
scopes: Iterable[str],
roles: Iterable[Role],
groups: Iterable[str],
app_ids: Iterable[str],
use_cache: bool = True,
):
super().__init__(dependency=dependency, use_cache=use_cache, scopes=tuple(scopes))
object.__setattr__(self, "roles", frozenset(roles))
object.__setattr__(self, "groups", frozenset(groups))
object.__setattr__(self, "app_ids", frozenset(app_ids))
def Requires(
*,
audience: str | Iterable[str],
roles: Iterable[Role | str],
scopes: Iterable[str] = EMPTY,
app_ids: Iterable[str | UUID] = EMPTY,
groups: Iterable[str] = EMPTY,
keys: KeysType = DEFAULT_PUBLIC_KEYS,
issuer: str | Iterable[str] = DEFAULT_ISSUERS,
):
if isinstance(issuer, str):
issuer = (issuer,)
if isinstance(audience, str):
audience = (audience,)
audience_set = frozenset(audience)
check_audience_values(audience_set)
roles_set = frozenset(Role(r) for r in roles)
scopes_set = frozenset(scopes)
groups_set = frozenset(groups)
app_ids_set = frozenset(map(str, app_ids))
for role in roles_set:
if reason := deprecated_role_reason(role):
warnings.warn(
f"Role.{role.name} is deprecated: {reason}",
category=DeprecationWarning,
stacklevel=2,
)
if app_ids_set and Role.APP2APP not in roles_set:
msg = "When setting app_ids, you must include Role.APP2APP in your roles."
raise ValueError(msg)
elif Role.APP2APP in roles_set and not app_ids_set:
msg = (
"When allowing Role.APP2APP, you must set 'app_ids' "
"to say which specific apps may call this endpoint."
)
raise ValueError(msg)
elif Role.APP2APP not in roles_set and not scopes:
msg = "You must set required 'scopes' unless using Role.APP2APP"
raise ValueError(msg)
auth_context_from_token = TokenScheme(
audiences=audience_set,
roles=roles_set,
scopes=scopes_set,
groups=groups_set,
app_ids=app_ids_set,
keys=keys,
issuers=issuer,
)
return SecurityRequires(
auth_context_from_token,
roles=roles_set,
scopes=scopes_set,
groups=groups_set,
app_ids=app_ids_set,
)