diff --git a/client/pkg/requirements.pip b/client/pkg/requirements.pip
index ff1b4f35cc968bbea017f2c3aceeac3321c72a11..6f2954ab2a3fb3526b6bda7a425b3e8fa10142b5 100644
--- a/client/pkg/requirements.pip
+++ b/client/pkg/requirements.pip
@@ -3,6 +3,9 @@ simplejson
u1db
scrypt
pycryptopp
+cchardet
+taskthread
+zope.proxy
#
# leap deps
@@ -21,5 +24,3 @@ oauth
# pysqlite should not be a dep, see #2945
pysqlite
-cchardet
-taskthread
diff --git a/client/src/leap/soledad/client/__init__.py b/client/src/leap/soledad/client/__init__.py
index 5f1d1a983cdf745574a2bc6d26aac35c27dcacdd..116a59e450ba9f0bbe717fac4fd36623bcc3c2c4 100644
--- a/client/src/leap/soledad/client/__init__.py
+++ b/client/src/leap/soledad/client/__init__.py
@@ -324,7 +324,7 @@ class Soledad(object):
self._bootstrap() # might raise BootstrapSequenceError()
# initialize syncing queue encryption pool
- self._sync_pool = SyncEncrypterPool(self._crypto, self._sync_db)
+ self._sync_enc_pool = SyncEncrypterPool(self._crypto, self._sync_db)
self._sync_watcher = TimerTask(self._encrypt_syncing_docs, delay=10)
self._sync_watcher.start()
@@ -1145,7 +1145,7 @@ class Soledad(object):
if self._db:
return self._db.resolve_doc(doc, conflicted_doc_revs)
- def sync(self):
+ def sync(self, decrypt_inline=False):
"""
Synchronize the local encrypted replica with a remote replica.
@@ -1155,11 +1155,15 @@ class Soledad(object):
:param url: the url of the target replica to sync with
:type url: str
- :return: the local generation before the synchronisation was
- performed.
+ :param decrypt_inline: Whether to do the decryption of received
+ messages inline or not.
+ :type decrypt_inline: bool
+
+ :return: The local generation before the synchronisation was
+ performed.
:rtype: str
"""
- #return
+ print "SYNC: inline? ", decrypt_inline
local_gen = None
if self._db:
# acquire lock before attempt to sync
@@ -1176,8 +1180,9 @@ class Soledad(object):
local_gen = self._db.sync(
urlparse.urljoin(
self.server_url, 'user-%s' % self._uuid),
- creds=self._creds, autocreate=True)
- #signal(SOLEDAD_DONE_DATA_SYNC, self._uuid)
+ creds=self._creds, autocreate=True,
+ decrypt_inline=decrypt_inline)
+ signal(SOLEDAD_DONE_DATA_SYNC, self._uuid)
except Exception as exc:
logger.error("error during soledad sync")
logger.exception(exc)
@@ -1388,7 +1393,7 @@ class Soledad(object):
return self._passphrase.encode('utf-8')
#
- # Symmetric encryption / decryption
+ # Symmetric encryption of syncing docs
#
def _encrypt_syncing_docs(self):
@@ -1396,6 +1401,8 @@ class Soledad(object):
Process the syncing queue and send the documents there
to be encrypted in the sync db. They will be read by the
SoledadSyncTarget during the sync_exchange.
+
+ Called periodical from the TimerTask self._sync_watcher.
"""
lock = self.encrypting_lock
# optional wait flag used to avoid blocking
@@ -1406,7 +1413,7 @@ class Soledad(object):
try:
while not queue.empty():
doc = queue.get_nowait()
- self._sync_pool.encrypt_doc(doc)
+ self._sync_enc_pool.encrypt_doc(doc)
except Exception as exc:
logger.error("Error while encrypting docs to sync")
logger.exception(exc)
diff --git a/client/src/leap/soledad/client/crypto.py b/client/src/leap/soledad/client/crypto.py
index 2ada4937f3937e66b1f94c9abf4a73612673770d..6d1fab37d50c327ef0663d492d48903c11d35a14 100644
--- a/client/src/leap/soledad/client/crypto.py
+++ b/client/src/leap/soledad/client/crypto.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
# crypto.py
-# Copyright (C) 2013,2014 LEAP
+# Copyright (C) 2013, 2014 LEAP
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -24,12 +24,15 @@ import hashlib
import json
import logging
import multiprocessing
+import threading
from pycryptopp.cipher.aes import AES
from pycryptopp.cipher.xsalsa20 import XSalsa20
+from zope.proxy import sameProxiedObjects
from leap.soledad.common import soledad_assert
from leap.soledad.common import soledad_assert_type
+from leap.soledad.common.document import SoledadDocument
from leap.soledad.common.crypto import (
@@ -346,14 +349,13 @@ def encrypt_docstr(docstr, doc_id, doc_rev, key, secret):
})
-# XXX change to docstr...
-def decrypt_doc(crypto, doc):
+def decrypt_doc_dict(doc_dict, doc_id, doc_rev, key, secret):
"""
Decrypt C{doc}'s content.
Return the JSON string representation of the document's decrypted content.
- The content of the document should have the following structure:
+ The passed doc_dict argument should have the following structure:
{
ENC_JSON_KEY: '<enc_blob>',
@@ -369,52 +371,67 @@ def decrypt_doc(crypto, doc):
EncryptionSchemes.SYMKEY and C{enc_method} is
EncryptionMethods.AES_256_CTR.
- :param crypto: A SoledadCryto instance to perform the encryption.
- :type crypto: leap.soledad.crypto.SoledadCrypto
- :param doc: The document to be decrypted.
- :type doc: SoledadDocument
+ :param doc_dict: The content of the document to be decrypted.
+ :type doc_dict: dict
+
+ :param doc_id: The document id.
+ :type doc_id: str
+
+ :param doc_rev: The document revision.
+ :type doc_rev: str
+
+ :param key: The key used to encrypt ``data`` (must be 256 bits long).
+ :type key: str
+
+ :param secret:
+ :type secret:
:return: The JSON serialization of the decrypted content.
:rtype: str
"""
- soledad_assert(doc.is_tombstone() is False)
- soledad_assert(ENC_JSON_KEY in doc.content)
- soledad_assert(ENC_SCHEME_KEY in doc.content)
- soledad_assert(ENC_METHOD_KEY in doc.content)
- soledad_assert(MAC_KEY in doc.content)
- soledad_assert(MAC_METHOD_KEY in doc.content)
+ # TODO where should we move these assertions, now that we're passed the
+ # string?
+ #soledad_assert(doc.is_tombstone() is False)
+
+ soledad_assert(ENC_JSON_KEY in doc_dict)
+ soledad_assert(ENC_SCHEME_KEY in doc_dict)
+ soledad_assert(ENC_METHOD_KEY in doc_dict)
+ soledad_assert(MAC_KEY in doc_dict)
+ soledad_assert(MAC_METHOD_KEY in doc_dict)
+
# verify MAC
ciphertext = binascii.a2b_hex( # content is stored as hex.
- doc.content[ENC_JSON_KEY])
+ doc_dict[ENC_JSON_KEY])
mac = mac_doc(
- doc.doc_id, doc.rev,
+ doc_id, doc_rev,
ciphertext,
- doc.content[MAC_METHOD_KEY], crypto.secret)
+ doc_dict[MAC_METHOD_KEY], secret)
# we compare mac's hashes to avoid possible timing attacks that might
# exploit python's builtin comparison operator behaviour, which fails
# immediatelly when non-matching bytes are found.
doc_mac_hash = hashlib.sha256(
binascii.a2b_hex( # the mac is stored as hex
- doc.content[MAC_KEY])).digest()
+ doc_dict[MAC_KEY])).digest()
calculated_mac_hash = hashlib.sha256(mac).digest()
if doc_mac_hash != calculated_mac_hash:
raise WrongMac('Could not authenticate document\'s contents.')
# decrypt doc's content
- enc_scheme = doc.content[ENC_SCHEME_KEY]
+ enc_scheme = doc_dict[ENC_SCHEME_KEY]
plainjson = None
if enc_scheme == EncryptionSchemes.SYMKEY:
- enc_method = doc.content[ENC_METHOD_KEY]
+ enc_method = doc_dict[ENC_METHOD_KEY]
if enc_method == EncryptionMethods.AES_256_CTR:
- soledad_assert(ENC_IV_KEY in doc.content)
- plainjson = crypto.decrypt_sym(
- ciphertext,
- crypto.doc_passphrase(doc.doc_id),
+ soledad_assert(ENC_IV_KEY in doc_dict)
+ plainjson = decrypt_sym(
+ ciphertext, key,
method=enc_method,
- iv=doc.content[ENC_IV_KEY])
+ iv=doc_dict[ENC_IV_KEY])
else:
raise UnknownEncryptionMethod(enc_method)
else:
raise UnknownEncryptionScheme(enc_scheme)
+
+ print "PLAIN: ", plainjson
return plainjson
@@ -451,6 +468,9 @@ class SyncEncryptDecryptPool(object):
:param sync_db: a database connection handle
:type sync_db: handle
+
+ :param insert_doc_cb: Optional callback for inserting doc.
+ :type insert_doc_cb: callable
"""
self._pool = multiprocessing.Pool(self.WORKERS)
self._crypto = crypto
@@ -492,9 +512,9 @@ class SyncEncrypterPool(SyncEncryptDecryptPool):
def encrypt_doc_cb(self, result):
doc_id, doc_rev, content = result
- self.insert_encrypted_doc(doc_id, doc_rev, content)
+ self.insert_encrypted_local_doc(doc_id, doc_rev, content)
- def insert_encrypted_doc(self, doc_id, doc_rev, content):
+ def insert_encrypted_local_doc(self, doc_id, doc_rev, content):
"""
Insert the contents of the encrypted doc into the local sync
database.
@@ -512,19 +532,216 @@ class SyncEncrypterPool(SyncEncryptDecryptPool):
self._sync_db.commit()
+def decrypt_doc_task(doc_id, doc_rev, content, gen, trans_id, key, secret):
+ decrypted_content = decrypt_doc_dict(
+ content, doc_id, doc_rev, key, secret)
+ return doc_id, doc_rev, decrypted_content, gen, trans_id
+
+
+def get_insertable_docs_by_gen(expected, got):
+ """
+ Return a list of documents ready to be inserted. This list is computed
+ by aligning the expected list with the already gotten docs, and returning
+ the maximum number of docs that can be processed in the expected order
+ before finding a gap.
+
+ :param expected: A list of generations to be inserted.
+ :type expected: list
+
+ :param got: A dictionary whose values are the docs to be inserted.
+ :type got: dict
+ """
+ ordered = [got.get(i) for i in expected]
+ if None in ordered:
+ return ordered[:ordered.index(None)]
+ else:
+ return ordered
+
+
class SyncDecrypterPool(SyncEncryptDecryptPool):
"""
Pool of workers that spawn subprocesses to execute the symmetric decryption
of documents that were received.
+
+ The decryption of the received documents is done in two steps:
+
+ 1. All the encrypted docs are collected, together with their generation
+ and transaction-id
+ 2. The docs are enqueued for decryption. When completed, they are
+ inserted following the generation order.
"""
WORKERS = 10
TABLE_NAME = "docs_received"
FIELD_NAMES = "doc_id, rev, content, gen, trans_id"
- def decrypt_doc(self, doc_id, rev):
+ write_encrypted_lock = threading.Lock()
+
+ def __init__(self, *args, **kwargs):
+ """
+ Initialize the decrypter pool, and setup a dict for putting the
+ results of the decrypted docs until they are picked by the insert
+ routine that gets them in order.
+ """
+ self._insert_doc_cb = kwargs.pop("insert_doc_cb")
+ SyncEncryptDecryptPool.__init__(self, *args)
+ self.decrypted_docs = {}
+
+ def insert_encrypted_received_doc(self, doc_id, doc_rev, content,
+ gen, trans_id):
+ """
+ Insert a received message with encrypted content, to be decrypted later
+ on.
+ """
+ docstr = json.dumps(content)
+ c = self._sync_db.cursor()
+ sql_ins = "INSERT INTO '%s' VALUES (?, ?, ?, ?, ?)" % (
+ self.TABLE_NAME,)
+ c.execute(sql_ins, (doc_id, doc_rev, docstr, gen, trans_id))
+ self._sync_db.commit()
+
+ def delete_encrypted_received_doc(self, doc_id, doc_rev):
+ """
+ Delete a encrypted received doc after it was inserted into the local
+ db.
+
+ :param doc_id: Document ID.
+ :type doc_id: str
+ :param doc_rev: Document revision.
+ :type doc_rev: str
+ """
+ c = self._sync_db.cursor()
+ sql_del = "DELETE FROM '%s' WHERE doc_id=? AND rev=?" % (
+ self.TABLE_NAME,)
+ c.execute(sql_del, (doc_id, doc_rev))
+ self._sync_db.commit()
+
+ def decrypt_doc(self, doc_id, rev, source_replica_uid):
"""
Symmetrically decrypt a document.
- :param doc: The document with contents to be encrypted.
- :type doc: SoledadDocument
+ :param doc_id: The ID for the document with contents to be encrypted.
+ :type doc: str
+ :param rev: The revision of the document.
+ :type rev: str
+ :param source_replica_uid:
+ :type source_replica_uid: str
+ """
+ self.source_replica_uid = source_replica_uid
+ if sameProxiedObjects(self._insert_doc_cb.get(source_replica_uid),
+ None):
+ print self._insert_doc_cb
+ logger.warning("No insert_doc callback, skipping decryption.")
+ return
+
+ # XXX move to get_doc function...
+ c = self._sync_db.cursor()
+ sql = "SELECT * FROM '%s' WHERE doc_id=? AND rev=?" % (
+ self.TABLE_NAME,)
+ c.execute(sql, (doc_id, rev))
+ res = c.fetchone()
+ if res is None:
+ logger.debug("Doc %s:%s does not exist in sync db" % (doc_id, rev))
+ return
+
+ doc_id, rev, docstr, gen, trans_id = res
+ content = json.loads(docstr)
+ key = self._crypto.doc_passphrase(doc_id)
+ secret = self._crypto.secret
+
+ args = doc_id, rev, content, gen, trans_id, key, secret
+
+ try:
+ self._pool.apply_async(decrypt_doc_task, args,
+ callback=self.decrypt_doc_cb)
+ except Exception as exc:
+ logger.exception(exc)
+
+ def decrypt_doc_cb(self, result):
+ """
+ Temporarily store the decryption result in a dictionary where it will
+ be picked by process_decrypted.
+
+ :param result: the result of the decryption routine.
+ :type result: tuple
+ """
+ doc_id, rev, content, gen, trans_id = result
+ self.decrypted_docs[gen] = result
+
+ def get_docs_by_generation(self):
+ """
+ Get all documents in the received table from the sync db,
+ ordered by generation.
+
+ :return: list of doc_id, rev, generation
+ """
+ c = self._sync_db.cursor()
+ sql = "SELECT doc_id, rev, gen FROM %s ORDER BY gen" % (
+ self.TABLE_NAME,)
+ c.execute(sql)
+ return c.fetchall()
+
+ def count_received_encrypted_docs(self):
+ """
+ Count how many documents we have in the table for received and
+ encrypted docs.
+
+ :return: The count of documents.
+ :rtype: int
+ """
+ c = self._sync_db.cursor()
+ sql = "SELECT COUNT(*) FROM %s" % (self.TABLE_NAME,)
+ c.execute(sql)
+ res = c.fetchone()
+ print "res"
+ if res is not None:
+ print ">>>>>>>>>> GOT %s received encrypted docs" % res[0]
+ return res[0]
+ else:
+ return 0
+
+ def decrypt_received_docs(self, source_replica_uid):
"""
+ Get all the encrypted documents from the sync database and dispatch a
+ decrypt worker to decrypt each one of them.
+ """
+ docs_by_generation = self.get_docs_by_generation()
+ for doc_id, rev, gen in docs_by_generation:
+ self.decrypt_doc(doc_id, rev, source_replica_uid)
+
+ def process_decrypted(self):
+ """
+ Process the already decrypted documents, and insert as many documents
+ as can be taken from the expected order without finding a gap.
+ """
+ # Acquire the lock to avoid processing while we're still
+ # getting data from the syncing stream, to avoid InvalidGeneration
+ # problems.
+ with self.write_encrypted_lock:
+ docs = self.get_docs_by_generation()
+ expected = [gen for doc_id, rev, gen in docs]
+ docs_to_insert = get_insertable_docs_by_gen(
+ expected, self.decrypted_docs)
+ for doc_fields in docs_to_insert:
+ self.insert_decrypted_local_doc(*doc_fields)
+
+ def insert_decrypted_local_doc(self, doc_id, doc_rev, content,
+ gen, trans_id):
+ """
+ Insert the decrypted document into the local sqlcipher database.
+ Makes use of the passed callback `return_doc_cb` passed to the caller
+ by u1db sync.
+ """
+ print "TRY TO INSERT GEN --->", gen
+ # could pass source_replica in params for callback chain
+ insert_fun = self._insert_doc_cb[self.source_replica_uid]
+ try:
+ doc = SoledadDocument(doc_id, doc_rev, content)
+ insert_fun(doc, int(gen), trans_id)
+ except Exception as exc:
+ logger.error("Error while inserting decrypted doc into local db")
+ logger.exception(exc)
+ else:
+ # If no errors found, remove it from the local temporary dict
+ # and from the received database.
+ self.decrypted_docs.pop(gen)
+ self.delete_encrypted_received_doc(doc_id, doc_rev)
diff --git a/client/src/leap/soledad/client/sqlcipher.py b/client/src/leap/soledad/client/sqlcipher.py
index c7cf79a2f7ece69cb03fa8a491cf34789d58df69..4e18847e51f92b8f052f0ab1661750e3ed535f24 100644
--- a/client/src/leap/soledad/client/sqlcipher.py
+++ b/client/src/leap/soledad/client/sqlcipher.py
@@ -54,6 +54,7 @@ from u1db.sync import Synchronizer
from u1db import errors as u1db_errors
from leap.soledad.client.target import SoledadSyncTarget
+from leap.soledad.client.target import PendingReceivedDocsSyncError
from leap.soledad.common.document import SoledadDocument
logger = logging.getLogger(__name__)
@@ -339,7 +340,7 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
crypto=crypto, raw_key=raw_key, cipher=cipher,
kdf_iter=kdf_iter, cipher_page_size=cipher_page_size)
- def sync(self, url, creds=None, autocreate=True):
+ def sync(self, url, creds=None, autocreate=True, decrypt_inline=False):
"""
Synchronize documents with remote replica exposed at url.
@@ -355,12 +356,21 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
:rtype: int
"""
print "***********************"
- print "SQLCIPHER: sync started"
+ print "SQLCIPHER: sync started. inline?", decrypt_inline
if not self.syncer:
self._create_syncer(url, creds=creds)
+ old_decrypt_inline = self.syncer.sync_target.decrypt_inline
+ print "SETTING TARGET decrypt_inline to ", decrypt_inline
+ self.syncer.sync_target.set_decrypt_inline(decrypt_inline)
+
try:
res = self.syncer.sync(autocreate=autocreate)
+
+ except PendingReceivedDocsSyncError:
+ logger.warning("Local sync db is not clear, skipping sync...")
+ return
+
except httplib.CannotSendRequest:
# raised when you reuse httplib.HTTP object for new request
# while you havn't called its getresponse()
@@ -371,10 +381,16 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
self._syncer = None
self._create_syncer(url, creds=creds)
print "SQLCIPHER: syncer created, about to sync..."
+ print "SETTING TARGET decrypt_inline to ", decrypt_inline
+ self.syncer.sync_target.set_decrypt_inline(decrypt_inline)
res = self.syncer.sync(autocreate=autocreate)
except Exception:
logger.error("error SQLITE sync")
raise
+ finally:
+ # restore the original decrypt inline behav
+ self.syncer.sync_target.set_decrypt_inline(old_decrypt_inline)
+
print "SQLCIPHER: sync DONE"
return res
diff --git a/client/src/leap/soledad/client/target.py b/client/src/leap/soledad/client/target.py
index dc2a0420b42d5158e9ca016b041b8df4a9b8aeb7..9e65b2df5445830d942082a00bcafcd770889f65 100644
--- a/client/src/leap/soledad/client/target.py
+++ b/client/src/leap/soledad/client/target.py
@@ -22,30 +22,33 @@ import cStringIO
import gzip
import logging
import os
+import re
import sqlite3
import urllib
-import simplejson as json
+from collections import defaultdict
from time import sleep
+import simplejson as json
+from taskthread import TimerTask
from u1db.remote import utils, http_errors
from u1db.errors import BrokenSyncStream
from u1db import errors
from u1db.remote.http_target import HTTPSyncTarget
from u1db.remote.http_client import _encode_query_parameter
+from zope.proxy import ProxyBase
+from zope.proxy import sameProxiedObjects, setProxiedObject
from leap.soledad.common.document import SoledadDocument
from leap.soledad.client.auth import TokenBasedAuth
-from leap.soledad.client.crypto import is_symmetrically_encrypted, decrypt_doc
+from leap.soledad.client.crypto import is_symmetrically_encrypted
+from leap.soledad.client.crypto import encrypt_docstr, decrypt_doc_dict
+from leap.soledad.client.crypto import SyncEncrypterPool, SyncDecrypterPool
from leap.common.check import leap_check
logger = logging.getLogger(__name__)
-#
-# Exceptions
-#
-
def _gunzip(data):
"""
@@ -65,10 +68,14 @@ def _gunzip(data):
return data
+class PendingReceivedDocsSyncError(Exception):
+ pass
+
#
# SoledadSyncTarget
#
+
class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
"""
A SyncTarget that encrypts data before sending and decrypts data after
@@ -80,6 +87,10 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
written to the main database.
"""
+ # will later keep a reference to the insert-doc callback
+ # passed to sync_exchange
+ _insert_doc_cb = defaultdict(lambda: ProxyBase(None))
+
#
# Modified HTTPSyncTarget methods.
#
@@ -109,10 +120,32 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
HTTPSyncTarget.__init__(self, url, creds)
self._crypto = crypto
+ print "URL : ", url
+ self.source_replica_uid = re.findall("user-([0-9a-fA-F]+)", url)[0]
+ print "uid -->", self.source_replica_uid
+
self._sync_db = None
if sync_db_path is not None:
self._init_sync_db(sync_db_path)
+ # whether to bypass the received messages decryption deferral
+ self._decrypt_inline = False
+
+ # initialize syncing queue decryption pool
+ self._sync_decr_pool = SyncDecrypterPool(
+ self._crypto, self._sync_db,
+ insert_doc_cb=self._insert_doc_cb)
+ self._sync_watcher = TimerTask(
+ self._decrypt_syncing_received_docs, delay=10)
+ self._sync_watcher.start()
+
+ def set_decrypt_inline(self, value):
+ self._decrypt_inline = value
+
+ @property
+ def decrypt_inline(self):
+ return self._decrypt_inline
+
@staticmethod
def connect(url, crypto=None):
return SoledadSyncTarget(url, crypto=crypto)
@@ -151,31 +184,57 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
raise BrokenSyncStream
data = parts[1:-1]
comma = False
+
+ queue_for_decrypt = (not self.decrypt_inline or
+ self._sync_db is None)
+ if queue_for_decrypt:
+ self._sync_decr_pool.write_encrypted_lock.acquire()
if data:
line, comma = utils.check_and_strip_comma(data[0])
res = json.loads(line)
if ensure_callback and 'replica_uid' in res:
ensure_callback(res['replica_uid'])
+
+ # XXX check that writing_incoming lock is not acquired ---------
+
for entry in data[1:]:
if not comma: # missing in between comma
raise BrokenSyncStream
line, comma = utils.check_and_strip_comma(entry)
entry = json.loads(line)
+ gen, trans_id = entry['gen'], entry['trans_id']
#-------------------------------------------------------------
# symmetric decryption of document's contents
#-------------------------------------------------------------
- # if arriving content was symmetrically encrypted, we decrypt
- # it.
+ # If arriving content was symmetrically encrypted, we decrypt
+ # it. We do it inline if decrypt_inline flag is True or no
+ # sync_db was defined, otherwise we defer it writing it to the
+ # received docs table.
+
doc = SoledadDocument(
entry['id'], entry['rev'], entry['content'])
- if doc.content and ENC_SCHEME_KEY in doc.content:
- if doc.content[ENC_SCHEME_KEY] == \
- EncryptionSchemes.SYMKEY:
- doc.set_json(decrypt_doc(self._crypto, doc))
+
+ if is_symmetrically_encrypted(doc):
+ if queue_for_decrypt:
+ print "ENQUEUING DECRYPT -----------------------"
+ self._save_encrypted_received_doc(doc, gen, trans_id)
+ else:
+ print "INLINE DECRYPT -------------------------"
+ # force inline decrypt, or no-db fallback, for tests
+ key = self._crypto.doc_passphrase(doc.doc_id)
+ secret = self._crypto.secret
+ doc.set_json(decrypt_doc_dict(
+ doc.content, doc.doc_id, doc.rev,
+ key, secret))
+ # XXX should release lock in the decrypt pool
+
#-------------------------------------------------------------
# end of symmetric decryption
#-------------------------------------------------------------
- return_doc_cb(doc, entry['gen'], entry['trans_id'])
+ if not queue_for_decrypt:
+ return_doc_cb(doc, gen, trans_id)
+ if queue_for_decrypt:
+ self._sync_decr_pool.write_encrypted_lock.release()
if parts[-1] != ']':
try:
partdic = json.loads(parts[-1])
@@ -304,6 +363,16 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
:return: The new generation and transaction id of the target replica.
:rtype: tuple
"""
+ self.source_replica_uid = source_replica_uid
+ print "SETTING SOURCE REPLICA UID to", source_replica_uid
+ # let the decrypter pool access the passed callback to insert docs
+ print "SETTING PROXY TO ------------>", return_doc_cb
+ setProxiedObject(self._insert_doc_cb[source_replica_uid],
+ return_doc_cb)
+
+ if not self.clear_to_sync():
+ raise PendingReceivedDocsSyncError
+
self._ensure_connection()
if self._trace_hook: # for tests
self._trace_hook('sync_exchange')
@@ -340,9 +409,12 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
if not doc.is_tombstone():
if self._sync_db is None:
# fallback case, for tests
+ key = self._crypto.doc_passphrase(doc.doc_id)
+ secret = self._crypto.secret
+
doc_json = encrypt_docstr(
json.dumps(doc.get_json()),
- doc.doc_id, doc.rev, self._crypto.secret)
+ doc.doc_id, doc.rev, key, secret)
else:
try:
doc_json = self.get_encrypted_doc_from_db(
@@ -378,6 +450,7 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
self.delete_encrypted_docs_from_db(synced)
data = None
+ print "SYNC EXCHANGE FINISHED: new generation -> %s" % res['new_generation']
return res['new_generation'], res['new_transaction_id']
#
@@ -439,9 +512,10 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
:param doc_rev: The document revision
:type doc_rev: str
"""
+ encr = SyncEncrypterPool
c = self._sync_db.cursor()
- # XXX interpolate table name
- sql = ("SELECT content FROM docs_tosync WHERE doc_id=? and rev=?")
+ sql = ("SELECT content FROM %s WHERE doc_id=? and rev=?" % (
+ encr.TABLE_NAME,))
c.execute(sql, (doc_id, doc_rev))
res = c.fetchall()
if len(res) != 0:
@@ -456,10 +530,51 @@ class SoledadSyncTarget(HTTPSyncTarget, TokenBasedAuth):
to be deleted.
:type docs_ids: any iterable of tuples of str
"""
+ encr = SyncEncrypterPool
c = self._sync_db.cursor()
for doc_id, doc_rev in docs_ids:
- # XXX interpolate table name
- sql = ("DELETE FROM docs_tosync "
- "WHERE doc_id=? and rev=?")
+ sql = ("DELETE FROM %s WHERE doc_id=? and rev=?" % (
+ encr.TABLE_NAME,))
c.execute(sql, (doc_id, doc_rev))
self._sync_db.commit()
+
+ def _save_encrypted_received_doc(self, doc, gen, trans_id):
+ """
+ Save an incoming document into the received docs table in the sync db.
+
+ :param doc: The document to save.
+ :type doc: SoledadDocument
+ :param gen: The generation.
+ :type gen: str
+ :param trans_id: Transacion id.
+ :type gen: str
+ """
+ self._sync_decr_pool.insert_encrypted_received_doc(
+ doc.doc_id, doc.rev, doc.content, gen, trans_id)
+
+ #
+ # Symmetric decryption of syncing docs
+ #
+
+ def clear_to_sync(self):
+ """
+ Return True if sync can proceed (ie, the received db table is empty).
+ :rtype: bool
+ """
+ return self._sync_decr_pool.count_received_encrypted_docs() == 0
+
+ def _decrypt_syncing_received_docs(self):
+ """
+ Decrypt the documents received from remote replica and insert them
+ into the local one.
+
+ Called periodically from TimerTask self._sync_watcher.
+ """
+ if sameProxiedObjects(self._insert_doc_cb.get(self.source_replica_uid),
+ None):
+ logger.warning("No insert_doc callback, skipping decryption.")
+ return
+
+ decrypter = self._sync_decr_pool
+ decrypter.decrypt_received_docs(self.source_replica_uid)
+ decrypter.process_decrypted()