mirror of
https://github.com/GSA/notifications-api.git
synced 2026-03-01 22:40:02 -05:00
Merge pull request #1079 from alphagov/sms_whitelist
retrieve sms ip whitelist from credentials on paas
This commit is contained in:
@@ -47,11 +47,14 @@ def restrict_ip_sms():
|
||||
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"
|
||||
ip_list = request.headers.get("X-Forwarded-For")
|
||||
ip = ip_list.split(',')[0].strip()
|
||||
current_app.logger.info("Inbound sms ip list {}".format(ip_list))
|
||||
# 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:
|
||||
ip = ip_list[len(ip_list) - 3]
|
||||
current_app.logger.info("Inbound sms ip route list {}".format(ip_route))
|
||||
|
||||
if ip in current_app.config.get('ALLOW_IP_INBOUND_SMS'):
|
||||
if ip in current_app.config.get('SMS_INBOUND_WHITELIST'):
|
||||
current_app.logger.info("Inbound sms ip addresses {} passed ".format(ip))
|
||||
return
|
||||
else:
|
||||
|
||||
@@ -43,6 +43,7 @@ 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['PERFORMANCE_PLATFORM_TOKEN'] = notify_config['credentials'].get('performance_platform_token', '')
|
||||
os.environ['SMS_INBOUND_WHITELIST'] = notify_config['credentials']['allow_ip_inbound_sms']
|
||||
|
||||
|
||||
def extract_notify_aws_config(aws_config):
|
||||
|
||||
@@ -262,7 +262,7 @@ class Config(object):
|
||||
|
||||
FREE_SMS_TIER_FRAGMENT_COUNT = 250000
|
||||
|
||||
ALLOW_IP_INBOUND_SMS = []
|
||||
SMS_INBOUND_WHITELIST = os.environ.get('SMS_INBOUND_WHITELIST', [])
|
||||
|
||||
|
||||
######################
|
||||
|
||||
@@ -313,8 +313,7 @@ def __create_token(service_id):
|
||||
def restrict_ip_sms_app():
|
||||
app = flask.Flask(__name__)
|
||||
app.config['TESTING'] = True
|
||||
app.config['ALLOW_IP_INBOUND_SMS'] = ['134.213.243.188']
|
||||
|
||||
app.config['SMS_INBOUND_WHITELIST'] = ['111.111.111.111', '100.100.100.100']
|
||||
blueprint = flask.Blueprint('restrict_ip_sms_app', __name__)
|
||||
|
||||
@blueprint.route('/')
|
||||
@@ -332,7 +331,7 @@ def test_allow_valid_ips(restrict_ip_sms_app):
|
||||
response = restrict_ip_sms_app.get(
|
||||
path='/',
|
||||
headers=[
|
||||
('X-Forwarded-For', '134.213.243.188, 222.222.222.222, 127.0.0.1'),
|
||||
('X-Forwarded-For', '111.111.111.111, 222.222.222.222, 127.0.0.1'),
|
||||
]
|
||||
)
|
||||
|
||||
@@ -345,7 +344,20 @@ def test_reject_invalid_ips(restrict_ip_sms_app):
|
||||
restrict_ip_sms_app.get(
|
||||
path='/',
|
||||
headers=[
|
||||
('X-Forwarded-For', '222.222.222.222, 111.111.111.111, 127.0.0.1')
|
||||
('X-Forwarded-For', '222.222.222.222, 333.333.333.333, 127.0.0.1')
|
||||
]
|
||||
)
|
||||
|
||||
assert exc_info.value.short_message == 'Unknown source IP address from the SMS provider'
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason='Currently not blocking invalid IPs', strict=True)
|
||||
def test_illegitimate_ips(restrict_ip_sms_app):
|
||||
with pytest.raises(AuthError) as exc_info:
|
||||
restrict_ip_sms_app.get(
|
||||
path='/',
|
||||
headers=[
|
||||
('X-Forwarded-For', '111.111.111.111, 999.999.999.999, 333.333.333.333, 127.0.0.1')
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
@@ -16,7 +16,8 @@ def notify_config():
|
||||
'admin_client_secret': 'admin client secret',
|
||||
'secret_key': 'secret key',
|
||||
'dangerous_salt': 'dangerous salt',
|
||||
'performance_platform_token': 'performance platform token'
|
||||
'performance_platform_token': 'performance platform token',
|
||||
'allow_ip_inbound_sms': ['111.111.111.111', '100.100.100.100']
|
||||
}
|
||||
}
|
||||
|
||||
@@ -197,3 +198,10 @@ def test_redis_config():
|
||||
|
||||
assert os.environ['REDIS_ENABLED'] == '1'
|
||||
assert os.environ['REDIS_URL'] == 'redis url'
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('os_environ', 'cloudfoundry_environ')
|
||||
def test_sms_inbound_config():
|
||||
extract_cloudfoundry_config()
|
||||
|
||||
assert os.environ['SMS_INBOUND_WHITELIST'] == ['111.111.111.111', '100.100.100.100']
|
||||
|
||||
Reference in New Issue
Block a user