mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-24 01:11:38 -05:00
Merge pull request #1503 from alphagov/vb-remove-ip-whitelist
remove ip whitelist inbound sms codes
This commit is contained in:
@@ -119,7 +119,6 @@ def register_blueprint(application):
|
||||
|
||||
# delivery receipts
|
||||
# TODO: make sure research mode can still trigger sms callbacks, then re-enable this
|
||||
# sms_callback_blueprint.before_request(restrict_ip_sms)
|
||||
sms_callback_blueprint.before_request(requires_no_auth)
|
||||
application.register_blueprint(sms_callback_blueprint)
|
||||
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
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
|
||||
@@ -44,43 +42,6 @@ def requires_no_auth():
|
||||
pass
|
||||
|
||||
|
||||
def restrict_ip_sms():
|
||||
# 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"
|
||||
# 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(',')
|
||||
|
||||
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].strip())
|
||||
|
||||
# IP whitelist
|
||||
allowed_ips = current_app.config.get('SMS_INBOUND_WHITELIST')
|
||||
|
||||
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': inbound_ip
|
||||
}
|
||||
})
|
||||
|
||||
if allowed:
|
||||
return
|
||||
else:
|
||||
raise AuthError('Unknown source IP address from the SMS provider', 403)
|
||||
|
||||
raise AuthError('Traffic from unknown source or route', 403)
|
||||
|
||||
|
||||
def requires_admin_auth():
|
||||
auth_token = get_auth_token(request)
|
||||
client = __get_token_issuer(auth_token)
|
||||
|
||||
@@ -5,7 +5,6 @@ from datetime import datetime
|
||||
from tests.conftest import set_config_values
|
||||
|
||||
import pytest
|
||||
import flask
|
||||
from flask import json, current_app
|
||||
from freezegun import freeze_time
|
||||
from notifications_python_client.authentication import create_jwt_token
|
||||
@@ -13,7 +12,6 @@ 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
|
||||
|
||||
|
||||
# Test the require_admin_auth and require_auth methods
|
||||
@@ -310,71 +308,6 @@ def __create_token(service_id):
|
||||
client_id=str(service_id))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
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.0/24']
|
||||
blueprint = flask.Blueprint('restrict_ip_sms_app', __name__)
|
||||
|
||||
@blueprint.route('/')
|
||||
def test_endpoint():
|
||||
return 'OK', 200
|
||||
|
||||
blueprint.before_request(restrict_ip_sms)
|
||||
app.register_blueprint(blueprint)
|
||||
|
||||
with app.test_request_context(), app.test_client() as client:
|
||||
yield client
|
||||
|
||||
|
||||
def test_allow_valid_ips(restrict_ip_sms_app):
|
||||
response = restrict_ip_sms_app.get(
|
||||
path='/',
|
||||
headers=[
|
||||
('X-Forwarded-For', '111.111.111.111, 222.222.222.222, 127.0.0.1'),
|
||||
]
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_reject_invalid_ips(restrict_ip_sms_app):
|
||||
with pytest.raises(AuthError) as exc_info:
|
||||
restrict_ip_sms_app.get(
|
||||
path='/',
|
||||
headers=[
|
||||
('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'
|
||||
|
||||
|
||||
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, 123.123.123.123, 333.333.333.333, 127.0.0.1')
|
||||
]
|
||||
)
|
||||
|
||||
assert exc_info.value.short_message == 'Unknown source IP address from the SMS provider'
|
||||
|
||||
|
||||
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='/',
|
||||
headers=[
|
||||
('X-Forwarded-For', '200.200.200.222, 222.222.222.222, 127.0.0.1'),
|
||||
]
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.parametrize('check_proxy_header,header_value,expected_status', [
|
||||
(True, 'key_1', 200),
|
||||
(True, 'wrong_key', 403),
|
||||
|
||||
@@ -23,7 +23,6 @@ from tests.app.conftest import sample_service
|
||||
def firetext_post(client, data, auth=True, password='testkey'):
|
||||
headers = [
|
||||
('Content-Type', 'application/x-www-form-urlencoded'),
|
||||
('X-Forwarded-For', '203.0.113.195, 70.41.3.18, 150.172.238.178')
|
||||
]
|
||||
|
||||
if auth:
|
||||
@@ -40,7 +39,6 @@ def firetext_post(client, data, auth=True, password='testkey'):
|
||||
def mmg_post(client, data, auth=True, password='testkey'):
|
||||
headers = [
|
||||
('Content-Type', 'application/json'),
|
||||
('X-Forwarded-For', '203.0.113.195, 70.41.3.18, 150.172.238.178')
|
||||
]
|
||||
|
||||
if auth:
|
||||
|
||||
Reference in New Issue
Block a user