162 lines
5.7 KiB
Python
162 lines
5.7 KiB
Python
import calendar
|
||
import functools
|
||
import json
|
||
|
||
from django.core.exceptions import ObjectDoesNotExist
|
||
from django.http import HttpResponse, HttpResponseForbidden
|
||
from django.utils.decorators import method_decorator
|
||
from django.views.decorators.csrf import csrf_exempt
|
||
from django.views.decorators.http import require_http_methods
|
||
from oauth2_provider.models import get_access_token_model
|
||
from oauth2_provider.oauth2_backends import OAuthLibCore
|
||
from oauth2_provider.views import ClientProtectedScopedResourceView
|
||
from oauthlib.oauth2 import Server
|
||
|
||
from apps.gooyal_oauth2.validators import IntrospectOAuth2Validator
|
||
|
||
|
||
@method_decorator(csrf_exempt, name="dispatch")
|
||
class IntrospectTokenView(ClientProtectedScopedResourceView):
|
||
"""
|
||
Implements an endpoint for token introspection based
|
||
on RFC 7662 https://tools.ietf.org/html/rfc7662
|
||
|
||
To access this view the request must pass a OAuth2 Bearer Token
|
||
which is allowed to access the scope `introspection`.
|
||
"""
|
||
required_scopes = ["introspection"]
|
||
|
||
@staticmethod
|
||
def get_token_response(token_value=None):
|
||
try:
|
||
token = get_access_token_model().objects.get(token=token_value)
|
||
except ObjectDoesNotExist:
|
||
return HttpResponse(
|
||
content=json.dumps({"active": False}),
|
||
status=401,
|
||
content_type="application/json"
|
||
)
|
||
else:
|
||
if token.is_valid():
|
||
data = {
|
||
"active": True,
|
||
"scope": token.scope,
|
||
"exp": int(calendar.timegm(token.expires.timetuple())),
|
||
}
|
||
if token.application:
|
||
data["client_id"] = token.application.client_id
|
||
if token.user:
|
||
data["username"] = token.user.get_username()
|
||
return HttpResponse(content=json.dumps(data), status=200, content_type="application/json")
|
||
else:
|
||
return HttpResponse(content=json.dumps({
|
||
"active": False,
|
||
}), status=200, content_type="application/json")
|
||
|
||
def get(self, request, *args, **kwargs):
|
||
"""
|
||
Get the token from the URL parameters.
|
||
URL: https://example.com/introspect?token=mF_9.B5f-4.1JqM
|
||
|
||
:param request:
|
||
:param args:
|
||
:param kwargs:
|
||
:return:
|
||
"""
|
||
return self.get_token_response(request.GET.get("token", None))
|
||
|
||
def post(self, request, *args, **kwargs):
|
||
"""
|
||
Get the token from the body form parameters.
|
||
Body: token=mF_9.B5f-4.1JqM
|
||
|
||
:param request:
|
||
:param args:
|
||
:param kwargs:
|
||
:return:
|
||
"""
|
||
return self.get_token_response(request.POST.get("token", None))
|
||
|
||
|
||
def protected_resource(scopes=None):
|
||
"""
|
||
Implementation of protected_resource decorator that saves the client on the
|
||
request for the view function to use.
|
||
|
||
Cribbed from django-oauth-toolkit.
|
||
"""
|
||
_scopes = scopes or []
|
||
|
||
def decorator(view_func):
|
||
@functools.wraps(view_func)
|
||
def _validate(request, *args, **kwargs):
|
||
validator = IntrospectOAuth2Validator()
|
||
core = OAuthLibCore(Server(validator))
|
||
valid, oauthlib_req = core.verify_request(request, scopes=_scopes)
|
||
if valid:
|
||
request.client = oauthlib_req.client
|
||
request.resource_owner = oauthlib_req.user
|
||
return view_func(request, *args, **kwargs)
|
||
return HttpResponseForbidden()
|
||
|
||
return _validate
|
||
|
||
return decorator
|
||
|
||
|
||
@require_http_methods(['GET', 'POST'])
|
||
@csrf_exempt
|
||
@protected_resource(scopes=['introspection'])
|
||
def introspect_token(request):
|
||
"""
|
||
Version of the introspection view protected by a regular scope instead of
|
||
read-write scopes.
|
||
|
||
Also allows for the required scope to be changed using a setting.
|
||
"""
|
||
if request.method == 'GET':
|
||
token = request.GET.get("token", None)
|
||
else:
|
||
token = request.POST.get("token", None)
|
||
return IntrospectTokenView.get_token_response(token)
|
||
|
||
|
||
# @require_POST
|
||
# @csrf_exempt
|
||
# @protected_resource(scopes=[settings.REGISTER_SCOPE_SCOPE])
|
||
# def register_scope(request):
|
||
# """
|
||
# Implements an endpoint for registering a scope.
|
||
# """
|
||
# # Get the scope data from the request body
|
||
# scope_data = json.loads(request.body) if request.body else {}
|
||
# try:
|
||
# try:
|
||
# # If a scope with the given name already exists, find it
|
||
# scope = Scope.objects.get(name=scope_data['name'])
|
||
# except Scope.DoesNotExist:
|
||
# # If no scope with the given name exists, create it
|
||
# _ = Scope.objects.create(
|
||
# application=request.client,
|
||
# name=scope_data['name'],
|
||
# description=scope_data['description'],
|
||
# is_default=scope_data.get('is_default', False)
|
||
# )
|
||
# # Respond with a 201 Created
|
||
# return HttpResponse(status=201)
|
||
# except KeyError as exc:
|
||
# # A key missing in the data should be reported as a bad request
|
||
# return HttpResponse(
|
||
# status=400,
|
||
# content="'{}' must be given in request data".format(exc.args[0]),
|
||
# content_type='text/plain'
|
||
# )
|
||
# # If the scope does exist, check that the current application is the
|
||
# # owner of the scope before updating it
|
||
# if scope.application and scope.application == request.client:
|
||
# scope.description = scope_data['description']
|
||
# scope.is_default = scope_data.get('is_default', False)
|
||
# scope.save()
|
||
# return HttpResponse(status=200)
|
||
# else:
|
||
# return HttpResponse(status=403)
|