from __future__ import with_statement
+import os as _OS
+
import catacomb as _C
import gdbm as _G
###--------------------------------------------------------------------------
### Backend storage.
+class StorageBackendRefusal (Exception):
+ """
+ I signify that a StorageBackend subclass has refused to open a file.
+
+ This is used by the StorageBackend.open class method.
+ """
+ pass
+
+class StorageBackendClass (type):
+ """
+ I am a metaclass for StorageBackend classes.
+
+ My main feature is that I register my concrete instances (with a `NAME'
+ which is not `None') with the StorageBackend class.
+ """
+ def __init__(me, name, supers, dict):
+ """
+ Register a new concrete StorageBackend subclass.
+ """
+ super(StorageBackendClass, me).__init__(name, supers, dict)
+ if me.NAME is not None: StorageBackend.register_concrete_subclass(me)
+
class StorageBackend (object):
"""
I provide basic protocol for password storage backends.
I'm an abstract class: you want one of my subclasses if you actually want
- to do something useful.
+ to do something useful. But I maintain a list of my subclasses and can
+ choose an appropriate one to open a database file you've found lying about.
Backends are responsible for storing and retrieving stuff, but not for the
cryptographic details. Backends need to store two kinds of information:
BE._iter_passwds() Return an iterator over the password (LABEL, PAYLOAD)
pairs; used by `iter_passwds'.
+
+ Also, concrete subclasses should define the following class attributes.
+
+ NAME The name of the backend, so that the user can select
+ it when creating a new database.
+
+ PRIO An integer priority: backends are tried in decreasing
+ priority order when opening an existing database.
"""
+ __metaclass__ = StorageBackendClass
+ NAME = None
+ PRIO = 10
+
+ ## The registry of subclasses.
+ CLASSES = {}
+
FAIL = ['FAIL']
- ## Life cycle methods.
+ @staticmethod
+ def register_concrete_subclass(sub):
+ """Register a concrete subclass, so that `open' can try it."""
+ StorageBackend.CLASSES[sub.NAME] = sub
+
+ @staticmethod
+ def byname(name):
+ """
+ Return the concrete subclass with the given NAME.
+
+ Raise `KeyError' if the name isn't found.
+ """
+ return StorageBackend.CLASSES[name]
+
+ @staticmethod
+ def classes():
+ """Return an iterator over the concrete subclasses."""
+ return StorageBackend.CLASSES.itervalues()
+
+ @staticmethod
+ def open(file, writep = False):
+ """Open a database FILE, using some appropriate backend."""
+ _OS.stat(file)
+ for cls in sorted(StorageBackend.CLASSES.values(), reverse = True,
+ key = lambda cls: cls.PRIO):
+ try: return cls(file, writep)
+ except StorageBackendRefusal: pass
+ raise StorageBackendRefusal
@classmethod
def create(cls, file):
`_open' method, which is invoked in the usual case.
"""
super(StorageBackend, me).__init__(*args, **kw)
+ if me.NAME is None: raise ValueError, 'abstract class'
if _magic is not None: _magic(me)
elif file is None: raise ValueError, 'missing file parameter'
else: me._open(file, writep)
metadata item name begins with `$'.
"""
+ NAME = 'gdbm'
+
def _open(me, file, writep):
try: me._db = _G.open(file, writep and 'w' or 'r')
except _G.error, e: raise StorageBackendRefusal, e
"""
## Open the database.
- me.db = GDBMStorageBackend(file, writep)
+ me.db = StorageBackend.open(file, writep)
## Find out what crypto to use.
c = _C.gcciphers[me.db.get_meta('cipher')]
me.magic = me.k.decrypt(me.db.get_meta('magic'))
@classmethod
- def create(cls, file, tag, c, h, m):
+ def create(cls, dbcls, file, tag, c, h, m):
"""
- Create and initialize a new database FILE.
+ Create and initialize a new database FILE using StorageBackend DBCLS.
We want a GCipher subclass C, a GHash subclass H, and a GMAC subclass M;
and a Pixie passphrase TAG.
## Set up and initialize the database.
kct = ppk.encrypt(_C.WriteBuffer().putblk16(ck).putblk16(mk))
- with GDBM.StorageBackend.create(file) as db:
+ with dbcls.create(file) as db:
db.put_meta('tag', tag)
db.put_meta('salt', ppk.salt)
db.put_meta('cipher', c.name)
## Parse the options.
try:
- opts, args = getopt(av, 'c:h:m:', ['cipher=', 'mac=', 'hash='])
+ opts, args = getopt(av, 'c:d:h:m:',
+ ['cipher=', 'database=', 'mac=', 'hash='])
except GetoptError:
return 1
+ dbty = 'gdbm'
for o, a in opts:
if o in ('-c', '--cipher'): cipher = a
elif o in ('-m', '--mac'): mac = a
elif o in ('-h', '--hash'): hash = a
+ elif o in ('-d', '--database'): dbty = a
else: raise 'Barf!'
if len(args) > 2: return 1
if len(args) >= 1: tag = args[0]
## Set up the database.
if mac is None: mac = hash + '-hmac'
- PW.create(file, C.gcciphers[cipher], C.gchashes[hash], C.gcmacs[mac], tag)
+ try: dbcls = StorageBackend.byname(dbty)
+ except KeyError: die("Unknown database backend `%s'" % dbty)
+ PW.create(dbcls, file, tag,
+ C.gcciphers[cipher], C.gchashes[hash], C.gcmacs[mac])
def cmd_changepp(av):
if len(av) != 0: return 1
except KeyError, exc: die("Password `%s' not found" % exc.args[0])
commands = { 'create': [cmd_create,
- '[-c CIPHER] [-h HASH] [-m MAC] [PP-TAG]'],
+ '[-c CIPHER] [-d DBTYPE] [-h HASH] [-m MAC] [PP-TAG]'],
'find' : [cmd_find, 'LABEL'],
'store' : [cmd_store, 'LABEL [VALUE]'],
'list' : [cmd_list, '[GLOB-PATTERN]'],
def version():
print '%s 1.0.0' % prog
+ print 'Backend types: %s' % \
+ ' '.join([c.NAME for c in StorageBackend.classes()])
def usage(fp):
print >>fp, 'Usage: %s COMMAND [ARGS...]' % prog
~mdw / catacomb-python / commitdiff