Anomaly found in Network Session Traffic (ASIM Network Session schema)

Status: available
Severity: medium
Time window: 14d
Group by: NetworkCustomAnalyticsExists, NetworkSummary_Protocol_Exists, exists
Source: github.com/Azure/Azure-Sentinel

The rule identifies anomalous pattern in network session traffic based on previously seen data, different Device Action, Network Protocol, Network Direction or overall volume. The rule utilize ASIM normalization, and is applied to any source which supports the ASIM Network Session schema. This rule leverages log summaries generated by a Summary Rule or Summarized Playbook. If no such summaries are available, the rule falls back to direct analysis using ASIM function.

MITRE ATT&CK coverage

Tactic	Techniques
Discovery	`T1046` Network Service Discovery
Lateral Movement	`T1210` Exploitation of Remote Services
Command & Control	`T1071` Application Layer Protocol `T1095` Non-Application Layer Protocol
Exfiltration	`T1030` Data Transfer Size Limits

Telemetry coverage

Provider	Record / event type
Sysmon	Event ID 3: Network connection
Security-Auditing	Event ID 5152: The Windows Filtering Platform blocked a packet.
Security-Auditing	Event ID 5154: The Windows Filtering Platform has permitted an application or service to listen on a port for incoming connections.
Security-Auditing	Event ID 5155: The Windows Filtering Platform has blocked an application or service from listening on a port for incoming connections.
Security-Auditing	Event ID 5156: The Windows Filtering Platform has permitted a connection.
Security-Auditing	Event ID 5157: The Windows Filtering Platform has blocked a connection.
Security-Auditing	Event ID 5158: The Windows Filtering Platform has permitted a bind to a local port.
Security-Auditing	Event ID 5159: The Windows Filtering Platform has blocked a bind to a local port.

Rule body

id: cd6def0d-3ef0-4d55-a7e3-faa96c46ba12
name: Anomaly found in Network Session Traffic (ASIM Network Session schema)
description: |
  'The rule identifies anomalous pattern in network session traffic based on previously seen data, different Device Action, Network Protocol, Network Direction or overall volume. The rule utilize [ASIM](https://aka.ms/AboutASIM) normalization, and is applied to any source which supports the ASIM Network Session schema. This rule leverages log summaries generated by a Summary Rule or Summarized Playbook. If no such summaries are available, the rule falls back to direct analysis using ASIM function.'
severity: Medium
status: Available 
tags:
  - Schema: ASimNetworkSessions
    SchemaVersion: 0.2.4
requiredDataConnectors:
  - connectorId: AWSS3
    dataTypes:
      - AWSVPCFlow
  - connectorId: MicrosoftThreatProtection
    dataTypes:
      - DeviceNetworkEvents
  - connectorId: SecurityEvents
    dataTypes:
      - SecurityEvent
  - connectorId: WindowsSecurityEvents
    dataTypes:
      - SecurityEvent
  - connectorId: WindowsForwardedEvents
    dataTypes:
      - WindowsEvent
  - connectorId: Zscaler
    dataTypes:
      - CommonSecurityLog
  - connectorId: MicrosoftSysmonForLinux
    dataTypes:
      - Syslog
  - connectorId: PaloAltoNetworks
    dataTypes:
      - CommonSecurityLog
  - connectorId: AzureMonitor(VMInsights)
    dataTypes:
      - VMConnection
  - connectorId: AzureFirewall
    dataTypes:
      - AzureDiagnostics
  - connectorId: AzureNSG
    dataTypes:
      - AzureDiagnostics
  - connectorId: CiscoASA
    dataTypes:
      - CommonSecurityLog
  - connectorId: CiscoAsaAma
    dataTypes:
      - CommonSecurityLog
  - connectorId: Corelight
    dataTypes:
      - Corelight_CL
  - connectorId: AIVectraStream
    dataTypes:
      - VectraStream
  - connectorId: CheckPoint
    dataTypes:
      - CommonSecurityLog
  - connectorId: Fortinet
    dataTypes:
      - CommonSecurityLog
  - connectorId: CiscoMeraki
    dataTypes:
      - Syslog
      - CiscoMerakiNativePoller

