From e08d726f0579e023c565f5cf42dff5dc95e41a0d Mon Sep 17 00:00:00 2001 From: Ben Thorner Date: Tue, 27 Jul 2021 17:25:22 +0100 Subject: [PATCH] DRY-up and simplify creating JWT tokens in tests 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. --- .../app/authentication/test_authentication.py | 232 ++++++++++++------ 1 file changed, 151 insertions(+), 81 deletions(-) diff --git a/tests/app/authentication/test_authentication.py b/tests/app/authentication/test_authentication.py index 33048e249..43b5c4f02 100644 --- a/tests/app/authentication/test_authentication.py +++ b/tests/app/authentication/test_authentication.py @@ -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): + client, + 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 == [