This repository has been archived by the owner on May 26, 2020. It is now read-only.
/
serializers.py
135 lines (105 loc) · 4.41 KB
/
serializers.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
from calendar import timegm
from datetime import datetime, timedelta
import jwt
from django.contrib.auth import authenticate
from rest_framework import serializers
from .compat import Serializer
from rest_framework_jwt import utils
from rest_framework_jwt.settings import api_settings
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
jwt_decode_handler = api_settings.JWT_DECODE_HANDLER
jwt_get_user_id_from_payload = api_settings.JWT_PAYLOAD_GET_USER_ID_HANDLER
class JSONWebTokenSerializer(Serializer):
"""
Serializer class used to validate a username and password.
'username' is identified by the custom UserModel.USERNAME_FIELD.
Returns a JSON Web Token that can be used to authenticate later calls.
"""
password = serializers.CharField(write_only=True)
def __init__(self, *args, **kwargs):
"""Dynamically add the USERNAME_FIELD to self.fields."""
super(JSONWebTokenSerializer, self).__init__(*args, **kwargs)
self.fields[self.username_field] = serializers.CharField()
@property
def username_field(self):
User = utils.get_user_model()
try:
return User.USERNAME_FIELD
except AttributeError:
return 'username'
def validate(self, attrs):
credentials = {self.username_field: attrs.get(self.username_field),
'password': attrs.get('password')}
if all(credentials.values()):
user = authenticate(**credentials)
if user:
if not user.is_active:
msg = 'User account is disabled.'
raise serializers.ValidationError(msg)
payload = jwt_payload_handler(user)
# Include original issued at time for a brand new token,
# to allow token refresh
if api_settings.JWT_ALLOW_REFRESH:
payload['orig_iat'] = timegm(
datetime.utcnow().utctimetuple()
)
return {
'token': jwt_encode_handler(payload)
}
else:
msg = 'Unable to login with provided credentials.'
raise serializers.ValidationError(msg)
else:
msg = 'Must include "username" and "password"'
raise serializers.ValidationError(msg)
class RefreshJSONWebTokenSerializer(Serializer):
"""
Check an access token
"""
token = serializers.CharField()
def validate(self, attrs):
User = utils.get_user_model()
token = attrs['token']
# Check payload valid (based off of JSONWebTokenAuthentication,
# may want to refactor)
try:
payload = jwt_decode_handler(token)
except jwt.ExpiredSignature:
msg = 'Signature has expired.'
raise serializers.ValidationError(msg)
except jwt.DecodeError:
msg = 'Error decoding signature.'
raise serializers.ValidationError(msg)
# Make sure user exists (may want to refactor this)
try:
user_id = jwt_get_user_id_from_payload(payload)
if user_id is not None:
user = User.objects.get(pk=user_id, is_active=True)
else:
msg = 'Invalid payload'
raise serializers.ValidationError(msg)
except User.DoesNotExist:
msg = "User doesn't exist"
raise serializers.ValidationError(msg)
# Get and check 'orig_iat'
orig_iat = payload.get('orig_iat')
if orig_iat:
# Verify expiration
refresh_limit = api_settings.JWT_REFRESH_EXPIRATION_DELTA
if isinstance(refresh_limit, timedelta):
refresh_limit = (refresh_limit.days * 24 * 3600 +
refresh_limit.seconds)
expiration_timestamp = orig_iat + int(refresh_limit)
now_timestamp = timegm(datetime.utcnow().utctimetuple())
if now_timestamp > expiration_timestamp:
msg = 'Refresh has expired'
raise serializers.ValidationError(msg)
else:
msg = 'orig_iat field is required'
raise serializers.ValidationError(msg)
new_payload = jwt_payload_handler(user)
new_payload['orig_iat'] = orig_iat
return {
'token': jwt_encode_handler(new_payload)
}