queryFrequency: 1d
queryPeriod: 14d
triggerOperator: gt
triggerThreshold: 0
tactics:
  - CommandAndControl
  - Discovery
  - Exfiltration
  - LateralMovement
relevantTechniques:
  - T1095
  - T1071
  - T1046
  - T1030
  - T1210
query: |
  let min_t = ago(14d);
  let max_t = now();
  let dt = 1d;
  let fieldForDvcAction = "DvcAction";
  let fieldForNetworkDirection = "NetworkDirection";
  let fieldForNetworkProtocol = "NetworkProtocol";
  let AnomalyThreshold = 2.5;
  let eps = materialize (_Im_NetworkSession
      | project TimeGenerated
      | where TimeGenerated > ago(5m)
      | count
      | extend Count = Count / 300);
  let maxSummarizedTime = toscalar (
      union isfuzzy=true
          (
          NetworkSummary_Protocol_CL
          | where EventTime > min_t
          | summarize max_TimeGenerated=max(EventTime)
          | extend max_TimeGenerated = datetime_add('minute', 20, max_TimeGenerated)
          ),
          (
          NetworkCustomAnalytics_protocol_CL
          | where EventTime_t > min_t
          | summarize max_TimeGenerated=max(EventTime_t)
          | extend max_TimeGenerated = datetime_add('minute', 20, max_TimeGenerated)
          ),
          (
          print(min_t)
          | project max_TimeGenerated = print_0
          )
      | summarize maxTimeGenerated = max(max_TimeGenerated) 
      );
  let NetworkSummary_Protocol = materialize(
      union isfuzzy=true 
          (
          NetworkSummary_Protocol_CL
          | where EventTime > ago(1d)
          | project v = int(2)
          ),
          (
          print int(1) 
          | project v = print_0
          )
      | summarize maxv = max(v)
      | extend
          NetworkSummary_Protocol_Exists = (maxv > 1),
          NetworkCustomAnalyticsExists = false
      );
  let NetworkCustomAnalytics = materialize(
      union isfuzzy=true 
          (
          NetworkCustomAnalytics_protocol_CL
          | where EventTime_t > ago(1d) 
          | project v = int(2)
          ),
          (
          print int(1) 
          | project v = print_0
          )
      | summarize maxv = max(v)
      | extend
          NetworkCustomAnalyticsExists = (maxv > 1),
          NetworkSummary_Protocol_Exists = false
      );
  let allData = union isfuzzy=true 
          (
          (datatable(
      exists: int,
      NetworkSummary_Protocol_Exists: bool,
      NetworkCustomAnalyticsExists: bool
  )[
      1, false, false
  ]
          | where toscalar(eps) > 500
          | join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists
          | join (NetworkCustomAnalytics) on NetworkCustomAnalyticsExists)
          | join (
              _Im_NetworkSession(starttime=todatetime(ago(3d)), endtime=now())
              | where TimeGenerated > maxSummarizedTime
              | summarize Count=count()
                  by
                  NetworkProtocol,
                  DstPortNumber,
                  DstAppName,
                  NetworkDirection,
                  DvcAction,
                  bin(TimeGenerated, 10m)
              | extend
                  EventTime = TimeGenerated,
                  Count = toint(Count),
                  DstPortNumber = toint(DstPortNumber),
                  exists=int(1)
              )
              on exists
          | project-away exists*, maxv, NetworkSummary_Protocol_Exists*, NetworkCustomAnalyticsExists*
          ),
          (
          (datatable(
      exists: int,
      NetworkSummary_Protocol_Exists: bool,
      NetworkCustomAnalyticsExists: bool
  )[
      1, false, false
  ]
          | where toscalar(eps) <= 500
          | join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists
          | join (NetworkCustomAnalytics) on NetworkCustomAnalyticsExists
          )
          | join (
              _Im_NetworkSession(starttime=todatetime(ago(4d)), endtime=now())
              | where TimeGenerated > maxSummarizedTime
              | summarize Count=count()
                  by
                  NetworkProtocol,
                  DstPortNumber,
                  DstAppName,
                  NetworkDirection,
                  DvcAction,
                  bin(TimeGenerated, 10m)
              | extend
                  EventTime = TimeGenerated,
                  Count = toint(Count),
                  DstPortNumber = toint(DstPortNumber),
                  exists=int(1)
              )
              on exists
          | project-away exists*, maxv, NetworkSummary_Protocol_Exists*, NetworkCustomAnalyticsExists*
          ),
          (
          (datatable(exists: int, NetworkSummary_Protocol_Exists: bool)[1, false]
          | join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists
          )
          | join (
              NetworkCustomAnalytics_protocol_CL
              | where EventTime_t > min_t
              | project
                  NetworkProtocol=NetworkProtocol_s,
                  DstPortNumber=DstPortNumber_d,
                  DstAppName=DstAppName_s,
                  NetworkDirection=NetworkDirection_s,
                  DvcAction=DvcAction_s,
                  Count=count__d,
                  EventTime=EventTime_t,
                  TimeGenerated,
                  Type
              | extend Count = toint(Count), DstPortNumber = toint(DstPortNumber), exists=int(1)
              )
              on exists
          | project-away exists*, maxv, NetworkSummary_Protocol_Exists*, NetworkCustomAnalyticsExists*
          ),
          (
          NetworkSummary_Protocol_CL
          | where EventTime > min_t
          | project
              NetworkProtocol,
              DstPortNumber,
              DstAppName,
              NetworkDirection,
              DvcAction,
              Count=count_,
              EventTime,
              TimeGenerated,
              Type
          | extend Count = toint(Count), DstPortNumber = toint(DstPortNumber)
          )
  ;
  let findVolumneBasedAnomaly = allData
      | make-series total=sum(Count) on EventTime from min_t to max_t step dt
      | extend (anomalies, score, baseline) = series_decompose_anomalies(total, AnomalyThreshold, -1, 'linefit') 
      | mv-expand anomalies, score, baseline, EventTime, total
      | extend
          anomalies = toint(anomalies),
          score = toint(score),
          baseline = toint(baseline),
          EventTime = todatetime(EventTime),
          total = tolong(total)
      | where EventTime >= ago(1d)
      | where score >= 2 * AnomalyThreshold
  ;
  let findAnomalies = (field: string) {
      allData
      | where isnotempty(column_ifexists(field, ""))
      | make-series total=sum(Count) on EventTime from min_t to max_t step dt by column_ifexists(field, "")
      | extend (anomalies, score, baseline) = series_decompose_anomalies(total, AnomalyThreshold, -1, 'linefit')
      | mv-expand anomalies, score, baseline, EventTime, total
      | extend
          anomalies = toint(anomalies),
          score = toint(score),
          baseline = toint(baseline),
          EventTime = todatetime(EventTime),
          total = tolong(total)
      | where EventTime >= ago(1d)
      | where score >= 2 * AnomalyThreshold
  };
  union
      findAnomalies(fieldForDvcAction),
      findAnomalies(fieldForNetworkDirection),
      findAnomalies(fieldForNetworkProtocol),
      findVolumneBasedAnomaly
  | extend
      anomalyFieldType  = case (
                          isnotempty(column_ifexists(fieldForDvcAction, "")),
                          "DvcAction",
                          isnotempty(column_ifexists(fieldForNetworkDirection, "")),
                          "NetworkDirection",
                          isnotempty(column_ifexists(fieldForNetworkProtocol, "")),
                          "NetworkProtocol",
                          "TotalVolume"
                      ),
      anomalyFieldValue  = case (
                          isnotempty(column_ifexists(fieldForDvcAction, "")),
                          column_ifexists(fieldForDvcAction, ""),
                          isnotempty(column_ifexists(fieldForNetworkDirection, "")),
                          column_ifexists(fieldForNetworkDirection, ""),
                          isnotempty(column_ifexists(fieldForNetworkProtocol, "")),
                          column_ifexists(fieldForNetworkProtocol, ""),
                          "Overall"
                      )
