/
views.py
146 lines (120 loc) · 5.1 KB
/
views.py
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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
import json
from datetime import timedelta
from django.http import HttpResponseNotAllowed, HttpResponseRedirect
from django.urls import reverse
from django.utils import timezone
from django.utils.http import urlencode
from django.views.decorators.csrf import csrf_exempt
from allauth.socialaccount.internal import jwtkit
from allauth.socialaccount.models import SocialToken
from allauth.socialaccount.providers.oauth2.views import (
OAuth2Adapter,
OAuth2CallbackView,
OAuth2LoginView,
)
from allauth.utils import build_absolute_uri, get_request_param
from .apple_session import get_apple_session
from .client import AppleOAuth2Client
class AppleOAuth2Adapter(OAuth2Adapter):
client_class = AppleOAuth2Client
provider_id = "apple"
access_token_url = "https://appleid.apple.com/auth/token"
authorize_url = "https://appleid.apple.com/auth/authorize"
public_key_url = "https://appleid.apple.com/auth/keys"
@classmethod
def get_verified_identity_data(cls, provider, id_token):
data = jwtkit.verify_and_decode(
credential=id_token,
keys_url=cls.public_key_url,
issuer="https://appleid.apple.com",
audience=provider.get_auds(),
lookup_kid=jwtkit.lookup_kid_jwk,
)
return data
def parse_token(self, data):
token = SocialToken(
token=data["access_token"],
)
token.token_secret = data.get("refresh_token", "")
expires_in = data.get(self.expires_in_key)
if expires_in:
token.expires_at = timezone.now() + timedelta(seconds=int(expires_in))
# `user_data` is a big flat dictionary with the parsed JWT claims
# access_tokens, and user info from the apple post.
identity_data = AppleOAuth2Adapter.get_verified_identity_data(
self.get_provider(), data["id_token"]
)
token.user_data = {**data, **identity_data}
return token
def complete_login(self, request, app, token, **kwargs):
extra_data = token.user_data
login = self.get_provider().sociallogin_from_response(
request=request, response=extra_data
)
login.state["id_token"] = token.user_data
# We can safely remove the apple login session now
# Note: The cookie will remain, but it's set to delete on browser close
get_apple_session(request).delete()
return login
def get_user_scope_data(self, request):
user_scope_data = request.apple_login_session.get("user", "")
try:
return json.loads(user_scope_data)
except json.JSONDecodeError:
# We do not care much about user scope data as it maybe blank
# so return blank dictionary instead
return {}
def get_access_token_data(self, request, app, client, pkce_code_verifier=None):
"""We need to gather the info from the apple specific login"""
apple_session = get_apple_session(request)
# Exchange `code`
code = get_request_param(request, "code")
access_token_data = client.get_access_token(
code, pkce_code_verifier=pkce_code_verifier
)
id_token = access_token_data.get("id_token", None)
# In case of missing id_token in access_token_data
if id_token is None:
id_token = apple_session.store.get("id_token")
return {
**access_token_data,
**self.get_user_scope_data(request),
"id_token": id_token,
}
@csrf_exempt
def apple_post_callback(request, finish_endpoint_name="apple_finish_callback"):
"""
Apple uses a `form_post` response type, which due to
CORS/Samesite-cookie rules means this request cannot access
the request since the session cookie is unavailable.
We work around this by storing the apple response in a
separate, temporary session and redirecting to a more normal
oauth flow.
args:
finish_endpoint_name (str): The name of a defined URL, which can be
overridden in your url configuration if you have more than one
callback endpoint.
"""
if request.method != "POST":
return HttpResponseNotAllowed(["POST"])
apple_session = get_apple_session(request)
# Add regular OAuth2 params to the URL - reduces the overrides required
keys_to_put_in_url = ["code", "state", "error"]
url_params = {}
for key in keys_to_put_in_url:
value = get_request_param(request, key, "")
if value:
url_params[key] = value
# Add other params to the apple_login_session
keys_to_save_to_session = ["user", "id_token"]
for key in keys_to_save_to_session:
apple_session.store[key] = get_request_param(request, key, "")
url = build_absolute_uri(request, reverse(finish_endpoint_name))
response = HttpResponseRedirect(
"{url}?{query}".format(url=url, query=urlencode(url_params))
)
apple_session.save(response)
return response
oauth2_login = OAuth2LoginView.adapter_view(AppleOAuth2Adapter)
oauth2_callback = apple_post_callback
oauth2_finish_login = OAuth2CallbackView.adapter_view(AppleOAuth2Adapter)