john
e5cbe80644
The proxy now forwards any api.weixin.qq.com/cgi-bin/* path (drafts, material upload, etc.), binds to the 10.10.0.1 tunnel address, and uses a 120s timeout. wechat-egress-fetch.mjs rewrites WeChat API URLs to the egress base when MEMIND_WECHAT_EGRESS_BASE_URL is set. Co-authored-by: Cursor <cursoragent@cursor.com>
66 lines
2.4 KiB
Python
Executable File
66 lines
2.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""105 固定 IP 出站：仅转发 api.weixin.qq.com/cgi-bin/* 到微信官方 API。"""
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
import os
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
ALLOWED_PREFIX = "/cgi-bin/"
|
|
TIMEOUT_SEC = 120
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
if self.path == "/healthz" or self.path.startswith("/healthz?"):
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/plain; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(b"ok")
|
|
return
|
|
self._proxy("GET")
|
|
|
|
def do_POST(self):
|
|
self._proxy("POST")
|
|
|
|
def _proxy(self, method):
|
|
parsed = urllib.parse.urlsplit(self.path)
|
|
if not parsed.path.startswith(ALLOWED_PREFIX):
|
|
self.send_error(403)
|
|
return
|
|
|
|
length = int(self.headers.get("Content-Length", "0")) if method == "POST" else 0
|
|
body = self.rfile.read(length) if length else None
|
|
url = "https://api.weixin.qq.com" + self.path
|
|
headers = {}
|
|
content_type = self.headers.get("Content-Type")
|
|
if content_type:
|
|
headers["Content-Type"] = content_type
|
|
req = urllib.request.Request(url, data=body, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=TIMEOUT_SEC) as resp:
|
|
payload = resp.read()
|
|
self.send_response(resp.status)
|
|
out_type = resp.headers.get("Content-Type", "application/json")
|
|
self.send_header("Content-Type", out_type)
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
except urllib.error.HTTPError as err:
|
|
payload = err.read()
|
|
self.send_response(err.code)
|
|
out_type = err.headers.get("Content-Type", "application/json")
|
|
self.send_header("Content-Type", out_type)
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
def log_message(self, fmt, *args):
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
host = os.environ.get("WECHAT_EGRESS_HOST", "10.10.0.1")
|
|
port = int(os.environ.get("WECHAT_EGRESS_PORT", "19090"))
|
|
HTTPServer((host, port), Handler).serve_forever()
|