`detect`

Detection rules for logs: describe what suspicious looks like as a map (or load a Sigma rule), run it over records from the logs library, and get alerts back. Thresholds ("5 failures within 60 seconds from one IP") and built-in hunts for brute force, port scans and rare values are included.

import detect

Example

import logs
import detect

let records = logs.read("apps/data/auth.log", "auth")

# A rule is just a map: what to match, how many, how fast, grouped by what.
let rule = detect.rule({
  "name": "Root login attempts",
  "level": "high",
  "where": {"action": "failed_login", "user": "root"},
  "count": 3,
  "within": 300,
  "by": "ip",
})
for alert in detect.run([rule], records) {
  say(f"{alert.level}: {alert.rule} from {alert.by} ({alert.count} tries)")
}

for alert in detect.brute_force(records) {
  say(f"{alert.rule}: {alert.by}, then logged in as {alert.success_user}")
}

Functions

rule(map)

returns map

Check a rule and fill in its defaults. A rule is a map: "name": what to call the alert; "level": informational/low/medium (default)/high/critical; "where": {field: value, ...} - every field must match. A value can be exact text (any case), text with * and ? wildcards, "re:pattern" for a regular expression, a number, nil (field missing), or a list (any of them). Add a modifier to the field name: "path|contains", "agent|startswith", "file|endswith", "bytes|gt", "ip|cidr" ("10.0.0.0/8"), "cmd|re", "x|exists". where can also be a function that gets a record and gives back true or false; "unless": {field: value} - skip records that match this (to remove false positives); "count": how many matches make an alert (default 1); "within": seconds (or "5m") they must fall in; "by": a field (or list of fields) to count separately, like "ip".

run(rules, records)

returns list

Run one rule or a list of rules over log records and give back the alerts, most serious first. Each alert is a map: {"rule", "level", "by" (the value of the by field, like the IP), "count", "first", "last" (timestamps), "first_when", "last_when" (as text), "description", "records" (the first few matching records)}. With count and within, each burst of activity gives one alert. A record's "repeat" field (from "message repeated 3 times") counts as that many records.

match(rule, record)

returns bool

True if one record matches a rule's where (or Sigma detection), ignoring count and within. Handy for testing rules.

brute_force(records, attempts?, within?)

returns list

Find password guessing: at least attempts failed logins (default 5) from one IP within within seconds (default 600). Works on auth.log (SSH, sudo) and Windows 4625/4771 events, or any record with success = false. Each alert also has "users" (the accounts tried) and "success_after": true if the same IP then logged in successfully - that alert becomes critical, because the attacker got in.

failed_logins(records)

returns list

Just the failed logins: records with success = false (SSH, sudo, su, Windows 4625/4771...). sshd's extra "Invalid user" and pam_unix lines are left out, so each guess counts once. brute_force() counts these.

port_scan(records, ports?, within?)

returns list

Find port scans in connection or firewall logs: one source IP touching at least ports different destination ports (default 10) within within seconds (default 60). Uses the fields src_ip / ip for the source and dst_port (or dest_port, DestinationPort, dpt) for the destination; UFW and iptables lines from the logs library have them. Alerts list the "ports" and "targets" seen.

rare(records, field, max_count?)

returns list

Values of a field that show up at most max_count times (default 1), like a user agent or process seen only once: rare things are worth a look. Gives back low-level alerts, rarest first.

sigma(text_or_path)

returns map

Load a Sigma rule (the shared YAML format for detections) from a .yml file or from text, as a rule for detect.run(). Supported: title, id, level, description, tags, logsource, timeframe, and detection with named selections (maps, lists of maps, keyword lists), the modifiers contains, startswith, endswith, all, re, cidr, gt/gte/lt/lte, exists and windash, wildcards * and ?, and conditions with and, or, not, brackets, "1 of selection*", "all of them" and "| count() by field > N". logsource limits the rule to matching record formats (product: windows -> Windows events). Anything else gives a clear error.

sigma_folder(path)

returns list

Load every .yml / .yaml Sigma rule in a folder (and its sub-folders), sorted by file name.

parse_yaml(text)

returns value

Read the simple YAML that Sigma rules use (maps, lists, quoted and plain values, | and > text blocks) into TB maps and lists.

Values

Name	Value	What it is
`detect.levels`	`["informational", "low", "medium", "high", "critical"]`	Alert levels from least to most serious: informational, low, medium, high, critical.