SAST专项之使用LSP语言服务器寻找污点函数调用链

前言

在我的上一篇文章

SAST专项之使用tree-sitter进行SINK点扫描——diff正则扫描器

已经做到了使用AST语法树把危险操作且变量可能可控的地方找出来了

那么第二步我们肯定是会去想，这个变量它真的可控吗？这个变量它又是从哪来的呢？

正常我们在挖洞的时候肯定就是打开IDE，然后对着这个函数右键接着一直查找用法，看看最上头的调用点是否由我们所控制来判断image-20260122032001970

这篇文章就是要把这一过程自动化，所使用的技术技术LSP（语言服务器）也是我们现代编译器的核心机制

什么是LSP

关于什么是LSP不多赘述

LSP - 语言服务器开发指南

只需要知道它是一个服务器，给我们的编辑器作为客户端可以查看代码中一些函数的定义，调用关系，位置信息，以及语法错误信息等等

安装和连接LSP服务器

1
npm install -g intelephense

启动LSP服务

1
intelephense --stdio

这里我们使用io输入和输出的方式和服务器进行通信

简单了解一下请求体，每个请求体都会包含如下四个字段

1
2
3
4
5
6
{
  "jsonrpc": "2.0",        // 【暗号】版本号，必须固定写 "2.0"
  "id": 1,                 // 【流水号】请求的ID (整数或字符串)
  "method": "initialize",  // 【动作】你想让服务器干什么
  "params": { ... }        // 【参数】干这件事需要的材料
}

jsonrpc: 就像写信的抬头，告诉对方“我说的是 JSON-RPC 2.0 这种语言，别搞错了”。

id: 非常重要。

  • 因为 LSP 是异步的，你可能连续发了 10 个请求。
  • 服务器回给你的时候，顺序可能是乱的。
  • 你需要靠这个 ID 把“请求”和“响应”对上号。（比如你发 ID: 100，服务器回 ID: 100，你就知道这是回复刚才那个请求的）。

method: 方法名。比如 initialize（初始化）、textDocument/hover（鼠标悬停）、shutdown（关闭）。

params: 具体的参数对象。不同的 method 需要不同的 params。

初始化包initialize

目的：认门。告诉服务器项目在哪，建立了索引，我才能查。

关键字段：

  • rootUri: 必填。项目的根目录，格式必须是 file:///C:/Users/...。如果不填或填错，它就变成了“单文件模式”，没法跨文件找调用。

使用这个包我们可以做一下初始连接测试，看看我们自己写的客户端是否可以和LSP服务器进行通信

如下测试代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import subprocess
import json
import os

LSP_CMD = "intelephense --stdio"

print(f"[*] 正在启动 LSP 服务: {LSP_CMD}")

process = subprocess.Popen(
    LSP_CMD,
    shell=True,      
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE 
)

request_data = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
        "processId": os.getpid(),
        "rootUri": None, # 这里先不指定项目路径，单纯测试连接
        "capabilities": {}
    }
}

json_str = json.dumps(request_data)

# 消息体前面必须带 Content-Length 头，并用 \r\n\r\n 分隔
# 格式：Content-Length: <数字>\r\n\r\n<JSON内容>
content_length = len(json_str)
full_message = f"Content-Length: {content_length}\r\n\r\n{json_str}"

print(f"[*] 发送请求:\n{full_message.strip()}")

# 写入 stdin，注意要 encode 成 bytes
process.stdin.write(full_message.encode('utf-8'))
process.stdin.flush() # 必须刷新缓冲区，否则数据可能卡在管道里发不出去

print("[*] 等待响应...")

# 读取第一行：Content-Length: xxx
header = process.stdout.readline().decode('utf-8')
print(f"[收到头] {header.strip()}")

if header.startswith("Content-Length:"):
    # 提取长度数字
    length = int(header.split(":")[1].strip())
    
    # 读取第二行：它应该是一个空行 (\r\n)
    _ = process.stdout.readline()
    
    # 读取真正的 JSON 内容
    body = process.stdout.read(length).decode('utf-8')
    
    print("-" * 20)
    print("[成功] 收到 LSP 服务器响应：")
    
    resp_json = json.loads(body)
    print(json.dumps(resp_json, indent=2))
    print("-" * 20)
else:
    print("[失败] 未收到正确的 Content-Length 头")
    err = process.stderr.read().decode('utf-8')
    if err:
        print(f"[错误输出] {err}")

process.kill()

