如何基于CSV文件的UPN列表编写KQL查询获取指定MFA登录日志

我有一个包含50000个用户主体名称（UPN）的CSV文件，需要提取这些UPN对应的、通过指定多因素认证（MFA）条件访问策略的至少一次成功登录日志。目前已编写初步KQL查询，但不知道如何针对CSV中的UPN列表进行查询，同时使用project算子指定列的效果未达预期。已知Azure Log Analytics有30000条记录上限，计划将CSV拆分为更小批次处理。我是KQL新手，仅学习数日。

已尝试的查询代码 #

SigninLogs
| mvexpand ConditionalAccessPolicies
| where TimeGenerated <= ago(13d)
| where ConditionalAccessPolicies.id == "4603056e-3712-4d0f-8d51-59ec0bd66261"
| where ConditionalAccessStatus == "success"
| where UserPrincipalName contains "@my.tccd.edu"
| project TimeGenerated, UserPrincipalName, AppDisplayName, ConditionalAccessPolicies.displayName
| summarize count=count() by UserPrincipalName

1. 针对CSV中的UPN列表查询 #

方式1：小批次静态UPN列表（适合手动拆分的小批量） #

把拆分后的批次UPN整理成动态数组，用in操作符精准匹配：

// 替换为当前批次的UPN列表
let targetUPNs = dynamic(["user1@my.tccd.edu", "user2@my.tccd.edu", "user3@my.tccd.edu"]);
SigninLogs
| mvexpand ConditionalAccessPolicies
| where TimeGenerated <= ago(13d)
| where ConditionalAccessPolicies.id == "4603056e-3712-4d0f-8d51-59ec0bd66261"
| where ConditionalAccessStatus == "success"
// 精准匹配目标UPN，替换原有的contains条件
| where UserPrincipalName in (targetUPNs)
// 给嵌套的策略名称起别名，避免显示问题
| project TimeGenerated, UserPrincipalName, AppDisplayName, CAPolicyName = ConditionalAccessPolicies.displayName
// 聚合出每个用户是否有至少一次成功登录
| summarize HasSuccessfulMFA = count() > 0 by UserPrincipalName, CAPolicyName

方式2：从存储账户读取CSV批次文件（适合自动化批量处理） #

将拆分后的CSV上传到Azure存储账户，用externaldata算子读取：

// 读取存储账户中的UPN批次文件，需配置存储账户权限
let targetUPNs = externaldata(UserPrincipalName:string)
[
    h@"https://your-storage-account.blob.core.windows.net/your-container/upn-batch-1.csv"
] with (format="csv", ignoreFirstRecord=true);
SigninLogs
| mvexpand ConditionalAccessPolicies
| where TimeGenerated <= ago(13d)
| where ConditionalAccessPolicies.id == "4603056e-3712-4d0f-8d51-59ec0bd66261"
| where ConditionalAccessStatus == "success"
// 内连接匹配目标UPN
| join kind=inner targetUPNs on UserPrincipalName
| project TimeGenerated, UserPrincipalName, AppDisplayName, CAPolicyName = ConditionalAccessPolicies.displayName
| summarize HasSuccessfulMFA = count() > 0 by UserPrincipalName, CAPolicyName

2. 解决project算子的显示问题 #

原查询中ConditionalAccessPolicies.displayName是嵌套字段，直接使用可能导致列显示异常，给它起一个明确的别名（比如CAPolicyName）即可解决。如果需要保留原始登录记录而非聚合结果，去掉summarize语句即可：

let targetUPNs = dynamic(["user1@my.tccd.edu"]);
SigninLogs
| mvexpand ConditionalAccessPolicies
| where TimeGenerated <= ago(13d)
| where ConditionalAccessPolicies.id == "4603056e-3712-4d0f-8d51-59ec0bd66261"
| where ConditionalAccessStatus == "success"
| where UserPrincipalName in (targetUPNs)
| project TimeGenerated, UserPrincipalName, AppDisplayName, CAPolicyName = ConditionalAccessPolicies.displayName

3. 处理30000条记录上限 #

• 将CSV拆分为每批次20000条以内的UPN，避免查询返回的原始日志记录数超出上限。
• 使用summarize聚合用户的登录状态（是否有成功记录），可以大幅减少返回的结果行数，避免触发记录上限。

内容的提问来源于stack exchange，提问作者MK11-DRM