公网穿透原理 · 可复现指南

场景：阿里云 DSW 实例（无公网 IP）上跑 LLM 推理服务（llama-server / vLLM），需要任何设备、无需阿里云登录 Cookie 直接调用。
本文档从原理到脚本完整可复现，所有组件均已在本机与实例上实测。

一、为什么需要"穿透"

结论：从实例内主动建立反向隧道把服务"推出去"。

二、隧道选型（免费方案实测）

设计思路：固定地址 + 长请求能力 = 两条隧道分工

外部设备
    │
    ▼
[serveo 固定前门 qwen38fn.serveousercontent.com]
    │  8097 redirect_front.py
    │  收到请求 → 307 重定向 → 毫秒级完成，永不触发隧道超时
    ▼
[localhost.run 随机长通道 6aab2cca7c7a64.lhr.life]
    │  8098 hb_proxy.py（心跳代理）
    │  非流式：先发 HTTP 头 + 每 2 秒空白 chunk 保活，最后补 JSON
    │  流式：边读上游边转发，2s 无数据发 SSE 注释 ": hb"
    ▼
[llama-server / vLLM  8099]   ← API 密钥校验在此

三、三个脚本（完整可复现）

3.1 心跳代理 hb_proxy.py（实例内 8098）

解决的问题：免费 SSH 隧道对"连接后无数据流动"约 5 秒判定死亡，但 LLM 推理 prefill 阶段可达 60~100 秒无输出。

原理（两种请求模式）：

非流式：

1. 立刻发 HTTP 200 头 + Transfer-Encoding: chunked（告诉客户端"正在传数据"）
2. 开后台线程请求上游 llama-server（3600s 超时）
3. 主循环每 2 秒发一个 1 字节空白 chunk 保活（JSON 规范允许前导空白）
4. 上游完成后，把完整 JSON 作为最后一个 chunk 发出
5. 客户端 httpx / requests / curl / OpenAI SDK 均正确解析（忽略前导空白）

流式：

1. 立刻发 200 + Content-Type: text/event-stream + chunked
2. 后台线程读上游 SSE 事件 → 队列
3. 主循环从队列取，有数据立即转发；2 秒无数据发一条 SSE 注释行 ": hb"
   （SSE 规范：以 ":" 开头的行被客户端忽略，不影响消息流）
4. 上游 EOF 后发终止 chunk

密钥传递（为 307 跳转设计）：

跨主机 307 重定向会丢失 Authorization 头（httpx/requests/curl 均如此，安全设计），
但不会丢 URL 路径。所以本代理支持 /k/<key>/v1/chat/completions 形式：
  - 解析 URL 中 /k/<key> 前缀
  - 剥掉前缀，把 key 注入 Authorization: Bearer <key>
  - 再转发到上游
两种入参均兼容：标准 Bearer 头 或 /k/<key>/ 路径前缀

完整代码：

#!/usr/bin/env python3
# 心跳代理：让请求穿过对"静默连接"有 5s 超时的免费 SSH 隧道
# 非流式 → chunked 空白保活 + 上游完成后补 JSON
# 流式   → 逐行转发 SSE，2s 无数据发 ": hb" 注释行保活
# 密钥   → /k/<key>/... 路径形式（为 307 跨主机跳转设计）
# 部署：隧道指向 127.0.0.1:8098（本代理），本代理转发到 8099（llama-server）
import json, queue, re, threading, urllib.parse, urllib.request, urllib.error, http.server

UPSTREAM = 'http://127.0.0.1:8099'
LISTEN = ('127.0.0.1', 8098)
SKIP_REQ = {'host', 'connection', 'content-length', 'accept-encoding', 'transfer-encoding'}
KEYPATH = re.compile(r'^/k/([^/]+)(/.*)?$')


