1 # This file is part of Wolnelektury, licensed under GNU Affero GPLv3 or later.
2 # Copyright © Fundacja Nowoczesna Polska. See NOTICE for more information.
5 from oauthlib.oauth1 import RequestValidator
6 from api.models import Consumer, Nonce, Token
9 class PistonRequestValidator(RequestValidator):
10 timestamp_threshold = 300
12 dummy_access_token = '!'
16 # It'd be a little more kosher to use test client with secure=True.
19 # iOS app generates 8-char nonces.
22 # Because Token.key is char(18).
23 request_token_length = 18, 32
24 access_token_length = 18, 32
25 # TODO: oauthlib request-access switch.
27 def check_client_key(self, client_key):
28 """We control the keys anyway."""
31 def get_request_token_secret(self, client_key, token, request):
32 return request.token.secret
34 def get_access_token_secret(self, client_key, token, request):
36 return request.token.secret
39 token = Token.objects.get(
40 token_type=Token.ACCESS,
41 consumer__key=client_key,
47 def get_default_realms(self, client_key, request):
50 def validate_request_token(self, client_key, token, request):
52 token = Token.objects.get(
53 token_type=Token.REQUEST,
54 consumer__key=client_key,
58 except Token.DoesNotExist:
64 def validate_access_token(self, client_key, token, request):
66 token = Token.objects.get(
67 token_type=Token.ACCESS,
68 consumer__key=client_key,
71 except Token.DoesNotExist:
77 def validate_timestamp_and_nonce(self, client_key, timestamp, nonce,
78 request, request_token=None, access_token=None):
79 if abs(time.time() - int(timestamp)) > self.timestamp_threshold:
81 token = request_token or access_token
82 # Yes, this is what Piston did.
86 nonce, created = Nonce.objects.get_or_create(consumer_key=client_key,
91 def validate_client_key(self, client_key, request):
93 request.oauth_consumer = Consumer.objects.get(key=client_key)
94 except Consumer.DoesNotExist:
98 def validate_realms(self, client_key, token, request, uri=None, realms=None):
101 def validate_requested_realms(self, *args, **kwargs):
104 def validate_redirect_uri(self, *args, **kwargs):
107 def validate_verifier(self, client_key, token, verifier, request):
110 def get_client_secret(self, client_key, request):
111 return request.oauth_consumer.secret
113 def save_request_token(self, token, request):
114 Token.objects.create(
115 token_type=Token.REQUEST,
116 timestamp=request.timestamp,
117 key=token['oauth_token'],
118 secret=token['oauth_token_secret'],
119 consumer=request.oauth_consumer,
122 def save_access_token(self, token, request):
123 Token.objects.create(
124 token_type=Token.ACCESS,
125 timestamp=request.timestamp,
126 key=token['oauth_token'],
127 secret=token['oauth_token_secret'],
128 consumer=request.oauth_consumer,
129 user=request.token.user,
132 def verify_request_token(self, token, request):
133 return Token.objects.filter(
134 token_type=Token.REQUEST, key=token, is_approved=False
137 def get_realms(self, *args, **kwargs):
140 def save_verifier(self, token, verifier, request):
141 Token.objects.filter(
142 token_type=Token.REQUEST,
147 user=verifier['user']
150 def get_redirect_uri(self, token, request):
151 return request.redirect_uri
153 def invalidate_request_token(self, client_key, request_token, request):
154 Token.objects.filter(
155 token_type=Token.REQUEST,
157 consumer__key=client_key,