mirror of
https://github.com/GSA/notifications-api.git
synced 2026-09-06 20:58:25 -04:00
Minor change in how we inteprete Incoming IP
This commit is contained in:
@@ -1,14 +1,13 @@
|
|||||||
|
from ipaddress import IPv4Address, IPv4Network
|
||||||
|
|
||||||
from flask import request, _request_ctx_stack, current_app, g
|
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.exc import DataError
|
||||||
from sqlalchemy.orm.exc import NoResultFound
|
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 app.dao.services_dao import dao_fetch_service_by_id_with_api_keys
|
||||||
|
|
||||||
from ipaddress import IPv4Interface, ip_address
|
|
||||||
|
|
||||||
|
|
||||||
class AuthError(Exception):
|
class AuthError(Exception):
|
||||||
def __init__(self, message, code):
|
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)
|
# 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_route = request.headers.get("X-Forwarded-For")
|
||||||
ip_list = ip_route.split(',')
|
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 {}"
|
current_app.logger.info("Inbound sms ip route list {}"
|
||||||
.format(ip_route))
|
.format(ip_route))
|
||||||
|
if len(ip_list) >= 3:
|
||||||
|
inbound_ip = IPv4Address(ip_list[len(ip_list) - 3])
|
||||||
|
|
||||||
# IP whitelist
|
# IP whitelist
|
||||||
allowed_ips = current_app.config.get('SMS_INBOUND_WHITELIST')
|
allowed_ips = current_app.config.get('SMS_INBOUND_WHITELIST')
|
||||||
allowed = False
|
|
||||||
|
|
||||||
for allowed_ip in allowed_ips:
|
allowed = any(
|
||||||
masked_bits = ''
|
inbound_ip in IPv4Network(allowed_ip)
|
||||||
if (len(allowed_ip.split('/')) > 1):
|
for allowed_ip in allowed_ips
|
||||||
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)
|
|
||||||
|
|
||||||
current_app.logger.info({
|
# if allowed:
|
||||||
'message': 'Inbound sms ip address',
|
# return
|
||||||
'log_contents': {
|
# else:
|
||||||
'passed': allowed,
|
# raise AuthError('Unknown source IP address from the SMS provider', 403)
|
||||||
'ip_address': ip
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
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():
|
def requires_admin_auth():
|
||||||
|
|||||||
@@ -313,7 +313,7 @@ def __create_token(service_id):
|
|||||||
def restrict_ip_sms_app():
|
def restrict_ip_sms_app():
|
||||||
app = flask.Flask(__name__)
|
app = flask.Flask(__name__)
|
||||||
app.config['TESTING'] = True
|
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 = flask.Blueprint('restrict_ip_sms_app', __name__)
|
||||||
|
|
||||||
@blueprint.route('/')
|
@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'
|
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
|
# Test an address that match the first 24 bits only
|
||||||
response = restrict_ip_sms_app.get(
|
response = restrict_ip_sms_app.get(
|
||||||
path='/',
|
path='/',
|
||||||
|
|||||||
Reference in New Issue
Block a user