mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-02-27 22:00:59 -05:00
acfe8092fc
Currently requests to the API made from the admin app are going from PaaS admin app to the nginx router ELB, which then routes them back to the api app on PaaS. This makes sense for external requests, but for requests made from the admin app we could skip nginx and go directly to the api PaaS host, which should reduce load on the nginx instances and potentially reduce latency of the api requests. API apps on PaaS are checking the X-Custom-Forwarder header (which is set by nginx on proxy_pass requests) to only allow requests going through the proxy. This adds the custom header to the API client requests, so that they can pass that header check without going through nginx.
58 lines
2.0 KiB
Python
58 lines
2.0 KiB
Python
from flask_login import current_user
|
|
from flask import has_request_context, request, abort
|
|
from notifications_python_client.base import BaseAPIClient
|
|
from notifications_python_client import __version__
|
|
|
|
|
|
def _attach_current_user(data):
|
|
return dict(
|
|
created_by=current_user.id,
|
|
**data
|
|
)
|
|
|
|
|
|
class NotifyAdminAPIClient(BaseAPIClient):
|
|
def init_app(self, app):
|
|
self.base_url = app.config['API_HOST_NAME']
|
|
self.service_id = app.config['ADMIN_CLIENT_USER_NAME']
|
|
self.api_key = app.config['ADMIN_CLIENT_SECRET']
|
|
self.route_secret = app.config['ROUTE_SECRET_KEY_1']
|
|
|
|
def generate_headers(self, api_token):
|
|
headers = {
|
|
"Content-type": "application/json",
|
|
"Authorization": "Bearer {}".format(api_token),
|
|
"X-Custom-Forwarder": self.route_secret,
|
|
"User-agent": "NOTIFY-API-PYTHON-CLIENT/{}".format(__version__)
|
|
}
|
|
return self._add_request_id_header(headers)
|
|
|
|
@staticmethod
|
|
def _add_request_id_header(headers):
|
|
if not has_request_context():
|
|
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)
|