azure-kusto-graph

Kusto 图语义

使用 KQL 图运算符从表格数据构建临时图和持久化图。此技能将自然语言转换为以边为先的图构建模式和图查询运算符。

激活条件

在以下情况下使用此技能：

• 想从表格数据构建图（make-graph）
• 想在数据中查找模式、路径或关系
• 提及 graph-match、graph-shortest-paths、graph-to-table、graph-mark-components
• 想创建持久化图模型或快照
• 说“构建图”“查找最短路径”“查找连通分量”“显示关系”
• 询问临时图与持久化图的区别

这不是自然语言到 KQL 的转换器。 输入通常应是一个可运行的 KQL 查询，且用户希望将其结果转换为图，并附上对所需图结构的自然语言描述。仅当基础自然语言数据源请求可直接映射到具有明显列的已知表时，才支持此类请求。对于一般的自然语言到 KQL 转换，请使用专用的查询生成技能（另行提供）。

互补技能：

• azure-kusto-irql -- 可组合的安全查询原语，用于生成图的表格输入
• azure-kusto-irql-graph -- IRQL 的 Lift_To_Graph JSON 映射系统，用于在 Kusto Explorer 中创建类型丰富、带图标的图

以边为先的方法

在 Kusto 中构建图的基本模式：

1. Define your EDGES       -> src --> dest, with relationship type/properties
2. Define your NODE LOOKUPS -> display names, types, properties for each node ID
3. Union edge types         -> if you have multiple relationship types
4. Union node lookups       -> if you have multiple node types
5. Call make-graph          -> edges | make-graph Source --> Target with nodes on nodeId

这是理解 make-graph 的方式。边表示你关注的关系。节点是查找表，为这些 ID 提供具象信息——显示名称、类型和属性。

图运算符参考

make-graph —— 从表构建图

Edges | make-graph SourceId --> TargetId with Nodes on NodeId

• Edges：表格数据源，每行表示一条边
• SourceId --> TargetId：包含源节点和目标节点 ID 的列
• with Nodes on NodeId：可选的节点属性表，按 ID 联接
• 支持多个节点表：with Nodes1 on Id1, Nodes2 on Id2
• 边中出现但节点表中缺失的节点，其属性为空

graph-match —— 查找模式

G | graph-match (a)-[e]->(b) where <constraints> project <output>

模式表示法：

元素	命名	匿名
节点	(n)	()
边（左到右）	-[e]->	-->
边（右到左）	<-[e]-	<--
任意方向	-[e]-	--
可变长度	-[e*1..5]->	-[*1..5]->

多跳模式：(a)-[e1]->(b)-[e2]->(c) 星形模式：(a)--(center)--(b), (c)--(center)--(d) 循环控制：cycles = all | none | unique_edges（默认值：unique_edges）

graph-shortest-paths —— 查找最短路径

G | graph-shortest-paths (start)-[e*1..20]->(end)
      where start.name == "Alice" and end.name == "Server01"
      project Path = e, Length = array_length(e)

• 至少需要一条可变长度边
• output = any（默认值，每对节点返回一条路径）或 output = all（返回所有长度相同的最短路径）
• 可变长度边的属性以动态数组形式返回

graph-to-table —— 将图导出为表

G | graph-to-table nodes                                     // export nodes
G | graph-to-table edges                                     // export edges
G | graph-to-table nodes as N, edges as E                    // export both
G | graph-to-table nodes with_node_id=Id                     // include node hash ID
G | graph-to-table edges with_source_id=Src with_target_id=Tgt  // include edge endpoint IDs

graph-mark-components —— 查找连通分量

G | graph-mark-components with_component_id=ComponentId
  | graph-to-table nodes
  | summarize Members = make_list(name) by ComponentId

为每个节点分配一个 ComponentId。同一连通分量中的节点共享相同的 ID。

graph() 函数 —— 查询持久化图

graph("MyGraphModel")                              // latest snapshot
graph("MyGraphModel", "Snapshot_2025_01")           // specific snapshot
graph("MyGraphModel", true)                         // transient from model definition

临时图

在查询执行过程中动态创建。无需设置。适用于临时分析、探索和原型设计。

模板：基本的双实体图

// 1. Define edges
let edges = <SourceTable>
    | summarize <aggregations> by SourceCol, TargetCol;
// 2. Define node lookups
let source_nodes = edges
    | distinct SourceCol
    | project nodeId = SourceCol, label = SourceCol, nodeType = "<SourceType>";
