heavy hitters analysis, moving avg + CMS

Add a heavy hitters analysis plugin that identifies heavy hitters
by comparing frequency per identifier in a sample window against
a calculated average for the window. The frequency counts are
stored in a count-min sketch data structure.

Author: Aaron Meihm

CMakeLists.txt

@@ -68,6 +68,7 @@ aws
 bloom_filter
 circular_buffer
 cjson
+cms
 compat
 cuckoo_filter
 elasticsearch
@@ -95,6 +96,7 @@ ssl
 struct
 syslog
 systemd
+xxhash
 zlib
 )
 

cms/CMakeLists.txt

@@ -0,0 +1,8 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+cmake_minimum_required(VERSION 3.0)
+project(cms VERSION 0.0.1 LANGUAGES C)
+set(CPACK_PACKAGE_DESCRIPTION_SUMMARY "Count-Min Sketch implementation")
+include(sandbox_module)

cms/index.md

@@ -0,0 +1,4 @@
+# Lua Count-Min Sketch implementation
+
+* http://theory.stanford.edu/~tim/s15/l/l2.pdf
+* https://github.com/mikalsande/lua-count-min

cms/modules/cms/cms.lua

@@ -0,0 +1,196 @@
+--[[
+# Count-min Sketch implementation in Lua
+
+Provides a probabilistic data structure to inform frequency calculation in an
+event stream.
+
+## Functions
+
+### new
+
+Create and return a new CMS data structure. Typical operations on this value will
+involve calling the add and check functions.
+
+*Arguments*
+- epsilon (number) - CMS epsilon parameter
+- delta (number) - CMS delta parameter
+
+*Return*
+- Returns initialized CMS structure
+--]]
+
+local math      = require("math")
+local xxhash    = require("xxhash")
+local hash      = xxhash.xxh32
+local ostime    = require("os").time
+
+local type          = type
+local assert        = assert
+local setmetatable  = setmetatable
+
+local M = {}
+setfenv(1, M)
+
+math.randomseed(ostime())
+
+function new(epsilon, delta)
+  assert(type(epsilon) == "number", "epsilon must be a number")
+  assert(epsilon > 0, "epsilon must be bigger than zero")
+  assert(type(delta) == "number", "delta must be a number")
+  assert(delta> 0, "delta must be bigger than zero")
+
+  -- Count Min Sketch variables
+  local w = math.ceil(2.718281828459045 / epsilon)
+  local d = math.ceil(math.log(1 / delta))
+  local l = d * w
+
+  -- main datastructure
+  local array = {}
+
+  -- hash variables
+  local seed1
+  local seed2
+
+  -- counter
+  local addedItems
+  local maxval = 2^53 - 1
+
+  -- temporary index table
+  local temp_index = {}
+  for x = 1, d do
+    temp_index[x] = 0
+  end
+
+  local query = function(input, update)
+    assert(type(input) == "string", 'input must be a string')
+    assert(type(update) == "boolean", 'update must be a boolean')
+
+    -- make two different hash values from the input with different seeds
+    local h1 = hash(input, seed1)
+    local h2 = hash(input, seed2)
+
+    -- track whether or not we have added a new value,  0 if added 1 if not added
+    local existed = 1
+
+    -- create hashes and compute array index and bit index for them to get
+    -- the individual bits
+    local min = maxval
+    for i = 1, d do
+      -- set offset, we are using one big array instead of many smaller ones
+      local offset = (i - 1) * w
+
+      -- use enhanced double hashing (Kirsh & Mitzenmacher) to create the hash
+      -- value used for the array index. +1 to account for Lua arrays
+      local x = ((h1 + (i * h2) + i^2) % w) + 1
+      temp_index[i] = x
+      local num = array[x + offset]
+
+      -- if the bit is zero we know it did not exist already just return if we are
+      -- not updating it
+      if num == 0 then
+        existed = 0
+        if not update then
+          return 0
+        end
+      end
+
+      -- update the smallest seen value
+      if num < min then
+        min = num
+      end
+    end
+
+    -- update the item counter if we added a new item
+    if update then
+      -- conservative update, only update values that smaller than min + 1
+      for i = 1, d do
+        local offset = (i - 1) * w
+        local index = temp_index[i] + offset
+        if array[index] < (min + 1) and min < maxval then
+          array[index] = array[index] + 1
+        end
+      end
+
+      -- update min in order to return the updated count
+      min = min + 1
+
+      -- update the items counter if we addded a new item
+      if existed == 0 then
+        addedItems = addedItems + 1
+      end
+    end
+
+    return min
+  end
+
+  --[[
+  Adds a new entry to the set. returns 1 if the element
+  already existed and 0 if it does not exist in the set.
+  ]]
+  local add = function(input)
+    return query(input, true)
+  end
+
+  --[[
+  checks whether or not an element is in the set.
+  returns 1 if the element exists and 0 if it does
+  not exist in the set.
+  ]]
+  local check = function(input)
+    return query(input, false)
+  end
+
+  --[[ resets the arrays used in the filter ]]
+  local reset = function()
+    addedItems = 0
+
+    -- since 2*rand and 3*rand is done modulo a prime these number will always
+    -- be different and should therefore safe to use as seeds to generate two
+    -- different hashes for the same input.
+    local rand = math.random(-2147483648, 2147483647)
+    seed1 = (2*rand) % 2147483647
+    seed2 = (3*rand) % 2147483647
+
+    for x = 1, l do
+      array[x] = 0
+    end
+  end
+
+  -- [[ get number of unique added items ]]
+  local getNumItems = function()
+    return addedItems
+  end
+
+  -- [[ get number of bits this instance uses ]]
+  local getDepth = function()
+    return d
+  end
+
+  -- [[ get number of keys this instance uses ]]
+  local getWidth = function()
+    return w
+  end
+
+  -- [[ get number of keys this instance uses ]]
+  local getAccumulatedError = function()
+    return addedItems * epsilon
+  end
+
+  -- reset the array to initialize this instance
+  reset()
+
+  -- methods we expose for this instance
+  local ret = {
+    add = add,
+    check = check,
+    reset = reset,
+    getNumItems = getNumItems,
+    getDepth = getDepth,
+    getWidth = getWidth,
+    getAccumulatedError = getAccumulatedError
+  }
+  setmetatable(ret, {})
+  return ret
+end
+
+return M

