From dc6ca211ae526c9e40087cbba0747955ff785178 Mon Sep 17 00:00:00 2001 From: jim800121chen Date: Thu, 9 Jul 2026 04:56:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(ws):=20tunnel=20WS=20forward=20=E2=80=94?= =?UTF-8?q?=20=E6=8E=A8=E8=AB=96=E7=B5=90=E6=9E=9C=E7=B6=93=20tunnel=20?= =?UTF-8?q?=E6=8E=A8=E5=9B=9E=E9=9B=B2=E7=AB=AF=EF=BC=88=E5=BE=8C=E7=AB=AF?= =?UTF-8?q?=E5=A1=8A2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 推論工作區後端塊2：實作 WS forward，讓 local agent 的推論結果 WS （inference:）經 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) --- .../all_endpoints_require_auth_test.go | 34 +++-- visionA-backend/internal/api/api.go | 11 +- visionA-backend/internal/api/camera.go | 29 +++- .../internal/api/camera_ws_test.go | 132 +++++++++++++++++ visionA-backend/internal/api/proxy.go | 134 +++++++++++++++++ visionA-backend/internal/api/stubs.go | 24 ++-- visionA-backend/internal/session/forwarder.go | 116 +++++++++++++-- .../internal/session/forwarder_test.go | 135 +++++++++++++++++- 8 files changed, 578 insertions(+), 37 deletions(-) create mode 100644 visionA-backend/internal/api/camera_ws_test.go diff --git a/visionA-backend/cmd/api-server/all_endpoints_require_auth_test.go b/visionA-backend/cmd/api-server/all_endpoints_require_auth_test.go index a7e813d..976f37b 100644 --- a/visionA-backend/cmd/api-server/all_endpoints_require_auth_test.go +++ b/visionA-backend/cmd/api-server/all_endpoints_require_auth_test.go @@ -43,23 +43,41 @@ import ( // - /api/auth/callback → OIDC IdP 302 回來，user 還沒登入 // - /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 保護。 var publicPaths = map[string]bool{ - "GET /healthz": true, - "GET /api/auth/login": true, - "GET /api/auth/callback": true, - "POST /api/pairing/exchange": true, + "GET /healthz": true, + "GET /api/auth/login": true, + "GET /api/auth/callback": 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」的清單。 // // - /storage/* — 用 HMAC presigned URL 驗簽（api-spec.md §10），不是 cookie -// - /ws/* — 雛形 stub 一律 501，註冊在 r 而非 apiGroup（stubs.go:70-85）。 -// 目前無認證 → 501；**未來補實作 WebSocket proxy 時必須套 auth**， -// 屆時應從這份清單移除。TODO(B7): 移到 protected。 +// +// 為什麼這裡「不」放 /ws/：/ws/* 底下同時有 501 stub（無 auth）與已升級為 authed 的 +// route（/ws/devices/:id/inference，掛 wsAuthGroup）。用 prefix 一次放行會連 authed +// route 一起豁免、讓回歸測試對它失效。故 /ws/ 的 stub 逐條列在 publicPaths、authed +// route 則刻意不列 → 自動被主測試納入「必須 401」檢查。 var publicPrefixes = []string{ "/storage/", - "/ws/", } // pathParamReplacements 把 gin route 的 path param（:id / :token / *filepath） diff --git a/visionA-backend/internal/api/api.go b/visionA-backend/internal/api/api.go index e4fadf8..84f0e76 100644 --- a/visionA-backend/internal/api/api.go +++ b/visionA-backend/internal/api/api.go @@ -233,9 +233,18 @@ func NewRouter(deps Deps) *gin.Engine { // Pairing Token 本身就是這個 endpoint 的憑證。詳見 security.md §1.2。 registerPairingPublicRoutes(r, deps) - // /ws/* 雛形全部 501；B7 補齊 WebSocket proxy + // /ws/* 雛形大多仍 501；已實作的 WS tunnel proxy（/ws/devices/:id/inference） + // 改掛在下方 wsAuthGroup（AuthMiddleware group）。 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）： // - GET /api/auth/login — 起始登入流程（user 還沒登入） // - GET /api/auth/callback — OIDC IdP 302 回來 diff --git a/visionA-backend/internal/api/camera.go b/visionA-backend/internal/api/camera.go index 831a8b0..993c8a9 100644 --- a/visionA-backend/internal/api/camera.go +++ b/visionA-backend/internal/api/camera.go @@ -37,8 +37,11 @@ import ( // registerCameraRoutes 註冊 /api/camera/* 與 /api/media/* 的 tunnel proxy routes。 // // 掛在 apiGroup（已套 AuthMiddleware）底下，所以每條路徑都受 OIDC session 保護。 -// MJPEG 帶不了 Authorization header 的認證問題（R-M3/R-C4）由前端 + api-server -// 的 query-token / cookie 機制處理，不在本檔範圍（本檔只負責把已認證請求中繼出去）。 +// MJPEG 帶不了 Authorization header 的認證問題（R-M3/R-C4）靠 same-origin +// cookie 解決：瀏覽器對 same-origin 的 會自動帶 visiona_session cookie， +// AuthMiddleware 驗 cookie 即放行。**刻意不支援 token-in-URL**（security 定案： +// long-lived session token 進 URL 會落 access log → 帳號接管風險）。本檔只負責把 +// 已認證請求中繼出去。 func registerCameraRoutes(g *gin.RouterGroup, deps Deps) { // request-response 類：body / response 一次讀完即可。 proxy := newProxyHandler(deps, proxyOptions{}) @@ -46,10 +49,10 @@ func registerCameraRoutes(g *gin.RouterGroup, deps Deps) { streamProxy := newProxyHandler(deps, proxyOptions{streaming: true}) // --- Camera（即時攝影機） --- - g.GET("/camera/list", proxy) // 列出可用攝影機 - g.POST("/camera/start", proxy) // 開 camera + 起推論，回 streamUrl - g.POST("/camera/stop", proxy) // 停止 pipeline - g.GET("/camera/stream", streamProxy) // MJPEG multipart/x-mixed-replace 長連線 + g.GET("/camera/list", proxy) // 列出可用攝影機 + g.POST("/camera/start", proxy) // 開 camera + 起推論，回 streamUrl + g.POST("/camera/stop", proxy) // 停止 pipeline + g.GET("/camera/stream", streamProxy) // MJPEG multipart/x-mixed-replace 長連線 // --- Media（圖片 / 影片 / 批次推論） --- // 注意：media 的「結果畫面」也走上面的 /camera/stream（pipeline.go 共用同一個 @@ -60,3 +63,17 @@ func registerCameraRoutes(g *gin.RouterGroup, deps Deps) { g.GET("/media/batch-images/:index", proxy) // 回單張 jpeg（非 streaming） 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)) +} diff --git a/visionA-backend/internal/api/camera_ws_test.go b/visionA-backend/internal/api/camera_ws_test.go new file mode 100644 index 0000000..3e619b8 --- /dev/null +++ b/visionA-backend/internal/api/camera_ws_test.go @@ -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）") +} diff --git a/visionA-backend/internal/api/proxy.go b/visionA-backend/internal/api/proxy.go index c4b062b..8d8cae4 100644 --- a/visionA-backend/internal/api/proxy.go +++ b/visionA-backend/internal/api/proxy.go @@ -17,6 +17,7 @@ import ( "io" "net/http" "strings" + "sync" "time" "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:` 走這條回顯）。 +// +// 認證：本 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。 // // 雛形邏輯（單一 user + 單一 agent）：走 Store.List，過濾 userID 對得上的第一筆。 diff --git a/visionA-backend/internal/api/stubs.go b/visionA-backend/internal/api/stubs.go index 670ea17..f2d66df 100644 --- a/visionA-backend/internal/api/stubs.go +++ b/visionA-backend/internal/api/stubs.go @@ -4,17 +4,17 @@ import ( "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 / // 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） // - Clusters 寫入類（Phase 1） // - Converter（Phase 1） // -// Camera / Media 已在 camera.go（registerCameraRoutes）換成真正的 tunnel proxy 宣告。 -// - WebSocket endpoints（B7 TODO — 需要 Hijack + WS relay） +// WebSocket 的 501 stub 在 registerWebSocketStubs（下方），不在此函式。 // // 讓前端對錯誤路徑能拿到 501 而非 404，減少除錯成本。 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")) } -// registerWebSocketStubs 註冊 /ws/* 的 stub。WebSocket proxy 在 B5 雛形不實作， -// 留 501 讓前端能收到明確錯誤，由 B7 補齊。 +// registerWebSocketStubs 註冊「尚未實作」的 /ws/* endpoint，一律回 501，讓前端能收到 +// 明確錯誤，由 B7 補齊。 // -// 為什麼不做 WS proxy：實作 WS relay 需要在 api-server 端做 Hijack、雙向 io.Copy， -// 而且 Forwarder.ForwardWebSocket 尚未實作（見 forwarder.go §ForwardWebSocket）。 -// 加這條路徑會顯著擴張 B5 範圍；按 prompt 指示先留 TODO。 +// WS tunnel proxy 的基礎設施（Forwarder.ForwardWebSocket、api-server 端 Hijack + 雙向 +// io.Copy）已實作（見 forwarder.go §ForwardWebSocket、proxy.go newWebSocketProxyHandler）， +// 且 /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。 func registerWebSocketStubs(r *gin.Engine) { @@ -69,7 +71,9 @@ func registerWebSocketStubs(r *gin.Engine) { // 用 GET（WebSocket upgrade 的初始 HTTP request） 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/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/system", stub("ws.system — pending B7")) r.GET("/ws/clusters/:id/inference", stub("ws.clusters.inference — pending B7")) diff --git a/visionA-backend/internal/session/forwarder.go b/visionA-backend/internal/session/forwarder.go index 02a9119..4d2277c 100644 --- a/visionA-backend/internal/session/forwarder.go +++ b/visionA-backend/internal/session/forwarder.go @@ -261,17 +261,115 @@ func (f *Forwarder) ForwardHTTP(ctx context.Context, sessionToken string, req *h return resp, nil } -// ForwardWebSocket 預留 — B5 接前端 WS 時實作。 +// WebSocketConn 是 ForwardWebSocket 成功後回傳的結果。 // -// 預期實作（草稿）： -// - OpenStream 拿到 raw conn -// - 把 WS upgrade request 透過 conn 寫過去 -// - 等 101 response 回來 -// - Hijack browser 端連線，與 conn 雙向 pipe +// Conn 是「已完成 WS upgrade（101 已讀掉）」的 tunnel 連線，語意上等同 net.Conn： +// 後續讀寫的都是 WebSocket frame bytes（api-server 不解析 frame，只做透明 byte pipe）。 +// Resp 是 local agent 回的 101 Switching Protocols response（含 Sec-WebSocket-Accept +// 等 header）；handler 需把它原樣寫回 browser 端 hijacked 連線，才能完成 browser↔agent +// 的 WS 握手。 // -// 雛形先回 ErrNotImplemented，避免被誤用。 -func (f *Forwarder) ForwardWebSocket(ctx context.Context, sessionToken string, req *http.Request) (net.Conn, error) { - return nil, errors.New("session: ForwardWebSocket not implemented yet (TODO B5)") +// caller 必須負責 Conn.Close()。 +type WebSocketConn struct { + 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 } // ---------------------------------------------------------------------- diff --git a/visionA-backend/internal/session/forwarder_test.go b/visionA-backend/internal/session/forwarder_test.go index 5cccefb..508c897 100644 --- a/visionA-backend/internal/session/forwarder_test.go +++ b/visionA-backend/internal/session/forwarder_test.go @@ -1,13 +1,18 @@ package session import ( + "bufio" "context" "errors" + "io" + "net" "net/http" "net/http/httptest" + "strings" "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) // TestForwarder_OpenStream_NoProxyHost 驗證 baseURL 為空時直接拒絕。 @@ -24,14 +29,138 @@ func TestForwarder_OpenStream_EmptyToken(t *testing.T) { assert.Error(t, err) } -// TestForwarder_ForwardWebSocket_NotImplemented 驗證 ForwardWebSocket 仍是 stub。 -func TestForwarder_ForwardWebSocket_NotImplemented(t *testing.T) { +// TestForwarder_ForwardWebSocket_NilReq 驗證 nil req 直接拒絕。 +func TestForwarder_ForwardWebSocket_NilReq(t *testing.T) { 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) 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 // （session 不存在時的雛形行為）→ 包裝成 ErrSessionNotFound。 //