This repository was archived by the owner on Oct 23, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 6
Expand file tree
/
Copy pathauth.py
More file actions
140 lines (116 loc) · 4.55 KB
/
Copy pathauth.py
File metadata and controls
140 lines (116 loc) · 4.55 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
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
136
137
138
139
140
"""Authentication operations."""
import logging
import requests
from objectrocket import bases
from objectrocket import errors
logger = logging.getLogger(__name__)
class Auth(bases.BaseAuthLayer):
"""Authentication operations.
:param objectrocket.client.Client base_client: An objectrocket.client.Client instance.
"""
def __init__(self, base_client):
self.__username = None
self.__password = None
self.__token = None
super(Auth, self).__init__(base_client=base_client)
#####################
# Public interface. #
#####################
def authenticate(self, username, password):
"""Authenticate against the ObjectRocket API.
:param str username: The username to perform basic authentication against the API with.
:param str password: The password to perform basic authentication against the API with.
:returns: A token used for authentication against token protected resources.
:rtype: str
"""
# Update the username and password bound to this instance for re-authentication needs.
self._username = username
self._password = password
# Attempt to authenticate.
resp = requests.get(
self._url,
auth=(username, password),
**self._default_request_kwargs
)
# Attempt to extract authentication data.
try:
if resp.status_code == 200:
json_data = resp.json()
token = json_data['data']['token']
elif resp.status_code == 401:
raise errors.AuthFailure(resp.json().get('message', 'Authentication Failure.'))
else:
raise errors.AuthFailure(
"Unknown exception while authenticating: '{}'".format(resp.text)
)
except errors.AuthFailure:
raise
except Exception as ex:
logging.exception(ex)
raise errors.AuthFailure('{}: {}'.format(ex.__class__.__name__, ex))
# Update the token bound to this instance for use by other client operations layers.
self._token = token
logger.info('New API token received: "{}".'.format(token))
return token
######################
# Private interface. #
######################
@property
def _default_request_kwargs(self):
"""The default request keyword arguments to be passed to the requests library."""
return super(Auth, self)._default_request_kwargs
@property
def _password(self):
"""The password currently being used for authentication."""
return self.__password
@_password.setter
def _password(self, new_password):
"""Update the password to be used for authentication."""
self.__password = new_password
def _refresh(self):
"""Refresh the API token using the currently bound credentials.
This is simply a convenience method to be invoked automatically if authentication fails
during normal client use.
"""
# Request and set a new API token.
new_token = self.authenticate(self._username, self._password)
self._token = new_token
logger.info('New API token received: "{}".'.format(new_token))
return self._token
@property
def _token(self):
"""The API token this instance is currently using."""
return self.__token
@_token.setter
def _token(self, new_token):
"""Update the API token which this instance is to use."""
self.__token = new_token
return self.__token
@property
def _url(self):
"""The base URL for authentication operations."""
return self._client._url + 'tokens/'
@property
def _username(self):
"""The username currently being used for authentication."""
return self.__username
@_username.setter
def _username(self, new_username):
"""Update the username to be used for authentication."""
self.__username = new_username
def _verify(self, token):
"""Verify that the given token is valid.
:param str token: The API token to verify.
:returns: The token's corresponding user model as a dict, or None if invalid.
:rtype: dict
"""
# Attempt to authenticate.
url = '{}{}/'.format(self._url, 'verify')
resp = requests.post(
url,
json={'token': token},
**self._default_request_kwargs
)
if resp.status_code == 200:
return resp.json().get('data', None)
return None