abdallah
/
server-env
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[REF] data_encryption: Black python code

This commit is contained in:
Thomas Binsfeld 2020-10-02 11:21:36 +02:00 committed by Florian da Costa
parent 36016cbff1
commit df274a2e41
4 changed files with 109 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
data_encryption/__manifest__.py
View File

@ -4,21 +4,14 @@
"name": "Encryption data", "name": "Encryption data",
"summary": "Store accounts and credentials encrypted by environment", "summary": "Store accounts and credentials encrypted by environment",
"version": "12.0.1.0.0", "version": "12.0.1.0.0",
"development_status": 'Alpha', "development_status": "Alpha",
"category": "Tools", "category": "Tools",
"website": "https://github/oca/server-env", "website": "https://github/oca/server-env",
"author": "Akretion, Odoo Community Association (OCA)", "author": "Akretion, Odoo Community Association (OCA)",
"license": "AGPL-3", "license": "AGPL-3",
"application": False, "application": False,
"installable": True, "installable": True,
"external_dependencies": { "external_dependencies": {"python": ["cryptography"],},
"python": [ "depends": ["base",],
'cryptography'], "data": ["security/ir.model.access.csv",],
},
"depends": [
"base",
],
"data": [
"security/ir.model.access.csv",
],
} }

90
data_encryption/models/encrypted_data.py
View File

