ntopng/scripts/lua/modules/notifications/endpoints/flowtriq.lua

--
-- (C) 2024 - ntop.org
--
-- Flowtriq notification endpoint
-- Sends DDoS-related alerts to Flowtriq's inbound webhook API.
--

require "lua_utils"
local json = require "dkjson"
local alert_utils = require "alert_utils"

local endpoint_key = "flowtriq"

local flowtriq = {
   name = "Flowtriq",

   -- (1) Endpoint
   endpoint_params = {
      { param_name = "flowtriq_url" },
      { param_name = "flowtriq_api_key" },
   },
   endpoint_template = {
      script_key = endpoint_key,
      template_name = "flowtriq_endpoint.template"
   },

   -- (2) Recipient
   recipient_params = {
   },
   recipient_template = {
      script_key = endpoint_key,
      template_name = "flowtriq_recipient.template"
   },
}

flowtriq.EXPORT_FREQUENCY = 5
flowtriq.prio = 400
flowtriq.REQUEST_TIMEOUT = 10
flowtriq.ITERATION_TIMEOUT = 3
local MAX_ALERTS_PER_REQUEST = 10

-- ##############################################

-- @brief Returns the desired formatted output for recipient params
function flowtriq.format_recipient_params(recipient_params)
   return string.format("(%s)", flowtriq.name)
end

-- ##############################################

local function readSettings(recipient)
   local settings = {
      url = recipient.endpoint_conf.flowtriq_url,
      api_key = recipient.endpoint_conf.flowtriq_api_key,
   }

   return settings
end

-- ##############################################

local function formatAlertPayload(alert)
   local decoded_alert = json.decode(alert)

   if decoded_alert and decoded_alert.json then
      local json_decoded = json.decode(decoded_alert.json)

      if json_decoded and json_decoded.flow_risk_info and type(json_decoded.flow_risk_info) == "string" then
         json_decoded.flow_risk_info = json.decode(json_decoded.flow_risk_info)
      end
      if json_decoded and json_decoded.alert_generation and json_decoded.alert_generation.flow_risk_info and type(json_decoded.alert_generation.flow_risk_info) == "string" then
         json_decoded.alert_generation.flow_risk_info = json.decode(json_decoded.alert_generation.flow_risk_info)
      end

      decoded_alert.json = json_decoded
      decoded_alert.metadata = {}
   end

   return decoded_alert
end

-- ##############################################

function flowtriq.sendMessage(alerts, settings)
   if isEmptyString(settings.url) or isEmptyString(settings.api_key) then
      return false
   end

   local message = {
      source = "ntopng",
      timestamp = os.time(),
      alerts = alerts,
   }

   local json_message = json.encode(message)

   local rc = false
   local retry_attempts = 3
   while retry_attempts > 0 do
      -- Use the bearer token (6th param) for API key authentication
      if ntop.postHTTPJsonData("", "", settings.url, json_message, flowtriq.REQUEST_TIMEOUT, settings.api_key) then
         rc = true
         break
      end
      retry_attempts = retry_attempts - 1
   end

   return rc
end

-- ##############################################

function flowtriq.dequeueRecipientAlerts(recipient, budget)
   local start_time = os.time()
   local sent = 0
   local budget_used = 0
   local settings = readSettings(recipient)

   local more_available = true
   local success = true
   local error_message = nil
   local delivered = 0
   local discarded = 0
   local failures = 0

   -- Dequeue alerts up to budget x MAX_ALERTS_PER_REQUEST
   while budget_used <= budget and more_available do
      local diff = os.time() - start_time
      if diff >= flowtriq.ITERATION_TIMEOUT then
         break
      end

      -- Dequeue MAX_ALERTS_PER_REQUEST notifications
      local notifications = {}
      local i = 0
      while i < MAX_ALERTS_PER_REQUEST do
         local notification = ntop.recipient_dequeue(recipient.recipient_id)
         if notification then
            if alert_utils.filter_notification(notification, recipient.recipient_id) then
               notifications[#notifications + 1] = notification.alert
               i = i + 1
            else
               discarded = discarded + 1
            end
         else
            break
         end
      end

      if not notifications or #notifications == 0 then
         more_available = false
         break
      end

      local alerts = {}

      for _, json_message in ipairs(notifications) do
         local alert = formatAlertPayload(json_message)
         table.insert(alerts, alert)
      end

      if not flowtriq.sendMessage(alerts, settings) then
         success = false
         error_message = "Unable to send alerts to Flowtriq"
         failures = failures + #notifications
         goto done
      else
         delivered = delivered + #notifications
      end

      -- Remove the processed messages from the queue
      budget_used = budget_used + #notifications
      sent = sent + 1
   end

  ::done::
   return {
      success = success,
      error_message = error_message,
      delivered = delivered,
      discarded = discarded,
      failures  = failures,
      more_available = more_available,
   }
end

-- ##############################################

function flowtriq.runTest(recipient)
   local message_info

   local settings = readSettings(recipient)

   local success = flowtriq.sendMessage({}, settings)

   if success then
      message_info = i18n("notification_endpoint.flowtriq.flowtriq_sent_successfully")
   else
      message_info = i18n("notification_endpoint.flowtriq.flowtriq_send_error")
   end

   return success, message_info
end

-- ##############################################

return flowtriq