diff --git a/app/notify_client/__init__.py b/app/notify_client/__init__.py index 609798de5..fb0cb7f19 100644 --- a/app/notify_client/__init__.py +++ b/app/notify_client/__init__.py @@ -1,5 +1,5 @@ from flask_login import current_user -from flask import has_request_context, request +from flask import has_request_context, request, abort from notifications_python_client.base import BaseAPIClient from notifications_python_client.version import __version__ @@ -26,3 +26,25 @@ class NotifyAdminAPIClient(BaseAPIClient): return headers headers['NotifyRequestID'] = request.request_id return headers + + def check_inactive_service(self): + # this file is imported in app/__init__.py before current_service is initialised, so need to import later + # to prevent cyclical imports + from app import current_service + + # if the current service is inactive and the user isn't a platform admin, we should block them from making any + # stateful modifications to that service + if current_service and not current_service['active'] and not current_user.platform_admin: + abort(403) + + def post(self, *args, **kwargs): + self.check_inactive_service() + return super().post(*args, **kwargs) + + def put(self, *args, **kwargs): + self.check_inactive_service() + return super().put(*args, **kwargs) + + def delete(self, *args, **kwargs): + self.check_inactive_service() + return super().delete(*args, **kwargs) diff --git a/tests/app/notify_client/test_notify_admin_api_client.py b/tests/app/notify_client/test_notify_admin_api_client.py new file mode 100644 index 000000000..2e0fe9ef5 --- /dev/null +++ b/tests/app/notify_client/test_notify_admin_api_client.py @@ -0,0 +1,96 @@ +import uuid +import pytest +from unittest.mock import patch + +import werkzeug + +from tests import service_json +from tests.conftest import api_user_active, platform_admin_user +from app.notify_client import NotifyAdminAPIClient + + +@pytest.mark.parametrize('method', [ + 'put', + 'post', + 'delete' +]) +@pytest.mark.parametrize('user', [ + api_user_active(str(uuid.uuid4())), + platform_admin_user(str(uuid.uuid4())) +], ids=['api_user', 'platform_admin']) +@pytest.mark.parametrize('service', [ + service_json(active=True), + None +], ids=['active_service', 'no_service']) +def test_active_service_can_be_modified(app_, method, user, service): + api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id') + + with app_.test_request_context() as request_context, app_.test_client() as client: + client.login(user) + request_context.service = service + + with patch.object(api_client, 'request') as request: + ret = getattr(api_client, method)('url', 'data') + + assert request.called + assert ret == request.return_value + + +@pytest.mark.parametrize('method', [ + 'put', + 'post', + 'delete' +]) +def test_inactive_service_cannot_be_modified_by_normal_user(app_, api_user_active, method): + api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id') + + with app_.test_request_context() as request_context, app_.test_client() as client: + client.login(api_user_active) + request_context.service = service_json(active=False) + + with patch.object(api_client, 'request') as request: + with pytest.raises(werkzeug.exceptions.Forbidden): + getattr(api_client, method)('url', 'data') + + assert not request.called + + +@pytest.mark.parametrize('method', [ + 'put', + 'post', + 'delete' +]) +def test_inactive_service_can_be_modified_by_platform_admin(app_, platform_admin_user, method): + api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id') + + with app_.test_request_context() as request_context, app_.test_client() as client: + client.login(platform_admin_user) + request_context.service = service_json(active=False) + + with patch.object(api_client, 'request') as request: + ret = getattr(api_client, method)('url', 'data') + + assert request.called + assert ret == request.return_value + + +def test_generate_headers_sets_standard_headers(): + api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id') + + # with patch('app.notify_client.has_request_context', return_value=False): + headers = api_client.generate_headers('api_token') + + assert set(headers.keys()) == {'Authorization', 'Content-type', 'User-agent'} + assert headers['Authorization'] == 'Bearer api_token' + assert headers['Content-type'] == 'application/json' + assert headers['User-agent'].startswith('NOTIFY-API-PYTHON-CLIENT') + + +def test_generate_headers_sets_request_id_if_in_request_context(app_): + api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id') + + with app_.test_request_context() as request_context: + headers = api_client.generate_headers('api_token') + + assert set(headers.keys()) == {'Authorization', 'Content-type', 'User-agent', 'NotifyRequestID'} + assert headers['NotifyRequestID'] == request_context.request.request_id