~mvo/ubuntu-sso-client/strawman-lp711413

# -*- coding: utf-8 -*-

# 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/>.
"""Tests for txkeyring."""

import uuid

import dbus.service

from ubuntuone.devtools.testcase import DBusTestCase
from twisted.internet.defer import inlineCallbacks, returnValue

from ubuntu_sso.utils import txsecrets

# DBus exported objects have different naming conventions than vanilla python
# pylint: disable=C0103

# pylint, my clock says... 2010. Do you need a tutorial on decorators?
# pylint: disable=C0322

# pylint complains when things are a little too dynamic
# pylint: disable=E1101


class SampleMiscException(Exception):
    """An exception that will be turned into a DBus Exception."""


class ItemMock(dbus.service.Object):
    """An item contains a secret, lookup attributes and has a label."""
    get_secret_fail = False
    delete_fail = False
    delete_prompt = False
    dismissed = False

    def __init__(self, collection, label, attributes, value, *args, **kwargs):
        """Initialize this instance."""
        super(ItemMock, self).__init__(*args, **kwargs)
        self.collection = collection
        self.label = label
        self.attributes = attributes
        self.value = value

    @dbus.service.method(dbus_interface=txsecrets.ITEM_IFACE,
                         out_signature="o")
    def Delete(self):
        """Delete this item."""
        if self.delete_fail:
            raise SampleMiscException()
        self.collection.items.remove(self)
        if self.delete_prompt:
            prompt_path = create_object_path("/org/freedesktop/secrets/prompt")
            prompt = self.dbus_publish(prompt_path, PromptMock,
                                       result="",
                                       dismissed=self.dismissed)
            return prompt
        else:
            return "/"

    @dbus.service.method(dbus_interface=txsecrets.ITEM_IFACE,
                         in_signature="o", out_signature="(oayay)")
    def GetSecret(self, session):
        """Retrieve the secret for this item."""
        if self.get_secret_fail:
            raise SampleMiscException()
        return (session, "", self.value)


class PromptMock(dbus.service.Object):
    """A prompt necessary to complete an operation."""

    def __init__(self, dismissed=True,
                 result=dbus.String("", variant_level=1), *args, **kwargs):
        """Initialize this instance."""
        super(PromptMock, self).__init__(*args, **kwargs)
        self.dismissed = dismissed
        self.result = result

    @dbus.service.method(dbus_interface=txsecrets.PROMPT_IFACE,
                         in_signature="s")
    def Prompt(self, window_id):
        """Perform the prompt."""
        self.Completed(self.dismissed, self.result)

    @dbus.service.signal(dbus_interface=txsecrets.PROMPT_IFACE,
                         signature="bv")
    def Completed(self, dismissed, result):
        """The prompt and operation completed."""


class CollectionMock(dbus.service.Object):
    """A collection of items containing secrets."""
    SUPPORTS_MULTIPLE_OBJECT_PATHS = True
    SUPPORTS_MULTIPLE_CONNECTIONS = True
    create_item_prompt = False
    dismissed = False
    create_item_fail = False
    locked = False

    def __init__(self, label, *args, **kwargs):
        """Initialize this instance."""
        super(CollectionMock, self).__init__(*args, **kwargs)
        self.items = []
        self.label = label

    @dbus.service.method(dbus_interface=txsecrets.COLLECTION_IFACE,
                         in_signature="a{sv}(oayay)b", out_signature="oo",
                         byte_arrays=True)
    def CreateItem(self, properties, secret, replace):
        """Create an item with the given attributes, secret and label.

        If replace is set, then it replaces an item already present with the
        same values for the attributes.
        """
        if self.create_item_fail:
            raise SampleMiscException()
        attributes = properties[txsecrets.ATTRIBUTES_PROPERTY]
        item_label = properties[txsecrets.LABEL_PROPERTY]
        session, parameters, value = secret
        item_path = create_object_path("/org/freedesktop/secrets/collection/" +
                                                                    self.label)
        item = self.dbus_publish(item_path, ItemMock, self, item_label,
                                 attributes, value)
        self.items.append(item)
        if self.create_item_prompt:
            prompt_path = create_object_path("/org/freedesktop/secrets/prompt")
            prompt = self.dbus_publish(prompt_path, PromptMock,
                                       result=item,
                                       dismissed=self.dismissed)
            return "/", prompt
        else:
            return item, "/"


