Pickle反序列化

Learnd From: https://goodapple.top/archives/1069

什么是Pickle？

pickle是Python中一个能够序列化和反序列化对象的模块。和其他语言类似，Python也提供了序列化和反序列化这一功能，其中一个实现模块就是pickle。在Python中，*“Pickling”* 是将 Python 对象及其所拥有的层次结构转化为一个**二进制字节流**的过程，也就是我们常说的序列化，而 *“unpickling”* 是相反的操作，会将字节流转化回一个对象层次结构。

当然在Python 中并不止pickle一个模块能够进行这一操作，更原始的序列化模块如`marshal`，同样能够完成序列化的任务，不过两者的侧重点并不相同，`marshal`存在主要是为了支持 Python 的`.pyc`文件。现在开发时一般首选pickle。

pickle实际上可以看作一种**独立的语言**，通过对`opcode`的编写可以进行Python代码执行、覆盖变量等操作。直接编写的`opcode`灵活性比使用pickle序列化生成的代码更高，并且有的代码不能通过pickle序列化得到（pickle解析能力大于pickle生成能力）。

既然opcode能够执行Python代码，那自然就免不了安全问题。以下是Python在pickle文档中的警告。

常用opcode

在Python的pickle.py中，我们能够找到所有的opcode及其解释，常用的opcode如下，这里我们以V0版本为例

指令	描述	具体写法	栈上的变化

c	获取一个全局对象或import一个模块	c[module]\n[instance]\n	获得的对象入栈
o	寻找栈中的上一个MARK，以之间的第一个数据（必须为函数）为callable，第二个到第n个数据为参数，执行该函数（或实例化一个对象）	o	这个过程中涉及到的数据都出栈，函数的返回值（或生成的对象）入栈
i	相当于c和o的组合，先获取一个全局函数，然后寻找栈中的上一个MARK，并组合之间的数据为元组，以该元组为参数执行全局函数（或实例化一个对象）	i[module]\n[callable]\n	这个过程中涉及到的数据都出栈，函数返回值（或生成的对象）入栈
N	实例化一个None	N	获得的对象入栈
S	实例化一个字符串对象	S’xxx’\n（也可以使用双引号、'等python字符串形式）	获得的对象入栈
V	实例化一个UNICODE字符串对象	Vxxx\n	获得的对象入栈
I	实例化一个int对象	Ixxx\n	获得的对象入栈
F	实例化一个float对象	Fx.x\n	获得的对象入栈
R	选择栈上的第一个对象作为函数、第二个对象作为参数（第二个对象必须为元组），然后调用该函数	R	函数和参数出栈，函数的返回值入栈
.	程序结束，栈顶的一个元素作为pickle.loads()的返回值	.	无
(	向栈中压入一个MARK标记	(	MARK标记入栈
t	寻找栈中的上一个MARK，并组合之间的数据为元组	t	MARK标记以及被组合的数据出栈，获得的对象入栈
)	向栈中直接压入一个空元组	)	空元组入栈
l	寻找栈中的上一个MARK，并组合之间的数据为列表	l	MARK标记以及被组合的数据出栈，获得的对象入栈
]	向栈中直接压入一个空列表	]	空列表入栈
d	寻找栈中的上一个MARK，并组合之间的数据为字典（数据必须有偶数个，即呈key-value对）	d	MARK标记以及被组合的数据出栈，获得的对象入栈
}	向栈中直接压入一个空字典	}	空字典入栈
p	将栈顶对象储存至memo_n	pn\n	无
g	将memo_n的对象压栈	gn\n	对象被压栈
0	丢弃栈顶对象	0	栈顶对象被丢弃
b	使用栈中的第一个元素（储存多个属性名: 属性值的字典）对第二个元素（对象实例）进行属性设置	b	栈上第一个元素出栈
s	将栈的第一个和第二个对象作为key-value对，添加或更新到栈的第三个对象（必须为列表或字典，列表以数字作为key）中	s	第一、二个元素出栈，第三个元素（列表或字典）添加新值或被更新
u	寻找栈中的上一个MARK，组合之间的数据（数据必须有偶数个，即呈key-value对）并全部添加或更新到该MARK之前的一个元素（必须为字典）中	u	MARK标记以及被组合的数据出栈，字典被更新
a	将栈的第一个元素append到第二个元素(列表)中	a	栈顶元素出栈，第二个元素（列表）被更新
e	寻找栈中的上一个MARK，组合之间的数据并extends到该MARK之前的一个元素（必须为列表）中	e	MARK标记以及被组合的数据出栈，列表被更新