moz_security/sandboxes/heka/analysis/moz_security_hh_cms.lua

@@ -0,0 +1,154 @@
+-- This Source Code Form is subject to the terms of the Mozilla Public
+-- License, v. 2.0. If a copy of the MPL was not distributed with this
+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+--[[
+#  Mozilla Security Heavy Hitters, CMS + moving average
+
+For events matching the message_matcher, this analysis plugin attempts to calculate typical
+request rates while identifying and flagging anomolous request patterns.
+
+Request counts within a given window are stored in a Count-Min sketch data structure; based on
+a sample encountered during the window this data structure is used to consult request rates for
+given identifiers, where an identifier exceeds a threshold it is added to an analysis list for
+submission by the plugin.
+
+## Sample Configuration
+```lua
+filename = "moz_security_hh_cms.lua"
+message_matcher = "Logger == 'input.nginx'"
+ticker_interval = 60
+preserve_data = false -- This module cannot keep state at this time
+
+id_field = "Fields[remote_addr]" -- field to use as the identifier
+-- id_field_capture = ",? *([^,]+)$",  -- optional e.g. extract the last entry in a comma delimited list
+
+sample_min_id = 5000 -- Minimum distinct identifers before a sample will be calculated
+sample_max_id = 10000 -- Maximum identifiers to use for sample calculation within a window
+sample_min_ev = 50000 -- Minimum number of events sampler must consume before calculation
+sample_window = 60 -- Sample window size
+sample_ticks = 2500 -- Recalculate sample every sample_ticks events
+threshold_cap = 10 -- Threshold will be calculated average + (calculated average * cap)
+-- cms_epsilon = 1 / 10000 -- optional CMS value for epsilon
+-- cms_delta = 0.0001 -- optional CMS value for delta
+```
+--]]
+
+require "string"
+require "table"
+
+local c         = require "cms.cms"
+local ostime    = require "os".time
+
+local sample_min_id = read_config("sample_min_id") or error("sample_min_id must be configured")
+local sample_max_id = read_config("sample_max_id") or error("sample_max_id must be configured")
+local sample_min_ev = read_config("sample_min_ev") or error("sample_min_ev must be configured")
+local sample_window = read_config("sample_window") or error("sample_window must be configured")
+local sample_ticks  = read_config("sample_ticks") or error ("sample_ticks must be configured")
+local threshold_cap = read_config("threshold_cap") or error("threshold_cap must be configured")
+local id_field      = read_config("id_field") or error("id_field must be configured")
+local id_fieldc     = read_config("id_field_capture")
+local cms_epsilon   = read_config("cms_epsilon") or 1 / 10000
+local cms_delta     = read_config("cms_delta") or 0.0001
+
+local cms = c.new(cms_epsilon, cms_delta)
+
+local alist = {}
+
+function alist:reset()
+    self.l = {}
+end
+
+function alist:add(i, c)
+    if not i or not c then
+        error("analysis list received nil argument")
+    end
+    self.l[i] = c
+end
+
+function alist:flush(t)
+    for k,v in pairs(self.l) do
+        if v < t then self.l[k] = nil end
+    end
+end
+
+local sampler = {}
+
+function sampler:reset()
+    self.s = {}
+    self.n = 0
+    self.start_time = ostime()
+    self.threshold = 0
+    self.validtick = 0
+    self.evcount = 0
+end
+
+function sampler:calc()
+    if not cms or self.n < sample_min_id or self.evcount < sample_min_ev then
+        return
+    end
+    if self.validtick < sample_ticks then
+        return
+    end
+    self.validtick = 0
+    local cnt = 0
+    local t = 0
+    for k,v in pairs(self.s) do
+        t = t + cms.check(k)
+        cnt = cnt + 1
+    end
+    self.threshold = t / cnt
+    self.threshold = self.threshold + (self.threshold * threshold_cap)
+    -- Remove any elements in the analysis list that no longer conform
+    -- to the set threshold
+    alist:flush(self.threshold)
+end
+
+function sampler:add(x)
+    if self.start_time + sample_window < ostime() then
+        self:reset()
+        alist:reset()
+        cms = c.new(cms_epsilon, cms_delta)
+    end
+    self.evcount = self.evcount + 1
+    self.validtick = self.validtick + 1
+    if self.n >= sample_max_id then
+        return
+    end
+    -- If x is already present in the sample, don't add it again
+    if self.s[x] then return end
+    self.s[x] = 1
+    self.n = self.n + 1
+end
+
+sampler:reset()
+alist:reset()
+
+function process_message()
+    local id = read_message(id_field)
+    if not id then return -1, "no id_field" end
+    if id_fieldc then
+        id = string.match(id, id_fieldc)
+        if not id then return 1 end -- no error as the capture may intentionally reject entries
+    end
+
+    sampler:add(id)
+    sampler:calc()
+    local q = cms.add(id)
+    if sampler.threshold ~= 0 and q > sampler.threshold then
+        alist:add(id, q)
+    end
+    return 0
+end
+
+function timer_event(ns)
+    -- For now, just generate a tsv here but this could be modified to submit violations
+    -- with a configured confidence to Tigerblood
+    add_to_payload("sampler_threshold", "\t", sampler.threshold, "\n")
+    add_to_payload("sampler_size", "\t", sampler.n, "\n")
+    add_to_payload("sampler_evcount", "\t", sampler.evcount, "\n")
+    for k,v in pairs(alist.l) do
+        add_to_payload(k, "\t", v, "\n")
+    end
+    inject_payload("tsv", "statistics")
+end

moz_security/tests/hindsight/run/analysis/hh_cms.cfg

@@ -0,0 +1,12 @@
+filename = "moz_security_hh_cms.lua"
+message_matcher = "Logger == 'input.hh_cms'"
+ticker_interval = 0
+
+id_field = "Fields[id]"
+
+sample_min_id = 100
+sample_max_id = 10000
+sample_min_ev = 0
+sample_window = 60
+sample_ticks = 1000
+threshold_cap = 2

moz_security/tests/hindsight/run/input/generate_hh_cms.cfg

@@ -0,0 +1 @@
+filename = "generate_hh_cms.lua"