eventGroupingSettings:
  aggregationKind: AlertPerResult
customDetails:
  AnomalyFieldType: anomalyFieldType
  AnomalyFieldValue: anomalyFieldValue
  Score: score

alertDetailsOverride:
  alertDisplayNameFormat: Anomaly was observed with {{anomalyFieldValue}} Traffic
  alertDescriptionFormat: 'Based on past data, anomaly was observed in {{anomalyFieldValue}} Traffic with a score of {{score}}.'
version: 1.3.0
kind: Scheduled

Stages and Predicates

Stage 0: `let`

let min_t = ago(14d);
let max_t = now();
let dt = 1d;
let fieldForDvcAction = "DvcAction";
let fieldForNetworkDirection = "NetworkDirection";
let fieldForNetworkProtocol = "NetworkProtocol";
let AnomalyThreshold = 2.5;
let eps = materialize (_Im_NetworkSession
    | project TimeGenerated
    | where TimeGenerated > ago(5m)
    | count
    | extend Count = Count / 300);
let maxSummarizedTime = toscalar (
    union isfuzzy=true
        (
        NetworkSummary_Protocol_CL
        | where EventTime > min_t
        | summarize max_TimeGenerated=max(EventTime)
        | extend max_TimeGenerated = datetime_add('minute', 20, max_TimeGenerated)
        ),
        (
        NetworkCustomAnalytics_protocol_CL
        | where EventTime_t > min_t
        | summarize max_TimeGenerated=max(EventTime_t)
        | extend max_TimeGenerated = datetime_add('minute', 20, max_TimeGenerated)
        ),
        (
        print(min_t)
        | project max_TimeGenerated = print_0
        )
    | summarize maxTimeGenerated = max(max_TimeGenerated) 
    );
