-
Notifications
You must be signed in to change notification settings - Fork 256
Expand file tree
/
Copy pathBearerAuth.py
More file actions
63 lines (52 loc) · 2.17 KB
/
Copy pathBearerAuth.py
File metadata and controls
63 lines (52 loc) · 2.17 KB
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
__author__ = "Andrea Biancini, geduldig"
__date__ = "January 3, 2014"
__license__ = "MIT"
from .constants import *
import base64
import requests
OAUTH2_SUBDOMAIN = 'api'
OAUTH2_ENDPOINT = 'oauth2/token'
class BearerAuth(requests.auth.AuthBase):
"""Request bearer access token for oAuth2 authentication.
:param consumer_key: Twitter application consumer key
:param consumer_secret: Twitter application consumer secret
:param proxies: Dictionary of proxy URLs (see documentation for python-requests).
"""
def __init__(self, consumer_key, consumer_secret, proxies=None, user_agent=None):
self._consumer_key = consumer_key
self._consumer_secret = consumer_secret
self.proxies = proxies
self.user_agent = user_agent
self._bearer_token = self._get_access_token()
def _get_access_token(self):
token_url = '%s://%s.%s/%s' % (PROTOCOL,
OAUTH2_SUBDOMAIN,
DOMAIN,
OAUTH2_ENDPOINT)
auth = self._consumer_key + ':' + self._consumer_secret
b64_bearer_token_creds = base64.b64encode(auth.encode('utf8'))
params = {'grant_type': 'client_credentials'}
headers = {}
headers['User-Agent'] = self.user_agent
headers['Authorization'] = 'Basic ' + b64_bearer_token_creds.decode('utf8')
headers['Content-Type'] = 'application/x-www-form-urlencoded;charset=UTF-8'
try:
response = requests.post(
token_url,
params=params,
headers=headers,
proxies=self.proxies)
data = response.json()
return data['access_token']
except Exception as e:
raise Exception('Error requesting bearer access token: %s' % e)
def __call__(self, r):
auth_list = [
self._consumer_key,
self._consumer_secret,
self._bearer_token]
if all(auth_list):
r.headers['Authorization'] = "Bearer %s" % self._bearer_token
return r
else:
raise Exception('Not enough keys passed to Bearer token manager.')