First Seen Network Flow Exporter Followed by Suspicious Source Activity

  1[metadata]
  2creation_date = "2026/08/24"
  3integration = ["netflow", "goflow2"]
  4maturity = "production"
  5updated_date = "2026/09/18"
  6
  7[rule]
  8author = ["Elastic"]
  9description = """
 10Identifies a newly observed NetFlow, IPFIX, or sFlow exporter IP followed by another detection alert with
 11medium-or-higher severity or an elevated risk score, where that exporter IP is the source of the detected activity in
 12the same data stream namespace. This correlation adds behavioral evidence that can help distinguish routine exporter
 13onboarding from a potentially unauthorized or compromised exporter introduced as part of defense evasion.
 14"""
 15false_positives = [
 16    """
 17    Newly deployed network devices may legitimately generate other detection alerts during commissioning, testing,
 18    vulnerability scanning, or monitoring validation. Shared addresses, network address translation, and incomplete
 19    asset attribution can also cause unrelated activity to correlate to the exporter IP.
 20    """,
 21]
 22from = "now-45m"
 23interval = "15m"
 24language = "esql"
 25license = "Elastic License v2"
 26name = "First Seen Network Flow Exporter Followed by Suspicious Source Activity"
 27note = """## Triage and analysis
 28
 29### Investigating First Seen Network Flow Exporter Followed by Suspicious Source Activity
 30
 31This higher-order rule correlates the building-block alert "First Seen Network Flow Exporter" with a later detection
 32alert that has medium-or-higher severity or an elevated risk score within 30 minutes. The new exporter's `observer.ip`
 33must equal the later alert's `source.ip`, and both alerts must share the same `data_stream.namespace`.
 34
 35This relationship provides stronger evidence than suspicious flows merely reported by the exporter. It indicates that
 36the exporter address itself was subsequently identified as the source of separately detected activity. It does not by
 37itself prove telemetry injection, exporter compromise, or the addition of a rogue collector. A rogue collector receives
 38flow exports and generally requires network-device configuration or audit telemetry to detect directly.
 39
 40### Possible investigation steps
 41
 42- Review the `observer.ip`, `source.ip`, correlation timestamps, and suspicious rule names and IDs surfaced in the
 43  alert. Pivot to the contributing alerts to review their risk score, destination, host, and user context.
 44- Confirm that `observer.ip` belongs to the newly observed exporter and that NAT, shared addressing, or stale asset
 45  records did not cause the correlation.
 46- Examine the second alert's rule name and determine what behavior associated the exporter address with `source.ip`.
 47- Verify the exporter's owner, device type, software version, management plane, recent configuration changes, and
 48  approved collector destinations.
 49- Review authentication, process, network-device audit, and configuration telemetry for the exporter around the
 50  correlation window.
 51- Compare observation domains, exporter version, uptime, sequence values, templates, sampling configuration, and flow
 52  volume with established exporters.
 53- Search for additional alerts involving the exporter address and for malformed, implausible, or conflicting flow
 54  records received from it.
 55
 56### False positive analysis
 57
 58- Validate recent device onboarding, migrations, disaster-recovery activation, penetration tests, vulnerability scans,
 59  and monitoring validation.
 60- Determine whether `source.ip` represents the exporter itself or a shared NAT, proxy, scanner, or management address.
 61- Confirm that the correlated source alert is a meaningful behavioral detection rather than an expected administrative
 62  or discovery event.
 63
 64### Response and remediation
 65
 66- If the exporter is unauthorized or compromised, restrict it at the collector boundary and isolate its management
 67  plane as appropriate.
 68- Preserve flow records, source alerts, configuration history, authentication events, and network evidence.
 69- Remove unauthorized export configuration, restore trusted device configuration, rotate affected credentials, and
 70  review collector ACLs and authentication.
 71"""
 72references = [
 73    "https://www.elastic.co/docs/solutions/security/detect-and-alert/about-building-block-rules",
 74    "https://www.elastic.co/docs/reference/integrations/netflow",
 75    "https://www.elastic.co/docs/reference/integrations/goflow2",
 76    "https://datatracker.ietf.org/doc/html/rfc7011",
 77    "https://sflow.org/sflow_version_5.txt",
 78]
 79risk_score = 73
 80rule_id = "882c8edc-1e39-4a60-80b2-485491b6c91c"
 81setup = """## Setup
 82
 83Enable the building-block rule "First Seen Network Flow Exporter" and detection rules that can identify suspicious
 84activity with `source.ip` populated. The building-block rule requires decoded NetFlow, IPFIX, or sFlow records from the
 85Elastic NetFlow (`netflow.log`) or GoFlow2 (`goflow2.sflow`) integration.
 86
 87The higher-order rule searches Elastic Security alert indices. The correlated source alert must preserve the exporter
 88address in `source.ip` and preserve `data_stream.namespace`; alerts without either field cannot contribute to this
 89correlation. Building-block, higher-order, machine-learning, new-terms, threat-match, and deprecated-rule alerts are
 90excluded as the second event.
 91"""
 92severity = "high"
 93tags = [
 94    "Domain: Network",
 95    "Use Case: Network Security Monitoring",
 96    "Use Case: Threat Detection",
 97    "Tactic: Defense Evasion",
 98    "Rule Type: Higher-Order Rule",
 99    "Data Source: Elastic Security",
100    "Data Source: NetFlow",
101    "Data Source: GoFlow2",
102    "Resources: Investigation Guide",
103    "Rule Type: ES|QL",
104]
105timestamp_override = "event.ingested"
106type = "esql"
107
108query = '''
109FROM .alerts-security.*
110| WHERE event.kind == "signal"
111    AND data_stream.namespace IS NOT NULL
112    AND (
113      (kibana.alert.rule.rule_id == "dfe3f626-4224-417e-aff1-8ef9a72c3191" AND observer.ip IS NOT NULL)
114      OR
115      (source.ip IS NOT NULL
116        AND kibana.alert.rule.name IS NOT NULL
117        AND kibana.alert.rule.rule_id IS NOT NULL
118        AND kibana.alert.rule.rule_id != "dfe3f626-4224-417e-aff1-8ef9a72c3191"
119        AND (kibana.alert.risk_score >= 47 OR kibana.alert.severity IN ("medium", "high", "critical"))
120        AND KQL("""NOT kibana.alert.building_block_type : *""")
121        AND NOT kibana.alert.rule.type IN ("machine_learning", "new_terms", "threat_match")
122        AND NOT kibana.alert.rule.name LIKE "Deprecated - *"
123        AND NOT KQL("""kibana.alert.rule.tags : "Rule Type: Higher-Order Rule" """))
124    )
125| EVAL
126    Esql.is_exporter_alert = kibana.alert.rule.rule_id == "dfe3f626-4224-417e-aff1-8ef9a72c3191",
127    Esql.is_suspicious_source_alert = kibana.alert.rule.rule_id != "dfe3f626-4224-417e-aff1-8ef9a72c3191",
128    Esql.correlation_ip = CASE(Esql.is_exporter_alert, observer.ip, source.ip),
129    Esql.exporter_alert_timestamp = CASE(Esql.is_exporter_alert, @timestamp, null),
130    Esql.suspicious_source_alert_timestamp = CASE(Esql.is_suspicious_source_alert, @timestamp, null),
131    Esql.suspicious_rule_name = CASE(Esql.is_suspicious_source_alert, kibana.alert.rule.name, null),
132    Esql.suspicious_rule_id = CASE(Esql.is_suspicious_source_alert, kibana.alert.rule.rule_id, null)
133| WHERE Esql.correlation_ip IS NOT NULL
134| STATS
135    Esql.exporter_alert_count = SUM(CASE(Esql.is_exporter_alert, 1, 0)),
136    Esql.suspicious_source_alert_count = SUM(CASE(Esql.is_suspicious_source_alert, 1, 0)),
137    observer.ip = MAX(CASE(Esql.is_exporter_alert, Esql.correlation_ip, null)),
138    source.ip = MAX(CASE(Esql.is_suspicious_source_alert, Esql.correlation_ip, null)),
139    Esql.exporter_alert_timestamp = MIN(Esql.exporter_alert_timestamp),
140    Esql.suspicious_source_alert_timestamp = MAX(Esql.suspicious_source_alert_timestamp),
141    Esql.suspicious_rule_name_values = VALUES(Esql.suspicious_rule_name),
142    Esql.suspicious_rule_id_values = VALUES(Esql.suspicious_rule_id)
143  BY data_stream.namespace, Esql.correlation_ip
144| EVAL Esql.time_diff_seconds = DATE_DIFF(
145    "second", Esql.exporter_alert_timestamp, Esql.suspicious_source_alert_timestamp
146  )
147| WHERE Esql.exporter_alert_count > 0
148    AND Esql.suspicious_source_alert_count > 0
149    AND Esql.time_diff_seconds >= 0
150    AND Esql.time_diff_seconds <= 1800
151| KEEP
152    data_stream.namespace,
153    observer.ip,
154    source.ip,
155    Esql.correlation_ip,
156    Esql.exporter_alert_count,
157    Esql.suspicious_source_alert_count,
158    Esql.exporter_alert_timestamp,
159    Esql.suspicious_source_alert_timestamp,
160    Esql.suspicious_rule_name_values,
161    Esql.suspicious_rule_id_values,
162    Esql.time_diff_seconds
163'''
164
165
166[[rule.threat]]
167framework = "MITRE ATT&CK"
168[[rule.threat.technique]]
169id = "T1562"
170name = "Impair Defenses"
171reference = "https://attack.mitre.org/techniques/T1562/"
172
173
174[rule.threat.tactic]
175id = "TA0005"
176name = "Defense Evasion"
177reference = "https://attack.mitre.org/tactics/TA0005/"
178
179[rule.investigation_fields]
180field_names = [
181    "observer.ip",
182    "source.ip",
183    "data_stream.namespace",
184    "Esql.exporter_alert_timestamp",
185    "Esql.suspicious_source_alert_timestamp",
186    "Esql.suspicious_rule_name_values",
187    "Esql.suspicious_rule_id_values",
188    "Esql.time_diff_seconds",
189]