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

First attempt at securing the endpoints.

Browse Source 
Started with adding a before_request event to the service_blueprint, which executes the requires_admin_auth method rather than the require_auth method.

Obviously this is not done but want to get this in front of people to get an opinion.
This commit is contained in:
Rebecca Law
2017-03-16 10:42:45 +00:00
parent fd00351ad2
commit f880604c85
2 changed files with 179 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -13,38 +13,33 @@ from app.dao.api_key_dao import get_unsigned_secrets, save_model_api_key, get_un
from app.models import ApiKey, KEY_TYPE_NORMAL from app.models import ApiKey, KEY_TYPE_NORMAL
def test_should_not_allow_request_with_no_token(notify_api): # Test the require_admin_auth and require_auth methods
with notify_api.test_request_context(): @pytest.mark.parametrize('url', ['/service', '/notifications'])
with notify_api.test_client() as client: def test_should_not_allow_request_with_no_token(client, url):
response = client.get('/service') response = client.get(url)
assert response.status_code == 401 assert response.status_code == 401
data = json.loads(response.get_data()) data = json.loads(response.get_data())
assert data['message'] == {"token": ['Unauthorized, authentication token must be provided']} assert data['message'] == {"token": ['Unauthorized, authentication token must be provided']}
def test_should_not_allow_request_with_incorrect_header(notify_api): @pytest.mark.parametrize('url', ['/service', '/notifications'])
with notify_api.test_request_context(): def test_should_not_allow_request_with_incorrect_header(client, url):
with notify_api.test_client() as client: response = client.get(url, headers={'Authorization': 'Basic 1234'})
response = client.get( assert response.status_code == 401
'/service', data = json.loads(response.get_data())
headers={'Authorization': 'Basic 1234'}) assert data['message'] == {"token": ['Unauthorized, authentication bearer scheme must be used']}
assert response.status_code == 401
data = json.loads(response.get_data())
assert data['message'] == {"token": ['Unauthorized, authentication bearer scheme must be used']}
def test_should_not_allow_request_with_incorrect_token(notify_api, sample_user): @pytest.mark.parametrize('url', ['/service', '/notifications'])
with notify_api.test_request_context(): def test_should_not_allow_request_with_incorrect_token(client, url):
with notify_api.test_client() as client: response = client.get(url, headers={'Authorization': 'Bearer 1234'})
response = client.get( assert response.status_code == 403
'/service', data = json.loads(response.get_data())
headers={'Authorization': 'Bearer 1234'}) assert data['message'] == {"token": ['Invalid token: signature']}
assert response.status_code == 403
data = json.loads(response.get_data())
assert data['message'] == {"token": ['Invalid token: signature']}
def test_should_not_allow_request_with_no_iss(client): @pytest.mark.parametrize('url', ['/service', '/notifications'])
def test_should_not_allow_request_with_no_iss(client, url):
# code copied from notifications_python_client.authentication.py::create_jwt_token # code copied from notifications_python_client.authentication.py::create_jwt_token
headers = { headers = {
"typ": 'JWT', "typ": 'JWT',
@@ -58,13 +53,14 @@ def test_should_not_allow_request_with_no_iss(client):
token = jwt.encode(payload=claims, key=str(uuid.uuid4()), headers=headers).decode() token = jwt.encode(payload=claims, key=str(uuid.uuid4()), headers=headers).decode()
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)}) response = client.get(url, headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 403 assert response.status_code == 403
data = json.loads(response.get_data()) data = json.loads(response.get_data())
assert data['message'] == {"token": ['Invalid token: iss field not provided']} assert data['message'] == {"token": ['Invalid token: iss field not provided']}
def test_should_not_allow_request_with_no_iat(client, sample_api_key): @pytest.mark.parametrize('url', ['/service', '/notifications'])
def test_should_not_allow_request_with_no_iat(client, sample_api_key, url):
# code copied from notifications_python_client.authentication.py::create_jwt_token # code copied from notifications_python_client.authentication.py::create_jwt_token
headers = { headers = {
"typ": 'JWT', "typ": 'JWT',
@@ -78,264 +74,214 @@ def test_should_not_allow_request_with_no_iat(client, sample_api_key):
token = jwt.encode(payload=claims, key=str(uuid.uuid4()), headers=headers).decode() token = jwt.encode(payload=claims, key=str(uuid.uuid4()), headers=headers).decode()
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)}) response = client.get(url, headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 403 assert response.status_code == 403
data = json.loads(response.get_data()) data = json.loads(response.get_data())
assert data['message'] == {"token": ['Invalid token: signature, api token is not valid']} assert data['message'] == {"token": ['Invalid token: signature, api token is not valid']}
def test_should_not_allow_invalid_secret(notify_api, sample_api_key): @pytest.mark.parametrize('url', ['/service', '/notifications'])
with notify_api.test_request_context(): def test_should_not_allow_invalid_secret(client, sample_api_key, url):
with notify_api.test_client() as client: token = create_jwt_token(
token = create_jwt_token( secret="not-so-secret",
secret="not-so-secret", client_id=str(sample_api_key.service_id))
client_id=str(sample_api_key.service_id)) response = client.get(
response = client.get( url,
'/service', headers={'Authorization': "Bearer {}".format(token)}
headers={'Authorization': "Bearer {}".format(token)} )
) assert response.status_code == 403
assert response.status_code == 403 data = json.loads(response.get_data())
data = json.loads(response.get_data()) assert data['message'] == {"token": ['Invalid token: signature, api token is not valid']}
assert data['message'] == {"token": ['Invalid token: signature, api token is not valid']}
@pytest.mark.parametrize('scheme', ['bearer', 'Bearer']) @pytest.mark.parametrize('scheme', ['bearer', 'Bearer'])
def test_should_allow_valid_token(notify_api, sample_api_key, scheme): def test_should_allow_valid_token(client, sample_api_key, scheme):
with notify_api.test_request_context(): token = __create_token(sample_api_key.service_id)
with notify_api.test_client() as client: response = client.get('/notifications', headers={'Authorization': '{} {}'.format(scheme, token)})
token = __create_get_token(sample_api_key.service_id) assert response.status_code == 200
response = client.get(
'/service/{}'.format(str(sample_api_key.service_id)),
headers={'Authorization': '{} {}'.format(scheme, token)}
)
assert response.status_code == 200
def test_should_not_allow_service_id_that_is_not_the_wrong_data_type(notify_api, sample_api_key): @pytest.mark.parametrize('url', ['/service', '/notifications'])
with notify_api.test_request_context(): def test_should_not_allow_service_id_that_is_not_the_wrong_data_type(client, sample_api_key, url):
with notify_api.test_client() as client: token = create_jwt_token(secret=get_unsigned_secrets(sample_api_key.service_id)[0],
token = create_jwt_token(secret=get_unsigned_secrets(sample_api_key.service_id)[0], client_id=str('not-a-valid-id'))
client_id=str('not-a-valid-id')) response = client.get(
response = client.get( url,
'/service', headers={'Authorization': "Bearer {}".format(token)}
headers={'Authorization': "Bearer {}".format(token)} )
) assert response.status_code == 403
assert response.status_code == 403 data = json.loads(response.get_data())
data = json.loads(response.get_data()) assert data['message'] == {"token": ['Invalid token: service id is not the right data type']}
assert data['message'] == {"token": ['Invalid token: service id is not the right data type']}
def test_should_allow_valid_token_for_request_with_path_params(notify_api, sample_api_key): def test_should_allow_valid_token_for_request_with_path_params_for_public_url(client, sample_api_key):
with notify_api.test_request_context(): token = __create_token(sample_api_key.service_id)
with notify_api.test_client() as client: response = client.get('/notifications', headers={'Authorization': 'Bearer {}'.format(token)})
token = __create_get_token(sample_api_key.service_id) assert response.status_code == 200
response = client.get(
'/service/{}'.format(str(sample_api_key.service_id)),
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
def test_should_allow_valid_token_when_service_has_multiple_keys(notify_api, sample_api_key): def test_should_allow_valid_token_for_request_with_path_params_for_admin_url(client):
with notify_api.test_request_context(): token = create_jwt_token(current_app.config['ADMIN_CLIENT_SECRET'], current_app.config['ADMIN_CLIENT_USER_NAME'])
with notify_api.test_client() as client: response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)})
data = {'service': sample_api_key.service, assert response.status_code == 200
'name': 'some key name',
'created_by': sample_api_key.created_by,
'key_type': KEY_TYPE_NORMAL
}
api_key = ApiKey(**data)
save_model_api_key(api_key)
token = __create_get_token(sample_api_key.service_id)
response = client.get(
'/service/{}'.format(str(sample_api_key.service_id)),
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
def test_authentication_passes_admin_client_token(notify_api, def test_should_allow_valid_token_when_service_has_multiple_keys(client, sample_api_key):
sample_api_key): data = {'service': sample_api_key.service,
with notify_api.test_request_context(): 'name': 'some key name',
with notify_api.test_client() as client: 'created_by': sample_api_key.created_by,
token = create_jwt_token( 'key_type': KEY_TYPE_NORMAL
secret=current_app.config.get('ADMIN_CLIENT_SECRET'), }
client_id=current_app.config.get('ADMIN_CLIENT_USER_NAME')) api_key = ApiKey(**data)
response = client.get( save_model_api_key(api_key)
'/service', token = __create_token(sample_api_key.service_id)
headers={'Authorization': 'Bearer {}'.format(token)}) response = client.get(
assert response.status_code == 200 '/notifications'.format(str(sample_api_key.service_id)),
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
def test_authentication_passes_when_service_has_multiple_keys_some_expired( def test_authentication_passes_when_service_has_multiple_keys_some_expired(
notify_api, client,
notify_db,
notify_db_session,
sample_api_key): sample_api_key):
with notify_api.test_request_context(): expired_key_data = {'service': sample_api_key.service,
with notify_api.test_client() as client: 'name': 'expired_key',
expired_key_data = {'service': sample_api_key.service, 'expiry_date': datetime.utcnow(),
'name': 'expired_key', 'created_by': sample_api_key.created_by,
'expiry_date': datetime.utcnow(), 'key_type': KEY_TYPE_NORMAL
'created_by': sample_api_key.created_by, }
'key_type': KEY_TYPE_NORMAL expired_key = ApiKey(**expired_key_data)
} save_model_api_key(expired_key)
expired_key = ApiKey(**expired_key_data) another_key = {'service': sample_api_key.service,
save_model_api_key(expired_key) 'name': 'another_key',
another_key = {'service': sample_api_key.service, 'created_by': sample_api_key.created_by,
'name': 'another_key', 'key_type': KEY_TYPE_NORMAL
'created_by': sample_api_key.created_by, }
'key_type': KEY_TYPE_NORMAL api_key = ApiKey(**another_key)
} save_model_api_key(api_key)
api_key = ApiKey(**another_key) token = create_jwt_token(
save_model_api_key(api_key) secret=get_unsigned_secret(api_key.id),
token = create_jwt_token( client_id=str(sample_api_key.service_id))
secret=get_unsigned_secret(api_key.id), response = client.get(
client_id=str(sample_api_key.service_id)) '/notifications',
response = client.get( headers={'Authorization': 'Bearer {}'.format(token)})
'/service', assert response.status_code == 200
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
def test_authentication_returns_token_expired_when_service_uses_expired_key_and_has_multiple_keys(notify_api, def test_authentication_returns_token_expired_when_service_uses_expired_key_and_has_multiple_keys(client,
notify_db,
notify_db_session,
sample_api_key): sample_api_key):
with notify_api.test_request_context(): expired_key = {'service': sample_api_key.service,
with notify_api.test_client() as client: 'name': 'expired_key',
expired_key = {'service': sample_api_key.service, 'created_by': sample_api_key.created_by,
'name': 'expired_key', 'key_type': KEY_TYPE_NORMAL
'created_by': sample_api_key.created_by, }
'key_type': KEY_TYPE_NORMAL expired_api_key = ApiKey(**expired_key)
} save_model_api_key(expired_api_key)
expired_api_key = ApiKey(**expired_key) another_key = {'service': sample_api_key.service,
save_model_api_key(expired_api_key) 'name': 'another_key',
another_key = {'service': sample_api_key.service, 'created_by': sample_api_key.created_by,
'name': 'another_key', 'key_type': KEY_TYPE_NORMAL
'created_by': sample_api_key.created_by, }
'key_type': KEY_TYPE_NORMAL api_key = ApiKey(**another_key)
} save_model_api_key(api_key)
api_key = ApiKey(**another_key) token = create_jwt_token(
save_model_api_key(api_key) secret=get_unsigned_secret(expired_api_key.id),
token = create_jwt_token( client_id=str(sample_api_key.service_id))
secret=get_unsigned_secret(expired_api_key.id), expire_api_key(service_id=sample_api_key.service_id, api_key_id=expired_api_key.id)
client_id=str(sample_api_key.service_id)) response = client.get(
expire_api_key(service_id=sample_api_key.service_id, api_key_id=expired_api_key.id) '/notifications',
response = client.get( headers={'Authorization': 'Bearer {}'.format(token)})
'/service', assert response.status_code == 403
headers={'Authorization': 'Bearer {}'.format(token)}) data = json.loads(response.get_data())
assert response.status_code == 403 assert data['message'] == {"token": ['Invalid token: API key revoked']}
data = json.loads(response.get_data())
assert data['message'] == {"token": ['Invalid token: API key revoked']}
def test_authentication_returns_error_when_admin_client_has_no_secrets(notify_api, def test_authentication_returns_error_when_admin_client_has_no_secrets(client):
sample_service): api_secret = current_app.config.get('ADMIN_CLIENT_SECRET')
with notify_api.test_request_context(): token = create_jwt_token(
with notify_api.test_client() as client: secret=api_secret,
api_secret = notify_api.config.get('ADMIN_CLIENT_SECRET') client_id=current_app.config.get('ADMIN_CLIENT_USER_NAME')
token = create_jwt_token( )
secret=api_secret, current_app.config['ADMIN_CLIENT_SECRET'] = ''
client_id=notify_api.config.get('ADMIN_CLIENT_USER_NAME') response = client.get(
) '/service',
notify_api.config['ADMIN_CLIENT_SECRET'] = '' headers={'Authorization': 'Bearer {}'.format(token)})
response = client.get( assert response.status_code == 403
'/service', error_message = json.loads(response.get_data())
headers={'Authorization': 'Bearer {}'.format(token)}) assert error_message['message'] == {"token": ['Invalid token: signature']}
assert response.status_code == 403 current_app.config['ADMIN_CLIENT_SECRET'] = api_secret
error_message = json.loads(response.get_data())
assert error_message['message'] == {"token": ['Invalid token: signature']}
notify_api.config['ADMIN_CLIENT_SECRET'] = api_secret
def test_authentication_returns_error_when_service_doesnt_exit( def test_authentication_returns_error_when_service_doesnt_exit(
notify_api, client,
sample_api_key sample_api_key
): ):
with notify_api.test_request_context(), notify_api.test_client() as client: # get service ID and secret the wrong way around
# get service ID and secret the wrong way around token = create_jwt_token(
token = create_jwt_token( secret=str(sample_api_key.service_id),
secret=str(sample_api_key.service_id), client_id=str(sample_api_key.id))
client_id=str(sample_api_key.id))
response = client.get( response = client.get(
'/service', '/notifications',
headers={'Authorization': 'Bearer {}'.format(token)} headers={'Authorization': 'Bearer {}'.format(token)}
) )
assert response.status_code == 403 assert response.status_code == 403
error_message = json.loads(response.get_data()) error_message = json.loads(response.get_data())
assert error_message['message'] == {'token': ['Invalid token: service not found']} assert error_message['message'] == {'token': ['Invalid token: service not found']}
def test_authentication_returns_error_when_service_inactive(client, sample_api_key): def test_authentication_returns_error_when_service_inactive(client, sample_api_key):
sample_api_key.service.active = False sample_api_key.service.active = False
token = create_jwt_token(secret=str(sample_api_key.id), client_id=str(sample_api_key.service_id)) token = create_jwt_token(secret=str(sample_api_key.id), client_id=str(sample_api_key.service_id))
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)}) response = client.get('/notifications', headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 403 assert response.status_code == 403
error_message = json.loads(response.get_data()) error_message = json.loads(response.get_data())
assert error_message['message'] == {'token': ['Invalid token: service is archived']} assert error_message['message'] == {'token': ['Invalid token: service is archived']}
def test_authentication_returns_error_when_service_has_no_secrets(notify_api, def test_authentication_returns_error_when_service_has_no_secrets(client,
sample_service, sample_service,
fake_uuid): fake_uuid):
with notify_api.test_request_context(): token = create_jwt_token(
with notify_api.test_client() as client: secret=fake_uuid,
token = create_jwt_token( client_id=str(sample_service.id))
secret=fake_uuid,
client_id=str(sample_service.id))
response = client.get( response = client.get(
'/service', '/notifications',
headers={'Authorization': 'Bearer {}'.format(token)}) headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 403 assert response.status_code == 403
error_message = json.loads(response.get_data()) error_message = json.loads(response.get_data())
assert error_message['message'] == {'token': ['Invalid token: service has no API keys']} assert error_message['message'] == {'token': ['Invalid token: service has no API keys']}
def test_should_attach_the_current_api_key_to_current_app(notify_api, sample_service, sample_api_key): def test_should_attach_the_current_api_key_to_current_app(notify_api, sample_service, sample_api_key):
with notify_api.test_request_context() as context, notify_api.test_client() as client: with notify_api.test_request_context() as context, notify_api.test_client() as client:
with pytest.raises(AttributeError): token = __create_token(sample_api_key.service_id)
print(api_user)
token = __create_get_token(sample_api_key.service_id)
response = client.get( response = client.get(
'/service/{}'.format(str(sample_api_key.service_id)), '/notifications',
headers={'Authorization': 'Bearer {}'.format(token)} headers={'Authorization': 'Bearer {}'.format(token)}
) )
assert response.status_code == 200 assert response.status_code == 200
assert api_user == sample_api_key assert api_user == sample_api_key
def test_should_return_403_when_token_is_expired(notify_api, def test_should_return_403_when_token_is_expired(client,
sample_api_key): sample_api_key):
with notify_api.test_request_context(): with freeze_time('2001-01-01T12:00:00'):
with notify_api.test_client() as client: token = __create_token(sample_api_key.service_id)
with freeze_time('2001-01-01T12:00:00'): with freeze_time('2001-01-01T12:00:40'):
token = __create_get_token(sample_api_key.service_id) response = client.get(
with freeze_time('2001-01-01T12:00:40'): '/notifications',
response = client.get( headers={'Authorization': 'Bearer {}'.format(token)})
'/service', assert response.status_code == 403
headers={'Authorization': 'Bearer {}'.format(token)}) error_message = json.loads(response.get_data())
assert response.status_code == 403 assert error_message['message'] == {'token': [
error_message = json.loads(response.get_data()) 'Invalid token: expired, check that your system clock is accurate'
assert error_message['message'] == {'token': [ ]}
'Invalid token: expired, check that your system clock is accurate'
]}
def __create_get_token(service_id): def __create_token(service_id):
if service_id: return create_jwt_token(secret=get_unsigned_secrets(service_id)[0],
return create_jwt_token(secret=get_unsigned_secrets(service_id)[0], client_id=str(service_id))
client_id=str(service_id))
else:
return create_jwt_token(secret=get_unsigned_secrets(service_id)[0],
client_id=service_id)
def __create_post_token(service_id, request_body):
return create_jwt_token(
secret=get_unsigned_secrets(service_id)[0],
client_id=str(service_id)
)

8
tests/app/service/test_rest.py
View File

@@ -1418,7 +1418,7 @@ def test_get_notification_billable_unit_count(client, notify_db, notify_db_sessi
notification = create_sample_notification(notify_db, notify_db_session) notification = create_sample_notification(notify_db, notify_db_session)
response = client.get( response = client.get(
'/service/{}/billable-units?year=2012'.format(notification.service_id), '/service/{}/billable-units?year=2012'.format(notification.service_id),
headers=[create_authorization_header(service_id=notification.service_id)] headers=[create_authorization_header()]
) )
assert response.status_code == 200 assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == { assert json.loads(response.get_data(as_text=True)) == {
@@ -1429,7 +1429,7 @@ def test_get_notification_billable_unit_count(client, notify_db, notify_db_sessi
def test_get_notification_billable_unit_count_missing_year(client, sample_service): def test_get_notification_billable_unit_count_missing_year(client, sample_service):
response = client.get( response = client.get(
'/service/{}/billable-units'.format(sample_service.id), '/service/{}/billable-units'.format(sample_service.id),
headers=[create_authorization_header(service_id=sample_service.id)] headers=[create_authorization_header()]
) )
assert response.status_code == 400 assert response.status_code == 400
assert json.loads(response.get_data(as_text=True)) == { assert json.loads(response.get_data(as_text=True)) == {
@@ -1451,7 +1451,7 @@ def test_get_service_provider_aggregate_statistics(
): ):
response = client.get( response = client.get(
'/service/{}/fragment/aggregate_statistics{}'.format(sample_service.id, query_string), '/service/{}/fragment/aggregate_statistics{}'.format(sample_service.id, query_string),
headers=[create_authorization_header(service_id=sample_service.id)] headers=[create_authorization_header()]
) )
assert response.status_code == expected_status assert response.status_code == expected_status
assert json.loads(response.get_data(as_text=True)) == expected_json assert json.loads(response.get_data(as_text=True)) == expected_json
@@ -1496,7 +1496,7 @@ def test_get_template_stats_by_month_returns_error_for_incorrect_year(
): ):
response = client.get( response = client.get(
'/service/{}/notifications/templates/monthly{}'.format(sample_service.id, query_string), '/service/{}/notifications/templates/monthly{}'.format(sample_service.id, query_string),
headers=[create_authorization_header(service_id=sample_service.id)] headers=[create_authorization_header()]
) )
assert response.status_code == expected_status assert response.status_code == expected_status
assert json.loads(response.get_data(as_text=True)) == expected_json assert json.loads(response.get_data(as_text=True)) == expected_json