let target_nodes = edges
    | distinct TargetCol
    | project nodeId = TargetCol, label = TargetCol, nodeType = "<TargetType>";
let all_nodes = union source_nodes, target_nodes;
// 3. Build and query the graph
edges
| make-graph SourceCol --> TargetCol with all_nodes on nodeId
| graph-match (s)-[e]->(t)
    where <constraints>
    project Source = s.label, Target = t.label, <edge properties>

模板：多关系图

// Multiple edge types -> union them with a common schema
let auth_edges = AuthEvents
    | project Source = username, Target = hostname, edgeType = "authenticates", ts = timestamp;
let net_edges = NetworkEvents
    | project Source = src_ip, Target = url, edgeType = "connects", ts = timestamp;
let all_edges = union auth_edges, net_edges;
// Node lookups from all sources
let user_nodes = Employees | project nodeId = username, label = name, nodeType = "User";
let host_nodes = AuthEvents | distinct hostname | project nodeId = hostname, label = hostname, nodeType = "Host";
let all_nodes = union user_nodes, host_nodes;
all_edges
| make-graph Source --> Target with all_nodes on nodeId

持久化图

适用于大规模、可重复使用的图。存储在数据库元数据中。支持创建快照以进行历史比较。

安全提示： 创建或更改图模型和快照会修改数据库。执行 .create-or-alter graph_model 或 .make graph_snapshot 前，务必向用户展示确切命令并征得确认。

步骤 1：创建图模型

.create-or-alter graph_model SecurityGraph
{
  "Schema": {
    "Nodes": {
      "User": {"name": "string", "role": "string"},
      "Host": {"hostname": "string"},
      "IP":   {"ip": "string"}
    },
    "Edges": {
      "AuthenticatesTo": {"timestamp": "datetime", "result": "string"},
      "ConnectsFrom":    {"timestamp": "datetime"}
    }
  },
  "Definition": {
    "Steps": [
      {
        "Kind": "AddNodes",
        "Query": "Employees | project name, role",
        "NodeIdColumn": "name",
        "Labels": ["User"]
      },
      {
        "Kind": "AddNodes",
        "Query": "AuthenticationEvents | distinct hostname | project hostname",
        "NodeIdColumn": "hostname",
        "Labels": ["Host"]
      },
      {
        "Kind": "AddEdges",
        "Query": "AuthenticationEvents | project username, hostname, timestamp, result",
        "SourceColumn": "username",
        "TargetColumn": "hostname",
        "Labels": ["AuthenticatesTo"]
      }
    ]
  }
}

步骤 2：创建快照

.make graph_snapshot SecurityGraph Snapshot_2025_07

步骤 3：查询快照

graph("SecurityGraph")
| graph-match (user)-[auth]->(host)
    where user.role == "Admin" and auth.result == "Failed Login"
    project User = user.name, Host = host.hostname, Time = auth.timestamp

管理命令

安全提示： 以下所有控制命令都会修改或删除数据库对象。切勿自动执行 .drop、.create-or-alter graph_model 或 .make graph_snapshot。始终展示确切的命令、群集、数据库和受影响的对象，然后要求用户明确确认后再执行。

.show graph_models                        // list all models
.show graph_model SecurityGraph           // show model details
.show graph_snapshots SecurityGraph       // list snapshots
.drop graph_snapshot SecurityGraph Snapshot_2025_07  // delete a snapshot (CONFIRM FIRST)
.drop graph_model SecurityGraph           // delete model and all snapshots (CONFIRM FIRST)

瞬态与持久化：何时使用哪种方式

因素	瞬态（make-graph）	持久化（graph()）
设置	无——在查询中内联定义	创建模型和快照
生命周期	仅在查询执行期间	存储在数据库元数据中
数据新鲜度	始终为最新	创建快照时的数据
规模	受查询内存限制	企业级规模
复用	每次查询都重新构建	在用户和查询之间共享
最适用于	临时搜寻、原型设计	生产工作流、仪表板

安全与威胁搜寻示例

身份验证图：谁从哪里登录了什么

let auth_edges = AuthenticationEvents
    | summarize
        logins = count(),
        fails = countif(result == "Failed Login")
      by src_ip, username, hostname;
let ip_nodes = auth_edges | distinct src_ip
    | project nodeId = src_ip, label = src_ip, nodeType = "IP";
let user_nodes = auth_edges | distinct username
    | project nodeId = username, label = username, nodeType = "User";
let host_nodes = auth_edges | distinct hostname
    | project nodeId = hostname, label = hostname, nodeType = "Host";
