1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
import secrets
from urllib.parse import urlparse
from django.contrib.auth.mixins import LoginRequiredMixin
from django.http import HttpResponseRedirect, JsonResponse
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):
self.allowed_schemes = [urlparse(redirect_uri).scheme]
super().__init__(redirect_uri + f"?{key}={value}")
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(),
)
# 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
|
