first implementation of client blobs

1 files changed, 181 insertions, 0 deletions

diff --git a/client/src/leap/soledad/client/_blobs.py b/client/src/leap/soledad/client/_blobs.py
new file mode 100644
index 00000000..952f1da7
--- /dev/null
+++ b/client/src/leap/soledad/client/_blobs.py
@@ -0,0 +1,181 @@
+"""
+Clientside BlobBackend Storage.
+"""
+
+from uuid import uuid4
+import os.path
+
+from io import BytesIO
+from functools import partial
+
+from sqlite3 import Binary
+
+from twisted.logger import Logger
+from twisted.enterprise import adbapi
+from twisted.internet import defer, reactor
+
+import treq
+
+from leap.soledad.client.sqlcipher import SQLCipherOptions
+from leap.soledad.client import pragmas
+
+from _crypto import docinfo, BlobEncryptor, BlobDecryptor
+
+
+logger = Logger()
+
+
+class BlobManager(object):
+
+    def __init__(self, local_path, remote, key, secret):
+        self.local = SQLiteBlobBackend(local_path, key)
+        self.remote = remote
+        self.secret = secret
+
+    @defer.inlineCallbacks
+    def put(self, doc):
+        fd = doc.blob_fd
+        yield self.local.put(doc.blob_id, fd)
+        fd.seek(0)
+        up = BytesIO()
+        doc_info = docinfo(doc.doc_id, doc.rev)
+
+        # TODO ------------------------------------------
+        # this is wrong, is doing 2 stages. Cutting corners!
+        # We should connect the pipeline, use Tubes: 
+        # the crypto producer can be passed to 
+        # the uploader and react as data is written.
+        # ------------------------------------------------
+        yield self._encrypt(doc_info, fd, up)
+        yield self._upload(doc.blob_id, up)
+
+    @defer.inlineCallbacks
+    def get(self, blob_id, doc_id, rev):
+        print "IN MANAGER: GETTING BLOB..."
+        local_blob = yield self.local.get(blob_id)
+        if local_blob:
+            print 'LOCAL BLOB', local_blob
+            defer.returnValue(local_blob)
+
+        print "NO LOCAL BLOB, WILL DOWNLOAD"
+
+        # TODO pass the fd to the downloader, possible?
+        remoteblob = yield self._download(blob_id)
+        ciphertext = BytesIO(str(remoteblob))
+
+        print 'remote ciphertext', remoteblob[:10], '[...]'
+        logger.debug('got remote blob %s [...]' % remoteblob[:100])
+        del remoteblob
+
+        doc_info = docinfo(doc_id, rev)
+        blob = yield self._decrypt(doc_info, ciphertext)
+        if blob:
+            print 'GOT DECRYPTED BLOB', type(blob)
+            blob.seek(0)
+            print 'SAVING BLOB IN LOCAL STORE'
+            yield self.local.put(blob_id, blob)
+            blob.seek(0)
+            defer.returnValue(blob)
+        else:
+        # XXX we shouldn't get here, but we will...
+        # lots of ugly error handling possible:
+        # 1. retry, might be network error
+        # 2. try later, maybe didn't finished streaming
+        # 3.. resignation, might be error while verifying
+            logger.error('sorry, dunno what happened')
+
+
+    @defer.inlineCallbacks
+    def _encrypt(self, doc_info, payload, result):
+        # TODO WE SHOULD SKIP THE BASE64 STEP!!!!
+        # this is going to be uploaded in binary mode
+        crypter = BlobEncryptor(doc_info, payload, result=result, secret=self.secret)
+        yield crypter.encrypt()
+
+
+    @defer.inlineCallbacks
+    def _decrypt(self, doc_info, ciphertext):
+        decrypter = BlobDecryptor(doc_info, ciphertext, secret=self.secret)
+        blob = yield decrypter.decrypt()
+        defer.returnValue(blob)
+
+
+    @defer.inlineCallbacks
+    def _upload(self, blob_id, payload_fd):
+        uri = self.remote + 'put'
+        yield treq.post(uri, data={'dafile_filename': blob_id},
+                        files={'dafile': payload_fd})
+
+    @defer.inlineCallbacks
+    def _download(self, blob_id):
+        uri = self.remote + 'blobs/' + blob_id
+        blob_resp = yield treq.get(uri)
+        blob = yield treq.text_content(blob_resp)
+        defer.returnValue(blob)
+
+
+
+# --------------------8<----------------------------------------------
+class BlobDoc(object):
+
+    # TODO probably not needed, but convenient for testing for now.
+
+    def __init__(self, doc_id, rev, content, blob_id=None):
+
+        self.doc_id = doc_id
+        self.rev = rev
+        self.is_blob = True
+        self.blob_fd = content
+        if blob_id is None:
+            blob_id = uuid4().get_hex()
+        self.blob_id = blob_id
+# --------------------8<----------------------------------------------
+
+
+class SQLiteBlobBackend(object):
+
+    def __init__(self, path, key=None):
+        self.path = os.path.abspath(
+            os.path.join(path, 'soledad_blob.db'))
+        if not key:
+            raise ValueError('key cannot be None')
+        backend = 'pysqlcipher.dbapi2'
+        opts = SQLCipherOptions('/tmp/ignored', key)
+        pragmafun = partial(pragmas.set_init_pragmas, opts=opts)
+        openfun = _sqlcipherInitFactory(pragmafun)
+
+        self.dbpool = dbpool = adbapi.ConnectionPool(
+            backend, self.path, check_same_thread=False, timeout=5,
+            cp_openfun=openfun, cp_min=1, cp_max=2, cp_name='blob_pool')
+
+
+    @defer.inlineCallbacks
+    def put(self, blob_id, blob_fd):
+        insert = str('INSERT INTO blobs VALUES (?, ?)')
+        print 'inserting...'
+        raw = blob_fd.getvalue()
+        yield self.dbpool.runQuery(insert, (blob_id, Binary(raw)))
+        
+    @defer.inlineCallbacks
+    def get(self, blob_id):
+        select = 'SELECT payload FROM blobs WHERE blob_id = ?'
+        result = yield self.dbpool.runQuery(select, (blob_id,))
+        if result:
+            defer.returnValue(BytesIO(str(result[0][0])))
+
+
+
+def _init_blob_table(conn):
+    maybe_create = (
+        "CREATE TABLE IF NOT EXISTS "
+        "blobs ("
+        "blob_id PRIMARY KEY, "
+        "payload BLOB)")
+    conn.execute(maybe_create)
+
+
+def _sqlcipherInitFactory(fun):
+    def _initialize(conn):
+        fun(conn)
+        _init_blob_table(conn)
+    return _initialize