class SessionMock(dbus.service.Object):
    """A session tracks state between the service and a client application."""

    @dbus.service.method(dbus_interface=txsecrets.SESSION_IFACE)
    def Close(self):
        """Close this session."""


class SecretServiceMock(dbus.service.Object):
    """The Secret Service manages all the sessions and collections."""
    create_collection_prompt = False
    create_collection_fail = False
    open_session_fail = False
    unlock_prompts = False
    dismissed = False

    def __init__(self, *args, **kwargs):
        """Initialize this instance."""
        super(SecretServiceMock, self).__init__(*args, **kwargs)
        self.sessions = {}
        self.collections = {}

    @dbus.service.method(dbus_interface=txsecrets.SERVICE_IFACE,
                         in_signature="sv", out_signature="vo")
    def OpenSession(self, algorithm, algorithm_parameters):
        """Open a unique session for the caller application."""
        if self.open_session_fail:
            raise SampleMiscException()
        session_path = create_object_path("/org/freedesktop/secrets/session")
        session = self.dbus_publish(session_path, SessionMock)
        self.sessions[session_path] = session
        return True, session

    @dbus.service.method(dbus_interface=txsecrets.SERVICE_IFACE,
                         in_signature="a{sv}", out_signature="oo")
    def CreateCollection(self, properties):
        """Create a new collection with the specified properties."""
        if self.create_collection_fail:
            raise SampleMiscException()
        label = str(properties[txsecrets.LABEL_PROPERTY])
        if len(self.collections):
            coll_path = "/org/freedesktop/secrets/collection/" + label
        else:
            coll_path = txsecrets.DEFAULT_COLLECTION
        collection = self.dbus_publish(coll_path, CollectionMock, label)
        self.collections[label] = collection

        if self.create_collection_prompt:
            prompt_path = create_object_path("/org/freedesktop/secrets/prompt")
            prompt = self.dbus_publish(prompt_path, PromptMock,
                                       result=collection,
                                       dismissed=self.dismissed)
            return "/", prompt
        else:
            return collection, "/"

    @dbus.service.method(dbus_interface=txsecrets.SERVICE_IFACE,
                         in_signature="a{ss}", out_signature="aoao")
    def SearchItems(self, attributes):
        """Find items in any collection."""
        unlocked_items = []
        locked_items = []
        for c in self.collections.values():
            if c.locked:
                append_item = locked_items.append
            else:
                append_item = unlocked_items.append
            for i in c.items:
                # TODO: check if attrs match (not needed for now)
                append_item(i)

        return unlocked_items, locked_items

    def unlock_objects(self, objects):
        """Unlock the objects or its containers."""
        for c in self.collections.values():
            if c.__dbus_object_path__ in objects:
                c.locked = False
            for i in c.items:
                if i.__dbus_object_path__ in objects:
                    c.locked = False

    @dbus.service.method(dbus_interface=txsecrets.SERVICE_IFACE,
                         in_signature="ao", out_signature="aoo")
    def Unlock(self, objects):
        """Unlock the specified objects."""
        if self.unlock_prompts:
            prompt_path = create_object_path("/org/freedesktop/secrets/prompt")
            self.unlock_objects(objects)
            prompt = self.dbus_publish(prompt_path, PromptMock,
                                       result=objects,
                                       dismissed=self.dismissed)
            return "/", prompt
        else:
            self.unlock_objects(objects)
            return objects, "/"


