Extracted serialiser for encryption into a flask module

- allows mocking easier
- shared across methods
- not built everytime

app/__init__.py

@@ -13,13 +13,14 @@ from utils import logging
 from notify_client import NotifyAPIClient
 from app.celery.celery import NotifyCelery
 from app.clients.sms.twilio import TwilioClient
+from app.encryption import Encryption
 
 db = SQLAlchemy()
 ma = Marshmallow()
 notify_alpha_client = NotifyAPIClient()
 notify_celery = NotifyCelery()
 twilio_client = TwilioClient()
-
+encryption = Encryption()
 
 api_user = LocalProxy(lambda: _request_ctx_stack.top.api_user)
 
@@ -36,6 +37,7 @@ def create_app(config_name, config_overrides=None):
     logging.init_app(application)
     twilio_client.init_app(application)
     notify_celery.init_app(application)
+    encryption.init_app(application)
 
     from app.service.rest import service as service_blueprint
     from app.user.rest import user as user_blueprint

app/celery/tasks.py

@@ -1,5 +1,5 @@
 from itsdangerous import URLSafeSerializer
-from app import notify_celery, twilio_client, db
+from app import notify_celery, twilio_client, db, encryption
 from app.clients.sms.twilio import TwilioClientException
 from app.dao.templates_dao import get_model_templates
 from app.models import Notification
@@ -7,10 +7,8 @@ from flask import current_app
 
 
 @notify_celery.task(name="send-sms", bind="True")
-def send_sms(service_id, notification_id, encrypted_notification, secret_key, salt):
+def send_sms(service_id, notification_id, encrypted_notification):
-    serializer = URLSafeSerializer(secret_key)
+    notification = encryption.decrypt(encrypted_notification)
-
-    notification = serializer.loads(encrypted_notification, salt=salt)
     template = get_model_templates(notification['template'])
 
     status = 'sent'

app/encryption.py

@@ -1,5 +1,20 @@
 from flask.ext.bcrypt import generate_password_hash, check_password_hash
 
+from itsdangerous import URLSafeSerializer
+
+
+class Encryption:
+
+    def init_app(self, app):
+        self.serializer = URLSafeSerializer(app.config.get('SECRET_KEY'))
+        self.salt = app.config.get('DANGEROUS_SALT')
+
+    def encrypt(self, thing_to_encrypt):
+        return self.serializer.dumps(thing_to_encrypt, self.salt)
+
+    def decrypt(self, thing_to_decrypt):
+        return self.serializer.loads(thing_to_decrypt, salt=self.salt)
+
 
 def hashpw(password):
     return generate_password_hash(password.encode('UTF-8'), 10)

app/notifications/rest.py

@@ -8,7 +8,7 @@ from flask import (
 )
 
 from itsdangerous import URLSafeSerializer
-from app import api_user
+from app import api_user, encryption
 from app.aws_sqs import add_notification_to_queue
 from app.dao import (templates_dao, notifications_dao)
 from app.schemas import (
@@ -37,8 +37,6 @@ def get_notifications(notification_id):
 
 @notifications.route('/sms', methods=['POST'])
 def create_sms_notification():
-    serializer = URLSafeSerializer(current_app.config.get('SECRET_KEY'))
-
     notification, errors = sms_template_notification_schema.load(request.get_json())
     if errors:
         return jsonify(result="error", message=errors), 400
@@ -49,14 +47,11 @@ def create_sms_notification():
         return jsonify(result="error", message={'template': ['Template not found']}), 400
 
     notification_id = create_notification_id()
-    encrypted_notification = serializer.dumps(notification, current_app.config.get('DANGEROUS_SALT'))
 
     send_sms.apply_async((
         api_user['client'],
         notification_id,
-        encrypted_notification,
+        encryption.encrypt(notification)))
-        current_app.config.get('SECRET_KEY'),
-        current_app.config.get('DANGEROUS_SALT')))
     return jsonify({'notification_id': notification_id}), 201
 
 

tests/app/notifications/test_rest.py

@@ -200,6 +200,7 @@ def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker
     with notify_api.test_request_context():
         with notify_api.test_client() as client:
             mocker.patch('app.celery.tasks.send_sms.apply_async')
+            mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
 
             data = {
                 'to': '+441234123123',
@@ -222,9 +223,7 @@ def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker
             app.celery.tasks.send_sms.apply_async.assert_called_once_with(
                 (str(sample_template.service_id),
                  notification_id,
-                 ANY,
+                 "something_encrypted")
-                 notify_api.config['SECRET_KEY'],
-                 notify_api.config['DANGEROUS_SALT'])
             )
             assert response.status_code == 201
             assert notification_id

tests/app/test_encryption.py

@@ -0,0 +1,20 @@
+from app.encryption import Encryption
+
+encryption = Encryption()
+
+
+def test_should_encrypt_content(notify_api):
+        encryption.init_app(notify_api)
+        assert encryption.encrypt("this") != "this"
+
+
+def test_should_decrypt_content(notify_api):
+        encryption.init_app(notify_api)
+        encrypted = encryption.encrypt("this")
+        assert encryption.decrypt(encrypted) == "this"
+
+
+def test_should_encrypt_json(notify_api):
+        encryption.init_app(notify_api)
+        encrypted = encryption.encrypt({"this": "that"})
+        assert encryption.decrypt(encrypted) == {"this": "that"}

tests/conftest.py

@@ -63,6 +63,7 @@ def notify_db_session(request):
 def notify_config(notify_api):
     notify_api.config['NOTIFY_API_ENVIRONMENT'] = 'test'
     notify_api.config.from_object(configs['test'])
+    return notify_api.config
 
 
 @pytest.fixture(scope='function')