Security Alert Alert Generation Anomaly

Query

let query_frequency = 1h;
let query_period = 7d;
let scan_step = 5m;
let consecutive_failures_threshold = 2;
let _SecurityAlertTimeSeries = (start_time: datetime, end_time: datetime) {
    let _Auxiliar = toscalar(
        SecurityAlert
        | where TimeGenerated between (start_time .. end_time)
        | make-series Count = count() default=0 on TimeGenerated step scan_step
        | extend
            TimeGenerated = array_slice(TimeGenerated, 0, toint(-(consecutive_failures_threshold * query_frequency / scan_step))),
            Count = array_slice(Count, 0, toint(-(consecutive_failures_threshold * query_frequency / scan_step)))
        | extend series_periods_detect(
            Count,
            0.0,
            toint(24h / scan_step),
            1)
        | summarize Period = take_any(toint(series_periods_detect_Count_periods[0]))
        );
    let _PeriodStep = scan_step * abs(coalesce(_Auxiliar, 1));
    SecurityAlert
    | where TimeGenerated between (start_time .. end_time)
    // _PeriodStep cannot be used with make-series, so summarize has to be used instead
    // | make-series Count = count() default=0 on TimeGenerated step _PeriodStep
    // summarize does not generate zero values for count(), baseline noise has to be added, it will be "deleted" afterwards
    | union (range TimeGenerated from start_time to end_time step _PeriodStep)
    | summarize Count = count() by bin_at(TimeGenerated, _PeriodStep, end_time)
    | extend Count = Count - 1
    | summarize TimeGenerated = make_list(TimeGenerated), Count = make_list(Count)
    | extend
        TimeGenerated = array_slice(TimeGenerated, 0, -2),
        Count = array_slice(Count, 0, -2)
    | extend series_decompose_anomalies(Count)
    | where array_sum(array_slice(series_decompose_anomalies_Count_ad_flag, -(consecutive_failures_threshold), -1)) == (-1 * consecutive_failures_threshold)
};
_SecurityAlertTimeSeries(ago(query_period), now())
// Uncomment the following line if you want only one alert (during the first query_frequency of the anomaly)
| where not(toscalar(_SecurityAlertTimeSeries(ago(query_period), ago(query_frequency)) | count) > 0)
| render timechart

Explanation

This KQL query is designed to detect anomalies in security alerts over a specified period. Here's a simplified explanation:

Parameters Setup:
- query_frequency: The frequency at which the query is run (1 hour).
- query_period: The total time period over which the data is analyzed (7 days).
- scan_step: The granularity of the time intervals for analysis (5 minutes).
- consecutive_failures_threshold: The number of consecutive anomalies needed to trigger an alert (2).
Data Preparation:
- The query defines a function _SecurityAlertTimeSeries that takes a start and end time as inputs.
- It calculates the number of security alerts generated within the specified time frame using a time series analysis.
- The data is sliced to remove the last few intervals based on the consecutive_failures_threshold.
Anomaly Detection:
- The query uses series_periods_detect to identify periodic patterns in the alert counts.
- It calculates a step size (_PeriodStep) based on the detected period and uses it to bin the data.
- A baseline noise is added to ensure that zero values are handled correctly.
Anomaly Analysis:
- The query uses series_decompose_anomalies to detect anomalies in the alert counts.
- It checks for consecutive anomalies (as defined by consecutive_failures_threshold) and flags them.
Output:
- The function _SecurityAlertTimeSeries is called to analyze data from the past 7 days up to the current time.
- Optionally, a line can be uncommented to filter out alerts that have already been detected in the last hour.
- The results are rendered as a time chart for visualization.

In summary, this query identifies periods with unusual spikes in security alerts over the past week, highlighting potential security incidents that may require further investigation.