Conditional parser

[aphilas]

TLDR: How can I conditionally parse logs or event fields with regex, to assign "codes" to events (i.e. map certain strings to specific integers).

Background

I am creating a SIEM tool loosely based on DSIEM and OSSIM. Coming up with a basic set of plugins (ingest) and directives (triggers for alarms) is non-trivial thus I have decided to adopt the base OSSIM plugins and directives.

Events are ingested by Fluentbit (in this case, say logs), and then a "correlation engine" raises an alarm if say 10_000 requests hit the Apache server in 1s from the same IP address. This an example of a directive. The directives reference plugin_ids and plugin_sids to work. For example, we may christen a Fluentbit config/parser as plugin 1 for Apache. Then, name the plugin_sids based on the HTTP code. Thus log lines requests with status code 200 end up with "200" as the plugin_sid which is used in the directives. Here are the directives docs for DSIEM.

Problem

How can I conditionally add keys to events based on regex?

In pseudocode:

map = {
  'new user'=>1
  'add'=>2
  'failed adding user'=>3
  'new group'=>10
  'group added'=>11
  _=>20000000
}

plugin_sid=map[$message]

Example

Here is an OSSIM plugin that reads /var/log/auth.log. Each plugin has one or more rules. Each rule (from L48) has a regexp which is matched against the log line. In this case, events which contain "failed adding user" are tagged with plugin_sid 3. If the rule regex does not match, the next regex is evaluated. Here are the docs on OSSIM/USM plugins:

• Choose a pre-check string. The plugin uses this string to search the log line before applying the rule. This acts as a filter to avoid applying the rule to a line that cannot match, improving plugin performance. For an example, see the example following step 3 in Add a New Rule to a Plugin. The fourth statement from the top shows Pre-check="Teardown", followed by the regex.
• Order the rules starting with 0001 and finishing with 9999. Create groups numbered 0002, 0003, and so forth, leaving room for future expressions.
• Your rules based on specific matching criteria should always be the lowest numbers and the more generic rules, the highest. This helps avoid event masking, which can occur when USM Appliance loads generic rules before specific rules.
• When a log line matches a rule, USM Appliance generates an event and does not match it to any other the rule in the queue.

Solutions I have considered

1. Use filters.modify/filters.grep - not expressive enough

2. Use filter.parser with parser.regex for a specific field

• + use any type of input (http, tail, unix socket)
• - fields might not match 1-to-1 with OSSIM
• ? how to map named regex match to plugin_sid?

3. Create custom input parsers with parser.regex

• + closer to ossim plugins
• - throws out all the work done on existing parsers/inputs
• ? how to map named regex match to plugin_sid?

4. Use filters.lua (with existing inputs) to parse event fields

• + expressive - replicate OSSIM "translation tables"
• - complexity - need to learn (basic) lua, complex maintenance
• - lua does not support POSIX regex internally. Pattern matching, the alternative, is pretty limited.

5. Stream processors - a brief check on the docs suggests these are for "aggregations"

ETA: on lua regex support

[agup006]

Note you can add custom lua libraries for regular expressions though you need to build the package manually to consume

[aphilas]

Thanks, this sounds like a feasible approach. I'll give it a try.