libsecret/library/tests/mock/service.py

499 lines
17 KiB
Python
Raw Normal View History

#!/usr/bin/env python
#
# Copyright 2011 Stef Walter
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation; either version 2 of the licence or (at
# your option) any later version.
#
# See the included COPYING file for more information.
#
2012-01-11 06:44:32 +00:00
import getopt
import os
import sys
2012-01-11 06:44:32 +00:00
import time
import unittest
import aes
import dh
import hkdf
import dbus
import dbus.service
import dbus.glib
import gobject
COLLECTION_PREFIX = "/org/freedesktop/secrets/collection/"
bus_name = 'org.freedesktop.Secret.MockService'
ready_pipe = -1
objects = { }
class NotSupported(dbus.exceptions.DBusException):
def __init__(self, msg):
dbus.exceptions.DBusException.__init__(self, msg, name="org.freedesktop.DBus.Error.NotSupported")
class InvalidArgs(dbus.exceptions.DBusException):
def __init__(self, msg):
dbus.exceptions.DBusException.__init__(self, msg, name="org.freedesktop.DBus.Error.InvalidArgs")
class IsLocked(dbus.exceptions.DBusException):
def __init__(self, msg):
dbus.exceptions.DBusException.__init__(self, msg, name="org.freedesktop.Secret.Error.IsLocked")
unique_identifier = 0
def next_identifier():
global unique_identifier
unique_identifier += 1
return unique_identifier
def hex_encode(string):
return "".join([hex(ord(c))[2:].zfill(2) for c in string])
class PlainAlgorithm():
def negotiate(self, service, sender, param):
if type (param) != dbus.String:
raise InvalidArgs("invalid argument passed to OpenSession")
session = SecretSession(service, sender, self, None)
return (dbus.String("", variant_level=1), session)
def encrypt(self, key, data):
return ("", data)
class AesAlgorithm():
def negotiate(self, service, sender, param):
if type (param) != dbus.ByteArray:
raise InvalidArgs("invalid argument passed to OpenSession")
privat, publi = dh.generate_pair()
peer = dh.bytes_to_number(param)
# print "mock publi: ", hex(publi)
# print " mock peer: ", hex(peer)
ikm = dh.derive_key(privat, peer)
# print " mock ikm: ", hex_encode(ikm)
key = hkdf.hkdf(ikm, 16)
# print " mock key: ", hex_encode(key)
session = SecretSession(service, sender, self, key)
return (dbus.ByteArray(dh.number_to_bytes(publi), variant_level=1), session)
def encrypt(self, key, data):
key = map(ord, key)
data = aes.append_PKCS7_padding(data)
keysize = len(key)
iv = [ord(i) for i in os.urandom(16)]
mode = aes.AESModeOfOperation.modeOfOperation["CBC"]
moo = aes.AESModeOfOperation()
(mode, length, ciph) = moo.encrypt(data, mode, key, keysize, iv)
return ("".join([chr(i) for i in iv]),
"".join([chr(i) for i in ciph]))
2012-01-11 06:44:32 +00:00
class SecretPrompt(dbus.service.Object):
def __init__(self, service, sender, prompt_name=None, delay=0,
dismiss=False, result=dbus.String("", variant_level=1),
action=None):
self.sender = sender
self.service = service
self.delay = 0
self.dismiss = False
self.result = result
self.action = action
self.completed = False
if prompt_name:
self.path = "/org/freedesktop/secrets/prompts/%s" % prompt_name
else:
self.path = "/org/freedesktop/secrets/prompts/p%d" % next_identifier()
dbus.service.Object.__init__(self, service.bus_name, self.path)
service.add_prompt(self)
assert self.path not in objects
objects[self.path] = self
def _complete(self):
if self.completed:
return
self.completed = True
self.Completed(self.dismiss, self.result)
self.remove_from_connection()
@dbus.service.method('org.freedesktop.Secret.Prompt')
def Prompt(self, window_id):
if self.action:
self.action()
gobject.timeout_add(self.delay * 1000, self._complete)
@dbus.service.method('org.freedesktop.Secret.Prompt')
def Dismiss(self):
self._complete()
@dbus.service.signal(dbus_interface='org.freedesktop.Secret.Prompt', signature='bv')
def Completed(self, dismiss, result):
pass
class SecretSession(dbus.service.Object):
def __init__(self, service, sender, algorithm, key):
self.sender = sender
self.service = service
self.algorithm = algorithm
self.key = key
self.path = "/org/freedesktop/secrets/sessions/%d" % next_identifier()
dbus.service.Object.__init__(self, service.bus_name, self.path)
service.add_session(self)
objects[self.path] = self
def encode_secret(self, secret, content_type):
(params, data) = self.algorithm.encrypt(self.key, secret)
# print " mock iv: ", hex_encode(params)
# print " mock ciph: ", hex_encode(data)
return dbus.Struct((dbus.ObjectPath(self.path), dbus.ByteArray(params),
dbus.ByteArray(data), dbus.String(content_type)),
signature="oayays")
@dbus.service.method('org.freedesktop.Secret.Session')
def Close(self):
self.remove_from_connection()
self.service.remove_session(self)
class SecretItem(dbus.service.Object):
def __init__(self, collection, identifier, label="Item", attributes={ },
2012-01-11 06:44:32 +00:00
secret="", confirm=False, content_type="text/plain"):
self.collection = collection
self.identifier = identifier
self.label = label
self.secret = secret
self.attributes = attributes
self.content_type = content_type
self.path = "%s/%s" % (collection.path, identifier)
2012-01-11 06:44:32 +00:00
self.confirm = confirm
self.created = self.modified = time.time()
dbus.service.Object.__init__(self, collection.service.bus_name, self.path)
collection.items[identifier] = self
objects[self.path] = self
def match_attributes(self, attributes):
for (key, value) in attributes.items():
if not self.attributes.get(key) == value:
return False
return True
2012-01-26 13:34:37 +00:00
def get_locked(self):
return self.collection.locked
def perform_xlock(self, lock):
return self.collection.perform_xlock(lock)
2012-01-11 06:44:32 +00:00
def perform_delete(self):
del self.collection.items[self.identifier]
del objects[self.path]
self.remove_from_connection()
2012-01-11 06:44:32 +00:00
@dbus.service.method('org.freedesktop.Secret.Item', sender_keyword='sender')
def GetSecret(self, session_path, sender=None):
session = objects.get(session_path, None)
if not session or session.sender != sender:
raise InvalidArgs("session invalid: %s" % session_path)
2012-01-26 13:34:37 +00:00
if self.get_locked():
raise IsLocked("secret is locked: %s" % self.path)
return session.encode_secret(self.secret, self.content_type)
2012-01-11 06:44:32 +00:00
@dbus.service.method('org.freedesktop.Secret.Item', sender_keyword='sender')
def Delete(self, sender=None):
if self.confirm:
prompt = SecretPrompt(self.collection.service, sender,
dismiss=False, action=self.perform_delete)
return dbus.ObjectPath(prompt.path)
else:
self.perform_delete()
return dbus.ObjectPath("/")
@dbus.service.method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v')
def Get(self, interface_name, property_name):
return self.GetAll(interface_name)[property_name]
@dbus.service.method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}')
def GetAll(self, interface_name):
if interface_name == 'org.freedesktop.Secret.Item':
return {
2012-01-26 13:34:37 +00:00
'Locked': self.get_locked(),
'Attributes': dbus.Dictionary(self.attributes, signature='ss'),
'Label': self.label,
'Created': dbus.UInt64(self.created),
'Modified': dbus.UInt64(self.modified)
}
else:
raise InvalidArgs('Unknown %s interface' % interface_name)
@dbus.service.method(dbus.PROPERTIES_IFACE, in_signature='ssv')
def Set(self, interface_name, property_name, new_value):
if interface_name != 'org.freedesktop.Secret.Item':
raise InvalidArgs('Unknown %s interface' % interface_name)
if property_name == "Label":
self.label = str(new_value)
elif property_name == "Attributes":
self.attributes = dict(new_value)
else:
raise InvalidArgs('Unknown %s interface' % property_name)
self.PropertiesChanged(interface_name, { property_name: new_value }, [])
@dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as')
def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties):
self.modified = time.time()
class SecretCollection(dbus.service.Object):
def __init__(self, service, identifier, label="Collection", locked=False, confirm=False):
self.service = service
self.identifier = identifier
self.label = label
self.locked = locked
self.items = { }
self.confirm = confirm
self.created = self.modified = time.time()
self.path = "%s%s" % (COLLECTION_PREFIX, identifier)
dbus.service.Object.__init__(self, service.bus_name, self.path)
service.collections[identifier] = self
objects[self.path] = self
def search_items(self, attributes):
results = []
for item in self.items.values():
if item.match_attributes(attributes):
results.append(item)
return results
2012-01-26 13:34:37 +00:00
def get_locked(self):
return self.locked
def perform_xlock(self, lock):
self.locked = lock
for item in self.items.values():
self.PropertiesChanged('org.freedesktop.Secret.Item', { "Locked" : lock }, [])
self.PropertiesChanged('org.freedesktop.Secret.Collection', { "Locked" : lock }, [])
def perform_delete(self):
for item in self.items.values():
item.perform_delete()
del self.service.collections[self.identifier]
del objects[self.path]
self.remove_from_connection()
@dbus.service.method('org.freedesktop.Secret.Collection', sender_keyword='sender')
def Delete(self, sender=None):
if self.confirm:
prompt = SecretPrompt(self.collection.service, sender,
dismiss=False, action=self.perform_delete)
return dbus.ObjectPath(prompt.path)
else:
self.perform_delete()
return dbus.ObjectPath("/")
@dbus.service.method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v')
def Get(self, interface_name, property_name):
return self.GetAll(interface_name)[property_name]
@dbus.service.method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}')
def GetAll(self, interface_name):
if interface_name == 'org.freedesktop.Secret.Collection':
return {
2012-01-26 13:34:37 +00:00
'Locked': self.get_locked(),
'Label': self.label,
'Created': dbus.UInt64(self.created),
'Modified': dbus.UInt64(self.modified),
2012-01-26 13:34:37 +00:00
'Items': dbus.Array([dbus.ObjectPath(i.path) for i in self.items.values()], signature='o')
}
else:
raise InvalidArgs('Unknown %s interface' % interface_name)
@dbus.service.method(dbus.PROPERTIES_IFACE, in_signature='ssv')
def Set(self, interface_name, property_name, new_value):
if interface_name != 'org.freedesktop.Secret.Collection':
raise InvalidArgs('Unknown %s interface' % interface_name)
if property_name == "Label":
self.label = str(new_value)
else:
2012-01-26 13:34:37 +00:00
raise InvalidArgs('Not a writable property %s' % property_name)
self.PropertiesChanged(interface_name, { property_name: new_value }, [])
@dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as')
def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties):
self.modified = time.time()
class SecretService(dbus.service.Object):
algorithms = {
'plain': PlainAlgorithm(),
"dh-ietf1024-sha256-aes128-cbc-pkcs7": AesAlgorithm(),
}
def __init__(self, name=None):
if name == None:
name = bus_name
bus = dbus.SessionBus()
self.bus_name = dbus.service.BusName(name, allow_replacement=True, replace_existing=True)
dbus.service.Object.__init__(self, self.bus_name, '/org/freedesktop/secrets')
self.sessions = { }
2012-01-11 06:44:32 +00:00
self.prompts = { }
self.collections = { }
def on_name_owner_changed(owned, old_owner, new_owner):
if not new_owner:
for session in list(self.sessions.get(old_owner, [])):
session.Close()
bus.add_signal_receiver(on_name_owner_changed,
'NameOwnerChanged',
'org.freedesktop.DBus')
def add_standard_objects(self):
2012-01-26 13:34:37 +00:00
collection = SecretCollection(self, "english", label="Collection One", locked=False)
SecretItem(collection, "item_one", label="Item One", attributes={ "number": "1", "string": "one", "even": "false" }, secret="111")
SecretItem(collection, "item_two", attributes={ "number": "2", "string": "two", "even": "true" }, secret="222")
SecretItem(collection, "item_three", attributes={ "number": "3", "string": "three", "even": "false" }, secret="3333")
collection = SecretCollection(self, "spanish", locked=True)
SecretItem(collection, "item_one", attributes={ "number": "1", "string": "uno", "even": "false" }, secret="111")
SecretItem(collection, "item_two", attributes={ "number": "2", "string": "dos", "even": "true" }, secret="222")
SecretItem(collection, "item_three", attributes={ "number": "3", "string": "tres", "even": "false" }, secret="3333")
collection = SecretCollection(self, "empty", locked=False)
def listen(self):
global ready_pipe
loop = gobject.MainLoop()
if ready_pipe >= 0:
os.write(ready_pipe, "GO")
os.close(ready_pipe)
ready_pipe = -1
loop.run()
def add_session(self, session):
if session.sender not in self.sessions:
self.sessions[session.sender] = []
self.sessions[session.sender].append(session)
def remove_session(self, session):
self.sessions[session.sender].remove(session)
2012-01-11 06:44:32 +00:00
def add_prompt(self, prompt):
if prompt.sender not in self.prompts:
self.prompts[prompt.sender] = []
self.prompts[prompt.sender].append(prompt)
def remove_prompt (self, prompt):
self.prompts[prompt.sender].remove(prompt)
def find_item(self, object):
if object.startswith(COLLECTION_PREFIX):
parts = object[len(COLLECTION_PREFIX):].split("/", 1)
if len(parts) == 2 and parts[0] in self.collections:
return self.collections[parts[0]].get(parts[1], None)
return None
2012-01-26 13:34:37 +00:00
@dbus.service.method('org.freedesktop.Secret.Service', sender_keyword='sender')
def Lock(self, paths, lock=True, sender=None):
locked = []
prompts = []
for path in paths:
if path not in objects:
continue
object = objects[path]
if object.get_locked() == lock:
locked.append(path)
elif not object.confirm:
object.perform_xlock(lock)
locked.append(path)
else:
prompts.append(object)
def prompt_callback():
for object in prompts:
object.perform_xlock(lock)
locked = dbus.Array(locked, signature='o')
if prompts:
prompt = SecretPrompt(self, sender, dismiss=False, action=prompt_callback,
result=dbus.Array([o.path for o in prompts], signature='o'))
return (locked, dbus.ObjectPath(prompt.path))
else:
return (locked, dbus.ObjectPath("/"))
@dbus.service.method('org.freedesktop.Secret.Service', sender_keyword='sender')
def Unlock(self, paths, sender=None):
return self.Lock(paths, lock=False, sender=sender)
@dbus.service.method('org.freedesktop.Secret.Service', byte_arrays=True, sender_keyword='sender')
def OpenSession(self, algorithm, param, sender=None):
assert type(algorithm) == dbus.String
if algorithm not in self.algorithms:
raise NotSupported("algorithm %s is not supported" % algorithm)
return self.algorithms[algorithm].negotiate(self, sender, param)
@dbus.service.method('org.freedesktop.Secret.Service')
def SearchItems(self, attributes):
locked = [ ]
unlocked = [ ]
items = [ ]
for collection in self.collections.values():
items = collection.search_items(attributes)
2012-01-26 13:34:37 +00:00
if collection.get_locked():
locked.extend(items)
else:
unlocked.extend(items)
return (dbus.Array(unlocked, "o"), dbus.Array(locked, "o"))
@dbus.service.method('org.freedesktop.Secret.Service', sender_keyword='sender')
def GetSecrets(self, item_paths, session_path, sender=None):
session = objects.get(session_path, None)
if not session or session.sender != sender:
raise InvalidArgs("session invalid: %s" % session_path)
results = dbus.Dictionary(signature="o(oayays)")
for item_path in item_paths:
item = objects.get(item_path, None)
2012-01-26 13:34:37 +00:00
if item and not item.get_locked():
results[item_path] = item.GetSecret(session_path, sender)
return results
2012-01-26 13:34:37 +00:00
@dbus.service.method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v')
def Get(self, interface_name, property_name):
return self.GetAll(interface_name)[property_name]
@dbus.service.method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}')
def GetAll(self, interface_name):
if interface_name == 'org.freedesktop.Secret.Service':
return {
'Collections': dbus.Array([dbus.ObjectPath(c.path) for c in self.collections.values()], signature='o')
}
else:
raise InvalidArgs('Unknown %s interface' % interface_name)
@dbus.service.method(dbus.PROPERTIES_IFACE, in_signature='ssv')
def Set(self, interface_name, property_name, new_value):
if interface_name != 'org.freedesktop.Secret.Collection':
raise InvalidArgs('Unknown %s interface' % interface_name)
raise InvalidArgs('Not a writable property %s' % property_name)
2012-01-11 06:44:32 +00:00
def parse_options(args):
global bus_name, ready_pipe
try:
opts, args = getopt.getopt(args, "nr", ["name=", "ready="])
except getopt.GetoptError, err:
print str(err)
sys.exit(2)
for o, a in opts:
if o in ("-n", "--name"):
bus_name = a
elif o in ("-r", "--ready"):
ready_pipe = int(a)
else:
assert False, "unhandled option"
return args
parse_options(sys.argv[1:])