wait-embedded_mitm.py

# embedded_mitm_fixed.py (only show modified parts, integrated in your script)
import asyncio
import threading
import sys
import signal
import os
import json
from mitmproxy import http
from mitmproxy.options import Options
from mitmproxy.tools.dump import DumpMaster

OUTFILE = "session_token.txt"
TARGET_HOST = "discount.wxpapp.wechatpay.cn"

class SaveTokenAddon:
    def __init__(self, target_host=TARGET_HOST, outfile=OUTFILE):
        self.target_host = target_host
        self.outfile = outfile
        print(f"[SaveTokenAddon] init target_host={self.target_host} outfile={self.outfile}")

    def response(self, flow: http.HTTPFlow) -> None:
        try:
            print(f"[SaveTokenAddon] got request host={flow.request.host!r} pretty_host={flow.request.pretty_host!r} path={flow.request.path}")
            if self.target_host not in flow.request.pretty_host:
                return
            text = flow.response.get_text()
            print(f"[SaveTokenAddon] response length={len(text)} snippet={text[:200]!r}")
            data = json.loads(text)
        except Exception as e:
            print("[SaveTokenAddon] parse response error:", repr(e))
            return

        token = None
        if isinstance(data, dict):
            d = data.get("data") or {}
            if isinstance(d, dict):
                token = d.get("session_token") or d.get("sessionToken") or d.get("session-token")
            if not token:
                token = data.get("session_token") or data.get("token")

        if token:
            try:
                with open(self.outfile, "w", encoding="utf-8") as f:
                    f.write(token)
                print(f"[SaveTokenAddon] saved session_token -> {self.outfile}")
            except Exception as e:
                print("[SaveTokenAddon] write error:", e)
        else:
            print("[SaveTokenAddon] no token found in response")

def start_proxy(listen_host="127.0.0.1", listen_port=8080):
    opts = Options(listen_host=listen_host, listen_port=listen_port)

    # 创建一个新的 asyncio loop 并设为当前线程的 loop
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    # 将 loop 传给 DumpMaster，避免 RuntimeError: no running event loop
    m = DumpMaster(opts, event_loop=loop)
    addon = SaveTokenAddon()
    m.addons.add(addon)

    def run_master():
        try:
            print(f"[embedded_mitm] starting mitmproxy on {listen_host}:{listen_port}")
            # 有些 mitmproxy 版本把 m.run 实现为 coroutine，有些为普通函数；两种都兼容
            try:
                loop.run_until_complete(m.run())
            except TypeError:
                # 如果 m.run 不是 coroutine，直接调用（会阻塞）
                m.run()
        except KeyboardInterrupt:
            print("[embedded_mitm] KeyboardInterrupt caught in run_master")
        except Exception as e:
            print("[embedded_mitm] DumpMaster.run exception:", e)
        finally:
            try:
                # Shutdown 清理
                loop.run_until_complete(m.shutdown())
            except Exception:
                try:
                    m.shutdown()
                except Exception:
                    pass
            try:
                loop.stop()
            except Exception:
                pass
            print("[embedded_mitm] mitmproxy stopped")

    t = threading.Thread(target=run_master, daemon=True)
    t.start()
    return m, t