Outgoing connection attempts stateful anomaly on database

let monitoredStatementsThreshold = 1;           // Minimal number of monitored statements in the slice to trigger an anomaly.
let trainingSlicesThreshold = 5;                // The maximal amount of slices with monitored statements in the training window before anomaly detection is throttled.
let timeSliceSize = 1h;                         // The size of the single timeSlice for individual aggregation.
let detectionWindow = 1h;                       // The size of the recent detection window for detecting anomalies.  
let trainingWindow = detectionWindow + 14d;     // The size of the training window before the detection window for learning the normal state.
let hotwords = pack_array('http:', 'https:', 'ftp:', 'onion.pet'); // List of monitored hot words.
let processedData = materialize (
    AzureDiagnostics
    | where TimeGenerated >= ago(trainingWindow)
    | where Category == 'SQLSecurityAuditEvents' and action_id_s has_any ("RCM", "BCM") // Keep only SQL affected rows
    | project TimeGenerated, PrincipalName = server_principal_name_s, ClientIp = client_ip_s, HostName = host_name_s, ResourceId,
              ApplicationName = application_name_s, ActionName = action_name_s, Database = strcat(LogicalServerName_s, '/', database_name_s),
              IsSuccess = succeeded_s, AffectedRows = affected_rows_d,
              ResponseRows = response_rows_d, Statement = statement_s,
              Error = case( additional_information_s has 'error_code', toint(extract("<error_code>([0-9.]+)", 1, additional_information_s))
                    , additional_information_s has 'failure_reason', toint(extract("<failure_reason>Err ([0-9.]+)", 1, additional_information_s))
                    , 0),
              State = case( additional_information_s has 'error_state', toint(extract("<error_state>([0-9.]+)", 1, additional_information_s))
                    , additional_information_s has 'failure_reason', toint(extract("<failure_reason>Err ([0-9.]+), Level ([0-9.]+)", 2, additional_information_s))
                    , 0),
              AdditionalInfo = additional_information_s, timeSlice = floor(TimeGenerated, timeSliceSize)
    | extend hasHotword = iff(Statement has_any (hotwords), 1, 0)
    | summarize countEvents = count(), countStatements = dcount(Statement)
        , countStatementsWithHotwords = dcountif(Statement, hasHotword == 1)
        , countFailedStatementsWithHotwords = dcountif(Statement, (hasHotword == 1) and (Error > 0))
        , countSuccessfulStatementsWithHotwords = dcountif(Statement, ((hasHotword == 1)) and (Error == 0))
        , anyMonitoredStatement = anyif(Statement, (hasHotword == 1))
        , anySuccessfulMonitoredStatement = anyif(Statement, (hasHotword == 1) and (Error == 0))
        , anyInfo = anyif(AdditionalInfo, hasHotword == 1)
        , hotWord = anyif(extract(strcat_array(hotwords, '|'), 0, tolower(Statement)), hasHotword == 1)
        by Database, ClientIp, ApplicationName, PrincipalName, timeSlice,HostName,ResourceId   
    | extend WindowType = case( timeSlice >= ago(detectionWindow), 'detection',
                                           (ago(trainingWindow) <= timeSlice and timeSlice < ago(detectionWindow)), 'training', 'other')
    | where WindowType in ('detection', 'training'));
let trainingSet =
    processedData
    | where WindowType == 'training'
    | summarize countSlicesWithHotwords = dcountif(timeSlice, countStatementsWithHotwords >= monitoredStatementsThreshold)
        by Database;
processedData
| where WindowType == 'detection' 
| join kind = inner (trainingSet) on Database
| extend IsHotwordAnomalyOnStatement = iff(((countStatementsWithHotwords >= monitoredStatementsThreshold) and (countSlicesWithHotwords <= trainingSlicesThreshold)), true, false)
    , anomalyScore = round(countStatementsWithHotwords/monitoredStatementsThreshold, 0)
| where IsHotwordAnomalyOnStatement == 'true'
| sort by countSuccessfulStatementsWithHotwords desc,  anomalyScore desc, timeSlice desc
| extend Name = tostring(split(PrincipalName,'@',0)[0]), UPNSuffix = tostring(split(PrincipalName,'@',1)[0])    