ISCTF2024【新闻系统】

前面就是session 伪造不多说

源码：

from flask import *
import pickle
import base64

app = Flask(__name__)
app.config["SECRET_KEY"] = "W3l1com_isCTF"

class News:
    def __init__(self, title, content) -> None:
        self.title = title
        self.content = content

    def __repr__(self) -> str:
        return f"news(name={self.title}, words={self.content})"

class NewsList:
    def __init__(self) -> None:
        self.news_list = []

    def create_news(self, title, content) -> None:
        news = News(title,content)
        self.news_list.append(news)
    
    def export_news(self, news_title) -> str | None:
        news = self.get_news(news_title)
        if news is not None:
            self.news_list.remove(news)
            return '删除成功'
        return None
    
    def add_news(self, serialized_news) -> None:
        try:
            news_data = base64.b64decode(serialized_news)
            black_list = ['create_news','export_news','add_news','get_news']
            for i in black_list:
                if i in str(news_data):
                    return False
            news = pickle.loads(news_data) #使用 pickle.loads 将解码后的数据反序列化成 Python 对象
           
            if isinstance(news,News):
                for i in self.news_list:
                    if i.title == news.title:
                        return False
                self.news_list.append(news)
                return True
            return False
        except Exception:
            return False
    
    def get_news(self, news_title) -> News | None:
        for news in self.news_list:
            if str(news.title) == news_title:
                return news
        return None


newslist = NewsList()

@app.route("/")
def index():
    return redirect("/login")

@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        username = request.form.get('username')
        password = request.form.get('password')
        if username == 'test' and password == 'test111':
            session['username'] = username
            session['password'] = password
            session['status'] = 'user'
            return redirect('/news')
        else:
            session['login_error'] = True               
    return render_template("login.html")

@app.route("/news")
def news():
    news = newslist.news_list
    return render_template("news.html",news = news)

@app.route('/admin',methods=['GET','POST'])
def admin():
    if session.get('status') != 'admin' or session.get('username') != 'admin' or session.get('password') != 'admin222':
        return redirect("/login")
    news = newslist.news_list
    return render_template("admin.html",news = news)
    
@app.route("/create", methods=["POST"])
def create_news():
    if session.get('status') != 'admin' or session.get('username') != 'admin' or session.get('password') != 'admin222':
        return redirect("/login")   
    title = request.form.get('title')
    content = request.form.get('content')
    newslist.create_news(title,content)
    return redirect("/admin")

@app.route("/export", methods=["POST"])
def export_news():
    if session.get('status') != 'admin' or session.get('username') != 'admin' or session.get('password') != 'admin222':
        return redirect("/login")
    news_title = request.form["title"]
    result = newslist.export_news(news_title)
    if result is not None:
        return jsonify({"result": result})
    else:
        return jsonify({"error": "news not found"})

@app.route("/add", methods=["POST"])
def add_news():
    if session.get('status') != 'admin' or session.get('username') != 'admin' or session.get('password') != 'admin222':
        return redirect("/login")    
    serialized_news = request.form["serialized_news"]
    if newslist.add_news(serialized_news):
        return redirect("/admin")
    else:
        return jsonify({"error": "Failed to add news"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8888, debug=False, threaded=True)

伪造进入/admin

news_data = base64.b64decode(serialized_news)

serialized_news = request.form["serialized_news"]
    if newslist.add_news(serialized_news):
        return redirect("/admin")

这里可以看到add路由可以进行反序列化

参考了 https://xz.aliyun.com/t/16227?time__1311=GuD%3DPRxGrh8D%2FWNiQGkDuQrKj8DmE%2BbD#toc-7

自己暂时还不会构造 然后 可以得到flag