Synnefo

# 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 collections import defaultdict

from functools import wraps, partial

from traceback import format_exc

try:

    from astakosclient import AstakosClient

except ImportError:

    AstakosClient = None

from base import (DEFAULT_ACCOUNT_QUOTA, DEFAULT_CONTAINER_QUOTA,

                  DEFAULT_CONTAINER_VERSIONING, NotAllowedError, QuotaError,

                  BaseBackend, AccountExists, ContainerExists, AccountNotEmpty,

                  ContainerNotEmpty, ItemNotExists, VersionNotExists,

                  InvalidHash)

class DisabledAstakosClient(object):

    def __init__(self, *args, **kwargs):

        self.args = args

        self.kwargs = kwargs

    def __getattr__(self, name):

        m = ("AstakosClient has been disabled, "

             "yet an attempt to access it was made")

        raise AssertionError(m)

# 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_BLOCK_SIZE = 4 * 1024 * 1024  # 4MB

DEFAULT_HASH_ALGORITHM = 'sha256'

#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)

QUOTA_POLICY = 'quota'

VERSIONING_POLICY = 'versioning'

PROJECT = 'project'

inf = float('inf')

ULTIMATE_ANSWER = 42

DEFAULT_DISKSPACE_RESOURCE = 'pithos.diskspace'

logger = logging.getLogger(__name__)

def backend_method(func):

    @wraps(func)

    def wrapper(self, *args, **kw):

        # if we are inside a database transaction

        # just proceed with the method execution

        # otherwise manage a new transaction

        if self.in_transaction:

            return func(self, *args, **kw)

        try:

            self.pre_exec()

            result = func(self, *args, **kw)

            success_status = True

            return result

        except:

            success_status = False

            raise

        finally:

            self.post_exec(success_status)

    return wrapper

def debug_method(func):

    @wraps(func)

    def wrapper(self, *args, **kw):

        try:

            result = func(self, *args, **kw)

            return result

        except:

            result = format_exc()

            raise

        finally:

            all_args = map(repr, args)

            map(all_args.append, ('%s=%s' % (k, v) for k, v in kw.iteritems()))

            logger.debug(">>> %s(%s) <<< %s" % (

                func.__name__, ', '.join(all_args).rstrip(', '), result))

    return wrapper

def check_allowed_paths(action):

    """Decorator for backend methods checking path access granted to user.

    The 1st argument of the decorated method is expected to be a

    ModularBackend instance, the 2nd the user performing the request and

    the path join of the rest arguments is supposed to be the requested path.

    The decorator checks whether the requested path is among the user's allowed

    cached paths.

    If this is the case, the decorator returns immediately to reduce the

    interactions with the database.

    Otherwise, it proceeds with the execution of the decorated method and if

    the method returns successfully (no exceptions are raised), the requested

    path is added to the user's cached allowed paths.

    :param action: (int) 0 for reads / 1 for writes

    :raises NotAllowedError: the user does not have access to the path

    """

    def decorator(func):

        @wraps(func)

        def wrapper(self, *args):

            user = args[0]

            if action == self.READ:

                d = self.read_allowed_paths

            else:

                d = self.write_allowed_paths

            path = '/'.join(args[1:])

            if path in d.get(user, []):

                return  # access is already checked

            else:

                func(self, *args)   # proceed with access check

                d[user].add(path)  # add path in the allowed user paths

        return wrapper

    return decorator

