Synnefo » Pithos

# Copyright 2011 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.

from pithos.lib.hashfiler import (Filer, ROOTNODE,

                                  SERIAL, PARENT, PATH, SIZE, POPULATION,

                                  POPSIZE, SOURCE, MTIME, CLUSTER)

from base import NotAllowedError

from binascii import hexlify, unhexlify

from os import makedirs, mkdir

from os.path import exists, isdir, split

from itertools import chain

from collections import defaultdict

from heapq import merge

debug = 0

READ  = 0

WRITE = 1

MAIN_CLUSTER  = 0

TRASH_CLUSTER = 1

def backend_method(func=None, name=None, autocommit=1):

    if func is None:

        def fn(func):

            return backend_method(func, name=name, autocommit=autocommit)

        return fn

    name = func.__name__ if name is None else name

    thefunc = None

    if debug:

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

            print "%s %s %s" % (name, args, kw)

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

        thefunc = fn

    else:

        thefunc = func

    #XXX: Can't set __doc__ in methods?!

    #

    #doc = func.__doc__

    #if doc is None:

    #    doc = ''

    #func.__doc__ = getattr(BaseBackend, name).__doc__ + '\n***\n' + doc

    if not autocommit:

        return func

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

        with self.hf:

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

    return method

class HashFilerBack(object):

    def __init__(self, db):

        head, tail = split(db)

        if not exists(head):

            makedirs(head)

        if not isdir(head):

            raise RuntimeError("Cannot open database at '%s'" % (db,))

        blocksize = 4*1024*1024

        hashtype = 'sha256'

        maps = head + '/maps'

        blocks = head + '/blocks'

        if not exists(root):

            mkdir(root)

        db = head + '/db'

        params = {  'filepath': db,

                    'mappath' : maps,

                    'blockpath': blocks,

                    'hashtype': hashtype,

                    'blocksize': blocksize  }

        hf = Filer(**params)

        self.hf = hf

        self.block_size = blocksize

        self.hash_algorithm = hashtype

    def quit(self):

        self.hf.quit()

    def get_account(self, account, root=ROOTNODE):

        #XXX: how are accounts created?

        r = self.hf.file_lookup(root, account, create=1)

        if r is None:

            raise NameError("account '%s' not found" % (account,))

        return r[0]

    def get_container(self, serial, name):

        r = self.hf.file_lookup(serial, name, create=0)

        if r is None:

            raise NameError("container '%s' not found" % (name,))

        return r[0]

    @staticmethod

    def insert_properties(meta, props):

        (serial, parent, path, size,

         population, popsize, source, mtime, cluster) = props

        meta['version'] = serial

        meta['name'] = path

        meta['bytes'] = size + popsize

        meta['count'] = population

        meta['created'] = mtime

        meta['modified'] = mtime

        return meta

    def lookup_file(self, account, container, name,

                    create=0, size=0, authorize=0, before=None):

        # XXX: account/container lookup cache?

        serial = self.get_account(account)

        serial = self.get_container(serial, container)

        if before is None:

            before = float('inf')

        props = self.hf.file_lookup(serial, name,

                                    create=create, size=size, before=before)

        if props is None:

            raise NameError("File '%s/%s/%s' does not exist" %

                            (account, container, name))

        return props

    def can_read(self, user, account, container, path):

        path = "%s:%s:%s" % (account, container, path)

        check = self.hf.access_check

        return check(READ, path, user) or check(WRITE, path, user)

    def can_write(self, user, account, container, path):

        path = "%s:%s:%s" % (account, container, path)

        check = self.hf.access_check

        return check(WRITE, path, user)

    @backend_method

    def get_account_meta(self, user, account, until=None):

        if user != account:

            raise NotAllowedError

        hf = self.hf

        before = float("inf") if until is None else until

        props = hf.file_lookup(ROOTNODE, account, create=0, before=before)

        if props is None:

            #XXX: why not return error?

            meta = {'name': account,

                    'count': 0,

                    'bytes': 0}

            if until is not None:

                meta['until_timestamp'] = 0

            return meta

        serial = props[SERIAL]

        meta = dict(hf.file_get_attributes(serial))

        meta['name'] = account

        self.insert_properties(meta, props)

        if until is not None:

            meta['until_timestamp'] = props[MTIME]

        return meta

    @backend_method

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

        if user != account:

            raise NotAllowedError

        hf = self.hf

        serial = self.get_account(account)

        if replace:

            hf.file_del_attributes(serial)

        hf.file_set_attributes(serial, meta.iteritems())

    @backend_method

    def get_account_groups(self, user, account):

        if user != account:

            raise NotAllowedError

        groups = defaultdict(list)

        grouplist = self.hf.group_list("%s:" % user)

        for name, member in grouplist:

            groups[name].append(member)

        return groups

    @backend_method

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

        if user != account:

            raise NotAllowedError

        hf = self.hf

        group_addmany = hf.group_addmany

        for name, members in groups:

            name = "%s:%s" % (user, name)

            if not members:

                members = (name,)

            group_addmany(name, members)

        if not replace:

            return

        groupnames = hf.group_list("%s:" % user)

        group_destroy = hf.group_destroy

        for name in groupnames:

            if name not in groups:

                group_destroy(name)

    @backend_method

    def delete_account(self, user, account):

        if user != account:

            raise NotAllowedError

        hf = self.hf

        r = hf.file_lookup(ROOTNODE, account)

        if r is None or r[SERIAL] is None:

            raise NameError

        if r[POPULATION]:

            raise IndexError

        hf.file_remove(r[SERIAL])

    @backend_method

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

        if user != account:

            raise NotAllowedError

        parent = self.get_account(account)

        hf = self.hf

        r = hf.file_lookup(parent, name)

        if r is not None:

            raise NameError("%s:%s already exists" % (account, name))

        serial, mtime = hf.file_create(parent, name, 0)

        return serial, mtime

    @backend_method

    def delete_container(self, user, account, name):

        if user != account:

            raise NotAllowedError

        serial = self.get_account(account)

        serial = self.get_container(serial, name)

        r = self.hf.file_remove(serial)

        if r is None:

            raise AssertionError("This should not have happened.")

        elif r == 0:

            #XXX: if container has only trashed objects, is it empty or not?

            msg = "'%s:%s' not empty. will not delete." % (account, name)

            raise IndexError(msg)

    @backend_method

    def get_container_meta(self, user, account, name, until=None):

        if user != account:

            raise NotAllowedError

        serial = self.get_account(account)

        hf = self.hf

        before = float('inf') if until is None else until

        props = hf.file_lookup(serial, name, before=before)

        if props is None:

            raise NameError("%s:%s does not exist" % (account, name))

        serial = props[0]

        r = hf.file_get_attributes(serial)

        meta = dict(r)

        self.insert_properties(meta, props)

        if until:

            meta['until_timestamp'] = props[MTIME]

        return meta

    @backend_method

    def update_container_meta(self, user, account, name, meta, replace=False):

        if user != account:

            raise NotAllowedError

        serial = self.get_account(account)

        serial = self.get_container(serial, name)

        hf = self.hf

        if replace:

            hf.file_del_attributes(serial)

        hf.file_set_attributes(serial, meta.iteritems())

    @backend_method

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

        return {}

    @backend_method

    def update_container_policy(self, user, account,

                                container, policy, replace=0):

        return

    @backend_method

    def list_containers(self, user, account,

                        marker=None, limit=10000, until=None):

        if not marker:

            marker = ''

        if limit is None:

            limit = -1

        serial = self.get_account(account)

        r = self.hf.file_list(serial, '', start=marker, limit=limit)

        objects, prefixes = r

        return [(o[PATH], o[SERIAL]) for o in objects]

    @backend_method

    def list_objects(self, user, account, container, prefix='',

                     delimiter=None, marker=None, limit=10000,

                     virtual=True, keys=[], until=None):

        if user != account:

            raise NotAllowedError

        if not marker:

            #XXX: '' is a valid string, should be None instead

            marker = None

        if limit is None:

            #XXX: limit should be an integer

            limit = -1

        if not delimiter:

            delimiter = None

        serial = self.get_account(account)

        serial = self.get_container(serial, container)

        hf = self.hf

        filterq = ','.join(keys) if keys else None

        r = hf.file_list(serial, prefix=prefix, start=marker,

                         versions=0, filterq=filterq,

                         delimiter=delimiter, limit=limit)

        objects, prefixes = r

        # XXX: virtual is not needed since we get them anyway

        #if virtual:

        #    return [(p, None) for p in prefixes]

        objects = ((o[PATH], o[MTIME]) for o in objects)

        prefixes = ((p, None) for p in prefixes) if virtual else ()

        return list(merge(objects, prefixes))

    @backend_method

    def list_object_meta(self, user, account, container, until=None):

        serial = self.get_account(account)

        serial = self.get_container(serial, container)

        return self.hf.file_list_attributes(serial)

    @backend_method

    def get_object_meta(self, user, account, container, name, until=None):

        props = self.lookup_file(account, container, name,

                                 before=until, authorize=2)

        serial = props[0]

        meta = dict(self.hf.file_get_attributes(serial))

        self.insert_properties(meta, props)

        meta['version_timestamp'] = props[MTIME]

        meta['modified_by'] = 'fuck you'

        return meta

    @backend_method

    def update_object_meta(self, user, account, container,

                           name, meta, replace=False):

        props = self.lookup_file(account, container, name, authorize=1)

        serial = props[0]

        hf = self.hf

        if replace:

            hf.file_del_attributes(serial)

        hf.file_set_attributes(serial, meta.iteritems())

    @backend_method

    def get_object_permissions(self, user, account, container, name):

        path = "%s/%s/%s" % (account, container, name)

        perms = defaultdict(list)

        for access, ident in self.hf.access_list(path):

            if access == READ and ident:

                perms['read'].append(ident)

            if access == WRITE and ident:

                perms['write'].append(ident)

        return path, perms

    @backend_method

    def update_object_permissions(self, user, account, container, name, permissions):

        if user != account:

            return NotAllowedError

        hf = self.hf

        path = "%s/%s/%s" % (account, container, name)

        feature = hf.feature_create()

        hf.access_grant(READ, path, members=permissions['read'])

        hf.access_grant(WRITE, path, members=permissions['write'])

        r = hf.xfeature_bestow(path, feature)

        assert(not r)

    @backend_method

    def get_object_public(self, user, account, container, name):

        """Return the public URL of the object if applicable."""

        return None

    @backend_method

    def update_object_public(self, user, account, container, name, public):

        """Update the public status of the object."""

        return

    @backend_method

    def get_object_hashmap(self, user, account, container, name, version=None):

        hf = self.hf

        if version is None:

            props = self.lookup_file(account, container, name, authorize=2)

        else:

            props = hf.file_get_properties(version)

        serial = props[SERIAL]

        size = props[SIZE]

        hashes = [hexlify(h) for h in hf.map_retr(serial)]

        return size, hashes

    @backend_method

    def update_object_hashmap(self, user, account, container,

                              name, size, hashmap, meta=None,

                              replace_meta=0, permissions=None):

        hf = self.hf

        props = self.lookup_file(account, container, name,

                                 create=1, size=size, authorize=1)

        serial = props[SERIAL]

        if size != props[SIZE]:

            hf.file_increment(serial, size)

        hf.map_stor(serial, (unhexlify(h) for h in hashmap))

        if replace_meta:

            hf.file_del_attributes(serial)

        if meta:

            hf.file_set_attributes(serial, meta.iteritems())

    @backend_method

    def copy_object(self, user, account, src_container, src_name,

                    dest_container, dest_name, dest_meta={},

                    replace_meta=False, permissions=None, src_version=None):

        hf = self.hf

        #FIXME: authorization

        if src_version:

            props = hf.file_get_properties(src_version)

        else:

            acc = self.get_account(account)

            srccont = self.get_container(acc, src_container)

            props = hf.file_lookup(srccont, src_name)

        if props is None:

            raise NameError

        srcobj = props[SERIAL]

        dstcont = self.get_container(acc, dest_container)

        serial, mtime = hf.file_copy(srcobj, dstcont, dest_name)

        if replace_meta:

            hf.file_del_attributes(serial)

        hf.file_set_attributes(serial, dest_meta.iteritems())

    @backend_method

    def move_object(self, user, account, src_container, src_name,

                    dest_container, dest_name, dest_meta={},

                    replace_meta=False, permissions=None, src_version=None):

        hf = self.hf

        #FIXME: authorization

        if src_version:

            props = hf.file_get_properties(src_version)

        else:

            acc = self.get_account(account)

            srccont = self.get_container(acc, src_container)

            props = hf.file_lookup(srccont, src_name)

        if props is None:

            raise NameError

        srcobj = props[SERIAL]

        dstcont = self.get_container(acc, dest_container)

        serial, mtime = hf.file_copy(srcobj, dstcont, dest_name)

        if replace_meta:

            hf.file_del_attributes(serial)

        hf.file_set_attributes(serial, dest_meta.iteritems())

        hf.file_remove(srcobj)

    @backend_method

    def delete_object(self, user, account, container, name):

        if user != account:

            raise NotAllowedError

        props = self.lookup_file(account, container, name, authorize=WRITE)

        serial = props[SERIAL]

        hf = self.hf

        #XXX: see delete_container, for empty-trashed ambiguty

        #     for now, we will just remove the object

        hf.file_remove(serial)

        #hf.file_recluster(serial, TRASH_CLUSTER)

        #XXX: Where to unset sharing? on file trash or on trash purge?

        #     We'll do it on file trash, but should we trash all versions?

        hf.xfeature_disinherit("%s/%s/%s" % (account, container, name))

#     @backend_method

#     def purge_trash(self, user, acount, container):

#         if user != account:

#             raise NotAllowedError

# 

#         serial = self.get_account(account)

#         serial = self.get_container(serial, container)

#         self.hf.file_purge_cluster(serial, cluster=TRASH_CLUSTER)

    @backend_method(autocommit=0)

    def get_block(self, blkhash):

        blocks = self.hf.block_retr((unhexlify(blkhash),))

        if not blocks:

            raise NameError

        return blocks[0]

    @backend_method(autocommit=0)

    def put_block(self, data):

        hashes, absent = self.hf.block_stor((data,))

        return hexlify(hashes[0])

    @backend_method(autocommit=0)

    def update_block(self, blkhash, data, offset=0):

        if offset == 0 and len(data) == self.block_size:

            return self.put_block(data)

        h, e = self.hf.block_delta(unhexlify(blkhash), ((offset, data),))

        return hexlify(h)