entityMappings:
- entityType: Account
  fieldMappings:
  - columnName: Name
    identifier: Name
  - columnName: UPNSuffix
    identifier: UPNSuffix
- entityType: IP
  fieldMappings:
  - columnName: ClientIp
    identifier: Address
- entityType: Host
  fieldMappings:
  - columnName: HostName
    identifier: HostName
- entityType: CloudApplication
  fieldMappings:
  - columnName: ApplicationName
    identifier: Name
- entityType: AzureResource
  fieldMappings:
  - columnName: ResourceId
    identifier: ResourceId
requiredDataConnectors:
- dataTypes:
  - AzureDiagnostics
  connectorId: AzureSql
queryPeriod: 14d
query: |
  let monitoredStatementsThreshold = 1;           // Minimal number of monitored statements in the slice to trigger an anomaly.
  let trainingSlicesThreshold = 5;                // The maximal amount of slices with monitored statements in the training window before anomaly detection is throttled.
  let timeSliceSize = 1h;                         // The size of the single timeSlice for individual aggregation.
  let detectionWindow = 1h;                       // The size of the recent detection window for detecting anomalies.  
  let trainingWindow = detectionWindow + 14d;     // The size of the training window before the detection window for learning the normal state.
  let hotwords = pack_array('http:', 'https:', 'ftp:', 'onion.pet'); // List of monitored hot words.
  let processedData = materialize (
      AzureDiagnostics
      | where TimeGenerated >= ago(trainingWindow)
      | where Category == 'SQLSecurityAuditEvents' and action_id_s has_any ("RCM", "BCM") // Keep only SQL affected rows
      | project TimeGenerated, PrincipalName = server_principal_name_s, ClientIp = client_ip_s, HostName = host_name_s, ResourceId,
                ApplicationName = application_name_s, ActionName = action_name_s, Database = strcat(LogicalServerName_s, '/', database_name_s),
                IsSuccess = succeeded_s, AffectedRows = affected_rows_d,
                ResponseRows = response_rows_d, Statement = statement_s,
                Error = case( additional_information_s has 'error_code', toint(extract("<error_code>([0-9.]+)", 1, additional_information_s))
                      , additional_information_s has 'failure_reason', toint(extract("<failure_reason>Err ([0-9.]+)", 1, additional_information_s))
                      , 0),
                State = case( additional_information_s has 'error_state', toint(extract("<error_state>([0-9.]+)", 1, additional_information_s))
                      , additional_information_s has 'failure_reason', toint(extract("<failure_reason>Err ([0-9.]+), Level ([0-9.]+)", 2, additional_information_s))
                      , 0),
                AdditionalInfo = additional_information_s, timeSlice = floor(TimeGenerated, timeSliceSize)
      | extend hasHotword = iff(Statement has_any (hotwords), 1, 0)
      | summarize countEvents = count(), countStatements = dcount(Statement)
          , countStatementsWithHotwords = dcountif(Statement, hasHotword == 1)
          , countFailedStatementsWithHotwords = dcountif(Statement, (hasHotword == 1) and (Error > 0))
          , countSuccessfulStatementsWithHotwords = dcountif(Statement, ((hasHotword == 1)) and (Error == 0))
          , anyMonitoredStatement = anyif(Statement, (hasHotword == 1))
          , anySuccessfulMonitoredStatement = anyif(Statement, (hasHotword == 1) and (Error == 0))
          , anyInfo = anyif(AdditionalInfo, hasHotword == 1)
          , hotWord = anyif(extract(strcat_array(hotwords, '|'), 0, tolower(Statement)), hasHotword == 1)
          by Database, ClientIp, ApplicationName, PrincipalName, timeSlice,HostName,ResourceId   
      | extend WindowType = case( timeSlice >= ago(detectionWindow), 'detection',
                                             (ago(trainingWindow) <= timeSlice and timeSlice < ago(detectionWindow)), 'training', 'other')
      | where WindowType in ('detection', 'training'));
  let trainingSet =
      processedData
      | where WindowType == 'training'
      | summarize countSlicesWithHotwords = dcountif(timeSlice, countStatementsWithHotwords >= monitoredStatementsThreshold)
          by Database;
  processedData
  | where WindowType == 'detection' 
  | join kind = inner (trainingSet) on Database
  | extend IsHotwordAnomalyOnStatement = iff(((countStatementsWithHotwords >= monitoredStatementsThreshold) and (countSlicesWithHotwords <= trainingSlicesThreshold)), true, false)
      , anomalyScore = round(countStatementsWithHotwords/monitoredStatementsThreshold, 0)
  | where IsHotwordAnomalyOnStatement == 'true'
  | sort by countSuccessfulStatementsWithHotwords desc,  anomalyScore desc, timeSlice desc
  | extend Name = tostring(split(PrincipalName,'@',0)[0]), UPNSuffix = tostring(split(PrincipalName,'@',1)[0])      
