"""
Utilities for managing OIDC settings for `Flask`_ applications
.. -`Flask`: https://flask.palletsprojects.com/en/1.1.x/
"""
import collections
import functools
import typing
from authlib.integrations.flask_oauth2 import ResourceProtector
import flask
from fictive.flask.extension import AbstractFlaskExtension
from fictive.oidc.validator import OIDCBearerTokenValidator
[docs]class OIDC(AbstractFlaskExtension):
"""
A `Flask`_ extension for managing OIDC settings
.. -`Flask`: https://flask.palletsprojects.com/en/1.1.x/
"""
EXTENSION_KEY = 'OIDC'
PROTECTOR_CLASS = ResourceProtector
VALIDATOR_CLASS = OIDCBearerTokenValidator
@property
def resource_protector(self) -> ResourceProtector:
"""
an appropriate resource protector configured for the current application
"""
app = self.app or flask.current_app
appenv = self.appenv(app)
storage = self.context_storage(app)
try:
return storage.protector
except AttributeError:
nested_config = flask.Config(None)
nested_config.update(appenv.kwargs)
protector_kwargs = nested_config.get('protector', {})
protector_kwargs.update(nested_config.get_namespace('protector_'))
storage.protector = self.PROTECTOR_CLASS(**protector_kwargs)
validator_kwargs = nested_config.get('validator', {})
validator_kwargs.update(nested_config.get_namespace('validator_'))
storage.protector.register_token_validator(
self.VALIDATOR_CLASS(**validator_kwargs))
return storage.protector
[docs] def require_jwt(
self,
scopes: typing.Union[str, collections.abc.Iterable[str]] = None,
optional: bool = False) -> typing.Callable:
"""
dynamically invoke the resource protector for the current flask app context
"""
def view_decorator(view_func: collections.abc.Callable) -> typing.Any:
@functools.wraps(view_func)
def decorated_view(
*args: typing.Any, **kwargs: typing.Any) -> typing.Any:
protector = self.resource_protector(
scopes=scopes, optional=optional)
return protector(view_func)(*args, **kwargs)
return decorated_view
return view_decorator