文章目录 展开 收起

项目

PyCG论文复现

2026-10-19

参考老师的评分标准,让 AI 筛选好复现的论文,结果过度追求容易复现,在复现难度、工作量上面不够。大部分时间在理解论文的流程上,在 AI 的帮助下复现是比较容易的。

本身的方向和程序分析不一样,就当是锻炼 presentation 的能力了。了解了 Pyhton 动态特性(运行时产能确定变量类型)、语法抽象树是 Python 解释器把 Python 源码到执行程序的一个中间产物。其他倒是想不出学到了什么。

仓库地址:https://github.com/zhanghao0806/PyCG-reproduction

论文链接:https://arxiv.org/html/2103.00587

基础知识

论文工作

这篇论文干什么的?

提出了一种新的方法,生成 Python 程序的调用图。

调用图:节点是函数,边表示谁可能调用谁;

重要条件是:在纯静态环境下(只看源码),把调用图画出来。

  • 为什么不通过动态方法(运行代码)分析?

    1 运行程序有代价;

    2 而且覆盖不到所有的分支语句。所以静态条件下画调用图很有难度。更难的是 Python 是动态语言(不像 C++ Java 变量类型已经写死)。

Python AST

  • Abstract Syntax Tree 抽象语法树;

    Python 代码执行前会被 Python 解释器 翻译成一系列中间产物,AST 是其中的一种。

  • AST 有很多节点,节点算是 Python 对象吧,有很多字段。

    常见节点:FunctionDef(对应源码中的 def)、 Call (对应函数调用)、Assign(赋值语句);

  • 重要的函数 parse(),dump();

    Python的 dump() 很多类都有:大体作用都是把内存中的某种数据“输出”,便于查看、保存。

import  ast
code = "x = 1 + 2"
tree = ast.parse(code)
print(ast.dump(tree, indent=2))

'''
Module(
  body=[
    Assign(
      targets=[
        Name(id='x', ctx=Store())],
      value=BinOp(
        left=Constant(value=1),
        op=Add(),
        right=Constant(value=2)))],
  type_ignores=[])
'''

Pyhton 作用域机制

LEGB 原则;

函数是一等对象,可以赋值给变量,可以作为函数参数、可以作为函数返回值;但是注意不要带括号,只要函数名就行了;

def test(func, arg):
    func(arg)
    func(arg)
test(print, "hello world")
'''
hello world
hello world
'''

论文给出了 python 语言特性带来的挑战:现有静态方法没法解决这些问题。又举了现有两个开源静态分析器生成函数调用图的例子,显示他们的效果很差。

关键设计:按照命名空间区分属性访问。

使用的方法

通过带上命名空间区分相同名字的字段;

图三:设计自己的语法结构,表示程序源码,相当于简化版的 AST;

论文如何将一个 Python 程序变成可分析的结构?Pipeline 是什么?

笼统两步:先得到赋值图,再得到调用图。

Pipeline

1 源码变成中间表示:

先把 python 代码变成解析成 AST;

但是 AST 还不够简洁,于是论文自定义了一种中间表示;

2 维护 4 个核心状态域(数据结构):

得到中间表示之后,维护一些数据结构:作用域树、当前命名空间、赋值图;

作用域树,表示 谁 定义在 谁里面;

3 不动点迭代分析(核心计算):

作者定义了状态转移规则:不断更新第二步的四个核心状态域直到收敛;

收敛:反复运用论文图六的状态转移规则,往作用域树、赋值图等添加新边,直到没有一条新边可以加。

最难理解的地方;

4 生成调用图:

第三步稳定后,生成了赋值图;这一步重新遍历中间表示,在赋值图的帮助下得到调用图。

论文定义一个算法流程。

调用图生成流程

总体来说单单复现的话比较简单,有 GPT 帮忙。关键是搞懂论文吧。

样例源码

接下来以一个代表用例,详细介绍作者的程序是怎么生成调用图的。

关键:这个样例展示了:函数可以赋值给变量,函数还能作为形参传递,这些关系必须被追踪到。

parameters/nested_call:

def nested_func():
    pass

def param_func(a):
    a()

def func(a):
    a(nested_func)

b = param_func
c = func
c(b)
  • 重要区别:第八行,a(nested_func) 和 a(nested_func())。前者传进去的是函数本身,后者是先执行函数,然后传进去的是返回值。

    人写的话,这个调用图很简单。

第一步:

读取 main.py,用 ast.parse() 转成抽象语法树,再查看和保存它。

from pathlib import Path
import ast

source_path = Path(
    "../"
    "reference/paper_artifact/data/micro-benchmark/"
    "parameters/nested_call/main.py"
)
source_code = source_path.read_text(encoding="utf-8")
# 带行号显示源码,方便后面对照 AST
for line_number, line in enumerate(source_code.splitlines(), start=1):
    print(f"{line_number:2}: {line}")

