74 lines
2.9 KiB
Python
74 lines
2.9 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
|
|
)
|
|
|
|
log = logging.getLogger("oauth2_provider")
|
|
|
|
|
|
class IsAuthenticatedOrTokenMatchesOASRequirements(TokenMatchesOASRequirements):
|
|
def has_permission(self, request, view):
|
|
is_authenticated = IsAuthenticated().has_permission(request, view)
|
|
oauth2authenticated = False
|
|
if is_authenticated:
|
|
oauth2authenticated = isinstance(request.successful_authenticator, OAuth2Authentication)
|
|
|
|
token_has_scope = TokenMatchesOASRequirements()
|
|
return (is_authenticated and not oauth2authenticated) or token_has_scope.has_permission(request, view)
|
|
|
|
|
|
class ActionMatchesOASRequirements(BasePermission):
|
|
def has_permission(self, request, view):
|
|
token = request.auth
|
|
|
|
if not token:
|
|
return False
|
|
|
|
if hasattr(token, "scope"): # OAuth 2
|
|
# {'get': 'retrieve', 'put': 'update', 'patch': 'partial_update', 'delete': 'destroy', 'head': 'retrieve'}
|
|
|
|
required_action_scopes = self.get_required_action_scopes(request, view)
|
|
|
|
action = request.action.upper()
|
|
if action in required_action_scopes:
|
|
log.debug(
|
|
"Required scopes actions to access resource: {0}".format(
|
|
required_action_scopes[action]
|
|
)
|
|
)
|
|
for alt in required_action_scopes[action]:
|
|
if token.is_valid(alt):
|
|
return True
|
|
return False
|
|
else:
|
|
log.warning("no scope alternates defined for method {0}".format(m))
|
|
return False
|
|
|
|
assert False, (
|
|
"ActionMatchesOASRequirements 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(
|
|
"ActionMatchesOASRequirements requires the view to"
|
|
" define the required_action_scopes attribute"
|
|
)
|
|
|
|
|
|
class IsAuthenticatedOrActionMatchesOASRequirements(ActionMatchesOASRequirements):
|
|
def has_permission(self, request, view):
|
|
is_authenticated = IsAuthenticated().has_permission(request, view)
|
|
oauth2authenticated = False
|
|
if is_authenticated:
|
|
oauth2authenticated = isinstance(request.successful_authenticator, OAuth2Authentication)
|
|
|
|
action_has_scope = ActionMatchesOASRequirements()
|
|
return (is_authenticated and not oauth2authenticated) or action_has_scope.has_permission(request, view)
|