import secrets from urllib.parse import urlparse, urlunparse from django.contrib.auth.mixins import LoginRequiredMixin from django.http import HttpResponseRedirect, JsonResponse from django.shortcuts import render from django.utils.decorators import method_decorator from django.views.decorators.csrf import csrf_exempt from django.views.generic import TemplateView, View from api.models import Application, Token from api.parser import FormOrJsonParser class OauthRedirect(HttpResponseRedirect): def __init__(self, redirect_uri, key, value): url_parts = urlparse(redirect_uri) self.allowed_schemes = [url_parts.scheme] # Either add or join the query section url_parts = list(url_parts) if url_parts[4]: url_parts[4] = url_parts[4] + f"&{key}={value}" else: url_parts[4] = f"{key}={value}" super().__init__(urlunparse(url_parts)) class AuthorizationView(LoginRequiredMixin, TemplateView): """ Asks the user to authorize access. Could maybe be a FormView, but things are weird enough we just handle the POST manually. """ template_name = "api/oauth_authorize.html" def get_context_data(self): redirect_uri = self.request.GET["redirect_uri"] scope = self.request.GET.get("scope", "read") try: application = Application.objects.get( client_id=self.request.GET["client_id"] ) except (Application.DoesNotExist, KeyError): return OauthRedirect(redirect_uri, "error", "invalid_application") return { "application": application, "redirect_uri": redirect_uri, "scope": scope, "identities": self.request.user.identities.all(), } def post(self, request): post_data = FormOrJsonParser().parse_body(request) # Grab the application and other details again redirect_uri = post_data["redirect_uri"] scope = post_data["scope"] application = Application.objects.get(client_id=post_data["client_id"]) # Get the identity identity = self.request.user.identities.get(pk=post_data["identity"]) # Make a token token = Token.objects.create( application=application, user=self.request.user, identity=identity, token=secrets.token_urlsafe(32), code=secrets.token_urlsafe(16), scopes=scope.split(), ) # If it's an out of band request, show the code if redirect_uri == "urn:ietf:wg:oauth:2.0:oob": return render(request, "api/oauth_code.html", {"code": token.code}) # Redirect with the token's code return OauthRedirect(redirect_uri, "code", token.code) @method_decorator(csrf_exempt, name="dispatch") class TokenView(View): def post(self, request): post_data = FormOrJsonParser().parse_body(request) grant_type = post_data["grant_type"] try: application = Application.objects.get(client_id=post_data["client_id"]) except (Application.DoesNotExist, KeyError): return JsonResponse({"error": "invalid_client_id"}, status=400) # TODO: Implement client credentials flow if grant_type == "client_credentials": return JsonResponse({"error": "invalid_grant_type"}, status=400) elif grant_type == "authorization_code": code = post_data["code"] # Retrieve the token by code # TODO: Check code expiry based on created date try: token = Token.objects.get(code=code, application=application) except Token.DoesNotExist: return JsonResponse({"error": "invalid_code"}, status=400) # Update the token to remove its code token.code = None token.save() # Return them the token return JsonResponse( { "access_token": token.token, "token_type": "Bearer", "scope": " ".join(token.scopes), "created_at": int(token.created.timestamp()), } ) class RevokeTokenView(View): pass