let NetworkSummary_Protocol = materialize(
    union isfuzzy=true 
        (
        NetworkSummary_Protocol_CL
        | where EventTime > ago(1d)
        | project v = int(2)
        ),
        (
        print int(1) 
        | project v = print_0
        )
    | summarize maxv = max(v)
    | extend
        NetworkSummary_Protocol_Exists = (maxv > 1),
        NetworkCustomAnalyticsExists = false
    );
let NetworkCustomAnalytics = materialize(
    union isfuzzy=true 
        (
        NetworkCustomAnalytics_protocol_CL
        | where EventTime_t > ago(1d) 
        | project v = int(2)
        ),
        (
        print int(1) 
        | project v = print_0
        )
    | summarize maxv = max(v)
    | extend
        NetworkCustomAnalyticsExists = (maxv > 1),
        NetworkSummary_Protocol_Exists = false
    );
let allData = union isfuzzy=true 
        (
        (datatable(
    exists: int,
    NetworkSummary_Protocol_Exists: bool,
    NetworkCustomAnalyticsExists: bool
)[
    1, false, false
]
        | where toscalar(eps) > 500
        | join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists
        | join (NetworkCustomAnalytics) on NetworkCustomAnalyticsExists)
        | join (
            _Im_NetworkSession(starttime=todatetime(ago(3d)), endtime=now())
            | where TimeGenerated > maxSummarizedTime
            | summarize Count=count()
                by
                NetworkProtocol,
                DstPortNumber,
                DstAppName,
                NetworkDirection,
                DvcAction,
                bin(TimeGenerated, 10m)
            | extend
                EventTime = TimeGenerated,
                Count = toint(Count),
                DstPortNumber = toint(DstPortNumber),
                exists=int(1)
            )
            on exists
        | project-away exists*, maxv, NetworkSummary_Protocol_Exists*, NetworkCustomAnalyticsExists*
        ),
        (
        (datatable(
    exists: int,
    NetworkSummary_Protocol_Exists: bool,
    NetworkCustomAnalyticsExists: bool
)[
    1, false, false
]
        | where toscalar(eps) <= 500
        | join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists
        | join (NetworkCustomAnalytics) on NetworkCustomAnalyticsExists
        )
        | join (
            _Im_NetworkSession(starttime=todatetime(ago(4d)), endtime=now())
            | where TimeGenerated > maxSummarizedTime
            | summarize Count=count()
                by
                NetworkProtocol,
                DstPortNumber,
                DstAppName,
                NetworkDirection,
                DvcAction,
                bin(TimeGenerated, 10m)
            | extend
                EventTime = TimeGenerated,
                Count = toint(Count),
                DstPortNumber = toint(DstPortNumber),
                exists=int(1)
            )
            on exists
        | project-away exists*, maxv, NetworkSummary_Protocol_Exists*, NetworkCustomAnalyticsExists*
        ),
        (
        (datatable(exists: int, NetworkSummary_Protocol_Exists: bool)[1, false]
        | join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists
        )
        | join (
            NetworkCustomAnalytics_protocol_CL
            | where EventTime_t > min_t
            | project
                NetworkProtocol=NetworkProtocol_s,
                DstPortNumber=DstPortNumber_d,
                DstAppName=DstAppName_s,
                NetworkDirection=NetworkDirection_s,
                DvcAction=DvcAction_s,
                Count=count__d,
                EventTime=EventTime_t,
                TimeGenerated,
                Type
            | extend Count = toint(Count), DstPortNumber = toint(DstPortNumber), exists=int(1)
            )
            on exists
        | project-away exists*, maxv, NetworkSummary_Protocol_Exists*, NetworkCustomAnalyticsExists*
        ),
        (
        NetworkSummary_Protocol_CL
        | where EventTime > min_t
        | project
            NetworkProtocol,
            DstPortNumber,
            DstAppName,
            NetworkDirection,
            DvcAction,
            Count=count_,
            EventTime,
            TimeGenerated,
            Type
        | extend Count = toint(Count), DstPortNumber = toint(DstPortNumber)
        )