可以看到我们就可以正常连接刚刚下好的PHP的LSP服务器了

image-20260122191020499

使用LSP追寻调用关系

回归正题，我们的目的就是使用LSP把手动在IDE寻找调用的这一操作自动化

这还需要我们下面两个请求，分别对应方法textDocument/didOpen以及textDocument/references

textDocument/didOpen打开文档

这是一个通知型请求，不需要回信，所以也就不需要填写ID字段，只需要把我们的代码段和URI发送过去即可

1
2
3
4
5
6
7
8
9
10
11
{
  "method": "textDocument/didOpen",
  "params": {
    "textDocument": {
      "uri": "file:///c:/project/vuln.php",  // [1] 文件的唯一身份证
      "languageId": "php",                    // [2] 语言类型
      "version": 1,                           // [3] 版本号
      "text": "<?php function..."             // [4] 文件完整内容
    }
  }
}

需要同时写uri和text的原因是为了让服务器处理更快一点，不需要服务器再去磁盘去读取IO

textDocument/references查找引用

这是需要回信的请求，所以需要填写ID字段，初次之外，为了定位我们的函数，我们需要提供行号和列数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
{
  "id": 2,
  "method": "textDocument/references",
  "params": {
    "textDocument": {
      "uri": "file:///c:/project/vuln.php"   // [1] 在哪个文件里找
    },
    "position": {
      "line": 1,                              // [2] 行号 (0开始)
      "character": 10                         // [3] 列号
    },
    "context": {
      "includeDeclaration": false             // [4] 是否包含定义本身
    }
  }
}

可以注意到我们这里是只需要提供行号和起始列数，不需要提供结束列数，是因为LSP本来就是面向鼠标点击的一个传输协议，所以只要鼠标放在那个范围，就可以去查看引用，也是比较方便

所以我们可以写如下脚本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
import subprocess
import json
import os
import time
from pathlib import Path

LSP_CMD = "intelephense --stdio"
PROJECT_ROOT = r"E:\1devEnv\CODE\project\ChainSeeker\ChainLsp" # 修改你的路径
TARGET_FILE = os.path.join(PROJECT_ROOT, "test.php")
TARGET_LINE = 1  
TARGET_CHAR = 11

def make_request(method, params, msg_id=None):
    body = {"jsonrpc": "2.0", "method": method, "params": params}
    if msg_id is not None: body["id"] = msg_id
    json_bytes = json.dumps(body).encode('utf-8')
    return f"Content-Length: {len(json_bytes)}\r\n\r\n".encode('utf-8') + json_bytes

def get_response_by_id(process, expected_id):
    """循环读取直到拿到目标ID，自动过滤通知"""
    while True:
        header = process.stdout.readline().decode('utf-8', errors='ignore')
        if not header: return None
        if header.startswith("Content-Length:"):
            length = int(header.split(":")[1].strip())
            process.stdout.readline()
            msg = json.loads(process.stdout.read(length).decode('utf-8'))
            
            if msg.get("id") == expected_id:
                return msg

def main():
    print(f"[*] 启动 LSP...")
    process = subprocess.Popen(
        LSP_CMD, shell=True,
        stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE
    )

    req_init = make_request("initialize", {
        "processId": os.getpid(),
        "rootUri": Path(PROJECT_ROOT).as_uri(),
        "capabilities": {} 
    }, 1)
    process.stdin.write(req_init)
    process.stdin.flush()
    
    get_response_by_id(process, 1)
    process.stdin.write(make_request("initialized", {}, None)) # 发送确认通知
    process.stdin.flush()

    with open(TARGET_FILE, 'r') as f:
        file_content = f.read()
    
    process.stdin.write(make_request("textDocument/didOpen", {
        "textDocument": {
            "uri": Path(TARGET_FILE).as_uri(),
            "languageId": "php",
            "version": 1,
            "text": file_content
        }
    }, None))
    process.stdin.flush()

    print("[*] 等待索引建立 (2秒)...")
    time.sleep(2)
    
    req_ref = make_request("textDocument/references", {
        "textDocument": {"uri": Path(TARGET_FILE).as_uri()},
        "position": {"line": TARGET_LINE, "character": TARGET_CHAR},
        "context": {
            "includeDeclaration": False # 我们只想找"哪里用了它"，不想找"它在哪里定义的"
        }
    }, 2)
    process.stdin.write(req_ref)
    process.stdin.flush()

    resp = get_response_by_id(process, 2)
    
    print("\n" + "="*30)
    if resp and "result" in resp:
        locations = resp["result"]
        print(f"发现 {len(locations)} 处调用\n")
        
        for loc in locations:
            file_uri = loc['uri']
            range_info = loc['range']
            start_line = range_info['start']['line'] + 1
            start_char = range_info['start']['character'] + 1
            end_line = range_info['end']['line'] + 1
            end_char = range_info['end']['character'] + 1
            
            # 简单的 URI 转路径处理
            file_path = file_uri.replace("file:///", "").replace("%3A", ":")
            
            print(f" -> 调用者位置: {file_path}")
            print(f"    位置: 行 {start_line} 列 {start_char} -> 行 {end_line} 列 {end_char}")
            print("-" * 20)
    else:
        print("未找到任何引用。")

    process.kill()