此时得到包含源码的字符串,接下来生成 AST

tree = ast.parse(
    source_code,
    filename=str(source_path),
    mode="exec",
)
print(type(tree))
ast_text = ast.dump(
    tree,
    indent=4,
    include_attributes=False,
)
print(ast_text)

输出:

<class 'ast.Module'>
Module(
    body=[
        FunctionDef(
            name='nested_func',
            args=arguments(
                posonlyargs=[],
                args=[],
                kwonlyargs=[],
                kw_defaults=[],
                defaults=[]),
            body=[
                Pass()],
            decorator_list=[]),
        FunctionDef(
            name='param_func',
            args=arguments(
                posonlyargs=[],
                args=[
                    arg(arg='a')],
                kwonlyargs=[],
                kw_defaults=[],
                defaults=[]),
            body=[
                Expr(
                    value=Call(
                        func=Name(id='a', ctx=Load()),
                        args=[],
                        keywords=[]))],
            decorator_list=[]),
        FunctionDef(
            name='func',
            args=arguments(
                posonlyargs=[],
                args=[
                    arg(arg='a')],
                kwonlyargs=[],
                kw_defaults=[],
                defaults=[]),
            body=[
                Expr(
                    value=Call(
                        func=Name(id='a', ctx=Load()),
                        args=[
                            Name(id='nested_func', ctx=Load())],
                        keywords=[]))],
            decorator_list=[]),
        Assign(
            targets=[
                Name(id='b', ctx=Store())],
            value=Name(id='param_func', ctx=Load())),
        Assign(
            targets=[
                Name(id='c', ctx=Store())],
            value=Name(id='func', ctx=Load())),
        Expr(
            value=Call(
                func=Name(id='c', ctx=Load()),
                args=[
                    Name(id='b', ctx=Load())],
                keywords=[]))],
    type_ignores=[])

根据输出结果可以看到:有三种节点类型:函数、赋值语句、表达式语句;

从最后面的表达式语句,AST 只能看出c的参数是 b,不知道 c 指向哪个函数、b 指向哪个函数,甚至不一定知道 b 是函数。

第二步:

调用作者的PreProcessor,将 AST 变成作者的中间表示;

import json

from pycg.pycg import CallGraphGenerator
from pycg.processing.preprocessor import PreProcessor
from pycg.utils.constants import CALL_GRAPH_OP, NAME_DEF

# 使用前面已经定义的 source_path
source_path = source_path.resolve()

# 创建分析器及其内部的数据管理对象
analyzer = CallGraphGenerator(
    entry_points=[str(source_path)],
    package=str(source_path.parent),
    max_iter=-1,
    operation=CALL_GRAPH_OP,
)

# 只运行预处理这一遍
try:
    analyzer.do_pass(
        PreProcessor,
        True,
        analyzer.import_manager,
        analyzer.scope_manager,
        analyzer.def_manager,
        analyzer.class_manager,
        analyzer.module_manager,
    )
finally:
    analyzer.remove_import_hooks()

运行完预处理,把数据打印出来看一下:

definitions = {}

for name, definition in analyzer.def_manager.get_defs().items():
    pointer = definition.get_name_pointer()

    definitions[name] = {
        "type": definition.get_type(),
        "points_to": sorted(pointer.get()),
        "arguments": {
            key: sorted(values)
            for key, values in pointer.get_args().items()
        },
    }

print(json.dumps(definitions, indent=2, ensure_ascii=False))

输出:

