Changeset 446 for elixir

Timestamp:

02/02/09 17:25:51 (3 years ago)

Author:

ged

Message:

- Fixed encrypted extension to not encrypt several times an instance attributes

when that instance is flushed several times before being expunged from the
session.

- added note about how the encrypted extension works

Location:

elixir/trunk

Files:

• elixir/ext/encrypted.py (modified) (3 diffs)
• tests/test_encryption.py (modified) (4 diffs)

elixir/trunk/elixir/ext/encrypted.py

@@ -25 +25 @@
 ssn columns on save, update, and load.  Different secrets can be specified on
 an entity by entity basis, for added security.
+
+**Important note**: instance attributes are encrypted in-place. This means that
+if one of the encrypted attributes of an instance is accessed after the
+instance has been flushed to the database (and thus encrypted), the value for
+that attribute will be crypted in the in-memory object in addition to the
+database row.
 '''
 
-from Crypto.Cipher          import Blowfish
-from elixir.statements      import Statement
-from sqlalchemy.orm         import MapperExtension, EXT_CONTINUE
+from Crypto.Cipher import Blowfish
+from elixir.statements import Statement
+from sqlalchemy.orm import MapperExtension, EXT_CONTINUE, EXT_STOP
 
 try:
@@ -62 +68 @@
     def __init__(self, entity, for_fields=[], with_secret='abcdef'):
 
-        def perform_encryption(instance, decrypt=False):
+        def perform_encryption(instance, encrypt=True):
+            encrypted = getattr(instance, '_elixir_encrypted', None)
+            if encrypted is encrypt:
+                # skipping encryption or decryption, as it is already done
+                return
+            else:
+                # marking instance as already encrypted/decrypted
+                instance._elixir_encrypted = encrypt
+
+            if encrypt:
+                func = encrypt_value
+            else:
+                func = decrypt_value
+
             for column_name in for_fields:
                 current_value = getattr(instance, column_name)
                 if current_value:
-                    if decrypt:
-                        new_value = decrypt_value(current_value, with_secret)
-                    else:
-                        new_value = encrypt_value(current_value, with_secret)
-                    setattr(instance, column_name, new_value)
+                    setattr(instance, column_name,
+                            func(current_value, with_secret))
 
         def perform_decryption(instance):
-            perform_encryption(instance, decrypt=True)
+            perform_encryption(instance, encrypt=False)
 
         class EncryptedMapperExtension(MapperExtension):
@@ -85 +101 @@
                 return EXT_CONTINUE
 
-        if SA05orlater:
-            def reconstruct_instance(self, mapper, instance):
-                perform_decryption(instance)
-                return True
-            EncryptedMapperExtension.reconstruct_instance = reconstruct_instance
-        else:
-            def populate_instance(self, mapper, selectcontext, row, instance,
-                                  *args, **kwargs):
-                mapper.populate_instance(selectcontext, instance, row,
-                                         *args, **kwargs)
-                perform_decryption(instance)
-                return True
-            EncryptedMapperExtension.populate_instance = populate_instance
+            if SA05orlater:
+                def reconstruct_instance(self, mapper, instance):
+                    perform_decryption(instance)
+                    # no special return value is required for
+                    # reconstruct_instance, but you never know...
+                    return EXT_CONTINUE
+            else:
+                def populate_instance(self, mapper, selectcontext, row,
+                                      instance, *args, **kwargs):
+                    mapper.populate_instance(selectcontext, instance, row,
+                                             *args, **kwargs)
+                    perform_decryption(instance)
+                    # EXT_STOP because we already did populate the instance and
+                    # the normal processing should not happen
+                    return EXT_STOP
 
         # make sure that the entity's mapper has our mapper extension

elixir/trunk/tests/test_encryption.py

@@ -21 +21 @@
 
     metadata.bind = 'sqlite:///'
+    setup_all()
 
 
@@ -29 +30 @@
 class TestEncryption(object):
     def setup(self):
-        setup_all(True)
+        create_all()
 
     def teardown(self):
         drop_all()
-        session.clear()
+        session.close()
 
     def test_encryption(self):
         jonathan = Person(
-            name=u'Jonathan LaCour',
-            password=u's3cr3tw0RD',
-            ssn=u'123-45-6789'
+            name='Jonathan LaCour',
+            password='s3cr3tw0RD',
+            ssn='123-45-6789'
         )
         winston = Pet(
@@ -51 +52 @@
         jonathan.pets = [winston, nelson]
 
-        session.commit(); session.clear()
+        session.commit()
+        session.clear()
 
         p = Person.get_by(name='Jonathan LaCour')
@@ -63 +65 @@
         p.password = 'N3wpAzzw0rd'
 
-        session.commit(); session.clear()
+        session.commit()
+        session.clear()
 
         p = Person.get_by(name='Jonathan LaCour')
         assert p.password == 'N3wpAzzw0rd'
-        p.name = 'Jon LaCour'
 
-        session.commit(); session.clear()
+    def test_two_consecutive_updates(self):
+        jonathan = Person(
+            name='Jonathan LaCour',
+            password='s3cr3tw0RD',
+            ssn='123-45-6789'
+        )
+        session.commit()
+        session.clear()
+
+        p = Person.get_by(name='Jonathan LaCour')
+        assert p.password == 's3cr3tw0RD'
+        p.name = 'JONATHAN LACOUR'
+        session.flush()
+
+        assert p.password == 'r\\x9d\\xa8\\xb4\\x8d|\\xffp\\xf5\\x0e'
+
+        p.name = 'Jonathan LaCour'
+        session.flush()
+
+        # check that it is not further encrypted
+        assert p.password == 'r\\x9d\\xa8\\xb4\\x8d|\\xffp\\xf5\\x0e'
+
+        session.commit()
+        session.clear()
+
+        p = Person.get_by(name='Jonathan LaCour')
+        assert p.password == 's3cr3tw0RD'
+