def create_object_path(base):
    """Create a random object path given a base path."""
    random = uuid.uuid4().hex
    return base + "/" + random


class BaseTestCase(DBusTestCase):
    """Base class for DBus tests."""
    timeout = 10

    def setUp(self):
        super(BaseTestCase, self).setUp()
        self.session_bus = dbus.SessionBus()
        self.mock_service = self.dbus_publish(txsecrets.SECRETS_SERVICE,
                                              SecretServiceMock)
        self.secretservice = txsecrets.SecretService()

    def dbus_publish(self, object_path, object_class, *args, **kwargs):
        """Create an object and publish it on the bus."""
        name = self.session_bus.request_name(txsecrets.BUS_NAME,
                                             dbus.bus.NAME_FLAG_DO_NOT_QUEUE)
        self.assertNotEqual(name, dbus.bus.REQUEST_NAME_REPLY_EXISTS)
        mock_object = object_class(*args, object_path=object_path,
                                   conn=self.session_bus, **kwargs)
        self.addCleanup(mock_object.remove_from_connection)
        mock_object.dbus_publish = self.dbus_publish
        return mock_object

    @inlineCallbacks
    def create_sample_collection(self, label):
        """Create a collection with a given label."""
        coll = yield self.secretservice.create_collection(label)
        returnValue(coll)


class SecretServiceTestCase(BaseTestCase):
    """Test the Secret Service class."""

    @inlineCallbacks
    def test_open_session(self):
        """The secret service session is opened."""
        result = yield self.secretservice.open_session()
        self.assertEqual(result, self.secretservice)

    @inlineCallbacks
    def test_open_session_throws_dbus_error_as_failure(self):
        """The secret service open session throws a dbus error as a failure."""
        d = self.secretservice.open_session()
        self.mock_service.open_session_fail = True
        yield self.assertFailure(d, dbus.exceptions.DBusException)

    @inlineCallbacks
    def test_open_session_fails_before_opening_as_failure(self):
        """A dbus error before opening the session is thrown as a failure."""

        def fail(*args, **kwargs):
            """Throw a DBus exception."""
            raise dbus.exceptions.DBusException()

        self.patch(txsecrets.dbus, "SessionBus", fail)
        d = self.secretservice.open_session()
        self.mock_service.open_session_fail = True
        yield self.assertFailure(d, dbus.exceptions.DBusException)

    @inlineCallbacks
    def test_create_collection(self):
        """The secret service creates a collection."""
        yield self.secretservice.open_session()
        collection_label = "sample_keyring"
        yield self.create_sample_collection(collection_label)
        self.assertIn(collection_label, self.mock_service.collections)

    @inlineCallbacks
    def test_create_collection_prompt(self):
        """The secret service creates a collection after a prompt."""
        yield self.secretservice.open_session()
        self.mock_service.create_collection_prompt = True
        collection_label = "sample_keyring"
        yield self.create_sample_collection(collection_label)
        self.assertIn(collection_label, self.mock_service.collections)

    @inlineCallbacks
    def test_create_collection_prompt_dismissed(self):
        """The service fails to create collection when prompt dismissed."""
        yield self.secretservice.open_session()
        self.mock_service.create_collection_prompt = True
        self.mock_service.dismissed = True
        collection_label = "sample_keyring"
        yield self.assertFailure(
                           self.create_sample_collection(collection_label),
                           txsecrets.UserCancelled)

    @inlineCallbacks
    def test_create_collection_throws_dbus_error(self):
        """The service fails to create collection on a DBus error."""
        yield self.secretservice.open_session()
        self.mock_service.create_collection_fail = True
        collection_label = "sample_keyring"
        yield self.assertFailure(
                           self.create_sample_collection(collection_label),
                           dbus.exceptions.DBusException)

    @inlineCallbacks
    def test_prompt_accepted(self):
        """A prompt is accepted."""
        yield self.secretservice.open_session()
        expected_result = "hello world"
        prompt_path = "/prompt"
        self.dbus_publish(prompt_path, PromptMock, result=expected_result,
                          dismissed=False)
        result = yield self.secretservice.do_prompt(prompt_path)
        self.assertEqual(result, expected_result)

    @inlineCallbacks
    def test_prompt_dismissed(self):
        """A prompt is dismissed with a UserCancelled failure."""
        yield self.secretservice.open_session()
        expected_result = "hello world2"
        prompt_path = "/prompt"
        self.dbus_publish(prompt_path, PromptMock, result=expected_result,
                          dismissed=True)
        d = self.secretservice.do_prompt(prompt_path)
        self.assertFailure(d, txsecrets.UserCancelled)

    @inlineCallbacks
    def test_search_unlocked_items(self):
        """The secret service searchs for unlocked items."""
        yield self.secretservice.open_session()
        coll = yield self.create_sample_collection("sample_keyring")
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret83!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 1)
        value = yield items[0].get_value()
        self.assertEqual(value, sample_secret)

    @inlineCallbacks
    def test_search_locked_items(self):
        """The secret service searchs for locked items."""
        yield self.secretservice.open_session()
        collection_name = "sample_keyring"
        coll = yield self.create_sample_collection(collection_name)
        mock_collection = self.mock_service.collections[collection_name]
        mock_collection.locked = True
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret99!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 1)
        value = yield items[0].get_value()
        self.assertEqual(value, sample_secret)

    @inlineCallbacks
    def test_search_locked_items_prompts(self):
        """The secret service searchs for locked items after a prompt."""
        yield self.secretservice.open_session()
        collection_name = "sample_keyring"
        coll = yield self.create_sample_collection(collection_name)
        mock_collection = self.mock_service.collections[collection_name]
        mock_collection.locked = True
        self.mock_service.unlock_prompts = True
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret99!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 1)
        value = yield items[0].get_value()
        self.assertEqual(value, sample_secret)

    @inlineCallbacks
    def test_search_locked_items_prompts_dismissed(self):
        """Service fails search for locked items after dismissed prompt."""
        yield self.secretservice.open_session()
        collection_name = "sample_keyring"
        coll = yield self.create_sample_collection(collection_name)
        mock_collection = self.mock_service.collections[collection_name]
        mock_collection.locked = True
        self.mock_service.unlock_prompts = True
        self.mock_service.dismissed = True
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret99!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        d = self.secretservice.search_items(attr)
        yield self.assertFailure(d, txsecrets.UserCancelled)


