93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
import logging
|
|
|
|
from django.core.exceptions import ImproperlyConfigured
|
|
from oauth2_provider.contrib.rest_framework import TokenMatchesOASRequirements, OAuth2Authentication
|
|
from rest_framework.permissions import (
|
|
IsAuthenticated, BasePermission
|
|
)
|
|
|
|
logger = logging.getLogger("oauth2_provider")
|
|
|
|
|
|
class IsAuthenticatedOrTokenMatchesOASRequirements(TokenMatchesOASRequirements):
|
|
def has_permission(self, request, view):
|
|
logger.debug(f'try to authenticate {request} for {view} in IsAuthenticatedOrTokenMatchesOASRequirements')
|
|
is_authenticated = IsAuthenticated().has_permission(request, view)
|
|
logger.debug(f'is_authenticated: {is_authenticated}')
|
|
oauth2authenticated = False
|
|
if is_authenticated:
|
|
oauth2authenticated = isinstance(request.successful_authenticator, OAuth2Authentication)
|
|
|
|
logger.debug(f'oauth2authenticated: {oauth2authenticated}')
|
|
|
|
token_has_scope = TokenMatchesOASRequirements()
|
|
logger.debug(f'token_has_scope: {token_has_scope}')
|
|
|
|
result = (is_authenticated and not oauth2authenticated) or token_has_scope.has_permission(request, view)
|
|
logger.debug(f'authentication result: {result}')
|
|
return result
|
|
|
|
|
|
class TokenMatchesViewSetActions(BasePermission):
|
|
"""
|
|
:attr:action_required_scopes: dict keyed by view set action name with value: iterable action scope lists
|
|
|
|
This fulfills the [Open API Specification (OAS; formerly Swagger)](https://www.openapis.org/)
|
|
list of alternative Security Requirements Objects for oauth2 or openIdConnect:
|
|
When a list of Security Requirement Objects is defined on the Open API object or Operation Object,
|
|
only one of Security Requirement Objects in the list needs to be satisfied to authorize the request.
|
|
[1](https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.0.md#securityRequirementObject)
|
|
|
|
For each method, a list of lists of allowed scopes is tried in order and the first to match succeeds.
|
|
|
|
@example
|
|
required_action_scopes = {
|
|
'list': [['read']],
|
|
'create': [['create1','scope2'], ['alt-scope3'], ['alt-scope4','alt-scope5']],
|
|
}
|
|
|
|
TODO: DRY: subclass TokenHasScope and iterate over values of required_scope?
|
|
"""
|
|
|
|
def has_permission(self, request, view):
|
|
token = request.auth
|
|
|
|
if not token:
|
|
return False
|
|
|
|
if hasattr(token, "scope"): # OAuth 2
|
|
required_action_scopes = self.get_required_action_scopes(request, view)
|
|
|
|
# TODO: use action map instead to analyze method
|
|
action_map = view.action_map
|
|
a = view.action
|
|
|
|
if a in required_action_scopes:
|
|
logger.debug(
|
|
"Required scopes alternatives to access resource: {0}".format(
|
|
required_action_scopes[a]
|
|
)
|
|
)
|
|
for alt in required_action_scopes[a]:
|
|
if token.is_valid(alt):
|
|
return True
|
|
return False
|
|
else:
|
|
logger.warning("no scope action defined for action {0}".format(a))
|
|
return False
|
|
|
|
assert False, (
|
|
"TokenMatchesViewSetActions requires the"
|
|
"`oauth2_provider.rest_framework.OAuth2Authentication` authentication "
|
|
"class to be used."
|
|
)
|
|
|
|
def get_required_action_scopes(self, request, view):
|
|
try:
|
|
return getattr(view, "required_action_scopes")
|
|
except AttributeError:
|
|
raise ImproperlyConfigured(
|
|
"TokenMatchesViewSetActions requires the view to"
|
|
" define the required_action_scopes attribute"
|
|
)
|
|
|