code.grnet.gr Git - pithos/blob - pithos/backends/simple.py

   1 # Copyright 2011 GRNET S.A. All rights reserved.
   2 #
   3 # Redistribution and use in source and binary forms, with or
   4 # without modification, are permitted provided that the following
   5 # conditions are met:
   6 #
   7 #   1. Redistributions of source code must retain the above
   8 #      copyright notice, this list of conditions and the following
   9 #      disclaimer.
  10 #
  11 #   2. Redistributions in binary form must reproduce the above
  12 #      copyright notice, this list of conditions and the following
  13 #      disclaimer in the documentation and/or other materials
  14 #      provided with the distribution.
  15 #
  16 # THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  17 # OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  18 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  19 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  20 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  21 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  22 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  23 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  24 # AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  25 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  26 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  27 # POSSIBILITY OF SUCH DAMAGE.
  28 #
  29 # The views and conclusions contained in the software and
  30 # documentation are those of the authors and should not be
  31 # interpreted as representing official policies, either expressed
  32 # or implied, of GRNET S.A.
  33
  34 import os
  35 import time
  36 import sqlite3
  37 import logging
  38 import hashlib
  39 import binascii
  40
  41 from base import NotAllowedError, BaseBackend
  42 from lib.hashfiler import Mapper, Blocker
  43 from django.utils.encoding import smart_unicode, smart_str
  44
  45 logger = logging.getLogger(__name__)
  46
  47 def backend_method(func=None, autocommit=1):
  48     if func is None:
  49         def fn(func):
  50             return backend_method(func, autocommit)
  51         return fn
  52
  53     if not autocommit:
  54         return func
  55     def fn(self, *args, **kw):
  56         self.con.execute('begin deferred')
  57         try:
  58             ret = func(self, *args, **kw)
  59             self.con.commit()
  60             return ret
  61         except:
  62             self.con.rollback()
  63             raise
  64     return fn
  65
  66
  67 class SimpleBackend(BaseBackend):
  68     """A simple backend.
  69
  70     Uses SQLite for storage.
  71     """
  72
  73     # TODO: Create account if not present in all functions.
  74
  75     def __init__(self, db, db_options):
  76         self.hash_algorithm = 'sha256'
  77         self.block_size = 4 * 1024 * 1024 # 4MB
  78
  79         self.default_policy = {'quota': 0, 'versioning': 'auto'}
  80
  81         basepath = os.path.split(db)[0]
  82         if basepath and not os.path.exists(basepath):
  83             os.makedirs(basepath)
  84         if not os.path.isdir(basepath):
  85             raise RuntimeError("Cannot open database at '%s'" % (db,))
  86
  87         self.con = sqlite3.connect(basepath + '/db', check_same_thread=False)
  88
  89         sql = '''pragma foreign_keys = on'''
  90         self.con.execute(sql)
  91
  92         sql = '''create table if not exists versions (
  93                     version_id integer primary key,
  94                     name text,
  95                     user text,
  96                     tstamp integer not null,
  97                     size integer default 0,
  98                     hide integer default 0)'''
  99         self.con.execute(sql)
 100         sql = '''create table if not exists metadata (
 101                     version_id integer,
 102                     key text,
 103                     value text,
 104                     primary key (version_id, key)
 105                     foreign key (version_id) references versions(version_id)
 106                     on delete cascade)'''
 107         self.con.execute(sql)
 108         sql = '''create table if not exists policy (
 109                     name text, key text, value text, primary key (name, key))'''
 110         self.con.execute(sql)
 111
 112         # Access control tables.
 113         sql = '''create table if not exists groups (
 114                     account text, gname text, user text)'''
 115         self.con.execute(sql)
 116         sql = '''create table if not exists permissions (
 117                     name text, op text, user text)'''
 118         self.con.execute(sql)
 119         sql = '''create table if not exists public (
 120                     name text, primary key (name))'''
 121         self.con.execute(sql)
 122
 123         self.con.commit()
 124
 125         params = {'blocksize': self.block_size,
 126                   'blockpath': basepath + '/blocks',
 127                   'hashtype': self.hash_algorithm}
 128         self.blocker = Blocker(**params)
 129
 130         params = {'mappath': basepath + '/maps',
 131                   'namelen': self.blocker.hashlen}
 132         self.mapper = Mapper(**params)
 133
 134     @backend_method
 135     def list_accounts(self, user, marker=None, limit=10000):
 136         """Return a list of accounts the user can access."""
 137
 138         allowed = self._allowed_accounts(user)
 139         start, limit = self._list_limits(allowed, marker, limit)
 140         return allowed[start:start + limit]
 141
 142     @backend_method
 143     def get_account_meta(self, user, account, until=None):
 144         """Return a dictionary with the account metadata."""
 145
 146         logger.debug("get_account_meta: %s %s", account, until)
 147         if user != account:
 148             if until or account not in self._allowed_accounts(user):
 149                 raise NotAllowedError
 150         else:
 151             self._create_account(user, account)
 152         try:
 153             version_id, mtime = self._get_accountinfo(account, until)
 154         except NameError:
 155             # Account does not exist before until.
 156             version_id = None
 157             mtime = until
 158         count, bytes, tstamp = self._get_pathstats(account, until)
 159         if mtime > tstamp:
 160             tstamp = mtime
 161         if until is None:
 162             modified = tstamp
 163         else:
 164             modified = self._get_pathstats(account)[2] # Overall last modification
 165             if mtime > modified:
 166                 modified = mtime
 167
 168         # Proper count.
 169         sql = 'select count(name) from (%s) where name glob ? and not name glob ?'
 170         sql = sql % self._sql_until(until)
 171         c = self.con.execute(sql, (account + '/*', account + '/*/*'))
 172         row = c.fetchone()
 173         count = row[0]
 174
 175         if user != account:
 176             meta = {'name': account}
 177         else:
 178             meta = self._get_metadata(account, version_id)
 179             meta.update({'name': account, 'count': count, 'bytes': bytes})
 180             if until is not None:
 181                 meta.update({'until_timestamp': tstamp})
 182         meta.update({'modified': modified})
 183         return meta
 184
 185     @backend_method
 186     def update_account_meta(self, user, account, meta, replace=False):
 187         """Update the metadata associated with the account."""
 188
 189         logger.debug("update_account_meta: %s %s %s", account, meta, replace)
 190         if user != account:
 191             raise NotAllowedError
 192         self._put_metadata(user, account, meta, replace, False)
 193
 194     @backend_method
 195     def get_account_groups(self, user, account):
 196         """Return a dictionary with the user groups defined for this account."""
 197
 198         logger.debug("get_account_groups: %s", account)
 199         if user != account:
 200             if account not in self._allowed_accounts(user):
 201                 raise NotAllowedError
 202             return {}
 203         self._create_account(user, account)
 204         return self._get_groups(account)
 205
 206     @backend_method
 207     def update_account_groups(self, user, account, groups, replace=False):
 208         """Update the groups associated with the account."""
 209
 210         logger.debug("update_account_groups: %s %s %s", account, groups, replace)
 211         if user != account:
 212             raise NotAllowedError
 213         self._create_account(user, account)
 214         self._check_groups(groups)
 215         self._put_groups(account, groups, replace)
 216
 217     @backend_method
 218     def put_account(self, user, account):
 219         """Create a new account with the given name."""
 220
 221         logger.debug("put_account: %s", account)
 222         if user != account:
 223             raise NotAllowedError
 224         try:
 225             version_id, mtime = self._get_accountinfo(account)
 226         except NameError:
 227             pass
 228         else:
 229             raise NameError('Account already exists')
 230         self._put_version(account, user)
 231
 232     @backend_method
 233     def delete_account(self, user, account):
 234         """Delete the account with the given name."""
 235
 236         logger.debug("delete_account: %s", account)
 237         if user != account:
 238             raise NotAllowedError
 239         count = self._get_pathstats(account)[0]
 240         if count > 0:
 241             raise IndexError('Account is not empty')
 242         sql = 'delete from versions where name = ?'
 243         self.con.execute(sql, (account,))
 244         self._del_groups(account)
 245
 246     @backend_method
 247     def list_containers(self, user, account, marker=None, limit=10000, shared=False, until=None):
 248         """Return a list of containers existing under an account."""
 249
 250         logger.debug("list_containers: %s %s %s %s", account, marker, limit, until)
 251         if user != account:
 252             if until or account not in self._allowed_accounts(user):
 253                 raise NotAllowedError
 254             allowed = self._allowed_containers(user, account)
 255             start, limit = self._list_limits(allowed, marker, limit)
 256             return allowed[start:start + limit]
 257         else:
 258             if shared:
 259                 allowed = [x.split('/', 2)[1] for x in self._shared_paths(account)]
 260                 start, limit = self._list_limits(allowed, marker, limit)
 261                 return allowed[start:start + limit]
 262         return [x[0] for x in self._list_objects(account, '', '/', marker, limit, False, [], until)]
 263
 264     @backend_method
 265     def get_container_meta(self, user, account, container, until=None):
 266         """Return a dictionary with the container metadata."""
 267
 268         logger.debug("get_container_meta: %s %s %s", account, container, until)
 269         if user != account:
 270             if until or container not in self._allowed_containers(user, account):
 271                 raise NotAllowedError
 272         path, version_id, mtime = self._get_containerinfo(account, container, until)
 273         count, bytes, tstamp = self._get_pathstats(path, until)
 274         if mtime > tstamp:
 275             tstamp = mtime
 276         if until is None:
 277             modified = tstamp
 278         else:
 279             modified = self._get_pathstats(path)[2] # Overall last modification
 280             if mtime > modified:
 281                 modified = mtime
 282
 283         if user != account:
 284             meta = {'name': container, 'modified': modified}
 285         else:
 286             meta = self._get_metadata(path, version_id)
 287             meta.update({'name': container, 'count': count, 'bytes': bytes, 'modified': modified})
 288             if until is not None:
 289                 meta.update({'until_timestamp': tstamp})
 290         return meta
 291
 292     @backend_method
 293     def update_container_meta(self, user, account, container, meta, replace=False):
 294         """Update the metadata associated with the container."""
 295
 296         logger.debug("update_container_meta: %s %s %s %s", account, container, meta, replace)
 297         if user != account:
 298             raise NotAllowedError
 299         path, version_id, mtime = self._get_containerinfo(account, container)
 300         self._put_metadata(user, path, meta, replace, False)
 301
 302     @backend_method
 303     def get_container_policy(self, user, account, container):
 304         """Return a dictionary with the container policy."""
 305
 306         logger.debug("get_container_policy: %s %s", account, container)
 307         if user != account:
 308             if container not in self._allowed_containers(user, account):
 309                 raise NotAllowedError
 310             return {}
 311         path = self._get_containerinfo(account, container)[0]
 312         return self._get_policy(path)
 313
 314     @backend_method
 315     def update_container_policy(self, user, account, container, policy, replace=False):
 316         """Update the policy associated with the account."""
 317
 318         logger.debug("update_container_policy: %s %s %s %s", account, container, policy, replace)
 319         if user != account:
 320             raise NotAllowedError
 321         path = self._get_containerinfo(account, container)[0]
 322         self._check_policy(policy)
 323         if replace:
 324             for k, v in self.default_policy.iteritems():
 325                 if k not in policy:
 326                     policy[k] = v
 327         for k, v in policy.iteritems():
 328             sql = 'insert or replace into policy (name, key, value) values (?, ?, ?)'
 329             self.con.execute(sql, (path, k, v))
 330
 331     @backend_method
 332     def put_container(self, user, account, container, policy=None):
 333         """Create a new container with the given name."""
 334
 335         logger.debug("put_container: %s %s %s", account, container, policy)
 336         if user != account:
 337             raise NotAllowedError
 338         try:
 339             path, version_id, mtime = self._get_containerinfo(account, container)
 340         except NameError:
 341             pass
 342         else:
 343             raise NameError('Container already exists')
 344         if policy:
 345             self._check_policy(policy)
 346         path = '/'.join((account, container))
 347         version_id = self._put_version(path, user)[0]
 348         for k, v in self.default_policy.iteritems():
 349             if k not in policy:
 350                 policy[k] = v
 351         for k, v in policy.iteritems():
 352             sql = 'insert or replace into policy (name, key, value) values (?, ?, ?)'
 353             self.con.execute(sql, (path, k, v))
 354
 355     @backend_method
 356     def delete_container(self, user, account, container, until=None):
 357         """Delete/purge the container with the given name."""
 358
 359         logger.debug("delete_container: %s %s %s", account, container, until)
 360         if user != account:
 361             raise NotAllowedError
 362         path, version_id, mtime = self._get_containerinfo(account, container)
 363
 364         if until is not None:
 365             sql = '''select version_id from versions where name like ? and tstamp <= ?
 366                         and version_id not in (select version_id from (%s))'''
 367             sql = sql % self._sql_until() # Do not delete current versions.
 368             c = self.con.execute(sql, (path + '/%', until))
 369             for v in [x[0] for x in c.fetchall()]:
 370                 self._del_version(v)
 371             return
 372
 373         count = self._get_pathstats(path)[0]
 374         if count > 0:
 375             raise IndexError('Container is not empty')
 376         sql = 'delete from versions where name = ? or name like ?' # May contain hidden items.
 377         self.con.execute(sql, (path, path + '/%',))
 378         sql = 'delete from policy where name = ?'
 379         self.con.execute(sql, (path,))
 380         self._copy_version(user, account, account, True, False) # New account version (for timestamp update).
 381
 382     @backend_method
 383     def list_objects(self, user, account, container, prefix='', delimiter=None, marker=None, limit=10000, virtual=True, keys=[], shared=False, until=None):
 384         """Return a list of objects existing under a container."""
 385
 386         logger.debug("list_objects: %s %s %s %s %s %s %s %s %s %s", account, container, prefix, delimiter, marker, limit, virtual, keys, shared, until)
 387         allowed = []
 388         if user != account:
 389             if until:
 390                 raise NotAllowedError
 391             allowed = self._allowed_paths(user, '/'.join((account, container)))
 392             if not allowed:
 393                 raise NotAllowedError
 394         else:
 395             if shared:
 396                 allowed = self._shared_paths('/'.join((account, container)))
 397         path, version_id, mtime = self._get_containerinfo(account, container, until)
 398         return self._list_objects(path, prefix, delimiter, marker, limit, virtual, keys, until, allowed)
 399
 400     @backend_method
 401     def list_object_meta(self, user, account, container, until=None):
 402         """Return a list with all the container's object meta keys."""
 403
 404         logger.debug("list_object_meta: %s %s %s", account, container, until)
 405         allowed = []
 406         if user != account:
 407             if until:
 408                 raise NotAllowedError
 409             allowed = self._allowed_paths(user, '/'.join((account, container)))
 410             if not allowed:
 411                 raise NotAllowedError
 412         path, version_id, mtime = self._get_containerinfo(account, container, until)
 413         sql = '''select distinct m.key from (%s) o, metadata m
 414                     where m.version_id = o.version_id and o.name like ?'''
 415         sql = sql % self._sql_until(until)
 416         param = (path + '/%',)
 417         if allowed:
 418             for x in allowed:
 419                 sql += ' and o.name like ?'
 420                 param += (x,)
 421         c = self.con.execute(sql, param)
 422         return [x[0] for x in c.fetchall()]
 423
 424     @backend_method
 425     def get_object_meta(self, user, account, container, name, version=None):
 426         """Return a dictionary with the object metadata."""
 427
 428         logger.debug("get_object_meta: %s %s %s %s", account, container, name, version)
 429         self._can_read(user, account, container, name)
 430         path, version_id, muser, mtime, size = self._get_objectinfo(account, container, name, version)
 431         if version is None:
 432             modified = mtime
 433         else:
 434             modified = self._get_version(path, version)[2] # Overall last modification
 435
 436         meta = self._get_metadata(path, version_id)
 437         meta.update({'name': name, 'bytes': size})
 438         meta.update({'version': version_id, 'version_timestamp': mtime})
 439         meta.update({'modified': modified, 'modified_by': muser})
 440         return meta
 441
 442     @backend_method
 443     def update_object_meta(self, user, account, container, name, meta, replace=False):
 444         """Update the metadata associated with the object."""
 445
 446         logger.debug("update_object_meta: %s %s %s %s %s", account, container, name, meta, replace)
 447         self._can_write(user, account, container, name)
 448         path, version_id, muser, mtime, size = self._get_objectinfo(account, container, name)
 449         return self._put_metadata(user, path, meta, replace)
 450
 451     @backend_method
 452     def get_object_permissions(self, user, account, container, name):
 453         """Return the path from which this object gets its permissions from,\
 454         along with a dictionary containing the permissions."""
 455
 456         logger.debug("get_object_permissions: %s %s %s", account, container, name)
 457         self._can_read(user, account, container, name)
 458         path = self._get_objectinfo(account, container, name)[0]
 459         return self._get_permissions(path)
 460
 461     @backend_method
 462     def update_object_permissions(self, user, account, container, name, permissions):
 463         """Update the permissions associated with the object."""
 464
 465         logger.debug("update_object_permissions: %s %s %s %s", account, container, name, permissions)
 466         if user != account:
 467             raise NotAllowedError
 468         path = self._get_objectinfo(account, container, name)[0]
 469         r, w = self._check_permissions(path, permissions)
 470         self._put_permissions(path, r, w)
 471
 472     @backend_method
 473     def get_object_public(self, user, account, container, name):
 474         """Return the public URL of the object if applicable."""
 475
 476         logger.debug("get_object_public: %s %s %s", account, container, name)
 477         self._can_read(user, account, container, name)
 478         path = self._get_objectinfo(account, container, name)[0]
 479         if self._get_public(path):
 480             return '/public/' + path
 481         return None
 482
 483     @backend_method
 484     def update_object_public(self, user, account, container, name, public):
 485         """Update the public status of the object."""
 486
 487         logger.debug("update_object_public: %s %s %s %s", account, container, name, public)
 488         self._can_write(user, account, container, name)
 489         path = self._get_objectinfo(account, container, name)[0]
 490         self._put_public(path, public)
 491
 492     @backend_method
 493     def get_object_hashmap(self, user, account, container, name, version=None):
 494         """Return the object's size and a list with partial hashes."""
 495
 496         logger.debug("get_object_hashmap: %s %s %s %s", account, container, name, version)
 497         self._can_read(user, account, container, name)
 498         path, version_id, muser, mtime, size = self._get_objectinfo(account, container, name, version)
 499         hashmap = self.mapper.map_retr(version_id)
 500         return size, [binascii.hexlify(x) for x in hashmap]
 501
 502     @backend_method
 503     def update_object_hashmap(self, user, account, container, name, size, hashmap, meta={}, replace_meta=False, permissions=None):
 504         """Create/update an object with the specified size and partial hashes."""
 505
 506         logger.debug("update_object_hashmap: %s %s %s %s %s", account, container, name, size, hashmap)
 507         if permissions is not None and user != account:
 508             raise NotAllowedError
 509         self._can_write(user, account, container, name)
 510         missing = self.blocker.block_ping([binascii.unhexlify(x) for x in hashmap])
 511         if missing:
 512             ie = IndexError()
 513             ie.data = missing
 514             raise ie
 515         path = self._get_containerinfo(account, container)[0]
 516         path = '/'.join((path, name))
 517         if permissions is not None:
 518             r, w = self._check_permissions(path, permissions)
 519         src_version_id, dest_version_id = self._copy_version(user, path, path, not replace_meta, False)
 520         sql = 'update versions set size = ? where version_id = ?'
 521         self.con.execute(sql, (size, dest_version_id))
 522         self.mapper.map_stor(dest_version_id, [binascii.unhexlify(x) for x in hashmap])
 523         for k, v in meta.iteritems():
 524             sql = 'insert or replace into metadata (version_id, key, value) values (?, ?, ?)'
 525             self.con.execute(sql, (dest_version_id, k, v))
 526         if permissions is not None:
 527             self._put_permissions(path, r, w)
 528         return dest_version_id
 529
 530     @backend_method
 531     def copy_object(self, user, account, src_container, src_name, dest_container, dest_name, dest_meta={}, replace_meta=False, permissions=None, src_version=None):
 532         """Copy an object's data and metadata."""
 533
 534         logger.debug("copy_object: %s %s %s %s %s %s %s %s %s", account, src_container, src_name, dest_container, dest_name, dest_meta, replace_meta, permissions, src_version)
 535         if permissions is not None and user != account:
 536             raise NotAllowedError
 537         self._can_read(user, account, src_container, src_name)
 538         self._can_write(user, account, dest_container, dest_name)
 539         self._get_containerinfo(account, src_container)
 540         if src_version is None:
 541             src_path = self._get_objectinfo(account, src_container, src_name)[0]
 542         else:
 543             src_path = '/'.join((account, src_container, src_name))
 544         dest_path = self._get_containerinfo(account, dest_container)[0]
 545         dest_path = '/'.join((dest_path, dest_name))
 546         if permissions is not None:
 547             r, w = self._check_permissions(dest_path, permissions)
 548         src_version_id, dest_version_id = self._copy_version(user, src_path, dest_path, not replace_meta, True, src_version)
 549         for k, v in dest_meta.iteritems():
 550             sql = 'insert or replace into metadata (version_id, key, value) values (?, ?, ?)'
 551             self.con.execute(sql, (dest_version_id, k, v))
 552         if permissions is not None:
 553             self._put_permissions(dest_path, r, w)
 554         return dest_version_id
 555
 556     @backend_method
 557     def move_object(self, user, account, src_container, src_name, dest_container, dest_name, dest_meta={}, replace_meta=False, permissions=None):
 558         """Move an object's data and metadata."""
 559
 560         logger.debug("move_object: %s %s %s %s %s %s %s %s", account, src_container, src_name, dest_container, dest_name, dest_meta, replace_meta, permissions)
 561         dest_version_id = self.copy_object(user, account, src_container, src_name, dest_container, dest_name, dest_meta, replace_meta, permissions, None)
 562         self.delete_object(user, account, src_container, src_name)
 563         return dest_version_id
 564
 565     @backend_method
 566     def delete_object(self, user, account, container, name, until=None):
 567         """Delete/purge an object."""
 568
 569         logger.debug("delete_object: %s %s %s %s", account, container, name, until)
 570         if user != account:
 571             raise NotAllowedError
 572
 573         if until is not None:
 574             path = '/'.join((account, container, name))
 575             sql = '''select version_id from versions where name = ? and tstamp <= ?'''
 576             c = self.con.execute(sql, (path, until))
 577             for v in [x[0] in c.fetchall()]:
 578                 self._del_version(v)
 579             try:
 580                 version_id = self._get_version(path)[0]
 581             except NameError:
 582                 pass
 583             else:
 584                 self._del_sharing(path)
 585             return
 586
 587         path = self._get_objectinfo(account, container, name)[0]
 588         self._put_version(path, user, 0, 1)
 589         self._del_sharing(path)
 590
 591     @backend_method
 592     def list_versions(self, user, account, container, name):
 593         """Return a list of all (version, version_timestamp) tuples for an object."""
 594
 595         logger.debug("list_versions: %s %s %s", account, container, name)
 596         self._can_read(user, account, container, name)
 597         # This will even show deleted versions.
 598         path = '/'.join((account, container, name))
 599         sql = '''select distinct version_id, tstamp from versions where name = ? and hide = 0'''
 600         c = self.con.execute(sql, (path,))
 601         return [(int(x[0]), int(x[1])) for x in c.fetchall()]
 602
 603     @backend_method(autocommit=0)
 604     def get_block(self, hash):
 605         """Return a block's data."""
 606
 607         logger.debug("get_block: %s", hash)
 608         blocks = self.blocker.block_retr((binascii.unhexlify(hash),))
 609         if not blocks:
 610             raise NameError('Block does not exist')
 611         return blocks[0]
 612
 613     @backend_method(autocommit=0)
 614     def put_block(self, data):
 615         """Store a block and return the hash."""
 616
 617         logger.debug("put_block: %s", len(data))
 618         hashes, absent = self.blocker.block_stor((data,))
 619         return binascii.hexlify(hashes[0])
 620
 621     @backend_method(autocommit=0)
 622     def update_block(self, hash, data, offset=0):
 623         """Update a known block and return the hash."""
 624
 625         logger.debug("update_block: %s %s %s", hash, len(data), offset)
 626         if offset == 0 and len(data) == self.block_size:
 627             return self.put_block(data)
 628         h, e = self.blocker.block_delta(binascii.unhexlify(hash), ((offset, data),))
 629         return binascii.hexlify(h)
 630
 631     def _sql_until(self, until=None):
 632         """Return the sql to get the latest versions until the timestamp given."""
 633         if until is None:
 634             until = int(time.time())
 635         sql = '''select version_id, name, tstamp, size from versions v
 636                     where version_id = (select max(version_id) from versions
 637                                         where v.name = name and tstamp <= %s)
 638                     and hide = 0'''
 639         return sql % (until,)
 640
 641     def _get_pathstats(self, path, until=None):
 642         """Return count and sum of size of everything under path and latest timestamp."""
 643
 644         sql = 'select count(version_id), total(size), max(tstamp) from (%s) where name like ?'
 645         sql = sql % self._sql_until(until)
 646         c = self.con.execute(sql, (path + '/%',))
 647         row = c.fetchone()
 648         tstamp = row[2] if row[2] is not None else 0
 649         return int(row[0]), int(row[1]), int(tstamp)
 650
 651     def _get_version(self, path, version=None):
 652         if version is None:
 653             sql = '''select version_id, user, tstamp, size, hide from versions where name = ?
 654                         order by version_id desc limit 1'''
 655             c = self.con.execute(sql, (path,))
 656             row = c.fetchone()
 657             if not row or int(row[4]):
 658                 raise NameError('Object does not exist')
 659         else:
 660             # The database (sqlite) will not complain if the version is not an integer.
 661             sql = '''select version_id, user, tstamp, size from versions where name = ?
 662                         and version_id = ?'''
 663             c = self.con.execute(sql, (path, version))
 664             row = c.fetchone()
 665             if not row:
 666                 raise IndexError('Version does not exist')
 667         return smart_str(row[0]), smart_str(row[1]), int(row[2]), int(row[3])
 668
 669     def _put_version(self, path, user, size=0, hide=0):
 670         tstamp = int(time.time())
 671         sql = 'insert into versions (name, user, tstamp, size, hide) values (?, ?, ?, ?, ?)'
 672         id = self.con.execute(sql, (path, user, tstamp, size, hide)).lastrowid
 673         return str(id), tstamp
 674
 675     def _copy_version(self, user, src_path, dest_path, copy_meta=True, copy_data=True, src_version=None):
 676         if src_version is not None:
 677             src_version_id, muser, mtime, size = self._get_version(src_path, src_version)
 678         else:
 679             # Latest or create from scratch.
 680             try:
 681                 src_version_id, muser, mtime, size = self._get_version(src_path)
 682             except NameError:
 683                 src_version_id = None
 684                 size = 0
 685         if not copy_data:
 686             size = 0
 687         dest_version_id = self._put_version(dest_path, user, size)[0]
 688         if copy_meta and src_version_id is not None:
 689             sql = 'insert into metadata select %s, key, value from metadata where version_id = ?'
 690             sql = sql % dest_version_id
 691             self.con.execute(sql, (src_version_id,))
 692         if copy_data and src_version_id is not None:
 693             # TODO: Copy properly.
 694             hashmap = self.mapper.map_retr(src_version_id)
 695             self.mapper.map_stor(dest_version_id, hashmap)
 696         return src_version_id, dest_version_id
 697
 698     def _get_versioninfo(self, account, container, name, until=None):
 699         """Return path, latest version, associated timestamp and size until the timestamp given."""
 700
 701         p = (account, container, name)
 702         try:
 703             p = p[:p.index(None)]
 704         except ValueError:
 705             pass
 706         path = '/'.join(p)
 707         sql = '''select version_id, tstamp, size from (%s) where name = ?'''
 708         sql = sql % self._sql_until(until)
 709         c = self.con.execute(sql, (path,))
 710         row = c.fetchone()
 711         if row is None:
 712             raise NameError('Path does not exist')
 713         return path, str(row[0]), int(row[1]), int(row[2])
 714
 715     def _get_accountinfo(self, account, until=None):
 716         try:
 717             path, version_id, mtime, size = self._get_versioninfo(account, None, None, until)
 718             return version_id, mtime
 719         except:
 720             raise NameError('Account does not exist')
 721
 722     def _get_containerinfo(self, account, container, until=None):
 723         try:
 724             path, version_id, mtime, size = self._get_versioninfo(account, container, None, until)
 725             return path, version_id, mtime
 726         except:
 727             raise NameError('Container does not exist')
 728
 729     def _get_objectinfo(self, account, container, name, version=None):
 730         path = '/'.join((account, container, name))
 731         version_id, muser, mtime, size = self._get_version(path, version)
 732         return path, version_id, muser, mtime, size
 733
 734     def _create_account(self, user, account):
 735         try:
 736             self._get_accountinfo(account)
 737         except NameError:
 738             self._put_version(account, user)
 739
 740     def _get_metadata(self, path, version):
 741         sql = 'select key, value from metadata where version_id = ?'
 742         c = self.con.execute(sql, (version,))
 743         return dict(c.fetchall())
 744
 745     def _put_metadata(self, user, path, meta, replace=False, copy_data=True):
 746         """Create a new version and store metadata."""
 747
 748         src_version_id, dest_version_id = self._copy_version(user, path, path, not replace, copy_data)
 749         for k, v in meta.iteritems():
 750             if not replace and v == '':
 751                 sql = 'delete from metadata where version_id = ? and key = ?'
 752                 self.con.execute(sql, (dest_version_id, k))
 753             else:
 754                 sql = 'insert or replace into metadata (version_id, key, value) values (?, ?, ?)'
 755                 self.con.execute(sql, (dest_version_id, k, v))
 756         return dest_version_id
 757
 758     def _check_policy(self, policy):
 759         for k in policy.keys():
 760             if policy[k] == '':
 761                 policy[k] = self.default_policy.get(k)
 762         for k, v in policy.iteritems():
 763             if k == 'quota':
 764                 q = int(v) # May raise ValueError.
 765                 if q < 0:
 766                     raise ValueError
 767             elif k == 'versioning':
 768                 if v not in ['auto', 'manual', 'none']:
 769                     raise ValueError
 770             else:
 771                 raise ValueError
 772
 773     def _get_policy(self, path):
 774         sql = 'select key, value from policy where name = ?'
 775         c = self.con.execute(sql, (path,))
 776         return dict(c.fetchall())
 777
 778     def _list_limits(self, listing, marker, limit):
 779         start = 0
 780         if marker:
 781             try:
 782                 start = listing.index(marker) + 1
 783             except ValueError:
 784                 pass
 785         if not limit or limit > 10000:
 786             limit = 10000
 787         return start, limit
 788
 789     def _list_objects(self, path, prefix='', delimiter=None, marker=None, limit=10000, virtual=True, keys=[], until=None, allowed=[]):
 790         cont_prefix = path + '/'
 791         if keys and len(keys) > 0:
 792             sql = '''select distinct o.name, o.version_id from (%s) o, metadata m where o.name like ? and
 793                         m.version_id = o.version_id and m.key in (%s)'''
 794             sql = sql % (self._sql_until(until), ', '.join('?' * len(keys)))
 795             param = (cont_prefix + prefix + '%',) + tuple(keys)
 796             if allowed:
 797                 sql += ' and (' + ' or '.join(('o.name like ?',) * len(allowed)) + ')'
 798                 param += tuple([x + '%' for x in allowed])
 799             sql += ' order by o.name'
 800         else:
 801             sql = 'select name, version_id from (%s) where name like ?'
 802             sql = sql % self._sql_until(until)
 803             param = (cont_prefix + prefix + '%',)
 804             if allowed:
 805                 sql += ' and (' + ' or '.join(('name like ?',) * len(allowed)) + ')'
 806                 param += tuple([x + '%' for x in allowed])
 807             sql += ' order by name'
 808         c = self.con.execute(sql, param)
 809         objects = [(x[0][len(cont_prefix):], x[1]) for x in c.fetchall()]
 810         if delimiter:
 811             pseudo_objects = []
 812             for x in objects:
 813                 pseudo_name = x[0]
 814                 i = pseudo_name.find(delimiter, len(prefix))
 815                 if not virtual:
 816                     # If the delimiter is not found, or the name ends
 817                     # with the delimiter's first occurence.
 818                     if i == -1 or len(pseudo_name) == i + len(delimiter):
 819                         pseudo_objects.append(x)
 820                 else:
 821                     # If the delimiter is found, keep up to (and including) the delimiter.
 822                     if i != -1:
 823                         pseudo_name = pseudo_name[:i + len(delimiter)]
 824                     if pseudo_name not in [y[0] for y in pseudo_objects]:
 825                         if pseudo_name == x[0]:
 826                             pseudo_objects.append(x)
 827                         else:
 828                             pseudo_objects.append((pseudo_name, None))
 829             objects = pseudo_objects
 830
 831         start, limit = self._list_limits([x[0] for x in objects], marker, limit)
 832         return objects[start:start + limit]
 833
 834     def _del_version(self, version):
 835         self.mapper.map_remv(version)
 836         sql = 'delete from versions where version_id = ?'
 837         self.con.execute(sql, (version,))
 838
 839     # Access control functions.
 840
 841     def _check_groups(self, groups):
 842         # Example follows.
 843         # for k, v in groups.iteritems():
 844         #     if True in [False or ',' in x for x in v]:
 845         #         raise ValueError('Bad characters in groups')
 846         pass
 847
 848     def _get_groups(self, account):
 849         sql = 'select gname, user from groups where account = ?'
 850         c = self.con.execute(sql, (account,))
 851         groups = {}
 852         for gname, user in c.fetchall():
 853             if gname not in groups:
 854                 groups[gname] = []
 855             groups[gname].append(user)
 856         return groups
 857
 858     def _put_groups(self, account, groups, replace=False):
 859         if replace:
 860             self._del_groups(account)
 861         for k, v in groups.iteritems():
 862             sql = 'delete from groups where account = ? and gname = ?'
 863             self.con.execute(sql, (account, k))
 864             if v:
 865                 sql = 'insert into groups (account, gname, user) values (?, ?, ?)'
 866                 self.con.executemany(sql, [(account, k, x) for x in v])
 867
 868     def _del_groups(self, account):
 869         sql = 'delete from groups where account = ?'
 870         self.con.execute(sql, (account,))
 871
 872     def _check_permissions(self, path, permissions):
 873         # Check for existing permissions.
 874         sql = '''select name from permissions
 875                     where name != ? and (name like ? or ? like name || ?)'''
 876         c = self.con.execute(sql, (path, path + '%', path, '%'))
 877         rows = c.fetchall()
 878         if rows:
 879             ae = AttributeError()
 880             ae.data = rows
 881             raise ae
 882
 883         # Format given permissions.
 884         if len(permissions) == 0:
 885             return [], []
 886         r = permissions.get('read', [])
 887         w = permissions.get('write', [])
 888         # Examples follow.
 889         # if True in [False or ',' in x for x in r]:
 890         #     raise ValueError('Bad characters in read permissions')
 891         # if True in [False or ',' in x for x in w]:
 892         #     raise ValueError('Bad characters in write permissions')
 893         return r, w
 894
 895     def _get_permissions(self, path):
 896         # Check for permissions at path or above.
 897         sql = 'select name, op, user from permissions where ? like name || ?'
 898         c = self.con.execute(sql, (path, '%'))
 899         name = path
 900         perms = {} # Return nothing, if nothing is set.
 901         for row in c.fetchall():
 902             name = row[0]
 903             op = row[1]
 904             user = row[2]
 905             if op not in perms:
 906                 perms[op] = []
 907             perms[op].append(user)
 908         return name, perms
 909
 910     def _put_permissions(self, path, r, w):
 911         sql = 'delete from permissions where name = ?'
 912         self.con.execute(sql, (path,))
 913         sql = 'insert into permissions (name, op, user) values (?, ?, ?)'
 914         if r:
 915             self.con.executemany(sql, [(path, 'read', x) for x in r])
 916         if w:
 917             self.con.executemany(sql, [(path, 'write', x) for x in w])
 918
 919     def _get_public(self, path):
 920         sql = 'select name from public where name = ?'
 921         c = self.con.execute(sql, (path,))
 922         row = c.fetchone()
 923         if not row:
 924             return False
 925         return True
 926
 927     def _put_public(self, path, public):
 928         if not public:
 929             sql = 'delete from public where name = ?'
 930         else:
 931             sql = 'insert or replace into public (name) values (?)'
 932         self.con.execute(sql, (path,))
 933
 934     def _del_sharing(self, path):
 935         sql = 'delete from permissions where name = ?'
 936         self.con.execute(sql, (path,))
 937         sql = 'delete from public where name = ?'
 938         self.con.execute(sql, (path,))
 939
 940     def _is_allowed(self, user, account, container, name, op='read'):
 941         if smart_unicode(user) == smart_unicode(account):
 942             return True
 943         path = '/'.join((account, container, name))
 944         if op == 'read' and self._get_public(path):
 945             return True
 946         perm_path, perms = self._get_permissions(path)
 947
 948         # Expand groups.
 949         for x in ('read', 'write'):
 950             g_perms = set()
 951             for y in perms.get(x, []):
 952                 if ':' in y:
 953                     g_account, g_name = y.split(':', 1)
 954                     groups = self._get_groups(g_account)
 955                     if g_name in groups.keys():
 956                         g_perms.update(groups[g_name])
 957                 else:
 958                     g_perms.add(y)
 959             perms[x] = g_perms
 960
 961         user = smart_unicode(user, strings_only=True)
 962         if op == 'read' and ('*' in perms['read'] or user in perms['read']):
 963             return True
 964         if '*' in perms['write'] or user in perms['write']:
 965             return True
 966         return False
 967
 968     def _can_read(self, user, account, container, name):
 969         if not self._is_allowed(user, account, container, name, 'read'):
 970             raise NotAllowedError
 971
 972     def _can_write(self, user, account, container, name):
 973         if not self._is_allowed(user, account, container, name, 'write'):
 974             raise NotAllowedError
 975
 976     def _allowed_paths(self, user, prefix=None):
 977         sql = '''select distinct name from permissions where (user = ?
 978                     or user in (select account || ':' || gname from groups where user = ?))'''
 979         param = (user, user)
 980         if prefix:
 981             sql += ' and name like ?'
 982             param += (prefix + '/%',)
 983         c = self.con.execute(sql, param)
 984         return [x[0] for x in c.fetchall()]
 985
 986     def _allowed_accounts(self, user):
 987         allow = set()
 988         for path in self._allowed_paths(user):
 989             allow.add(path.split('/', 1)[0])
 990         return sorted(allow)
 991
 992     def _allowed_containers(self, user, account):
 993         allow = set()
 994         for path in self._allowed_paths(user, account):
 995             allow.add(path.split('/', 2)[1])
 996         return sorted(allow)
 997
 998     def _shared_paths(self, prefix):
 999         sql = 'select distinct name from permissions where name like ?'
1000         c = self.con.execute(sql, (prefix + '/%',))
1001         return [x[0] for x in c.fetchall()]