rapid7-insightidr 1.12.0: ingest LEQL query results as alerts
A tenant whose basic detection rules only notify — email, Slack — creates no
investigation and no alert any API can read: /investigations stays empty and the
Alert Triage API is MDR-gated. The rule's LEQL query is the only artefact left,
so replay it on a schedule and turn every matching line into an alert.
query_log and query_log_set gain:
- an ingest block (results_path events, dedup_key event_id, watermark start_time);
- event_id = log_id + the line's sequence number, the line's own address in Log
Search. It has to be stable across runs because the scheduler deliberately
re-reads the last minute of the previous window on every tick, and anything
derived from the fetch time would make that overlap a second incident. A hash
of (log_id, timestamp, message) covers a source that does not number entries;
- start_time read as ISO-8601, UNIX seconds or milliseconds alike. The API only
takes milliseconds, but which one the scheduler sends is an operator setting on
the rule — the feed must not fetch the wrong window because of a dropdown;
- a bounded page walk (limit, default 1000): unattended every few minutes, a
query matching more than intended would otherwise hold the run open long
enough for the next tick to find it in flight and skip itself;
- detection_name, stamped on every event. The log line does not know which rule
it is being replayed for, and without it an incident is titled by its own first
characters where the source console shows an alert name.
Plus mappers/query_log{,_set}.yaml: the parsed event lives in `message` as a
JSON-encoded string, so account, addresses, asset and process come out through
$parseJson. A raw unparsed line is not JSON — those entries resolve to nothing
and are skipped, and custom:rapid7_log_line still carries the line verbatim.
Severity is left unmapped: a log line has no priority of its own, so the alert
rule's default applies.
Known limits, by construction: this endpoint rejects calculations, so a rule with
a threshold (N occurrences in M minutes) cannot be reproduced — only pattern
rules are 1:1 — and an inactivity rule never can, since it fires on the absence
of events. The query also becomes a copy: editing the rule in InsightIDR leaves
the feed reading the old one.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UjtRtCzw6FEajDDHJ3mTQf
This commit is contained in:
@@ -1,7 +1,11 @@
|
||||
import json, os, sys, time, urllib.request, urllib.parse, urllib.error
|
||||
import datetime, hashlib, json, os, sys, time, urllib.request, urllib.parse, urllib.error
|
||||
|
||||
_UNITS = {"second": 1, "minute": 60, "hour": 3600, "day": 86400, "week": 604800, "month": 2592000, "year": 31536000}
|
||||
|
||||
# Beyond this a single value can only be milliseconds: 1e11 seconds is the year 5138,
|
||||
# and 1e11 milliseconds was 1973. Nothing a feed can legitimately ask for sits between.
|
||||
_MS_FLOOR = 100000000000
|
||||
|
||||
|
||||
def _cfg():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
@@ -30,6 +34,27 @@ def _rel_ms_range(text):
|
||||
return now_ms - num * _UNITS[unit] * 1000, now_ms
|
||||
|
||||
|
||||
def _to_ms(v):
|
||||
"""Epoch milliseconds, whatever the caller wrote.
|
||||
|
||||
As an ingestion feed this command receives start_time from the scheduler's
|
||||
high-water mark, and the mark's format is an operator setting on the rule
|
||||
(iso8601, unix, unix_ms). The API only takes milliseconds. Reading all three
|
||||
here is what keeps the feed from silently fetching the wrong window — or
|
||||
nothing at all — because of a dropdown three screens away.
|
||||
"""
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v).strip()
|
||||
if s.isdigit():
|
||||
n = int(s)
|
||||
return n if n >= _MS_FLOOR else n * 1000
|
||||
try:
|
||||
return int(datetime.datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def request(method, url, params=None):
|
||||
base, headers = _cfg()
|
||||
full = url if url.startswith("http") else base + url.lstrip("/")
|
||||
@@ -43,26 +68,73 @@ def request(method, url, params=None):
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _collect(first):
|
||||
# Follow the query's pagination links until no more pages remain.
|
||||
def _collect(first, limit):
|
||||
"""Walk the pagination links, stopping at `limit`.
|
||||
|
||||
The walk is bounded because this runs unattended every few minutes: a query
|
||||
that matches more than it was meant to would otherwise hold the run open for
|
||||
as many pages as the window contains, and the next tick would find the rule
|
||||
still in flight and skip itself.
|
||||
"""
|
||||
events = []
|
||||
queue = [first]
|
||||
while queue:
|
||||
while queue and len(events) < limit:
|
||||
page = queue.pop(0)
|
||||
events.extend(page.get("events", []) or [])
|
||||
for link in page.get("links", []) or []:
|
||||
href = link.get("href")
|
||||
if href:
|
||||
queue.append(request("GET", href))
|
||||
return events
|
||||
return events[:limit]
|
||||
|
||||
|
||||
def _identify(events, fallback_log_id, detection_name):
|
||||
"""Give every event a stable id, and the name of the rule it stands for.
|
||||
|
||||
dedup_key needs a field that is identical across runs for the same log line.
|
||||
The scheduler deliberately re-reads the last minute of the previous window on
|
||||
every run (so an event written while a poll was in flight is not lost), which
|
||||
means the same line WILL come back — anything derived from the fetch time, or
|
||||
from the position in the page, would make it a second incident every time.
|
||||
log_id + sequence_number is the line's own address in Log Search; the hash is
|
||||
the fallback for a source that does not number its entries.
|
||||
|
||||
detection_name is stamped here because the log line has no idea which basic
|
||||
detection rule it is being replayed for. Without it an incident is titled by
|
||||
its own first characters, and an analyst reads a log line where the source
|
||||
console shows an alert name.
|
||||
"""
|
||||
out = []
|
||||
for e in events:
|
||||
if not isinstance(e, dict):
|
||||
continue
|
||||
log_id = e.get("log_id") or fallback_log_id or ""
|
||||
seq = e.get("sequence_number")
|
||||
if seq in (None, ""):
|
||||
digest = hashlib.sha1(
|
||||
("%s|%s|%s" % (log_id, e.get("timestamp"), e.get("message"))).encode("utf-8", "replace")
|
||||
).hexdigest()
|
||||
e["event_id"] = "%s:%s" % (log_id, digest)
|
||||
else:
|
||||
e["event_id"] = "%s:%s" % (log_id, seq)
|
||||
if not e.get("log_id"):
|
||||
e["log_id"] = log_id
|
||||
if detection_name:
|
||||
e["detection_name"] = detection_name
|
||||
out.append(e)
|
||||
return out
|
||||
|
||||
|
||||
def run():
|
||||
inp = _inputs()
|
||||
start_time = inp.get("start_time")
|
||||
end_time = inp.get("end_time")
|
||||
start_time = _to_ms(inp.get("start_time"))
|
||||
end_time = _to_ms(inp.get("end_time"))
|
||||
if inp.get("time_range"):
|
||||
start_time, end_time = _rel_ms_range(inp.get("time_range"))
|
||||
try:
|
||||
limit = int(inp.get("limit") or 1000)
|
||||
except (TypeError, ValueError):
|
||||
limit = 1000
|
||||
params = {
|
||||
"query": inp.get("query"),
|
||||
"from": start_time,
|
||||
@@ -72,7 +144,8 @@ def run():
|
||||
}
|
||||
log_id = inp.get("log_id", "")
|
||||
first = request("GET", "log_search/query/logs/" + urllib.parse.quote(log_id, safe=""), params=params)
|
||||
print(json.dumps({"events": _collect(first)}))
|
||||
events = _identify(_collect(first, limit), log_id, inp.get("detection_name"))
|
||||
print(json.dumps({"events": events, "count": len(events), "from": start_time, "to": end_time}))
|
||||
|
||||
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user