Fix imports.
[wolnelektury.git] / src / api / request_validator.py
1 # This file is part of Wolnelektury, licensed under GNU Affero GPLv3 or later.
2 # Copyright © Fundacja Nowoczesna Polska. See NOTICE for more information.
3 #
4 import time
5 from oauthlib.oauth1 import RequestValidator
6 from api.models import Consumer, Nonce, Token
7
8
9 class PistonRequestValidator(RequestValidator):
10     timestamp_threshold = 300
11
12     dummy_access_token = '!'
13     realms = ['API']
14
15     # Just for the tests.
16     # It'd be a little more kosher to use test client with secure=True.
17     enforce_ssl = False
18
19     # iOS app generates 8-char nonces.
20     nonce_length = 8, 250
21
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.
26
27     def check_client_key(self, client_key):
28         """We control the keys anyway."""
29         return True
30
31     def get_request_token_secret(self, client_key, token, request):
32         return request.token.secret
33
34     def get_access_token_secret(self, client_key, token, request):
35         if request.token:
36             return request.token.secret
37         else:
38             try:
39                 token = Token.objects.get(
40                     token_type=Token.ACCESS,
41                     consumer__key=client_key,
42                     key=token
43                 )
44             except: return None
45             return token.secret
46
47     def get_default_realms(self, client_key, request):
48         return ['API']
49
50     def validate_request_token(self, client_key, token, request):
51         try:
52             token = Token.objects.get(
53                 token_type=Token.REQUEST,
54                 consumer__key=client_key,
55                 key=token,
56                 is_approved=True,
57             )
58         except Token.DoesNotExist:
59             return False
60         else:
61             request.token = token
62             return True
63
64     def validate_access_token(self, client_key, token, request):
65         try:
66             token = Token.objects.get(
67                 token_type=Token.ACCESS,
68                 consumer__key=client_key,
69                 key=token
70             )
71         except Token.DoesNotExist:
72             return False
73         else:
74             request.token = token
75             return True
76
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:
80             return False
81         token = request_token or access_token
82         # Yes, this is what Piston did.
83         if token is None:
84             return True
85
86         nonce, created = Nonce.objects.get_or_create(consumer_key=client_key,
87                                                      token_key=token,
88                                                      key=nonce)
89         return created
90
91     def validate_client_key(self, client_key, request):
92         try:
93             request.oauth_consumer = Consumer.objects.get(key=client_key)
94         except Consumer.DoesNotExist:
95             return False
96         return True
97
98     def validate_realms(self, client_key, token, request, uri=None, realms=None):
99         return True
100
101     def validate_requested_realms(self, *args, **kwargs):
102         return True
103
104     def validate_redirect_uri(self, *args, **kwargs):
105         return True
106
107     def validate_verifier(self, client_key, token, verifier, request):
108         return True
109
110     def get_client_secret(self, client_key, request):
111         return request.oauth_consumer.secret
112
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,
120         )
121
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,
130         )
131
132     def verify_request_token(self, token, request):
133         return Token.objects.filter(
134             token_type=Token.REQUEST, key=token, is_approved=False
135         ).exists()
136
137     def get_realms(self, *args, **kwargs):
138         return []
139
140     def save_verifier(self, token, verifier, request):
141         Token.objects.filter(
142             token_type=Token.REQUEST,
143             key=token,
144             is_approved=False
145         ).update(
146             is_approved=True,
147             user=verifier['user']
148         )
149
150     def get_redirect_uri(self, token, request):
151         return request.redirect_uri
152
153     def invalidate_request_token(self, client_key, request_token, request):
154         Token.objects.filter(
155             token_type=Token.REQUEST,
156             key=request_token,
157             consumer__key=client_key,
158         )