/usr/lib/python2.7/dist-packages/secretstorage/collection.py is in python-secretstorage 2.0.0-1ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | # SecretStorage module for Python
# Access passwords using the SecretService DBus API
# Author: Dmitry Shachnev, 2013
# License: BSD
"""Collection is a place where secret items are stored. Normally, only
the default collection should be used, but this module allows to use any
registered collection. Use :func:`get_default_collection` to get the
default collection (and create it, if necessary).
Collections are usually automatically unlocked when user logs in, but
collections can also be locked and unlocked using
:meth:`Collection.lock` and :meth:`Collection.unlock` methods (unlocking
requires showing the unlocking prompt to user and can be synchronous or
asynchronous). Creating new items and editing existing ones is possible
only in unlocked collection."""
import dbus
from secretstorage.defines import SS_PREFIX, SS_PATH, SECRETS
from secretstorage.exceptions import LockedException, ItemNotFoundException
from secretstorage.item import Item
from secretstorage.util import bus_get_object, InterfaceWrapper, \
exec_prompt, exec_prompt_glib, format_secret, open_session, \
to_unicode
COLLECTION_IFACE = SS_PREFIX + 'Collection'
SERVICE_IFACE = SS_PREFIX + 'Service'
DEFAULT_COLLECTION = '/org/freedesktop/secrets/aliases/default'
SESSION_COLLECTION = '/org/freedesktop/secrets/collection/session'
class Collection(object):
"""Represents a collection."""
def __init__(self, bus, collection_path=DEFAULT_COLLECTION, session=None):
collection_obj = bus_get_object(bus, SECRETS, collection_path)
self.bus = bus
self.session = session
self.collection_path = collection_path
self.collection_iface = InterfaceWrapper(collection_obj,
COLLECTION_IFACE)
self.collection_props_iface = InterfaceWrapper(collection_obj,
dbus.PROPERTIES_IFACE)
self.collection_props_iface.Get(COLLECTION_IFACE, 'Label',
signature='ss')
def is_locked(self):
"""Returns :const:`True` if item is locked, otherwise
:const:`False`."""
return bool(self.collection_props_iface.Get(
COLLECTION_IFACE, 'Locked', signature='ss'))
def ensure_not_locked(self):
"""If collection is locked, raises
:exc:`~secretstorage.exceptions.LockedException`."""
if self.is_locked():
raise LockedException('Collection is locked!')
def unlock(self, callback=None):
"""Requests unlocking the collection. If `callback` is specified,
calls it when unlocking is complete (see
:func:`~secretstorage.util.exec_prompt` description for
details) and returns a boolean representing whether the operation was
dismissed. Otherwise, uses loop from GLib API."""
service_obj = bus_get_object(self.bus, SECRETS, SS_PATH)
service_iface = InterfaceWrapper(service_obj, SERVICE_IFACE)
prompt = service_iface.Unlock([self.collection_path], signature='ao')[1]
if len(prompt) > 1:
if callback:
exec_prompt(self.bus, prompt, callback)
else:
return exec_prompt_glib(self.bus, prompt)[0]
elif callback:
# We still need to call it.
callback(False, [])
def lock(self):
"""Locks the collection."""
service_obj = bus_get_object(self.bus, SECRETS, SS_PATH)
service_iface = InterfaceWrapper(service_obj, SERVICE_IFACE)
service_iface.Lock([self.collection_path], signature='ao')
def delete(self):
"""Deletes the collection and all items inside it."""
self.ensure_not_locked()
self.collection_iface.Delete(signature='')
def get_all_items(self):
"""Returns a generator of all items in the collection."""
for item_path in self.collection_props_iface.Get(
COLLECTION_IFACE, 'Items', signature='ss'):
yield Item(self.bus, item_path, self.session)
def search_items(self, attributes):
"""Returns a generator of items with the given attributes.
`attributes` should be a dictionary."""
result = self.collection_iface.SearchItems(attributes,
signature='a{ss}')
if isinstance(result, tuple):
# bug in GNOME Keyring <= 3.7.5
result = result[0] + result[1]
for item_path in result:
yield Item(self.bus, item_path, self.session)
def get_label(self):
"""Returns the collection label."""
label = self.collection_props_iface.Get(COLLECTION_IFACE,
'Label', signature='ss')
return to_unicode(label)
def set_label(self, label):
"""Sets collection label to `label`."""
self.ensure_not_locked()
self.collection_props_iface.Set(COLLECTION_IFACE, 'Label',
label, signature='ssv')
def create_item(self, label, attributes, secret, replace=False,
content_type='text/plain'):
"""Creates a new :class:`~secretstorage.item.Item` with given
`label` (unicode string), `attributes` (dictionary) and `secret`
(bytestring). If `replace` is :const:`True`, replaces the existing
item with the same attributes. If `content_type` is given, also
sets the content type of the secret (``text/plain`` by default).
Returns the created item."""
self.ensure_not_locked()
if not self.session:
self.session = open_session(self.bus)
secret = format_secret(self.session, secret, content_type)
attributes = dbus.Dictionary(attributes, signature='ss')
properties = {
SS_PREFIX+'Item.Label': label,
SS_PREFIX+'Item.Attributes': attributes
}
new_item, prompt = self.collection_iface.CreateItem(properties,
secret, replace, signature='a{sv}(oayays)b')
return Item(self.bus, new_item, self.session)
def create_collection(bus, label, alias='', session=None):
"""Creates a new :class:`Collection` with the given `label` and `alias`
and returns it. This action requires prompting. If prompt is dismissed,
raises :exc:`~secretstorage.exceptions.ItemNotFoundException`. This is
synchronous function, uses loop from GLib API."""
if not session:
session = open_session(bus)
properties = {SS_PREFIX+'Collection.Label': label}
service_obj = bus_get_object(bus, SECRETS, SS_PATH)
service_iface = dbus.Interface(service_obj, SERVICE_IFACE)
collection_path, prompt = service_iface.CreateCollection(properties,
alias, signature='a{sv}s')
if len(collection_path) > 1:
return Collection(bus, collection_path, session=session)
dismissed, unlocked = exec_prompt_glib(bus, prompt)
if dismissed:
raise ItemNotFoundException('Prompt dismissed.')
return Collection(bus, unlocked, session=session)
def get_all_collections(bus):
"""Returns a generator of all available collections."""
service_obj = bus_get_object(bus, SECRETS, SS_PATH)
service_props_iface = dbus.Interface(service_obj,
dbus.PROPERTIES_IFACE)
for collection_path in service_props_iface.Get(SERVICE_IFACE,
'Collections', signature='ss'):
yield Collection(bus, collection_path)
def get_default_collection(bus, session=None):
"""Returns the default collection. If it doesn't exist,
creates it."""
try:
return Collection(bus)
except ItemNotFoundException:
return create_collection(bus, 'Default',
'default', session)
def get_any_collection(bus):
"""Returns any collection, in the following order of preference:
- The default collection;
- The "session" collection (usually temporary);
- The first collection in the collections list."""
try:
return Collection(bus)
except ItemNotFoundException:
pass
try:
return Collection(bus, SESSION_COLLECTION)
except ItemNotFoundException:
pass
collections = list(get_all_collections(bus))
if collections:
return collections[0]
else:
raise ItemNotFoundException('No collections found.')
def get_collection_by_alias(bus, alias):
"""Returns the collection with the given `alias`. If there is no
such collection, raises
:exc:`~secretstorage.exceptions.ItemNotFoundException`."""
service_obj = bus_get_object(bus, SECRETS, SS_PATH)
service_iface = dbus.Interface(service_obj, SERVICE_IFACE)
collection_path = service_iface.ReadAlias(alias, signature='s')
if len(collection_path) <= 1:
raise ItemNotFoundException('No collection with such alias.')
return Collection(bus, collection_path)
def search_items(bus, attributes):
"""Returns a generator of items in all collections with the given
attributes. `attributes` should be a dictionary."""
service_obj = bus_get_object(bus, SECRETS, SS_PATH)
service_iface = dbus.Interface(service_obj, SERVICE_IFACE)
locked, unlocked = service_iface.SearchItems(attributes,
signature='a{ss}')
for item_path in locked + unlocked:
yield Item(bus, item_path)
|