From c285ab0b45bfa75e70e5a4e01c499933ed7af3ba Mon Sep 17 00:00:00 2001 From: venusbb Date: Wed, 13 Sep 2017 11:29:11 +0100 Subject: [PATCH 1/3] inbound sms monitoring 24bit mask --- app/authentication/auth.py | 24 +++++++++++++++---- .../app/authentication/test_authentication.py | 12 ++++++++++ 2 files changed, 31 insertions(+), 5 deletions(-) diff --git a/app/authentication/auth.py b/app/authentication/auth.py index b3e254b94..baadb32f0 100644 --- a/app/authentication/auth.py +++ b/app/authentication/auth.py @@ -44,6 +44,12 @@ def requires_no_auth(): def restrict_ip_sms(): + # 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 {}".format(request.headers.get("X-Custom-forwarder"))) + + # Check IP of SMS providers ip = '' if request.headers.get("X-Forwarded-For"): # X-Forwarded-For looks like "203.0.113.195, 70.41.3.18, 150.172.238.178" @@ -54,21 +60,29 @@ def restrict_ip_sms(): ip = ip_list[len(ip_list) - 3] current_app.logger.info("Inbound sms ip route list {}" .format(ip_route)) + p0 = ip.split('.') - # Temporary custom header for route security - to experiment if the header passes through - if request.headers.get("X-Custom-forwarder"): - current_app.logger.info("X-Custom-forwarder {}".format(request.headers.get("X-Custom-forwarder"))) + # IP whitelist + allowed_ips = current_app.config.get('SMS_INBOUND_WHITELIST') + allowed = False + + for allowed_ip in allowed_ips: + p1 = allowed_ip.split('.') + if p0[0] == p1[0] and p0[1] == p1[1] and p0[2] == p1[2]: + allowed = True + # return + # else: + # raise AuthError('Unknown source IP address from the SMS provider', 403) current_app.logger.info({ 'message': 'Inbound sms ip address', 'log_contents': { - 'passed': ip in current_app.config.get('SMS_INBOUND_WHITELIST'), + 'passed': allowed, 'ip_address': ip } }) return - # raise AuthError('Unknown source IP address from the SMS provider', 403) def requires_admin_auth(): diff --git a/tests/app/authentication/test_authentication.py b/tests/app/authentication/test_authentication.py index f0420e755..a0fac818d 100644 --- a/tests/app/authentication/test_authentication.py +++ b/tests/app/authentication/test_authentication.py @@ -362,3 +362,15 @@ def test_illegitimate_ips(restrict_ip_sms_app): ) assert exc_info.value.short_message == 'Unknown IP route not from known SMS provider' + + +def test_allow_valid_ips_24bits(restrict_ip_sms_app): + # Test an address that match the first 24 bits only + response = restrict_ip_sms_app.get( + path='/', + headers=[ + ('X-Forwarded-For', '111.111.111.119, 222.222.222.222, 127.0.0.1'), + ] + ) + + assert response.status_code == 200 From 9efc17a941020561c8507d65ed375eab178beffa Mon Sep 17 00:00:00 2001 From: venusbb Date: Wed, 13 Sep 2017 14:08:23 +0100 Subject: [PATCH 2/3] Use ipaddress library for the masked bits --- app/authentication/auth.py | 13 +++++++++---- tests/app/authentication/test_authentication.py | 4 ++-- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/app/authentication/auth.py b/app/authentication/auth.py index baadb32f0..f7866a943 100644 --- a/app/authentication/auth.py +++ b/app/authentication/auth.py @@ -7,6 +7,8 @@ from notifications_python_client.errors import TokenDecodeError, TokenExpiredErr from app.dao.services_dao import dao_fetch_service_by_id_with_api_keys +from ipaddress import IPv4Interface, ip_address + class AuthError(Exception): def __init__(self, message, code): @@ -57,20 +59,23 @@ def restrict_ip_sms(): ip_route = request.headers.get("X-Forwarded-For") ip_list = ip_route.split(',') if len(ip_list) >= 3: - ip = ip_list[len(ip_list) - 3] + inbound_ip = ip_list[len(ip_list) - 3] current_app.logger.info("Inbound sms ip route list {}" .format(ip_route)) - p0 = ip.split('.') # IP whitelist allowed_ips = current_app.config.get('SMS_INBOUND_WHITELIST') allowed = False for allowed_ip in allowed_ips: - p1 = allowed_ip.split('.') - if p0[0] == p1[0] and p0[1] == p1[1] and p0[2] == p1[2]: + masked_bits = '' + if (len(allowed_ip.split('/')) > 1): + masked_bits = allowed_ip.split('/')[1] + inbound_ip_str = inbound_ip + '/' + masked_bits + if IPv4Interface(allowed_ip).network == IPv4Interface(inbound_ip_str).network: allowed = True # return + break # else: # raise AuthError('Unknown source IP address from the SMS provider', 403) diff --git a/tests/app/authentication/test_authentication.py b/tests/app/authentication/test_authentication.py index a0fac818d..000d08211 100644 --- a/tests/app/authentication/test_authentication.py +++ b/tests/app/authentication/test_authentication.py @@ -313,7 +313,7 @@ def __create_token(service_id): def restrict_ip_sms_app(): app = flask.Flask(__name__) app.config['TESTING'] = True - app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111', '100.100.100.100'] + app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.200/24'] blueprint = flask.Blueprint('restrict_ip_sms_app', __name__) @blueprint.route('/') @@ -369,7 +369,7 @@ def test_allow_valid_ips_24bits(restrict_ip_sms_app): response = restrict_ip_sms_app.get( path='/', headers=[ - ('X-Forwarded-For', '111.111.111.119, 222.222.222.222, 127.0.0.1'), + ('X-Forwarded-For', '200.200.200.222, 222.222.222.222, 127.0.0.1'), ] ) From 160b8787458805df0cbe35dcff7ba3f5b6449489 Mon Sep 17 00:00:00 2001 From: venusbb Date: Wed, 13 Sep 2017 17:23:23 +0100 Subject: [PATCH 3/3] Minor change in how we inteprete Incoming IP --- app/authentication/auth.py | 59 ++++++++++--------- .../app/authentication/test_authentication.py | 4 +- 2 files changed, 32 insertions(+), 31 deletions(-) diff --git a/app/authentication/auth.py b/app/authentication/auth.py index f7866a943..d24153bcd 100644 --- a/app/authentication/auth.py +++ b/app/authentication/auth.py @@ -1,14 +1,13 @@ +from ipaddress import IPv4Address, IPv4Network + from flask import request, _request_ctx_stack, current_app, g +from notifications_python_client.authentication import decode_jwt_token, get_token_issuer +from notifications_python_client.errors import TokenDecodeError, TokenExpiredError, TokenIssuerError from sqlalchemy.exc import DataError from sqlalchemy.orm.exc import NoResultFound -from notifications_python_client.authentication import decode_jwt_token, get_token_issuer -from notifications_python_client.errors import TokenDecodeError, TokenExpiredError, TokenIssuerError - from app.dao.services_dao import dao_fetch_service_by_id_with_api_keys -from ipaddress import IPv4Interface, ip_address - class AuthError(Exception): def __init__(self, message, code): @@ -58,36 +57,38 @@ def restrict_ip_sms(): # Counting backwards and look at the IP at the 3rd last hop - hence, hop(end-3) ip_route = request.headers.get("X-Forwarded-For") ip_list = ip_route.split(',') - if len(ip_list) >= 3: - inbound_ip = ip_list[len(ip_list) - 3] + current_app.logger.info("Inbound sms ip route list {}" .format(ip_route)) + if len(ip_list) >= 3: + inbound_ip = IPv4Address(ip_list[len(ip_list) - 3]) - # IP whitelist - allowed_ips = current_app.config.get('SMS_INBOUND_WHITELIST') - allowed = False + # IP whitelist + allowed_ips = current_app.config.get('SMS_INBOUND_WHITELIST') - for allowed_ip in allowed_ips: - masked_bits = '' - if (len(allowed_ip.split('/')) > 1): - masked_bits = allowed_ip.split('/')[1] - inbound_ip_str = inbound_ip + '/' + masked_bits - if IPv4Interface(allowed_ip).network == IPv4Interface(inbound_ip_str).network: - allowed = True - # return - break - # else: - # raise AuthError('Unknown source IP address from the SMS provider', 403) + allowed = any( + inbound_ip in IPv4Network(allowed_ip) + for allowed_ip in allowed_ips + ) - current_app.logger.info({ - 'message': 'Inbound sms ip address', - 'log_contents': { - 'passed': allowed, - 'ip_address': ip - } - }) + # if allowed: + # return + # else: + # raise AuthError('Unknown source IP address from the SMS provider', 403) - return + current_app.logger.info({ + 'message': 'Inbound sms ip address', + 'log_contents': { + 'passed': allowed, + 'ip_address': ip + } + }) + return + + current_app.logger.error('Traffic from unknown source or route, X-Forwarded-For="{}"'.format( + request.headers.get("X-Forwarded-For")) + ) + # raise AuthError('Traffic from unknown source or route', 403) def requires_admin_auth(): diff --git a/tests/app/authentication/test_authentication.py b/tests/app/authentication/test_authentication.py index 000d08211..15eb7f2ec 100644 --- a/tests/app/authentication/test_authentication.py +++ b/tests/app/authentication/test_authentication.py @@ -313,7 +313,7 @@ def __create_token(service_id): def restrict_ip_sms_app(): app = flask.Flask(__name__) app.config['TESTING'] = True - app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.200/24'] + app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111/32', '200.200.200.0/24'] blueprint = flask.Blueprint('restrict_ip_sms_app', __name__) @blueprint.route('/') @@ -364,7 +364,7 @@ def test_illegitimate_ips(restrict_ip_sms_app): assert exc_info.value.short_message == 'Unknown IP route not from known SMS provider' -def test_allow_valid_ips_24bits(restrict_ip_sms_app): +def test_allow_valid_ips_bits(restrict_ip_sms_app): # Test an address that match the first 24 bits only response = restrict_ip_sms_app.get( path='/',