Stored Data Manipulation — Unauthorized Bulk Modification of Production Database Records

// THREAT: Stored data manipulation - anomalous bulk UPDATE/DELETE volume + service-account interactive tooling (T1565.001)
// Requires Azure SQL Database Auditing ingested into AzureDiagnostics (category SQLSecurityAuditEvents)
let LookbackWindow = 1h;
let BulkWriteThreshold = 100; // UPDATE/DELETE statements from one principal in the window
let KnownServiceAccounts = dynamic(["svc_appserver", "svc_etl", "svc_reporting"]); // populate with legitimate application/ETL service principals
// Signal 1: bulk UPDATE/DELETE volume spike from a single principal against production tables
let BulkWriteSpike = AzureDiagnostics
| where Category == "SQLSecurityAuditEvents"
| where TimeGenerated > ago(LookbackWindow)
| where action_name_s in ("UPDATE", "DELETE", "BATCH UPDATE", "BATCH DELETE")
| summarize WriteCount=count(), TablesTouched=make_set(object_name_s, 20) by server_principal_name_s, database_name_s, bin(TimeGenerated, 15m)
| where WriteCount >= BulkWriteThreshold
| extend Indicator = "BulkUpdateDeleteVolumeSpike"
| extend RiskScore = iff(server_principal_name_s !in~ (KnownServiceAccounts), 90, 70)
| project TimeGenerated, server_principal_name_s, database_name_s, WriteCount, TablesTouched, Indicator, RiskScore;
// Signal 2: service account launching an interactive DB query tool (indicates manual, off-process access)
let ServiceAccountInteractiveTool = DeviceProcessEvents
| where Timestamp > ago(LookbackWindow)
| where FileName in~ ("ssms.exe", "azuredatastudio.exe", "mysql.exe", "psql.exe", "dbeaver.exe")
| where AccountName in~ (KnownServiceAccounts)
| extend Indicator = "ServiceAccountLaunchedInteractiveQueryTool"
| extend RiskScore = 85
| project Timestamp, DeviceName, AccountName, FileName, ProcessCommandLine, Indicator, RiskScore;
union BulkWriteSpike, ServiceAccountInteractiveTool
| sort by RiskScore desc, TimeGenerated desc

(
  index=azure sourcetype="azure:diagnostics" Category="SQLSecurityAuditEvents"
  action_name IN ("UPDATE","DELETE","BATCH UPDATE","BATCH DELETE")
  | bucket _time span=15m
  | stats count AS WriteCount, values(object_name) AS TablesTouched by server_principal_name, database_name, _time
  | where WriteCount >= 100
  | eval Indicator="BulkUpdateDeleteVolumeSpike"
  | eval RiskScore=if(server_principal_name IN ("svc_appserver","svc_etl","svc_reporting"), 70, 90)
  | table _time, server_principal_name, database_name, WriteCount, TablesTouched, Indicator, RiskScore
)
| append [
    search index=mde sourcetype="DeviceProcessEvents"
    FileName IN ("ssms.exe","azuredatastudio.exe","mysql.exe","psql.exe","dbeaver.exe")
    AccountName IN ("svc_appserver","svc_etl","svc_reporting")
    | eval Indicator="ServiceAccountLaunchedInteractiveQueryTool", RiskScore=85
    | table _time, DeviceName, AccountName, FileName, ProcessCommandLine, Indicator, RiskScore
  ]
| sort - RiskScore

sequence by process.parent.user.name with maxspan=1h
  [process where event.type == "start" and process.name : ("ssms.exe", "azuredatastudio.exe", "mysql.exe", "psql.exe", "dbeaver.exe")]

SELECT
  DATEFORMAT(starttime, 'YYYY-MM-dd HH:mm:ss') AS EventTime,
  username AS Principal,
  UTF8(payload) AS RawEvent,
  COUNT(*) OVER (PARTITION BY username) AS WriteCount
FROM events
WHERE
  LOGSOURCETYPEID IN (400, 401)
  AND (UTF8(payload) ILIKE '%UPDATE%' OR UTF8(payload) ILIKE '%DELETE%')
  AND starttime > NOW() - 3600000
GROUP BY username, starttime, payload
HAVING COUNT(*) >= 100
ORDER BY WriteCount DESC

_sourceCategory=*/Azure/SQLAudit*
| parse field=_raw "action_name: *\r" as ActionName nodrop
| parse field=_raw "server_principal_name: *\r" as Principal nodrop
| parse field=_raw "database_name: *\r" as DatabaseName nodrop
| where (ActionName = "UPDATE" or ActionName = "DELETE" or ActionName = "BATCH UPDATE" or ActionName = "BATCH DELETE")
| timeslice 15m
| stats count as WriteCount by Principal, DatabaseName, _timeslice
| where WriteCount >= 100
| eval Indicator = "BulkUpdateDeleteVolumeSpike"
| sort - WriteCount

rule stored_data_manipulation_bulk_writes {
  meta:
    author = "df00tech Detection Engineering"
    description = "Detects anomalous bulk UPDATE/DELETE volume against production database tables by a single principal"
    severity = "HIGH"
    priority = "HIGH"
    mitre_attack_tactic = "Impact"
    mitre_attack_technique = "T1565.001"
    created = "2026-07-18"

  events:
    $db.metadata.event_type = "DATABASE_UPDATE" or $db.metadata.event_type = "DATABASE_DELETE"
    $db.principal.user.userid = $principal

  match:
    $principal over 15m

  condition:
    #db >= 100
}

#event_simpleName=ProcessRollup2
| FileName = /(?i)(ssms|azuredatastudio|mysql|psql|dbeaver)(\.exe)?$/
| UserName = /(?i)^svc_/
| groupBy([ComputerName, UserName, FileName], function=[count(as=EventCount)])
| sort(EventCount, order=desc)