add metrics ability to the lua bouncer

lib/plugins/crowdsec/config.lua

@@ -64,7 +64,6 @@ function config.loadConfig(file)
         ['APPSEC_FAILURE_ACTION'] = "passthrough",
         ['SSL_VERIFY'] = "true",
         ['ALWAYS_SEND_TO_APPSEC'] = "false",
-
     }
     for line in io.lines(file) do
         local isOk = false

lib/plugins/crowdsec/iputils.lua

@@ -212,4 +212,4 @@ end
 local band = bit.band
 local sar = bit.arshift
 
-return _M
+return _M

lib/plugins/crowdsec/live.lua

@@ -0,0 +1,91 @@
+local cjson = require "cjson"
+local utils = require "plugins.crowdsec.utils"
+
+local live = {}
+live.__index = live
+
+live.cache = ngx.shared.crowdsec_cache
+
+--- Create a new live object
+-- Create a new live object to query the live API
+-- @return live: the live object
+
+function live:new()
+  return self
+end
+
+--- Live query the API to get the decision for the IP
+-- Query the live API to get the decision for the IP in real time
+-- @param ip string: the IP to query
+-- @param api_url string: the URL of the LAPI
+-- @param timeout number: the timeout of the request to lapi
+-- @param cache_expiration number: the expiration time of the cache
+-- @param api_key_header string: the authorization header to use for the lapi request
+-- @param api_key string: the API key to use for the lapi request
+-- @param user_agent string: the user agent to use for the lapi request
+-- @param ssl_verify boolean: whether to verify the SSL certificate or not
+-- @param bouncing_on_type string: the type of decision to bounce on
+-- @return boolean: true if the IP is allowed, false if the IP is blocked
+-- @return string: the type of the decision
+-- @return string: the origin of the decision
+-- @return string: the error message if any
+
+function live:live_query(ip, api_url, timeout, cache_expiration, api_key_header, api_key, user_agent, ssl_verify, bouncing_on_type)
+
+  local link = api_url .. "/v1/decisions?ip=" .. ip
+  -- function M.get_remediation_http_request(link,timeout, api_key_header, api_key, user_agent,ssl_verify)
+
+  local res, err = utils.get_remediation_http_request(link, timeout, api_key_header, api_key, user_agent, ssl_verify)
+  if not res then
+    return true, nil, nil, "request failed: ".. err
+  end
+  -- debug: wip
+  ngx.log(ngx.INFO, "request" .. res.body)
+  local status = res.status
+  local body = res.body
+  if status~=200 then
+    return true, nil, nil, "Http error " .. status .. " while talking to LAPI (" .. link .. ")"
+  end
+  if body == "null" then -- no result from API, no decision for this IP
+    -- set ip in cache and DON'T block it
+    local key,_ = utils.item_to_string(ip, "ip")
+    local succ, err, forcible = live.cache:set(key, "none", cache_expiration, 1)
+    --
+    ngx.log(ngx.INFO, "Adding '" .. key .. "' in cache for '" .. cache_expiration .. "' seconds") --debug
+    if not succ then
+      ngx.log(ngx.ERR, "failed to add ip '" .. ip .. "' in cache: ".. err)
+    end
+    if forcible then
+      ngx.log(ngx.ERR, "Lua shared dict (crowdsec cache) is full, please increase dict size in config")
+    end
+    return true, nil, nil, nil
+  end
+  local decision = cjson.decode(body)[1]
+
+  -- debug: wip
+  ngx.log(ngx.INFO, "Decision: " .. decision.type .. " | " .. decision.value .. " | " .. decision.origin .. " | " .. decision.duration)
+  ngx.log(ngx.INFO, "Bouncing on type: " .. bouncing_on_type)
+  if bouncing_on_type == decision.type or bouncing_on_type == "all" then
+    if decision.origin == "lists" and decision.scenario ~= nil then
+      decision.origin = "lists:" .. decision.scenario
+    end
+    local cache_value = decision.type .. "/" .. decision.origin
+    local key,_ = utils.item_to_string(decision.value, decision.scope)
+    local succ, err, forcible = live.cache:set(key, cache_value, cache_expiration, 0)
+    ngx.log(ngx.INFO, "Adding '" .. key .. "' in cache for '" .. cache_expiration .. "' seconds with decision type'" .. decision.type .. "'with origin'" .. decision.origin ) --debug
+    if not succ then
+      ngx.log(ngx.ERR, "failed to add ".. decision.value .." : "..err)
+    end
+    if forcible then
+      ngx.log(ngx.ERR, "Lua shared dict (crowdsec cache) is full, please increase dict size in config")
+    end
+    -- debug: wip
+    ngx.log(ngx.DEBUG, "Adding '" .. key .. "' in cache for '" .. cache_expiration .. "' seconds")
+    ngx.log(ngx.INFO, "Adding '" .. key .. "' in cache for '" .. cache_expiration .. "' seconds")
+    return false, decision.type, decision.origin, nil
+  else
+    return true, nil, nil, nil
+  end
+end
+
+return live

