Detection rules › Kusto
TI Map URL Entity to EmailUrlInfo
This query identifies any URL indicators of compromise (IOCs) from threat intelligence (TI) by searching for matches in EmailUrlInfo.
MITRE ATT&CK coverage
| Tactic | Techniques |
|---|---|
| Command & Control |
Telemetry coverage
Rule body
id: 9e32e545-e60c-47de-9941-f9ca1ada0a42
name: TI Map URL Entity to EmailUrlInfo
description: |
'This query identifies any URL indicators of compromise (IOCs) from threat intelligence (TI) by searching for matches in EmailUrlInfo.'
severity: Medium
requiredDataConnectors:
- connectorId: MicrosoftThreatProtection
dataTypes:
- EmailUrlInfo
- connectorId: ThreatIntelligence
dataTypes:
- ThreatIntelIndicators
- connectorId: ThreatIntelligenceTaxii
dataTypes:
- ThreatIntelIndicators
- connectorId: MicrosoftDefenderThreatIntelligence
dataTypes:
- ThreatIntelIndicators
queryFrequency: 1h
queryPeriod: 14d
triggerOperator: gt
triggerThreshold: 0
tactics:
- CommandAndControl
relevantTechniques:
- T1071
query: |
let dt_lookBack = 1h;
let ioc_lookBack = 14d;
let EmailUrlInfo_ = materialize(EmailUrlInfo
| where TimeGenerated >= ago(dt_lookBack)
| extend EmailUrlInfo_TimeGenerated = TimeGenerated);
let EmailUrls = EmailUrlInfo_ | distinct Url | summarize make_list(Url);
let EmailUrlDomains = EmailUrlInfo_ | distinct UrlDomain | summarize make_list(UrlDomain);
let EmailEvents_ = materialize(EmailEvents
| where TimeGenerated >= ago(dt_lookBack));
let TI = materialize(ThreatIntelIndicators
//extract key part of kv pair
| extend IndicatorType = replace(@"\[|\]|\""", "", tostring(split(ObservableKey, ":", 0)))
| where IndicatorType == "url"
| extend Url = ObservableValue
| extend TrafficLightProtocolLevel = tostring(parse_json(AdditionalFields).TLPLevel)
| extend IndicatorId = tostring(split(Id, "--")[2])
| where TimeGenerated >= ago(ioc_lookBack)
| where tolower(Url) in (EmailUrls) or tolower(Url) in (EmailUrlDomains)
| summarize LatestIndicatorTime = arg_max(TimeGenerated, *) by Id, ObservableValue
| where IsActive and (ValidUntil > now() or isempty(ValidUntil)));
(union
(TI | join kind=innerunique (EmailUrlInfo_) on Url),
(TI | join kind=innerunique (EmailUrlInfo_) on $left.Url == $right.UrlDomain))
| where EmailUrlInfo_TimeGenerated < ValidUntil
| summarize EmailUrlInfo_TimeGenerated = arg_max(EmailUrlInfo_TimeGenerated, *) by IndicatorId, Url
| extend Description = tostring(parse_json(Data).description)
| extend ActivityGroupNames = extract(@"ActivityGroup:(\S+)", 1, tostring(parse_json(Data).labels))
| project EmailUrlInfo_TimeGenerated, Description, ActivityGroupNames, IndicatorId, Type, ValidUntil, Confidence, Url, UrlDomain, UrlLocation, NetworkMessageId
| extend timestamp = EmailUrlInfo_TimeGenerated
| join kind=inner (EmailEvents_) on NetworkMessageId
| where DeliveryAction !has "Blocked"
| extend Name = tostring(split(RecipientEmailAddress, '@', 0)[0]), UPNSuffix = tostring(split(RecipientEmailAddress, '@', 1)[0])
entityMappings:
- entityType: Account
fieldMappings:
- identifier: FullName
columnName: RecipientEmailAddress
- identifier: Name
columnName: Name
- identifier: UPNSuffix
columnName: UPNSuffix
- entityType: URL
fieldMappings:
- identifier: Url
columnName: Url
version: 1.0.5
kind: Scheduled
Stages and Predicates
Parameters
let dt_lookBack = 1h;
let ioc_lookBack = 14d;
Let binding: EmailUrlInfo_
let EmailUrlInfo_ = materialize(EmailUrlInfo
| where TimeGenerated >= ago(dt_lookBack)
| extend EmailUrlInfo_TimeGenerated = TimeGenerated);
Let binding: EmailUrls
let EmailUrls = EmailUrlInfo_ | distinct Url | summarize make_list(Url);
Let binding: EmailUrlDomains
let EmailUrlDomains = EmailUrlInfo_ | distinct UrlDomain | summarize make_list(UrlDomain);
Let binding: EmailEvents_
let EmailEvents_ = materialize(EmailEvents
| where TimeGenerated >= ago(dt_lookBack));
union (2 sources)
Each leg below queries one source; the rule matches if any leg does. Sources: TI, TI
Leg 1: TI
TI | join kind=innerunique (EmailUrlInfo_) on Url
Leg 2: TI
TI | join kind=innerunique (EmailUrlInfo_) on $left.Url == $right.UrlDomain
Applied to the combined result
| where EmailUrlInfo_TimeGenerated < ValidUntil
| summarize EmailUrlInfo_TimeGenerated = arg_max(EmailUrlInfo_TimeGenerated, *) by IndicatorId, Url
| extend Description = tostring(parse_json(Data).description)
| extend ActivityGroupNames = extract(@"ActivityGroup:(\S+)", 1, tostring(parse_json(Data).labels))
| project EmailUrlInfo_TimeGenerated, Description, ActivityGroupNames, IndicatorId, Type, ValidUntil, Confidence, Url, UrlDomain, UrlLocation, NetworkMessageId
| extend timestamp = EmailUrlInfo_TimeGenerated
| join kind=inner (EmailEvents_) on NetworkMessageId
| where DeliveryAction !has "Blocked"
| extend Name = tostring(split(RecipientEmailAddress, '@', 0)[0]), UPNSuffix = tostring(split(RecipientEmailAddress, '@', 1)[0])
Exclusions
The rule actively suppresses these predicates.
| Field | Kind | Excluded values | Search |
|---|---|---|---|
DeliveryAction | match | Blocked | excludes:DeliveryAction field:"DeliveryAction" value:"Blocked" |
Indicators
These rows show field, operator, and value matches.
| Field | Kind | Values | Search |
|---|---|---|---|
EmailUrlInfo_TimeGenerated | cross_field_compare |
| field:"EmailUrlInfo_TimeGenerated" kind:cross_field_compare value:"ValidUntil" |
IndicatorType | eq |
| field:"IndicatorType" kind:eq value:"url" |
Url | in |
| field:"Url" kind:in |
ValidUntil | is_null | field:"ValidUntil" kind:is_null |
Output fields
These fields are emitted when the rule matches.
| Field | Source |
|---|---|
ActivityGroupNames | project |
Confidence | project |
Description | project |
EmailUrlInfo_TimeGenerated | project |
IndicatorId | project |
NetworkMessageId | project |
Type | project |
Url | project |
UrlDomain | project |
UrlLocation | project |
ValidUntil | project |
timestamp | extend |
Name | extend |
UPNSuffix | extend |