if __name__ == "__main__":
    main()

最后也是找到的调用处的具体位置

image-20260122200614172

建立调用图

我们已经实现了功能的最小单元，就是寻找到某一函数的所有调用位置，但是我们知道一个函数在一个大项目中肯定会有很多次调用，然后这个调用它的函数往上也会有很多调用处，那我们如果找出所有的链路呢？

我们先从最理想的情况开始考虑

image-20260122201651982

首先我们使用扫描器找到了system(‘calc’)，接着我们用AST语法树找到它所在的函数D，然后对它所在的函数去调用LSP去找到它的调用点，又使用AST去找到它的调用函数C…..这样的一个递归过程，最后找到了A，没有地方再调用A，所以找到一个没有地方调用的函数，作为我们的递归出口

那么我们先来实现一次递归的功能

查找某行代码所在的函数

首先我们要实现的点就是，如何找到system()所在的函数D，这里我们选择把这个功能还在在原本使用go的AST扫描器上开发

因为是我们python程序调用go程序的API，得提前商议好通信的字段，首先我们相互传的是一个函数的位置信息，首先是python给go的字段，应该包含

  • uri:函数所在的文件地址
  • line:函数所在的行号
  • name:函数名称

然后是go返回给python的信息，应该包含

  • name:函数名称
  • parent_name:所在函数的名称
  • line:所在函数的行号
  • start_column:起始列数
  • end_column:结束列数

接下来就是完善我们的AST查找逻辑

我们使用的是向上攀爬的逻辑

我们需要找到包含目标行号的最具体的那个节点。 AST 是一个层级结构，父节点包含子节点。我使用了一个递归函数 walk 从根节点开始向下“钻”：

  • 判定范围：检查当前节点的起始行和结束行是否包含目标行。
  • 贪心查找：如果当前节点包含目标行，我就记录它为 targetNode，然后继续检查它的所有子节点。因为子节点一定比父节点更小、更具体。最终我们会停留在树最深处那个包含了目标行的节点上
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 核心定位逻辑
walk = func(n *sitter.Node) {
    start := n.StartPoint().Row
    end := n.EndPoint().Row

    // 如果节点范围覆盖了目标行
    if targetLine >= start && targetLine <= end {
        targetNode = n // 暂存当前节点
        // 继续尝试找更深层的子节点
        count := n.ChildCount()
        for i := uint32(0); i < count; i++ {
            walk(n.Child(int(i)))
        }
    }
}

一旦找到了最具体的 targetNode，我沿着树向上爬

  • 匹配类型：在回溯过程中，检查每一个祖先节点的类型 (Type())。
    • function_definition: 对应普通函数 function foo() { ... }
    • method_declaration: 对应类方法 public function bar() { ... }
  • 提取信息：一旦命中这两种类型，就提取它的名字、起始行和结束行
  • 兜底：如果爬到了树顶都没找到，说明这行代码在全局作用域，返回 {main}，这也是我们之后递归出去的出口
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
func (a *Analyzer) findParentFunction(n *sitter.Node, source []byte) *FunctionLocation {
    curr := n.Parent()
    for curr != nil {
        switch curr.Type() {
        case "function_definition":
            // 找到普通函数定义，提取名字和位置
            // ...
            return &FunctionLocation{...}
        case "method_declaration":
            // 找到类方法定义
            // ...
            return &FunctionLocation{...}
        }
        curr = curr.Parent() // 继续向上找
    }
    return &FunctionLocation{Name: "{main}"} // 全局作用域
}

一次递归

我们以发现的这个点为例向上寻找调用链

image-20260123142605890

输出的位置信息如下