lib/plugins/crowdsec/metrics.lua

@@ -0,0 +1,197 @@
+local cjson = require "cjson"
+local http = require "resty.http"
+local utils = require "plugins.crowdsec.utils"
+local osinfo = require "plugins.crowdsec.osinfo"
+local metrics = {}
+
+metrics.__index = metrics
+metrics.cache = ngx.shared.crowdsec_cache
+
+
+-- Constructor for the store
+function metrics:new(userAgent)
+  local info = osinfo.get_os_info()
+  self.cache:set("metrics_data", cjson.encode({
+    version = userAgent,
+    os = {
+      name = info["NAME"];
+      version = info["VERSION_ID"];
+    },
+    type="lua-bouncer",
+    name="nginx bouncer",
+    utc_startup_timestamp = ngx.time(),
+  }))
+end
+
+
+-- Increment the value of a key or initialize it if it does not exist
+-- @param key: the key to increment
+-- @param increment: the value to increment the key by
+-- @param labels: a table of labels to add to the key
+-- @return the new value of the key
+function metrics:increment(key, increment, labels)
+    increment = increment or 1
+    if labels == nil then
+      ngx.log(ngx.INFO, "no labels")
+    end
+
+    -- keys could look like:
+    -- processed/ip_version=ipv4&
+    -- active_decisions/ip_version=ipv4&decision_type=ban&
+    key = key .. "/" .. utils.table_to_string(labels)
+    local value, err, forcible = self.cache:incr("metrics_" .. key, increment, 0)
+    metrics:add_to_metrics(key)
+    if err then
+        ngx.log(ngx.ERR, "failed to increment key: ", err)
+    end
+    if forcible then
+      ngx.log(ngx.ERR, "Lua shared dict (crowdsec cache) is full, please increase dict size in config")
+    end
+    return value
+end
+
+-- Get all metrics as a table (key-value pairs)
+function metrics:get_all_keys()
+    local keys = metrics.cache:get("metrics_all")
+    return utils.split_on_delimiter(keys, ",")
+end
+
+-- Add a metric key to the `metrics_all` list
+function metrics:add_to_metrics(key)
+    local metrics_all = self.cache:get("metrics_all") or ""
+    if not metrics_all:find(key) then
+        metrics_all = metrics_all .. key .. ","
+        local success, err, forcible = self.cache:set("metrics_all", metrics_all)
+        if not success then
+          ngx.log(ngx.ERR, "failed to set key metrics_all: ", err)
+        end
+        if forcible then
+          ngx.log(ngx.ERR, "Lua shared dict (crowdsec cache) is full, please increase dict size in config")
+        end
+
+    end
+end
+
+--- Get the labels from a cache key
+--- As labels are stored in the cache key
+--- we need to extract them from the key
+--- @param key string key: the cache key to extract the labels from
+--- @return string the key without the labels and the labels as a table
+--- @return table labels as a table
+local function get_labels_from_key(key)
+  local table = utils.split_on_delimiter(key, "/")
+  local labels = {}
+  if table == nil then
+    return "", {}
+  else
+    if table[2] ~= nil then
+      labels = utils.string_to_table(table[2])
+    end
+  end
+  return table[1], labels
+end
+
+-- Export the store data to JSON
+function metrics:toJson(window)
+  local metrics_array = {}
+  local metrics_data = self.cache:get("metrics_data")
+  local keys = metrics:get_all_keys()
+  for  _,key in ipairs(keys) do
+    local cache_key = "metrics_" .. key
+    local value = self.cache:get(cache_key)
+    ngx.log(ngx.INFO, "cache_key: " .. cache_key .. " value: " .. tostring(self.cache:get(cache_key)))--debug
+    if value ~= nil then
+      ngx.log(ngx.INFO, "key: " .. key)
+      local final_key, labels = get_labels_from_key(key)
+      ngx.log(ngx.INFO, "final_key: " .. final_key)
+      ngx.log(ngx.INFO, "value: " .. value)
+      if labels ~= nil then
+        for k, v in pairs(labels) do
+          ngx.log(ngx.INFO, "label: " .. k .. " " .. v)
+        end
+      end
+
+      if final_key == "processed" then
+        table.insert(metrics_array, {
+                       name = "processed",
+                       value = value,
+                       unit = "request",
+                       labels = labels
+        })
+      elseif final_key == "active_decisions" then
+        table.insert(metrics_array, {
+                       name = final_key,
+                       value = value,
+                       unit = "ip",
+                       labels = labels
+        })
+      else
+        table.insert(metrics_array, {
+                       name = final_key,
+                       value = value,
+                       unit = "request",
+                       labels = labels
+        })
+
+      end
+
+      if final_key ~= "active_decisions" and final_key ~= "processed" then
+        local success, err = self.cache:delete(cache_key)
+        if success then
+          ngx.log(ngx.INFO, "Cache key '", cache_key, "' deleted successfully")
+        else
+          ngx.log(ngx.INFO, "Failed to delete cache key '", cache_key, "': ", err)
+        end
+      else
+        if final_key == "processed" then
+          self.cache:set(cache_key, 0)
+        end
+      end
+    end
+  end
+  --setmetatable(metrics_data, cjson.array_mt)
+    -- for k, v in pairs(metrics_data) do
+    --   remediation_components[k] = v
+    -- end
+    --
+
+  local remediation_components = {}
+  local remediation_component = cjson.decode(metrics_data)
+  remediation_component["feature_flags"] = setmetatable({}, cjson.array_mt)
+  remediation_component["metrics"]= {
+    {
+      items = metrics_array,
+      meta = {
+        utc_now_timestamp = ngx.time(),
+        window_size_seconds = window
+      }
+    }
+  }
+  table.insert(remediation_components, remediation_component)
+  return cjson.encode({log_processors = cjson.null, remediation_components = remediation_components})
+end
+
+function metrics:sendMetrics(link, headers, ssl, window)
+  local body = self:toJson(window) .. "\n"
+  ngx.log(ngx.DEBUG, "Sending metrics to " .. link .. "/v1/usage-metrics")
+  ngx.log(ngx.DEBUG, "metrics: " .. body)
+  local httpc = http.new()
+  local res, err = httpc:request_uri(link .. "/v1/usage-metrics", {
+    body = body,
+    method = "POST",
+    headers = headers,
+    ssl_verify = ssl
+  })
+  httpc:close()
+  if not res then
+    ngx.log(ngx.ERR, "failed to send metrics: ", err)
+  else
+    ngx.log(ngx.INFO, "metrics sent: " .. res.status)
+    ngx.log(ngx.INFO, "metrics response: " .. body)
+  end
+
+end
+
+-- Function to retrieve all keys that start with a given prefix
+
+return metrics

