path: root/gdatastore/datastore.py

#
# Author: Sascha Silbe <sascha-pgp@silbe.org>
#
# 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 <http://www.gnu.org/licenses/>.
"""
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)