modular.py

# Copyright 2011-2012 GRNET S.A. All rights reserved.
#
# Redistribution and use in source and binary forms, with or
# without modification, are permitted provided that the following
# conditions are met:
#
#   1. Redistributions of source code must retain the above
#      copyright notice, this list of conditions and the following
#      disclaimer.
#
#   2. Redistributions in binary form must reproduce the above
#      copyright notice, this list of conditions and the following
#      disclaimer in the documentation and/or other materials
#      provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# The views and conclusions contained in the software and
# documentation are those of the authors and should not be
# interpreted as representing official policies, either expressed
# or implied, of GRNET S.A.

import sys
import uuid as uuidlib
import logging
import hashlib
import binascii

from synnefo.lib.quotaholder import QuotaholderClient

from base import (DEFAULT_ACCOUNT_QUOTA, DEFAULT_CONTAINER_QUOTA,
                  DEFAULT_CONTAINER_VERSIONING, NotAllowedError, QuotaError,
                  BaseBackend, AccountExists, ContainerExists, AccountNotEmpty,
                  ContainerNotEmpty, ItemNotExists, VersionNotExists)

# Stripped-down version of the HashMap class found in tools.


class HashMap(list):

    def __init__(self, blocksize, blockhash):
        super(HashMap, self).__init__()
        self.blocksize = blocksize
        self.blockhash = blockhash

    def _hash_raw(self, v):
        h = hashlib.new(self.blockhash)
        h.update(v)
        return h.digest()

    def hash(self):
        if len(self) == 0:
            return self._hash_raw('')
        if len(self) == 1:
            return self.__getitem__(0)

        h = list(self)
        s = 2
        while s < len(h):
            s = s * 2
        h += [('\x00' * len(h[0]))] * (s - len(h))
        while len(h) > 1:
            h = [self._hash_raw(h[x] + h[x + 1]) for x in range(0, len(h), 2)]
        return h[0]

# Default modules and settings.
DEFAULT_DB_MODULE = 'pithos.backends.lib.sqlalchemy'
DEFAULT_DB_CONNECTION = 'sqlite:///backend.db'
DEFAULT_BLOCK_MODULE = 'pithos.backends.lib.hashfiler'
DEFAULT_BLOCK_PATH = 'data/'
DEFAULT_BLOCK_UMASK = 0o022
#DEFAULT_QUEUE_MODULE = 'pithos.backends.lib.rabbitmq'
DEFAULT_BLOCK_PARAMS = { 'mappool': None, 'blockpool': None }
#DEFAULT_QUEUE_HOSTS = '[amqp://guest:guest@localhost:5672]'
#DEFAULT_QUEUE_EXCHANGE = 'pithos'
DEFAULT_PUBLIC_URL_ALPHABET = ('0123456789'
                               'abcdefghijklmnopqrstuvwxyz'
                               'ABCDEFGHIJKLMNOPQRSTUVWXYZ')
DEFAULT_PUBLIC_URL_SECURITY = 16

QUEUE_MESSAGE_KEY_PREFIX = 'pithos.%s'
QUEUE_CLIENT_ID = 'pithos'
QUEUE_INSTANCE_ID = '1'

(CLUSTER_NORMAL, CLUSTER_HISTORY, CLUSTER_DELETED) = range(3)

inf = float('inf')

ULTIMATE_ANSWER = 42


logger = logging.getLogger(__name__)


def backend_method(func=None, autocommit=1):
    if func is None:
        def fn(func):
            return backend_method(func, autocommit)
        return fn

    if not autocommit:
        return func

    def fn(self, *args, **kw):
        self.wrapper.execute()
        serials = []
        self.serials = serials
        self.messages = []

        try:
            ret = func(self, *args, **kw)
            for m in self.messages:
                self.queue.send(*m)
            if serials:
                self.quotaholder_serials.insert_many(serials)

                # commit to ensure that the serials are registered
                # even if accept commission fails
                self.wrapper.commit()
                self.wrapper.execute()

                self.quotaholder.accept_commission(
                            context     =   {},
                            clientkey   =   'pithos',
                            serials     =   serials)

                self.quotaholder_serials.delete_many(serials)

            self.wrapper.commit()
            return ret
        except:
            if serials:
                self.quotaholder.reject_commission(
                            context     =   {},
                            clientkey   =   'pithos',
                            serials     =   serials)
            self.wrapper.rollback()
            raise
    return fn


