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", "{}")) region = str(s.get("region") or "us").strip().lower() base = "https://" + region + ".api.insight.rapid7.com/" headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"} return base, headers def _inputs(): return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) def _rel_ms_range(text): if not text: return None, None num = unit = None for t in str(text).lower().replace("last", "").split(): if t.isdigit(): num = int(t) elif t.rstrip("s") in _UNITS: unit = t.rstrip("s") if num is None or not unit: return None, None now_ms = int(time.time() * 1000) 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("/") if params: clean = {k: v for k, v in params.items() if v not in (None, "")} if clean: full += ("&" if "?" in full else "?") + urllib.parse.urlencode(clean, doseq=True) req = urllib.request.Request(full, headers=headers, method=method) with urllib.request.urlopen(req, timeout=90) as r: raw = r.read() return json.loads(raw) if raw else {} 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 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[: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 = _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, "to": end_time, "per_page": inp.get("logs_per_page"), "sequence_number": inp.get("sequence_number"), } log_set_id = inp.get("log_set_id", "") first = request("GET", "log_search/query/logsets/" + urllib.parse.quote(log_set_id, safe=""), params=params) # No fallback log id: a log set spans several logs and every event names its own. events = _identify(_collect(first, limit), "", inp.get("detection_name")) print(json.dumps({"events": events, "count": len(events), "from": start_time, "to": end_time})) try: run() except urllib.error.HTTPError as e: print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) sys.exit(1) except Exception as e: print(json.dumps({"error": str(e)})) sys.exit(1)