let all_nodes = union ip_nodes, user_nodes, host_nodes;
// IP -> User edges
let ip_user = auth_edges
    | project Source = src_ip, Target = username, logins, fails;
// User -> Host edges
let user_host = auth_edges
    | project Source = username, Target = hostname, logins, fails;
union ip_user, user_host
| make-graph Source --> Target with all_nodes on nodeId
| graph-match (ip)-[e1]->(user)-[e2]->(host)
    where e2.fails > 20
    project
        IP = ip.label,
        User = user.label,
        Host = host.label,
        Failures = e2.fails
| order by Failures desc

横向移动检测：共享受感染主机的用户

// Pattern: (user1)-[auth1]->(host)<-[auth2]-(user2)
// Two users both failing on the same host = possible credential spray
let edges = AuthenticationEvents
    | summarize fails = countif(result == "Failed Login"), logins = count()
      by username, hostname;
let nodes = union
    (edges | distinct username | project nodeId = username, nodeType = "User"),
    (edges | distinct hostname | project nodeId = hostname, nodeType = "Host");
edges
| make-graph username --> hostname with nodes on nodeId
| graph-match (u1)-[e1]->(h)<-[e2]-(u2)
    where u1.nodeId != u2.nodeId and e1.fails > 10 and e2.fails > 10
    project
        User1 = u1.nodeId, User2 = u2.nodeId,
        SharedHost = h.nodeId,
        User1Fails = e1.fails, User2Fails = e2.fails
| distinct User1, SharedHost, User2, User1Fails, User2Fails
| order by User1Fails + User2Fails desc

最短攻击路径

let edges = SecurityEvents
    | project Source = source_entity, Target = target_entity, action, timestamp;
let nodes = union
    (edges | distinct Source | project nodeId = Source),
    (edges | distinct Target | project nodeId = Target);
edges
| make-graph Source --> Target with nodes on nodeId
| graph-shortest-paths (start)-[e*1..10]->(end)
    where start.nodeId == "ExternalIP_1.2.3.4" and end.nodeId == "DatabaseServer"
    project
        PathLength = array_length(e),
        Actions = e.action,
        Hops = e.Target

连通分量：查找孤立的集群

let edges = NetworkFlows
    | project Source = src_ip, Target = dst_ip;
let nodes = union
    (edges | distinct Source | project nodeId = Source),
    (edges | distinct Target | project nodeId = Target);
edges
| make-graph Source --> Target with nodes on nodeId
| graph-mark-components with_component_id = ComponentId
| graph-to-table nodes
| summarize Members = make_list(nodeId), Size = count() by ComponentId
| order by Size desc

在 Kusto Explorer 中可视化

以 make-graph 结束查询（不通过管道传递到 graph-match），即可触发 Kusto Explorer 的交互式图可视化窗口：

edges
| make-graph Source --> Target with all_nodes on nodeId
// <- stop here. Kusto Explorer renders the graph visually.

要展平回表格以用于仪表板或导出，请通过管道传递到 graph-match | project 或 graph-to-table。

与 IRQL 配合使用

处理安全数据时，可以考虑使用 azure-kusto-irql 技能中的 IRQL 选择器（Get_*）作为数据源。IRQL 提供统一架构，无需记忆原始表名或列映射。若要使用带图标和节点折叠功能的丰富可视化，azure-kusto-irql-graph 技能中的 Lift_To_Graph 是更快捷的途径。

方式	最适用于
原始 make-graph（本技能）	完全控制、持久化模型、最短路径、连通分量、自定义架构
Lift_To_Graph（azure-kusto-irql-graph）	在 Kusto Explorer 中快速创建带图标的可视化、节点折叠
IRQL Get_* -> make-graph	以 IRQL 的统一架构为输入，再使用原始图运算符进行分析
IRQL Get_* -> Lift_To_Graph -> Graph_Render_View	从问题到可视化图的最快途径

注意： Lift_To_Graph、Graph_Render_View 和 Graph_Fold_By_Property 是存储函数，而非内置运算符。它们已预先部署在 kc7001 示例群集上，但可能需要在其他群集上部署。函数定义和部署说明请参阅 azure-kusto-irql-graph/references/DEPLOY_IRQL_FUNCTIONS.md。

示例：IRQL 选择器 -> make-graph -> 最短路径

IRQL 负责检索数据；make-graph 负责图分析。此示例通过身份验证事件，查找从外部 IP 到邮件服务器的最短路径：

