// Copyright (C) 2026 Dcntrlized // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published // by the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. // // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . package dataplane import ( "context" "errors" "fmt" "io" "net/http" "net/http/httptest" "strings" "sync" "testing" "time" "git.dcntrlized.org/Dcntrlized/Maestro/lib/serviceproxy" "github.com/gorilla/websocket" ) // recorder captures what the fake service saw when the socket was opened. type recorder struct { mu sync.Mutex path string query string headers http.Header } func (rec *recorder) capture(r *http.Request) { rec.mu.Lock() defer rec.mu.Unlock() rec.path = r.URL.Path rec.query = r.URL.RawQuery rec.headers = r.Header.Clone() } func (rec *recorder) snapshot() (path, query string, headers http.Header) { rec.mu.Lock() defer rec.mu.Unlock() return rec.path, rec.query, rec.headers } // startUpstream runs a fake service socket server. run executes after a // successful upgrade and returning from it ends the connection. func startUpstream(t *testing.T, run func(c *websocket.Conn)) (*recorder, string) { t.Helper() rec := &recorder{} upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { rec.capture(r) conn, err := upgrader.Upgrade(w, r, nil) if err != nil { return } defer func() { _ = conn.Close() }() run(conn) })) t.Cleanup(srv.Close) return rec, srv.URL } // startProxy builds the same route table main.go builds, with a fake resolver. func startProxy(t *testing.T, resolve Resolver) *httptest.Server { t.Helper() mux := http.NewServeMux() RegisterWith(mux, resolve) srv := httptest.NewServer(mux) t.Cleanup(srv.Close) return srv } func runningPeer(baseURL string) *serviceproxy.Peer { return &serviceproxy.Peer{Service: "alarm", Running: true, BaseURL: baseURL} } // resolverAnswer resolves every reference to peer, or fails with err. func resolverAnswer(peer *serviceproxy.Peer, err error) Resolver { return func(context.Context, string) (*serviceproxy.Peer, error) { if err != nil { return nil, err } return peer, nil } } func wsURL(t *testing.T, srv *httptest.Server, path string) string { t.Helper() return "ws" + strings.TrimPrefix(srv.URL, "http") + path } // dialProxy connects a test client through the proxy. func dialProxy(t *testing.T, srv *httptest.Server, path string, header http.Header) *websocket.Conn { t.Helper() conn, resp, err := websocket.DefaultDialer.Dial(wsURL(t, srv, path), header) if err != nil { status := 0 if resp != nil { status = resp.StatusCode } t.Fatalf("dial %s: %v (status %d)", path, err, status) } t.Cleanup(func() { _ = conn.Close() }) return conn } // readText reads one text frame, failing the test on timeout or wrong type. func readText(t *testing.T, conn *websocket.Conn) string { t.Helper() if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { t.Fatalf("set read deadline: %v", err) } messageType, data, err := conn.ReadMessage() if err != nil { t.Fatalf("read frame: %v", err) } if messageType != websocket.TextMessage { t.Fatalf("frame type = %d, want text", messageType) } return string(data) } // TestFramesBothWays is (a): a frame from the service reaches the caller and a // frame from the caller reaches the service, through the real upgrade path. func TestFramesBothWays(t *testing.T) { _, upstreamURL := startUpstream(t, func(c *websocket.Conn) { _ = c.WriteMessage(websocket.TextMessage, []byte("alarm-frame-from-service")) for { messageType, data, err := c.ReadMessage() if err != nil { return } if err := c.WriteMessage(messageType, data); err != nil { return } } }) srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil)) conn := dialProxy(t, srv, "/api/alarm/ws/alarm", nil) if got := readText(t, conn); got != "alarm-frame-from-service" { t.Fatalf("service frame = %q, want alarm-frame-from-service", got) } if err := conn.WriteMessage(websocket.TextMessage, []byte("ack-from-client")); err != nil { t.Fatalf("write client frame: %v", err) } if got := readText(t, conn); got != "ack-from-client" { t.Fatalf("echoed client frame = %q, want ack-from-client", got) } } // TestForwardsQueryAndHeaders is (b): the service sees the caller's query string // verbatim and the access token header, and never the caller's cookies. func TestForwardsQueryAndHeaders(t *testing.T) { const query = "alertId=0f8fad5b-0000-4000-8000-000000000001&token=SUBTOKEN&deviceId=pixel-7&accessToken=ACCESSTOKEN" rec, upstreamURL := startUpstream(t, func(c *websocket.Conn) { time.Sleep(50 * time.Millisecond) // hold the socket open long enough to be captured }) srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil)) header := http.Header{} header.Set("X-Access-Token", "ACCESSTOKEN") header.Set("Cookie", "maestro_access=operator-secret") header.Set("Origin", "http://127.0.0.1:8081") _ = dialProxy(t, srv, "/api/alarm/ws/alarm?"+query, header) path, gotQuery, headers := rec.snapshot() if path != "/ws/alarm" { t.Fatalf("upstream path = %q, want /ws/alarm", path) } if gotQuery != query { t.Fatalf("upstream query = %q, want %q", gotQuery, query) } if got := headers.Get("X-Access-Token"); got != "ACCESSTOKEN" { t.Fatalf("upstream X-Access-Token = %q, want ACCESSTOKEN", got) } if got := headers.Get("Origin"); got != "http://127.0.0.1:8081" { t.Fatalf("upstream Origin = %q, want it forwarded", got) } if got := headers.Get("Cookie"); got != "" { t.Fatalf("upstream Cookie = %q, want none — the owner session must never ride into a service", got) } // The dialer owns the handshake headers (Host/Upgrade/Connection/ // Sec-WebSocket-Key), so the upstream MUST see a valid upgrade even though we // never forwarded those from the caller. if got := headers.Get("Upgrade"); !strings.EqualFold(got, "websocket") { t.Fatalf("upstream Upgrade = %q, want the dialer's own handshake", got) } } // TestStatusCodes is (c), (d), (g), (h) and (i) as a table: the proxy answers // with a real HTTP status before it hijacks anything. func TestStatusCodes(t *testing.T) { rejectingUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusUnauthorized) _, _ = io.WriteString(w, `{"error":{"code":"ERR_UNAUTHORIZED","message":"missing access token"}}`) })) t.Cleanup(rejectingUpstream.Close) tests := []struct { name string resolve Resolver path string upgrade bool wantStatus int wantBody string // wantCode is Maestro's own error code; empty means the body is the // service's own shape, mirrored verbatim (never reformatted). wantCode string }{ { name: "non-upgrade request is rejected", resolve: resolverAnswer(runningPeer("http://127.0.0.1:1"), nil), path: "/api/alarm/ws", wantStatus: http.StatusBadRequest, wantBody: "only accepts WebSocket upgrades", wantCode: "ERR_INVALID_REQUEST", }, { name: "unknown service", resolve: resolverAnswer(nil, fmt.Errorf("%w: %q", serviceproxy.ErrNotFound, "alarm")), path: "/api/alarm/ws", upgrade: true, wantStatus: http.StatusNotFound, wantBody: "unknown service", // the quoted name is JSON-escaped inside the envelope wantCode: "ERR_NOT_FOUND", }, { name: "service not running", resolve: resolverAnswer(&serviceproxy.Peer{BaseURL: "http://127.0.0.1:1", Running: false}, nil), path: "/api/alarm/ws", upgrade: true, wantStatus: http.StatusConflict, wantBody: "is not running", wantCode: "ERR_NOT_RUNNING", }, { name: "docker unavailable", resolve: resolverAnswer(nil, fmt.Errorf("%w: cannot connect", serviceproxy.ErrDocker)), path: "/api/alarm/ws", upgrade: true, wantStatus: http.StatusInternalServerError, wantBody: "failed to resolve service", wantCode: "ERR_INTERNAL", }, { name: "upstream refuses the upgrade", resolve: resolverAnswer(runningPeer(rejectingUpstream.URL), nil), path: "/api/alarm/ws", upgrade: true, wantStatus: http.StatusUnauthorized, wantBody: "err_unauthorized", // the service's own envelope, lower-cased by its code }, { name: "upstream is down", resolve: resolverAnswer(runningPeer("http://127.0.0.1:1"), nil), path: "/api/alarm/ws", upgrade: true, wantStatus: http.StatusBadGateway, wantBody: "failed to reach the socket", wantCode: "ERR_UPSTREAM_UNREACHABLE", }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { srv := startProxy(t, test.resolve) var resp *http.Response if test.upgrade { conn, handshake, err := websocket.DefaultDialer.Dial(wsURL(t, srv, test.path), nil) if conn != nil { _ = conn.Close() } if err == nil { t.Fatalf("handshake succeeded, want status %d", test.wantStatus) } resp = handshake } else { var err error resp, err = http.Get(srv.URL + test.path) if err != nil { t.Fatalf("GET %s: %v", test.path, err) } } if resp == nil { t.Fatal("no HTTP response") } defer func() { _ = resp.Body.Close() }() body, err := io.ReadAll(resp.Body) if err != nil { t.Fatalf("read body: %v", err) } if resp.StatusCode != test.wantStatus { t.Fatalf("status = %d, want %d (body: %s)", resp.StatusCode, test.wantStatus, body) } if !strings.Contains(strings.ToLower(string(body)), strings.ToLower(test.wantBody)) { t.Fatalf("body = %q, want it to contain %q", body, test.wantBody) } if test.wantCode != "" { // Maestro's own errors must be the uniform envelope: the phone // clients unwrap {error:{code,message}} and render the message. if got := resp.Header.Get("Content-Type"); !strings.Contains(got, "application/json") { t.Fatalf("Content-Type = %q, want application/json for %s", got, test.wantCode) } if !strings.Contains(string(body), `"code":"`+test.wantCode+`"`) { t.Fatalf("body = %q, want code %s", body, test.wantCode) } if !strings.Contains(string(body), `"message":"`) { t.Fatalf("body = %q, want a human-readable message", body) } } }) } } // TestCloseCodePropagates is (e): a clean close from the service reaches the // caller with the same code and reason, so "the admin disarmed the alert" is // distinguishable from "the network died". func TestCloseCodePropagates(t *testing.T) { _, upstreamURL := startUpstream(t, func(c *websocket.Conn) { if _, _, err := c.ReadMessage(); err != nil { return } _ = c.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "admin disarmed the alert"), time.Now().Add(time.Second)) }) srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil)) conn := dialProxy(t, srv, "/api/alarm/ws/alarm", nil) if err := conn.WriteMessage(websocket.TextMessage, []byte("hello")); err != nil { t.Fatalf("write: %v", err) } if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { t.Fatalf("set read deadline: %v", err) } _, _, err := conn.ReadMessage() var closeErr *websocket.CloseError if !errors.As(err, &closeErr) { t.Fatalf("read = %v, want a close error", err) } if closeErr.Code != websocket.CloseNormalClosure { t.Fatalf("close code = %d, want %d", closeErr.Code, websocket.CloseNormalClosure) } if !strings.Contains(closeErr.Text, "admin disarmed") { t.Fatalf("close reason = %q, want it to carry the service's reason", closeErr.Text) } } // TestIdleConnectionSurvives is (f): no proxy deadline kills a quiet socket. // A 2.5s idle gap is far beyond anything a naive 1s proxy deadline would allow, // and the session must still work afterwards. func TestIdleConnectionSurvives(t *testing.T) { _, upstreamURL := startUpstream(t, func(c *websocket.Conn) { for { messageType, data, err := c.ReadMessage() if err != nil { return } if err := c.WriteMessage(messageType, data); err != nil { return } } }) srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil)) conn := dialProxy(t, srv, "/api/alarm/ws/alarm", nil) // Prove the socket is quiet but alive: say nothing for 2.5s, then use it. time.Sleep(2500 * time.Millisecond) if err := conn.WriteMessage(websocket.TextMessage, []byte("after-idle")); err != nil { t.Fatalf("write after idle: %v", err) } if got := readText(t, conn); got != "after-idle" { t.Fatalf("frame after idle = %q, want after-idle", got) } } // TestPingRelayedNotAnswered is the liveness test: control frames are forwarded // to the other hop instead of being answered by Maestro. // // The caller has a custom ping handler that records the ping it receives, so a // proxy that answered pings itself would leave that channel empty (the alarm // service's ping would never reach the phone). The service's pong channel proves // the return trip: phone pong → proxy → service. func TestPingRelayedNotAnswered(t *testing.T) { servicePongs := make(chan string, 1) _, upstreamURL := startUpstream(t, func(c *websocket.Conn) { c.SetPongHandler(func(appData string) error { select { case servicePongs <- appData: default: } return nil }) if err := c.WriteControl(websocket.PingMessage, []byte("service-ping"), time.Now().Add(time.Second)); err != nil { return } // Keep reading so the pong can arrive. _ = c.SetReadDeadline(time.Now().Add(5 * time.Second)) for { if _, _, err := c.ReadMessage(); err != nil { return } } }) srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil)) conn := dialProxy(t, srv, "/api/alarm/ws/alarm", nil) callerPings := make(chan string, 1) conn.SetPingHandler(func(appData string) error { select { case callerPings <- appData: default: } return conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(time.Second)) }) // The caller must read for control frames to be processed at all. go func() { for { if _, _, err := conn.ReadMessage(); err != nil { return } } }() select { case got := <-callerPings: if got != "service-ping" { t.Fatalf("relayed ping payload = %q, want service-ping", got) } case <-time.After(5 * time.Second): t.Fatal("the service's ping never reached the caller — the proxy answered it locally") } select { case got := <-servicePongs: if got != "service-ping" { t.Fatalf("pong payload = %q, want service-ping", got) } case <-time.After(5 * time.Second): t.Fatal("the caller's pong never reached the service — the ping/pong path is not end-to-end") } } // TestPathScoping is (3): only the service's /ws namespace is reachable. func TestPathScoping(t *testing.T) { // Unit level: the only paths that can be produced start at /ws. paths := []struct { sub string wantPath string wantOK bool }{ {"", "/ws", true}, {"alarm", "/ws/alarm", true}, {"alarm/v2", "/ws/alarm/v2", true}, {"/alarm", "/ws/alarm", true}, {"../config", "", false}, {"a/../config", "", false}, {"..", "", false}, {"a//b", "", false}, {"alarm?x=1", "", false}, {"alarm#x", "", false}, {"%2e%2e", "", false}, } for _, test := range paths { gotPath, ok := upstreamPath(namespaceWS, test.sub) if ok != test.wantOK || gotPath != test.wantPath { t.Fatalf("upstreamPath(ws, %q) = (%q, %v), want (%q, %v)", test.sub, gotPath, ok, test.wantPath, test.wantOK) } } // Route level: /api/{service}/config is not a proxy route (404), and the // sub-path variant reaches the service's own /ws/alarm path. rec, upstreamURL := startUpstream(t, func(c *websocket.Conn) { time.Sleep(50 * time.Millisecond) }) srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil)) _ = dialProxy(t, srv, "/api/alarm/ws/alarm", nil) if path, _, _ := rec.snapshot(); path != "/ws/alarm" { t.Fatalf("upstream path = %q, want /ws/alarm", path) } for _, path := range []string{"/api/alarm/config", "/api/typo"} { resp, err := http.Get(srv.URL + path) if err != nil { t.Fatalf("GET %s: %v", path, err) } body, readErr := io.ReadAll(resp.Body) _ = resp.Body.Close() if readErr != nil { t.Fatalf("read %s body: %v", path, readErr) } if resp.StatusCode != http.StatusNotFound { t.Fatalf("GET %s = %d, want 404 — this route must not tunnel to /config", path, resp.StatusCode) } // The fallback's own 404 must use Maestro's JSON envelope, not plain text: // the phone clients unwrap {error:{code,message}} into a rendered sentence. if got := resp.Header.Get("Content-Type"); !strings.Contains(got, "application/json") { t.Fatalf("GET %s Content-Type = %q, want application/json", path, got) } if !strings.Contains(string(body), `"code":"ERR_NOT_FOUND"`) || !strings.Contains(string(body), `"message":"unknown API path"`) { t.Fatalf("GET %s body = %q, want the ERR_NOT_FOUND envelope with a message", path, body) } } } // TestConcurrentSessions keeps -race honest: several proxied sessions at once, // each with its own frames. func TestConcurrentSessions(t *testing.T) { _, upstreamURL := startUpstream(t, func(c *websocket.Conn) { for { messageType, data, err := c.ReadMessage() if err != nil { return } if err := c.WriteMessage(messageType, data); err != nil { return } } }) srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil)) var wg sync.WaitGroup for i := 0; i < 4; i++ { wg.Add(1) go func(id int) { defer wg.Done() conn, _, err := websocket.DefaultDialer.Dial(wsURL(t, srv, "/api/alarm/ws"), nil) if err != nil { t.Errorf("session %d dial: %v", id, err) return } defer func() { _ = conn.Close() }() payload := fmt.Sprintf("session-%d", id) if err := conn.WriteMessage(websocket.TextMessage, []byte(payload)); err != nil { t.Errorf("session %d write: %v", id, err) return } _ = conn.SetReadDeadline(time.Now().Add(5 * time.Second)) _, data, err := conn.ReadMessage() if err != nil { t.Errorf("session %d read: %v", id, err) return } if string(data) != payload { t.Errorf("session %d got %q, want %q", id, data, payload) } }(i) } wg.Wait() } // TestProductionPatternSetDoesNotPanic constructs the SAME pattern set main.go // constructs, because a ServeMux conflict is a boot panic that `go build` and // `go vet` cannot see. The data plane is a single "/api/" fallback precisely so // that no pattern in this set can collide with the legacy action route. func TestProductionPatternSetDoesNotPanic(t *testing.T) { marker := func(name string) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = io.WriteString(w, name) }) } mux := http.NewServeMux() // The console's own routes, exactly as main.go registers them. mux.Handle("/api/instance", marker("instance")) mux.Handle("/api/containers", marker("containers")) mux.Handle("/api/actions/{containerId}/{actionName}", marker("legacy-action")) mux.Handle("/api/services/{service}/config", marker("config")) mux.Handle("/api/ws", marker("maestro-ws")) // Panics here if the data plane ever regresses to per-shape patterns. RegisterWith(mux, resolverAnswer(runningPeer("http://127.0.0.1:1"), nil)) serve := func(path string) *httptest.ResponseRecorder { rec := httptest.NewRecorder() mux.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "http://x"+path, nil)) return rec } if rec := serve("/api/actions/abc/createAlert"); rec.Body.String() != "legacy-action" { t.Fatalf("legacy action route was shadowed: body = %q", rec.Body.String()) } if rec := serve("/api/services/alarm/config"); rec.Body.String() != "config" { t.Fatalf("config route was shadowed: body = %q", rec.Body.String()) } if rec := serve("/api/instance"); rec.Body.String() != "instance" { t.Fatalf("instance route was shadowed: body = %q", rec.Body.String()) } // The data plane answers its own namespaces through the fallback. if rec := serve("/api/alarm/ws"); rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "only accepts WebSocket upgrades") { t.Fatalf("/api/alarm/ws = %d %q, want the data plane's 400", rec.Code, rec.Body.String()) } if rec := serve("/api/alarm/config"); rec.Code != http.StatusNotFound { t.Fatalf("/api/alarm/config = %d, want 404 — the data plane must not tunnel to /config", rec.Code) } } // TestNaivePatternSetPanics pins WHY the fallback exists: the literal patterns for // this surface cannot be registered next to the legacy action route. If a future // Go release makes this set legal, this test fails and the fallback can be // replaced by real patterns; until then, the fallback is the only option. func TestNaivePatternSetPanics(t *testing.T) { marker := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) for _, naive := range []string{ "/api/{service}/action/{actionName}", "/api/{service}/auth/{path...}", "/api/{service}/ws/{path...}", } { t.Run(naive, func(t *testing.T) { defer func() { if recover() == nil { t.Fatalf("registering %q next to the legacy action route did NOT panic; "+ "the /api/ fallback can be reconsidered", naive) } }() mux := http.NewServeMux() mux.Handle("/api/actions/{containerId}/{actionName}", marker) mux.Handle(naive, marker) }) } } // TestUpgradeOnAnotherNamespaceUsesWSRelay pins the single dispatch rule: upgrade // presence, not the namespace name, selects the WebSocket relay. func TestUpgradeOnAnotherNamespaceUsesWSRelay(t *testing.T) { rec, upstreamURL := startUpstream(t, func(c *websocket.Conn) { _ = c.WriteMessage(websocket.TextMessage, []byte("events-frame")) time.Sleep(100 * time.Millisecond) }) srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil)) conn := dialProxy(t, srv, "/api/alarm/auth/events", nil) if got := readText(t, conn); got != "events-frame" { t.Fatalf("frame = %q, want events-frame", got) } if path, _, _ := rec.snapshot(); path != "/auth/events" { t.Fatalf("upstream path = %q, want /auth/events", path) } }