class ModularBackend(BaseBackend):
    """A modular backend.

    Uses modules for SQL functions and storage.
    """

    def __init__(self, db_module=None, db_connection=None,
                 block_module=None, block_path=None, block_umask=None,
                 queue_module=None, queue_hosts=None, queue_exchange=None,
                 quotaholder_enabled=False,
                 quotaholder_url=None, quotaholder_token=None,
                 quotaholder_client_poolsize=None,
                 free_versioning=True, block_params=None,
                 public_url_security=None,
                 public_url_alphabet=None,
                 account_quota_policy=None,
                 container_quota_policy=None,
                 container_versioning_policy=None):
        db_module = db_module or DEFAULT_DB_MODULE
        db_connection = db_connection or DEFAULT_DB_CONNECTION
        block_module = block_module or DEFAULT_BLOCK_MODULE
        block_path = block_path or DEFAULT_BLOCK_PATH
        block_umask = block_umask or DEFAULT_BLOCK_UMASK
        block_params = block_params or DEFAULT_BLOCK_PARAMS
        #queue_module = queue_module or DEFAULT_QUEUE_MODULE
        account_quota_policy = account_quota_policy or DEFAULT_ACCOUNT_QUOTA
        container_quota_policy = container_quota_policy \
            or DEFAULT_CONTAINER_QUOTA
        container_versioning_policy = container_versioning_policy \
            or DEFAULT_CONTAINER_VERSIONING

        self.default_account_policy = {'quota': account_quota_policy}
        self.default_container_policy = {
            'quota': container_quota_policy,
            'versioning': container_versioning_policy
        }
        #queue_hosts = queue_hosts or DEFAULT_QUEUE_HOSTS
        #queue_exchange = queue_exchange or DEFAULT_QUEUE_EXCHANGE

        self.public_url_security = public_url_security or DEFAULT_PUBLIC_URL_SECURITY
        self.public_url_alphabet = public_url_alphabet or DEFAULT_PUBLIC_URL_ALPHABET

        self.hash_algorithm = 'sha256'
        self.block_size = 4 * 1024 * 1024  # 4MB
        self.free_versioning = free_versioning

        def load_module(m):
            __import__(m)
            return sys.modules[m]

        self.db_module = load_module(db_module)
        self.wrapper = self.db_module.DBWrapper(db_connection)
        params = {'wrapper': self.wrapper}
        self.permissions = self.db_module.Permissions(**params)
        self.config = self.db_module.Config(**params)
        self.quotaholder_serials = self.db_module.QuotaholderSerial(**params)
        for x in ['READ', 'WRITE']:
            setattr(self, x, getattr(self.db_module, x))
        self.node = self.db_module.Node(**params)
        for x in ['ROOTNODE', 'SERIAL', 'HASH', 'SIZE', 'TYPE', 'MTIME', 'MUSER', 'UUID', 'CHECKSUM', 'CLUSTER', 'MATCH_PREFIX', 'MATCH_EXACT']:
            setattr(self, x, getattr(self.db_module, x))

        self.block_module = load_module(block_module)
        self.block_params = block_params
        params = {'path': block_path,
                  'block_size': self.block_size,
                  'hash_algorithm': self.hash_algorithm,
                  'umask': block_umask}
        params.update(self.block_params)
        self.store = self.block_module.Store(**params)

        if queue_module and queue_hosts:
            self.queue_module = load_module(queue_module)
            params = {'hosts': queue_hosts,
                      'exchange': queue_exchange,
                      'client_id': QUEUE_CLIENT_ID}
            self.queue = self.queue_module.Queue(**params)
        else:
            class NoQueue:
                def send(self, *args):
                    pass

                def close(self):
                    pass

            self.queue = NoQueue()

        self.quotaholder_enabled = quotaholder_enabled
        if quotaholder_enabled:
            self.quotaholder_url = quotaholder_url
            self.quotaholder_token = quotaholder_token
            self.quotaholder = QuotaholderClient(
                                    quotaholder_url,
                                    token=quotaholder_token,
                                    poolsize=quotaholder_client_poolsize)

        self.serials = []
        self.messages = []

    def close(self):
        self.wrapper.close()
        self.queue.close()

    @property
    def using_external_quotaholder(self):
        return self.quotaholder_enabled

    @backend_method
    def list_accounts(self, user, marker=None, limit=10000):
        """Return a list of accounts the user can access."""

        logger.debug("list_accounts: %s %s %s", user, marker, limit)
        allowed = self._allowed_accounts(user)
        start, limit = self._list_limits(allowed, marker, limit)
        return allowed[start:start + limit]

    @backend_method
    def get_account_meta(
            self, user, account, domain, until=None, include_user_defined=True,
            external_quota=None):
        """Return a dictionary with the account metadata for the domain."""

        logger.debug(
            "get_account_meta: %s %s %s %s", user, account, domain, until)
        path, node = self._lookup_account(account, user == account)
        if user != account:
            if until or node is None or account not in self._allowed_accounts(user):
                raise NotAllowedError
        try:
            props = self._get_properties(node, until)
            mtime = props[self.MTIME]
        except NameError:
            props = None
            mtime = until
        count, bytes, tstamp = self._get_statistics(node, until)
        tstamp = max(tstamp, mtime)
        if until is None:
            modified = tstamp
        else:
            modified = self._get_statistics(
                node)[2]  # Overall last modification.
            modified = max(modified, mtime)

        if user != account:
            meta = {'name': account}
        else:
            meta = {}
            if props is not None and include_user_defined:
                meta.update(
                    dict(self.node.attribute_get(props[self.SERIAL], domain)))
            if until is not None:
                meta.update({'until_timestamp': tstamp})
            meta.update({'name': account, 'count': count, 'bytes': bytes})
            if self.using_external_quotaholder:
                external_quota = external_quota or {}
                meta['bytes'] = external_quota.get('currValue', 0)
        meta.update({'modified': modified})
        return meta

    @backend_method
    def update_account_meta(self, user, account, domain, meta, replace=False):
        """Update the metadata associated with the account for the domain."""

        logger.debug("update_account_meta: %s %s %s %s %s", user,
                     account, domain, meta, replace)
        if user != account:
            raise NotAllowedError
        path, node = self._lookup_account(account, True)
        self._put_metadata(user, node, domain, meta, replace)

    @backend_method
    def get_account_groups(self, user, account):
        """Return a dictionary with the user groups defined for this account."""

        logger.debug("get_account_groups: %s %s", user, account)
        if user != account:
            if account not in self._allowed_accounts(user):
                raise NotAllowedError
            return {}
        self._lookup_account(account, True)
        return self.permissions.group_dict(account)

    @backend_method
    def update_account_groups(self, user, account, groups, replace=False):
        """Update the groups associated with the account."""

        logger.debug("update_account_groups: %s %s %s %s", user,
                     account, groups, replace)
        if user != account:
            raise NotAllowedError
        self._lookup_account(account, True)
        self._check_groups(groups)
        if replace:
            self.permissions.group_destroy(account)
        for k, v in groups.iteritems():
            if not replace:  # If not already deleted.
                self.permissions.group_delete(account, k)
            if v:
                self.permissions.group_addmany(account, k, v)

    @backend_method
    def get_account_policy(self, user, account, external_quota=None):
        """Return a dictionary with the account policy."""

        logger.debug("get_account_policy: %s %s", user, account)
        if user != account:
            if account not in self._allowed_accounts(user):
                raise NotAllowedError
            return {}
        path, node = self._lookup_account(account, True)
        policy = self._get_policy(node, is_account_policy=True)
        if self.using_external_quotaholder:
            external_quota = external_quota or {}
            policy['quota'] = external_quota.get('maxValue', 0)
        return policy

    @backend_method
    def update_account_policy(self, user, account, policy, replace=False):
        """Update the policy associated with the account."""

        logger.debug("update_account_policy: %s %s %s %s", user,
                     account, policy, replace)
        if user != account:
            raise NotAllowedError
        path, node = self._lookup_account(account, True)
        self._check_policy(policy, is_account_policy=True)
        self._put_policy(node, policy, replace, is_account_policy=True)

    @backend_method
    def put_account(self, user, account, policy=None):
        """Create a new account with the given name."""

        logger.debug("put_account: %s %s %s", user, account, policy)
        policy = policy or {}
        if user != account:
            raise NotAllowedError
        node = self.node.node_lookup(account)
        if node is not None:
            raise AccountExists('Account already exists')
        if policy:
            self._check_policy(policy, is_account_policy=True)
        node = self._put_path(user, self.ROOTNODE, account)
        self._put_policy(node, policy, True, is_account_policy=True)

    @backend_method
    def delete_account(self, user, account):
        """Delete the account with the given name."""

        logger.debug("delete_account: %s %s", user, account)
        if user != account:
            raise NotAllowedError
        node = self.node.node_lookup(account)
        if node is None:
            return
        if not self.node.node_remove(node):
            raise AccountNotEmpty('Account is not empty')
        self.permissions.group_destroy(account)

    @backend_method
    def list_containers(self, user, account, marker=None, limit=10000, shared=False, until=None, public=False):
        """Return a list of containers existing under an account."""

        logger.debug("list_containers: %s %s %s %s %s %s %s", user,
                     account, marker, limit, shared, until, public)
        if user != account:
            if until or account not in self._allowed_accounts(user):
                raise NotAllowedError
            allowed = self._allowed_containers(user, account)
            start, limit = self._list_limits(allowed, marker, limit)
            return allowed[start:start + limit]
        if shared or public:
            allowed = set()
            if shared:
                allowed.update([x.split('/', 2)[1] for x in self.permissions.access_list_shared(account)])
            if public:
                allowed.update([x[0].split('/', 2)[1] for x in self.permissions.public_list(account)])
            allowed = sorted(allowed)
            start, limit = self._list_limits(allowed, marker, limit)
            return allowed[start:start + limit]
        node = self.node.node_lookup(account)
        containers = [x[0] for x in self._list_object_properties(
            node, account, '', '/', marker, limit, False, None, [], until)]
        start, limit = self._list_limits(
            [x[0] for x in containers], marker, limit)
        return containers[start:start + limit]

    @backend_method
    def list_container_meta(self, user, account, container, domain, until=None):
        """Return a list with all the container's object meta keys for the domain."""

        logger.debug("list_container_meta: %s %s %s %s %s", user,
                     account, container, domain, until)
        allowed = []
        if user != account:
            if until:
                raise NotAllowedError
            allowed = self.permissions.access_list_paths(
                user, '/'.join((account, container)))
            if not allowed:
                raise NotAllowedError
        path, node = self._lookup_container(account, container)
        before = until if until is not None else inf
        allowed = self._get_formatted_paths(allowed)
        return self.node.latest_attribute_keys(node, domain, before, CLUSTER_DELETED, allowed)

    @backend_method
    def get_container_meta(self, user, account, container, domain, until=None, include_user_defined=True):
        """Return a dictionary with the container metadata for the domain."""

        logger.debug("get_container_meta: %s %s %s %s %s", user,
                     account, container, domain, until)
        if user != account:
            if until or container not in self._allowed_containers(user, account):
                raise NotAllowedError
        path, node = self._lookup_container(account, container)
        props = self._get_properties(node, until)
        mtime = props[self.MTIME]
        count, bytes, tstamp = self._get_statistics(node, until)
        tstamp = max(tstamp, mtime)
        if until is None:
            modified = tstamp
        else:
            modified = self._get_statistics(
                node)[2]  # Overall last modification.
            modified = max(modified, mtime)

        if user != account:
            meta = {'name': container}
        else:
            meta = {}
            if include_user_defined:
                meta.update(
                    dict(self.node.attribute_get(props[self.SERIAL], domain)))
            if until is not None:
                meta.update({'until_timestamp': tstamp})
            meta.update({'name': container, 'count': count, 'bytes': bytes})
        meta.update({'modified': modified})
        return meta

    @backend_method
    def update_container_meta(self, user, account, container, domain, meta, replace=False):
        """Update the metadata associated with the container for the domain."""

        logger.debug("update_container_meta: %s %s %s %s %s %s",
                     user, account, container, domain, meta, replace)
        if user != account:
            raise NotAllowedError
        path, node = self._lookup_container(account, container)
        src_version_id, dest_version_id = self._put_metadata(
            user, node, domain, meta, replace)
        if src_version_id is not None:
            versioning = self._get_policy(
                node, is_account_policy=False)['versioning']
            if versioning != 'auto':
                self.node.version_remove(src_version_id)

    @backend_method
    def get_container_policy(self, user, account, container):
        """Return a dictionary with the container policy."""

        logger.debug(
            "get_container_policy: %s %s %s", user, account, container)
        if user != account:
            if container not in self._allowed_containers(user, account):
                raise NotAllowedError
            return {}
        path, node = self._lookup_container(account, container)
        return self._get_policy(node, is_account_policy=False)

    @backend_method
    def update_container_policy(self, user, account, container, policy, replace=False):
        """Update the policy associated with the container."""

        logger.debug("update_container_policy: %s %s %s %s %s",
                     user, account, container, policy, replace)
        if user != account:
            raise NotAllowedError
        path, node = self._lookup_container(account, container)
        self._check_policy(policy, is_account_policy=False)
        self._put_policy(node, policy, replace, is_account_policy=False)

    @backend_method
    def put_container(self, user, account, container, policy=None):
        """Create a new container with the given name."""

        logger.debug(
            "put_container: %s %s %s %s", user, account, container, policy)
        policy = policy or {}
        if user != account:
            raise NotAllowedError
        try:
            path, node = self._lookup_container(account, container)
        except NameError:
            pass
        else:
            raise ContainerExists('Container already exists')
        if policy:
            self._check_policy(policy, is_account_policy=False)
        path = '/'.join((account, container))
        node = self._put_path(
            user, self._lookup_account(account, True)[1], path)
        self._put_policy(node, policy, True, is_account_policy=False)

    @backend_method
    def delete_container(self, user, account, container, until=None, prefix='', delimiter=None):
        """Delete/purge the container with the given name."""

        logger.debug("delete_container: %s %s %s %s %s %s", user,
                     account, container, until, prefix, delimiter)
        if user != account:
            raise NotAllowedError
        path, node = self._lookup_container(account, container)

        if until is not None:
            hashes, size, serials = self.node.node_purge_children(
                node, until, CLUSTER_HISTORY)
            for h in hashes:
                self.store.map_delete(h)
            self.node.node_purge_children(node, until, CLUSTER_DELETED)
            if not self.free_versioning:
                self._report_size_change(
                    user, account, -size, {
                        'action':'container purge',
                        'path': path,
                        'versions': ','.join(str(i) for i in serials)
                    }
                )
            return

        if not delimiter:
            if self._get_statistics(node)[0] > 0:
                raise ContainerNotEmpty('Container is not empty')
            hashes, size, serials = self.node.node_purge_children(
                node, inf, CLUSTER_HISTORY)
            for h in hashes:
                self.store.map_delete(h)
            self.node.node_purge_children(node, inf, CLUSTER_DELETED)
            self.node.node_remove(node)
            if not self.free_versioning:
                self._report_size_change(
                    user, account, -size, {
                        'action':'container purge',
                        'path': path,
                        'versions': ','.join(str(i) for i in serials)
                    }
                )
        else:
            # remove only contents
            src_names = self._list_objects_no_limit(user, account, container, prefix='', delimiter=None, virtual=False, domain=None, keys=[], shared=False, until=None, size_range=None, all_props=True, public=False)
            paths = []
            for t in src_names:
                path = '/'.join((account, container, t[0]))
                node = t[2]
                src_version_id, dest_version_id = self._put_version_duplicate(user, node, size=0, type='', hash=None, checksum='', cluster=CLUSTER_DELETED)
                del_size = self._apply_versioning(
                    account, container, src_version_id)
                self._report_size_change(
                        user, account, -del_size, {
                                'action': 'object delete',
                                'path': path,
                        'versions': ','.join([str(dest_version_id)])
                     }
                )
                self._report_object_change(
                    user, account, path, details={'action': 'object delete'})
                paths.append(path)
            self.permissions.access_clear_bulk(paths)

    def _list_objects(self, user, account, container, prefix, delimiter, marker, limit, virtual, domain, keys, shared, until, size_range, all_props, public):
        if user != account and until:
            raise NotAllowedError
        if shared and public:
            # get shared first
            shared_paths = self._list_object_permissions(
                user, account, container, prefix, shared=True, public=False)
            objects = set()
            if shared_paths:
                path, node = self._lookup_container(account, container)
                shared_paths = self._get_formatted_paths(shared_paths)
                objects |= set(self._list_object_properties(node, path, prefix, delimiter, marker, limit, virtual, domain, keys, until, size_range, shared_paths, all_props))

            # get public
            objects |= set(self._list_public_object_properties(
                user, account, container, prefix, all_props))
            objects = list(objects)

            objects.sort(key=lambda x: x[0])
            start, limit = self._list_limits(
                [x[0] for x in objects], marker, limit)
            return objects[start:start + limit]
        elif public:
            objects = self._list_public_object_properties(
                user, account, container, prefix, all_props)
            start, limit = self._list_limits(
                [x[0] for x in objects], marker, limit)
            return objects[start:start + limit]

        allowed = self._list_object_permissions(
            user, account, container, prefix, shared, public)
        if shared and not allowed:
            return []
        path, node = self._lookup_container(account, container)
        allowed = self._get_formatted_paths(allowed)
        objects = self._list_object_properties(node, path, prefix, delimiter, marker, limit, virtual, domain, keys, until, size_range, allowed, all_props)
        start, limit = self._list_limits(
            [x[0] for x in objects], marker, limit)
        return objects[start:start + limit]

    def _list_public_object_properties(self, user, account, container, prefix, all_props):
        public = self._list_object_permissions(
            user, account, container, prefix, shared=False, public=True)
        paths, nodes = self._lookup_objects(public)
        path = '/'.join((account, container))
        cont_prefix = path + '/'
        paths = [x[len(cont_prefix):] for x in paths]
        props = self.node.version_lookup_bulk(nodes, all_props=all_props)
        objects = [(path,) + props for path, props in zip(paths, props)]
        return objects

    def _list_objects_no_limit(self, user, account, container, prefix, delimiter, virtual, domain, keys, shared, until, size_range, all_props, public):
        objects = []
        while True:
            marker = objects[-1] if objects else None
            limit = 10000
            l = self._list_objects(user, account, container, prefix, delimiter, marker, limit, virtual, domain, keys, shared, until, size_range, all_props, public)
            objects.extend(l)
            if not l or len(l) < limit:
                break
        return objects

    def _list_object_permissions(self, user, account, container, prefix, shared, public):
        allowed = []
        path = '/'.join((account, container, prefix)).rstrip('/')
        if user != account:
            allowed = self.permissions.access_list_paths(user, path)
            if not allowed:
                raise NotAllowedError
        else:
            allowed = set()
            if shared:
                allowed.update(self.permissions.access_list_shared(path))
            if public:
                allowed.update(
                    [x[0] for x in self.permissions.public_list(path)])
            allowed = sorted(allowed)
            if not allowed:
                return []
        return allowed

    @backend_method
    def list_objects(self, user, account, container, prefix='', delimiter=None, marker=None, limit=10000, virtual=True, domain=None, keys=None, shared=False, until=None, size_range=None, public=False):
        """Return a list of object (name, version_id) tuples existing under a container."""

        logger.debug("list_objects: %s %s %s %s %s %s %s %s %s %s %s %s %s %s", user, account, container, prefix, delimiter, marker, limit, virtual, domain, keys, shared, until, size_range, public)
        keys = keys or []
        return self._list_objects(user, account, container, prefix, delimiter, marker, limit, virtual, domain, keys, shared, until, size_range, False, public)

    @backend_method
    def list_object_meta(self, user, account, container, prefix='', delimiter=None, marker=None, limit=10000, virtual=True, domain=None, keys=None, shared=False, until=None, size_range=None, public=False):
        """Return a list of object metadata dicts existing under a container."""

        logger.debug("list_object_meta: %s %s %s %s %s %s %s %s %s %s %s %s %s %s", user, account, container, prefix, delimiter, marker, limit, virtual, domain, keys, shared, until, size_range, public)
        keys = keys or []
        props = self._list_objects(user, account, container, prefix, delimiter, marker, limit, virtual, domain, keys, shared, until, size_range, True, public)
        objects = []
        for p in props:
            if len(p) == 2:
                objects.append({'subdir': p[0]})
            else:
                objects.append({'name': p[0],
                                'bytes': p[self.SIZE + 1],
                                'type': p[self.TYPE + 1],
                                'hash': p[self.HASH + 1],
                                'version': p[self.SERIAL + 1],
                                'version_timestamp': p[self.MTIME + 1],
                                'modified': p[self.MTIME + 1] if until is None else None,
                                'modified_by': p[self.MUSER + 1],
                                'uuid': p[self.UUID + 1],
                                'checksum': p[self.CHECKSUM + 1]})
        return objects

    @backend_method
    def list_object_permissions(self, user, account, container, prefix=''):
        """Return a list of paths that enforce permissions under a container."""

        logger.debug("list_object_permissions: %s %s %s %s", user,
                     account, container, prefix)
        return self._list_object_permissions(user, account, container, prefix, True, False)

    @backend_method
    def list_object_public(self, user, account, container, prefix=''):
        """Return a dict mapping paths to public ids for objects that are public under a container."""

        logger.debug("list_object_public: %s %s %s %s", user,
                     account, container, prefix)
        public = {}
        for path, p in self.permissions.public_list('/'.join((account, container, prefix))):
            public[path] = p
        return public

    @backend_method
    def get_object_meta(self, user, account, container, name, domain, version=None, include_user_defined=True):
        """Return a dictionary with the object metadata for the domain."""

        logger.debug("get_object_meta: %s %s %s %s %s %s", user,
                     account, container, name, domain, version)
        self._can_read(user, account, container, name)
        path, node = self._lookup_object(account, container, name)
        props = self._get_version(node, version)
        if version is None:
            modified = props[self.MTIME]
        else:
            try:
                modified = self._get_version(
                    node)[self.MTIME]  # Overall last modification.
            except NameError:  # Object may be deleted.
                del_props = self.node.version_lookup(
                    node, inf, CLUSTER_DELETED)
                if del_props is None:
                    raise ItemNotExists('Object does not exist')
                modified = del_props[self.MTIME]

        meta = {}
        if include_user_defined:
            meta.update(
                dict(self.node.attribute_get(props[self.SERIAL], domain)))
        meta.update({'name': name,
                     'bytes': props[self.SIZE],
                     'type': props[self.TYPE],
                     'hash': props[self.HASH],
                     'version': props[self.SERIAL],
                     'version_timestamp': props[self.MTIME],
                     'modified': modified,
                     'modified_by': props[self.MUSER],
                     'uuid': props[self.UUID],
                     'checksum': props[self.CHECKSUM]})
        return meta

    @backend_method