1
# -*- coding: utf-8 -*-
3
# Author: Alejandro J. Cura <alecu@canonical.com>
5
# Copyright 2010 Canonical Ltd.
7
# This program is free software: you can redistribute it and/or modify it
8
# under the terms of the GNU General Public License version 3, as published
9
# by the Free Software Foundation.
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranties of
13
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
14
# PURPOSE. See the GNU General Public License for more details.
16
# You should have received a copy of the GNU General Public License along
17
# with this program. If not, see <http://www.gnu.org/licenses/>.
18
"""Tests for txkeyring."""
24
from ubuntuone.devtools.testcase import DBusTestCase
25
from twisted.internet.defer import inlineCallbacks, returnValue
27
from ubuntu_sso.utils import txsecrets
29
# DBus exported objects have different naming conventions than vanilla python
30
# pylint: disable=C0103
32
# pylint, my clock says... 2010. Do you need a tutorial on decorators?
33
# pylint: disable=C0322
35
# pylint complains when things are a little too dynamic
36
# pylint: disable=E1101
39
class SampleMiscException(Exception):
40
"""An exception that will be turned into a DBus Exception."""
43
class ItemMock(dbus.service.Object):
44
"""An item contains a secret, lookup attributes and has a label."""
45
get_secret_fail = False
50
def __init__(self, collection, label, attributes, value, *args, **kwargs):
51
"""Initialize this instance."""
52
super(ItemMock, self).__init__(*args, **kwargs)
53
self.collection = collection
55
self.attributes = attributes
58
@dbus.service.method(dbus_interface=txsecrets.ITEM_IFACE,
61
"""Delete this item."""
63
raise SampleMiscException()
64
self.collection.items.remove(self)
65
if self.delete_prompt:
66
prompt_path = create_object_path("/org/freedesktop/secrets/prompt")
67
prompt = self.dbus_publish(prompt_path, PromptMock,
69
dismissed=self.dismissed)
74
@dbus.service.method(dbus_interface=txsecrets.ITEM_IFACE,
75
in_signature="o", out_signature="(oayay)")
76
def GetSecret(self, session):
77
"""Retrieve the secret for this item."""
78
if self.get_secret_fail:
79
raise SampleMiscException()
80
return (session, "", self.value)
83
class PromptMock(dbus.service.Object):
84
"""A prompt necessary to complete an operation."""
86
def __init__(self, dismissed=True,
87
result=dbus.String("", variant_level=1), *args, **kwargs):
88
"""Initialize this instance."""
89
super(PromptMock, self).__init__(*args, **kwargs)
90
self.dismissed = dismissed
93
@dbus.service.method(dbus_interface=txsecrets.PROMPT_IFACE,
95
def Prompt(self, window_id):
96
"""Perform the prompt."""
97
self.Completed(self.dismissed, self.result)
99
@dbus.service.signal(dbus_interface=txsecrets.PROMPT_IFACE,
101
def Completed(self, dismissed, result):
102
"""The prompt and operation completed."""
105
class CollectionMock(dbus.service.Object):
106
"""A collection of items containing secrets."""
107
SUPPORTS_MULTIPLE_OBJECT_PATHS = True
108
SUPPORTS_MULTIPLE_CONNECTIONS = True
109
create_item_prompt = False
111
create_item_fail = False
114
def __init__(self, label, *args, **kwargs):
115
"""Initialize this instance."""
116
super(CollectionMock, self).__init__(*args, **kwargs)
120
@dbus.service.method(dbus_interface=txsecrets.COLLECTION_IFACE,
121
in_signature="a{sv}(oayay)b", out_signature="oo",
123
def CreateItem(self, properties, secret, replace):
124
"""Create an item with the given attributes, secret and label.
126
If replace is set, then it replaces an item already present with the
127
same values for the attributes.
129
if self.create_item_fail:
130
raise SampleMiscException()
131
attributes = properties[txsecrets.ATTRIBUTES_PROPERTY]
132
item_label = properties[txsecrets.LABEL_PROPERTY]
133
session, parameters, value = secret
134
item_path = create_object_path("/org/freedesktop/secrets/collection/" +
136
item = self.dbus_publish(item_path, ItemMock, self, item_label,
138
self.items.append(item)
139
if self.create_item_prompt:
140
prompt_path = create_object_path("/org/freedesktop/secrets/prompt")
141
prompt = self.dbus_publish(prompt_path, PromptMock,
143
dismissed=self.dismissed)
149
class SessionMock(dbus.service.Object):
150
"""A session tracks state between the service and a client application."""
152
@dbus.service.method(dbus_interface=txsecrets.SESSION_IFACE)
154
"""Close this session."""
157
class SecretServiceMock(dbus.service.Object):
158
"""The Secret Service manages all the sessions and collections."""
159
create_collection_prompt = False
160
create_collection_fail = False
161
open_session_fail = False
162
unlock_prompts = False
165
def __init__(self, *args, **kwargs):
166
"""Initialize this instance."""
167
super(SecretServiceMock, self).__init__(*args, **kwargs)
169
self.collections = {}
171
@dbus.service.method(dbus_interface=txsecrets.SERVICE_IFACE,
172
in_signature="sv", out_signature="vo")
173
def OpenSession(self, algorithm, algorithm_parameters):
174
"""Open a unique session for the caller application."""
175
if self.open_session_fail:
176
raise SampleMiscException()
177
session_path = create_object_path("/org/freedesktop/secrets/session")
178
session = self.dbus_publish(session_path, SessionMock)
179
self.sessions[session_path] = session
182
@dbus.service.method(dbus_interface=txsecrets.SERVICE_IFACE,
183
in_signature="a{sv}", out_signature="oo")
184
def CreateCollection(self, properties):
185
"""Create a new collection with the specified properties."""
186
if self.create_collection_fail:
187
raise SampleMiscException()
188
label = str(properties[txsecrets.LABEL_PROPERTY])
189
if len(self.collections):
190
coll_path = "/org/freedesktop/secrets/collection/" + label
192
coll_path = txsecrets.DEFAULT_COLLECTION
193
collection = self.dbus_publish(coll_path, CollectionMock, label)
194
self.collections[label] = collection
196
if self.create_collection_prompt:
197
prompt_path = create_object_path("/org/freedesktop/secrets/prompt")
198
prompt = self.dbus_publish(prompt_path, PromptMock,
200
dismissed=self.dismissed)
203
return collection, "/"
205
@dbus.service.method(dbus_interface=txsecrets.SERVICE_IFACE,
206
in_signature="a{ss}", out_signature="aoao")
207
def SearchItems(self, attributes):
208
"""Find items in any collection."""
211
for c in self.collections.values():
213
append_item = locked_items.append
215
append_item = unlocked_items.append
217
# TODO: check if attrs match (not needed for now)
220
return unlocked_items, locked_items
222
def unlock_objects(self, objects):
223
"""Unlock the objects or its containers."""
224
for c in self.collections.values():
225
if c.__dbus_object_path__ in objects:
228
if i.__dbus_object_path__ in objects:
231
@dbus.service.method(dbus_interface=txsecrets.SERVICE_IFACE,
232
in_signature="ao", out_signature="aoo")
233
def Unlock(self, objects):
234
"""Unlock the specified objects."""
235
if self.unlock_prompts:
236
prompt_path = create_object_path("/org/freedesktop/secrets/prompt")
237
self.unlock_objects(objects)
238
prompt = self.dbus_publish(prompt_path, PromptMock,
240
dismissed=self.dismissed)
243
self.unlock_objects(objects)
247
def create_object_path(base):
248
"""Create a random object path given a base path."""
249
random = uuid.uuid4().hex
250
return base + "/" + random
253
class BaseTestCase(DBusTestCase):
254
"""Base class for DBus tests."""
258
super(BaseTestCase, self).setUp()
259
self.session_bus = dbus.SessionBus()
260
self.mock_service = self.dbus_publish(txsecrets.SECRETS_SERVICE,
262
self.secretservice = txsecrets.SecretService()
264
def dbus_publish(self, object_path, object_class, *args, **kwargs):
265
"""Create an object and publish it on the bus."""
266
name = self.session_bus.request_name(txsecrets.BUS_NAME,
267
dbus.bus.NAME_FLAG_DO_NOT_QUEUE)
268
self.assertNotEqual(name, dbus.bus.REQUEST_NAME_REPLY_EXISTS)
269
mock_object = object_class(*args, object_path=object_path,
270
conn=self.session_bus, **kwargs)
271
self.addCleanup(mock_object.remove_from_connection)
272
mock_object.dbus_publish = self.dbus_publish
276
def create_sample_collection(self, label):
277
"""Create a collection with a given label."""
278
coll = yield self.secretservice.create_collection(label)
282
class SecretServiceTestCase(BaseTestCase):
283
"""Test the Secret Service class."""
286
def test_open_session(self):
287
"""The secret service session is opened."""
288
result = yield self.secretservice.open_session()
289
self.assertEqual(result, self.secretservice)
292
def test_open_session_throws_dbus_error_as_failure(self):
293
"""The secret service open session throws a dbus error as a failure."""
294
d = self.secretservice.open_session()
295
self.mock_service.open_session_fail = True
296
yield self.assertFailure(d, dbus.exceptions.DBusException)
299
def test_open_session_fails_before_opening_as_failure(self):
300
"""A dbus error before opening the session is thrown as a failure."""
302
def fail(*args, **kwargs):
303
"""Throw a DBus exception."""
304
raise dbus.exceptions.DBusException()
306
self.patch(txsecrets.dbus, "SessionBus", fail)
307
d = self.secretservice.open_session()
308
self.mock_service.open_session_fail = True
309
yield self.assertFailure(d, dbus.exceptions.DBusException)
312
def test_create_collection(self):
313
"""The secret service creates a collection."""
314
yield self.secretservice.open_session()
315
collection_label = "sample_keyring"
316
yield self.create_sample_collection(collection_label)
317
self.assertIn(collection_label, self.mock_service.collections)
320
def test_create_collection_prompt(self):
321
"""The secret service creates a collection after a prompt."""
322
yield self.secretservice.open_session()
323
self.mock_service.create_collection_prompt = True
324
collection_label = "sample_keyring"
325
yield self.create_sample_collection(collection_label)
326
self.assertIn(collection_label, self.mock_service.collections)
329
def test_create_collection_prompt_dismissed(self):
330
"""The service fails to create collection when prompt dismissed."""
331
yield self.secretservice.open_session()
332
self.mock_service.create_collection_prompt = True
333
self.mock_service.dismissed = True
334
collection_label = "sample_keyring"
335
yield self.assertFailure(
336
self.create_sample_collection(collection_label),
337
txsecrets.UserCancelled)
340
def test_create_collection_throws_dbus_error(self):
341
"""The service fails to create collection on a DBus error."""
342
yield self.secretservice.open_session()
343
self.mock_service.create_collection_fail = True
344
collection_label = "sample_keyring"
345
yield self.assertFailure(
346
self.create_sample_collection(collection_label),
347
dbus.exceptions.DBusException)
350
def test_prompt_accepted(self):
351
"""A prompt is accepted."""
352
yield self.secretservice.open_session()
353
expected_result = "hello world"
354
prompt_path = "/prompt"
355
self.dbus_publish(prompt_path, PromptMock, result=expected_result,
357
result = yield self.secretservice.do_prompt(prompt_path)
358
self.assertEqual(result, expected_result)
361
def test_prompt_dismissed(self):
362
"""A prompt is dismissed with a UserCancelled failure."""
363
yield self.secretservice.open_session()
364
expected_result = "hello world2"
365
prompt_path = "/prompt"
366
self.dbus_publish(prompt_path, PromptMock, result=expected_result,
368
d = self.secretservice.do_prompt(prompt_path)
369
self.assertFailure(d, txsecrets.UserCancelled)
372
def test_search_unlocked_items(self):
373
"""The secret service searchs for unlocked items."""
374
yield self.secretservice.open_session()
375
coll = yield self.create_sample_collection("sample_keyring")
376
attr = {"key-type": "Ubuntu SSO credentials"}
377
sample_secret = "secret83!"
378
yield coll.create_item("Cucaracha", attr, sample_secret)
379
items = yield self.secretservice.search_items(attr)
380
self.assertEqual(len(items), 1)
381
value = yield items[0].get_value()
382
self.assertEqual(value, sample_secret)
385
def test_search_locked_items(self):
386
"""The secret service searchs for locked items."""
387
yield self.secretservice.open_session()
388
collection_name = "sample_keyring"
389
coll = yield self.create_sample_collection(collection_name)
390
mock_collection = self.mock_service.collections[collection_name]
391
mock_collection.locked = True
392
attr = {"key-type": "Ubuntu SSO credentials"}
393
sample_secret = "secret99!"
394
yield coll.create_item("Cucaracha", attr, sample_secret)
395
items = yield self.secretservice.search_items(attr)
396
self.assertEqual(len(items), 1)
397
value = yield items[0].get_value()
398
self.assertEqual(value, sample_secret)
401
def test_search_locked_items_prompts(self):
402
"""The secret service searchs for locked items after a prompt."""
403
yield self.secretservice.open_session()
404
collection_name = "sample_keyring"
405
coll = yield self.create_sample_collection(collection_name)
406
mock_collection = self.mock_service.collections[collection_name]
407
mock_collection.locked = True
408
self.mock_service.unlock_prompts = True
409
attr = {"key-type": "Ubuntu SSO credentials"}
410
sample_secret = "secret99!"
411
yield coll.create_item("Cucaracha", attr, sample_secret)
412
items = yield self.secretservice.search_items(attr)
413
self.assertEqual(len(items), 1)
414
value = yield items[0].get_value()
415
self.assertEqual(value, sample_secret)
418
def test_search_locked_items_prompts_dismissed(self):
419
"""Service fails search for locked items after dismissed prompt."""
420
yield self.secretservice.open_session()
421
collection_name = "sample_keyring"
422
coll = yield self.create_sample_collection(collection_name)
423
mock_collection = self.mock_service.collections[collection_name]
424
mock_collection.locked = True
425
self.mock_service.unlock_prompts = True
426
self.mock_service.dismissed = True
427
attr = {"key-type": "Ubuntu SSO credentials"}
428
sample_secret = "secret99!"
429
yield coll.create_item("Cucaracha", attr, sample_secret)
430
d = self.secretservice.search_items(attr)
431
yield self.assertFailure(d, txsecrets.UserCancelled)
434
class CollectionTestCase(BaseTestCase):
435
"""Test the Collection class."""
438
def test_create_item(self):
439
"""The collection creates an item."""
440
yield self.secretservice.open_session()
441
collection_label = "sample_keyring"
442
yield self.create_sample_collection(collection_label)
443
coll = self.secretservice.get_default_collection()
444
mock_collection = self.mock_service.collections[collection_label]
445
attr = {"key-type": "Ubuntu 242 credentials"}
446
sample_secret = "secret!"
447
yield coll.create_item("Cucaracha", attr, sample_secret)
448
self.assertEqual(len(mock_collection.items), 1)
449
self.assertEqual(mock_collection.items[0].value, sample_secret)
452
def test_create_item_prompt(self):
453
"""The collection creates an item after a prompt."""
454
yield self.secretservice.open_session()
455
collection_label = "sample_keyring"
456
yield self.create_sample_collection(collection_label)
457
coll = self.secretservice.get_default_collection()
458
mock_collection = self.mock_service.collections[collection_label]
459
mock_collection.create_item_prompt = True
460
attr = {"key-type": "Ubuntu 242 credentials"}
461
sample_secret = "secret2!"
462
yield coll.create_item("Cucaracha", attr, sample_secret)
463
self.assertEqual(len(mock_collection.items), 1)
464
self.assertEqual(mock_collection.items[0].value, sample_secret)
467
def test_create_item_prompt_dismissed(self):
468
"""The collection fails to create an item when prompt is dismissed."""
469
yield self.secretservice.open_session()
470
collection_label = "sample_keyring"
471
yield self.create_sample_collection(collection_label)
472
coll = self.secretservice.get_default_collection()
473
mock_collection = self.mock_service.collections[collection_label]
474
mock_collection.create_item_prompt = True
475
mock_collection.dismissed = True
476
attr = {"key-type": "Ubuntu 242 credentials"}
477
sample_secret = "secret3!"
478
yield self.assertFailure(coll.create_item("Cuca", attr, sample_secret),
479
txsecrets.UserCancelled)
482
def test_create_item_throws_dbus_error(self):
483
"""The collection fails to create an item when DBus fails."""
484
yield self.secretservice.open_session()
485
collection_label = "sample_keyring"
486
yield self.create_sample_collection(collection_label)
487
coll = self.secretservice.get_default_collection()
488
mock_collection = self.mock_service.collections[collection_label]
489
mock_collection.create_item_fail = True
490
attr = {"key-type": "Ubuntu 242 credentials"}
491
sample_secret = "secret4!"
492
yield self.assertFailure(coll.create_item("Cuca", attr, sample_secret),
493
dbus.exceptions.DBusException)
496
class ItemTestCase(BaseTestCase):
497
"""Test the Item class."""
500
def test_get_value(self):
501
"""The secret value is retrieved from the item."""
502
yield self.secretservice.open_session()
503
coll = yield self.create_sample_collection("sample_keyring")
504
attr = {"key-type": "Ubuntu SSO credentials"}
505
sample_secret = "secret83!"
506
yield coll.create_item("Cucaracha", attr, sample_secret)
507
items = yield self.secretservice.search_items(attr)
508
self.assertEqual(len(items), 1)
509
value = yield items[0].get_value()
510
self.assertEqual(value, sample_secret)
513
def test_get_value_throws_dbus_error(self):
514
"""The secret value is not retrieved if DBus fails."""
515
yield self.secretservice.open_session()
516
collection_label = "sample_keyring"
517
coll = yield self.create_sample_collection(collection_label)
518
attr = {"key-type": "Ubuntu SSO credentials"}
519
sample_secret = "secret83!"
520
yield coll.create_item("Cucaracha", attr, sample_secret)
521
items = yield self.secretservice.search_items(attr)
522
self.assertEqual(len(items), 1)
523
mock = self.mock_service.collections[collection_label].items[0]
524
mock.get_secret_fail = True
525
yield self.assertFailure(items[0].get_value(),
526
dbus.exceptions.DBusException)
529
def test_delete(self):
530
"""The item is deleted."""
531
yield self.secretservice.open_session()
532
coll = yield self.create_sample_collection("sample_keyring")
533
attr = {"key-type": "Ubuntu SSO credentials"}
534
sample_secret = "secret83!"
535
yield coll.create_item("Cucaracha", attr, sample_secret)
536
items = yield self.secretservice.search_items(attr)
537
self.assertEqual(len(items), 1)
538
yield items[0].delete()
539
items = yield self.secretservice.search_items(attr)
540
self.assertEqual(len(items), 0)
543
def test_delete_prompt(self):
544
"""The item is deleted after a prompt."""
545
yield self.secretservice.open_session()
546
collection_label = "sample_keyring"
547
coll = yield self.create_sample_collection(collection_label)
548
attr = {"key-type": "Ubuntu SSO credentials"}
549
sample_secret = "secret83!"
550
yield coll.create_item("Cucaracha", attr, sample_secret)
551
items = yield self.secretservice.search_items(attr)
552
self.assertEqual(len(items), 1)
553
mock_item = self.mock_service.collections[collection_label].items[0]
554
mock_item.delete_prompt = True
555
yield items[0].delete()
556
items = yield self.secretservice.search_items(attr)
557
self.assertEqual(len(items), 0)
560
def test_delete_prompt_dismissed(self):
561
"""The item is not deleted after a dismissed prompt."""
562
yield self.secretservice.open_session()
563
collection_label = "sample_keyring"
564
coll = yield self.create_sample_collection(collection_label)
565
attr = {"key-type": "Ubuntu SSO credentials"}
566
sample_secret = "secret83!"
567
yield coll.create_item("Cucaracha", attr, sample_secret)
568
items = yield self.secretservice.search_items(attr)
569
self.assertEqual(len(items), 1)
570
mock_item = self.mock_service.collections[collection_label].items[0]
571
mock_item.delete_prompt = True
572
mock_item.dismissed = True
573
yield self.assertFailure(items[0].delete(), txsecrets.UserCancelled)
576
def test_delete_throws_dbus_error(self):
577
"""The item is not deleted when a DBus error happens."""
578
yield self.secretservice.open_session()
579
collection_label = "sample_keyring"
580
coll = yield self.create_sample_collection(collection_label)
581
attr = {"key-type": "Ubuntu SSO credentials"}
582
sample_secret = "secret83!"
583
yield coll.create_item("Cucaracha", attr, sample_secret)
584
items = yield self.secretservice.search_items(attr)
585
self.assertEqual(len(items), 1)
586
mock_item = self.mock_service.collections[collection_label].items[0]
587
mock_item.delete_fail = True
588
yield self.assertFailure(items[0].delete(),
589
dbus.exceptions.DBusException)