lib/plugins/crowdsec/osinfo.lua

@@ -0,0 +1,67 @@
+local osinfo = {}
+
+
+local function parse_os_release()
+	local os_release = {}
+	local f = io.open("/etc/os-release", "r")
+	if f then
+		for line in f:lines() do
+			local key, value = line:match("([^=]+)=(.*)")
+			if key and value then
+				os_release[key] = value
+			end
+		end
+		f:close()
+	end
+	return os_release
+end
+
+local function parse_lsb_release()
+	local lsb_release = {}
+	local f = io.open("/etc/lsb-release", "r")
+	if f then
+		for line in f:lines() do
+			local key, value = line:match("([^=]+)=(.*)")
+			if key and value then
+				lsb_release[key] = value
+			end
+		end
+		f:close()
+	end
+	return lsb_release
+end
+
+
+function osinfo.get_os_info()
+
+	local os_info = parse_os_release()
+	if not os_info["NAME"] then
+		os_info = parse_lsb_release()
+	end
+
+	if not os_info["NAME"] then
+		os_info["NAME"] = "Unknown"
+	end
+
+	if not os_info["VERSION_ID"] then
+		os_info["VERSION_ID"] = "Unknown"
+	end
+
+	-- remove quotes
+	-- some distros have quotes around the values
+	-- e.g. "Ubuntu"
+	-- this is not the case for all distros
+	-- so we need to check if the first character is a quote
+	-- and remove it if it is
+	for k, _ in pairs(os_info) do
+		if string.sub(os_info[k], 1, 1) == '"' then
+			os_info[k] = string.sub(os_info[k], 2, -1)
+		end
+		if string.sub(os_info[k], -1) == '"' then
+			os_info[k] = string.sub(os_info[k], 1, -2)
+		end
+	end
+	return os_info
+end
+
+return osinfo