# # Author: Sascha Silbe # # 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 warranty of # MERCHANTABILITY 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 . """ Gdatastore D-Bus service API """ import hashlib import logging import os import shutil from subprocess import Popen, PIPE import tempfile import time import uuid import dbus import dbus.service import gconf from gdatastore.index import Index DBUS_SERVICE_NATIVE_V1 = 'org.silbe.GDataStore' DBUS_INTERFACE_NATIVE_V1 = 'org.silbe.GDataStore1' DBUS_PATH_NATIVE_V1 = '/org/silbe/GDataStore1' DBUS_SERVICE_SUGAR_V2 = 'org.laptop.sugar.DataStore' DBUS_INTERFACE_SUGAR_V2 = 'org.laptop.sugar.DataStore' DBUS_PATH_SUGAR_V2 = '/org/laptop/sugar/DataStore' DBUS_SERVICE_SUGAR_V3 = 'org.laptop.sugar.DataStore' DBUS_INTERFACE_SUGAR_V3 = 'org.laptop.sugar.DataStore2' DBUS_PATH_SUGAR_V3 = '/org/laptop/sugar/DataStore2' class DataStoreError(Exception): pass class GitError(DataStoreError): def __init__(self, returncode, stderr): self.returncode = returncode self.stderr = unicode(stderr) Exception.__init__(self) def __unicode__(self): return u'Git returned with exit code #%d: %s' % (self.returncode, self.stderr) def __str__(self): return self.__unicode__() class DBusApiSugarV2(dbus.service.Object): """Compatibility layer for the Sugar 0.84+ data store D-Bus API """ def __init__(self, internal_api): self._internal_api = internal_api bus_name = dbus.service.BusName(DBUS_SERVICE_SUGAR_V2, bus=dbus.SessionBus(), replace_existing=False, allow_replacement=False, do_not_queue=True) dbus.service.Object.__init__(self, bus_name, DBUS_PATH_SUGAR_V2) @dbus.service.method(DBUS_INTERFACE_SUGAR_V2, in_signature='a{sv}sb', out_signature='s', async_callbacks=('async_cb', 'async_err_cb'), byte_arrays=True) def create(self, props, file_path, transfer_ownership, async_cb, async_err_cb): def success_cb(tree_id, child_id): self.Created(tree_id) async_cb(tree_id) self._internal_api.save(tree_id='', parent_id='', metadata=props, path=file_path, delete_after=transfer_ownership, async_cb=success_cb, async_err_cb=async_err_cb) @dbus.service.signal(DBUS_INTERFACE_SUGAR_V2, signature='s') def Created(self, uid): # pylint: disable-msg=C0103 pass @dbus.service.method(DBUS_INTERFACE_SUGAR_V2, in_signature='sa{sv}sb', out_signature='', async_callbacks=('async_cb', 'async_err_cb'), byte_arrays=True) def update(self, uid, props, file_path, transfer_ownership, async_cb, async_err_cb): def success_cb(tree_id, child_id): self.Updated(tree_id) async_cb() latest_versions = self._get_latest(uid) if not latest_versions: raise ValueError('Trying to update non-existant entry %s - wanted' ' to use create()?' % (uid, )) parent = latest_versions[0] object_id = parent['tree_id'], parent['version_id'] if self._compare_checksums(parent, file_path): self._internal_api.change_metadata(object_id, props) return success_cb(uid, None) self._internal_api.save(tree_id=uid, parent_id=parent['version_id'], metadata=props, path=file_path, delete_after=transfer_ownership, async_cb=success_cb, async_err_cb=async_err_cb) @dbus.service.signal(DBUS_INTERFACE_SUGAR_V2, signature='s') def Updated(self, uid): # pylint: disable-msg=C0103 pass @dbus.service.method(DBUS_INTERFACE_SUGAR_V2, in_signature='a{sv}as', out_signature='aa{sv}u') def find(self, query, properties): if 'uid' in properties: properties.append('tree_id') properties.remove('uid') options = {'metadata': properties} for name in ['offset', 'limit', 'order_by']: if name in query: options[name] = query.pop(name) if 'uid' in query: query['tree_id'] = query.pop('uid') results, count = self._internal_api.find(query, options, query.pop('query', None)) if not properties or 'tree_id' in properties: for entry in results: entry['uid'] = entry.pop('tree_id') return results, count @dbus.service.method(DBUS_INTERFACE_SUGAR_V2, in_signature='s', out_signature='s', sender_keyword='sender') def get_filename(self, uid, sender=None): latest_versions = self._get_latest(uid) if not latest_versions: raise ValueError('Entry %s does not exist' % (uid, )) object_id = (uid, latest_versions[0]['version_id']) return self._internal_api.get_data_path(object_id, sender=sender) @dbus.service.method(DBUS_INTERFACE_SUGAR_V2, in_signature='s', out_signature='a{sv}') def get_properties(self, uid): latest_versions = self._get_latest(uid) if not latest_versions: raise ValueError('Entry %s does not exist' % (uid, )) latest_versions[0]['uid'] = latest_versions[0].pop('tree_id') del latest_versions[0]['version_id'] return latest_versions[0] @dbus.service.method(DBUS_INTERFACE_SUGAR_V2, in_signature='sa{sv}', out_signature='as') def get_uniquevaluesfor(self, propertyname, query=None): return self._internal_api.find_unique_values(query, propertyname) @dbus.service.method(DBUS_INTERFACE_SUGAR_V2, in_signature='s', out_signature='') def delete(self, uid): latest_versions = self._get_latest(uid) if not latest_versions: raise ValueError('Entry %s does not exist' % (uid, )) self._internal_api.delete((uid, latest_versions[0]['version_id'])) self.Deleted(uid) @dbus.service.signal(DBUS_INTERFACE_SUGAR_V2, signature='s') def Deleted(self, uid): # pylint: disable-msg=C0103 pass @dbus.service.method(DBUS_INTERFACE_SUGAR_V2, in_signature='', out_signature='aa{sv}') def mounts(self): return [{'id': 1}] @dbus.service.signal(DBUS_INTERFACE_SUGAR_V2, signature='a{sv}') def Mounted(self, descriptior): # pylint: disable-msg=C0103 pass @dbus.service.signal(DBUS_INTERFACE_SUGAR_V2, signature='a{sv}') def Unmounted(self, descriptor): # pylint: disable-msg=C0103 pass def _get_latest(self, uid): return self._internal_api.find({'tree_id': uid}, {'limit': 1, 'order_by': ['+timestamp']})[0] def _compare_checksums(self, parent, child_data_path): parent_object_id = (parent['tree_id'], parent['version_id']) parent_data_path = self._internal_api.get_data_path(parent_object_id) if bool(child_data_path) ^ bool(parent_data_path): return False elif not child_data_path: return True return False parent_checksum = self._internal_api.get_data_checksum( parent_object_id) child_checksum = calculate_checksum(child_data_path) return parent_checksum == child_checksum class InternalApi(object): def __init__(self, base_dir): self._base_dir = base_dir self._checkouts_dir = os.path.join(base_dir, 'checkouts') if not os.path.exists(self._checkouts_dir): os.makedirs(self._checkouts_dir) self._git_dir = os.path.join(base_dir, 'git') self._git_env = {} gconf_client = gconf.client_get_default() self._max_versions = gconf_client.get_int( '/desktop/sugar/datastore/max_versions') logging.debug('max_versions=%r', self._max_versions) self._index = Index(os.path.join(self._base_dir, 'index')) self._migrate() def change_metadata(self, object_id, metadata): logging.debug('change_metadata(%r, %r)', object_id, metadata) metadata['tree_id'], metadata['version_id'] = object_id if 'creation_time' not in metadata: old_metadata = self._index.retrieve(object_id) metadata['creation_time'] = old_metadata['creation_time'] self._index.store(object_id, metadata) def delete(self, object_id): logging.debug('delete(%r)', object_id) self._index.delete(object_id) self._git_call('update-ref', ['-d', _format_ref(*object_id)]) def get_data_path(self, (tree_id, version_id), sender=None): logging.debug('get_data_path((%r, %r), %r)', tree_id, version_id, sender) ref_name = _format_ref(tree_id, version_id) top_level_entries = self._git_call('ls-tree', [ref_name]).splitlines() if len(top_level_entries) == 1 and \ top_level_entries[0].endswith('\tdata'): blob_hash = top_level_entries[0].split('\t')[0].split(' ')[2] return self._checkout_file(blob_hash) return self._checkout_dir(ref_name) def find(self, query_dict, options, query_string=None): logging.debug('find(%r, %r, %r)', query_dict, options, query_string) entries, total_count = self._index.find(query_dict, query_string, options) #logging.debug('object_ids=%r', object_ids) property_names = options.pop('metadata', None) for entry in entries: for name in entry.keys(): if property_names and name not in property_names: del entry[name] elif isinstance(entry[name], str): entry[name] = dbus.ByteArray(entry[name]) return entries, total_count def find_unique_values(self, query, name): logging.debug('find_unique_values(%r, %r)', query, name) return ['org.sugarlabs.DataStoreTest1', 'org.sugarlabs.DataStoreTest2'] def save(self, tree_id, parent_id, metadata, path, delete_after, async_cb, async_err_cb): logging.debug('save(%r, %r, %r, %r, %r)', tree_id, parent_id, metadata, path, delete_after) if path: path = os.path.realpath(path) if not os.access(path, os.R_OK): raise ValueError('Invalid path given.') if delete_after and not os.access(os.path.dirname(path), os.W_OK): raise ValueError('Deletion requested for read-only directory') if (not tree_id) and parent_id: raise ValueError('tree_id is empty but parent_id is not') if tree_id and not parent_id: if self.find({'tree_id': tree_id}, {'limit': 1})[1]: raise ValueError('No parent_id given but tree_id already ' 'exists') elif parent_id: if not self._index.contains((tree_id, parent_id)): raise ValueError('Given parent does not exist') if not tree_id: tree_id = self._gen_uuid() child_id = metadata.get('version_id') if not child_id: child_id = self._gen_uuid() elif not tree_id: raise ValueError('No tree_id given but metadata contains' ' version_id') elif self._index.contains((tree_id, child_id)): raise ValueError('There is an existing entry with the same tree_id' ' and version_id') if 'timestamp' not in metadata: metadata['timestamp'] = time.time() if 'creation_time' not in metadata: metadata['creation_time'] = metadata['timestamp'] if os.path.isfile(path): metadata['filesize'] = str(os.stat(path).st_size) elif not path: metadata['filesize'] = '0' tree_id = str(tree_id) parent_id = str(parent_id) child_id = str(child_id) metadata['tree_id'] = tree_id metadata['version_id'] = child_id # TODO: check metadata for validity first (index?) self._store_entry(tree_id, child_id, parent_id, path, metadata) self._index.store((tree_id, child_id), metadata) async_cb(tree_id, child_id) def stop(self): logging.debug('stop()') self._index.close() def _add_to_index(self, index_path, path): if os.path.isdir(path): self._git_call('add', ['-A'], work_dir=path, index_path=index_path) elif os.path.isfile(path): object_hash = self._git_call('hash-object', ['-w', path]).strip() mode = os.stat(path).st_mode self._git_call('update-index', ['--add', '--cacheinfo', oct(mode), object_hash, 'data'], index_path=index_path) else: raise DataStoreError('Refusing to store special object %r' % (path, )) def _check_max_versions(self, tree_id): if not self._max_versions: return options = {'all_versions': True, 'offset': self._max_versions, 'metadata': ['tree_id', 'version_id', 'timestamp'], 'order_by': ['+timestamp']} old_versions = self.find({'tree_id': tree_id}, options)[0] logging.info('Deleting old versions: %r', old_versions) for entry in old_versions: self.delete((entry['tree_id'], entry['version_id'])) def _checkout_file(self, blob_hash): fd, file_name = tempfile.mkstemp(dir=self._checkouts_dir) try: self._git_call('cat-file', ['blob', blob_hash], stdout_fd=fd) finally: os.close(fd) return file_name def _checkout_dir(self, ref_name): # FIXME return '' def _create_repo(self): os.makedirs(self._git_dir) self._git_call('init', ['-q', '--bare']) def _find_git_parent(self, tree_id, parent_id): if not parent_id: return None return self._git_call('rev-parse', [_format_ref(tree_id, parent_id)]).strip() def _format_commit_message(self, metadata): return repr(metadata) def _gen_uuid(self): return str(uuid.uuid4()) def _git_call(self, command, args=None, input=None, input_fd=None, stdout_fd=None, work_dir=None, index_path=None): env = dict(self._git_env) if work_dir: env['GIT_WORK_TREE'] = work_dir if index_path: env['GIT_INDEX_FILE'] = index_path logging.debug('calling git %s, env=%r', ['git', command] + (args or []), env) pipe = Popen(['git', command] + (args or []), stdin=input_fd or PIPE, stdout=stdout_fd or PIPE, stderr=PIPE, close_fds=True, cwd=self._git_dir, env=env) stdout, stderr = pipe.communicate(input) if pipe.returncode: raise GitError(pipe.returncode, stderr) return stdout def _migrate(self): if not os.path.exists(self._git_dir): return self._create_repo() def _store_entry(self, tree_id, version_id, parent_id, path, metadata): parent_hash = self._find_git_parent(tree_id, parent_id) commit_message = self._format_commit_message(metadata) tree_hash = self._write_tree(path) commit_options = [tree_hash] if parent_hash: commit_options += ['-p', parent_hash] commit_hash = self._git_call('commit-tree', commit_options, input=commit_message).strip() self._git_call('update-ref', [_format_ref(tree_id, version_id), commit_hash]) def _write_tree(self, path): if not path: return self._git_call('hash-object', ['-w', '-t', 'tree', '--stdin'], input='').strip() index_dir = tempfile.mkdtemp(prefix='gdatastore-') index_path = os.path.join(index_dir, 'index') try: self._add_to_index(index_path, path) return self._git_call('write-tree', index_path=index_path).strip() finally: shutil.rmtree(index_dir) def calculate_checksum(path): checksum = hashlib.sha1() f = file(path) while True: chunk = f.read(65536) if not chunk: return checksum.hexdigest() checksum.update(chunk) def _format_ref(tree_id, version_id): return 'refs/gdatastore/%s/%s' % (tree_id, version_id)