1
2
3
4
5
6
7
8
9
{
  "uri": "E:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Js_packer.php",
  "line": 96,
  "name": "exec",
  "rule": "命令执行",
  "severity": "high",
  "description": "命令执行函数中存在变量，可能存在命令执行漏洞",
  "snippet": "$parser-\u003eexec($script)"
}

寻找所在的函数输出

1
2
3
4
5
6
7
8
9
10
[
  {
    "name": "exec",
    "parent_name": "_basicCompression",
    "uri": "E:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Js_packer.php",
    "line": 74,
    "start_column": 5,
    "end_column": 6
  }
]

使用LSP寻找_basicCompression的调用

之后就找不到了，所以这个命令执行只在一个不被调用的函数中使用了

下面我们找一个层级更多的

1
2
3
4
5
6
7
8
9
10
11
 [
  {
    "uri": "E:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 634,
    "name": "require",
    "rule": "文件包含",
    "severity": "medium",
    "description": "文件包含函数中存在变量，可能存在文件包含漏洞",
    "snippet": "require $file"
  }
]

文件包含，寻找所在函数

1
2
3
4
5
6
7
8
9
10
[
  {
    "name": "require",
    "parent_name": "get",
    "uri": "E:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 597,
    "start_column": 9,
    "end_column": 10
  }
]

寻找所在函数被调用的地方

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
[
  {
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 163,
    "name": "get"
  },
  {
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 670,
    "name": "get"
  },
  {
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 758,
    "name": "get"
  }
]

寻找他们所在的函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
[
  {
    "name": "get",
    "parent_name": "toform",
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 101,
    "start_column": 9,
    "end_column": 10
  },
  {
    "name": "get",
    "parent_name": "option",
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 664,
    "start_column": 9,
    "end_column": 10
  },
  {
    "name": "get",
    "parent_name": "get_value",
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 756,
    "start_column": 9,
    "end_column": 10
  }
]

寻找调用处

1
2
3
4
5
6
7
8
9
10
11
12
[
  {
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 785,
    "name": "get_value"
  },
  {
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 806,
    "name": "get_value"
  }
]

找到所在函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
[
  {
    "name": "get_value",
    "parent_name": "format_value",
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 774,
    "start_column": 9,
    "end_column": 10
  },
  {
    "name": "get_value",
    "parent_name": "format_value",
    "uri": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
    "line": 774,
    "start_column": 9,
    "end_column": 10
  }
]

完整图

那么我们如何把这个函数调用关系绘制成图呢？我们就需要一个程序作为两边API传递的枢纽，记录一下他们每次都传递了什么数据然后在中间绘制成图即可

新建一个draw.py

