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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
|
# -*- coding: utf-8 -*-
# Author: Natalia Bidart <natalia.bidart@canonical.com>
# Author: Alejandro J. Cura <alecu@canonical.com>
#
# Copyright 2010 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>.
"""Credential management utilities.
'Credentials' provides the following fault-tolerant methods:
* find_credentials
* clear_credentials
* store_credentials
* register
* login
The first three return a Deferred that will be fired when the operation was
completed.
The second two use the 'success_cb', 'error_cb' and 'denial_cb' to signal the
caller when the credentials were retrieved successfully, when there was an
error or when the user denied the access to the application, respectively.
For details, please read the Credentials class documentation.
"""
import sys
import traceback
import urllib2
from functools import wraps
from oauth import oauth
from twisted.internet.defer import inlineCallbacks, returnValue
from ubuntu_sso import NO_OP
from ubuntu_sso.keyring import Keyring
from ubuntu_sso.logger import setup_logging
logger = setup_logging('ubuntu_sso.credentials')
APP_NAME_KEY = 'app_name'
TC_URL_KEY = 'tc_url'
HELP_TEXT_KEY = 'help_text'
WINDOW_ID_KEY = 'window_id'
PING_URL_KEY = 'ping_url'
UI_MODULE_KEY = 'ui_module'
UI_CLASS_KEY = 'ui_class'
SUCCESS_CB_KEY = 'success_cb'
ERROR_CB_KEY = 'error_cb'
DENIAL_CB_KEY = 'denial_cb'
ERROR_KEY = 'error_message'
ERROR_DETAIL_KEY = 'detailed_error'
def handle_exceptions(msg):
"""Handle exceptions using 'msg' as error message."""
def middle(f):
"""Decorate 'f' to catch all errors."""
@wraps(f)
def inner(self, *a, **kw):
"""Call 'f' within a try-except block.
If any exception occurs, self.error_cb is called and the exception
is logged.
"""
result = None
try:
result = f(self, *a, **kw)
except: # pylint: disable=W0702
logger.exception('%s (app_name: %s): %s.',
f.__name__, self.app_name, msg)
logger.error('%s (app_name: %s): Calling error_cb at %r.',
f.__name__, self.app_name, self.error_cb)
error_dict = {ERROR_KEY: msg,
ERROR_DETAIL_KEY: traceback.format_exc()}
self.error_cb(error_dict)
return result
return inner
return middle
def handle_failures(msg):
"""Handle failures using 'msg' as error message."""
def middle(f):
"""Decorate 'f' to catch all errors."""
@wraps(f)
@inlineCallbacks
def inner(self, *a, **kw):
"""Call 'f' within a try-except block.
If any exception occurs, self.error_cb is called and the exception
is logged.
"""
result = None
try:
result = yield f(self, *a, **kw)
except Exception: # pylint: disable=W0703
logger.exception('%s (app_name: %s): %s.',
f.__name__, self.app_name, msg)
logger.error('%s (app_name: %s): Calling error_cb at %r.',
f.__name__, self.app_name, self.error_cb)
error_dict = {ERROR_KEY: msg,
ERROR_DETAIL_KEY: traceback.format_exc()}
self.error_cb(error_dict)
returnValue(result)
return inner
return middle
class Credentials(object):
"""Credentials management gateway."""
def __init__(self, app_name, tc_url=None, help_text='',
window_id=0, ping_url=None,
ui_module='ubuntu_sso.gtk.gui', ui_class='UbuntuSSOClientGUI',
success_cb=NO_OP, error_cb=NO_OP, denial_cb=NO_OP):
"""Return a Credentials management object.
'app_name' is the application name to be displayed in the GUI.
'tc_url' is the URL pointing to Terms & Conditions. If None, no
TOS agreement will be displayed.
'help_text' is an explanatory text for the end-users, will be shown
below the headers.
'window_id' is the id of the window which will be set as a parent of
the GUI. If 0, no parent will be set.
'ping_url' is the url that will be pinged when a user registers/logins
successfully. The user email will be attached to 'ping_url'.
'success_cb' will be called when the credentials were retrieved
successfully. Two params will be passed: the app_name and the
credentials per se. The credentials is a dictionary of the form:
{'token': <value>,
'token_secret': <value>,
'consumer_key': <value>,
'consumer_secret': <value>,
'name': <the token name, matches "[app_name] @ [host name]">}
'error_cb' will be called when the credentials retrieval failed. Two
params will be passed: the app_name, and an error dict with 2 keys:
the error message (user friendly, not translatable so far), and
the detailed error (usually the traceback).
'denial_cb' will be called when the user denied the credentials to the
caller. A single param is passed: the app_name.
"""
self.app_name = app_name
self.help_text = help_text
self.window_id = window_id
self.ping_url = ping_url
self.tc_url = tc_url
self.ui_module = ui_module
self.ui_class = ui_class
self._success_cb = success_cb
self._error_cb = error_cb
self.denial_cb = denial_cb
self.gui = None # will hold the GUI instance
@handle_failures(msg='Problem while retrieving credentials')
@inlineCallbacks
def _login_success_cb(self, app_name, email):
"""Store credentials when the login/registration succeeded.
Also, open self.ping_url/email to notify about this new token. If any
error occur, self.error_cb is called. Otherwise, self.success_cb is
called.
Return 0 on success, and a non-zero value (or None) on error.
"""
logger.info('Login/registration was successful for app %r, email %r',
app_name, email)
creds = yield self.find_credentials()
if creds is not None:
assert len(creds) > 0, 'Creds are empty! This should not happen'
# ping a server with the credentials if we were requested to
if self.ping_url is not None:
status = yield self._ping_url(app_name, email, creds)
if status is None:
yield self.clear_credentials()
return
self.success_cb(creds)
returnValue(0)
def _auth_denial_cb(self, app_name):
"""The user decided not to allow the registration or login."""
logger.warning('Login/registration was denied to app %r', app_name)
self.denial_cb(app_name)
@handle_failures(msg='Problem opening the ping_url')
@inlineCallbacks
def _ping_url(self, app_name, email, credentials):
"""Ping the self.ping_url with the email attached.
Sign the request with the user credentials. The self.ping_url must be
defined if this method is being called.
"""
logger.info('Pinging server for app_name "%s", ping_url: "%s", '
'email "%s".', app_name, self.ping_url, email)
url = self.ping_url + email
consumer = oauth.OAuthConsumer(credentials['consumer_key'],
credentials['consumer_secret'])
token = oauth.OAuthToken(credentials['token'],
credentials['token_secret'])
get_request = oauth.OAuthRequest.from_consumer_and_token
oauth_req = get_request(oauth_consumer=consumer, token=token,
http_method='GET', http_url=url)
oauth_req.sign_request(oauth.OAuthSignatureMethod_HMAC_SHA1(),
consumer, token)
request = urllib2.Request(url, headers=oauth_req.to_header())
logger.debug('Opening the url "%s" with urllib2.urlopen.',
request.get_full_url())
# This code is blocking, we should change this.
# I've tried with deferToThread an twisted.web.client.getPage
# but the returned deferred will never be fired (nataliabidart).
response = urllib2.urlopen(request)
logger.debug('Url opened. Response: %s.', response.code)
returnValue(response.code)
@handle_exceptions(msg='Problem opening the Ubuntu SSO user interface')
def _show_ui(self, login_only):
"""Shows the UI, connect outcome signals."""
__import__(self.ui_module)
gui = sys.modules[self.ui_module]
self.gui = getattr(gui, self.ui_class)(app_name=self.app_name,
tc_url=self.tc_url, help_text=self.help_text,
window_id=self.window_id, login_only=login_only)
self.gui.login_success_callback = self._login_success_cb
self.gui.registration_success_callback = self._login_success_cb
self.gui.user_cancellation_callback = self._auth_denial_cb
@handle_failures(msg='Problem while retrieving credentials')
@inlineCallbacks
def _login_or_register(self, login_only):
"""Get credentials if found else prompt the GUI."""
token = yield self.find_credentials()
if token is not None and len(token) > 0:
self.success_cb(token)
elif token == {}:
self._show_ui(login_only)
else:
# something went wrong with find_credentials, already handled.
logger.info('_login_or_register: call to "find_credentials" went '
'wrong, and was already handled.')
def error_cb(self, error_dict):
"""Handle error and notify the caller."""
logger.error('Calling error callback at %r (error is %r).',
self._error_cb, error_dict)
self._error_cb(self.app_name, error_dict)
def success_cb(self, creds):
"""Handle success and notify the caller."""
logger.debug('Calling success callback at %r.', self._success_cb)
self._success_cb(self.app_name, creds)
@inlineCallbacks
def find_credentials(self):
"""Get the credentials for 'self.app_name'. Return {} if not there."""
creds = yield Keyring().get_credentials(self.app_name)
logger.info('find_credentials: self.app_name "%s", '
'result is {}? %s', self.app_name, creds is None)
returnValue(creds if creds is not None else {})
@inlineCallbacks
def clear_credentials(self):
"""Clear the credentials for 'self.app_name'."""
yield Keyring().delete_credentials(self.app_name)
@inlineCallbacks
def store_credentials(self, token):
"""Store the credentials for 'self.app_name'."""
yield Keyring().set_credentials(self.app_name, token)
def register(self):
"""Get credentials if found else prompt the GUI to register."""
return self._login_or_register(login_only=False)
def login(self):
"""Get credentials if found else prompt the GUI to login."""
return self._login_or_register(login_only=True)
|