src/api/request_validator.py

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