class Handler(http.server.BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'
    server_version = 'hbproxy/1.0'

    def log_message(self, fmt, *args):
        pass

    def _fwd_headers(self):
        return {k: v for k, v in self.headers.items() if k.lower() not in SKIP_REQ}

    def _target(self):
        """识别 /k/<key>/... → 剥前缀并注入 Authorization 头"""
        path, key = self.path, None
        m = KEYPATH.match(self.path)
        if m:
            key = urllib.parse.unquote(m.group(1))
            path = m.group(2) or '/'
        hdrs = self._fwd_headers()
        if key:
            hdrs['Authorization'] = 'Bearer ' + key
        return path, hdrs

    def do_GET(self):
        try:
            path, hdrs = self._target()
            req = urllib.request.Request(UPSTREAM + path, headers=hdrs)
            with urllib.request.urlopen(req, timeout=300) as r:
                body, code = r.read(), r.status
        except urllib.error.HTTPError as e:
            body, code = e.read(), e.code
        except Exception as e:
            body, code = json.dumps({'error': str(e)}).encode(), 502
        self.send_response(code)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        n = int(self.headers.get('Content-Length') or 0)
        raw = self.rfile.read(n)
        try:
            payload = json.loads(raw.decode('utf-8') or '{}')
            if not isinstance(payload, dict):
                payload = {}
        except Exception:
            payload = {}
        path, hdrs = self._target()
        hdrs['Content-Type'] = 'application/json'

        if payload.get('stream'):
            # 流式：逐行转发 + SSE 注释保活
            up = dict(payload); up['stream'] = True
            req = urllib.request.Request(UPSTREAM + path, data=json.dumps(up).encode(), headers=hdrs)
            try:
                r = urllib.request.urlopen(req, timeout=3600)
            except urllib.error.HTTPError as e:
                b = e.read()
                self.send_response(e.code)
                self.send_header('Content-Length', str(len(b)))
                self.end_headers()
                self.wfile.write(b); return
            except Exception as e:
                b = json.dumps({'error': str(e)}).encode()
                self.send_response(502); self.send_header('Content-Length', str(len(b)))
                self.end_headers(); self.wfile.write(b); return

            self.send_response(200)
            self.send_header('Content-Type', 'text/event-stream')
            self.send_header('Transfer-Encoding', 'chunked')
            self.end_headers()
            q = queue.Queue()

            def reader():
                try:
                    while True:
                        line = r.readline()
                        if not line:
                            break
                        q.put(line)
                except Exception:
                    pass
                q.put(None)

            threading.Thread(target=reader, daemon=True).start()
            try:
                while True:
                    try:
                        line = q.get(timeout=2)
                    except queue.Empty:
                        # 2s 无数据 → 发 SSE 注释行保活
                        self.wfile.write(b'9\r\n: hb hb\n\n\r\n'); self.wfile.flush()
                        continue
                    if line is None:
                        break
                    self.wfile.write(('%x\r\n' % len(line)).encode() + line + b'\r\n')
                    self.wfile.flush()
            except Exception:
                pass
            try:
                self.wfile.write(b'0\r\n\r\n'); self.wfile.flush()
            except Exception:
                pass
            return

        # 非流式：立即发头 + 心跳保活，上游完成后补 JSON
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Transfer-Encoding', 'chunked')
        self.end_headers()
        done = {}

        def work():
            try:
                up = dict(payload); up['stream'] = False
                req = urllib.request.Request(UPSTREAM + path, data=json.dumps(up).encode(), headers=hdrs)
                with urllib.request.urlopen(req, timeout=3600) as r:
                    done['body'] = r.read()
            except urllib.error.HTTPError as e:
                done['body'] = e.read()
            except Exception as e:
                done['body'] = json.dumps({'error': str(e)}).encode()

        t = threading.Thread(target=work, daemon=True)
        t.start()
        hb = b'\n'
        try:
            while t.is_alive():
                self.wfile.write(('%x\r\n' % len(hb)).encode() + hb + b'\r\n')
                self.wfile.flush()
                t.join(2)
        except Exception:
            return
        body = done.get('body') or b'{}'
        try:
            self.wfile.write(('%x\r\n' % len(body)).encode() + body + b'\r\n')
            self.wfile.write(b'0\r\n\r\n')
            self.wfile.flush()
        except Exception:
            pass


if __name__ == '__main__':
    srv = http.server.ThreadingHTTPServer(LISTEN, Handler)
    print('hbproxy listening on %s -> %s' % (LISTEN, UPSTREAM), flush=True)
    srv.serve_forever()

部署：

# 实例内
setsid nohup python3 /root/hb_proxy.py > /root/hb_proxy.out 2>&1 &
# 验证
curl http://127.0.0.1:8098/health   # 应返回 {"status":"ok"}

3.2 重定向前门 redirect_front.py（实例内 8097）

解决的问题：serveo 固定地址只能做毫秒级响应（长请求 5s 即断），需要一个"稳定入口"把请求转给能扛长请求的 localhost.run 通道。

原理：

收到任何请求（GET/POST/PUT/DELETE）
    │
    ├── 如果是 GET /current → 直接返回当前 lhr.life 地址（几十字节，不触发上限）
    │
    └── 其他请求 → 307 临时重定向到当前 localhost.run 通道
          关键细节：
          ① 必须先读完请求体再发 307
             （否则 keep-alive 连接上残留的请求体会被当成下一个请求行，
              触发 "Bad request syntax" 400，实测长流式请求经此即失败）
          ② 密钥传递：把 Authorization: Bearer <key> 转成 URL 路径 /k/<key>/...
             （跨主机 307 丢 Authorization 头，但不会丢路径；
              对端 hb_proxy 解析 /k/<key>/ 前缀并重新注入 Authorization）

完整代码：

#!/usr/bin/env python3
# 稳定前门：任何请求 → 307 重定向到当前 localhost.run 长请求通道
# 设计: serveo 隧道指向 127.0.0.1:8097（本服务），本服务读 /root/lhr_url.txt 得知目标
# 密钥: 入参支持 ① 标准 Authorization 头  ② /k/<key>/ 路径前缀（为跨主机跳转设计）
import json, re, urllib.parse, http.server

CUR = '/root/lhr_url.txt'   # 由 tunnel_watch.sh 每次隧道重连时写入
KEYPATH = re.compile(r'^/k/[^/]+/')


def current_base():
    try:
        with open(CUR) as f:
            u = f.read().strip()
        return u if u.startswith('http') else None
    except Exception:
        return None


class Handler(http.server.BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'
    server_version = 'frontdoor/1.0'

    def log_message(self, fmt, *args):
        pass

    def _target_path(self):
        """已有 /k/<key>/ 则原样；若用标准 Authorization，把 key 转成路径前缀"""
        if KEYPATH.match(self.path):
            return self.path
        auth = self.headers.get('Authorization') or ''
        if auth.lower().startswith('bearer '):
            k = urllib.parse.quote(auth[7:].strip(), safe='')
            if k:
                return '/k/' + k + self.path
        return self.path

    def _redirect(self):
        # 关键：必须先读完请求体再发 307（keep-alive 连接残留体 → 400）
        n = int(self.headers.get('Content-Length') or 0)
        if n > 0:
            try:
                self.rfile.read(n)
            except Exception:
                pass
        elif (self.headers.get('Transfer-Encoding') or '').lower() == 'chunked':
            try:
                while True:
                    line = self.rfile.readline().strip()
                    size = int(line.split(b';')[0] or b'0', 16)
                    if size == 0:
                        self.rfile.readline()
                        break
                    self.rfile.read(size)
                    self.rfile.readline()
            except Exception:
                pass

        base = current_base()
        if not base:
            self.send_response(503)
            body = json.dumps({'error': 'upstream not ready'}).encode()
            self.send_header('Content-Type', 'application/json')
            self.send_header('Content-Length', str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return

        target = base + self._target_path()
        self.send_response(307)
        self.send_header('Location', target)
        self.send_header('Cache-Control', 'no-store')
        self.send_header('Content-Length', '0')
        self.end_headers()

    def do_GET(self):
        # /current 端点：返回当前长通道地址（几十字节，远低于 serveo ~128KB 上限）
        if self.path.split('?')[0].rstrip('/') in ('/current', '/k/current'):
            base = current_base()
            body = json.dumps({
                'current_long_channel': base or '',
                'hint': '这是可承载长请求的直连地址; 固定名地址只做跳转'
            }).encode()
            self.send_response(200 if base else 503)
            self.send_header('Content-Type', 'application/json')
            self.send_header('Content-Length', str(len(body)))
            self.send_header('Cache-Control', 'no-store')
            self.end_headers()
            self.wfile.write(body)
            return
        self._redirect()

    do_POST = _redirect
    do_PUT = _redirect
    do_DELETE = _redirect


if __name__ == '__main__':
    srv = http.server.ThreadingHTTPServer(('127.0.0.1', 8097), Handler)
    print('frontdoor listening on 127.0.0.1:8097', flush=True)
    srv.serve_forever()

部署：

setsid nohup python3 /root/redirect_front.py > /root/redirect_front.out 2>&1 &
# 验证（lhr_url.txt 有值时）
curl -s http://127.0.0.1:8097/current
# → {"current_long_channel":"https://xxxx.lhr.life","hint":"..."}

3.3 隧道守护 tunnel_watch.sh（实例内常驻）

职责：

1. 维持两条 SSH 反向隧道（断线 5s 自动重连）
2. 从隧道日志提取当前地址，写入 /root/lhr_url.txt
3. 地址变化时推送到 ntfy.sh（任何设备打开即可查最新端点，不刷屏）

#!/bin/bash
# 双隧道守护：A=localhost.run(长请求)  B=serveo(固定前门)  + 地址发布到 ntfy.sh
# 用法: setsid nohup /root/tunnel_watch.sh [serveo子域名] > /root/tunnel_daemon.log 2>&1 &
#       serveo 子域名：不注册则随机，注册后固定为 https://<子域名>.serveo.net
set -u
SUB="${1:-qwen38fn}"                    # serveo 固定子域名（未注册则用随机地址）
NTFY_TOPIC="${NTFY_TOPIC:-qwen38fn-kteax-7f3a9c}"   # ntfy.sh 公开主题名
KEY=/root/.ssh/id_ed25519               # 隧道 SSH 密钥（存 NAS 可跨实例恢复）
LOG=/root/tunnel_watch.log
LHR_LOG=/root/lhr_cur.log
SER_LOG=/root/serveo_cur.log
LHR_URL=/root/lhr_url.txt
SER_URL=/root/serveo_url.txt
LAST_PUB=/root/ntfy_last.txt
# SSH 保活参数：ServerAliveInterval=30 每 30s 发心跳；CountMax=3 三次无响应断线
SSH_OPT="-o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \
        -o ServerAliveInterval=30 -o ServerAliveCountMax=3 \
        -o ExitOnForwardFailure=yes -o ConnectTimeout=15"

log() { echo "$(date '+%F %T') $*" >> "$LOG"; }
log "守护启动 (serveo 子域名=$SUB, 发布主题=$NTFY_TOPIC)"

# ---- 通道 A：localhost.run（长请求，随机地址）----
(
  while true; do
    : > "$LHR_LOG"
    # -R 80:localhost:8098 → 把本代理 8098 通过随机 .lhr.life 端口 80 暴露出去
    ssh $SSH_OPT -i "$KEY" -R 80:localhost:8098 [email protected] >> "$LHR_LOG" 2>&1
    log "A 通道(localhost.run)断开, 5s 后重连"
    sleep 5
  done
) &

# ---- 通道 B：serveo（固定前门，只做 307 跳转）----
(
  while true; do
    : > "$SER_LOG"
    # -R ${SUB}:80:localhost:8097 → 把 8097 重定向前门暴露为 https://<SUB>.serveousercontent.com
    ssh $SSH_OPT -i "$KEY" -R "${SUB}:80:localhost:8097" serveo.net >> "$SER_LOG" 2>&1
    log "B 通道(serveo)断开, 5s 后重连"
    sleep 5
  done
) &

# ---- 地址提取 + 健康自检 + 变化时发布 ----
(
  while true; do
    # 从隧道日志提取最新地址
    u=$(grep -oE 'https://[a-z0-9]+\.lhr\.life' "$LHR_LOG" 2>/dev/null | tail -1)
    s=$(grep -oE 'https://[a-zA-Z0-9.-]+\.serveousercontent\.com' "$SER_LOG" 2>/dev/null | tail -1)
    [ -n "$u" ] && echo "$u" > "$LHR_URL"
    [ -n "$s" ] && echo "$s" > "$SER_URL"

    # 健康检查：curl /health（经隧道走完整路径，验证全链路）
    if [ -n "$u" ]; then
      h=$(curl -s -m 8 "$u/health" 2>/dev/null)
      case "$h" in *ok*) st=正常 ;; *) st=异常 ;; esac
      cur="$u|$s"

      # 只在地址变化时推送（不刷屏）
      if [ "$(cat "$LAST_PUB" 2>/dev/null)" != "$cur" ]; then
        curl -s -m 10 \
          -H "Title: Qwen3.8-Flash-Next 端点" \
          -H "Tags: robot" \
          -d "长请求通道(直接用这个): $u
稳定前门(307重定向): ${s:-未就绪}
服务状态: $st
更新时间: $(date '+%F %T')" \
          "https://ntfy.sh/$NTFY_TOPIC" > /dev/null 2>&1
        echo "$cur" > "$LAST_PUB"
        log "已发布端点: $u (前门 ${s:-无})"
      fi
    fi
    sleep 15
  done
) &
wait

部署：

# 首次生成 SSH 密钥（存 NAS 可跨实例恢复）
ssh-keygen -t ed25519 -N "" -f /root/.ssh/id_ed25519
cp /root/.ssh/id_ed25519 /mnt/workspace/qwen38fn/tunnel_key

# 启动守护（serveo 子域名 qwen38fn；已注册固定名可传参）
setsid nohup /root/tunnel_watch.sh qwen38fn > /root/tunnel_daemon.log 2>&1 &

# 查看当前地址
cat /root/lhr_url.txt       # 长请求通道（直接用这个）
cat /root/serveo_url.txt   # 固定前门（307 跳转）

# 任何设备查询最新地址（浏览器打开）
# https://ntfy.sh/qwen38fn-kteax-7f3a9c

四、完整链路（从外部到推理引擎）

外部设备（任意）
  │  调用: https://qwen38fn.serveousercontent.com/v1/chat/completions
  │        （固定地址，Authorization: Bearer <key>）
  │
  ▼
[serveo 隧道 B]  ──  ssh -R qwen38fn:80:localhost:8097  ──
  │
  ▼
[8097 redirect_front.py]
  │  读完请求体 → 307 Location: https://xxxx.lhr.life/k/<key>/v1/chat/completions
  │  （key 从 Authorization 转成 URL 路径，跨主机不丢）
  │
  ▼
[localhost.run 隧道 A]  ──  ssh -R 80:localhost:8098  ──
  │
  ▼
[8098 hb_proxy.py]
  │  解析 /k/<key>/ 前缀 → 注入 Authorization 头
  │  非流式: chunked 空白保活 + 补 JSON
  │  流式:   SSE 逐行转发 + 2s 注释保活
  │
  ▼
[8099 llama-server / vLLM]
     校验 API 密钥 → 推理 → 响应

五、客户端调用方式（任何设备）

curl：

# 方法 1：直接走长请求通道（需从 ntfy.sh 查当前地址）
# 先查地址：curl -s https://ntfy.sh/qwen38fn-kteax-7f3a9c | grep "lhr.life"
LHR=$(curl -s https://ntfy.sh/qwen38fn-kteax-7f3a9c | grep -oP 'https://\S+\.lhr\.life' | tail -1)

curl "${LHR}/v1/chat/completions" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <key>" \
  -d '{"model":"qwen38","messages":[{"role":"user","content":"你好"}],"max_tokens":64}'

# 方法 2：走固定前门（自动 307 跳转，客户端无需知道长通道地址）
curl -L "https://qwen38fn.serveousercontent.com/v1/chat/completions" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <key>" \
  -d '{"model":"qwen38","messages":[{"role":"user","content":"你好"}],"max_tokens":64}'
# 注意：-L 让 curl 跟随 307；请求体会被保留（httpx/requests 行为一致）

Python OpenAI SDK：

from openai import OpenAI

# 固定前门方式（推荐，地址不变）
c = OpenAI(
    base_url="https://qwen38fn.serveousercontent.com/v1",
    api_key="<key>"
)
resp = c.chat.completions.create(
    model="qwen38",
    messages=[{"role": "user", "content": "你好"}],
    max_tokens=64
)
print(resp.choices[0].message.content)

OpenAI 兼容客户端（通用）：

Base URL:  https://qwen38fn.serveousercontent.com/v1
API Key:   <key>
Model:     qwen38（服务端单模型，名称随意）

六、新实例快速恢复（实例 8h 重启后）

# 1. 拉取持久化文件（NAS 不受实例重启影响）
mkdir -p /root/qwen38fn
cp /mnt/workspace/qwen38fn/tunnel_key /root/.ssh/id_ed25519
cp /mnt/workspace/qwen38fn/hb_proxy.py /root/
cp /mnt/workspace/qwen38fn/redirect_front.py /root/
cp /mnt/workspace/qwen38fn/tunnel_watch.sh /root/

# 2. 起推理引擎（llama-server 或 vllm）
# 3. 起三个服务
setsid nohup python3 /root/hb_proxy.py       > /root/hb_proxy.out 2>&1 &
setsid nohup python3 /root/redirect_front.py > /root/redirect_front.out 2>&1 &
setsid nohup /root/tunnel_watch.sh qwen38fn  > /root/tunnel_daemon.log 2>&1 &

# 4. 验证（地址写入 /root/lhr_url.txt 后）
sleep 15
cat /root/lhr_url.txt          # 应显示 https://xxxx.lhr.life
curl -s http://127.0.0.1:8097/current   # 应返回当前地址

七、已知坑与验证清单

验证清单（每次部署后逐项确认）：

# 1. 本机能访问
curl -s http://127.0.0.1:8099/health

# 2. 心跳代理通
curl -s http://127.0.0.1:8098/health

# 3. 前门通（需 lhr_url.txt 有值）
curl -s http://127.0.0.1:8097/current

# 4. 隧道链路通（外部地址）
curl -s "$(cat /root/lhr_url.txt)/health"

# 5. 鉴权（llama-server 约定：HTTP 200 + 错误体，不产生推理）
curl -s "$(cat /root/lhr_url.txt)/v1/chat/completions" \
  -H "Content-Type: application/json" \
  -d '{"model":"qwen38","messages":[{"role":"user","content":"hi"}],"max_tokens":8}'
# 无 key → 200 + {"error":{"message":"Invalid API Key","code":401}}
# 有 key → 200 + 正常 JSON

# 6. ntfy 推送（新地址触发后 15s 内）
curl -s https://ntfy.sh/qwen38fn-kteax-7f3a9c?limit=3

八、serveo 固定子域名注册（可选，一次性）

未注册时 serveo 地址随机（https://xxxx.serveousercontent.com），注册后固定为 https://qwen38fn.serveo.net。

注册入口（用当前隧道密钥指纹）：
  https://console.serveo.net/ssh/keys?add=<SHA256指纹>

  获取当前指纹：ssh-keygen -lf /root/.ssh/id_ed25519.pub
  示例：SHA256:5IJg/QOYHhgTP+JDKQWWrMtIzpt7hOYVGg7dSqcWaro

注册后，隧道命令改为：
  ssh ... -R "qwen38fn:80:localhost:8097" serveo.net
  → 固定前门地址变为 https://qwen38fn.serveo.net（永不变）

注意：未注册固定名时，serveo 前门地址也会随机变化，ntfy.sh 会同时发布两个地址；
客户端用 /current 端点或 ntfy.sh 均可获取最新的"可直接用"长请求通道。