;
let findVolumneBasedAnomaly = allData
    | make-series total=sum(Count) on EventTime from min_t to max_t step dt
    | extend (anomalies, score, baseline) = series_decompose_anomalies(total, AnomalyThreshold, -1, 'linefit') 
    | mv-expand anomalies, score, baseline, EventTime, total
    | extend
        anomalies = toint(anomalies),
        score = toint(score),
        baseline = toint(baseline),
        EventTime = todatetime(EventTime),
        total = tolong(total)
    | where EventTime >= ago(1d)
    | where score >= 2 * AnomalyThreshold
;
let findAnomalies = (field: string) {
    allData
    | where isnotempty(column_ifexists(field, ""))
    | make-series total=sum(Count) on EventTime from min_t to max_t step dt by column_ifexists(field, "")
    | extend (anomalies, score, baseline) = series_decompose_anomalies(total, AnomalyThreshold, -1, 'linefit')
    | mv-expand anomalies, score, baseline, EventTime, total
    | extend
        anomalies = toint(anomalies),
        score = toint(score),
        baseline = toint(baseline),
        EventTime = todatetime(EventTime),
        total = tolong(total)
    | where EventTime >= ago(1d)
    | where score >= 2 * AnomalyThreshold
};

Stage 1: `source`

let allData

Stage 2: `source`

let findVolumneBasedAnomaly

Stage 3: `union`

union of 4 branches

Stage 4: `source`

findAnomalies

Stage 5: `source`

findAnomalies

Stage 6: `source`

findAnomalies

Stage 7: `union`

union of 4 branches

Stage 8: `source`

datatable

Stage 9: `where`

where /* macro: (toscalar(eps) > 500) */

Stage 10: `join`

join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists

Stage 11: `join`

join (NetworkCustomAnalytics) on NetworkCustomAnalyticsExists

Stage 12: `join`

join (_Im_NetworkSession) on exists

Stage 13: `project-away`

project-away NetworkCustomAnalyticsExists*, NetworkSummary_Protocol_Exists*, exists*, maxv

Stage 14: `source`

datatable

Stage 15: `where`

where /* macro: (toscalar(eps) <= 500) */

Stage 16: `join`

join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists

Stage 17: `join`

join (NetworkCustomAnalytics) on NetworkCustomAnalyticsExists

Stage 18: `join`

join (_Im_NetworkSession) on exists

Stage 19: `project-away`

