Merge pull request #1368 from alphagov/vb-route-secret

initial logging for route protection

app/authentication/auth.py

@@ -7,6 +7,7 @@ from sqlalchemy.exc import DataError
 from sqlalchemy.orm.exc import NoResultFound
 
 from app.dao.services_dao import dao_fetch_service_by_id_with_api_keys
+from flask import jsonify
 
 
 class AuthError(Exception):
@@ -44,11 +45,55 @@ def requires_no_auth():
     pass
 
 
-def restrict_ip_sms():
+def check_route_secret():
     # Check route of inbound sms (Experimental)
-    # Temporary custom header for route security
+    # Custom header for route security
-    if request.headers.get("X-Custom-forwarder"):
+    auth_error_msg = ''
-        current_app.logger.info("X-Custom-forwarder received")
+    if request.headers.get("X-Custom-Forwarder"):
+        route_secret_key = request.headers.get("X-Custom-Forwarder")
+
+        if route_secret_key is None:
+            # Not blocking at the moment
+            # raise AuthError('invalid secret key', 403)
+            auth_error_msg = auth_error_msg + 'invalid secret key, '
+        else:
+
+            key_1 = current_app.config.get('ROUTE_SECRET_KEY_1')
+            key_2 = current_app.config.get('ROUTE_SECRET_KEY_2')
+
+            if key_1 == '' and key_2 == '':
+                # Not blocking at the moment
+                # raise AuthError('X-Custom-Forwarder, no secret was set on server', 503)
+                auth_error_msg = auth_error_msg + 'no secret was set on server, '
+            else:
+
+                key_used = None
+                route_allowed = False
+                if route_secret_key == key_1:
+                    key_used = 1
+                    route_allowed = True
+                elif route_secret_key == key_2:
+                    key_used = 2
+                    route_allowed = True
+
+                if not key_used:
+                    # Not blocking at the moment
+                    # raise AuthError('X-Custom-Forwarder, wrong secret', 403)
+                    auth_error_msg = auth_error_msg + 'wrong secret'
+
+        current_app.logger.info({
+            'message': 'X-Custom-Forwarder',
+            'log_contents': {
+                'passed': route_allowed,
+                'key_used': key_used,
+                'error': auth_error_msg
+            }
+        })
+        return jsonify(key_used=key_used), 200
+
+
+def restrict_ip_sms():
+    check_route_secret()
 
     # Check IP of SMS providers
     if request.headers.get("X-Forwarded-For"):

app/cloudfoundry_config.py

@@ -45,6 +45,8 @@ def extract_notify_config(notify_config):
     os.environ['SECRET_KEY'] = notify_config['credentials']['secret_key']
     os.environ['DANGEROUS_SALT'] = notify_config['credentials']['dangerous_salt']
     os.environ['SMS_INBOUND_WHITELIST'] = json.dumps(notify_config['credentials']['allow_ip_inbound_sms'])
+    os.environ['ROUTE_SECRET_KEY_1'] = notify_config['credentials']['route_secret_key_1']
+    os.environ['ROUTE_SECRET_KEY_2'] = notify_config['credentials']['route_secret_key_2']
 
 
 def extract_performance_platform_config(performance_platform_config):

app/config.py

@@ -287,6 +287,8 @@ class Config(object):
     FREE_SMS_TIER_FRAGMENT_COUNT = 250000
 
     SMS_INBOUND_WHITELIST = json.loads(os.environ.get('SMS_INBOUND_WHITELIST', '[]'))
+    ROUTE_SECRET_KEY_1 = os.environ.get('ROUTE_SECRET_KEY_1', '')
+    ROUTE_SECRET_KEY_2 = os.environ.get('ROUTE_SECRET_KEY_2', '')
 
     # Format is as follows:
     # {"dataset_1": "token_1", ...}

tests/app/authentication/test_authentication.py

@@ -12,7 +12,7 @@ from notifications_python_client.authentication import create_jwt_token
 from app import api_user
 from app.dao.api_key_dao import get_unsigned_secrets, save_model_api_key, get_unsigned_secret, expire_api_key
 from app.models import ApiKey, KEY_TYPE_NORMAL
-from app.authentication.auth import restrict_ip_sms, AuthError
+from app.authentication.auth import restrict_ip_sms, AuthError, check_route_secret
 
 
 # Test the require_admin_auth and require_auth methods
@@ -372,3 +372,158 @@ def test_allow_valid_ips_bits(restrict_ip_sms_app):
     )
 
     assert response.status_code == 200
