From 027cd6cd2980c1dd5943392fbb4f3eefb7ccfaed Mon Sep 17 00:00:00 2001
From: drebs <drebs@leap.se>
Date: Mon, 29 Apr 2013 17:36:22 -0300
Subject: Switch string literals to string constants.
---
src/leap/soledad/__init__.py | 34 ++++++++++++++++++++-----------
src/leap/soledad/backends/leap_backend.py | 27 ++++++++++++++++--------
src/leap/soledad/tests/test_crypto.py | 17 +++++++++-------
src/leap/soledad/tests/test_sqlcipher.py | 15 ++++++++------
4 files changed, 59 insertions(+), 34 deletions(-)
(limited to 'src')
diff --git a/src/leap/soledad/__init__.py b/src/leap/soledad/__init__.py
index 3b7aadea..6ae82b4d 100644
--- a/src/leap/soledad/__init__.py
+++ b/src/leap/soledad/__init__.py
@@ -28,9 +28,6 @@ remote storage in the server side.
import os
import string
-import hashlib
-import configparser
-import re
import binascii
import logging
try:
@@ -43,6 +40,7 @@ from hashlib import sha256
from leap.common import events
+from leap.common.check import leap_assert
from leap.soledad.config import SoledadConfig
from leap.soledad.backends import sqlcipher
from leap.soledad.backends.leap_backend import (
@@ -121,6 +119,12 @@ class Soledad(object):
The length of the secret used for symmetric encryption.
"""
+ SYMKEY_KEY = '_symkey'
+ ADDRESS_KEY = '_address'
+ """
+ Key used to access symmetric keys in recovery documents.
+ """
+
def __init__(self, address, passphrase, config_path=None,
secret_path=None, local_db_path=None,
shared_db_url=None, auth_token=None, bootstrap=True):
@@ -208,7 +212,7 @@ class Soledad(object):
else:
self._set_symkey(
self._crypto.decrypt_sym(
- doc.content['_symkey'],
+ doc.content[self.KEY_SYMKEY],
passphrase=self._address_hash()))
# Stage 2 - Keys synchronization
self._assert_server_keys()
@@ -416,20 +420,26 @@ class Soledad(object):
"""
Assert our key copies are the same as server's ones.
"""
- assert self._has_keys()
+ leap_assert(
+ self._has_keys(),
+ 'Tried to send keys to server but they don\'t exist in local '
+ 'storage.')
if not self._shared_db:
return
doc = self._fetch_keys_from_shared_db()
if doc:
remote_symkey = self.decrypt_sym(
- doc.content['_symkey'],
+ doc.content[self.SYMKEY_KEY],
passphrase=self._address_hash())
- assert remote_symkey == self._symkey
+ leap_assert(
+ remote_symkey == self._symkey,
+ 'Local and remote symmetric secrets differ!')
else:
events.signal(
events.events_pb2.SOLEDAD_UPLOADING_KEYS, self._address)
content = {
- '_symkey': self.encrypt_sym(self._symkey, self._passphrase),
+ self.SYMKEY_KEY: self.encrypt_sym(
+ self._symkey, self._passphrase),
}
doc = LeapDocument(doc_id=self._address_hash())
doc.content = content
@@ -744,8 +754,8 @@ class Soledad(object):
@rtype: str
"""
data = json.dumps({
- 'address': self._address,
- 'symkey': self._symkey,
+ self.ADDRESS_KEY: self._address,
+ self.SYMKEY_KEY: self._symkey,
})
if passphrase:
data = self._crypto.encrypt_sym(data, passphrase)
@@ -770,8 +780,8 @@ class Soledad(object):
if passphrase:
data = self._crypto.decrypt_sym(data, passphrase=passphrase)
data = json.loads(data)
- self._address = data['address']
- self._symkey = data['symkey']
+ self._address = data[self.ADDRESS_KEY]
+ self._symkey = data[self.SYMKEY_KEY]
self._crypto.symkey = self._symkey
self._store_symkey()
# TODO: make this work well with bootstrap.
diff --git a/src/leap/soledad/backends/leap_backend.py b/src/leap/soledad/backends/leap_backend.py
index 51c471eb..9750ffad 100644
--- a/src/leap/soledad/backends/leap_backend.py
+++ b/src/leap/soledad/backends/leap_backend.py
@@ -39,6 +39,10 @@ from leap.common.keymanager import KeyManager
from leap.common.check import leap_assert
+#
+# Exceptions
+#
+
class NoDefaultKey(Exception):
"""
Exception to signal that there's no default OpenPGP key configured.
@@ -84,6 +88,11 @@ class EncryptionSchemes(object):
# Crypto utilities for a LeapDocument.
#
+ENC_JSON_KEY = '_enc_json'
+ENC_SCHEME_KEY = '_enc_scheme'
+MAC_KEY = '_mac'
+
+
def encrypt_doc_json(crypto, doc_id, doc_json):
"""
Return a valid JSON string containing the C{doc} content encrypted to
@@ -92,7 +101,7 @@ def encrypt_doc_json(crypto, doc_id, doc_json):
The returned JSON string is the serialization of the following dictionary:
{
- '_encrypted_json': encrypt_sym(doc_content),
+ ENC_JSON_KEY: encrypt_sym(doc_content),
'_encryption_scheme: 'symkey',
}
@@ -112,8 +121,8 @@ def encrypt_doc_json(crypto, doc_id, doc_json):
if not crypto.is_encrypted_sym(ciphertext):
raise DocumentNotEncrypted('Failed encrypting document.')
return json.dumps({
- '_encrypted_json': ciphertext,
- '_encryption_scheme': EncryptionSchemes.SYMKEY,
+ ENC_JSON_KEY: ciphertext,
+ ENC_SCHEME_KEY: EncryptionSchemes.SYMKEY,
})
@@ -126,8 +135,8 @@ def decrypt_doc_json(crypto, doc_id, doc_json):
following dictionary:
{
- '_encrypted_json': enc_blob,
- '_encryption_scheme': enc_scheme,
+ ENC_JSON_KEY: enc_blob,
+ ENC_SCHEME_KEY: enc_scheme,
}
C{enc_blob} is the encryption of the JSON serialization of the document's
@@ -150,8 +159,8 @@ def decrypt_doc_json(crypto, doc_id, doc_json):
leap_assert(isinstance(doc_json, str))
leap_assert(doc_json != '')
content = json.loads(doc_json)
- ciphertext = content['_encrypted_json']
- enc_scheme = content['_encryption_scheme']
+ ciphertext = content[ENC_JSON_KEY]
+ enc_scheme = content[ENC_SCHEME_KEY]
plainjson = None
if enc_scheme == EncryptionSchemes.SYMKEY:
if not crypto.is_encrypted_sym(ciphertext):
@@ -315,8 +324,8 @@ class LeapSyncTarget(HTTPSyncTarget):
# if arriving content was symmetrically encrypted, we decrypt
# it.
doc = LeapDocument(entry['id'], entry['rev'], entry['content'])
- if doc.content and '_encryption_scheme' in doc.content:
- if doc.content['_encryption_scheme'] == \
+ if doc.content and ENC_SCHEME_KEY in doc.content:
+ if doc.content[ENC_SCHEME_KEY] == \
EncryptionSchemes.SYMKEY:
doc.set_json(
decrypt_doc_json(
diff --git a/src/leap/soledad/tests/test_crypto.py b/src/leap/soledad/tests/test_crypto.py
index 5d494818..101b5d83 100644
--- a/src/leap/soledad/tests/test_crypto.py
+++ b/src/leap/soledad/tests/test_crypto.py
@@ -35,6 +35,9 @@ from leap.soledad.backends.leap_backend import (
decrypt_doc_json,
EncryptionSchemes,
LeapSyncTarget,
+ ENC_JSON_KEY,
+ ENC_SCHEME_KEY,
+ MAC_KEY,
)
from leap.soledad.backends.couch import CouchDatabase
from leap.soledad import KeyAlreadyExists, Soledad
@@ -79,7 +82,7 @@ class EncryptedSyncTestCase(BaseSoledadTest):
enc_json = json.loads(
encrypt_doc_json(
self._soledad._crypto,
- doc1.doc_id, doc1.get_json()))['_encrypted_json']
+ doc1.doc_id, doc1.get_json()))[ENC_JSON_KEY]
self.assertEqual(
True,
self._soledad._crypto.is_encrypted_sym(enc_json),
@@ -182,8 +185,8 @@ class RecoveryDocumentTestCase(BaseSoledadTest):
rd = self._soledad.export_recovery_document(None)
self.assertEqual(
{
- 'address': self._soledad._address,
- 'symkey': self._soledad._symkey
+ self._soledad.ADDRESS_KEY: self._soledad._address,
+ self._soledad.SYMKEY_KEY: self._soledad._symkey
},
json.loads(rd),
"Could not export raw recovery document."
@@ -194,12 +197,12 @@ class RecoveryDocumentTestCase(BaseSoledadTest):
self.assertEqual(True,
self._soledad._crypto.is_encrypted_sym(rd))
data = {
- 'address': self._soledad._address,
- 'symkey': self._soledad._symkey,
+ self._soledad.ADDRESS_KEY: self._soledad._address,
+ self._soledad.SYMKEY_KEY: self._soledad._symkey,
}
- raw_data = json.loads(str(self._soledad._crypto.decrypt_sym(
+ raw_data = json.loads(self._soledad._crypto.decrypt_sym(
rd,
- passphrase='123456')))
+ passphrase='123456'))
self.assertEqual(
raw_data,
data,
diff --git a/src/leap/soledad/tests/test_sqlcipher.py b/src/leap/soledad/tests/test_sqlcipher.py
index 73388202..6b2889d6 100644
--- a/src/leap/soledad/tests/test_sqlcipher.py
+++ b/src/leap/soledad/tests/test_sqlcipher.py
@@ -31,6 +31,9 @@ from leap.soledad.backends.leap_backend import (
EncryptionSchemes,
decrypt_doc_json,
encrypt_doc_json,
+ ENC_JSON_KEY,
+ ENC_SCHEME_KEY,
+ MAC_KEY,
)
# u1db tests stuff.
@@ -518,7 +521,7 @@ class SQLCipherDatabaseSyncTests(
# make sure db2 now has the exact same thing
doc1 = self.db1.get_doc('doc')
doc2 = self.db1.get_doc('doc')
- if '_encryption_scheme' in doc2.content:
+ if ENC_SCHEME_KEY in doc2.content:
doc2.set_json(
decrypt_doc_json(
self._soledad._crypto, doc2, doc2.get_json()))
@@ -579,7 +582,7 @@ class SQLCipherDatabaseSyncTests(
self.assertFalse(doc2.has_conflicts)
self.sync(self.db2, db3)
doc3 = db3.get_doc('the-doc')
- if '_encryption_scheme' in doc3.content:
+ if ENC_SCHEME_KEY in doc3.content:
doc3.set_json(
decrypt_doc_json(
self._soledad._crypto, doc3.doc_id, doc3.get_json()))
@@ -595,7 +598,7 @@ class SQLCipherDatabaseSyncTests(
doc.doc_id, doc.rev, tests.simple_doc, False)
doc2 = self.db2.get_doc(doc.doc_id)
# decrypt to compare it it is the case
- if '_encryption_scheme' in doc2.content:
+ if ENC_SCHEME_KEY in doc2.content:
doc2 = self.db2.get_doc(doc.doc_id)
doc2.set_json(
decrypt_doc_json(
@@ -652,7 +655,7 @@ class SQLCipherSyncTargetTests(
last_known_trans_id=None, return_doc_cb=self.receive_doc)
# decrypt doc1 for comparison if needed
tmpdoc = self.db.get_doc('doc-id')
- if '_encryption_scheme' in tmpdoc.content:
+ if ENC_SCHEME_KEY in tmpdoc.content:
tmpdoc.set_json(
decrypt_doc_json(
self._soledad._crypto, tmpdoc.doc_id,
@@ -681,7 +684,7 @@ class SQLCipherSyncTargetTests(
last_known_trans_id=None, return_doc_cb=self.receive_doc)
# decrypt doc1 for comparison if needed
tmpdoc1 = self.db.get_doc('doc-id')
- if '_encryption_scheme' in tmpdoc1.content:
+ if ENC_SCHEME_KEY in tmpdoc1.content:
tmpdoc1.set_json(
decrypt_doc_json(
self._soledad._crypto, tmpdoc1.doc_id,
@@ -691,7 +694,7 @@ class SQLCipherSyncTargetTests(
self.assertFalse(tmpdoc1.has_conflicts)
# decrypt doc2 for comparison if needed
tmpdoc2 = self.db.get_doc('doc-id2')
- if '_encryption_scheme' in tmpdoc2.content:
+ if ENC_SCHEME_KEY in tmpdoc2.content:
tmpdoc2.set_json(
decrypt_doc_json(
self._soledad._crypto, tmpdoc2.doc_id,
--
cgit v1.2.3