description: |
    'This query detects batches of distinct SQL queries that execute (or attempt to) commands that could indicate potential security issues - such as attempts to access external sites or resources (e.g. for downloading malicious content).'
version: 1.1.1
name: Outgoing connection attempts stateful anomaly on database
relevantTechniques:
- T1190
tags:
- SQL
OriginalUri: https://github.com/Azure/Azure-Sentinel/blob/master/Solutions/Azure SQL Database solution for sentinel/Analytic Rules/Detection-HotwordsOutgoingStatefulAnomalyOnDatabase.yaml
kind: Scheduled
status: Available
queryFrequency: 1h
severity: Medium
tactics:
- InitialAccess
triggerOperator: gt
id: c105513d-e398-4a02-bd91-54b9b2d6fa7d
triggerThreshold: 0

{
  "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
  "contentVersion": "1.0.0.0",
  "parameters": {
    "workspace": {
      "type": "String"
    }
  },
  "resources": [
    {
      "apiVersion": "2023-02-01-preview",
      "id": "[concat(resourceId('Microsoft.OperationalInsights/workspaces/providers', parameters('workspace'), 'Microsoft.SecurityInsights'),'/alertRules/c105513d-e398-4a02-bd91-54b9b2d6fa7d')]",
      "kind": "Scheduled",
      "name": "[concat(parameters('workspace'),'/Microsoft.SecurityInsights/c105513d-e398-4a02-bd91-54b9b2d6fa7d')]",
      "properties": {
        "alertRuleTemplateName": "c105513d-e398-4a02-bd91-54b9b2d6fa7d",
        "customDetails": null,
        "description": "'This query detects batches of distinct SQL queries that execute (or attempt to) commands that could indicate potential security issues - such as attempts to access external sites or resources (e.g. for downloading malicious content).'\n",
        "displayName": "Outgoing connection attempts stateful anomaly on database",
        "enabled": true,
        "entityMappings": [
          {
            "entityType": "Account",
            "fieldMappings": [
              {
                "columnName": "Name",
                "identifier": "Name"
              },
              {
                "columnName": "UPNSuffix",
                "identifier": "UPNSuffix"
              }
            ]
          },
          {
            "entityType": "IP",
            "fieldMappings": [
              {
                "columnName": "ClientIp",
                "identifier": "Address"
              }
            ]
          },
          {
            "entityType": "Host",
            "fieldMappings": [
              {
                "columnName": "HostName",
                "identifier": "HostName"
              }
            ]
          },
          {
            "entityType": "CloudApplication",
            "fieldMappings": [
              {
                "columnName": "ApplicationName",
                "identifier": "Name"
              }
            ]
          },
          {
            "entityType": "AzureResource",
            "fieldMappings": [
              {
                "columnName": "ResourceId",
                "identifier": "ResourceId"
              }
            ]
          }
        ],
        "OriginalUri": "https://github.com/Azure/Azure-Sentinel/blob/master/Solutions/Azure SQL Database solution for sentinel/Analytic Rules/Detection-HotwordsOutgoingStatefulAnomalyOnDatabase.yaml",
        "query": "let monitoredStatementsThreshold = 1;           // Minimal number of monitored statements in the slice to trigger an anomaly.\nlet trainingSlicesThreshold = 5;                // The maximal amount of slices with monitored statements in the training window before anomaly detection is throttled.\nlet timeSliceSize = 1h;                         // The size of the single timeSlice for individual aggregation.\nlet detectionWindow = 1h;                       // The size of the recent detection window for detecting anomalies.  \nlet trainingWindow = detectionWindow + 14d;     // The size of the training window before the detection window for learning the normal state.\nlet hotwords = pack_array('http:', 'https:', 'ftp:', 'onion.pet'); // List of monitored hot words.\nlet processedData = materialize (\n    AzureDiagnostics\n    | where TimeGenerated >= ago(trainingWindow)\n    | where Category == 'SQLSecurityAuditEvents' and action_id_s has_any (\"RCM\", \"BCM\") // Keep only SQL affected rows\n    | project TimeGenerated, PrincipalName = server_principal_name_s, ClientIp = client_ip_s, HostName = host_name_s, ResourceId,\n              ApplicationName = application_name_s, ActionName = action_name_s, Database = strcat(LogicalServerName_s, '/', database_name_s),\n              IsSuccess = succeeded_s, AffectedRows = affected_rows_d,\n              ResponseRows = response_rows_d, Statement = statement_s,\n              Error = case( additional_information_s has 'error_code', toint(extract(\"<error_code>([0-9.]+)\", 1, additional_information_s))\n                    , additional_information_s has 'failure_reason', toint(extract(\"<failure_reason>Err ([0-9.]+)\", 1, additional_information_s))\n                    , 0),\n              State = case( additional_information_s has 'error_state', toint(extract(\"<error_state>([0-9.]+)\", 1, additional_information_s))\n                    , additional_information_s has 'failure_reason', toint(extract(\"<failure_reason>Err ([0-9.]+), Level ([0-9.]+)\", 2, additional_information_s))\n                    , 0),\n              AdditionalInfo = additional_information_s, timeSlice = floor(TimeGenerated, timeSliceSize)\n    | extend hasHotword = iff(Statement has_any (hotwords), 1, 0)\n    | summarize countEvents = count(), countStatements = dcount(Statement)\n        , countStatementsWithHotwords = dcountif(Statement, hasHotword == 1)\n        , countFailedStatementsWithHotwords = dcountif(Statement, (hasHotword == 1) and (Error > 0))\n        , countSuccessfulStatementsWithHotwords = dcountif(Statement, ((hasHotword == 1)) and (Error == 0))\n        , anyMonitoredStatement = anyif(Statement, (hasHotword == 1))\n        , anySuccessfulMonitoredStatement = anyif(Statement, (hasHotword == 1) and (Error == 0))\n        , anyInfo = anyif(AdditionalInfo, hasHotword == 1)\n        , hotWord = anyif(extract(strcat_array(hotwords, '|'), 0, tolower(Statement)), hasHotword == 1)\n        by Database, ClientIp, ApplicationName, PrincipalName, timeSlice,HostName,ResourceId   \n    | extend WindowType = case( timeSlice >= ago(detectionWindow), 'detection',\n                                           (ago(trainingWindow) <= timeSlice and timeSlice < ago(detectionWindow)), 'training', 'other')\n    | where WindowType in ('detection', 'training'));\nlet trainingSet =\n    processedData\n    | where WindowType == 'training'\n    | summarize countSlicesWithHotwords = dcountif(timeSlice, countStatementsWithHotwords >= monitoredStatementsThreshold)\n        by Database;\nprocessedData\n| where WindowType == 'detection' \n| join kind = inner (trainingSet) on Database\n| extend IsHotwordAnomalyOnStatement = iff(((countStatementsWithHotwords >= monitoredStatementsThreshold) and (countSlicesWithHotwords <= trainingSlicesThreshold)), true, false)\n    , anomalyScore = round(countStatementsWithHotwords/monitoredStatementsThreshold, 0)\n| where IsHotwordAnomalyOnStatement == 'true'\n| sort by countSuccessfulStatementsWithHotwords desc,  anomalyScore desc, timeSlice desc\n| extend Name = tostring(split(PrincipalName,'@',0)[0]), UPNSuffix = tostring(split(PrincipalName,'@',1)[0])    \n",
        "queryFrequency": "PT1H",
        "queryPeriod": "P14D",
        "severity": "Medium",
        "status": "Available",
        "suppressionDuration": "PT1H",
        "suppressionEnabled": false,
        "tactics": [
          "InitialAccess"
        ],
        "tags": [
          "SQL"
        ],
        "techniques": [
          "T1190"
        ],
        "templateVersion": "1.1.1",
        "triggerOperator": "GreaterThan",
        "triggerThreshold": 0
      },
      "type": "Microsoft.OperationalInsights/workspaces/providers/alertRules"
    }
  ]
}