class CollectionTestCase(BaseTestCase):
    """Test the Collection class."""

    @inlineCallbacks
    def test_create_item(self):
        """The collection creates an item."""
        yield self.secretservice.open_session()
        collection_label = "sample_keyring"
        yield self.create_sample_collection(collection_label)
        coll = self.secretservice.get_default_collection()
        mock_collection = self.mock_service.collections[collection_label]
        attr = {"key-type": "Ubuntu 242 credentials"}
        sample_secret = "secret!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        self.assertEqual(len(mock_collection.items), 1)
        self.assertEqual(mock_collection.items[0].value, sample_secret)

    @inlineCallbacks
    def test_create_item_prompt(self):
        """The collection creates an item after a prompt."""
        yield self.secretservice.open_session()
        collection_label = "sample_keyring"
        yield self.create_sample_collection(collection_label)
        coll = self.secretservice.get_default_collection()
        mock_collection = self.mock_service.collections[collection_label]
        mock_collection.create_item_prompt = True
        attr = {"key-type": "Ubuntu 242 credentials"}
        sample_secret = "secret2!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        self.assertEqual(len(mock_collection.items), 1)
        self.assertEqual(mock_collection.items[0].value, sample_secret)

    @inlineCallbacks
    def test_create_item_prompt_dismissed(self):
        """The collection fails to create an item when prompt is dismissed."""
        yield self.secretservice.open_session()
        collection_label = "sample_keyring"
        yield self.create_sample_collection(collection_label)
        coll = self.secretservice.get_default_collection()
        mock_collection = self.mock_service.collections[collection_label]
        mock_collection.create_item_prompt = True
        mock_collection.dismissed = True
        attr = {"key-type": "Ubuntu 242 credentials"}
        sample_secret = "secret3!"
        yield self.assertFailure(coll.create_item("Cuca", attr, sample_secret),
                                 txsecrets.UserCancelled)

    @inlineCallbacks
    def test_create_item_throws_dbus_error(self):
        """The collection fails to create an item when DBus fails."""
        yield self.secretservice.open_session()
        collection_label = "sample_keyring"
        yield self.create_sample_collection(collection_label)
        coll = self.secretservice.get_default_collection()
        mock_collection = self.mock_service.collections[collection_label]
        mock_collection.create_item_fail = True
        attr = {"key-type": "Ubuntu 242 credentials"}
        sample_secret = "secret4!"
        yield self.assertFailure(coll.create_item("Cuca", attr, sample_secret),
                                 dbus.exceptions.DBusException)