project-away NetworkCustomAnalyticsExists*, NetworkSummary_Protocol_Exists*, exists*, maxv

Stage 20: `source`

datatable

Stage 21: `join`

join (NetworkSummary_Protocol) on NetworkSummary_Protocol_Exists

Stage 22: `join`

join (NetworkCustomAnalytics_protocol_CL) on exists

Stage 23: `project-away`

project-away NetworkCustomAnalyticsExists*, NetworkSummary_Protocol_Exists*, exists*, maxv

Stage 24: `source`

NetworkSummary_Protocol_CL

Stage 25: `where`

where EventTime > min_t

Stage 26: `project`

project Count, DstAppName, DstPortNumber, DvcAction, EventTime, NetworkDirection, NetworkProtocol, TimeGenerated, Type

Stage 27: `extend`

extend Count, DstPortNumber

Stage 28: `summarize`

summarize total

Stage 29: `extend`

extend anomalies, baseline, score

Stage 30: `mv-expand`

mv-expand anomalies

Stage 31: `extend`

extend EventTime, anomalies, baseline, score, total

Stage 32: `where`

where EventTime >= ago(86400s)

Stage 33: `where`

where score >= 5

Stage 34: `extend`

extend anomalyFieldType, anomalyFieldValue

anomalyFieldType =

if/* macro: isnotempty(column_ifexists(fieldForDvcAction, "")) */"DvcAction"

elif/* macro: isnotempty(column_ifexists(fieldForNetworkDirection, "")) */"NetworkDirection"

elif/* macro: isnotempty(column_ifexists(fieldForNetworkProtocol, "")) */"NetworkProtocol"

else"TotalVolume"

anomalyFieldValue =

if/* macro: isnotempty(column_ifexists(fieldForDvcAction, "")) */column_ifexists(fieldForDvcAction, "")

elif/* macro: isnotempty(column_ifexists(fieldForNetworkDirection, "")) */column_ifexists(fieldForNetworkDirection, "")

elif/* macro: isnotempty(column_ifexists(fieldForNetworkProtocol, "")) */column_ifexists(fieldForNetworkProtocol, "")

else"Overall"

Indicators

These rows show field, operator, and value matches.

Field	Kind	Values	Search
`EventTime`	cross_field_compare	`min_t` transforms: `op:gt`	`field:"EventTime" kind:cross_field_compare value:"min_t"`
`EventTime_t`	cross_field_compare	`min_t` transforms: `op:gt` corpus 4 (kusto 4)	`field:"EventTime_t" kind:cross_field_compare value:"min_t"`
`TimeGenerated`	cross_field_compare	`maxSummarizedTime` transforms: `op:gt` corpus 4 (kusto 4)	`field:"TimeGenerated" kind:cross_field_compare value:"maxSummarizedTime"`
`score`	ge	`5` corpus 6 (kusto 6)	`field:"score" kind:ge value:"5"`

Output fields

These fields are emitted when the rule matches.

Field	Source
`total`	`extend`
`anomalies`	`extend`
`baseline`	`extend`
`score`	`extend`
`EventTime`	`extend`
`anomalyFieldType`	`extend`
`anomalyFieldValue`	`extend`

Anomaly found in Network Session Traffic (ASIM Network Session schema)

MITRE ATT&CK coverage

Telemetry coverage

Rule body

Stages and Predicates

Stage 0: let

Stage 1: source

Stage 2: source

Stage 3: union

Stage 4: source

Stage 5: source

Stage 6: source

Stage 7: union

Stage 8: source

Stage 9: where

Stage 10: join

Stage 11: join

Stage 12: join

Stage 13: project-away

Stage 14: source

Stage 15: where

Stage 16: join

Stage 17: join

Stage 18: join

Stage 19: project-away

Stage 20: source

Stage 21: join

Stage 22: join

Stage 23: project-away

Stage 24: source

Stage 25: where

Stage 26: project

Stage 27: extend

Stage 28: summarize

Stage 29: extend

Stage 30: mv-expand

Stage 31: extend

Stage 32: where

Stage 33: where

Stage 34: extend