import copy import logging from datetime import datetime from typing import Any import pytz import tornado import tornado_eventsource.handler from tornado import httputil from tornado.web import Application from core.enums import AlertType from core.utils import safe_json_dumps from data.lookup_credentials import extract_credentials logger = logging.getLogger(__name__) class APIAlertsHandler(tornado.web.RequestHandler): """API request handler for /api/v2/alerts""" def __init__( self, application: "Application", request: httputil.HTTPServerRequest, **kwargs: Any, ): self._alerts = None super().__init__(application, request, **kwargs) def initialize(self, alerts): self._alerts = alerts @staticmethod def _enrich(alerts, credentials): enriched = [] for alert in alerts: alert_copy = copy.deepcopy(alert) alert_copy.infer_missing(credentials) enriched.append(alert_copy) return enriched def get(self): try: # request.arguments contains lists for each param key because technically the client can supply multiple, # reduce that to just the first entry, and convert bytes to string query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()} # Fetch all alerts matching the query, then optionally enrich with online data credentials = extract_credentials(self.request.headers) data = get_alert_list_with_filters(self._alerts, query_params) fields = [f.strip() for f in query_params["fields"].split(",")] if "fields" in query_params else [] if credentials: data = self._enrich(data, credentials) # Filter for only the required fields, if necessary if fields: data = filter_fields(data, fields) self.write(safe_json_dumps(data)) self.set_status(200) except ValueError as e: self.write(safe_json_dumps(f"Bad request - {e!s}")) self.set_status(400) except Exception: logger.exception("Exception when handling client request to alerts API") self.write(safe_json_dumps("Error - an internal server error occurred.")) self.set_status(500) self.set_header("Cache-Control", "no-store") self.set_header("Content-Type", "application/json") class APIAlertsStreamHandler(tornado_eventsource.handler.EventSourceHandler): """API request handler for /api/v2/alerts/stream""" def __init__(self, application, request, **kwargs: Any): self._sse_alert_broadcaster = None self._query_params = None self._credentials = None self._fields = None super().__init__(application, request, **kwargs) def initialize(self, sse_alert_broadcaster): self._sse_alert_broadcaster = sse_alert_broadcaster def custom_headers(self): """Custom headers to avoid e.g. nginx reverse proxy from buffering SSE data""" return {"Cache-Control": "no-store", "X-Accel-Buffering": "no"} def open(self): try: # request.arguments contains lists for each param key because technically the client can supply multiple, # reduce that to just the first entry, and convert bytes to string self._query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()} self._credentials = extract_credentials(self.request.headers) self._fields = ( [f.strip() for f in self._query_params["fields"].split(",")] if "fields" in self._query_params else [] ) # Flush headers immediately so nginx doesn't time out waiting for a response self.write_message("keepalive", "") # Register to handle new alerts arriving. The callback() method will get called with the new alert as an # argument. self._sse_alert_broadcaster.register(self) except Exception: logger.exception("Exception when serving SSE socket") self.close() def close(self): """When the user closes the socket, deregister ourselves from the alert broadcaster""" self._sse_alert_broadcaster.unregister(self) super().close() def callback(self, alert): """Callback when a new alert arrives""" try: # If the new alert matches our param filters, send it to the client. If not, ignore it. if alert_allowed_by_query(alert, self._query_params): # Add lookup data if we have credentials if self._credentials: alert = copy.deepcopy(alert) alert.infer_missing(self._credentials) # Filter fields returned if necessary if self._fields: alert = filter_fields([alert], self._fields)[0] self.write_message(msg=safe_json_dumps(alert)) except Exception: logger.exception("Exception in SSE callback, connection will be closed") self.close() def get_alert_list_with_filters(all_alerts, query): """Utility method to apply filters to the overall alert list and return only a subset. Enables query parameters in the main "alerts" GET call.""" # Create a shallow copy of the alert list ordered by start time, then filter the list to reduce it only to alerts # that match the filter parameters in the query string. Finally, apply a limit to the number of alerts returned. # The list of query string filters is defined in the API docs. alert_ids = all_alerts.keys() alerts = [] for k in alert_ids: a = all_alerts.get(k) if a is not None: alerts.append(a) alerts = sorted(alerts, key=lambda alert: alert.start_time if alert and alert.start_time else 0) alerts = list(filter(lambda alert: alert_allowed_by_query(alert, query), alerts)) if "limit" in query: alerts = alerts[: int(query.get("limit"))] return alerts def alert_allowed_by_query(alert, query): """Given URL query params and an alert, figure out if the alert "passes" the requested filters or is rejected. The list of query parameters and their function is defined in the API docs.""" for k in query: match k: case "received_since": since = datetime.fromtimestamp(float(query.get(k)), pytz.UTC) if not alert.received_time or alert.received_time <= since: return False case "max_duration": max_duration = int(query.get(k)) # Check the duration if end_time is provided. If end_time is not provided, assume the activation is # "short", i.e. it always passes this check. If dxpeditions_skip_max_duration_check is true and # the alert is a dxpedition, or contests_skip_max_duration_check and the alert is a contest, it also # always passes the check. if ( alert.alert_type == AlertType.DXPEDITION and "dxpeditions_skip_max_duration_check" in query and query.get("dxpeditions_skip_max_duration_check").upper() == "TRUE" ): continue if ( alert.alert_type == AlertType.CONTEST and "contests_skip_max_duration_check" in query and query.get("contests_skip_max_duration_check").upper() == "TRUE" ): continue if alert.end_time and alert.start_time and alert.end_time - alert.start_time > max_duration: return False case "source": sources = query.get(k).split(",") if not alert.source or alert.source not in sources: return False case "sig": # If a list of sigs is provided, the alert must have a sig and it must match one of them. # The special "sig" "NO_SIG", when supplied in the list, mathches alerts with no sig. sigs = query.get(k).split(",") include_no_sig = "NO_SIG" in sigs if not alert.sig and not include_no_sig: return False if alert.sig and alert.sig not in sigs: return False case "dx_continent": dxconts = query.get(k).split(",") if not alert.dx_continent or alert.dx_continent not in dxconts: return False case "dx_call_includes": dx_call_includes = query.get(k).strip() if not alert.dx_call or dx_call_includes.upper() not in alert.dx_call.upper(): return False case "text_includes": text_includes = query.get(k).strip() if ( (not alert.dx_call or text_includes.upper() not in alert.dx_call.upper()) and (not alert.comment or text_includes.upper() not in alert.comment.upper()) and (not alert.freqs_modes or text_includes.upper() not in alert.freqs_modes.upper()) ): return False return True def filter_fields(alerts, fields): """Given a list of alert objects, return copies containing only the named fields.""" return [{k: v for k, v in alert.__dict__.items() if k in fields} for alert in alerts]