class ItemTestCase(BaseTestCase):
    """Test the Item class."""

    @inlineCallbacks
    def test_get_value(self):
        """The secret value is retrieved from the item."""
        yield self.secretservice.open_session()
        coll = yield self.create_sample_collection("sample_keyring")
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret83!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 1)
        value = yield items[0].get_value()
        self.assertEqual(value, sample_secret)

    @inlineCallbacks
    def test_get_value_throws_dbus_error(self):
        """The secret value is not retrieved if DBus fails."""
        yield self.secretservice.open_session()
        collection_label = "sample_keyring"
        coll = yield self.create_sample_collection(collection_label)
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret83!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 1)
        mock = self.mock_service.collections[collection_label].items[0]
        mock.get_secret_fail = True
        yield self.assertFailure(items[0].get_value(),
                                 dbus.exceptions.DBusException)

    @inlineCallbacks
    def test_delete(self):
        """The item is deleted."""
        yield self.secretservice.open_session()
        coll = yield self.create_sample_collection("sample_keyring")
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret83!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 1)
        yield items[0].delete()
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 0)

    @inlineCallbacks
    def test_delete_prompt(self):
        """The item is deleted after a prompt."""
        yield self.secretservice.open_session()
        collection_label = "sample_keyring"
        coll = yield self.create_sample_collection(collection_label)
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret83!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 1)
        mock_item = self.mock_service.collections[collection_label].items[0]
        mock_item.delete_prompt = True
        yield items[0].delete()
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 0)

    @inlineCallbacks
    def test_delete_prompt_dismissed(self):
        """The item is not deleted after a dismissed prompt."""
        yield self.secretservice.open_session()
        collection_label = "sample_keyring"
        coll = yield self.create_sample_collection(collection_label)
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret83!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 1)
        mock_item = self.mock_service.collections[collection_label].items[0]
        mock_item.delete_prompt = True
        mock_item.dismissed = True
        yield self.assertFailure(items[0].delete(), txsecrets.UserCancelled)

    @inlineCallbacks
    def test_delete_throws_dbus_error(self):
        """The item is not deleted when a DBus error happens."""
        yield self.secretservice.open_session()
        collection_label = "sample_keyring"
        coll = yield self.create_sample_collection(collection_label)
        attr = {"key-type": "Ubuntu SSO credentials"}
        sample_secret = "secret83!"
        yield coll.create_item("Cucaracha", attr, sample_secret)
        items = yield self.secretservice.search_items(attr)
        self.assertEqual(len(items), 1)
        mock_item = self.mock_service.collections[collection_label].items[0]
        mock_item.delete_fail = True
        yield self.assertFailure(items[0].delete(),
                                 dbus.exceptions.DBusException)