+
+
+@pytest.fixture
+def route_secret_app_with_key_1_only():
+    app = flask.Flask(__name__)
+    app.config['TESTING'] = True
+    app.config['ROUTE_SECRET_KEY_1'] = "key_1"
+    app.config['ROUTE_SECRET_KEY_2'] = ""
+    app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.0/24']
+    blueprint = flask.Blueprint('route_secret_app_with_key_1_only', __name__)
+
+    @blueprint.route('/')
+    def test_endpoint():
+        return 'OK', 200
+
+    blueprint.before_request(check_route_secret)
+    app.register_blueprint(blueprint)
+
+    with app.test_request_context(), app.test_client() as client:
+        yield client
+
+
+def test_route_secret_key_1_is_used(route_secret_app_with_key_1_only):
+    response = route_secret_app_with_key_1_only.get(
+        path='/',
+        headers=[
+            ('X-Custom-forwarder', 'key_1'),
+        ]
+    )
+    resp_json = json.loads(response.get_data(as_text=True))
+    assert response.status_code == 200
+    assert resp_json['key_used'] == 1
+
+
+# This tests for when we do key rotation when we use both keys
+@pytest.fixture
+def route_secret_app_both_keys():
+    app = flask.Flask(__name__)
+    app.config['TESTING'] = True
+    app.config['ROUTE_SECRET_KEY_1'] = "key_1"
+    app.config['ROUTE_SECRET_KEY_2'] = "key_2"
+    app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.0/24']
+    blueprint = flask.Blueprint('route_secret_app_both_keys', __name__)
+
+    @blueprint.route('/')
+    def test_endpoint():
+        return 'OK', 200
+
+    blueprint.before_request(check_route_secret)
+    app.register_blueprint(blueprint)
+
+    with app.test_request_context(), app.test_client() as client:
+        yield client
+
+
+@pytest.mark.parametrize('secret_header, expected_key_used', [
+    ('key_2', 2),
+    ('key_1', 1)
+])
+def test_can_use_either_secret_route_key(route_secret_app_both_keys, secret_header, expected_key_used):
+    print(secret_header)
+    response = route_secret_app_both_keys.get(
+        path='/',
+        headers=[
+            ('X-Custom-forwarder', secret_header),
+        ]
+    )
+    resp_json = json.loads(response.get_data(as_text=True))
+    assert response.status_code == 200
+    assert resp_json['key_used'] == expected_key_used
+
+
+@pytest.fixture
+def route_secret_app_with_key_2_only():
+    app = flask.Flask(__name__)
+    app.config['TESTING'] = True
+    app.config['ROUTE_SECRET_KEY_1'] = ""
+    app.config['ROUTE_SECRET_KEY_2'] = "key_2"
+    app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.0/24']
+    blueprint = flask.Blueprint('route_secret_app_with_key_2_only', __name__)
+
+    @blueprint.route('/')
+    def test_endpoint():
+        return 'OK', 200
+
+    blueprint.before_request(check_route_secret)
+    app.register_blueprint(blueprint)
+
+    with app.test_request_context(), app.test_client() as client:
+        yield client
+
+
+def test_route_secret_key_2_is_used(route_secret_app_with_key_2_only):
+    response = route_secret_app_with_key_2_only.get(
+        path='/',
+        headers=[
+            ('X-Custom-Forwarder', 'key_2'),
+        ]
+    )
+    resp_json = json.loads(response.get_data(as_text=True))
+    assert response.status_code == 200
+    assert resp_json['key_used'] == 2
+
+
+# TODO: expected to fail because we have not implement blocking yet
+@pytest.mark.parametrize('header_name, secret', [
+    pytest.mark.xfail(('some-header', 'some-value')),
+    pytest.mark.xfail(('X-Custom-Forwarder', 'wrong-value')),
+])
+def test_no_route_secret_raise_403(route_secret_app_with_key_2_only, header_name, secret):
+
+    response = route_secret_app_with_key_2_only.get(
+        path='/',
+        headers=[
+            (header_name, secret)
+        ]
+    )
+    assert response.status_code == 403
+
+
+@pytest.fixture
+def route_secret_app_with_no_key():
+    app = flask.Flask(__name__)
+    app.config['TESTING'] = True
+    app.config['ROUTE_SECRET_KEY_1'] = ""
+    app.config['ROUTE_SECRET_KEY_2'] = ""
+    app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.0/24']
+    blueprint = flask.Blueprint('route_secret_app_with_no_key', __name__)
+
+    @blueprint.route('/')
+    def test_endpoint():
+        return 'OK', 200
+
+    blueprint.before_request(check_route_secret)
+    app.register_blueprint(blueprint)
+
+    with app.test_request_context(), app.test_client() as client:
+        yield client
+
+
+# TODO: expected to fail because we have not implement blocking yet
+@pytest.mark.parametrize('secret', [
+    pytest.mark.xfail('some-header')
+])
+def test_route_secret_no_key_set_should_fail(route_secret_app_with_no_key, secret):
+    with pytest.raises(AuthError) as exc_info:
+        response = route_secret_app_with_no_key.get(
+            path='/',
+            headers=[
+                ('X-Custom-Forwarder', 'some_value'),
+            ]
+        )
+        resp_json = json.loads(response.get_data(as_text=True))
+        exc_info.value.short_message == 'X-Custom-Forwarder, no secret was set on server'
+        assert resp_json['key_used'] is None

tests/app/test_cloudfoundry_config.py

@@ -16,7 +16,9 @@ def notify_config():
             'admin_client_secret': 'admin client secret',
             'secret_key': 'secret key',
             'dangerous_salt': 'dangerous salt',
-            'allow_ip_inbound_sms': ['111.111.111.111', '100.100.100.100']
+            'allow_ip_inbound_sms': ['111.111.111.111', '100.100.100.100'],
+            'route_secret_key_1': "key_1",
+            'route_secret_key_2': ""
         }
     }