// IRQL provides unified columns (ClientIp, Hostname, Username, Result)
let auth = Get_Event_Authentication_All
    | where Result == "Failed Login";
let edges = auth
    | summarize Failures = count() by ClientIp, Hostname;
let nodes = union
    (edges | distinct ClientIp | project nodeId = ClientIp, nodeType = "IP"),
    (edges | distinct Hostname | project nodeId = Hostname, nodeType = "Host");
edges
| make-graph ClientIp --> Hostname with nodes on nodeId
| graph-shortest-paths (src)-[e*1..5]->(dest)
    where src.nodeType == "IP" and dest.nodeId == "MAIL-SERVER01"
    project
        SourceIP = src.nodeId,
        PathLength = array_length(e),
        Hops = e.Hostname

示例：IRQL 选择器 -> make-graph -> 连通分量

查找相互连接的 IP 和域名集群——它们可能是 C2 基础设施：

let dns = Get_Dns_All;
let edges = dns | project Source = ClientIp, Target = Domain;
let nodes = union
    (edges | distinct Source | project nodeId = Source, nodeType = "IP"),
    (edges | distinct Target | project nodeId = Target, nodeType = "Domain");
edges
| make-graph Source --> Target with nodes on nodeId
| graph-mark-components with_component_id = ComponentId
| graph-to-table nodes
| summarize
    IPs = make_set_if(nodeId, nodeType == "IP"),
    Domains = make_set_if(nodeId, nodeType == "Domain"),
    Size = count()
  by ComponentId
| where Size > 3
| order by Size desc

示例：IRQL + make-graph 集成

多源调查图示例将 IRQL 选择器与 make-graph 结合使用，以及 Lift_To_Graph 可视化图示例，请参阅 references/EXAMPLES.md。

实际使用场景

完整的演示示例请参阅 references/SCENARIOS.md，其中包括：

• 可达性分析（到关键资产的最短路径）
• 网络分段验证（连通分量）
• 受感染账户的影响范围（可变长度路径匹配）
• SOC 团队的持久化图模型（graph_model + 快照）

使用的 MCP 工具

工具	用途
kusto_query	执行 KQL 查询，包括 make-graph、graph-match 和管理命令
kusto_table_schema_get	在构建边/节点投影前发现表列
kusto_cluster_list	列出可用的 ADX 群集
kusto_database_list	列出群集中的数据库

在 Kusto Explorer 中打开查询（仅限 Windows）

可选的便捷功能。 默认工作流是在聊天中输出 KQL，让用户手动将其复制到 Kusto Explorer 或 VS Code Kusto 扩展中。自动启动仅供选择启用。

默认：在聊天中输出 KQL

始终输出完整的 KQL，并清楚标注步骤 1（连接）和步骤 2（查询）：

// Step 1: Connect to your cluster (skip if already connected)
// Example: uncomment to connect to the KC7 training cluster
// #connect cluster('kc7001.eastus.kusto.windows.net').database('ValdyTimes')
// Or replace with your own cluster:
// #connect cluster('<YOUR_CLUSTER>').database('<YOUR_DATABASE>')

// Step 2: Run the query below
<KQL_QUERY ending at make-graph>

然后紧接着输出一个 ADX Web Explorer 版本，在末尾追加 | graph-to-table nodes as N, edges as E，因为 ADX Web Explorer 无法直接呈现 make-graph：

// ADX Web Explorer version (tabular output):
<SAME_QUERY>
| graph-to-table nodes as N, edges as E

这样无需用户进行任何修改，输出就能同时用于 Kusto Explorer（图可视化）和 ADX Web Explorer（表格结果）。

可选：保存并启动

如果用户要求在 Kusto Explorer 中保存或打开查询，请按照 references/KUSTO_EXPLORER_LAUNCH.md 中的步骤操作。关键规则：

• 写入文件或启动可执行文件之前，始终使用 ask_user 进行确认
• 始终在聊天中显示文件内容，以便用户在打开前检查
• 绝不使用 shell 插值或 here-string — 通过 Set-Content/Add-Content 写入文件
• 绝不将查询编码到浏览器 URL 中
• 在 macOS/Linux 上，保存 .kql 文件，并建议使用 VS Code Kusto 扩展或 ADX Web Explorer

对于 make-graph 可视化（图窗口），查询必须以 make-graph 结尾 — 不要通过管道传递给 graph-match。只有当输出是图对象而非表格时，Kusto Explorer 才会打开图可视化窗口。