mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-07-16 03:10:16 -04:00
block inactive services from making stateful changes
in the NotifyAdminAPIClient, which all api traffic goes through, return
403 for any stateful requests (post, put and delete), if the following
criteria have been met:
* a current_service is set
(this prevents checks being carried out on non-service related
updates, eg editing user details)
* the service is not active
* the current user is not a platform admin
so platform admins can still update anything.
Note: Without any specific error handling, the user will see a generic
403 page. This is fine, probably - it's a relatively niche case that
you'll be editing a service you can't get to anyway
This commit is contained in:
@@ -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,29 @@ 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
|
||||
print('\n\n\n')
|
||||
print(current_service)
|
||||
print(current_user)
|
||||
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)
|
||||
|
||||
74
tests/app/notify_client/test_notify_admin_api_client.py
Normal file
74
tests/app/notify_client/test_notify_admin_api_client.py
Normal file
@@ -0,0 +1,74 @@
|
||||
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
|
||||
Reference in New Issue
Block a user