def list_method(func):

    @wraps(func)

    def wrapper(self, *args, **kw):

        marker = kw.get('marker')

        limit = kw.get('limit')

        result = func(self, *args, **kw)

        start, limit = self._list_limits(result, marker, limit)

        return result[start:start + limit]

    return wrapper

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,

                 block_size=None, hash_algorithm=None,

                 queue_module=None, queue_hosts=None, queue_exchange=None,

                 astakos_auth_url=None, service_token=None,

                 astakosclient_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

        block_size = block_size or DEFAULT_BLOCK_SIZE

        hash_algorithm = hash_algorithm or DEFAULT_HASH_ALGORITHM

        #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 = {}

        self.default_container_policy = {

            QUOTA_POLICY: container_quota_policy,

            VERSIONING_POLICY: container_versioning_policy,

            PROJECT: None

        }

        #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 = hash_algorithm

        self.block_size = block_size

        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.commission_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', 'NODE', 'HASH', 'SIZE', 'TYPE',

                  'MTIME', 'MUSER', 'UUID', 'CHECKSUM', 'CLUSTER',

                  'MATCH_PREFIX', 'MATCH_EXACT']:

            setattr(self, x, getattr(self.db_module, x))

        self.ALLOWED = ['read', 'write']

        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.astakos_auth_url = astakos_auth_url

        self.service_token = service_token

        if not astakos_auth_url or not AstakosClient:

            self.astakosclient = DisabledAstakosClient(

                service_token, astakos_auth_url,

                use_pool=True,

                pool_size=astakosclient_poolsize)

        else:

            self.astakosclient = AstakosClient(

                service_token, astakos_auth_url,

                use_pool=True,

                pool_size=astakosclient_poolsize)

        self.serials = []

        self.messages = []

        self._move_object = partial(self._copy_object, is_move=True)

        self.lock_container_path = False

        self.in_transaction = False

        self._reset_allowed_paths()

    def pre_exec(self, lock_container_path=False):

        self.lock_container_path = lock_container_path

        self.wrapper.execute()

        self.serials = []

        self._reset_allowed_paths()

        self.in_transaction = True

    def post_exec(self, success_status=True):

        if success_status:

            # send messages produced

            for m in self.messages:

                self.queue.send(*m)

            # register serials

            if self.serials:

                self.commission_serials.insert_many(

                    self.serials)

                # commit to ensure that the serials are registered

                # even if resolve commission fails

                self.wrapper.commit()

                # start new transaction

                self.wrapper.execute()

                r = self.astakosclient.resolve_commissions(

                    accept_serials=self.serials,

                    reject_serials=[])

                self.commission_serials.delete_many(

                    r['accepted'])

            self.wrapper.commit()

        else:

            if self.serials:

                r = self.astakosclient.resolve_commissions(

                    accept_serials=[],

                    reject_serials=self.serials)

                self.commission_serials.delete_many(

                    r['rejected'])

            self.wrapper.rollback()

        self.in_transaction = False

    def close(self):

        self.wrapper.close()

        self.queue.close()

    @property

    def using_external_quotaholder(self):

        return not isinstance(self.astakosclient, DisabledAstakosClient)

    @debug_method

    @backend_method

    @list_method

    def list_accounts(self, user, marker=None, limit=10000):

        """Return a list of accounts the user can access."""

        return self._allowed_accounts(user)

    @debug_method

    @backend_method

    def get_account_meta(self, user, account, domain=None, until=None,

                         include_user_defined=True):

        """Return a dictionary with the account metadata for the domain."""

        self._can_read_account(user, account)

        path, node = self._lookup_account(account, user == account)

        if user != account:

            if until or (node is None):

                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, compute=True)

        tstamp = max(tstamp, mtime)

        if until is None:

            modified = tstamp

        else:

            modified = self._get_statistics(

                node, compute=True)[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:

                if domain is None:

                    raise ValueError(

                        'Domain argument is obligatory for getting '

                        'user defined metadata')

                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 = self.astakosclient.service_get_quotas(

                    account)[account]

                meta['bytes'] = sum(d['pithos.diskspace']['usage'] for d in

                                    external_quota.values())

        meta.update({'modified': modified})

        return meta

    @debug_method

    @backend_method

    def update_account_meta(self, user, account, domain, meta, replace=False):

        """Update the metadata associated with the account for the domain."""

        self._can_write_account(user, account)

        path, node = self._lookup_account(account, True)

        self._put_metadata(user, node, domain, meta, replace,

                           update_statistics_ancestors_depth=-1)

    @debug_method

    @backend_method

    def get_account_groups(self, user, account):

        """Return a dictionary with the user groups defined for the account."""

        self._can_read_account(user, account)

        if user != account:

            return {}

        self._lookup_account(account, True)

        return self.permissions.group_dict(account)

    @debug_method

    @backend_method

    def update_account_groups(self, user, account, groups, replace=False):

        """Update the groups associated with the account."""

        self._can_write_account(user, account)

        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)

    @debug_method

    @backend_method

    def get_account_policy(self, user, account):

        """Return a dictionary with the account policy."""

        self._can_read_account(user, account)

        if user != account:

            return {}

        path, node = self._lookup_account(account, True)

        policy = self._get_policy(node, is_account_policy=True)

        if self.using_external_quotaholder:

            external_quota = self.astakosclient.service_get_quotas(

                account)[account]

            policy.update(dict(('%s-%s' % (QUOTA_POLICY, k),

                                v['pithos.diskspace']['limit']) for k, v in

                               external_quota.items()))

        return policy

    @debug_method

    @backend_method

    def update_account_policy(self, user, account, policy, replace=False):

        """Update the policy associated with the account."""

        self._can_write_account(user, account)

        path, node = self._lookup_account(account, True)

        self._put_policy(node, policy, replace, is_account_policy=True,

                         check=True)

    @debug_method

    @backend_method

    def put_account(self, user, account, policy=None):

        """Create a new account with the given name."""

        policy = policy or {}

        self._can_write_account(user, account)

        node = self.node.node_lookup(account)

        if node is not None:

            raise AccountExists('Account already exists')

        node = self._put_path(user, self.ROOTNODE, account,

                              update_statistics_ancestors_depth=-1)

        self._put_policy(node, policy, True, is_account_policy=True,

                         check=True if policy else False)

    @debug_method

    @backend_method

    def delete_account(self, user, account):

        """Delete the account with the given name."""

        self._can_write_account(user, account)

        node = self.node.node_lookup(account)

        if node is None:

            return

        if not self.node.node_remove(node,

                                     update_statistics_ancestors_depth=-1):

            raise AccountNotEmpty('Account is not empty')

        self.permissions.group_destroy(account)

        # remove all the cached allowed paths

        # removing the specific path could be more expensive

        self._reset_allowed_paths()

    @debug_method

    @backend_method

    @list_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."""

        self._can_read_account(user, account)

        if user != account:

            if until:

                raise NotAllowedError

            return self._allowed_containers(user, account)

        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)])

            return sorted(allowed)

        node = self.node.node_lookup(account)

        return [x[0] for x in self._list_object_properties(

            node, account, '', '/', marker, limit, False, None, [], until)]

    @debug_method

    @backend_method

    def list_container_meta(self, user, account, container, domain,

                            until=None):

        """Return a list of the container's object meta keys for a domain."""

        self._can_read_container(user, account, container)

        allowed = []

        if user != account:

            if until:

                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)

    @debug_method

    @backend_method

    def get_container_meta(self, user, account, container, domain=None,

                           until=None, include_user_defined=True):

        """Return a dictionary with the container metadata for the domain."""

        self._can_read_container(user, account, container)

        if user != account:

            if until:

                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:

                if domain is None:

                    raise ValueError(

                        'Domain argument is obligatory for getting '

                        'user defined metadata')

                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

    @debug_method

    @backend_method

    def update_container_meta(self, user, account, container, domain, meta,

                              replace=False):

        """Update the metadata associated with the container for the domain."""

        self._can_write_container(user, account, container)

        path, node = self._lookup_container(account, container)

        src_version_id, dest_version_id = self._put_metadata(

            user, node, domain, meta, replace,

            update_statistics_ancestors_depth=0)

        if src_version_id is not None:

            versioning = self._get_policy(

                node, is_account_policy=False)[VERSIONING_POLICY]

            if versioning != 'auto':

                self.node.version_remove(src_version_id,

                                         update_statistics_ancestors_depth=0)

    @debug_method

    @backend_method

    def get_container_policy(self, user, account, container):

        """Return a dictionary with the container policy."""

        self._can_read_container(user, account, container)

        if user != account:

            return {}

        path, node = self._lookup_container(account, container)

        return self._get_policy(node, is_account_policy=False)

    @debug_method

    @backend_method

    def update_container_policy(self, user, account, container, policy,

                                replace=False):

        """Update the policy associated with the container."""

        self._can_write_container(user, account, container)

        path, node = self._lookup_container(account, container)

        if PROJECT in policy:

            project = self._get_project(node)

            try:

                serial = self.astakosclient.issue_resource_reassignment(

                    holder=account,

                    from_source=project,

                    to_source=policy[PROJECT],

                    provisions={'pithos.diskspace': self.get_container_meta(

                        user, account, container,

                        include_user_defined=False)['bytes']})

            except BaseException, e:

                raise QuotaError(e)

            else:

                self.serials.append(serial)

        self._put_policy(node, policy, replace, is_account_policy=False,

                         default_project=account, check=True)

    @debug_method

    @backend_method

    def put_container(self, user, account, container, policy=None):

        """Create a new container with the given name."""

        policy = policy or {}

        self._can_write_container(user, account, container)

        try:

            path, node = self._lookup_container(account, container)

        except NameError:

            pass

        else:

            raise ContainerExists('Container already exists')

        path = '/'.join((account, container))

        node = self._put_path(

            user, self._lookup_account(account, True)[1], path,

            update_statistics_ancestors_depth=-1)

        self._put_policy(node, policy, True, is_account_policy=False,

                         default_project=account,

                         check=True if policy else False)

    @debug_method

    @backend_method

    def delete_container(self, user, account, container, until=None, prefix='',

                         delimiter=None):

        """Delete/purge the container with the given name."""

        self._can_write_container(user, account, container)

        path, node = self._lookup_container(account, container)

        project = self._get_project(node)

        if until is not None:

            hashes, size, serials = self.node.node_purge_children(

                node, until, CLUSTER_HISTORY,

                update_statistics_ancestors_depth=0)

            for h in hashes:

                self.store.map_delete(h)

            self.node.node_purge_children(node, until, CLUSTER_DELETED,

                                          update_statistics_ancestors_depth=0)

            if not self.free_versioning:

                self._report_size_change(

                    user, account, -size, project, {

                        '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,

                update_statistics_ancestors_depth=0)

            for h in hashes:

                self.store.map_delete(h)

            self.node.node_purge_children(node, inf, CLUSTER_DELETED,

                                          update_statistics_ancestors_depth=0)

            self.node.node_remove(node, update_statistics_ancestors_depth=0)

            if not self.free_versioning:

                self._report_size_change(

                    user, account, -size, project, {

                        '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]

                if not self._exists(node):

                    continue

                src_version_id, dest_version_id = self._put_version_duplicate(

                    user, node, size=0, type='', hash=None, checksum='',

                    cluster=CLUSTER_DELETED,

                    update_statistics_ancestors_depth=1)

                del_size = self._apply_versioning(

                    account, container, src_version_id,

                    update_statistics_ancestors_depth=1)

                self._report_size_change(

                    user, account, -del_size, project, {

                        '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)