{
  "main": {
    "type": "MODULEDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.nested_func": {
    "type": "FUNCTIONDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.param_func": {
    "type": "FUNCTIONDEF",
    "points_to": [],
    "arguments": {
      "a": [
        "main.param_func.a"
      ]
    }
  },
  "main.func": {
    "type": "FUNCTIONDEF",
    "points_to": [],
    "arguments": {
      "a": [
        "main.func.a"
      ]
    }
  },
  "main.nested_func.<RETURN>": {
    "type": "NAMEDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.param_func.<RETURN>": {
    "type": "NAMEDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.param_func.a": {
    "type": "NAMEDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.func.<RETURN>": {
    "type": "NAMEDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.func.a": {
    "type": "NAMEDEF",
    "points_to": [],
    "arguments": {
      "0": [
        "main.nested_func"
      ]
    }
  },
  "main.b": {
    "type": "NAMEDEF",
    "points_to": [
      "main.param_func"
    ],
    "arguments": {}
  },
  "main.c": {
    "type": "NAMEDEF",
    "points_to": [
      "main.func"
    ],
    "arguments": {
      "0": [
        "main.b"
      ]
    }
  }
}

输出结果大概对应作者定义的中间表示。

“points_to”: [] 含义:这个名字可能指向的对象合集; “main.b” 的 points_to": ["main.param_func"], 说明程序已经识别出 b 代表的是函数 parm_func;

还有一个值得关注的:main.param_func 和 main.func 都有形参 a,但是他们命名空间不同。这是论文重要的设计之一:按标识符所在的命名空间区分对象;

第三步:

传播关系直到收敛。

from pycg.processing.postprocessor import PostProcessor

# 先补全预处理阶段记录的参数关系
analyzer.def_manager.complete_definitions()

# 重复分析,直到状态不再变化
while not analyzer.has_converged():
    analyzer.state = analyzer.extract_state()
    analyzer.reset_counters()

    analyzer.do_pass(
        PostProcessor,
        False,
        analyzer.import_manager,
        analyzer.scope_manager,
        analyzer.def_manager,
        analyzer.class_manager,
        analyzer.module_manager,
    )

    analyzer.def_manager.complete_definitions()

# 重新读取收敛后的信息,更新之前的 definitions
definitions = {}

for name, definition in analyzer.def_manager.get_defs().items():
    pointer = definition.get_name_pointer()
    definitions[name] = {
        "type": definition.get_type(),
        "points_to": sorted(pointer.get()),
        "arguments": {
            key: sorted(values)
            for key, values in pointer.get_args().items()
        },
    }

print("是否收敛:", analyzer.has_converged())
print(json.dumps(definitions, indent=2, ensure_ascii=False))

输出:

是否收敛: True
{
  "main": {
    "type": "MODULEDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.nested_func": {
    "type": "FUNCTIONDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.param_func": {
    "type": "FUNCTIONDEF",
    "points_to": [],
    "arguments": {
      "a": [
        "main.param_func.a"
      ]
    }
  },
  "main.func": {
    "type": "FUNCTIONDEF",
    "points_to": [],
    "arguments": {
      "a": [
        "main.func.a"
      ]
    }
  },
  "main.nested_func.<RETURN>": {
    "type": "NAMEDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.param_func.<RETURN>": {
    "type": "NAMEDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.param_func.a": {
    "type": "NAMEDEF",
    "points_to": [
      "main.nested_func"
    ],
    "arguments": {}
  },
  "main.func.<RETURN>": {
    "type": "NAMEDEF",
    "points_to": [],
    "arguments": {}
  },
  "main.func.a": {
    "type": "NAMEDEF",
    "points_to": [
      "main.b"
    ],
    "arguments": {
      "0": [
        "main.nested_func"
      ]
    }
  },
  "main.b": {
    "type": "NAMEDEF",
    "points_to": [
      "main.param_func"
    ],
    "arguments": {
      "0": [
        "main.nested_func"
      ]
    }
  },
  "main.c": {
    "type": "NAMEDEF",
    "points_to": [
      "main.func"
    ],
    "arguments": {
      "0": [
        "main.b"
      ]
    }
  }
}

此时所有关系已经收敛,可以看到和第一步的输出不同的地方有:

main.func.a 的 points_to从空变成了 "main.b",说明程序已经识别出 func()的参数 a 其实是变量 b;

main.param_func.a 的 points_to从空变成了 "main.nested_func";说明程序识别出 函数 param_func 里面的 a 是函数nested_func;

第四步:

根据已收敛的状态(保存在 analyzer中),生成调用图。

from pycg.formats.simple import Simple
from pycg.processing.cgprocessor import CallGraphProcessor

analyzer.reset_counters()

analyzer.do_pass(
    CallGraphProcessor,
    False,
    analyzer.import_manager,
    analyzer.scope_manager,
    analyzer.def_manager,
    analyzer.class_manager,
    analyzer.module_manager,
    call_graph=analyzer.cg,
)

callgraph = Simple(analyzer).generate()

print(json.dumps(callgraph, indent=2, ensure_ascii=False))

输出:

{
  "main": [
    "main.func"
  ],
  "main.nested_func": [],
  "main.param_func": [
    "main.nested_func"
  ],
  "main.func": [
    "main.param_func"
  ]
}

得到这样的键值对。键表示函数名,对应的列表是他直接调用的函数; 把得到的输出,可视化成一张图:

import networkx as nx
import matplotlib.pyplot as plt

g = nx.DiGraph(callgraph)
pos = {name: (i, 0) for i, name in enumerate(nx.topological_sort(g))}

plt.figure(figsize=(10, 2))
nx.draw(
    g, pos, with_labels=True,
    node_size=0,
    font_size=10,
    arrowsize=20,
    min_source_margin=65,
    min_target_margin=65,
)
plt.show()

为什么论文中提到的方法不行

接下来用论文中提到的 Pyan 方法,对同样一个测试用例生成调用图,看结果是否不行;

程序

from pathlib import Path
import json
import subprocess
import sys

source_path = Path("../reference/paper_artifact/data/micro-benchmark/parameters/nested_call/main.py").resolve()
pyan_path = Path("../reference/paper_tools/pyan/pyan.py").resolve()

print(source_path.read_text(encoding="utf-8"))

输出:

def nested_func():
    pass

def param_func(a):
    a()

def func(a):
    a(nested_func)

b = param_func
c = func
c(b)

运行论文工件中的 Pyan,结果保存在内存中并直接显示。

result = subprocess.run(
    [sys.executable, "-B", str(pyan_path), str(source_path)],
    capture_output=True,
    text=True,
    check=True,
)
callgraph = json.loads(result.stdout)
print(json.dumps(callgraph, indent=2, ensure_ascii=False))

输出:

{
  "main": [
    "main.func",
    "main.param_func"
  ],
  "main.func": [
    "main.nested_func"
  ],
  "main.nested_func": [],
  "main.param_func": []
}

画出 Pyan 实际输出的关系。

import networkx as nx
import matplotlib.pyplot as plt

g = nx.DiGraph(callgraph)
pos = {}
for level, nodes in enumerate(nx.topological_generations(g)):
    nodes = sorted(nodes)
    for i, name in enumerate(nodes):
        pos[name] = (i - (len(nodes) - 1) / 2, -level)

plt.figure(figsize=(7, 4))
nx.draw(g, pos, with_labels=True, node_size=0, font_size=10,
        arrowsize=18, min_source_margin=22, min_target_margin=22)
plt.show()

实测结果:

Pyan 漏掉 main.func 调用 main.param_func、main.param_func 调用 main.nested_func 两条边;多出 main 使用 main.param_func、main.func 使用 main.nested_func 两条边。

为什么会漏边:

Pyan 能看到代码中调用了 a,但没有把跨函数传递的实参与形参关联起来,不知道形参 a 其实是param_func;

复现过程

整体复现内容

微基准:112 个最小 Python 程序真实项目基准:5 个开源 Python 项目
特点每个程序聚焦一种调用场景,规模小、便于定位错误包含多个模块、函数和类,调用关系更复杂
内容函数赋值、参数传递、返回值、闭包、继承、模块导入等,共 16 类autojump、fabric、asciinema、face_classification、Sublist3r
评价检查是否存在多余调用边或遗漏调用边计算精确率和召回率

112个小型测试用例

第一次运行,复现论文中提到的 112 个程序对应的调用图,复现结果和论文中提到的一致。

指标论文结果本次结果
Complete 检查:没有多余调用边111/112111/112
Sound 检查:没有漏掉应有调用边103/112103/112

PyCG 的局限性:

不支持 * 号解包

assignments/starred

def func1():
    pass
def func2():
    pass
def func3():
    pass
def func4():
    pass
a, *b, c = func1, func2, func3, func4
a()
b[0]()
b[1]()
c()

不支持处理星号解包的“中间收集、尾部匹配”规则,导致函数绑定错误并遗漏调用边;

不支持内置对象的方法

builtins/types

" ".join(["1", "2", "3"])
# 按空格拼接:"1 2 3"
"a b c".split(" ")
# 按空格拆分: ["a", "b", "c"]
d = {"a": 1}
d.items()
# 可视化键值对

一共调用了三个函数,PyCG无法识别:不包含任何调用包。

5个大型项目

先大致介绍项目的情况;

项目功能论文报告代码行数
autojump命令行目录跳转2,662
fabric远程执行与部署3,236
asciinema终端录屏1,409
face_classification人脸相关分类1,455
Sublist3r子域名枚举1,269
项目本次精确率作者精确率本次召回率作者召回率
autojump99.5%99.5%68.2%68.2%
fabric98.3%98.3%61.9%61.9%
asciinema100.0%100.0%68.3%68.3%
face_classification99.5%99.5%89.7%89.7%
Sublist3r98.8%98.8%61.6%61.6%
平均99.2%99.2%69.9%69.9%
分析器与环境112 个小用例5 个真实项目
0.0.2 + Python 3.10,未经兼容处理未对齐指标全部对齐
0.0.8 + Python 3.10对齐未对齐
0.0.2 + Python 3.10,仅临时修正下标解析全部分类指标对齐全部分类指标对齐

评价指标:

精确率:分析器生成的边,有多少是正确的?

Precision=∣Eactual∩Eexpected∣∣Eactual∣\text{Precision} = \frac{|E_{\text{actual}} \cap E_{\text{expected}}|}{|E_{\text{actual}}|}

召回率:真实的边,分析器找到了多少?

Recall=∣Eactual∩Eexpected∣∣Eexpected∣\text{Recall} = \frac{|E_{\text{actual}} \cap E_{\text{expected}}|}{|E_{\text{expected}}|}

把文件实验包可以整理一下了;