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
|
import logging
from rest_framework import generics, status, permissions
from rest_framework.response import Response
from django.conf import settings
from requests_oauthlib import OAuth2Session
from django.contrib.auth import get_user_model
from djoser.conf import settings as djoser_settings
from django.shortcuts import render
from django.http import HttpResponseNotFound
from djoser import utils
from djoser.serializers import TokenSerializer
from authAPI.serializers import TokenCreateSerializer
from blocks.settings import DOMAIN
Token = djoser_settings.TOKEN_MODEL
# Set up logging
logger = logging.getLogger(__name__)
def activate_user(request, uid, token):
"""
Used to activate accounts,
sends POST request to /api/auth/users/activation/ route
internally to activate account.
Link to this route is sent via email to user for verification
"""
protocol = 'https://' if request.is_secure() else 'http://'
web_url = protocol + request.get_host() + '/api/auth/users/activation/' # URL comes from Djoser library
return render(request, 'activate_user.html',
{'uid': uid,
'token': token,
'activation_url': web_url,
'redirect_url': settings.POST_ACTIVATE_REDIRECT_URL
})
def get_social_user(email, request, callback, service):
if not email:
logger.error(f'Email not found for {service} user')
return HttpResponseNotFound('<h1>Email not found</h1>')
user, created = get_user_model().objects.get_or_create(email=email)
if created:
user.username = email
user.save()
if not user.is_active:
user.is_active = True
user.save()
token, created = Token.objects.get_or_create(user=user)
protocol = 'https://' if request.is_secure() else 'http://'
web_url = protocol + DOMAIN + '/#/dashboard'
return render(request, callback,
{'token': token,
'url': web_url
})
def GoogleOAuth2(request):
state = request.GET.get('state', None)
code = request.GET.get('code', None)
if state is None or state == '' or code is None or code == '':
return HttpResponseNotFound('<h1>Page Not Found</h1>')
client_id = settings.SOCIAL_AUTH_GOOGLE_OAUTH2_KEY
client_secret = settings.SOCIAL_AUTH_GOOGLE_OAUTH2_SECRET
google = OAuth2Session(
client_id,
redirect_uri=settings.GOOGLE_OAUTH_REDIRECT_URI,
state=state
)
google.fetch_token(
'https://accounts.google.com/o/oauth2/token',
client_secret=client_secret,
code=code
)
user_info = google.get(
'https://www.googleapis.com/oauth2/v1/userinfo').json()
return get_social_user(user_info['email'], request, 'google_callback.html', 'google')
def GitHubOAuth2(request):
state = request.GET.get('state', None)
code = request.GET.get('code', None)
if state is None or state == '' or code is None or code == '':
return HttpResponseNotFound('<h1>Page Not Found</h1>')
client_id = settings.SOCIAL_AUTH_GITHUB_KEY
client_secret = settings.SOCIAL_AUTH_GITHUB_SECRET
github = OAuth2Session(
client_id,
redirect_uri=settings.GITHUB_OAUTH_REDIRECT_URI,
state=state
)
try:
github.fetch_token(
'https://github.com/login/oauth/access_token',
client_secret=client_secret,
code=code
)
except Exception as e:
logger.error(f'Failed to fetch token from GitHub: {e}')
return HttpResponseNotFound('<h1>Failed to authenticate with GitHub</h1>')
try:
user_info = github.get('https://api.github.com/user').json()
except Exception as e:
logger.error(f'Failed to get user info from GitHub: {e}')
return HttpResponseNotFound('<h1>Failed to get user info from GitHub</h1>')
primary_email = f"{user_info['id']}+{user_info['login']}@users.noreply.github.com"
return get_social_user(primary_email, request, 'github_callback.html', 'github')
class CustomTokenCreateView(utils.ActionViewMixin, generics.GenericAPIView):
"""
Use this endpoint to obtain user authentication token.
"""
serializer_class = TokenCreateSerializer
permission_classes = [permissions.AllowAny]
def _action(self, serializer):
token = utils.login_user(self.request, serializer.user)
token_serializer_class = TokenSerializer
data = {
'auth_token': token_serializer_class(token).data['auth_token'],
'user_id': serializer.user.id
}
return Response(data=data, status=status.HTTP_200_OK)
|
