darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-03 09:51:11 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1391 from alphagov/add_proxy_header_check

Browse Source 
Add proxy header check
This commit is contained in:
Sakis
2017-11-17 10:35:43 +00:00
committed by GitHub
parent 6d36a65b65 26ba4dad06
commit 9d2cecd001
5 changed files with 33 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
app/__init__.py
View File

@@ -3,14 +3,13 @@ import random
import string
import uuid
from flask import Flask, _request_ctx_stack
from flask import request, g, jsonify
from flask import Flask, _request_ctx_stack, request, g, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from monotonic import monotonic
from notifications_utils.clients.statsd.statsd_client import StatsdClient
from notifications_utils.clients.redis.redis_client import RedisClient
from notifications_utils import logging, request_id
from notifications_utils import logging, request_helper
from werkzeug.local import LocalProxy
from app.celery.celery import NotifyCelery
@@ -56,7 +55,7 @@ def create_app(app_name=None):
application.config['NOTIFY_APP_NAME'] = app_name
init_app(application)
request_id.init_app(application)
request_helper.init_app(application)
db.init_app(application)
ma.init_app(application)
statsd_client.init_app(application)
@@ -206,6 +205,8 @@ def init_app(app):
def record_user_agent():
statsd_client.incr("user-agent.{}".format(process_user_agent(request.headers.get('User-Agent', None))))
app.before_request(request_helper.check_proxy_header_before_request)
@app.before_request
def record_request_details():
g.start = monotonic()

49
app/authentication/auth.py
View File

@@ -45,56 +45,7 @@ def requires_no_auth():
pass
def check_route_secret():
# Check route of inbound sms (Experimental)
# Custom header for route security
auth_error_msg = ''
if request.headers.get("X-Custom-Forwarder"):
route_secret_key = request.headers.get("X-Custom-Forwarder")
if route_secret_key is None:
# Not blocking at the moment
# raise AuthError('invalid secret key', 403)
auth_error_msg = auth_error_msg + 'invalid secret key, '
else:
key_1 = current_app.config.get('ROUTE_SECRET_KEY_1')
key_2 = current_app.config.get('ROUTE_SECRET_KEY_2')
if key_1 == '' and key_2 == '':
# Not blocking at the moment
# raise AuthError('X-Custom-Forwarder, no secret was set on server', 503)
auth_error_msg = auth_error_msg + 'no secret was set on server, '
else:
key_used = None
route_allowed = False
if route_secret_key == key_1:
key_used = 1
route_allowed = True
elif route_secret_key == key_2:
key_used = 2
route_allowed = True
if not key_used:
# Not blocking at the moment
# raise AuthError('X-Custom-Forwarder, wrong secret', 403)
auth_error_msg = auth_error_msg + 'wrong secret'
current_app.logger.info({
'message': 'X-Custom-Forwarder',
'log_contents': {
'passed': route_allowed,
'key_used': key_used,
'error': auth_error_msg
}
})
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"

5
app/config.py
View File

@@ -121,6 +121,8 @@ class Config(object):
ONE_OFF_MESSAGE_FILENAME = 'Report'
MAX_VERIFY_CODE_COUNT = 10
CHECK_PROXY_HEADER = False
NOTIFY_SERVICE_ID = 'd6aa2c68-a2d9-4437-ab19-3ae8eb202553'
NOTIFY_USER_ID = '6af522d0-2915-4e52-83a3-3690455a5fe6'
INVITATION_EMAIL_TEMPLATE_ID = '4f46df42-f795-4cc4-83bb-65ca312f49cc'
@@ -371,6 +373,7 @@ class Preview(Config):
DVLA_RESPONSE_BUCKET_NAME = 'notify.works-ftp'
FROM_NUMBER = 'preview'
API_RATE_LIMIT_ENABLED = True
CHECK_PROXY_HEADER = True
class Staging(Config):
@@ -381,6 +384,7 @@ class Staging(Config):
STATSD_ENABLED = True
FROM_NUMBER = 'stage'
API_RATE_LIMIT_ENABLED = True
CHECK_PROXY_HEADER = True
class Live(Config):
@@ -394,6 +398,7 @@ class Live(Config):
FUNCTIONAL_TEST_PROVIDER_SMS_TEMPLATE_ID = 'ba9e1789-a804-40b8-871f-cc60d4c1286f'
PERFORMANCE_PLATFORM_ENABLED = True
API_RATE_LIMIT_ENABLED = True
CHECK_PROXY_HEADER = False
class CloudFoundryConfig(Config):

2
requirements.txt
View File

@@ -26,6 +26,6 @@ notifications-python-client==4.6.0
awscli>=1.11,<1.12
awscli-cwlogs>=1.4,<1.5
git+https://github.com/alphagov/notifications-utils.git@21.5.1#egg=notifications-utils==21.5.1
git+https://github.com/alphagov/notifications-utils.git@22.1.0#egg=notifications-utils==22.1.0
git+https://github.com/alphagov/boto.git@2.43.0-patch3#egg=boto==2.43.0-patch3

174
tests/app/authentication/test_authentication.py
View File

@@ -2,17 +2,19 @@ import jwt
import uuid
import time
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
from notifications_utils.request_helper import check_proxy_header_before_request
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, check_route_secret
from app.authentication.auth import restrict_ip_sms, AuthError
# Test the require_admin_auth and require_auth methods
@@ -374,156 +376,24 @@ def test_allow_valid_ips_bits(restrict_ip_sms_app):
assert response.status_code == 200
@pytest.fixture
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_with_key_1_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_1_is_used(route_secret_app_with_key_1_only):
response = route_secret_app_with_key_1_only.get(
path='/',
headers=[
('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_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_both_keys', __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
@pytest.mark.parametrize('secret_header, expected_key_used', [
('key_2', 2),
('key_1', 1)
@pytest.mark.parametrize('check_proxy_header,header_value,expected_status', [
(True, 'key_1', 200),
(True, 'wrong_key', 403),
(False, 'key_1', 200),
(False, 'wrong_key', 200),
])
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', 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
def test_route_correct_secret_key(notify_api, check_proxy_header, header_value, expected_status):
with set_config_values(notify_api, {
'ROUTE_SECRET_KEY_1': 'key_1',
'ROUTE_SECRET_KEY_2': '',
'CHECK_PROXY_HEADER': check_proxy_header,
}):
@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
@pytest.fixture
def route_secret_app_with_no_key():
app = flask.Flask(__name__)
app.config['TESTING'] = True
app.config['ROUTE_SECRET_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_with_no_key', __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
# TODO: expected to fail because we have not implement blocking yet
@pytest.mark.parametrize('secret', [
pytest.mark.xfail('some-header')
])
def test_route_secret_no_key_set_should_fail(route_secret_app_with_no_key, secret):
with pytest.raises(AuthError) as exc_info:
response = route_secret_app_with_no_key.get(
path='/',
headers=[
('X-Custom-Forwarder', 'some_value'),
]
)
resp_json = json.loads(response.get_data(as_text=True))
exc_info.value.short_message == 'X-Custom-Forwarder, no secret was set on server'
assert resp_json['key_used'] is None
with notify_api.test_client() as client:
response = client.get(
path='/_status',
headers=[
('X-Custom-Forwarder', header_value),
]
)
assert response.status_code == expected_status