Vanna CVE-2025-47277 看CodeQL Python 方法映射

1. 漏洞原理

漏洞并不复杂，整个链也不长

source：
https://github.com/vanna-ai/vanna/blob/a72b842d420cf1fa061e5f97d45ea08051651ebb/src/vanna/flask/__init__.py#L441

upload successful

sink：
https://github.com/vanna-ai/vanna/blob/4da8dea0ce14a0d1db5a0692a7921d873be91c5f/src/vanna/base/base.py#L2088

upload successful

问题代码抽象如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class VannaFlaskApp:
    flask_app = None
        def __init__(self, 
                    vn: VannaBase, 
                    cache: Cache = MemoryCache(),
                    auth: AuthInterface = NoAuth(),
                    ...
            self.vn = vn
            ...  
            @self.flask_app.route("/api/v0/generate_plotly_figure", methods=["GET"])
            @self.requires_auth
            @self.requires_cache(["df", "question", "sql"])
            def generate_plotly_figure(user: any, id: str, df, question, sql):
                ...
                try:
                    code = vn.generate_plotly_code(
                        question=question,
                        sql=sql,
                        df_metadata=f"Running df.dtypes gives:\n {df.dtypes}",
                    )
                    fig = vn.get_plotly_figure(plotly_code=code, df=df, dark_mode=False)

python 现在的CodeQL 支持还是有限，在复现过程中，发现以下三点问题：
1.Flask 框架不全，这几个参数都不是source点
2.疑似py 动态类型特性，导致一些nv.generate_plotly_code 方法无法跟踪
3.第三方规则不够，像ZhipuAI

1 和3都好解决，重点跟踪问题2 的原因，猜测可能原因
●方法调用call 无法定位到方法的实现function，最可能的原因是无法识别vn 的类型，导致无法跟踪

2. 步骤一：手动寻找vn.generate_plotly_code 对应的Call & Function是否存在

遇到第一个问题，无法发现vn.generate_plotly_code 这个Call

1
2
3
from Call_ e
where e.toString().matches("%generate_plotly_code%")
select e

但是Function 是存在的

upload successful
Function 存在，但是Call 不存在，需要进一步跟踪原因，猜测可能有以下原因
1.DB 构建的时候就没有这个Call_(为此跟踪了整个codeQL python DB 构建过程)
2.Function 和Call 关联的时候出问题了

省略问题1 中间的排查过程，结论如下：

upload successful

结论：DB 中所需的Call 是存在的，不过被识别为Attribute，有点类似Java 的Field，toString 统一为Attribute。。。

接下来，需要验证Function 和Call 的关联关系

3. 步骤二：寻找Call & Function 关联逻辑

在Java 中，为Call 和Callable，Python 有些差异，正常来讲，以java 为例，是Call 和Callable 是可以直接通过Call.getCallable 获取到对应的方法实现的，看看Python 中是否有这样的API

3.1 Call

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/** A call expression, such as `func(...)` */
class Call extends Call_ {
  /* syntax: Expr(...) */
  override Expr getASubExpression() {
    result = this.getAPositionalArg() or
    result = this.getAKeyword().getValue() or
    result = this.getFunc()
  }


/** INTERNAL: See the class `Call` for further information. */
class Call_ extends @py_Call, Expr {
  /** Gets the callable of this call expression. */
  Expr getFunc() { py_exprs(result, _, this, 2) }

注：
●这里是多继承，逻辑关系为取交集，也就是Call_ 是@py_Call 和 Expr 的交集，单独测试也是ok

1
2
3
4
py_exprs(unique int id : @py_expr,
    int kind: int ref,
    int parent : @py_expr_parent ref,
    int idx : int ref);

@py_Call 是最原始的基础类型，是直接作为基础类型在trap 文件中应用的

py_Call 小问题跟踪

upload successful

但是却没有py_Call.rel，难道是生成db 的时候漏了？
像py_Attribute 也是类似，在db 中并没有rel 文件，这是为什么？
实际上猜测是因为exprs 是一个集合，包含py_Call 和py_Attribute 等之类的

1
2
3
4
5
6
7
8
9
10
11
12
py_exprs(unique int id : @py_expr,
    int kind: int ref,
    int parent : @py_expr_parent ref,
    int idx : int ref);
    
case @py_expr.kind of
    0 = @py_Attribute
|   1 = @py_BinaryExpr
|   2 = @py_BoolExpr
|   3 = @py_Bytes
|   4 = @py_Call
|   5 = @py_ClassExpr

3.2 Function

1
2
3
4
/** INTERNAL: See the class `Call` for further information. */
class Call_ extends @py_Call, Expr {
  /** Gets the callable of this call expression. */
  Expr getFunc() { py_exprs(result, _, this, 2) }

虽然字面意思是获取对应callable，但是实际上是不是如此，只是获取对应的expr，举个例子

  • Call: nv.generate_plotly_code(question=…)
  • Call.getFunc(): nv.generate_plotly_code
  • Call.getFunc().(Attrbuite).getObject(): nv

此路不通，得去看最正宗的实现（也就是DFA 接口中的实现）

4. 步骤三：viableCallable 分析

viableCallable 是DFA 的接口API，直接分析Python 的实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/** Gets a viable run-time target for the call `call`. */
DataFlowCallable viableCallable(DataFlowCall call) {
  call instanceof ExtractedDataFlowCall and
  result = call.getCallable()
  or
  // A call to a library callable with a flow summary
  // In this situation we can not resolve the callable from the call,
  // as that would make data flow depend on type tracking.
  // Instead we resolve the call from the summary.
  exists(LibraryCallable callable |
    result = TLibraryCallable(callable) and
    call.getNode() = callable.getACall().getNode() and
    call instanceof PotentialLibraryCall
  )
}

upload successful

upload successful

可以看到：

  • DataFlowCall 有350条，筛选出来的和DataFlowCallable 关系只102条，那么中间是存在遗漏的

4.1 DataFlowCall & DataFlowCallable

看看DataFlowCall 的实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/** A call that is taken into account by the global data flow computation. */
abstract class DataFlowCall extends TDataFlowCall {
  /** Gets a textual representation of this element. */
  abstract string ();

  /** Get the callable to which this call goes. */
  abstract DataFlowCallable getCallable();

newtype TDataFlowCall =
  TNormalCall(CallNode call, Function target, CallType type) { resolveCall(call, target, type) } or
  TPotentialLibraryCall(CallNode call) or
  /** A synthesized call inside a summarized callable */
  TSummaryCall(
    FlowSummaryImpl::Public::SummarizedCallable c, FlowSummaryImpl::Private::SummaryNode receiver
  ) {
    FlowSummaryImpl::Private::summaryCallbackRange(c, receiver)
  }
  • 主要关TNormalCall，其中用到了resolveCall

resolveCall

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/**
 * Holds if `call` is a call to the `target`, with call-type `type`.
 */
cached
predicate resolveCall(CallNode call, Function target, CallType type) {
  Stages::DataFlow::ref() and
  (
    type instanceof CallTypePlainFunction and
    call.getFunction() = functionTracker(target).asCfgNode() and
    not exists(Class cls | cls.getAMethod() = target)
    or
    resolveMethodCall(call, target, type, _)
    or
    type instanceof CallTypeClass and
    exists(Class cls |
      resolveClassCall(call, cls) and
      target = invokedFunctionFromClassConstruction(cls, _)
    )
    or
    type instanceof CallTypeClassInstanceCall and
    resolveClassInstanceCall(call, target, _)
  )
}
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
predicate resolveMethodCall(CallNode call, Function target, CallType type, Node self) {
    (
      directCall(call, target, _, _, _, self)
      or
      callWithinMethodImplicitSelfOrCls(call, target, _, _, _, self)
      or
      fromSuper(call, target, _, _, _, self)
    ) and
    (
      // normal method call
      type instanceof CallTypeNormalMethod and
      (
        self = classInstanceTracker(_)
        or
        self = selfTracker(_)
      ) and
      not isStaticmethod(target) and
      not isClassmethod(target)
      or
      // method as plain function call
      type instanceof CallTypeMethodAsPlainFunction and
      self = classTracker(_) and
      not isStaticmethod(target) and
      not isClassmethod(target)
      or
      // staticmethod call
      type instanceof CallTypeStaticMethod and
      isStaticmethod(target)
      or
      // classmethod call
      type instanceof CallTypeClassMethod and
      isClassmethod(target)
    )
  }
}

callType

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
// =============================================================================
// call and argument resolution
// =============================================================================
newtype TCallType =
  /** A call to a function that is not part of a class. */
  CallTypePlainFunction() or
  /**
   * A call to an "normal" method on a class instance.
   * Does not include staticmethods or classmethods.
   */
  CallTypeNormalMethod() or
  /** A call to a staticmethod. */
  CallTypeStaticMethod() or
  /** A call to a classmethod. */
  CallTypeClassMethod() or
  /**
   * A call to method on a class, not going through an instance method, such as
   *
   * ```py
   * class Foo:
   *     def method(self, arg):
   *         pass
   *
   * foo = Foo()
   * Foo.method(foo, 42)
   * ```
   */
  CallTypeMethodAsPlainFunction() or
  /** A call to a class. */
  CallTypeClass() or
  /** A call on a class instance, that goes to the `__call__` method of the class */
  CallTypeClassInstanceCall()
  • CallTypePlainFunction: 调用最常见的方法，没有类修饰，例如json.dumps 就属于这种
    upload successful
  • CallTypeClass: 类初始化，也就是类的 init 方法
    upload successful
  • CallTypeClassInstanceCall：直接调用类方法，而不是从实例去调用，Demo 已经很详细了
    1
    2
    3
    4
    5
    class Foo:
         def method(self, arg):
             pass
    foo = Foo()
    Foo.method(foo, 42)
  • CallTypeNormalMethod: 普通的实例方法调用，需通过实例去调用，包括self
  • CallTypeMethodAsPlainFunction: 调用
    upload successful
  • CallTypeStaticMethod: 静态方法调用
    upload successful
  • CallTypeClassMethod: 具有classmethod 修饰的方法调用，包括new 等内置方法
    • classmethod 修饰符对应的函数不需要实例化，不需要 self 参数，但第一个参数需要是表示自身类的 cls 参数，可以来调用类的属性，类的方法，实例化对象等。

判断vn.generate_plotly_code 应该是属于directCall，type 类型为CallTypeNormalMethod，一步步跟进

directCall

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
/**
   * Holds if `call` is a call to a method `target` on an instance or class, where the
   * instance or class is not derived from an implicit `self`/`cls` argument to a method
   * -- for that, see `callWithinMethodImplicitSelfOrCls`.
   *
   * It is found by making an attribute read `attr` with the name `functionName` on a
   * reference to the class `cls`, or to an instance of the class `cls`. The reference the
   * attribute-read is made on is `self`.
   */
  pragma[nomagic]
  private predicate directCall(
    CallNode call, Function target, string functionName, Class cls, AttrRead attr, Node self
  ) {
    target = findFunctionAccordingToMroKnownStartingClass(cls, functionName) and
    directCall_join(call, functionName, cls, attr, self)
  }

directCall_join

1
2
3
4
5
6
7
8
9
/** Extracted to give good join order */
pragma[nomagic]
private predicate directCall_join(
  CallNode call, string functionName, Class cls, AttrRead attr, Node self
) {
  call.getFunction() = attrReadTracker(attr).asCfgNode() and
  attr.accesses(self, functionName) and
  self in [classTracker(cls), classInstanceTracker(cls)]
}

测试发现，这个条件是没问题的

  • attr.accesses(self, functionName)
    upload successful
    问题在于下面两个逻辑
  • self in [classTracker(cls), classInstanceTracker(cls)]
  • call.getFunction() = attrReadTracker(attr).asCfgNode()
    其中单步调试发现，这二个条件也不满足

4.2 问题一定位：classInstanceTracker

上文中的self in [classTracker(cls), classInstanceTracker(cls)]，self 是代码vn.generate_plotly_code 中的vn, 其原始赋值应该是从__init__(vn...) 中而来，cls 理应是VannaBase，猜测classInstanceTracker(VannaBase) 并不能传播至vn

patch classInstanceTracker(Class cls)

注：这个方法是跟踪Class cls 类能够传播到哪个实例Node 中，例如
●bar = Foo() // Class Foo 能传播到bar 这个Node 中

做个patch

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
private module TrackClassInstanceInput implements CallGraphConstruction::Simple::InputSig {
  class State = Class;

  predicate start(Node start, Class cls) {
    resolveClassCall(start.(CallCfgNode).asCfgNode(), cls)
    or
    // result of `super().__new__` as used in a `__new__` method implementation
    exists(Class classUsedInSuper |
      fromSuperNewCall(start.(CallCfgNode).asCfgNode(), classUsedInSuper, _, _) and
      classUsedInSuper = getADirectSuperclass*(cls)
    )
    // patched by m0d9, for such case
    // ```python
    // def foo(bar: cls):
    // ```
    // bar could be a start node
    or
    exists(Function f,Parameter p, int i|
      f.getArg(i) = p
      and p.getAnnotation().(Name).toString() = cls.getName()
      |
      start.asExpr() = p
    )
  }

测试增强之后的classInstanceTracker

1
2
3
4
predicate test_class_instance_track(Class c, Node n){
    n = classInstanceTracker(c)
    and c.getName() = "VannaBase"
}

upload successful

可以看到，classInstanceTracker(VannaBase) 是能够传播至参数vn的

再来试试directCall_join 中的逻辑
1.before

1
2
3
4
predicate test_attr_access(AttrRead attr, Node self, string functionName){
    attr.accesses(self, functionName) 
    and functionName = "generate_plotly_code"
}

upload successful
2.添加classInstanceTracker

1
2
3
4
5
predicate test_attr_access2(AttrRead attr, Node self, string functionName, Class cls){
    attr.accesses(self, functionName) 
    and functionName = "generate_plotly_code"
    and self in [classTracker(cls), classInstanceTracker(cls)]
}

结果为空，不行
3.对比classInstanceTracker 结果

upload successful

引入新问题：SynthCaptureNode VS ControlFlowNode

针对self in [classTracker(cls), classInstanceTracker(cls)]，getAQlClass 对比发现差异原因

upload successful

SynthCaptureNode && CapturedVariable && LocalVariable && ExprNode

1
2
3
4
5
6
SynthCaptureNode
    TSynthCaptureNode
    Node
        TNode
            ControlFlowNode
                @py_flow_node
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// semmle.python.dataflow.new.internal.DataFlowPrivate.qll
/**
 * A synthesized data flow node representing a closure object that tracks
 * captured variables.
 */
class SynthCaptureNode extends Node, TSynthCaptureNode {
  private VariableCapture::Flow::SynthesizedCaptureNode cn;

// shared.dataflow.codeql.dataflow.VariableCapture.qll
class SynthesizedCaptureNode extends ClosureNode, TSynthesizedCaptureNode {

private class TSynthesizedCaptureNode = TSynthRead or TSynthThisQualifier or TSynthPhi;
class ClosureNode extends TClosureNode {

private newtype TClosureNode =
    TSynthRead(CapturedVariable v, BasicBlock bb, int i, Boolean isPost) {
      synthRead(v, bb, i, _, _)
    } 

// semmle.python.dataflow.new.internal.VariableCapture.qll
class CapturedVariable extends LocalVariable {
    Function f;
1
2
3
4
5
6
7
ExprNode
    CfgNode
        TCfgNode
        Node
            TNode
            ControlFlowNode
                @py_flow_node
1
2
3
4
CapturedVariable
    LocalVariable
        Variable
            @py_variable

从SynthCaptureNode 到 ExprNode

  1. SynthCaptureNode 到 CapturedVariable
    1
    2
    3
    from SynthCaptureNode n, CapturedVariable v
    where 
        n.getSynthesizedCaptureNode().isVariableAccess(v)
  2. CapturedVariable 到 LocalVariable
  3. LocalVariable 到ExprNode

upload successful

  • attr 为SynthCaptureNode
  • v 为LocalVariable

4.3 问题二定位：attrReadTracker

attrReadTracker

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/** Gets a reference to the attribute read `attr` */
Node attrReadTracker(AttrRead attr) {
  CallGraphConstruction::Simple::Make<TrackAttrReadInput>::track(attr)
      .(LocalSourceNode)
      .flowsTo(result)
}

private module test_TrackAttrReadInput implements CallGraphConstruction::Simple::InputSig {
  class State = AttrRead;

  predicate start(Node start, AttrRead attr) {
    start = attr and
    // pragma[only_bind_into](attr.getObject()) in [
    //     classTracker(_), classInstanceTracker(_), selfTracker(_), clsArgumentTracker(_),
    //     superCallNoArgumentTracker(_), superCallTwoArgumentTracker(_, _)
    //   ]
    // and 
    attr.accesses(_, "generate_plotly_code")
    and attr.getLocation().getFile().getAbsolutePath().matches("%flask/__init__.py%")
  }

upload successful

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
pragma[nomagic]
private predicate test_directCall_join(
  CallNode call, string functionName, Class cls, AttrRead attr, Node self
) {
  (call.getFunction() = attrReadTracker(attr).asCfgNode() 
    or call.getFunction() = attr.asCfgNode()
  ) and
  attr.accesses(self, functionName) and
  // self in [classTracker(cls), classInstanceTracker(cls)]
  exists(Node clsis|
    clsis = [classTracker(cls), classInstanceTracker(cls)]
    |
    self = clsis
    // local variable
    or exists(LocalVariable v|
      clsis.(SynthCaptureNode).getSynthesizedCaptureNode().isVariableAccess(v)
      and self.asCfgNode().getNode().toString() = v.getId()
      and self.asCfgNode().getScope() = v.getAnAccess().getScope()
    )
  )
}

upload successful

是否有更优雅解？

4.4 更优解：variableCaptureLocalFlowStep？

flowsTo
1
2
3
4
5
6
7
LocalSources::flowsTo
    LocalSources::Cached::hasLocalSource
        LocalSources::Cached::localSourceFlowStep
            DataFlowPrivate::simpleLocalFlowStep
                simpleLocalFlowStepForTypetracking
                summaryLocalStep
                variableCaptureLocalFlowStep

upload successful
结果

upload successful

5. 其他问题

5.1 abstractmethod 抽象方法识别

Python 中的 abstractmethod 是在 abc 模块中引入的，该模块在Python 2.6 版本中引入，并随着Python 3 的发展而不断完善 https://blog.csdn.net/weixin_40907382/article/details/80277170 。abc 模块提供了抽象基类(Abstract Base Classes, ABC) 的支持，使得开发者能够定义具有抽象方法的类，从而实现更严格的面向对象设计。
简单来说，abstractmethod 是 abc 模块的一部分，用于标记抽象方法，这些方法必须在子类中被实现。它确保了子类必须提供特定功能的实现，提高了代码的可维护性和可扩展性。

upload successful

结论：当前CodeQL版本(v2.17.2) 支持

5.2 dataclass 数据类

https://docs.python.org/zh-cn/3.13/library/dataclasses.html
Python的dataclass在Python 3.7版本中被引入。它是通过dataclasses模块实现的，并且使用@dataclass装饰器来简化数据类的创建。
更详细的解释:
Python 3.7:
dataclass是作为Python标准库的一部分，在Python 3.7版本中引入的。
@dataclass:
这个装饰器可以自动生成一些特殊方法，例如__init__()，__repr__()，__eq__()等，这些方法通常是数据类所需要的。

upload successful

结论：当前CodeQL版本(v2.17.2) 不支持

初步思路：

  1. 需适配模拟init() 方法
  2. 需适配模拟field 的访问
  3. 需适配模拟str

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true