@ -19,22 +19,26 @@ except ImportError as err: # pragma: no cover
class EncryptedData(models.Model): class EncryptedData(models.Model):
"""Model to store encrypted data by environment (prod, preprod...)""" """Model to store encrypted data by environment (prod, preprod...)"""
_name = 'encrypted.data' _name = "encrypted.data"
_description = 'Store any encrypted data by environment' _description = "Store any encrypted data by environment"
name = fields.Char( name = fields.Char(
required=True, readonly=True, index=True, required=True, readonly=True, index=True, help="Technical name"
help="Technical name") )
environment = fields.Char( environment = fields.Char(
required=True, required=True,
index=True, index=True,
help="Concerned Odoo environment (prod, preprod...)") help="Concerned Odoo environment (prod, preprod...)",
)
encrypted_data = fields.Binary() encrypted_data = fields.Binary()
_sql_constraints = [ _sql_constraints = [
('name_environment_uniq', 'unique (name, environment)', (
'You can not store multiple encrypted data for the same record and \ "name_environment_uniq",
environment') "unique (name, environment)",
"You can not store multiple encrypted data for the same record and \
environment",
)
] ]
def _decrypt_data(self, env): def _decrypt_data(self, env):
@ -43,29 +47,34 @@ class EncryptedData(models.Model):
try: try:
return cipher.decrypt(self.encrypted_data).decode() return cipher.decrypt(self.encrypted_data).decode()
except InvalidToken: except InvalidToken:
raise ValidationError(_( raise ValidationError(
"Password has been encrypted with a different " _(
"key. Unless you can recover the previous key, " "Password has been encrypted with a different "
"this password is unreadable.")) "key. Unless you can recover the previous key, "
"this password is unreadable."
)
)
@api.model @api.model
@ormcache('self._uid', 'name', 'env') @ormcache("self._uid", "name", "env")
def _encrypted_get(self, name, env=None): def _encrypted_get(self, name, env=None):
if self.env.context.get('bin_size'): if self.env.context.get("bin_size"):
self = self.with_context(bin_size=False) self = self.with_context(bin_size=False)
if not self.env.user._is_superuser(): if not self.env.user._is_superuser():
raise AccessError( raise AccessError(
_("Encrypted data can only be read as superuser")) _("Encrypted data can only be read as superuser")
)
if not env: if not env:
env = self._retrieve_env() env = self._retrieve_env()
encrypted_rec = self.search( encrypted_rec = self.search(
[('name', '=', name), ('environment', '=', env)]) [("name", "=", name), ("environment", "=", env)]
)
if not encrypted_rec: if not encrypted_rec:
return None return None
return encrypted_rec._decrypt_data(env) return encrypted_rec._decrypt_data(env)
@api.model @api.model
@ormcache('self._uid', 'name', 'env') @ormcache("self._uid", "name", "env")
def _encrypted_read_json(self, name, env=None): def _encrypted_read_json(self, name, env=None):
data = self._encrypted_get(name, env=env) data = self._encrypted_get(name, env=env)
if not data: if not data:
@ -74,18 +83,22 @@ class EncryptedData(models.Model):
return json.loads(data) return json.loads(data)
except (ValueError, TypeError): except (ValueError, TypeError):
raise ValidationError( raise ValidationError(
_("The data you are trying to read are not in a json format")) _("The data you are trying to read are not in a json format")
)
@staticmethod @staticmethod
def _retrieve_env(): def _retrieve_env():
"""Return the current environment """Return the current environment
Raise if none is found Raise if none is found
""" """
current = config.get('running_env', False) current = config.get("running_env", False)
if not current: if not current:
raise ValidationError( raise ValidationError(
_('No environment found, please check your running_env ' _(
'entry in your config file.')) "No environment found, please check your running_env "
"entry in your config file."
)
)
return current return current
@classmethod @classmethod
@ -94,12 +107,15 @@ class EncryptedData(models.Model):
force_env = name of the env key. force_env = name of the env key.
Useful for encoding against one precise env Useful for encoding against one precise env
""" """
key_name = 'encryption_key_%s' % env key_name = "encryption_key_%s" % env
key_str = config.get(key_name) key_str = config.get(key_name)
if not key_str: if not key_str:
raise ValidationError(_( raise ValidationError(
"No '%s' entry found in config file. " _(
"Use a key similar to: %s") % (key_name, Fernet.generate_key()) "No '%s' entry found in config file. "
"Use a key similar to: %s"
)
% (key_name, Fernet.generate_key())
) )
# key should be in bytes format # key should be in bytes format
key = key_str.encode() key = key_str.encode()
@ -110,26 +126,28 @@ class EncryptedData(models.Model):
cipher = self._get_cipher(env) cipher = self._get_cipher(env)
if not isinstance(data, bytes): if not isinstance(data, bytes):
data = data.encode() data = data.encode()
return cipher.encrypt(data or '') return cipher.encrypt(data or "")
@api.model @api.model
def _encrypted_store(self, name, data, env=None): def _encrypted_store(self, name, data, env=None):
if not self.env.user._is_superuser(): if not self.env.user._is_superuser():
raise AccessError( raise AccessError(_("You can only encrypt data as superuser"))
_("You can only encrypt data as superuser"))
if not env: if not env:
env = self._retrieve_env() env = self._retrieve_env()
encrypted_data = self._encrypt_data(data, env) encrypted_data = self._encrypt_data(data, env)
existing_data = self.search( existing_data = self.search(
[('name', '=', name), ('environment', '=', env)]) [("name", "=", name), ("environment", "=", env)]
)
if existing_data: if existing_data:
existing_data.write({'encrypted_data': encrypted_data}) existing_data.write({"encrypted_data": encrypted_data})
else: else:
self.create({ self.create(
'name': name, {
'environment': env, "name": name,
'encrypted_data': encrypted_data, "environment": env,
}) "encrypted_data": encrypted_data,
}
)
self._encrypted_get.clear_cache(self) self._encrypted_get.clear_cache(self)
self._encrypted_read_json.clear_cache(self) self._encrypted_read_json.clear_cache(self)

16
data_encryption/tests/common.py
View File

@ -15,15 +15,14 @@ except ImportError as err: # pragma: no cover
class CommonDataEncrypted(TransactionCase): class CommonDataEncrypted(TransactionCase):
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.encrypted_data = self.env['encrypted.data'] self.encrypted_data = self.env["encrypted.data"]
self.set_new_key_env('test') self.set_new_key_env("test")
self.old_running_env = config.get('running_env', '') self.old_running_env = config.get("running_env", "")
config['running_env'] = 'test' config["running_env"] = "test"
self.crypted_data_name = 'test_model,1' self.crypted_data_name = "test_model,1"
def set_new_key_env(self, environment): def set_new_key_env(self, environment):
crypting_key = Fernet.generate_key() crypting_key = Fernet.generate_key()
@ -31,9 +30,8 @@ class CommonDataEncrypted(TransactionCase):
# the key com from the config file and is not in a binary format. # the key com from the config file and is not in a binary format.
# So we decode here to avoid having a special behavior because of # So we decode here to avoid having a special behavior because of
# the tests. # the tests.
config['encryption_key_{}'.format(environment)] = \ config["encryption_key_{}".format(environment)] = crypting_key.decode()
crypting_key.decode()
def tearDown(self): def tearDown(self):
config['running_env'] = self.old_running_env config["running_env"] = self.old_running_env
return super().tearDown() return super().tearDown()

71
data_encryption/tests/test_data_encrypt.py
View File

@ -17,83 +17,100 @@ except ImportError as err: # pragma: no cover
class TestDataEncrypted(CommonDataEncrypted): class TestDataEncrypted(CommonDataEncrypted):
def test_store_data_no_superuser(self): def test_store_data_no_superuser(self):
# only superuser can use this model # only superuser can use this model
admin = self.env.ref('base.user_admin') admin = self.env.ref("base.user_admin")
with self.assertRaises(AccessError): with self.assertRaises(AccessError):
self.encrypted_data.sudo(admin.id)._encrypted_store( self.encrypted_data.sudo(admin.id)._encrypted_store(
self.crypted_data_name, "My config") self.crypted_data_name, "My config"
)
def test_store_data_noenv_set(self): def test_store_data_noenv_set(self):
config.pop('running_env', None) config.pop("running_env", None)
with self.assertRaises(ValidationError): with self.assertRaises(ValidationError):
self.encrypted_data.sudo()._encrypted_store( self.encrypted_data.sudo()._encrypted_store(
self.crypted_data_name, "My config") self.crypted_data_name, "My config"
)
def test_store_data_nokey_set(self): def test_store_data_nokey_set(self):
config.pop('encryption_key_test', None) config.pop("encryption_key_test", None)
with self.assertRaises(ValidationError): with self.assertRaises(ValidationError):
self.encrypted_data.sudo()._encrypted_store( self.encrypted_data.sudo()._encrypted_store(
self.crypted_data_name, "My config") self.crypted_data_name, "My config"
)
def test_get_data_decrypted_and_cache(self): def test_get_data_decrypted_and_cache(self):
self.encrypted_data.sudo()._encrypted_store( self.encrypted_data.sudo()._encrypted_store(
'test_model,1', "My config") "test_model,1", "My config"
)
data = self.encrypted_data.sudo()._encrypted_get( data = self.encrypted_data.sudo()._encrypted_get(
self.crypted_data_name) self.crypted_data_name
)
self.assertEqual(data, "My config") self.assertEqual(data, "My config")
# Test cache really depends on user (super user) else any user could # Test cache really depends on user (super user) else any user could
# access the data # access the data
admin = self.env.ref('base.user_admin') admin = self.env.ref("base.user_admin")
with self.assertRaises(AccessError): with self.assertRaises(AccessError):
self.encrypted_data.sudo(admin)._encrypted_get( self.encrypted_data.sudo(admin)._encrypted_get(
self.crypted_data_name) self.crypted_data_name
)
# Change value should invalidate cache # Change value should invalidate cache
self.encrypted_data.sudo()._encrypted_store( self.encrypted_data.sudo()._encrypted_store(
'test_model,1', "Other Config") "test_model,1", "Other Config"
)
new_data = self.encrypted_data.sudo()._encrypted_get( new_data = self.encrypted_data.sudo()._encrypted_get(
self.crypted_data_name) self.crypted_data_name
)
self.assertEqual(new_data, "Other Config") self.assertEqual(new_data, "Other Config")
def test_get_data_wrong_key(self): def test_get_data_wrong_key(self):
self.encrypted_data.sudo()._encrypted_store( self.encrypted_data.sudo()._encrypted_store(
'test_model,1', "My config") "test_model,1", "My config"
)
new_key = Fernet.generate_key() new_key = Fernet.generate_key()
config['encryption_key_test'] = new_key.decode() config["encryption_key_test"] = new_key.decode()
with self.assertRaises(ValidationError): with self.assertRaises(ValidationError):
self.encrypted_data.sudo()._encrypted_get( self.encrypted_data.sudo()._encrypted_get(self.crypted_data_name)
self.crypted_data_name)
def test_get_empty_data(self): def test_get_empty_data(self):
empty_data = self.encrypted_data.sudo()._encrypted_get( empty_data = self.encrypted_data.sudo()._encrypted_get(
self.crypted_data_name) self.crypted_data_name
)
self.assertEqual(empty_data, None) self.assertEqual(empty_data, None)
def test_get_wrong_json(self): def test_get_wrong_json(self):
self.encrypted_data.sudo()._encrypted_store( self.encrypted_data.sudo()._encrypted_store(
self.crypted_data_name, 'config') self.crypted_data_name, "config"
)
with self.assertRaises(ValidationError): with self.assertRaises(ValidationError):
self.encrypted_data.sudo()._encrypted_read_json( self.encrypted_data.sudo()._encrypted_read_json(
self.crypted_data_name) self.crypted_data_name
)
def test_get_good_json(self): def test_get_good_json(self):
self.encrypted_data.sudo()._encrypted_store_json( self.encrypted_data.sudo()._encrypted_store_json(
self.crypted_data_name, {'key': 'value'}) self.crypted_data_name, {"key": "value"}
)
data = self.encrypted_data.sudo()._encrypted_read_json( data = self.encrypted_data.sudo()._encrypted_read_json(
self.crypted_data_name) self.crypted_data_name
self.assertEqual(data, {'key': 'value'}) )
self.assertEqual(data, {"key": "value"})
def test_get_empty_json(self): def test_get_empty_json(self):
data = self.encrypted_data.sudo()._encrypted_read_json( data = self.encrypted_data.sudo()._encrypted_read_json(
self.crypted_data_name) self.crypted_data_name
)
self.assertEqual(data, {}) self.assertEqual(data, {})
def test_get_data_with_bin_size_context(self): def test_get_data_with_bin_size_context(self):
self.encrypted_data.sudo()._encrypted_store( self.encrypted_data.sudo()._encrypted_store(
self.crypted_data_name, "test") self.crypted_data_name, "test"
data = self.encrypted_data.sudo().with_context(bin_size=True).\ )
_encrypted_get(self.crypted_data_name) data = (
self.encrypted_data.sudo()
.with_context(bin_size=True)
._encrypted_get(self.crypted_data_name)
)
self.assertEqual(data, "test") self.assertEqual(data, "test")