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 types
  39 import hashlib
  40 import shutil
  41 import pickle
  42
  43 from base import NotAllowedError, BaseBackend
  44
  45
  46 logger = logging.getLogger(__name__)
  47
  48
  49 class SimpleBackend(BaseBackend):
  50     """A simple backend.
  51
  52     Uses SQLite for storage.
  53     """
  54
  55     # TODO: Automatic/manual clean-up after a time interval.
  56
  57     def __init__(self, db):
  58         self.hash_algorithm = 'sha1'
  59         self.block_size = 128 * 1024 # 128KB
  60
  61         self.default_policy = {'quota': 0, 'versioning': 'auto'}
  62
  63         basepath = os.path.split(db)[0]
  64         if basepath and not os.path.exists(basepath):
  65             os.makedirs(basepath)
  66
  67         self.con = sqlite3.connect(db, check_same_thread=False)
  68
  69         sql = '''pragma foreign_keys = on'''
  70         self.con.execute(sql)
  71
  72         sql = '''create table if not exists versions (
  73                     version_id integer primary key,
  74                     name text,
  75                     user text,
  76                     tstamp datetime default current_timestamp,
  77                     size integer default 0,
  78                     hide integer default 0)'''
  79         self.con.execute(sql)
  80         sql = '''create table if not exists metadata (
  81                     version_id integer,
  82                     key text,
  83                     value text,
  84                     primary key (version_id, key)
  85                     foreign key (version_id) references versions(version_id)
  86                     on delete cascade)'''
  87         self.con.execute(sql)
  88         sql = '''create table if not exists hashmaps (
  89                     version_id integer,
  90                     pos integer,
  91                     block_id text,
  92                     primary key (version_id, pos)
  93                     foreign key (version_id) references versions(version_id)
  94                     on delete cascade)'''
  95         self.con.execute(sql)
  96         sql = '''create table if not exists blocks (
  97                     block_id text, data blob, primary key (block_id))'''
  98         self.con.execute(sql)
  99
 100         sql = '''create table if not exists policy (
 101                     name text, key text, value text, primary key (name, key))'''
 102         self.con.execute(sql)
 103
 104         sql = '''create table if not exists groups (
 105                     account text, name text, users text, primary key (account, name))'''
 106         self.con.execute(sql)
 107         sql = '''create table if not exists permissions (
 108                     name text, read text, write text, primary key (name))'''
 109         self.con.execute(sql)
 110         sql = '''create table if not exists public (
 111                     name text, primary key (name))'''
 112         self.con.execute(sql)
 113         self.con.commit()
 114
 115     def get_account_meta(self, user, account, until=None):
 116         """Return a dictionary with the account metadata."""
 117
 118         logger.debug("get_account_meta: %s %s", account, until)
 119         if user != account:
 120             raise NotAllowedError
 121         try:
 122             version_id, mtime = self._get_accountinfo(account, until)
 123         except NameError:
 124             version_id = None
 125             mtime = 0
 126         count, bytes, tstamp = self._get_pathstats(account, until)
 127         if mtime > tstamp:
 128             tstamp = mtime
 129         if until is None:
 130             modified = tstamp
 131         else:
 132             modified = self._get_pathstats(account)[2] # Overall last modification
 133             if mtime > modified:
 134                 modified = mtime
 135
 136         # Proper count.
 137         sql = 'select count(name) from (%s) where name glob ? and not name glob ?'
 138         sql = sql % self._sql_until(until)
 139         c = self.con.execute(sql, (account + '/*', account + '/*/*'))
 140         row = c.fetchone()
 141         count = row[0]
 142
 143         meta = self._get_metadata(account, version_id)
 144         meta.update({'name': account, 'count': count, 'bytes': bytes})
 145         if modified:
 146             meta.update({'modified': modified})
 147         if until is not None:
 148             meta.update({'until_timestamp': tstamp})
 149         return meta
 150
 151     def update_account_meta(self, user, account, meta, replace=False):
 152         """Update the metadata associated with the account."""
 153
 154         logger.debug("update_account_meta: %s %s %s", account, meta, replace)
 155         if user != account:
 156             raise NotAllowedError
 157         self._put_metadata(user, account, meta, replace)
 158
 159     def get_account_groups(self, user, account):
 160         """Return a dictionary with the user groups defined for this account."""
 161
 162         logger.debug("get_account_groups: %s", account)
 163         if user != account:
 164             raise NotAllowedError
 165         return self._get_groups(account)
 166
 167     def update_account_groups(self, user, account, groups, replace=False):
 168         """Update the groups associated with the account."""
 169
 170         logger.debug("update_account_groups: %s %s %s", account, groups, replace)
 171         if user != account:
 172             raise NotAllowedError
 173         for k, v in groups.iteritems():
 174             if True in [False or ',' in x for x in v]:
 175                 raise ValueError('Bad characters in groups')
 176         if replace:
 177             sql = 'delete from groups where account = ?'
 178             self.con.execute(sql, (account,))
 179         for k, v in groups.iteritems():
 180             if len(v) == 0:
 181                 if not replace:
 182                     sql = 'delete from groups where account = ? and name = ?'
 183                     self.con.execute(sql, (account, k))
 184             else:
 185                 sql = 'insert or replace into groups (account, name, users) values (?, ?, ?)'
 186                 self.con.execute(sql, (account, k, ','.join(v)))
 187         self.con.commit()
 188
 189     def put_account(self, user, account):
 190         """Create a new account with the given name."""
 191
 192         logger.debug("put_account: %s", account)
 193         if user != account:
 194             raise NotAllowedError
 195         try:
 196             version_id, mtime = self._get_accountinfo(account)
 197         except NameError:
 198             pass
 199         else:
 200             raise NameError('Account already exists')
 201         version_id = self._put_version(account, user)
 202         self.con.commit()
 203
 204     def delete_account(self, user, account):
 205         """Delete the account with the given name."""
 206
 207         logger.debug("delete_account: %s", account)
 208         if user != account:
 209             raise NotAllowedError
 210         if self._get_pathcount(account) > 0:
 211             raise IndexError('Account is not empty')
 212         sql = 'delete from versions where name = ?'
 213         self.con.execute(sql, (path,))
 214         sql = 'delete from groups where name = ?'
 215         self.con.execute(sql, (account,))
 216         self.con.commit()
 217
 218     def list_containers(self, user, account, marker=None, limit=10000, until=None):
 219         """Return a list of containers existing under an account."""
 220
 221         logger.debug("list_containers: %s %s %s %s", account, marker, limit, until)
 222         if user != account:
 223             raise NotAllowedError
 224         return self._list_objects(account, '', '/', marker, limit, False, [], until)
 225
 226     def get_container_meta(self, user, account, container, until=None):
 227         """Return a dictionary with the container metadata."""
 228
 229         logger.debug("get_container_meta: %s %s %s", account, container, until)
 230         if user != account:
 231             raise NotAllowedError
 232         path, version_id, mtime = self._get_containerinfo(account, container, until)
 233         count, bytes, tstamp = self._get_pathstats(path, until)
 234         if mtime > tstamp:
 235             tstamp = mtime
 236         if until is None:
 237             modified = tstamp
 238         else:
 239             modified = self._get_pathstats(path)[2] # Overall last modification
 240             if mtime > modified:
 241                 modified = mtime
 242
 243         meta = self._get_metadata(path, version_id)
 244         meta.update({'name': container, 'count': count, 'bytes': bytes, 'modified': modified})
 245         if until is not None:
 246             meta.update({'until_timestamp': tstamp})
 247         return meta
 248
 249     def update_container_meta(self, user, account, container, meta, replace=False):
 250         """Update the metadata associated with the container."""
 251
 252         logger.debug("update_container_meta: %s %s %s %s", account, container, meta, replace)
 253         if user != account:
 254             raise NotAllowedError
 255         path, version_id, mtime = self._get_containerinfo(account, container)
 256         self._put_metadata(user, path, meta, replace)
 257
 258     def get_container_policy(self, user, account, container):
 259         """Return a dictionary with the container policy."""
 260
 261         logger.debug("get_container_policy: %s %s", account, container)
 262         if user != account:
 263             raise NotAllowedError
 264         path = self._get_containerinfo(account, container)[0]
 265         return self._get_policy(path)
 266
 267     def update_container_policy(self, user, account, container, policy, replace=False):
 268         """Update the policy associated with the account."""
 269
 270         logger.debug("update_container_policy: %s %s %s %s", account, container, policy, replace)
 271         if user != account:
 272             raise NotAllowedError
 273         path = self._get_containerinfo(account, container)[0]
 274         self._check_policy(policy)
 275         if replace:
 276             for k, v in self.default_policy.iteritems():
 277                 if k not in policy:
 278                     policy[k] = v
 279         for k, v in policy.iteritems():
 280             sql = 'insert or replace into policy (name, key, value) values (?, ?, ?)'
 281             self.con.execute(sql, (path, k, v))
 282         self.con.commit()
 283
 284     def put_container(self, user, account, container, policy=None):
 285         """Create a new container with the given name."""
 286
 287         logger.debug("put_container: %s %s %s", account, container, policy)
 288         if user != account:
 289             raise NotAllowedError
 290         try:
 291             path, version_id, mtime = self._get_containerinfo(account, container)
 292         except NameError:
 293             pass
 294         else:
 295             raise NameError('Container already exists')
 296         if policy:
 297             self._check_policy(policy)
 298         path = os.path.join(account, container)
 299         version_id = self._put_version(path, user)
 300         for k, v in self.default_policy.iteritems():
 301             if k not in policy:
 302                 policy[k] = v
 303         for k, v in policy.iteritems():
 304             sql = 'insert or replace into policy (name, key, value) values (?, ?, ?)'
 305             self.con.execute(sql, (path, k, v))
 306         self.con.commit()
 307
 308     def delete_container(self, user, account, container, until=None):
 309         """Delete the container with the given name."""
 310
 311         logger.debug("delete_container: %s %s %s", account, container, until)
 312         if user != account:
 313             raise NotAllowedError
 314         path, version_id, mtime = self._get_containerinfo(account, container)
 315
 316         if until is not None:
 317             sql = '''select version_id from versions where name like ? and tstamp <= datetime(%s, 'unixepoch')'''
 318             c = self.con.execute(sql, (path + '/%', until))
 319             versions = [x[0] for x in c.fetchall()]
 320             for v in versions:
 321                 sql = 'delete from hashmaps where version_id = ?'
 322                 self.con.execute(sql, (v,))
 323                 sql = 'delete from versions where version_id = ?'
 324                 self.con.execute(sql, (v,))
 325             self.con.commit()
 326             return
 327
 328         if self._get_pathcount(path) > 0:
 329             raise IndexError('Container is not empty')
 330         sql = 'delete from versions where name = ?'
 331         self.con.execute(sql, (path,))
 332         sql = 'delete from policy where name = ?'
 333         self.con.execute(sql, (path,))
 334         self._copy_version(user, account, account, True, True) # New account version (for timestamp update).
 335
 336     def list_objects(self, user, account, container, prefix='', delimiter=None, marker=None, limit=10000, virtual=True, keys=[], until=None):
 337         """Return a list of objects existing under a container."""
 338
 339         logger.debug("list_objects: %s %s %s %s %s %s %s", account, container, prefix, delimiter, marker, limit, until)
 340         if user != account:
 341             raise NotAllowedError
 342         path, version_id, mtime = self._get_containerinfo(account, container, until)
 343         return self._list_objects(path, prefix, delimiter, marker, limit, virtual, keys, until)
 344
 345     def list_object_meta(self, user, account, container, until=None):
 346         """Return a list with all the container's object meta keys."""
 347
 348         logger.debug("list_object_meta: %s %s %s", account, container, until)
 349         if user != account:
 350             raise NotAllowedError
 351         path, version_id, mtime = self._get_containerinfo(account, container, until)
 352         sql = '''select distinct m.key from (%s) o, metadata m
 353                     where m.version_id = o.version_id and o.name like ?'''
 354         sql = sql % self._sql_until(until)
 355         c = self.con.execute(sql, (path + '/%',))
 356         return [x[0] for x in c.fetchall()]
 357
 358     def get_object_meta(self, user, account, container, name, version=None):
 359         """Return a dictionary with the object metadata."""
 360
 361         logger.debug("get_object_meta: %s %s %s %s", account, container, name, version)
 362         self._can_read(user, account, container, name)
 363         path, version_id, muser, mtime, size = self._get_objectinfo(account, container, name, version)
 364         if version is None:
 365             modified = mtime
 366         else:
 367             modified = self._get_version(path, version)[2] # Overall last modification
 368
 369         meta = self._get_metadata(path, version_id)
 370         meta.update({'name': name, 'bytes': size})
 371         meta.update({'version': version_id, 'version_timestamp': mtime})
 372         meta.update({'modified': modified, 'modified_by': muser})
 373         return meta
 374
 375     def update_object_meta(self, user, account, container, name, meta, replace=False):
 376         """Update the metadata associated with the object."""
 377
 378         logger.debug("update_object_meta: %s %s %s %s %s", account, container, name, meta, replace)
 379         self._can_write(user, account, container, name)
 380         path, version_id, muser, mtime, size = self._get_objectinfo(account, container, name)
 381         self._put_metadata(user, path, meta, replace)
 382
 383     def get_object_permissions(self, user, account, container, name):
 384         """Return the path from which this object gets its permissions from,\
 385         along with a dictionary containing the permissions."""
 386
 387         logger.debug("get_object_permissions: %s %s %s", account, container, name)
 388         self._can_read(user, account, container, name)
 389         path = self._get_objectinfo(account, container, name)[0]
 390         return self._get_permissions(path)
 391
 392     def update_object_permissions(self, user, account, container, name, permissions):
 393         """Update the permissions associated with the object."""
 394
 395         logger.debug("update_object_permissions: %s %s %s %s", account, container, name, permissions)
 396         if user != account:
 397             raise NotAllowedError
 398         path = self._get_objectinfo(account, container, name)[0]
 399         r, w = self._check_permissions(path, permissions)
 400         self._put_permissions(path, r, w)
 401
 402     def get_object_public(self, user, account, container, name):
 403         """Return the public URL of the object if applicable."""
 404
 405         logger.debug("get_object_public: %s %s %s", account, container, name)
 406         self._can_read(user, account, container, name)
 407         path = self._get_objectinfo(account, container, name)[0]
 408         if self._get_public(path):
 409             return '/public/' + path
 410         return None
 411
 412     def update_object_public(self, user, account, container, name, public):
 413         """Update the public status of the object."""
 414
 415         logger.debug("update_object_public: %s %s %s %s", account, container, name, public)
 416         self._can_write(user, account, container, name)
 417         path = self._get_objectinfo(account, container, name)[0]
 418         self._put_public(path, public)
 419
 420     def get_object_hashmap(self, user, account, container, name, version=None):
 421         """Return the object's size and a list with partial hashes."""
 422
 423         logger.debug("get_object_hashmap: %s %s %s %s", account, container, name, version)
 424         self._can_read(user, account, container, name)
 425         path, version_id, muser, mtime, size = self._get_objectinfo(account, container, name, version)
 426         sql = 'select block_id from hashmaps where version_id = ? order by pos asc'
 427         c = self.con.execute(sql, (version_id,))
 428         hashmap = [x[0] for x in c.fetchall()]
 429         return size, hashmap
 430
 431     def update_object_hashmap(self, user, account, container, name, size, hashmap, meta={}, replace_meta=False, permissions=None):
 432         """Create/update an object with the specified size and partial hashes."""
 433
 434         logger.debug("update_object_hashmap: %s %s %s %s %s", account, container, name, size, hashmap)
 435         if permissions is not None and user != account:
 436             raise NotAllowedError
 437         self._can_write(user, account, container, name)
 438         missing = []
 439         for i in range(len(hashmap)):
 440             sql = 'select count(*) from blocks where block_id = ?'
 441             c = self.con.execute(sql, (hashmap[i],))
 442             if c.fetchone()[0] == 0:
 443                 missing.append(hashmap[i])
 444         if missing:
 445             ie = IndexError()
 446             ie.data = missing
 447             raise ie
 448         path = self._get_containerinfo(account, container)[0]
 449         path = os.path.join(path, name)
 450         if permissions is not None:
 451             r, w = self._check_permissions(path, permissions)
 452         src_version_id, dest_version_id = self._copy_version(user, path, path, not replace_meta, False)
 453         sql = 'update versions set size = ? where version_id = ?'
 454         self.con.execute(sql, (size, dest_version_id))
 455         # TODO: Check for block_id existence.
 456         for i in range(len(hashmap)):
 457             sql = 'insert or replace into hashmaps (version_id, pos, block_id) values (?, ?, ?)'
 458             self.con.execute(sql, (dest_version_id, i, hashmap[i]))
 459         for k, v in meta.iteritems():
 460             sql = 'insert or replace into metadata (version_id, key, value) values (?, ?, ?)'
 461             self.con.execute(sql, (dest_version_id, k, v))
 462         if permissions is not None:
 463             sql = 'insert or replace into permissions (name, read, write) values (?, ?, ?)'
 464             self.con.execute(sql, (path, r, w))
 465         self.con.commit()
 466
 467     def copy_object(self, user, account, src_container, src_name, dest_container, dest_name, dest_meta={}, replace_meta=False, permissions=None, src_version=None):
 468         """Copy an object's data and metadata."""
 469
 470         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)
 471         if permissions is not None and user != account:
 472             raise NotAllowedError
 473         self._can_read(user, account, src_container, src_name)
 474         self._can_write(user, account, dest_container, dest_name)
 475         self._get_containerinfo(account, src_container)
 476         if src_version is None:
 477             src_path = self._get_objectinfo(account, src_container, src_name)[0]
 478         else:
 479             src_path = os.path.join(account, src_container, src_name)
 480         dest_path = self._get_containerinfo(account, dest_container)[0]
 481         dest_path = os.path.join(dest_path, dest_name)
 482         if permissions is not None:
 483             r, w = self._check_permissions(dest_path, permissions)
 484         src_version_id, dest_version_id = self._copy_version(user, src_path, dest_path, not replace_meta, True, src_version)
 485         for k, v in dest_meta.iteritems():
 486             sql = 'insert or replace into metadata (version_id, key, value) values (?, ?, ?)'
 487             self.con.execute(sql, (dest_version_id, k, v))
 488         if permissions is not None:
 489             sql = 'insert or replace into permissions (name, read, write) values (?, ?, ?)'
 490             self.con.execute(sql, (dest_path, r, w))
 491         self.con.commit()
 492
 493     def move_object(self, user, account, src_container, src_name, dest_container, dest_name, dest_meta={}, replace_meta=False, permissions=None):
 494         """Move an object's data and metadata."""
 495
 496         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)
 497         self.copy_object(user, account, src_container, src_name, dest_container, dest_name, dest_meta, replace_meta, permissions, None)
 498         self.delete_object(user, account, src_container, src_name)
 499
 500     def delete_object(self, user, account, container, name, until=None):
 501         """Delete an object."""
 502
 503         logger.debug("delete_object: %s %s %s %s", account, container, name, until)
 504         if user != account:
 505             raise NotAllowedError
 506         if until is None:
 507             path = self._get_objectinfo(account, container, name)[0]
 508             sql = 'select version_id from versions where name = ?'
 509             c = self.con.execute(sql, (path,))
 510         else:
 511             path = os.path.join(account, container, name)
 512             sql = '''select version_id from versions where name = ? and tstamp <= datetime(%s, 'unixepoch')'''
 513             c = self.con.execute(sql, (path, until))
 514         versions = [x[0] for x in c.fetchall()]
 515         for v in versions:
 516             sql = 'delete from hashmaps where version_id = ?'
 517             self.con.execute(sql, (v,))
 518             sql = 'delete from versions where version_id = ?'
 519             self.con.execute(sql, (v,))
 520
 521         # If no more versions exist, delete permissions/public.
 522         sql = 'select version_id from versions where name = ?'
 523         row = self.con.execute(sql, (path,)).fetchone()
 524         if row is None:
 525             self._del_sharing(path)
 526         self.con.commit()
 527
 528     def update_object_trash(self, user, account, container, name, trash=True):
 529         """Trash/untrash an object."""
 530
 531         logger.debug("trash_object: %s %s %s", account, container, name)
 532         if user != account:
 533             raise NotAllowedError
 534         if trash:
 535             path, version_id, muser, mtime, size = self._get_objectinfo(account, container, name)
 536             self._del_sharing(path)
 537         else:
 538             path = os.path.join(account, container, name)
 539             sql = '''select version_id, hide from versions where name = ?
 540                         order by version_id desc limit 1'''
 541             c = self.con.execute(sql, (path,))
 542             row = c.fetchone()
 543             if not row or not int(row[1]):
 544                 raise NameError('Object not in trash')
 545             version_id = row[0]
 546         hide = 1 if trash else 0
 547         sql = 'update versions set hide = ? where version_id = ?'
 548         self.con.execute(sql, (hide, version_id,))
 549         seld.con.commit()
 550
 551     def list_versions(self, user, account, container, name):
 552         """Return a list of all (version, version_timestamp) tuples for an object."""
 553
 554         logger.debug("list_versions: %s %s %s", account, container, name)
 555         self._can_read(user, account, container, name)
 556         # This will even show deleted versions.
 557         path = os.path.join(account, container, name)
 558         sql = '''select distinct version_id, strftime('%s', tstamp) from versions where name = ?'''
 559         c = self.con.execute(sql, (path,))
 560         return [(int(x[0]), int(x[1])) for x in c.fetchall()]
 561
 562     def get_block(self, hash):
 563         """Return a block's data."""
 564
 565         logger.debug("get_block: %s", hash)
 566         c = self.con.execute('select data from blocks where block_id = ?', (hash,))
 567         row = c.fetchone()
 568         if row:
 569             return str(row[0])
 570         else:
 571             raise NameError('Block does not exist')
 572
 573     def put_block(self, data):
 574         """Create a block and return the hash."""
 575
 576         logger.debug("put_block: %s", len(data))
 577         h = hashlib.new(self.hash_algorithm)
 578         h.update(data.rstrip('\x00'))
 579         hash = h.hexdigest()
 580         sql = 'insert or ignore into blocks (block_id, data) values (?, ?)'
 581         self.con.execute(sql, (hash, buffer(data)))
 582         self.con.commit()
 583         return hash
 584
 585     def update_block(self, hash, data, offset=0):
 586         """Update a known block and return the hash."""
 587
 588         logger.debug("update_block: %s %s %s", hash, len(data), offset)
 589         if offset == 0 and len(data) == self.block_size:
 590             return self.put_block(data)
 591         src_data = self.get_block(hash)
 592         bs = self.block_size
 593         if offset < 0 or offset > bs or offset + len(data) > bs:
 594             raise IndexError('Offset or data outside block limits')
 595         dest_data = src_data[:offset] + data + src_data[offset + len(data):]
 596         return self.put_block(dest_data)
 597
 598     def _sql_until(self, until=None):
 599         """Return the sql to get the latest versions until the timestamp given."""
 600         if until is None:
 601             until = int(time.time())
 602         sql = '''select version_id, name, strftime('%s', tstamp) as tstamp, size from versions v
 603                     where version_id = (select max(version_id) from versions
 604                                         where v.name = name and tstamp <= datetime(%s, 'unixepoch'))
 605                     and hide = 0'''
 606         return sql % ('%s', until)
 607
 608     def _get_pathstats(self, path, until=None):
 609         """Return count, sum of size and latest timestamp of everything under path (latest versions/no trash)."""
 610
 611         sql = 'select count(version_id), total(size), max(tstamp) from (%s) where name like ?'
 612         sql = sql % self._sql_until(until)
 613         c = self.con.execute(sql, (path + '/%',))
 614         row = c.fetchone()
 615         tstamp = row[2] if row[2] is not None else 0
 616         return int(row[0]), int(row[1]), int(tstamp)
 617
 618     def _get_pathcount(self, path):
 619         """Return count of everything under path (including versions/trash)."""
 620
 621         sql = 'select count(version_id) from versions where name like ?'
 622         c = self.con.execute(sql, (path + '/%',))
 623         row = c.fetchone()
 624         return int(row[0])
 625
 626     def _get_version(self, path, version=None):
 627         if version is None:
 628             sql = '''select version_id, user, strftime('%s', tstamp), size, hide from versions where name = ?
 629                         order by version_id desc limit 1'''
 630             c = self.con.execute(sql, (path,))
 631             row = c.fetchone()
 632             if not row or int(row[4]):
 633                 raise NameError('Object does not exist')
 634         else:
 635             # The database (sqlite) will not complain if the version is not an integer.
 636             sql = '''select version_id, user, strftime('%s', tstamp), size from versions where name = ?
 637                         and version_id = ?'''
 638             c = self.con.execute(sql, (path, version))
 639             row = c.fetchone()
 640             if not row:
 641                 raise IndexError('Version does not exist')
 642         return str(row[0]), str(row[1]), int(row[2]), int(row[3])
 643
 644     def _put_version(self, path, user, size=0):
 645         sql = 'insert into versions (name, user, size) values (?, ?, ?)'
 646         id = self.con.execute(sql, (path, user, size)).lastrowid
 647         self.con.commit()
 648         return str(id)
 649
 650     def _copy_version(self, user, src_path, dest_path, copy_meta=True, copy_data=True, src_version=None):
 651         if src_version is not None:
 652             src_version_id, muser, mtime, size = self._get_version(src_path, src_version)
 653         else:
 654             # Latest or create from scratch.
 655             try:
 656                 src_version_id, muser, mtime, size = self._get_version(src_path)
 657             except NameError:
 658                 src_version_id = None
 659                 size = 0
 660         if not copy_data:
 661             size = 0
 662         dest_version_id = self._put_version(dest_path, user, size)
 663         if copy_meta and src_version_id is not None:
 664             sql = 'insert into metadata select %s, key, value from metadata where version_id = ?'
 665             sql = sql % dest_version_id
 666             self.con.execute(sql, (src_version_id,))
 667         if copy_data and src_version_id is not None:
 668             sql = 'insert into hashmaps select %s, pos, block_id from hashmaps where version_id = ?'
 669             sql = sql % dest_version_id
 670             self.con.execute(sql, (src_version_id,))
 671         self.con.commit()
 672         return src_version_id, dest_version_id
 673
 674     def _get_versioninfo(self, account, container, name, until=None):
 675         """Return path, latest version, associated timestamp and size until the timestamp given."""
 676
 677         p = (account, container, name)
 678         try:
 679             p = p[:p.index(None)]
 680         except ValueError:
 681             pass
 682         path = os.path.join(*p)
 683         sql = '''select version_id, tstamp, size from (%s) where name = ?'''
 684         sql = sql % self._sql_until(until)
 685         c = self.con.execute(sql, (path,))
 686         row = c.fetchone()
 687         if row is None:
 688             raise NameError('Path does not exist')
 689         return path, str(row[0]), int(row[1]), int(row[2])
 690
 691     def _get_accountinfo(self, account, until=None):
 692         try:
 693             path, version_id, mtime, size = self._get_versioninfo(account, None, None, until)
 694             return version_id, mtime
 695         except:
 696             raise NameError('Account does not exist')
 697
 698     def _get_containerinfo(self, account, container, until=None):
 699         try:
 700             path, version_id, mtime, size = self._get_versioninfo(account, container, None, until)
 701             return path, version_id, mtime
 702         except:
 703             raise NameError('Container does not exist')
 704
 705     def _get_objectinfo(self, account, container, name, version=None):
 706         path = os.path.join(account, container, name)
 707         version_id, muser, mtime, size = self._get_version(path, version)
 708         return path, version_id, muser, mtime, size
 709
 710     def _get_metadata(self, path, version):
 711         sql = 'select key, value from metadata where version_id = ?'
 712         c = self.con.execute(sql, (version,))
 713         return dict(c.fetchall())
 714
 715     def _put_metadata(self, user, path, meta, replace=False):
 716         """Create a new version and store metadata."""
 717
 718         src_version_id, dest_version_id = self._copy_version(user, path, path, not replace, True)
 719         for k, v in meta.iteritems():
 720             if not replace and v == '':
 721                 sql = 'delete from metadata where version_id = ? and key = ?'
 722                 self.con.execute(sql, (dest_version_id, k))
 723             else:
 724                 sql = 'insert or replace into metadata (version_id, key, value) values (?, ?, ?)'
 725                 self.con.execute(sql, (dest_version_id, k, v))
 726         self.con.commit()
 727
 728     def _get_groups(self, account):
 729         sql = 'select name, users from groups where account = ?'
 730         c = self.con.execute(sql, (account,))
 731         return dict([(x[0], x[1].split(',')) for x in c.fetchall()])
 732
 733     def _check_policy(self, policy):
 734         for k in policy.keys():
 735             if policy[k] == '':
 736                 policy[k] = self.default_policy.get(k)
 737         for k, v in policy.iteritems():
 738             if k == 'quota':
 739                 q = int(v) # May raise ValueError.
 740                 if q < 0:
 741                     raise ValueError
 742             elif k == 'versioning':
 743                 if v not in ['auto', 'manual', 'none']:
 744                     raise ValueError
 745             else:
 746                 raise ValueError
 747
 748     def _get_policy(self, path):
 749         sql = 'select key, value from policy where name = ?'
 750         c = self.con.execute(sql, (path,))
 751         return dict(c.fetchall())
 752
 753     def _is_allowed(self, user, account, container, name, op='read'):
 754         if user == account:
 755             return True
 756         path = os.path.join(account, container, name)
 757         if op == 'read' and self._get_public(path):
 758             return True
 759         perm_path, perms = self._get_permissions(path)
 760
 761         # Expand groups.
 762         for x in ('read', 'write'):
 763             g_perms = []
 764             for y in perms.get(x, []):
 765                 groups = self._get_groups(account)
 766                 if y in groups: #it's a group
 767                     for g_name in groups[y]:
 768                         g_perms.append(g_name)
 769                 else: #it's a user
 770                     g_perms.append(y)
 771             perms[x] = g_perms
 772
 773         if op == 'read' and user in perms.get('read', []):
 774             return True
 775         if user in perms.get('write', []):
 776             return True
 777         return False
 778
 779     def _can_read(self, user, account, container, name):
 780         if not self._is_allowed(user, account, container, name, 'read'):
 781             raise NotAllowedError
 782
 783     def _can_write(self, user, account, container, name):
 784         if not self._is_allowed(user, account, container, name, 'write'):
 785             raise NotAllowedError
 786
 787     def _check_permissions(self, path, permissions):
 788         # Check for existing permissions.
 789         sql = '''select name from permissions
 790                     where name != ? and (name like ? or ? like name || ?)'''
 791         c = self.con.execute(sql, (path, path + '%', path, '%'))
 792         row = c.fetchone()
 793         if row:
 794             ae = AttributeError()
 795             ae.data = row[0]
 796             raise ae
 797
 798         # Format given permissions.
 799         if len(permissions) == 0:
 800             return '', ''
 801         r = permissions.get('read', [])
 802         w = permissions.get('write', [])
 803         if True in [False or ',' in x for x in r]:
 804             raise ValueError('Bad characters in read permissions')
 805         if True in [False or ',' in x for x in w]:
 806             raise ValueError('Bad characters in write permissions')
 807         return ','.join(r), ','.join(w)
 808
 809     def _get_permissions(self, path):
 810         # Check for permissions at path or above.
 811         sql = 'select name, read, write from permissions where ? like name || ?'
 812         c = self.con.execute(sql, (path, '%'))
 813         row = c.fetchone()
 814         if not row:
 815             return path, {}
 816
 817         name, r, w = row
 818         ret = {}
 819         if w != '':
 820             ret['write'] = w.split(',')
 821         if r != '':
 822             ret['read'] = r.split(',')
 823         return name, ret
 824
 825     def _put_permissions(self, path, r, w):
 826         if r == '' and w == '':
 827             sql = 'delete from permissions where name = ?'
 828             self.con.execute(sql, (path,))
 829         else:
 830             sql = 'insert or replace into permissions (name, read, write) values (?, ?, ?)'
 831             self.con.execute(sql, (path, r, w))
 832         self.con.commit()
 833
 834     def _get_public(self, path):
 835         sql = 'select name from public where name = ?'
 836         c = self.con.execute(sql, (path,))
 837         row = c.fetchone()
 838         if not row:
 839             return False
 840         return True
 841
 842     def _put_public(self, path, public):
 843         if not public:
 844             sql = 'delete from public where name = ?'
 845         else:
 846             sql = 'insert or replace into public (name) values (?)'
 847         self.con.execute(sql, (path,))
 848         self.con.commit()
 849
 850     def _del_sharing(self, path):
 851         sql = 'delete from permissions where name = ?'
 852         self.con.execute(sql, (path,))
 853         sql = 'delete from public where name = ?'
 854         self.con.execute(sql, (path,))
 855         self.con.commit()
 856
 857     def _list_objects(self, path, prefix='', delimiter=None, marker=None, limit=10000, virtual=True, keys=[], until=None):
 858         cont_prefix = path + '/'
 859         if keys and len(keys) > 0:
 860             sql = '''select distinct o.name, o.version_id from (%s) o, metadata m where o.name like ? and
 861                         m.version_id = o.version_id and m.key in (%s) order by o.name'''
 862             sql = sql % (self._sql_until(until), ', '.join('?' * len(keys)))
 863             param = (cont_prefix + prefix + '%',) + tuple(keys)
 864         else:
 865             sql = 'select name, version_id from (%s) where name like ? order by name'
 866             sql = sql % self._sql_until(until)
 867             param = (cont_prefix + prefix + '%',)
 868         c = self.con.execute(sql, param)
 869         objects = [(x[0][len(cont_prefix):], x[1]) for x in c.fetchall()]
 870         if delimiter:
 871             pseudo_objects = []
 872             for x in objects:
 873                 pseudo_name = x[0]
 874                 i = pseudo_name.find(delimiter, len(prefix))
 875                 if not virtual:
 876                     # If the delimiter is not found, or the name ends
 877                     # with the delimiter's first occurence.
 878                     if i == -1 or len(pseudo_name) == i + len(delimiter):
 879                         pseudo_objects.append(x)
 880                 else:
 881                     # If the delimiter is found, keep up to (and including) the delimiter.
 882                     if i != -1:
 883                         pseudo_name = pseudo_name[:i + len(delimiter)]
 884                     if pseudo_name not in [y[0] for y in pseudo_objects]:
 885                         if pseudo_name == x[0]:
 886                             pseudo_objects.append(x)
 887                         else:
 888                             pseudo_objects.append((pseudo_name, None))
 889             objects = pseudo_objects
 890
 891         start = 0
 892         if marker:
 893             try:
 894                 start = [x[0] for x in objects].index(marker) + 1
 895             except ValueError:
 896                 pass
 897         if not limit or limit > 10000:
 898             limit = 10000
 899         return objects[start:start + limit]