通过有向图的方式生成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def draw(test_dir="test", root_json="test.json", output_json="call_chain.json"):
    """
    根据test目录下的JSON文件整理函数调用链数据
    
    参数:
        test_dir: 包含depth相关JSON文件的目录
        root_json: 根节点JSON文件路径
        output_json: 输出JSON文件路径
        
    返回:
        包含nodes和edges的字典，适合前端可视化使用
    """
    
    # 读取根节点
    with open(root_json, 'r', encoding='utf-8') as f:
        root_data = json.load(f)
    
    # 数据结构：nodes列表和edges列表
    nodes = []
    edges = []
    node_map = {}  # 用于去重和快速查找: key -> node_index
    
    def make_node_key(name, uri, line):
        """生成节点唯一标识"""
        return f"{name}@{os.path.normpath(uri)}:{line}"
    
    def add_node(node_id, name, uri, line, node_type, **extra):
        """添加节点，返回节点索引"""
        if node_id in node_map:
            return node_map[node_id]
        
        node_index = len(nodes)
        node = {
            "id": node_id,
            "index": node_index,
            "name": name,
            "file": os.path.basename(uri),
            "full_path": os.path.normpath(uri),
            "line": line,
            "type": node_type,
            **extra
        }
        nodes.append(node)
        node_map[node_id] = node_index
        return node_index
    
    def add_edge(source_id, target_id, relation_type):
        """添加边"""
        if source_id not in node_map or target_id not in node_map:
            return
        
        edge = {
            "source": node_map[source_id],
            "target": node_map[target_id],
            "type": relation_type
        }
        edges.append(edge)
    
    # 1. 添加根节点（sink点）
    print("=" * 60)
    print("函数调用链分析")
    print("=" * 60)
    print("\n[ROOT] 漏洞Sink点:")
    
    for item in root_data:
        node_id = make_node_key(item['name'], item['uri'], item['line'])
        add_node(
            node_id, 
            item['name'], 
            item['uri'], 
            item['line'],
            "sink",
            rule=item.get('rule', ''),
            severity=item.get('severity', ''),
            description=item.get('description', ''),
            snippet=item.get('snippet', '')
        )
        print(f"  - {item['name']} @ {os.path.basename(item['uri'])}:{item['line']}")
        print(f"    规则: {item.get('rule', 'N/A')}")
        print(f"    描述: {item.get('description', 'N/A')}")
    
    # 2. 按深度处理调用链
    depth = 0
    previous_level_map = {}  # 上一层的映射关系: child_name -> parent_nodes
    
    # 初始化：根节点作为第0层的"子节点"
    for item in root_data:
        child_key = make_node_key(item['name'], item['uri'], item['line'])
        if item['name'] not in previous_level_map:
            previous_level_map[item['name']] = []
        previous_level_map[item['name']].append(child_key)
    
    while True:
        parent_file = os.path.join(test_dir, f"test_depth_{depth}_parents.json")
        call_file = os.path.join(test_dir, f"test_depth_{depth}_calls.json")
        
        if not os.path.exists(parent_file):
            break
        
        print(f"\n[DEPTH {depth}] 父函数定义:")
        
        # 读取parent文件
        with open(parent_file, 'r', encoding='utf-8') as f:
            parents = json.load(f)
        
        current_level_map = {}  # 当前层的映射
        
        # 处理parent关系
        for parent in parents:
            child_name = parent['name']
            parent_name = parent['parent_name']
            uri = parent['uri']
            line = parent['line']
            
            # 添加父函数节点
            parent_node_id = make_node_key(parent_name, uri, line)
            add_node(
                parent_node_id,
                parent_name,
                uri,
                line,
                "function",
                depth=depth
            )
            
            print(f"  - {parent_name} @ {os.path.basename(uri)}:{line}")
            print(f"    包含函数: {child_name}")
            
            # 建立边：parent -> child
            if child_name in previous_level_map:
                for child_node_id in previous_level_map[child_name]:
                    add_edge(parent_node_id, child_node_id, "contains")
            
            # 记录当前层
            if parent_name not in current_level_map:
                current_level_map[parent_name] = []
            current_level_map[parent_name].append(parent_node_id)
        
        # 读取calls文件
        if os.path.exists(call_file):
            print(f"\n[DEPTH {depth}] 调用位置:")
            
            with open(call_file, 'r', encoding='utf-8') as f:
                calls = json.load(f)
            
            next_level_map = {}  # 下一层的映射
            
            for call in calls:
                caller_name = call['name']
                uri = call['uri']
                line = call['line']
                
                # 添加调用点节点
                call_node_id = make_node_key(f"call_{caller_name}", uri, line)
                add_node(
                    call_node_id,
                    caller_name,
                    uri,
                    line,
                    "call_site",
                    depth=depth
                )
                
                print(f"  - 调用 {caller_name} @ {os.path.basename(uri)}:{line}")
                
                # 建立边：call_site -> function
                if caller_name in current_level_map:
                    for func_node_id in current_level_map[caller_name]:
                        add_edge(call_node_id, func_node_id, "calls")
                
                # 记录到下一层
                if caller_name not in next_level_map:
                    next_level_map[caller_name] = []
                next_level_map[caller_name].append(call_node_id)
            
            # 更新previous_level_map为下一轮准备
            previous_level_map = next_level_map
        else:
            # 如果没有calls文件，使用当前的parent作为下一层的输入
            previous_level_map = current_level_map
        
        depth += 1
    
    # 3. 构建最终数据结构
    result = {
        "nodes": nodes,
        "edges": edges,
        "metadata": {
            "total_nodes": len(nodes),
            "total_edges": len(edges),
            "max_depth": depth - 1,
            "root_count": len(root_data)
        }
    }
    
    # 4. 保存到JSON文件
    with open(output_json, 'w', encoding='utf-8') as f:
        json.dump(result, f, indent=2, ensure_ascii=False)
    
    print("\n" + "=" * 60)
    print(f"数据统计:")
    print(f"  - 总节点数: {len(nodes)}")
    print(f"  - 总边数: {len(edges)}")
    print(f"  - 最大深度: {depth - 1}")
    print(f"  - 数据已保存到: {output_json}")
    print("=" * 60)
    
    return result

if __name__ == '__main__':
    draw()

