darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-21 07:51:13 -05:00
Code Issues Packages Projects Releases Wiki Activity

DRY-up and simplify creating JWT tokens in tests

Browse Source 
Previously this was heavily duplicated but with the odd test using
a __create_token method. This adds some fixtures to remove all the
boilerplate and standardise how tokens are created in each test.
This commit is contained in:
Ben Thorner
2021-07-27 17:25:22 +01:00
parent 36afd4210f
commit e08d726f05
1 changed files with 151 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -34,6 +34,34 @@ def create_custom_jwt_token(headers=None, payload=None, key=None):
return jwt.encode(payload=payload, key=key or str(uuid.uuid4()), headers=headers)
@pytest.fixture
def service_jwt_secret(sample_api_key):
return get_unsigned_secrets(sample_api_key.service_id)[0]
@pytest.fixture
def service_jwt_token(sample_api_key, service_jwt_secret):
return create_jwt_token(
secret=service_jwt_secret,
client_id=str(sample_api_key.service_id),
)
@pytest.fixture
def admin_jwt_client_id():
return current_app.config['ADMIN_CLIENT_USER_NAME']
@pytest.fixture
def admin_jwt_secret(admin_jwt_client_id):
return current_app.config['INTERNAL_CLIENT_API_KEYS'][admin_jwt_client_id][0]
@pytest.fixture
def admin_jwt_token(admin_jwt_client_id, admin_jwt_secret):
return create_jwt_token(admin_jwt_secret, admin_jwt_client_id)
@pytest.mark.parametrize('auth_fn', [requires_auth, requires_admin_auth])
def test_should_not_allow_request_with_no_token(client, auth_fn):
request.headers = {}
@@ -93,13 +121,14 @@ def test_requires_auth_should_not_allow_request_with_non_hs256_algorithm(client,
assert exc.value.short_message == 'Invalid token: algorithm used is not HS256'
def test_requires_admin_auth_should_not_allow_request_with_no_iat(client):
client_id = current_app.config['ADMIN_CLIENT_USER_NAME']
secret = current_app.config['INTERNAL_CLIENT_API_KEYS'][client_id][0]
def test_requires_admin_auth_should_not_allow_request_with_no_iat(
client,
admin_jwt_client_id,
admin_jwt_secret,
):
token = create_custom_jwt_token(
payload={'iss': client_id},
key=secret
payload={'iss': admin_jwt_client_id},
key=admin_jwt_secret
)
request.headers = {'Authorization': 'Bearer {}'.format(token)}
@@ -108,13 +137,14 @@ def test_requires_admin_auth_should_not_allow_request_with_no_iat(client):
assert exc.value.short_message == "Unauthorized: API authentication token not found"
def test_requires_admin_auth_should_not_allow_request_with_old_iat(client):
client_id = current_app.config['ADMIN_CLIENT_USER_NAME']
secret = current_app.config['INTERNAL_CLIENT_API_KEYS'][client_id][0]
def test_requires_admin_auth_should_not_allow_request_with_old_iat(
client,
admin_jwt_client_id,
admin_jwt_secret,
):
token = create_custom_jwt_token(
payload={'iss': client_id, 'iat': int(time.time()) - 60},
key=secret
payload={'iss': admin_jwt_client_id, 'iat': int(time.time()) - 60},
key=admin_jwt_secret
)
request.headers = {'Authorization': 'Bearer {}'.format(token)}
@@ -123,16 +153,18 @@ def test_requires_admin_auth_should_not_allow_request_with_old_iat(client):
assert exc.value.short_message == "Invalid token: expired, check that your system clock is accurate"
def test_requires_auth_should_not_allow_request_with_extra_claims(client, sample_api_key):
key = get_unsigned_secrets(sample_api_key.service_id)[0]
def test_requires_auth_should_not_allow_request_with_extra_claims(
client,
sample_api_key,
service_jwt_secret,
):
token = create_custom_jwt_token(
payload={
'iss': str(sample_api_key.service_id),
'iat': int(time.time()),
'aud': 'notifications.service.gov.uk' # extra claim that we don't support
},
key=key
key=service_jwt_secret,
)
request.headers = {'Authorization': 'Bearer {}'.format(token)}
@@ -155,16 +187,30 @@ def test_requires_auth_should_not_allow_invalid_secret(client, sample_api_key):
@pytest.mark.parametrize('scheme', ['bearer', 'Bearer'])
def test_requires_auth_should_allow_valid_token(client, sample_api_key, scheme):
token = __create_token(sample_api_key.service_id)
def test_requires_auth_should_allow_valid_token(
client,
sample_api_key,
service_jwt_secret,
scheme,
):
token = create_jwt_token(
client_id=str(sample_api_key.service_id),
secret=service_jwt_secret,
)
response = client.get('/notifications', headers={'Authorization': '{} {}'.format(scheme, token)})
assert response.status_code == 200
@pytest.mark.parametrize('service_id', ['not-a-valid-id', 1234])
def test_requires_auth_should_not_allow_service_id_with_the_wrong_data_type(client, sample_api_key, service_id):
token = create_jwt_token(secret=get_unsigned_secrets(sample_api_key.service_id)[0],
client_id=service_id)
def test_requires_auth_should_not_allow_service_id_with_the_wrong_data_type(
client,
service_jwt_secret,
service_id
):
token = create_jwt_token(
client_id=service_id,
secret=service_jwt_secret,
)
response = client.get(
'/notifications',
headers={'Authorization': "Bearer {}".format(token)}
@@ -174,36 +220,43 @@ def test_requires_auth_should_not_allow_service_id_with_the_wrong_data_type(clie
assert data['message'] == {"token": ['Invalid token: service id is not the right data type']}
def test_requires_auth_should_allow_valid_token_for_request_with_path_params_for_public_url(client, sample_api_key):
token = __create_token(sample_api_key.service_id)
response = client.get('/notifications', headers={'Authorization': 'Bearer {}'.format(token)})
def test_requires_auth_should_allow_valid_token_for_request_with_path_params_for_public_url(
client,
service_jwt_token,
):
response = client.get('/notifications', headers={'Authorization': 'Bearer {}'.format(service_jwt_token)})
assert response.status_code == 200
def test_requires_admin_auth_should_allow_valid_token_for_request_with_path_params(client):
client_id = current_app.config['ADMIN_CLIENT_USER_NAME']
secret = current_app.config['INTERNAL_CLIENT_API_KEYS'][client_id][0]
token = create_jwt_token(secret, client_id)
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)})
def test_requires_admin_auth_should_allow_valid_token_for_request_with_path_params(
client,
admin_jwt_token
):
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(admin_jwt_token)})
assert response.status_code == 200
def test_requires_admin_auth_should_allow_valid_token_for_request_with_path_params_with_second_secret(client):
client_id = current_app.config['ADMIN_CLIENT_USER_NAME']
new_secrets = {client_id: ["secret1", "secret2"]}
def test_requires_admin_auth_should_allow_valid_token_for_request_with_path_params_with_second_secret(
client,
admin_jwt_client_id,
):
new_secrets = {admin_jwt_client_id: ["secret1", "secret2"]}
with set_config(client.application, 'INTERNAL_CLIENT_API_KEYS', new_secrets):
token = create_jwt_token("secret1", client_id)
token = create_jwt_token("secret1", admin_jwt_client_id)
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
token = create_jwt_token("secret2", client_id)
token = create_jwt_token("secret2", admin_jwt_client_id)
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
def test_requires_auth_should_allow_valid_token_when_service_has_multiple_keys(client, sample_api_key):
def test_requires_auth_should_allow_valid_token_when_service_has_multiple_keys(
client,
sample_api_key,
service_jwt_token,
):
data = {'service': sample_api_key.service,
'name': 'some key name',
'created_by': sample_api_key.created_by,
@@ -211,16 +264,16 @@ def test_requires_auth_should_allow_valid_token_when_service_has_multiple_keys(c
}
api_key = ApiKey(**data)
save_model_api_key(api_key)
token = __create_token(sample_api_key.service_id)
response = client.get(
'/notifications',
headers={'Authorization': 'Bearer {}'.format(token)})
headers={'Authorization': 'Bearer {}'.format(service_jwt_token)})
assert response.status_code == 200
def test_requires_auth_passes_when_service_has_multiple_keys_some_expired(
client,
sample_api_key):
sample_api_key,
):
expired_key_data = {'service': sample_api_key.service,
'name': 'expired_key',
'expiry_date': datetime.utcnow(),
@@ -237,8 +290,9 @@ def test_requires_auth_passes_when_service_has_multiple_keys_some_expired(
api_key = ApiKey(**another_key)
save_model_api_key(api_key)
token = create_jwt_token(
secret=get_unsigned_secret(api_key.id),
client_id=str(sample_api_key.service_id))
client_id=str(sample_api_key.service_id),
secret=get_unsigned_secret(api_key.id)
)
response = client.get(
'/notifications',
headers={'Authorization': 'Bearer {}'.format(token)})
@@ -246,7 +300,8 @@ def test_requires_auth_passes_when_service_has_multiple_keys_some_expired(
def test_requires_auth_returns_token_expired_when_service_uses_expired_key_and_has_multiple_keys(
client, sample_api_key
client,
sample_api_key
):
expired_key = {'service': sample_api_key.service,
'name': 'expired_key',
@@ -263,8 +318,9 @@ def test_requires_auth_returns_token_expired_when_service_uses_expired_key_and_h
api_key = ApiKey(**another_key)
save_model_api_key(api_key)
token = create_jwt_token(
secret=get_unsigned_secret(expired_api_key.id),
client_id=str(sample_api_key.service_id))
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)
request.headers = {'Authorization': 'Bearer {}'.format(token)}
with pytest.raises(AuthError) as exc:
@@ -274,39 +330,41 @@ def test_requires_auth_returns_token_expired_when_service_uses_expired_key_and_h
assert exc.value.api_key_id == expired_api_key.id
def test_requires_admin_auth_returns_error_with_no_secrets(client):
client_id = current_app.config.get('ADMIN_CLIENT_USER_NAME')
secret = current_app.config.get('INTERNAL_CLIENT_API_KEYS')[client_id][0]
token = create_jwt_token(secret, client_id)
new_secrets = {client_id: []}
def test_requires_admin_auth_returns_error_with_no_secrets(
client,
admin_jwt_client_id,
admin_jwt_token,
):
new_secrets = {admin_jwt_client_id: []}
with set_config(client.application, 'INTERNAL_CLIENT_API_KEYS', new_secrets):
response = client.get(
'/service',
headers={'Authorization': 'Bearer {}'.format(token)})
headers={'Authorization': 'Bearer {}'.format(admin_jwt_token)})
assert response.status_code == 401
error_message = json.loads(response.get_data())
assert error_message['message'] == {"token": ["Unauthorized: API authentication token not found"]}
def test_requires_admin_auth_returns_error_when_secret_is_invalid(client):
client_id = current_app.config.get('ADMIN_CLIENT_USER_NAME')
secret = current_app.config.get('INTERNAL_CLIENT_API_KEYS')[client_id][0]
token = create_jwt_token(secret, client_id)
new_secrets = {client_id: ['something-wrong']}
def test_requires_admin_auth_returns_error_when_secret_is_invalid(
client,
admin_jwt_client_id,
admin_jwt_token,
):
new_secrets = {admin_jwt_client_id: ['something-wrong']}
with set_config(client.application, 'INTERNAL_CLIENT_API_KEYS', new_secrets):
response = client.get(
'/service',
headers={'Authorization': 'Bearer {}'.format(token)})
headers={'Authorization': 'Bearer {}'.format(admin_jwt_token)})
assert response.status_code == 401
error_message = json.loads(response.get_data())
assert error_message['message'] == {"token": ["Unauthorized: API authentication token not found"]}
def test_requires_auth_returns_error_when_service_doesnt_exit(
def test_requires_auth_returns_error_when_service_doesnt_exist(
client,
sample_api_key
):
@@ -324,11 +382,13 @@ def test_requires_auth_returns_error_when_service_doesnt_exit(
assert error_message['message'] == {'token': ['Invalid token: service not found']}
def test_requires_auth_returns_error_when_service_inactive(client, sample_api_key):
def test_requires_auth_returns_error_when_service_inactive(
client,
sample_api_key,
service_jwt_token,
):
sample_api_key.service.active = False
token = create_jwt_token(secret=str(sample_api_key.id), client_id=str(sample_api_key.service_id))
response = client.get('/notifications', headers={'Authorization': 'Bearer {}'.format(token)})
response = client.get('/notifications', headers={'Authorization': 'Bearer {}'.format(service_jwt_token)})
assert response.status_code == 403
error_message = json.loads(response.get_data())
@@ -349,22 +409,31 @@ def test_requires_auth_returns_error_when_service_has_no_secrets(
assert exc.value.service_id == str(sample_service.id)
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,
service_jwt_token,
):
with notify_api.test_request_context(), notify_api.test_client() as client:
token = __create_token(sample_api_key.service_id)
response = client.get(
'/notifications',
headers={'Authorization': 'Bearer {}'.format(token)}
headers={'Authorization': 'Bearer {}'.format(service_jwt_token)}
)
assert response.status_code == 200
assert str(api_user.id) == str(sample_api_key.id)
def test_requires_auth_return_403_when_token_is_expired(
client, sample_api_key
client,
sample_api_key,
service_jwt_secret,
):
with freeze_time('2001-01-01T12:00:00'):
token = __create_token(sample_api_key.service_id)
token = create_jwt_token(
client_id=str(sample_api_key.service_id),
secret=service_jwt_secret,
)
with freeze_time('2001-01-01T12:00:40'):
with pytest.raises(AuthError) as exc:
request.headers = {'Authorization': 'Bearer {}'.format(token)}
@@ -374,11 +443,6 @@ def test_requires_auth_return_403_when_token_is_expired(
assert str(exc.value.api_key_id) == str(sample_api_key.id)
def __create_token(service_id):
return create_jwt_token(secret=get_unsigned_secrets(service_id)[0],
client_id=str(service_id))
@pytest.mark.parametrize('check_proxy_header,header_value', [
(True, 'key_1'),
(True, 'wrong_key'),
@@ -408,11 +472,13 @@ def test_requires_no_auth_proxy_key(notify_api, check_proxy_header, header_value
(False, 'key_1', 200),
(False, 'wrong_key', 200),
])
def test_requires_admin_auth_proxy_key(notify_api, check_proxy_header, header_value, expected_status):
client_id = current_app.config.get('ADMIN_CLIENT_USER_NAME')
secret = current_app.config.get('INTERNAL_CLIENT_API_KEYS')[client_id][0]
token = create_jwt_token(secret, client_id)
def test_requires_admin_auth_proxy_key(
notify_api,
check_proxy_header,
header_value,
expected_status,
admin_jwt_token,
):
with set_config_values(notify_api, {
'ROUTE_SECRET_KEY_1': 'key_1',
'ROUTE_SECRET_KEY_2': '',
@@ -424,13 +490,18 @@ def test_requires_admin_auth_proxy_key(notify_api, check_proxy_header, header_va
path='/service',
headers=[
('X-Custom-Forwarder', header_value),
('Authorization', 'Bearer {}'.format(token))
('Authorization', 'Bearer {}'.format(admin_jwt_token))
]
)
assert response.status_code == expected_status
def test_requires_auth_should_cache_service_and_api_key_lookups(mocker, client, sample_api_key):
def test_requires_auth_should_cache_service_and_api_key_lookups(
mocker,
client,
sample_api_key,
service_jwt_token
):
mock_get_api_keys = mocker.patch(
'app.serialised_models.get_model_api_keys',
wraps=get_model_api_keys,
@@ -441,9 +512,8 @@ def test_requires_auth_should_cache_service_and_api_key_lookups(mocker, client,
)
for _ in range(5):
token = __create_token(sample_api_key.service_id)
client.get('/notifications', headers={
'Authorization': f'Bearer {token}'
'Authorization': f'Bearer {service_jwt_token}'
})
assert mock_get_api_keys.call_args_list == [