From 81b8d62557b5f71abdfa5806fb0e7abc220d5e8d Mon Sep 17 00:00:00 2001 From: venusbb Date: Fri, 3 Nov 2017 18:19:59 +0000 Subject: [PATCH] create new method to validate secret header, new tests --- app/authentication/auth.py | 22 +++-- .../app/authentication/test_authentication.py | 87 +++++++++++++++---- 2 files changed, 88 insertions(+), 21 deletions(-) diff --git a/app/authentication/auth.py b/app/authentication/auth.py index d7adc51c7..8181f7309 100644 --- a/app/authentication/auth.py +++ b/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,16 +45,18 @@ 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 - # if request.headers.get("X-Custom-forwarder"): - # 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: + # raise AuthError('invalid secret key', 403) + key_1 = current_app.config.get('ROUTE_SECRET_KEY_1') key_2 = current_app.config.get('ROUTE_SECRET_KEY_2') - key_used = 0 + key_used = None route_allowed = False if route_secret_key == key_1: key_used = 1 @@ -63,12 +66,21 @@ def restrict_ip_sms(): route_allowed = True current_app.logger.info({ - 'message': 'X-Custom-forwarder key {} is used'.format(key_used), + 'message': 'X-Custom-Forwarder key {} is used'.format(key_used), 'log_contents': { 'passed': route_allowed, } }) + # if not key_used: + # raise AuthError('X-Custom-Forwarder, wrong secret', 403) + + 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"): # X-Forwarded-For looks like "203.0.113.195, 70.41.3.18, 150.172.238.178" diff --git a/tests/app/authentication/test_authentication.py b/tests/app/authentication/test_authentication.py index 8c58716cc..43fc284c4 100644 --- a/tests/app/authentication/test_authentication.py +++ b/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 @@ -375,63 +375,118 @@ def test_allow_valid_ips_bits(restrict_ip_sms_app): @pytest.fixture -def route_secret_app_1(): +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_1', __name__) + blueprint = flask.Blueprint('route_secret_app_with_key_1_only', __name__) @blueprint.route('/') def test_endpoint(): return 'OK', 200 - blueprint.before_request(restrict_ip_sms) + 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_1): - response = route_secret_app_1.get( +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', 'some key 1'), - ('X-Forwarded-For', '200.200.200.222, 222.222.222.222, 127.0.0.1'), + ('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_2(): +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_2', __name__) + blueprint = flask.Blueprint('route_secret_app_both_keys', __name__) @blueprint.route('/') def test_endpoint(): return 'OK', 200 - blueprint.before_request(restrict_ip_sms) + blueprint.before_request(check_route_secret) app.register_blueprint(blueprint) with app.test_request_context(), app.test_client() as client: yield client -def test_can_use_secret_route_key_2(route_secret_app_2): - - response = route_secret_app_2.get( +@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', 'key_2'), - ('X-Forwarded-For', '200.200.200.222, 222.222.222.222, 127.0.0.1'), + ('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