feat(ws): tunnel WS forward — 推論結果經 tunnel 推回雲端(後端塊2)
推論工作區後端塊2:實作 WS forward,讓 local agent 的推論結果 WS (inference:<deviceId>)經 tunnel 推回雲端瀏覽器 canvas overlay。照 edge-ai-platform POC relay/server.go proxyWebSocket 移植(唯讀參考)。 - forwarder.go ForwardWebSocket:OpenStream→寫 upgrade→讀 101→WebSocketConn (新增 WebSocketConn/wsUpgradeError/AsWSUpgradeError) - proxy.go newWebSocketProxyHandler:hijack browser→回寫 101→雙向 io.Copy cross-close(無 goroutine leak)+ copyWebSocketUpgradeHeaders(保留 Sec-WebSocket-*、剝 Authorization/Origin) - camera.go registerWebSocketRoutes:GET /ws/devices/:id/inference - api.go wsAuthGroup(/ws + AuthMiddleware):same-origin cookie 認證, 無 token-in-URL(security 定案) - stubs.go:移除 WS inference 501 stub、更新過時 doc comment(Mi-1/2) 架構差異:POC 單 binary,visionA api-server(auth)+remote-proxy 雙 binary, remote-proxy raw byte pipe 透明穿過 WS upgrade bytes。local agent 端未動。 認證/授權(security 定案 + S1/S2):same-origin cookie、剝 Auth/Origin、 WS 不套 300s timeout、走 pickActiveSessionToken(帶別人 deviceId 也只打到 自己 tunnel;多租戶嚴格綁定屬 Phase 1 M2 debt)。 Reviewer 0C/1M/3Mi 通過(修後)。Major-1 已修:收窄 all_endpoints_require_auth_test 的 /ws/ 白名單,讓 authed inference WS 納入「無 cookie 應 401」回歸檢查 (附守得住證明:移除 auth 測試即 FAIL)。+forwarder 5 test + camera_ws 端到端四層真連線雙向 pipe。build/vet/全回歸綠。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
ba8fd454ea
commit
dc6ca211ae
@ -43,23 +43,41 @@ import (
|
|||||||
// - /api/auth/callback → OIDC IdP 302 回來,user 還沒登入
|
// - /api/auth/callback → OIDC IdP 302 回來,user 還沒登入
|
||||||
// - /api/pairing/exchange → agent 還沒 session token,用 pairing token 換
|
// - /api/pairing/exchange → agent 還沒 session token,用 pairing token 換
|
||||||
//
|
//
|
||||||
|
// /ws/* stub endpoints(仍是 501、尚未實作 WebSocket proxy、註冊在 r 而非 wsAuthGroup,
|
||||||
|
// 故不走 AuthMiddleware)也逐條列在這裡。**逐條精確列出**、而非用 /ws/ prefix 一次放行,
|
||||||
|
// 是為了讓「已升級成 authed 的 WS route」(如 GET /ws/devices/:id/inference,掛在
|
||||||
|
// wsAuthGroup、走 same-origin cookie AuthMiddleware)自動落入主測試的「必須 401」檢查。
|
||||||
|
// 未來任一條 stub 補實作並套 auth 後,把它從這份清單移除即可納入回歸檢查。
|
||||||
|
//
|
||||||
// 任何往這份清單裡新加 endpoint 的 PR 都該特別 review — 你正在繞過 OIDC 保護。
|
// 任何往這份清單裡新加 endpoint 的 PR 都該特別 review — 你正在繞過 OIDC 保護。
|
||||||
var publicPaths = map[string]bool{
|
var publicPaths = map[string]bool{
|
||||||
"GET /healthz": true,
|
"GET /healthz": true,
|
||||||
"GET /api/auth/login": true,
|
"GET /api/auth/login": true,
|
||||||
"GET /api/auth/callback": true,
|
"GET /api/auth/callback": true,
|
||||||
"POST /api/pairing/exchange": true,
|
"POST /api/pairing/exchange": true,
|
||||||
|
|
||||||
|
// /ws/* 仍為 501 stub 的 endpoint(見 internal/api/stubs.go registerWebSocketStubs)。
|
||||||
|
// 注意:GET /ws/devices/:id/inference 已升級為 authed(wsAuthGroup),**刻意不在此清單**,
|
||||||
|
// 因此會被主測試納入「無 cookie 應回 401」的檢查。
|
||||||
|
"GET /ws/devices/events": true,
|
||||||
|
"GET /ws/devices/:id/flash-progress": true,
|
||||||
|
"GET /ws/server-logs": true,
|
||||||
|
"GET /ws/system": true,
|
||||||
|
"GET /ws/clusters/:id/inference": true,
|
||||||
|
"GET /ws/clusters/:id/flash-progress": true,
|
||||||
|
"GET /ws/pairing/status": true,
|
||||||
}
|
}
|
||||||
|
|
||||||
// publicPrefixes 是「整個 path prefix 都不走 OIDC AuthMiddleware」的清單。
|
// publicPrefixes 是「整個 path prefix 都不走 OIDC AuthMiddleware」的清單。
|
||||||
//
|
//
|
||||||
// - /storage/* — 用 HMAC presigned URL 驗簽(api-spec.md §10),不是 cookie
|
// - /storage/* — 用 HMAC presigned URL 驗簽(api-spec.md §10),不是 cookie
|
||||||
// - /ws/* — 雛形 stub 一律 501,註冊在 r 而非 apiGroup(stubs.go:70-85)。
|
//
|
||||||
// 目前無認證 → 501;**未來補實作 WebSocket proxy 時必須套 auth**,
|
// 為什麼這裡「不」放 /ws/:/ws/* 底下同時有 501 stub(無 auth)與已升級為 authed 的
|
||||||
// 屆時應從這份清單移除。TODO(B7): 移到 protected。
|
// route(/ws/devices/:id/inference,掛 wsAuthGroup)。用 prefix 一次放行會連 authed
|
||||||
|
// route 一起豁免、讓回歸測試對它失效。故 /ws/ 的 stub 逐條列在 publicPaths、authed
|
||||||
|
// route 則刻意不列 → 自動被主測試納入「必須 401」檢查。
|
||||||
var publicPrefixes = []string{
|
var publicPrefixes = []string{
|
||||||
"/storage/",
|
"/storage/",
|
||||||
"/ws/",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// pathParamReplacements 把 gin route 的 path param(:id / :token / *filepath)
|
// pathParamReplacements 把 gin route 的 path param(:id / :token / *filepath)
|
||||||
|
|||||||
@ -233,9 +233,18 @@ func NewRouter(deps Deps) *gin.Engine {
|
|||||||
// Pairing Token 本身就是這個 endpoint 的憑證。詳見 security.md §1.2。
|
// Pairing Token 本身就是這個 endpoint 的憑證。詳見 security.md §1.2。
|
||||||
registerPairingPublicRoutes(r, deps)
|
registerPairingPublicRoutes(r, deps)
|
||||||
|
|
||||||
// /ws/* 雛形全部 501;B7 補齊 WebSocket proxy
|
// /ws/* 雛形大多仍 501;已實作的 WS tunnel proxy(/ws/devices/:id/inference)
|
||||||
|
// 改掛在下方 wsAuthGroup(AuthMiddleware group)。
|
||||||
registerWebSocketStubs(r)
|
registerWebSocketStubs(r)
|
||||||
|
|
||||||
|
// WS tunnel proxy group:走 same-origin cookie AuthMiddleware(security 定案,
|
||||||
|
// 不放 token 到 URL)。目前只有推論結果 WS(/ws/devices/:id/inference)。
|
||||||
|
// 刻意獨立成 group 而非掛 /api:WS endpoint 對外路徑就是 /ws/*(對齊前端與
|
||||||
|
// api-spec);但認證邏輯與 /api 共用 AuthMiddleware。
|
||||||
|
wsAuthGroup := r.Group("/ws")
|
||||||
|
wsAuthGroup.Use(AuthMiddleware(deps))
|
||||||
|
registerWebSocketRoutes(wsAuthGroup, deps)
|
||||||
|
|
||||||
// OIDC public routes(不走 AuthMiddleware):
|
// OIDC public routes(不走 AuthMiddleware):
|
||||||
// - GET /api/auth/login — 起始登入流程(user 還沒登入)
|
// - GET /api/auth/login — 起始登入流程(user 還沒登入)
|
||||||
// - GET /api/auth/callback — OIDC IdP 302 回來
|
// - GET /api/auth/callback — OIDC IdP 302 回來
|
||||||
|
|||||||
@ -37,8 +37,11 @@ import (
|
|||||||
// registerCameraRoutes 註冊 /api/camera/* 與 /api/media/* 的 tunnel proxy routes。
|
// registerCameraRoutes 註冊 /api/camera/* 與 /api/media/* 的 tunnel proxy routes。
|
||||||
//
|
//
|
||||||
// 掛在 apiGroup(已套 AuthMiddleware)底下,所以每條路徑都受 OIDC session 保護。
|
// 掛在 apiGroup(已套 AuthMiddleware)底下,所以每條路徑都受 OIDC session 保護。
|
||||||
// MJPEG <img src> 帶不了 Authorization header 的認證問題(R-M3/R-C4)由前端 + api-server
|
// MJPEG <img src> 帶不了 Authorization header 的認證問題(R-M3/R-C4)靠 same-origin
|
||||||
// 的 query-token / cookie 機制處理,不在本檔範圍(本檔只負責把已認證請求中繼出去)。
|
// cookie 解決:瀏覽器對 same-origin 的 <img src> 會自動帶 visiona_session cookie,
|
||||||
|
// AuthMiddleware 驗 cookie 即放行。**刻意不支援 token-in-URL**(security 定案:
|
||||||
|
// long-lived session token 進 URL 會落 access log → 帳號接管風險)。本檔只負責把
|
||||||
|
// 已認證請求中繼出去。
|
||||||
func registerCameraRoutes(g *gin.RouterGroup, deps Deps) {
|
func registerCameraRoutes(g *gin.RouterGroup, deps Deps) {
|
||||||
// request-response 類:body / response 一次讀完即可。
|
// request-response 類:body / response 一次讀完即可。
|
||||||
proxy := newProxyHandler(deps, proxyOptions{})
|
proxy := newProxyHandler(deps, proxyOptions{})
|
||||||
@ -60,3 +63,17 @@ func registerCameraRoutes(g *gin.RouterGroup, deps Deps) {
|
|||||||
g.GET("/media/batch-images/:index", proxy) // 回單張 jpeg(非 streaming)
|
g.GET("/media/batch-images/:index", proxy) // 回單張 jpeg(非 streaming)
|
||||||
g.POST("/media/seek", proxy) // 影片 seek
|
g.POST("/media/seek", proxy) // 影片 seek
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// registerWebSocketRoutes 註冊需要 WS tunnel proxy 的 /ws/* endpoint。
|
||||||
|
//
|
||||||
|
// 目前只有 /ws/devices/:id/inference(推論結果即時推播 — camera overlay + media 結果
|
||||||
|
// 顯示的共用資料通道)。掛在 wsGroup(已套 AuthMiddleware)下,走 same-origin cookie
|
||||||
|
// 認證,不放 token 到 URL(security 定案)。
|
||||||
|
//
|
||||||
|
// 其餘 /ws/* 仍是 registerWebSocketStubs 的 501(events / flash-progress / server-logs /
|
||||||
|
// system / clusters / pairing),非本次範圍。
|
||||||
|
//
|
||||||
|
// wsGroup 必須是「path 前綴為 /ws 且套了 AuthMiddleware」的 group(見 api.go NewRouter)。
|
||||||
|
func registerWebSocketRoutes(wsGroup *gin.RouterGroup, deps Deps) {
|
||||||
|
wsGroup.GET("/devices/:id/inference", newWebSocketProxyHandler(deps))
|
||||||
|
}
|
||||||
|
|||||||
132
visionA-backend/internal/api/camera_ws_test.go
Normal file
132
visionA-backend/internal/api/camera_ws_test.go
Normal file
@ -0,0 +1,132 @@
|
|||||||
|
package api
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bufio"
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
|
"visiona-backend/internal/session"
|
||||||
|
)
|
||||||
|
|
||||||
|
// newWSFixture 建立一個掛了 registerWebSocketRoutes 的 router(含 injectStaticUserContext
|
||||||
|
// 模擬 AuthMiddleware 已放行)。deps 由 caller 提供以便測不同情境。
|
||||||
|
func newWSFixture(deps Deps) *gin.Engine {
|
||||||
|
r := gin.New()
|
||||||
|
r.Use(RequestIDMiddleware())
|
||||||
|
r.Use(injectStaticUserContext("demo-user", ""))
|
||||||
|
ws := r.Group("/ws")
|
||||||
|
registerWebSocketRoutes(ws, deps)
|
||||||
|
return r
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestWSInference_NoForwarder 驗證缺 Forwarder 時回 501(走 recorder,不需 hijack)。
|
||||||
|
func TestWSInference_NoForwarder(t *testing.T) {
|
||||||
|
r := newWSFixture(Deps{})
|
||||||
|
w := httptest.NewRecorder()
|
||||||
|
r.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/ws/devices/dev1/inference", nil))
|
||||||
|
assert.Equal(t, http.StatusNotImplemented, w.Code)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestWSInference_TunnelDisconnected 驗證無 active session 時回 502 TUNNEL_DISCONNECTED。
|
||||||
|
func TestWSInference_TunnelDisconnected(t *testing.T) {
|
||||||
|
r := newWSFixture(Deps{
|
||||||
|
SessionStore: &fakeSessionStore{}, // List 回空
|
||||||
|
Forwarder: session.NewForwarder("http://localhost:0", nil),
|
||||||
|
})
|
||||||
|
w := httptest.NewRecorder()
|
||||||
|
r.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/ws/devices/dev1/inference", nil))
|
||||||
|
assert.Equal(t, http.StatusBadGateway, w.Code)
|
||||||
|
assert.Contains(t, w.Body.String(), ErrCodeTunnelDisconnect)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestWSInference_HappyPath_BidirectionalPipe 是端到端的 WS forward 測試:
|
||||||
|
//
|
||||||
|
// browser(test client) ──► api-server handler(hijack) ──► Forwarder.OpenStream
|
||||||
|
// ──► fake remote-proxy(raw, hijack + 200 Connected) ──► fake local agent(101 + echo)
|
||||||
|
//
|
||||||
|
// 驗證:
|
||||||
|
// - handler 回 101 給 browser(升級成功)
|
||||||
|
// - 升級後 browser → agent 的 bytes 能雙向 pipe(送 ping 收 echo)
|
||||||
|
//
|
||||||
|
// 用真 httptest.NewServer(recorder 不支援 Hijack)。
|
||||||
|
func TestWSInference_HappyPath_BidirectionalPipe(t *testing.T) {
|
||||||
|
const clientMsg = "ping-from-browser"
|
||||||
|
|
||||||
|
// 1. 假 remote-proxy:hijack → 200 Connected → 模擬 local agent(讀 upgrade → 回 101 → echo)
|
||||||
|
fakeProxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
hj, ok := w.(http.Hijacker)
|
||||||
|
require.True(t, ok)
|
||||||
|
conn, _, err := hj.Hijack()
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer conn.Close()
|
||||||
|
_, _ = conn.Write([]byte("HTTP/1.1 200 Connected\r\n\r\n"))
|
||||||
|
|
||||||
|
// 模擬 local agent:讀 upgrade request → 回 101 → 之後把收到的 bytes echo 回去
|
||||||
|
br := bufio.NewReader(conn)
|
||||||
|
req, err := http.ReadRequest(br)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
assert.True(t, strings.EqualFold(req.Header.Get("Upgrade"), "websocket"))
|
||||||
|
_, _ = conn.Write([]byte(
|
||||||
|
"HTTP/1.1 101 Switching Protocols\r\n" +
|
||||||
|
"Upgrade: websocket\r\nConnection: Upgrade\r\n" +
|
||||||
|
"Sec-WebSocket-Accept: x\r\n\r\n",
|
||||||
|
))
|
||||||
|
// echo:讀 browser 送來的 bytes(透過 handler pipe)→ 原樣寫回
|
||||||
|
buf := make([]byte, len(clientMsg))
|
||||||
|
n, _ := br.Read(buf)
|
||||||
|
if n > 0 {
|
||||||
|
_, _ = conn.Write(buf[:n])
|
||||||
|
}
|
||||||
|
}))
|
||||||
|
defer fakeProxy.Close()
|
||||||
|
|
||||||
|
// 2. api-server router(有 active session token → pickActiveSessionToken 命中)
|
||||||
|
r := newWSFixture(Deps{
|
||||||
|
SessionStore: &fakeSessionStore{sessions: []*session.Summary{
|
||||||
|
{Token: "vAc_live", UserID: "demo-user"},
|
||||||
|
}},
|
||||||
|
Forwarder: session.NewForwarder(fakeProxy.URL, nil),
|
||||||
|
})
|
||||||
|
apiServer := httptest.NewServer(r)
|
||||||
|
defer apiServer.Close()
|
||||||
|
|
||||||
|
// 3. browser 端:raw TCP dial + 送 WS upgrade request
|
||||||
|
u := strings.TrimPrefix(apiServer.URL, "http://")
|
||||||
|
cli, err := net.Dial("tcp", u)
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer cli.Close()
|
||||||
|
|
||||||
|
upgrade := "GET /ws/devices/dev1/inference HTTP/1.1\r\n" +
|
||||||
|
"Host: " + u + "\r\n" +
|
||||||
|
"Upgrade: websocket\r\n" +
|
||||||
|
"Connection: Upgrade\r\n" +
|
||||||
|
"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" +
|
||||||
|
"Sec-WebSocket-Version: 13\r\n\r\n"
|
||||||
|
_, err = cli.Write([]byte(upgrade))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
// 讀 101 response
|
||||||
|
cbr := bufio.NewReader(cli)
|
||||||
|
resp, err := http.ReadResponse(cbr, nil)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode,
|
||||||
|
"handler 應回 101 給 browser")
|
||||||
|
|
||||||
|
// 升級後:送一段 bytes,期望 echo 回來(驗證雙向 pipe)
|
||||||
|
_, err = cli.Write([]byte(clientMsg))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
echo := make([]byte, len(clientMsg))
|
||||||
|
_, err = cbr.Read(echo)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, clientMsg, string(echo), "升級後 bytes 應雙向 pipe(收到 echo)")
|
||||||
|
}
|
||||||
@ -17,6 +17,7 @@ import (
|
|||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/gin-gonic/gin"
|
"github.com/gin-gonic/gin"
|
||||||
@ -118,6 +119,139 @@ func newProxyHandler(deps Deps, opts proxyOptions) gin.HandlerFunc {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// newWebSocketProxyHandler 產生一個 gin.HandlerFunc,把 browser 的 WebSocket 連線
|
||||||
|
// 經 tunnel 轉發到 local agent(推論結果 WS `inference:<deviceId>` 走這條回顯)。
|
||||||
|
//
|
||||||
|
// 認證:本 handler 掛在 apiGroup(AuthMiddleware)下,走 same-origin cookie 認證。
|
||||||
|
// 瀏覽器的 `new WebSocket(...)` 對 same-origin 會自動帶 visiona_session cookie,
|
||||||
|
// AuthMiddleware 驗 cookie → 放行。**刻意不支援 token-in-URL**(security 定案:
|
||||||
|
// long-lived session token 進 URL 會落 access log → 帳號接管風險,Critical)。
|
||||||
|
//
|
||||||
|
// 授權(S2 / IDOR):與所有 /api/devices/:id/* proxy 路徑同一套 posture —
|
||||||
|
// pickActiveSessionToken 只挑「當前 user 自己的」active session token,request 只會
|
||||||
|
// 被送到該 user 自己的 local agent。deviceId 只是透傳給 local agent 的 path 參數,
|
||||||
|
// 攻擊者帶別人的 deviceId 也只會打到自己的 tunnel(打不到別人的 agent)。多 user /
|
||||||
|
// 多 device 的 strict deviceId ↔ session 綁定屬 Phase 1(見 pickActiveSessionToken 註解)。
|
||||||
|
//
|
||||||
|
// 流程(照 POC edge-ai-platform relay/server.go:233-296 proxyWebSocket 移植後半段):
|
||||||
|
// 1. 找 user 的 active session token
|
||||||
|
// 2. Forwarder.ForwardWebSocket:OpenStream → 寫 upgrade req → 讀 101
|
||||||
|
// 3. Hijack browser 連線 → 回寫 101 → 與 tunnel conn 雙向 io.Copy
|
||||||
|
//
|
||||||
|
// WS payload 契約:透明轉發 raw bytes,api-server 不解析 / 不包 envelope
|
||||||
|
// (local agent 直接推 raw driver.InferenceResult JSON,前端自行解析)。
|
||||||
|
func newWebSocketProxyHandler(deps Deps) gin.HandlerFunc {
|
||||||
|
return func(c *gin.Context) {
|
||||||
|
if deps.Forwarder == nil || deps.SessionStore == nil {
|
||||||
|
WriteNotImplemented(c, "forwarder/session store not configured")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
uc, ok := UserContextFrom(c)
|
||||||
|
if !ok || uc.UserID == "" {
|
||||||
|
WriteError(c, http.StatusInternalServerError, ErrCodeInternalError,
|
||||||
|
"missing user context (auth middleware misconfigured?)", nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
token, err := pickActiveSessionToken(c.Request.Context(), deps.SessionStore, uc.UserID, deps.Logger)
|
||||||
|
if err != nil {
|
||||||
|
writeTunnelError(c, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// 建出「打給 local agent」的 upgrade request:沿用原 path + query + WS header。
|
||||||
|
outPath := c.Request.URL.Path
|
||||||
|
if c.Request.URL.RawQuery != "" {
|
||||||
|
outPath += "?" + c.Request.URL.RawQuery
|
||||||
|
}
|
||||||
|
// WS 是長連線,不套 defaultProxyRequestTimeout(S1:streaming 不設 timeout)。
|
||||||
|
outReq, err := http.NewRequestWithContext(c.Request.Context(), c.Request.Method, outPath, nil)
|
||||||
|
if err != nil {
|
||||||
|
WriteError(c, http.StatusInternalServerError, ErrCodeInternalError,
|
||||||
|
"ws proxy: build upstream request: "+err.Error(), nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// WS upgrade 依賴 Upgrade / Connection / Sec-WebSocket-* header,必須完整帶上。
|
||||||
|
// 不走 copyProxyRequestHeaders(那會剝 Connection/Upgrade 等 hop-by-hop)。
|
||||||
|
copyWebSocketUpgradeHeaders(c.Request.Header, outReq.Header)
|
||||||
|
|
||||||
|
wsConn, err := deps.Forwarder.ForwardWebSocket(c.Request.Context(), token, outReq)
|
||||||
|
if err != nil {
|
||||||
|
// local agent 拒絕 upgrade(非 101)→ 轉發原狀態碼給 browser。
|
||||||
|
if resp, ok := session.AsWSUpgradeError(err); ok {
|
||||||
|
defer resp.Body.Close()
|
||||||
|
writeProxyResponse(c, resp, false)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeTunnelError(c, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer wsConn.Conn.Close()
|
||||||
|
|
||||||
|
// Hijack browser 連線 → 回寫 101 → 雙向 pipe(照 POC proxyWebSocket 後半段)。
|
||||||
|
hijacker, ok := c.Writer.(http.Hijacker)
|
||||||
|
if !ok {
|
||||||
|
WriteError(c, http.StatusInternalServerError, ErrCodeInternalError,
|
||||||
|
"ws proxy: hijacking not supported", nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
clientConn, clientBuf, err := hijacker.Hijack()
|
||||||
|
if err != nil {
|
||||||
|
logOrDefault(deps.Logger).Warn("ws proxy: hijack failed",
|
||||||
|
"error", err, "request_id", RequestIDFrom(c))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer clientConn.Close()
|
||||||
|
|
||||||
|
// 把 local agent 的 101 response 原樣寫回 browser,完成 browser 端 WS 握手。
|
||||||
|
if err := wsConn.Resp.Write(clientBuf); err != nil {
|
||||||
|
logOrDefault(deps.Logger).Warn("ws proxy: write 101 to browser failed",
|
||||||
|
"error", err, "request_id", RequestIDFrom(c))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := clientBuf.Flush(); err != nil {
|
||||||
|
logOrDefault(deps.Logger).Warn("ws proxy: flush 101 to browser failed",
|
||||||
|
"error", err, "request_id", RequestIDFrom(c))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// 雙向 byte pipe:browser ↔ tunnel。任一方向結束就關掉另一邊。
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
wg.Add(2)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
_, _ = io.Copy(wsConn.Conn, clientConn)
|
||||||
|
_ = wsConn.Conn.Close()
|
||||||
|
}()
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
_, _ = io.Copy(clientConn, wsConn.Conn)
|
||||||
|
_ = clientConn.Close()
|
||||||
|
}()
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// copyWebSocketUpgradeHeaders 複製 WS upgrade 所需的 header 到 upstream request。
|
||||||
|
//
|
||||||
|
// 與 copyProxyRequestHeaders 的差別:**保留** Connection / Upgrade(WS 握手必需,
|
||||||
|
// 雖是 hop-by-hop),並帶上 Sec-WebSocket-*。同樣剝掉 Authorization / Origin
|
||||||
|
// (Origin 剝除理由同 copyProxyRequestHeaders:避免 local agent CORS 對 tunnel 中繼
|
||||||
|
// 請求誤判)。
|
||||||
|
func copyWebSocketUpgradeHeaders(src, dst http.Header) {
|
||||||
|
for name, values := range src {
|
||||||
|
if strings.EqualFold(name, "Authorization") || strings.EqualFold(name, "Origin") {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// 其餘(含 Connection / Upgrade / Sec-WebSocket-Key / -Version / -Protocol /
|
||||||
|
// -Extensions)全部保留,WS 握手需要。
|
||||||
|
for _, v := range values {
|
||||||
|
dst.Add(name, v)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// pickActiveSessionToken 找出當前使用者在雲端的 active session token。
|
// pickActiveSessionToken 找出當前使用者在雲端的 active session token。
|
||||||
//
|
//
|
||||||
// 雛形邏輯(單一 user + 單一 agent):走 Store.List,過濾 userID 對得上的第一筆。
|
// 雛形邏輯(單一 user + 單一 agent):走 Store.List,過濾 userID 對得上的第一筆。
|
||||||
|
|||||||
@ -4,17 +4,17 @@ import (
|
|||||||
"github.com/gin-gonic/gin"
|
"github.com/gin-gonic/gin"
|
||||||
)
|
)
|
||||||
|
|
||||||
// registerStubRoutes 註冊 B5 尚未實作 / Phase 1 才處理的 endpoint,一律回 501 NOT_IMPLEMENTED。
|
// registerStubRoutes 註冊尚未實作 / Phase 1 才處理的 HTTP endpoint,一律回 501 NOT_IMPLEMENTED。
|
||||||
//
|
//
|
||||||
// **為什麼只留這些**:Auth / Pairing 補齊 / Devices / Models / GET /clusters /
|
// **為什麼只留這些**:Auth / Pairing 補齊 / Devices / Models / GET /clusters /
|
||||||
// system/deps / /storage 都在 B5 補實作(見 auth.go / devices.go / models.go /
|
// system/deps / /storage 都在 B5 補實作(見 auth.go / devices.go / models.go /
|
||||||
// clusters.go / storage.go 各檔)。這裡只剩:
|
// clusters.go / storage.go 各檔);Camera / Media 已在 camera.go(registerCameraRoutes)
|
||||||
|
// 換成真正的 tunnel proxy 宣告。這裡只剩:
|
||||||
// - Cloud 裝置記錄(非 tunnel 的 CRUD,Phase 1)
|
// - Cloud 裝置記錄(非 tunnel 的 CRUD,Phase 1)
|
||||||
// - Clusters 寫入類(Phase 1)
|
// - Clusters 寫入類(Phase 1)
|
||||||
// - Converter(Phase 1)
|
// - Converter(Phase 1)
|
||||||
//
|
//
|
||||||
// Camera / Media 已在 camera.go(registerCameraRoutes)換成真正的 tunnel proxy 宣告。
|
// WebSocket 的 501 stub 在 registerWebSocketStubs(下方),不在此函式。
|
||||||
// - WebSocket endpoints(B7 TODO — 需要 Hijack + WS relay)
|
|
||||||
//
|
//
|
||||||
// 讓前端對錯誤路徑能拿到 501 而非 404,減少除錯成本。
|
// 讓前端對錯誤路徑能拿到 501 而非 404,減少除錯成本。
|
||||||
func registerStubRoutes(g *gin.RouterGroup, _ Deps) {
|
func registerStubRoutes(g *gin.RouterGroup, _ Deps) {
|
||||||
@ -52,12 +52,14 @@ func registerStubRoutes(g *gin.RouterGroup, _ Deps) {
|
|||||||
g.GET("/converter/jobs/:id/download", stub("converter.download — pending Phase 1"))
|
g.GET("/converter/jobs/:id/download", stub("converter.download — pending Phase 1"))
|
||||||
}
|
}
|
||||||
|
|
||||||
// registerWebSocketStubs 註冊 /ws/* 的 stub。WebSocket proxy 在 B5 雛形不實作,
|
// registerWebSocketStubs 註冊「尚未實作」的 /ws/* endpoint,一律回 501,讓前端能收到
|
||||||
// 留 501 讓前端能收到明確錯誤,由 B7 補齊。
|
// 明確錯誤,由 B7 補齊。
|
||||||
//
|
//
|
||||||
// 為什麼不做 WS proxy:實作 WS relay 需要在 api-server 端做 Hijack、雙向 io.Copy,
|
// WS tunnel proxy 的基礎設施(Forwarder.ForwardWebSocket、api-server 端 Hijack + 雙向
|
||||||
// 而且 Forwarder.ForwardWebSocket 尚未實作(見 forwarder.go §ForwardWebSocket)。
|
// io.Copy)已實作(見 forwarder.go §ForwardWebSocket、proxy.go newWebSocketProxyHandler),
|
||||||
// 加這條路徑會顯著擴張 B5 範圍;按 prompt 指示先留 TODO。
|
// 且 /ws/devices/:id/inference 已由 registerWebSocketRoutes 換成真正的 WS tunnel proxy。
|
||||||
|
// 這裡只剩尚未接上 proxy 的其餘 /ws/* 端點(events / flash-progress / server-logs /
|
||||||
|
// system / clusters / pairing)。
|
||||||
//
|
//
|
||||||
// 注意:ws endpoint 在 /ws 而非 /api/ws,所以由 NewRouter 直接註冊而非 apiGroup。
|
// 注意:ws endpoint 在 /ws 而非 /api/ws,所以由 NewRouter 直接註冊而非 apiGroup。
|
||||||
func registerWebSocketStubs(r *gin.Engine) {
|
func registerWebSocketStubs(r *gin.Engine) {
|
||||||
@ -69,7 +71,9 @@ func registerWebSocketStubs(r *gin.Engine) {
|
|||||||
// 用 GET(WebSocket upgrade 的初始 HTTP request)
|
// 用 GET(WebSocket upgrade 的初始 HTTP request)
|
||||||
r.GET("/ws/devices/events", stub("ws.devices.events — pending B7"))
|
r.GET("/ws/devices/events", stub("ws.devices.events — pending B7"))
|
||||||
r.GET("/ws/devices/:id/flash-progress", stub("ws.flash-progress — pending B7"))
|
r.GET("/ws/devices/:id/flash-progress", stub("ws.flash-progress — pending B7"))
|
||||||
r.GET("/ws/devices/:id/inference", stub("ws.inference — pending B7"))
|
// /ws/devices/:id/inference 已由 registerWebSocketRoutes 換成真正的 WS tunnel proxy
|
||||||
|
// (掛在 AuthMiddleware group 內,走 same-origin cookie 認證)。不在此註冊 stub,
|
||||||
|
// 避免 gin radix tree 同路徑重複註冊 panic。
|
||||||
r.GET("/ws/server-logs", stub("ws.server-logs — pending B7"))
|
r.GET("/ws/server-logs", stub("ws.server-logs — pending B7"))
|
||||||
r.GET("/ws/system", stub("ws.system — pending B7"))
|
r.GET("/ws/system", stub("ws.system — pending B7"))
|
||||||
r.GET("/ws/clusters/:id/inference", stub("ws.clusters.inference — pending B7"))
|
r.GET("/ws/clusters/:id/inference", stub("ws.clusters.inference — pending B7"))
|
||||||
|
|||||||
@ -261,17 +261,115 @@ func (f *Forwarder) ForwardHTTP(ctx context.Context, sessionToken string, req *h
|
|||||||
return resp, nil
|
return resp, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// ForwardWebSocket 預留 — B5 接前端 WS 時實作。
|
// WebSocketConn 是 ForwardWebSocket 成功後回傳的結果。
|
||||||
//
|
//
|
||||||
// 預期實作(草稿):
|
// Conn 是「已完成 WS upgrade(101 已讀掉)」的 tunnel 連線,語意上等同 net.Conn:
|
||||||
// - OpenStream 拿到 raw conn
|
// 後續讀寫的都是 WebSocket frame bytes(api-server 不解析 frame,只做透明 byte pipe)。
|
||||||
// - 把 WS upgrade request 透過 conn 寫過去
|
// Resp 是 local agent 回的 101 Switching Protocols response(含 Sec-WebSocket-Accept
|
||||||
// - 等 101 response 回來
|
// 等 header);handler 需把它原樣寫回 browser 端 hijacked 連線,才能完成 browser↔agent
|
||||||
// - Hijack browser 端連線,與 conn 雙向 pipe
|
// 的 WS 握手。
|
||||||
//
|
//
|
||||||
// 雛形先回 ErrNotImplemented,避免被誤用。
|
// caller 必須負責 Conn.Close()。
|
||||||
func (f *Forwarder) ForwardWebSocket(ctx context.Context, sessionToken string, req *http.Request) (net.Conn, error) {
|
type WebSocketConn struct {
|
||||||
return nil, errors.New("session: ForwardWebSocket not implemented yet (TODO B5)")
|
Conn net.Conn
|
||||||
|
Resp *http.Response
|
||||||
|
}
|
||||||
|
|
||||||
|
// ForwardWebSocket 把一個 WebSocket upgrade 請求經 tunnel 轉發到 local agent,
|
||||||
|
// 回傳「已升級(101 已讀)」的 tunnel 連線 + local agent 的 101 response。
|
||||||
|
//
|
||||||
|
// 架構說明(visionA 雙 binary vs POC 單 relay):
|
||||||
|
//
|
||||||
|
// POC(edge-ai-platform relay/server.go:233-296 proxyWebSocket)是單 binary,relay
|
||||||
|
// 直接持有 yamux session,在同一個 handler 內完成「寫 upgrade → 讀 101 → hijack
|
||||||
|
// browser → 雙向 copy」。
|
||||||
|
//
|
||||||
|
// visionA 拆成 api-server(面向瀏覽器 + auth)+ remote-proxy(relay)。api-server 透過
|
||||||
|
// OpenStream 對 remote-proxy 開一條 raw TCP,remote-proxy 端 (internal_forward_raw.go)
|
||||||
|
// 對這條連線做**透明 byte pipe、不解析 HTTP**,所以 WS upgrade bytes 原樣穿過
|
||||||
|
// remote-proxy 到 local agent(client.go:370 handleWebSocket 已實作本地端 pipe)。
|
||||||
|
//
|
||||||
|
// 因此 visionA 的 WS forward 分工:
|
||||||
|
// - 本函式(forwarder 層):OpenStream → 寫 upgrade req → 讀 101 → 回 conn+resp
|
||||||
|
// - handler 層(camera.go newWebSocketProxyHandler):hijack browser → 回寫 101 →
|
||||||
|
// 雙向 io.Copy(照 POC proxyWebSocket 的後半段移植)
|
||||||
|
//
|
||||||
|
// 失敗回傳的 error:
|
||||||
|
// - ErrSessionNotFound:無 active session(OpenStream 已映射)
|
||||||
|
// - 非 101:local agent 拒絕 upgrade(回傳含該 response 的 error 供 handler 轉發原狀態碼)
|
||||||
|
// - 其他 wrapped error:dial / 寫 / 讀失敗
|
||||||
|
func (f *Forwarder) ForwardWebSocket(ctx context.Context, sessionToken string, req *http.Request) (*WebSocketConn, error) {
|
||||||
|
if req == nil {
|
||||||
|
return nil, errors.New("session: ForwardWebSocket requires non-nil req")
|
||||||
|
}
|
||||||
|
|
||||||
|
conn, err := f.OpenStream(ctx, sessionToken)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// 改寫 req 為「打給 local agent」格式(同 ForwardHTTP)。
|
||||||
|
outReq := req.Clone(ctx)
|
||||||
|
if outReq.URL == nil {
|
||||||
|
outReq.URL = &url.URL{}
|
||||||
|
}
|
||||||
|
outReq.URL.Scheme = "http"
|
||||||
|
outReq.URL.Host = "127.0.0.1"
|
||||||
|
outReq.RequestURI = ""
|
||||||
|
if outReq.Host == "" {
|
||||||
|
outReq.Host = "127.0.0.1"
|
||||||
|
}
|
||||||
|
|
||||||
|
// 寫 upgrade request 到 tunnel(保留 Upgrade / Connection / Sec-WebSocket-* header —
|
||||||
|
// 這些是 hop-by-hop,但 WS upgrade 依賴它們,所以 req.Write 會原樣送出)。
|
||||||
|
if err := outReq.Write(conn); err != nil {
|
||||||
|
_ = conn.Close()
|
||||||
|
return nil, fmt.Errorf("session: write ws upgrade request: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 讀 upgrade response。用 bufio.Reader 讀,讀完 101 header 後若有多讀的 byte
|
||||||
|
// (WS frame 可能緊跟在 101 後面)要接回 conn,避免丟失。
|
||||||
|
br := bufio.NewReader(conn)
|
||||||
|
resp, err := http.ReadResponse(br, outReq)
|
||||||
|
if err != nil {
|
||||||
|
_ = conn.Close()
|
||||||
|
return nil, fmt.Errorf("session: read ws upgrade response: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if resp.StatusCode != http.StatusSwitchingProtocols {
|
||||||
|
// local agent 拒絕 upgrade(非 101)。把 conn 包進 response.Body 讓 handler
|
||||||
|
// 能讀出 body 轉發原狀態碼給 browser,再由 handler close。
|
||||||
|
resp.Body = &bodyWithConn{ReadCloser: resp.Body, conn: conn}
|
||||||
|
return nil, &wsUpgradeError{Resp: resp}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 101 成功。把 bufio 預讀但還沒被消費的 byte 接回 conn 開頭,回傳給 handler。
|
||||||
|
outConn := conn
|
||||||
|
if buffered := br.Buffered(); buffered > 0 {
|
||||||
|
peek, _ := br.Peek(buffered)
|
||||||
|
outConn = newPrefixConn(conn, append([]byte(nil), peek...))
|
||||||
|
}
|
||||||
|
return &WebSocketConn{Conn: outConn, Resp: resp}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// wsUpgradeError 代表「tunnel 通了,但 local agent 回的不是 101」。
|
||||||
|
// 帶著原 response 讓 handler 能把原狀態碼 / body 轉發回 browser。
|
||||||
|
type wsUpgradeError struct {
|
||||||
|
Resp *http.Response
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *wsUpgradeError) Error() string {
|
||||||
|
return fmt.Sprintf("session: ws upgrade rejected by local agent: %s", e.Resp.Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
// AsWSUpgradeError 若 err 是 ws upgrade 被拒(非 101),回傳該 response 與 true。
|
||||||
|
// handler 用它把 local agent 的原狀態碼轉發給 browser。
|
||||||
|
func AsWSUpgradeError(err error) (*http.Response, bool) {
|
||||||
|
var e *wsUpgradeError
|
||||||
|
if errors.As(err, &e) {
|
||||||
|
return e.Resp, true
|
||||||
|
}
|
||||||
|
return nil, false
|
||||||
}
|
}
|
||||||
|
|
||||||
// ----------------------------------------------------------------------
|
// ----------------------------------------------------------------------
|
||||||
|
|||||||
@ -1,13 +1,18 @@
|
|||||||
package session
|
package session
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bufio"
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"io"
|
||||||
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestForwarder_OpenStream_NoProxyHost 驗證 baseURL 為空時直接拒絕。
|
// TestForwarder_OpenStream_NoProxyHost 驗證 baseURL 為空時直接拒絕。
|
||||||
@ -24,14 +29,138 @@ func TestForwarder_OpenStream_EmptyToken(t *testing.T) {
|
|||||||
assert.Error(t, err)
|
assert.Error(t, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestForwarder_ForwardWebSocket_NotImplemented 驗證 ForwardWebSocket 仍是 stub。
|
// TestForwarder_ForwardWebSocket_NilReq 驗證 nil req 直接拒絕。
|
||||||
func TestForwarder_ForwardWebSocket_NotImplemented(t *testing.T) {
|
func TestForwarder_ForwardWebSocket_NilReq(t *testing.T) {
|
||||||
f := NewForwarder("http://localhost:9999", nil)
|
f := NewForwarder("http://localhost:9999", nil)
|
||||||
req, _ := http.NewRequest(http.MethodGet, "/ws", nil)
|
_, err := f.ForwardWebSocket(context.Background(), "vAc_x", nil)
|
||||||
|
assert.Error(t, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestForwarder_ForwardWebSocket_NoProxyHost 驗證 baseURL 為空時 OpenStream 就失敗。
|
||||||
|
func TestForwarder_ForwardWebSocket_NoProxyHost(t *testing.T) {
|
||||||
|
f := NewForwarder("", nil)
|
||||||
|
req, _ := http.NewRequest(http.MethodGet, "/ws/devices/x/inference", nil)
|
||||||
_, err := f.ForwardWebSocket(context.Background(), "vAc_x", req)
|
_, err := f.ForwardWebSocket(context.Background(), "vAc_x", req)
|
||||||
assert.Error(t, err)
|
assert.Error(t, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// fakeRawProxy 起一個假的 remote-proxy /internal/forward/raw:
|
||||||
|
// hijack 後回 "200 Connected",接著 handoff 給 onStream 模擬 local agent 端行為。
|
||||||
|
//
|
||||||
|
// 回傳 baseURL 供 NewForwarder。
|
||||||
|
func fakeRawProxy(t *testing.T, onStream func(conn net.Conn)) string {
|
||||||
|
t.Helper()
|
||||||
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
hj, ok := w.(http.Hijacker)
|
||||||
|
require.True(t, ok, "test server must support hijack")
|
||||||
|
conn, _, err := hj.Hijack()
|
||||||
|
require.NoError(t, err)
|
||||||
|
// 回 raw forward 握手
|
||||||
|
if _, err := conn.Write([]byte("HTTP/1.1 200 Connected\r\n\r\n")); err != nil {
|
||||||
|
_ = conn.Close()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
onStream(conn) // 模擬 tunnel 另一端(local agent)
|
||||||
|
}))
|
||||||
|
t.Cleanup(ts.Close)
|
||||||
|
return ts.URL
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestForwarder_ForwardWebSocket_101_Success 驗證 happy path:
|
||||||
|
// forwarder 寫 upgrade request → 假 local agent 回 101 → forwarder 回 WebSocketConn。
|
||||||
|
func TestForwarder_ForwardWebSocket_101_Success(t *testing.T) {
|
||||||
|
const framePayload = "hello-ws-frame"
|
||||||
|
|
||||||
|
baseURL := fakeRawProxy(t, func(conn net.Conn) {
|
||||||
|
defer conn.Close()
|
||||||
|
// 讀 upgrade request
|
||||||
|
req, err := http.ReadRequest(bufio.NewReader(conn))
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// 驗證 upgrade header 有被帶過來
|
||||||
|
assert.True(t, strings.EqualFold(req.Header.Get("Upgrade"), "websocket"),
|
||||||
|
"local agent 應收到 Upgrade: websocket")
|
||||||
|
// 回 101 + 一段 frame bytes(模擬升級後 local agent 主動推的資料)
|
||||||
|
_, _ = conn.Write([]byte(
|
||||||
|
"HTTP/1.1 101 Switching Protocols\r\n" +
|
||||||
|
"Upgrade: websocket\r\n" +
|
||||||
|
"Connection: Upgrade\r\n" +
|
||||||
|
"Sec-WebSocket-Accept: fake-accept\r\n\r\n" +
|
||||||
|
framePayload,
|
||||||
|
))
|
||||||
|
})
|
||||||
|
|
||||||
|
f := NewForwarder(baseURL, nil)
|
||||||
|
req, _ := http.NewRequest(http.MethodGet, "/ws/devices/dev1/inference", nil)
|
||||||
|
req.Header.Set("Upgrade", "websocket")
|
||||||
|
req.Header.Set("Connection", "Upgrade")
|
||||||
|
req.Header.Set("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
|
||||||
|
req.Header.Set("Sec-WebSocket-Version", "13")
|
||||||
|
|
||||||
|
wsConn, err := f.ForwardWebSocket(context.Background(), "vAc_x", req)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NotNil(t, wsConn)
|
||||||
|
defer wsConn.Conn.Close()
|
||||||
|
|
||||||
|
assert.Equal(t, http.StatusSwitchingProtocols, wsConn.Resp.StatusCode)
|
||||||
|
assert.Equal(t, "fake-accept", wsConn.Resp.Header.Get("Sec-WebSocket-Accept"))
|
||||||
|
|
||||||
|
// 驗證 101 後緊跟的 frame bytes 沒有被 bufio 預讀吞掉(prefixConn 接回)。
|
||||||
|
buf := make([]byte, len(framePayload))
|
||||||
|
n, err := io.ReadFull(wsConn.Conn, buf)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, framePayload, string(buf[:n]),
|
||||||
|
"101 後預讀的 frame bytes 應完整保留")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestForwarder_ForwardWebSocket_Non101_Rejected 驗證 local agent 回非 101 時,
|
||||||
|
// 回傳 wsUpgradeError 且能透過 AsWSUpgradeError 取出原 response。
|
||||||
|
func TestForwarder_ForwardWebSocket_Non101_Rejected(t *testing.T) {
|
||||||
|
baseURL := fakeRawProxy(t, func(conn net.Conn) {
|
||||||
|
defer conn.Close()
|
||||||
|
_, _ = http.ReadRequest(bufio.NewReader(conn))
|
||||||
|
// local agent 拒絕 upgrade:回 403
|
||||||
|
_, _ = conn.Write([]byte(
|
||||||
|
"HTTP/1.1 403 Forbidden\r\n" +
|
||||||
|
"Content-Type: text/plain\r\n" +
|
||||||
|
"Content-Length: 7\r\n\r\n" +
|
||||||
|
"denied!",
|
||||||
|
))
|
||||||
|
})
|
||||||
|
|
||||||
|
f := NewForwarder(baseURL, nil)
|
||||||
|
req, _ := http.NewRequest(http.MethodGet, "/ws/devices/dev1/inference", nil)
|
||||||
|
req.Header.Set("Upgrade", "websocket")
|
||||||
|
|
||||||
|
_, err := f.ForwardWebSocket(context.Background(), "vAc_x", req)
|
||||||
|
require.Error(t, err)
|
||||||
|
|
||||||
|
resp, ok := AsWSUpgradeError(err)
|
||||||
|
require.True(t, ok, "應為 wsUpgradeError")
|
||||||
|
assert.Equal(t, http.StatusForbidden, resp.StatusCode)
|
||||||
|
body, _ := io.ReadAll(resp.Body)
|
||||||
|
_ = resp.Body.Close()
|
||||||
|
assert.Equal(t, "denied!", string(body))
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestForwarder_ForwardWebSocket_NoSession 驗證無 session(remote-proxy 回 502)→
|
||||||
|
// 映射成 ErrSessionNotFound(沿用 OpenStream 的錯誤映射)。
|
||||||
|
func TestForwarder_ForwardWebSocket_NoSession(t *testing.T) {
|
||||||
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusBadGateway)
|
||||||
|
_, _ = w.Write([]byte(`{"error":{"code":"TUNNEL_DISCONNECTED"}}`))
|
||||||
|
}))
|
||||||
|
defer ts.Close()
|
||||||
|
|
||||||
|
f := NewForwarder(ts.URL, nil)
|
||||||
|
req, _ := http.NewRequest(http.MethodGet, "/ws/devices/dev1/inference", nil)
|
||||||
|
req.Header.Set("Upgrade", "websocket")
|
||||||
|
_, err := f.ForwardWebSocket(context.Background(), "vAc_dead", req)
|
||||||
|
assert.ErrorIs(t, err, ErrSessionNotFound)
|
||||||
|
}
|
||||||
|
|
||||||
// TestForwarder_OpenStream_502_TreatedAsNotFound 驗證當 remote-proxy 回 502
|
// TestForwarder_OpenStream_502_TreatedAsNotFound 驗證當 remote-proxy 回 502
|
||||||
// (session 不存在時的雛形行為)→ 包裝成 ErrSessionNotFound。
|
// (session 不存在時的雛形行為)→ 包裝成 ErrSessionNotFound。
|
||||||
//
|
//
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user