summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/leap/soledad/__init__.py179
-rw-r--r--src/leap/soledad/backends/leap_backend.py41
-rw-r--r--src/leap/soledad/backends/sqlcipher.py35
-rw-r--r--src/leap/soledad/tests/__init__.py7
-rw-r--r--src/leap/soledad/tests/test_couch.py1
-rw-r--r--src/leap/soledad/tests/test_crypto.py22
-rw-r--r--src/leap/soledad/tests/test_leap_backend.py6
-rw-r--r--src/leap/soledad/tests/test_soledad.py3
8 files changed, 90 insertions, 204 deletions
diff --git a/src/leap/soledad/__init__.py b/src/leap/soledad/__init__.py
index 3c4fc801..a27a586d 100644
--- a/src/leap/soledad/__init__.py
+++ b/src/leap/soledad/__init__.py
@@ -29,7 +29,7 @@ remote storage in the server side.
import os
import string
import random
-import hmac
+import hashlib
import configparser
import re
try:
@@ -38,9 +38,12 @@ except ImportError:
import json # noqa
+from binascii import b2a_base64
+from hashlib import sha256
+
+
from leap.common import events
#from leap.common.keymanager.gpgwrapper import GPGWrapper
-from leap.soledad.util import GPGWrapper
from leap.soledad.config import SoledadConfig
from leap.soledad.backends import sqlcipher
from leap.soledad.backends.leap_backend import (
@@ -176,7 +179,7 @@ class Soledad(object):
# TODO: log each bootstrap step.
# Stage 0 - Local environment setup
self._init_dirs()
- self._gpg = GPGWrapper(gnupghome=self._config.get_gnupg_home())
+ self._crypto = SoledadCrypto(self._config.get_gnupg_home())
if self._config.get_shared_db_url() and self._auth_token:
# TODO: eliminate need to create db here.
self._shared_db = SoledadSharedDatabase.open_database(
@@ -194,7 +197,7 @@ class Soledad(object):
self._init_keys()
else:
self._set_symkey(self.decrypt(doc.content['_symkey'],
- passphrase=self._user_hash()))
+ passphrase=self._hash_user()))
# Stage 2 - Keys synchronization
self._assert_server_keys()
# Stage 3 - Local database initialization
@@ -259,7 +262,7 @@ class Soledad(object):
self._symkey,
create=True,
document_factory=LeapDocument,
- soledad=self)
+ crypto=self._crypto)
def close(self):
"""
@@ -271,9 +274,6 @@ class Soledad(object):
# Management of secret for symmetric encryption
#-------------------------------------------------------------------------
- # TODO: refactor the following methods to somewhere out of here
- # (a new class SoledadCrypto, maybe?)
-
def _has_symkey(self):
"""
Verify if a key for symmetric encryption exists in a local encrypted
@@ -289,12 +289,12 @@ class Soledad(object):
# is it symmetrically encrypted?
f = open(self._config.get_secret_path(), 'r')
content = f.read()
- if not self.is_encrypted_sym(content):
+ if not self._crypto.is_encrypted_sym(content):
raise DocumentNotEncrypted(
"File %s is not encrypted!" % self._config.get_secret_path())
# can we decrypt it?
- result = self._gpg.decrypt(content, passphrase=self._passphrase)
- return result.status == 'decryption ok'
+ cyphertext = self._crypto.decrypt(content, passphrase=self._passphrase)
+ return bool(cyphertext)
def _load_symkey(self):
"""
@@ -305,8 +305,9 @@ class Soledad(object):
"encryption but it does not exist on disk.")
try:
with open(self._config.get_secret_path()) as f:
- self._symkey = str(
- self._gpg.decrypt(f.read(), passphrase=self._passphrase))
+ self._symkey = \
+ self._crypto.decrypt(f.read(), passphrase=self._passphrase)
+ self._crypto.symkey = self._symkey
except IOError:
raise IOError('Failed to open secret file %s.' %
self._config.get_secret_path())
@@ -333,11 +334,13 @@ class Soledad(object):
"symmetric encryption but it already "
"exists on disk.")
self._symkey = symkey
+ self._crypto.symkey = self._symkey
self._store_symkey()
def _store_symkey(self):
- ciphertext = self._gpg.encrypt(self._symkey, '', symmetric=True,
- passphrase=self._passphrase)
+ ciphertext = self._crypto.encrypt(
+ self._symkey, symmetric=True,
+ passphrase=self._passphrase)
f = open(self._config.get_secret_path(), 'w')
f.write(str(ciphertext))
f.close()
@@ -367,15 +370,16 @@ class Soledad(object):
"""
self._gen_symkey()
- def _user_hash(self):
+ def _hash_user(self):
"""
Calculate a hash for storing/retrieving key material on shared
- database, based on user's email.
+ database, based on user's address.
@return: the hash
@rtype: str
"""
- return hmac.new(self._user, 'user').hexdigest()
+ return b2a_base64(
+ sha256('user-%s' % self._user).digest())[:-1]
def _get_keys_doc(self):
"""
@@ -389,7 +393,7 @@ class Soledad(object):
# TODO: change below to raise appropriate exceptions
if not self._shared_db:
return None
- doc = self._shared_db.get_doc_unauth(self._user_hash())
+ doc = self._shared_db.get_doc_unauth(self._hash_user())
events.signal(events.events_pb2.SOLEDAD_DONE_DOWNLOADING_KEYS, self._user)
return doc
@@ -403,139 +407,19 @@ class Soledad(object):
doc = self._get_keys_doc()
if doc:
remote_symkey = self.decrypt(doc.content['_symkey'],
- passphrase=self._user_hash())
+ passphrase=self._hash_user())
assert remote_symkey == self._symkey
else:
events.signal(events.events_pb2.SOLEDAD_UPLOADING_KEYS, self._user)
content = {
'_symkey': self.encrypt(self._symkey),
}
- doc = LeapDocument(doc_id=self._user_hash(), soledad=self)
+ doc = LeapDocument(doc_id=self._hash_user(), crypto=self._crypto)
doc.content = content
self._shared_db.put_doc(doc)
events.signal(events.events_pb2.SOLEDAD_DONE_UPLOADING_KEYS, self._user)
#-------------------------------------------------------------------------
- # Data encryption and decryption
- #-------------------------------------------------------------------------
-
- def encrypt(self, data, fingerprint=None, sign=None, passphrase=None,
- symmetric=False):
- """
- Encrypt data.
-
- @param data: the data to be encrypted
- @type data: str
- @param sign: the fingerprint of key to be used for signature
- @type sign: str
- @param passphrase: the passphrase to be used for encryption
- @type passphrase: str
- @param symmetric: whether the encryption scheme should be symmetric
- @type symmetric: bool
-
- @return: the encrypted data
- @rtype: str
- """
- return str(self._gpg.encrypt(data, fingerprint, sign=sign,
- passphrase=passphrase,
- symmetric=symmetric))
-
- def encrypt_symmetric(self, doc_id, data, sign=None):
- """
- Encrypt data using a password.
-
- The password is derived from the document id and the secret for
- symmetric encryption previously generated/loaded.
-
- @param doc_id: the document id
- @type doc_id: str
- @param data: the data to be encrypted
- @type data: str
- @param sign: the fingerprint of key to be used for signature
- @type sign: str
-
- @return: the encrypted data
- @rtype: str
- """
- return self.encrypt(data, sign=sign,
- passphrase=self._hmac_passphrase(doc_id),
- symmetric=True)
-
- def decrypt(self, data, passphrase=None):
- """
- Decrypt data.
-
- @param data: the data to be decrypted
- @type data: str
- @param passphrase: the passphrase to be used for decryption
- @type passphrase: str
-
- @return: the decrypted data
- @rtype: str
- """
- return str(self._gpg.decrypt(data, passphrase=passphrase))
-
- def decrypt_symmetric(self, doc_id, data):
- """
- Decrypt data using symmetric secret.
-
- @param doc_id: the document id
- @type doc_id: str
- @param data: the data to be decrypted
- @type data: str
-
- @return: the decrypted data
- @rtype: str
- """
- return self.decrypt(data, passphrase=self._hmac_passphrase(doc_id))
-
- def _hmac_passphrase(self, doc_id):
- """
- Generate a passphrase for symmetric encryption.
-
- The password is derived from the document id and the secret for
- symmetric encryption previously generated/loaded.
-
- @param doc_id: the document id
- @type doc_id: str
-
- @return: the passphrase
- @rtype: str
- """
- return hmac.new(self._symkey, doc_id).hexdigest()
-
- def is_encrypted(self, data):
- """
- Test whether some chunk of data is a cyphertext.
-
- @param data: the data to be tested
- @type data: str
-
- @return: whether the data is a cyphertext
- @rtype: bool
- """
- return self._gpg.is_encrypted(data)
-
- def is_encrypted_sym(self, data):
- """
- Test whether some chunk of data was encrypted with a symmetric key.
-
- @return: whether data is encrypted to a symmetric key
- @rtype: bool
- """
- return self._gpg.is_encrypted_sym(data)
-
- def is_encrypted_asym(self, data):
- """
- Test whether some chunk of data was encrypted to an OpenPGP private
- key.
-
- @return: whether data is encrypted to an OpenPGP private key
- @rtype: bool
- """
- return self._gpg.is_encrypted_asym(data)
-
- #-------------------------------------------------------------------------
# Document storage, retrieval and sync
#-------------------------------------------------------------------------
@@ -699,7 +583,7 @@ class Soledad(object):
@rtype: bool
"""
# TODO: create auth scheme for sync with server
- target = LeapSyncTarget(url, creds=None, soledad=self)
+ target = LeapSyncTarget(url, creds=None, crypto=self._crypto)
info = target.get_sync_info(self._db._get_replica_uid())
# compare source generation with target's last known source generation
if self._db._get_generation() != info[4]:
@@ -743,9 +627,9 @@ class Soledad(object):
'symkey': self._symkey,
})
if passphrase:
- data = str(self._gpg.encrypt(data, None, sign=None,
- passphrase=passphrase,
- symmetric=True))
+ data = self._crypto.encrypt(data, None, sign=None,
+ passphrase=passphrase,
+ symmetric=True)
return data
def import_recovery_document(self, data, passphrase=None):
@@ -761,14 +645,15 @@ class Soledad(object):
if self._has_keys():
raise KeyAlreadyExists("You tried to import a recovery document "
"but secret keys are already present.")
- if passphrase and not self._gpg.is_encrypted_sym(data):
+ if passphrase and not self._crypto.is_encrypted_sym(data):
raise DocumentNotEncrypted("You provided a password but the "
"recovery document is not encrypted.")
if passphrase:
- data = str(self._gpg.decrypt(data, passphrase=passphrase))
+ data = str(self._crypto.decrypt(data, passphrase=passphrase))
data = json.loads(data)
self._user = data['user']
self._symkey = data['symkey']
+ self._crypto.symkey = self._symkey
self._store_symkey()
# TODO: make this work well with bootstrap.
self._load_keys()
diff --git a/src/leap/soledad/backends/leap_backend.py b/src/leap/soledad/backends/leap_backend.py
index 4ea131c0..687b59ef 100644
--- a/src/leap/soledad/backends/leap_backend.py
+++ b/src/leap/soledad/backends/leap_backend.py
@@ -42,7 +42,7 @@ class NoDefaultKey(Exception):
pass
-class NoSoledadInstance(Exception):
+class NoSoledadCryptoInstance(Exception):
"""
Exception to signal that no Soledad instance was found.
"""
@@ -69,7 +69,7 @@ class LeapDocument(Document):
"""
def __init__(self, doc_id=None, rev=None, json='{}', has_conflicts=False,
- encrypted_json=None, soledad=None, syncable=True):
+ encrypted_json=None, crypto=None, syncable=True):
"""
Container for handling an encryptable document.
@@ -84,14 +84,14 @@ class LeapDocument(Document):
@param encrypted_json: The encrypted JSON string for this document. If
given, the decrypted value supersedes any raw json string given.
@type encrypted_json: str
- @param soledad: An instance of Soledad so we can encrypt/decrypt
+ @param crypto: An instance of SoledadCrypto so we can encrypt/decrypt
document contents when syncing.
- @type soledad: soledad.Soledad
+ @type crypto: soledad.Soledad
@param syncable: Should this document be synced with remote replicas?
@type syncable: bool
"""
Document.__init__(self, doc_id, rev, json, has_conflicts)
- self._soledad = soledad
+ self._crypto = crypto
self._syncable = syncable
if encrypted_json:
self.set_encrypted_json(encrypted_json)
@@ -103,17 +103,20 @@ class LeapDocument(Document):
@return: The encrypted JSON serialization of document's contents.
@rtype: str
"""
- if not self._soledad:
- raise NoSoledadInstance()
- return self._soledad.encrypt_symmetric(self.doc_id,
- self.get_json())
+ if not self._crypto:
+ raise NoSoledadCryptoInstance()
+ return self._crypto.encrypt_symmetric(
+ self.get_json(),
+ self._crypto._hash_passphrase(self.doc_id))
def set_encrypted_content(self, cyphertext):
"""
Decrypt C{cyphertext} and set document's content.
contents.
"""
- plaintext = self._soledad.decrypt_symmetric(self.doc_id, cyphertext)
+ plaintext = self._crypto.decrypt_symmetric(
+ cyphertext,
+ self._crypto._hash_passphrase(self.doc_id))
self.set_json(plaintext)
def get_encrypted_json(self):
@@ -131,8 +134,8 @@ class LeapDocument(Document):
Set document's content based on a valid JSON string containing the
encrypted document's contents.
"""
- if not self._soledad:
- raise NoSoledadInstance()
+ if not self._crypto:
+ raise NoSoledadCryptoInstance()
cyphertext = json.loads(encrypted_json)['_encrypted_json']
self.set_encrypted_content(cyphertext)
@@ -196,7 +199,7 @@ class LeapSyncTarget(HTTPSyncTarget):
receiving.
"""
- def __init__(self, url, creds=None, soledad=None):
+ def __init__(self, url, creds=None, crypto=None):
"""
Initialize the LeapSyncTarget.
@@ -210,7 +213,7 @@ class LeapSyncTarget(HTTPSyncTarget):
@type soledad: soledad.Soledad
"""
HTTPSyncTarget.__init__(self, url, creds)
- self._soledad = soledad
+ self._crypto = crypto
def _parse_sync_stream(self, data, return_doc_cb, ensure_callback=None):
"""
@@ -244,15 +247,15 @@ class LeapSyncTarget(HTTPSyncTarget):
line, comma = utils.check_and_strip_comma(entry)
entry = json.loads(line)
# decrypt after receiving from server.
- if not self._soledad:
- raise NoSoledadInstance()
+ if not self._crypto:
+ raise NoSoledadCryptoInstance()
enc_json = json.loads(entry['content'])['_encrypted_json']
- if not self._soledad.is_encrypted_sym(enc_json):
+ if not self._crypto.is_encrypted_sym(enc_json):
raise DocumentNotEncrypted(
"Incoming document from sync is not encrypted.")
doc = LeapDocument(entry['id'], entry['rev'],
encrypted_json=entry['content'],
- soledad=self._soledad)
+ crypto=self._crypto)
return_doc_cb(doc, entry['gen'], entry['trans_id'])
if parts[-1] != ']':
try:
@@ -300,7 +303,7 @@ class LeapSyncTarget(HTTPSyncTarget):
# encrypt and verify before sending to server.
enc_json = json.loads(
doc.get_encrypted_json())['_encrypted_json']
- if not self._soledad.is_encrypted_sym(enc_json):
+ if not self._crypto.is_encrypted_sym(enc_json):
raise DocumentNotEncrypted(
"Could not encrypt document before sync.")
size += prepare(id=doc.doc_id, rev=doc.rev,
diff --git a/src/leap/soledad/backends/sqlcipher.py b/src/leap/soledad/backends/sqlcipher.py
index a4d53fa8..288680d4 100644
--- a/src/leap/soledad/backends/sqlcipher.py
+++ b/src/leap/soledad/backends/sqlcipher.py
@@ -38,7 +38,7 @@ from u1db import (
from leap.soledad.backends.leap_backend import LeapDocument
-def open(path, password, create=True, document_factory=None, soledad=None):
+def open(path, password, create=True, document_factory=None, crypto=None):
"""Open a database at the given location.
Will raise u1db.errors.DatabaseDoesNotExist if create=False and the
@@ -58,7 +58,7 @@ def open(path, password, create=True, document_factory=None, soledad=None):
"""
return SQLCipherDatabase.open_database(
path, password, create=create, document_factory=document_factory,
- soledad=soledad)
+ crypto=crypto)
class DatabaseIsNotEncrypted(Exception):
@@ -78,7 +78,7 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
db_handle.cursor().execute("PRAGMA key = '%s'" % key)
def __init__(self, sqlcipher_file, password, document_factory=None,
- soledad=None):
+ crypto=None):
"""
Create a new sqlcipher file.
@@ -89,23 +89,24 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
@param document_factory: A function that will be called with the same
parameters as Document.__init__.
@type document_factory: callable
- @param soledad: An instance of Soledad so we can encrypt/decrypt
+ @param crypto: An instance of SoledadCrypto so we can encrypt/decrypt
document contents when syncing.
- @type soledad: soledad.Soledad
+ @type crypto: soledad.crypto.SoledadCrypto
"""
self._check_if_db_is_encrypted(sqlcipher_file)
self._db_handle = dbapi2.connect(sqlcipher_file)
SQLCipherDatabase.set_pragma_key(self._db_handle, password)
self._real_replica_uid = None
self._ensure_schema()
- self._soledad = soledad
+ self._crypto = crypto
def factory(doc_id=None, rev=None, json='{}', has_conflicts=False,
encrypted_json=None, syncable=True):
return LeapDocument(doc_id=doc_id, rev=rev, json=json,
has_conflicts=has_conflicts,
encrypted_json=encrypted_json,
- syncable=syncable, soledad=self._soledad)
+ syncable=syncable,
+ crypto=self._crypto)
self.set_document_factory(factory)
def _check_if_db_is_encrypted(self, sqlcipher_file):
@@ -131,7 +132,7 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
@classmethod
def _open_database(cls, sqlcipher_file, password, document_factory=None,
- soledad=None):
+ crypto=None):
"""
Open a SQLCipher database.
@@ -142,9 +143,9 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
@param document_factory: A function that will be called with the same
parameters as Document.__init__.
@type document_factory: callable
- @param soledad: An instance of Soledad so we can encrypt/decrypt
+ @param crypto: An instance of SoledadCrypto so we can encrypt/decrypt
document contents when syncing.
- @type soledad: soledad.Soledad
+ @type crypto: soledad.crypto.SoledadCrypto
@return: The database object.
@rtype: SQLCipherDatabase
@@ -171,11 +172,11 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
time.sleep(cls.WAIT_FOR_PARALLEL_INIT_HALF_INTERVAL)
return SQLCipherDatabase._sqlite_registry[v](
sqlcipher_file, password, document_factory=document_factory,
- soledad=soledad)
+ crypto=crypto)
@classmethod
def open_database(cls, sqlcipher_file, password, create, backend_cls=None,
- document_factory=None, soledad=None):
+ document_factory=None, crypto=None):
"""
Open a SQLCipher database.
@@ -191,9 +192,9 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
@param document_factory: A function that will be called with the same
parameters as Document.__init__.
@type document_factory: callable
- @param soledad: An instance of Soledad so we can encrypt/decrypt
+ @param crypto: An instance of SoledadCrypto so we can encrypt/decrypt
document contents when syncing.
- @type soledad: soledad.Soledad
+ @type crypto: soledad.crypto.SoledadCrypto
@return: The database object.
@rtype: SQLCipherDatabase
@@ -201,7 +202,7 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
try:
return cls._open_database(sqlcipher_file, password,
document_factory=document_factory,
- soledad=soledad)
+ crypto=crypto)
except errors.DatabaseDoesNotExist:
if not create:
raise
@@ -211,7 +212,7 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
backend_cls = SQLCipherDatabase
return backend_cls(sqlcipher_file, password,
document_factory=document_factory,
- soledad=soledad)
+ crypto=crypto)
def sync(self, url, creds=None, autocreate=True):
"""
@@ -234,7 +235,7 @@ class SQLCipherDatabase(sqlite_backend.SQLitePartialExpandDatabase):
self,
LeapSyncTarget(url,
creds=creds,
- soledad=self._soledad)).sync(autocreate=autocreate)
+ crypto=self._crypto)).sync(autocreate=autocreate)
def _extra_schema_init(self, c):
"""
diff --git a/src/leap/soledad/tests/__init__.py b/src/leap/soledad/tests/__init__.py
index 7f1e9f28..396b2775 100644
--- a/src/leap/soledad/tests/__init__.py
+++ b/src/leap/soledad/tests/__init__.py
@@ -4,7 +4,7 @@ Tests to make sure Soledad provides U1DB functionality and more.
import u1db
from leap.soledad import Soledad
-from leap.soledad.util import GPGWrapper
+from leap.soledad.crypto import SoledadCrypto
from leap.soledad.backends.leap_backend import LeapDocument
from leap.common.testing.basetest import BaseLeapTest
@@ -33,8 +33,8 @@ class BaseSoledadTest(BaseLeapTest):
# initialize soledad by hand so we can control keys
self._soledad = self._soledad_instance(user=self.email)
self._soledad._init_dirs()
- self._soledad._gpg = GPGWrapper(gnupghome=self.gnupg_home)
#self._soledad._gpg.import_keys(PUBLIC_KEY)
+ self._soledad._crypto = SoledadCrypto(self.gnupg_home)
if not self._soledad._has_symkey():
self._soledad._gen_symkey()
self._soledad._load_symkey()
@@ -57,9 +57,6 @@ class BaseSoledadTest(BaseLeapTest):
local_db_path=self.tempdir+prefix+local_db_path,
bootstrap=bootstrap)
- def _gpgwrapper_instance(self):
- return GPGWrapper(gnupghome="%s/gnupg" % self.tempdir)
-
# Key material for testing
KEY_FINGERPRINT = "E36E738D69173C13D709E44F2F455E2824D18DDF"
diff --git a/src/leap/soledad/tests/test_couch.py b/src/leap/soledad/tests/test_couch.py
index a6171dd8..008c3ca4 100644
--- a/src/leap/soledad/tests/test_couch.py
+++ b/src/leap/soledad/tests/test_couch.py
@@ -60,7 +60,6 @@ class CouchDBWrapper(object):
os.mkdir(os.path.join(self.tempdir, 'lib'))
os.mkdir(os.path.join(self.tempdir, 'log'))
args = ['couchdb', '-n', '-a', confPath]
- print args
#null = open('/dev/null', 'w')
self.process = subprocess.Popen(
args, env=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
diff --git a/src/leap/soledad/tests/test_crypto.py b/src/leap/soledad/tests/test_crypto.py
index ee3b6c89..f762437a 100644
--- a/src/leap/soledad/tests/test_crypto.py
+++ b/src/leap/soledad/tests/test_crypto.py
@@ -32,7 +32,7 @@ from leap.soledad.tests import (
PRIVATE_KEY,
)
from leap.soledad import KeyAlreadyExists
-from leap.soledad.util import GPGWrapper
+from leap.soledad.crypto import SoledadCrypto
try:
import simplejson as json
@@ -49,11 +49,11 @@ class EncryptedSyncTestCase(BaseSoledadTest):
"""
Test getting and setting encrypted content.
"""
- doc1 = LeapDocument(soledad=self._soledad)
+ doc1 = LeapDocument(crypto=self._soledad._crypto)
doc1.content = {'key': 'val'}
doc2 = LeapDocument(doc_id=doc1.doc_id,
encrypted_json=doc1.get_encrypted_json(),
- soledad=self._soledad)
+ crypto=self._soledad._crypto)
res1 = doc1.get_json()
res2 = doc2.get_json()
self.assertEqual(res1, res2, 'incorrect document encryption')
@@ -62,12 +62,12 @@ class EncryptedSyncTestCase(BaseSoledadTest):
"""
Test for successful symmetric encryption.
"""
- doc1 = LeapDocument(soledad=self._soledad)
+ doc1 = LeapDocument(crypto=self._soledad._crypto)
doc1.content = {'key': 'val'}
enc_json = json.loads(doc1.get_encrypted_json())['_encrypted_json']
self.assertEqual(
True,
- self._soledad._gpg.is_encrypted_sym(enc_json),
+ self._soledad._crypto.is_encrypted_sym(enc_json),
"could not encrypt with passphrase.")
@@ -87,12 +87,12 @@ class RecoveryDocumentTestCase(BaseSoledadTest):
def test_export_recovery_document_crypt(self):
rd = self._soledad.export_recovery_document('123456')
self.assertEqual(True,
- self._soledad._gpg.is_encrypted_sym(rd))
+ self._soledad._crypto.is_encrypted_sym(rd))
data = {
'user': self._soledad._user,
'symkey': self._soledad._symkey,
}
- raw_data = json.loads(str(self._soledad._gpg.decrypt(
+ raw_data = json.loads(str(self._soledad._crypto.decrypt(
rd,
passphrase='123456')))
self.assertEqual(
@@ -111,7 +111,7 @@ class RecoveryDocumentTestCase(BaseSoledadTest):
gnupg_home = self.gnupg_home = "%s/gnupg2" % self.tempdir
s = self._soledad_instance(user='anotheruser@leap.se', prefix='/2')
s._init_dirs()
- s._gpg = GPGWrapper(gnupghome=gnupg_home)
+ s._crypto = SoledadCrypto(gnupg_home)
s.import_recovery_document(rd, None)
self.assertEqual(self._soledad._user,
s._user, 'Failed setting user email.')
@@ -124,7 +124,7 @@ class RecoveryDocumentTestCase(BaseSoledadTest):
gnupg_home = self.gnupg_home = "%s/gnupg2" % self.tempdir
s = self._soledad_instance(user='anotheruser@leap.se', prefix='3')
s._init_dirs()
- s._gpg = GPGWrapper(gnupghome=gnupg_home)
+ s._crypto = SoledadCrypto(gnupg_home)
s.import_recovery_document(rd, '123456')
self.assertEqual(self._soledad._user,
s._user, 'Failed setting user email.')
@@ -138,7 +138,7 @@ class CryptoMethodsTestCase(BaseSoledadTest):
def test__gen_symkey(self):
sol = self._soledad_instance(user='user@leap.se', prefix='/3')
sol._init_dirs()
- sol._gpg = GPGWrapper(gnupghome="%s/gnupg3" % self.tempdir)
+ sol._crypto = SoledadCrypto("%s/3/gnupg" % self.tempdir)
self.assertFalse(sol._has_symkey(), "Should not have a symkey at "
"this point")
sol._gen_symkey()
@@ -147,7 +147,7 @@ class CryptoMethodsTestCase(BaseSoledadTest):
def test__has_keys(self):
sol = self._soledad_instance(user='leap@leap.se', prefix='/5')
sol._init_dirs()
- sol._gpg = GPGWrapper(gnupghome=self.tempdir+"/5/gnupg")
+ sol._crypto = SoledadCrypto("%s/5/gnupg" % self.tempdir)
self.assertFalse(sol._has_keys())
sol._gen_symkey()
self.assertTrue(sol._has_keys())
diff --git a/src/leap/soledad/tests/test_leap_backend.py b/src/leap/soledad/tests/test_leap_backend.py
index 9056355f..fd9ef85d 100644
--- a/src/leap/soledad/tests/test_leap_backend.py
+++ b/src/leap/soledad/tests/test_leap_backend.py
@@ -28,7 +28,7 @@ def make_leap_document_for_test(test, doc_id, rev, content,
has_conflicts=False):
return leap_backend.LeapDocument(
doc_id, rev, content, has_conflicts=has_conflicts,
- soledad=test._soledad)
+ crypto=test._soledad._crypto)
def make_leap_encrypted_document_for_test(test, doc_id, rev, encrypted_content,
@@ -36,7 +36,7 @@ def make_leap_encrypted_document_for_test(test, doc_id, rev, encrypted_content,
return leap_backend.LeapDocument(
doc_id, rev, encrypted_json=encrypted_content,
has_conflicts=has_conflicts,
- soledad=test._soledad)
+ crypto=test._soledad.crypto)
LEAP_SCENARIOS = [
@@ -134,7 +134,7 @@ class TestLeapParsingSyncStream(test_remote_sync_target.TestParsingSyncStream):
self.assertRaises(u1db.errors.BrokenSyncStream,
tgt._parse_sync_stream, "[\r\n{},\r\n]", None)
- self.assertRaises(leap_backend.NoSoledadInstance,
+ self.assertRaises(leap_backend.NoSoledadCryptoInstance,
tgt._parse_sync_stream,
'[\r\n{},\r\n{"id": "i", "rev": "r", '
'"content": "{}", "gen": 3, "trans_id": "T-sid"}'
diff --git a/src/leap/soledad/tests/test_soledad.py b/src/leap/soledad/tests/test_soledad.py
index 61d131f1..b849c310 100644
--- a/src/leap/soledad/tests/test_soledad.py
+++ b/src/leap/soledad/tests/test_soledad.py
@@ -32,6 +32,7 @@ except ImportError:
from leap.soledad.tests import BaseSoledadTest
from leap.soledad import Soledad
+from leap.soledad.crypto import SoledadCrypto
class AuxMethodsTestCase(BaseSoledadTest):
@@ -49,7 +50,7 @@ class AuxMethodsTestCase(BaseSoledadTest):
def test__init_db(self):
sol = self._soledad_instance()
sol._init_dirs()
- sol._gpg = self._gpgwrapper_instance()
+ sol._crypto = SoledadCrypto(self.tempdir+'/gnupg')
#self._soledad._gpg.import_keys(PUBLIC_KEY)
if not sol._has_symkey():
sol._gen_symkey()
generated by cgit v1.2.3 (git 2.25.1) at 2025-10-31 04:52:46 +0000