把图的信息整理成新的json文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
{
  "nodes": [
    {
      "id": "require@E:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:634",
      "index": 0,
      "name": "require",
      "file": "Field.php",
      "full_path": "E:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 634,
      "type": "sink",
      "rule": "文件包含",
      "severity": "medium",
      "description": "文件包含函数中存在变量，可能存在文件包含漏洞",
      "snippet": "require $file"
    },
    {
      "id": "get@E:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:597",
      "index": 1,
      "name": "get",
      "file": "Field.php",
      "full_path": "E:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 597,
      "type": "function",
      "depth": 0
    },
    {
      "id": "call_get@e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:163",
      "index": 2,
      "name": "get",
      "file": "Field.php",
      "full_path": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 163,
      "type": "call_site",
      "depth": 0
    },
    {
      "id": "call_get@e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:670",
      "index": 3,
      "name": "get",
      "file": "Field.php",
      "full_path": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 670,
      "type": "call_site",
      "depth": 0
    },
    {
      "id": "call_get@e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:758",
      "index": 4,
      "name": "get",
      "file": "Field.php",
      "full_path": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 758,
      "type": "call_site",
      "depth": 0
    },
    {
      "id": "toform@e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:101",
      "index": 5,
      "name": "toform",
      "file": "Field.php",
      "full_path": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 101,
      "type": "function",
      "depth": 1
    },
    {
      "id": "option@e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:664",
      "index": 6,
      "name": "option",
      "file": "Field.php",
      "full_path": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 664,
      "type": "function",
      "depth": 1
    },
    {
      "id": "get_value@e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:756",
      "index": 7,
      "name": "get_value",
      "file": "Field.php",
      "full_path": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 756,
      "type": "function",
      "depth": 1
    },
    {
      "id": "call_get_value@e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:785",
      "index": 8,
      "name": "get_value",
      "file": "Field.php",
      "full_path": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 785,
      "type": "call_site",
      "depth": 1
    },
    {
      "id": "call_get_value@e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:806",
      "index": 9,
      "name": "get_value",
      "file": "Field.php",
      "full_path": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 806,
      "type": "call_site",
      "depth": 1
    },
    {
      "id": "format_value@e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php:774",
      "index": 10,
      "name": "format_value",
      "file": "Field.php",
      "full_path": "e:\\1CTFDB\\chanllenge\\myvulhub\\xunruicms-master\\dayrui\\Fcms\\Library\\Field.php",
      "line": 774,
      "type": "function",
      "depth": 2
    }
  ],
  "edges": [
    {
      "source": 1,
      "target": 0,
      "type": "contains"
    },
    {
      "source": 2,
      "target": 1,
      "type": "calls"
    },
    {
      "source": 3,
      "target": 1,
      "type": "calls"
    },
    {
      "source": 4,
      "target": 1,
      "type": "calls"
    },
    {
      "source": 5,
      "target": 2,
      "type": "contains"
    },
    {
      "source": 5,
      "target": 3,
      "type": "contains"
    },
    {
      "source": 5,
      "target": 4,
      "type": "contains"
    },
    {
      "source": 6,
      "target": 2,
      "type": "contains"
    },
    {
      "source": 6,
      "target": 3,
      "type": "contains"
    },
    {
      "source": 6,
      "target": 4,
      "type": "contains"
    },
    {
      "source": 7,
      "target": 2,
      "type": "contains"
    },
    {
      "source": 7,
      "target": 3,
      "type": "contains"
    },
    {
      "source": 7,
      "target": 4,
      "type": "contains"
    },
    {
      "source": 8,
      "target": 7,
      "type": "calls"
    },
    {
      "source": 9,
      "target": 7,
      "type": "calls"
    },
    {
      "source": 10,
      "target": 8,
      "type": "contains"
    },
    {
      "source": 10,
      "target": 9,
      "type": "contains"
    },
    {
      "source": 10,
      "target": 8,
      "type": "contains"
    },
    {
      "source": 10,
      "target": 9,
      "type": "contains"
    }
  ],
  "metadata": {
    "total_nodes": 11,
    "total_edges": 19,
    "max_depth": 2,
    "root_count": 1
  }
}

最后进行前端展示

image-20260123223356945

就能得到我们的调用图了

但是有时候如果扫到一个非常底层的函数，比如数据库连接，这意味着每一次进行数据库操作都会被调用一下，就会出现下面这样有点难绷的图

image-20260123223558476

这也是后期需要优化的一部分