diff --git a/src/handlers/actions/actions.go b/src/handlers/actions/actions.go
index d7095d1..b097c28 100644
--- a/src/handlers/actions/actions.go
+++ b/src/handlers/actions/actions.go
@@ -17,15 +17,14 @@ package actions
import (
"bytes"
- "context"
+ "errors"
"fmt"
"io"
"net/http"
"strings"
- "git.dcntrlized.org/Dcntrlized/Maestro/lib/authmgmt"
- "git.dcntrlized.org/Dcntrlized/Maestro/lib/docker"
"git.dcntrlized.org/Dcntrlized/Maestro/lib/logger"
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/serviceproxy"
)
func RunAction(w http.ResponseWriter, r *http.Request) {
@@ -35,43 +34,31 @@ func RunAction(w http.ResponseWriter, r *http.Request) {
logger.Debug("Received request to run action %s on container id %s", actionName, containerId)
- // Get IP address of the container using containerId
- dockerClient, err := docker.NewDockerClient()
+ // Resolve "where is this service" through the shared helper — the WebSocket
+ // data plane (/api/{service}/ws) resolves through the same code, so the two
+ // edges can never disagree about a service's address. The status codes and
+ // messages below are the pre-refactor ones: only the lookup moved.
+ baseURL, err := serviceproxy.Target(r.Context(), containerId)
if err != nil {
- http.Error(w, "Failed to create Docker client", http.StatusInternalServerError)
- return
- }
-
- inspectData, err := docker.ResolveContainer(context.Background(), dockerClient, containerId)
- if err != nil {
- http.Error(w, "Failed to inspect container", http.StatusInternalServerError)
- return
- }
-
- network, _, err := docker.ExtractNetworkInfo(context.Background(), dockerClient, inspectData)
- if err != nil {
- http.Error(w, "Failed to extract network info", http.StatusInternalServerError)
- return
- }
-
- // The service port can be declared via the org.dcntrlized.port label. Fall back to the
- // first mapped guest port when it is not set.
- port := inspectData.Config.Labels["org.dcntrlized.port"]
- if port == "" && len(network.Ports) > 0 {
- port = fmt.Sprintf("%d", network.Ports[0].Guest)
- }
- if port == "" {
- http.Error(w, "Container has no configured service port", http.StatusInternalServerError)
+ // Same status (500) and same message text as before the refactor; the real
+ // cause now goes to the debug log instead of being folded into the message.
+ logger.Debug("Failed to resolve container %s: %v", containerId, err)
+ switch {
+ case errors.Is(err, serviceproxy.ErrNoPort):
+ http.Error(w, "Container has no configured service port", http.StatusInternalServerError)
+ case errors.Is(err, serviceproxy.ErrDockerClient):
+ http.Error(w, "Failed to create Docker client", http.StatusInternalServerError)
+ case errors.Is(err, serviceproxy.ErrNoNetwork):
+ http.Error(w, "Failed to extract network info", http.StatusInternalServerError)
+ default:
+ http.Error(w, "Failed to inspect container", http.StatusInternalServerError)
+ }
return
}
// Construct the URL for the action endpoint, preserving the query string.
- // The service peer edge is mutual TLS (M2 §2): Maestro presents its own
- // identity cert and the service's listener requires a CA-issued client
- // cert, so only Maestro can drive actions (T4). In dev mode
- // (DCNTRLIZED_DEV=true) the satellites keep a plaintext listener for the
- // RN clients, so the action edge stays plaintext there too.
- actionURL := fmt.Sprintf("%s://%s:%s/action/%s", authmgmt.Scheme(), network.IPAddress.Address, port, actionName)
+ // Rev 9: the service edge is plaintext HTTP on the single :8080 listener.
+ actionURL := fmt.Sprintf("%s/action/%s", baseURL, actionName)
if r.URL.RawQuery != "" {
actionURL += "?" + r.URL.RawQuery
}
@@ -99,16 +86,8 @@ func RunAction(w http.ResponseWriter, r *http.Request) {
}
}
- // Maestro's own identity (mTLS) as the peer; plaintext in dev mode.
- transport, err := authmgmt.PeerTransport()
- if err != nil {
- logger.Debug("Failed to build peer transport for action %s on container %s: %v", actionName, containerId, err)
- http.Error(w, "Failed to build peer transport", http.StatusInternalServerError)
- return
- }
-
- // Forward the request to the container's action endpoint
- resp, err := (&http.Client{Transport: transport}).Do(req)
+ // Forward the request to the container's action endpoint (plaintext).
+ resp, err := (&http.Client{}).Do(req)
if err != nil {
logger.Debug("Failed to forward action request %s to container %s: %v", actionName, containerId, err)
http.Error(w, "Failed to forward request to container", http.StatusInternalServerError)
diff --git a/src/handlers/dataplane/dataplane.go b/src/handlers/dataplane/dataplane.go
new file mode 100644
index 0000000..cf9bc48
--- /dev/null
+++ b/src/handlers/dataplane/dataplane.go
@@ -0,0 +1,307 @@
+// 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 is Maestro's single data plane for every installed service.
+//
+// The model (user ruling, 2026-10-18): requests to a service are always expected
+// to be routed through Maestro, and may well be unauthenticated. Only one port
+// per instance therefore needs to be open, and setup stays simple. A service sets
+// up its own auth on top (typically via dcntrlized-auth); Maestro neither knows
+// nor cares whether a request is authenticated, and direct service addressing is
+// not a supported deployment.
+//
+// Reachable surface — exactly this, nothing else:
+//
+// /api/{service}/action/{actionName} → service /action/{actionName}
+// /api/{service}/auth/{path...} → service /auth/{path...}
+// /api/{service}/ws[/{path...}] → service /ws[/{path...}]
+//
+// ONE allowlist, ONE dispatcher, ONE passthrough: a request is relayed as a
+// WebSocket when it is an upgrade and as plain HTTP otherwise. Adding a service
+// namespace therefore takes no Maestro code beyond a name in this file, and the
+// two halves can never disagree about a service's address because both resolve
+// through lib/serviceproxy.
+//
+// WHAT THIS IS NOT: a general tunnel. /config and /_manifest are unreachable
+// because they are not namespaces here. /config in particular is Maestro's own
+// ES256-signed edge — Maestro mints the token and the service verifies it, so
+// exposing it through the data plane would let a caller skip that mint.
+//
+// NOT OWNER-GATED, deliberately (the reasoning lives on the registration in
+// main.go): the callers are apps and phones, which hold no Maestro owner session
+// and must never be handed one — that cookie belongs to the instance operator.
+// The service is the gate.
+//
+// Maestro envelopes ONLY what Maestro itself produced (400/404/409/500/502). A
+// service response — success, 4xx, 5xx, redirect, raw body — is mirrored
+// byte-for-byte: the service's body is the service's contract.
+package dataplane
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "strings"
+
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/logger"
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/serviceproxy"
+)
+
+// The allowlist. A namespace is reachable only if it appears here.
+const (
+ namespaceAction = "action"
+ namespaceAuth = "auth"
+ namespaceWS = "ws"
+)
+
+// Resolver resolves a service reference to its dialable peer. Production uses
+// serviceproxy.Resolve; tests inject a fake so no Docker daemon is required.
+type Resolver func(ctx context.Context, service string) (*serviceproxy.Peer, error)
+
+// Register mounts the service data plane on mux.
+func Register(mux *http.ServeMux) { RegisterWith(mux, serviceproxy.Resolve) }
+
+// RegisterWith is Register with an injectable resolver (tests).
+//
+// Why a single "/api/" fallback instead of one pattern per shape: ServeMux
+// panics at boot when two patterns match a path and neither is more specific.
+// The naive pattern for the action surface, /api/{service}/action/{actionName},
+// collides with the legacy route /api/actions/{containerId}/{actionName} on
+// "/api/actions/action/" (segment 2 is a wildcard in one pattern and the
+// literal "actions" in the other, so the literal "action" sits at different
+// positions) — and /api/{service}/auth/{path...} collides with it the same way.
+// Constructing the pattern set is the only way to see this: `go build` and
+// `go vet` cannot, because the mux is assembled inside main().
+//
+// The fallback is only reached when no concrete route matched, so every existing
+// console route keeps winning its own paths, and the requested public URLs are
+// exactly the ones in the package doc. TestProductionPatternSetDoesNotPanic
+// pins the pattern set; TestNaivePatternSetPanics pins the trap.
+func RegisterWith(mux *http.ServeMux, resolve Resolver) {
+ mux.Handle("/api/", fallback(resolve))
+}
+
+// fallback dispatches the data plane and answers 404 for every other /api/ path
+// that no concrete route claimed.
+func fallback(resolve Resolver) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ service, namespace, rest, ok := parse(r.URL.Path)
+ if !ok {
+ // Every other error under /api/ is a JSON envelope, and the phone
+ // clients unwrap {error:{code,message}} into a rendered sentence; a
+ // bare "404 page not found" would show up as raw text in the app.
+ writeAPIError(w, http.StatusNotFound, "ERR_NOT_FOUND", "unknown API path")
+ return
+ }
+
+ target, ok := upstreamPath(namespace, rest)
+ if !ok {
+ writeAPIError(w, http.StatusNotFound, "ERR_NOT_FOUND", unknownPathMessage(namespace))
+ return
+ }
+
+ // Upgrade presence decides the relay. A socket is a socket wherever the
+ // service puts it, so /api/{service}/auth/events upgrading is the same
+ // code path as /api/{service}/ws.
+ if isUpgrade(r) {
+ relayWebSocket(resolve, service, target, w, r)
+ return
+ }
+ if namespace == namespaceWS {
+ // A plain GET on a socket endpoint is a caller mistake, not a request
+ // for the service's (non-existent) HTTP view of /ws.
+ writeAPIError(w, http.StatusBadRequest, "ERR_INVALID_REQUEST", "this endpoint only accepts WebSocket upgrades")
+ return
+ }
+ relayHTTP(resolve, service, target, w, r)
+ })
+}
+
+// parse splits /api/{service}/{namespace}[/{rest...}].
+func parse(path string) (service, namespace, rest string, ok bool) {
+ trimmed, found := strings.CutPrefix(path, "/api/")
+ if !found {
+ return "", "", "", false
+ }
+ parts := strings.Split(trimmed, "/")
+ if len(parts) < 2 || !validServiceName(parts[0]) || !allowedNamespace(parts[1]) {
+ return "", "", "", false
+ }
+ return parts[0], parts[1], strings.Join(parts[2:], "/"), true
+}
+
+func allowedNamespace(namespace string) bool {
+ switch namespace {
+ case namespaceAction, namespaceAuth, namespaceWS:
+ return true
+ }
+ return false
+}
+
+// upstreamPath maps the namespace and its remainder onto the service's own path.
+// rest arrives as raw text, so it is validated here: this is the boundary between
+// "a path a caller chose" and "a URL Maestro dials".
+func upstreamPath(namespace, rest string) (string, bool) {
+ segments, ok := splitSegments(rest)
+ if !ok {
+ return "", false
+ }
+ switch namespace {
+ case namespaceWS:
+ if len(segments) == 0 {
+ return "/ws", true
+ }
+ return "/ws/" + strings.Join(segments, "/"), true
+ case namespaceAction:
+ // An action is one name; /action/{name}/{junk} is not a thing.
+ if len(segments) != 1 {
+ return "", false
+ }
+ return "/action/" + segments[0], true
+ case namespaceAuth:
+ // /auth/{path...} — login, register, logout, me, refresh, and any future
+ // auth sub-path, without Maestro needing to know them.
+ if len(segments) == 0 {
+ return "", false
+ }
+ return "/auth/" + strings.Join(segments, "/"), true
+ }
+ return "", false
+}
+
+func unknownPathMessage(namespace string) string {
+ switch namespace {
+ case namespaceAction:
+ return "an action path is /api/{service}/action/{action}"
+ case namespaceAuth:
+ return "an auth path is /api/{service}/auth/{path}"
+ case namespaceWS:
+ return "only the service's /ws namespace is proxied"
+ }
+ return "unknown API path"
+}
+
+// splitSegments validates a slash-separated remainder: no empty segments, no dot
+// segments and nothing outside an unescaped-URL-safe charset. Descriptions are
+// deliberately conservative — a namespace that accepted "../" would be a tunnel.
+func splitSegments(rest string) ([]string, bool) {
+ rest = strings.TrimPrefix(rest, "/")
+ if rest == "" {
+ return nil, true
+ }
+ segments := strings.Split(rest, "/")
+ for _, segment := range segments {
+ if segment == "" || segment == "." || segment == ".." {
+ return nil, false
+ }
+ for _, char := range segment {
+ switch {
+ case char >= 'a' && char <= 'z',
+ char >= 'A' && char <= 'Z',
+ char >= '0' && char <= '9',
+ char == '-', char == '_', char == '.', char == '~':
+ default:
+ return nil, false
+ }
+ }
+ }
+ return segments, true
+}
+
+// validServiceName keeps the reference a plain service identifier before it is
+// handed to Docker: no dot segments, no encodings, nothing that could be read as
+// a path elsewhere.
+func validServiceName(name string) bool {
+ if name == "" || name == "." || name == ".." || len(name) > 128 {
+ return false
+ }
+ for _, char := range name {
+ switch {
+ case char >= 'a' && char <= 'z',
+ char >= 'A' && char <= 'Z',
+ char >= '0' && char <= '9',
+ char == '-', char == '_', char == '.':
+ default:
+ return false
+ }
+ }
+ return true
+}
+
+// resolvePeer resolves the service and refuses to relay to a stopped container.
+func resolvePeer(resolve Resolver, service string, w http.ResponseWriter, r *http.Request) (*serviceproxy.Peer, bool) {
+ peer, err := resolve(r.Context(), service)
+ if err != nil {
+ writeResolveError(w, service, err)
+ return nil, false
+ }
+ if !peer.Running {
+ writeAPIError(w, http.StatusConflict, "ERR_NOT_RUNNING", fmt.Sprintf("service %q is not running", service))
+ return nil, false
+ }
+ return peer, true
+}
+
+// errorBody is Maestro's uniform API error shape (same envelope as the owner-gated
+// handlers, with a message so every error still says something human-readable).
+type errorBody struct {
+ Error struct {
+ Code string `json:"code"`
+ Message string `json:"message,omitempty"`
+ } `json:"error"`
+}
+
+func writeAPIError(w http.ResponseWriter, status int, code, message string) {
+ var body errorBody
+ body.Error.Code = code
+ body.Error.Message = message
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(status)
+ _ = json.NewEncoder(w).Encode(body)
+}
+
+// writeResolveError maps resolution failures onto the documented statuses.
+func writeResolveError(w http.ResponseWriter, service string, err error) {
+ switch {
+ case errors.Is(err, serviceproxy.ErrNotFound):
+ writeAPIError(w, http.StatusNotFound, "ERR_NOT_FOUND", fmt.Sprintf("unknown service %q", service))
+ case errors.Is(err, serviceproxy.ErrNotRunning):
+ writeAPIError(w, http.StatusConflict, "ERR_NOT_RUNNING", fmt.Sprintf("service %q is not running", service))
+ default:
+ logger.Debug("Service resolution failed for %q: %v", service, err)
+ // Maestro's own plumbing failed (dead daemon, no network, no port): a 502
+ // would send an operator to the service's logs for a failure that is not
+ // there.
+ writeAPIError(w, http.StatusInternalServerError, "ERR_INTERNAL",
+ fmt.Sprintf("failed to resolve service %q: %v", service, err))
+ }
+}
+
+// writeUnusableAddress answers a malformed base URL, which is Maestro's own
+// plumbing failing — never a 502.
+func writeUnusableAddress(w http.ResponseWriter, service string, err error) {
+ logger.Debug("Bad service base URL for %q: %v", service, err)
+ writeAPIError(w, http.StatusInternalServerError, "ERR_INTERNAL",
+ fmt.Sprintf("service %q has an unusable address", service))
+}
+
+// writeUpstreamUnreachable answers a transport failure to a resolved service.
+func writeUpstreamUnreachable(w http.ResponseWriter, service string, err error) {
+ logger.Debug("Service relay failed for %q: %v", service, err)
+ writeAPIError(w, http.StatusBadGateway, "ERR_UPSTREAM_UNREACHABLE",
+ fmt.Sprintf("failed to reach service %q: %v", service, err))
+}
diff --git a/src/handlers/dataplane/http.go b/src/handlers/dataplane/http.go
new file mode 100644
index 0000000..b923978
--- /dev/null
+++ b/src/handlers/dataplane/http.go
@@ -0,0 +1,148 @@
+// 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 (
+ "io"
+ "net"
+ "net/http"
+ "strings"
+ "time"
+
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/logger"
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/serviceproxy"
+)
+
+// connectTimeout bounds reaching the service. There is deliberately no overall
+// client timeout: an action may legitimately take minutes (a scan, a sync, a
+// long poll). The caller's request context bounds the request, exactly as it
+// would for any other handler.
+const connectTimeout = 10 * time.Second
+
+// relayClient is the plain-HTTP half of the data plane. A fresh http.Client per
+// request would leak connection pools; this one keeps them warm across calls.
+var relayClient = &http.Client{
+ Transport: &http.Transport{
+ DialContext: (&net.Dialer{Timeout: connectTimeout, KeepAlive: 30 * time.Second}).DialContext,
+ MaxIdleConns: 64,
+ MaxIdleConnsPerHost: 8,
+ IdleConnTimeout: 90 * time.Second,
+ ExpectContinueTimeout: 1 * time.Second,
+ },
+ // Mirror the service's redirects instead of following them: a 302 from a
+ // service is an answer to the caller, not an instruction to Maestro.
+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+}
+
+// relayHTTP forwards one request to the service and mirrors the answer.
+func relayHTTP(resolve Resolver, service, upstreamPath string, w http.ResponseWriter, r *http.Request) {
+ peer, ok := resolvePeer(resolve, service, w, r)
+ if !ok {
+ return
+ }
+
+ // The query string is forwarded verbatim (tokens live there and re-encoding
+ // would corrupt them).
+ target, err := serviceproxy.Endpoint(peer.BaseURL, upstreamPath, r.URL.RawQuery)
+ if err != nil {
+ writeUnusableAddress(w, service, err)
+ return
+ }
+
+ req, err := http.NewRequestWithContext(r.Context(), r.Method, target, r.Body)
+ if err != nil {
+ logger.Debug("Failed to build relay request for %q: %v", service, err)
+ writeAPIError(w, http.StatusInternalServerError, "ERR_INTERNAL",
+ "failed to build the request for service "+service)
+ return
+ }
+ req.ContentLength = r.ContentLength
+ req.Header = forwardableRequestHeaders(r.Header)
+
+ resp, err := relayClient.Do(req)
+ if err != nil {
+ writeUpstreamUnreachable(w, service, err)
+ return
+ }
+ defer func() { _ = resp.Body.Close() }()
+
+ // The service's status, headers and body ARE the answer: mirrored, never
+ // re-shaped and never enveloped (see the package doc). Only Maestro's own
+ // failures above use the envelope.
+ copyResponseHeaders(w.Header(), resp.Header)
+ w.WriteHeader(resp.StatusCode)
+ if _, err := io.Copy(w, resp.Body); err != nil {
+ logger.Debug("Relay of service %q response to client failed: %v", service, err)
+ }
+}
+
+// forwardableRequestHeaders copies the caller's headers, minus the ones a relay
+// must never forward.
+func forwardableRequestHeaders(src http.Header) http.Header {
+ dst := http.Header{}
+ for name, values := range src {
+ if !forwardableRequestHeader(name) {
+ continue
+ }
+ for _, value := range values {
+ dst.Add(name, value)
+ }
+ }
+ return dst
+}
+
+// forwardableRequestHeader is a SECURITY decision, not a convenience.
+//
+// Hop-by-hop headers are the relay's own business. Cookie and Authorization are
+// excluded deliberately twice over: this surface is not owner-gated, so a Maestro
+// session cookie must never ride into a service, and services authenticate with
+// their own tokens (X-Access-Token / their own Cookie issued by themselves, which
+// this relay can never forward because it never forwards cookies at all).
+// Sec-WebSocket-* is set by the WebSocket dialer itself; forwarding a caller's key
+// would corrupt the handshake.
+func forwardableRequestHeader(name string) bool {
+ switch strings.ToLower(name) {
+ case "host", "content-length", "connection", "keep-alive",
+ "proxy-authenticate", "proxy-authorization", "te", "trailer",
+ "transfer-encoding", "upgrade",
+ "cookie", "authorization":
+ return false
+ }
+ return !strings.HasPrefix(strings.ToLower(name), "sec-websocket-")
+}
+
+// copyResponseHeaders mirrors the service's response headers, minus hop-by-hop
+// ones and Set-Cookie: that cookie would be scoped to Maestro's own origin, so
+// mirroring it would let any service write cookies onto the console's origin.
+func copyResponseHeaders(dst, src http.Header) {
+ for name, values := range src {
+ if !forwardableResponseHeader(name) {
+ continue
+ }
+ for _, value := range values {
+ dst.Add(name, value)
+ }
+ }
+}
+
+func forwardableResponseHeader(name string) bool {
+ switch strings.ToLower(name) {
+ case "connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
+ "te", "trailer", "transfer-encoding", "upgrade", "set-cookie":
+ return false
+ }
+ return true
+}
diff --git a/src/handlers/dataplane/http_test.go b/src/handlers/dataplane/http_test.go
new file mode 100644
index 0000000..87ca1a1
--- /dev/null
+++ b/src/handlers/dataplane/http_test.go
@@ -0,0 +1,546 @@
+// 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"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "testing"
+
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/serviceproxy"
+ "github.com/gorilla/websocket"
+)
+
+// httpRecorder captures what the fake service saw for a plain-HTTP relay.
+type httpRecorder struct {
+ mu sync.Mutex
+ calls int
+ method string
+ path string
+ query string
+ headers http.Header
+ body string
+}
+
+func (rec *httpRecorder) record(r *http.Request) {
+ body, _ := io.ReadAll(r.Body)
+ rec.mu.Lock()
+ defer rec.mu.Unlock()
+ rec.calls++
+ rec.method = r.Method
+ rec.path = r.URL.Path
+ rec.query = r.URL.RawQuery
+ rec.headers = r.Header.Clone()
+ rec.body = string(body)
+}
+
+func (rec *httpRecorder) snapshot() (calls int, method, path, query string, headers http.Header, body string) {
+ rec.mu.Lock()
+ defer rec.mu.Unlock()
+ return rec.calls, rec.method, rec.path, rec.query, rec.headers, rec.body
+}
+
+// startHTTPService runs a fake service that records the request and answers with
+// the handler's response.
+func startHTTPService(t *testing.T, answer http.HandlerFunc) (*httpRecorder, string) {
+ t.Helper()
+ rec := &httpRecorder{}
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ rec.record(r)
+ answer(w, r)
+ }))
+ t.Cleanup(srv.Close)
+ return rec, srv.URL
+}
+
+// TestActionRelayed is the core of the model: an app calls
+// /api/{service}/action/{action} with no Maestro session, and the service receives
+// the action, the query, the body and the token — but never the operator's cookie
+// or an Authorization header.
+func TestActionRelayed(t *testing.T) {
+ const query = "alertId=0f8fad5b-0000-4000-8000-000000000001&accessToken=ACCESSTOKEN"
+ rec, upstreamURL := startHTTPService(t, func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusCreated)
+ _, _ = io.WriteString(w, `{"ok":true,"alert":{"id":"a1"}}`)
+ })
+ srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil))
+
+ req, err := http.NewRequest(http.MethodPost, srv.URL+"/api/alarm/action/createAlert?"+query,
+ strings.NewReader(`{"payload":{"alertId":"a1"}}`))
+ if err != nil {
+ t.Fatalf("build request: %v", err)
+ }
+ req.Header.Set("Content-Type", "application/json")
+ req.Header.Set("X-Access-Token", "ACCESSTOKEN")
+ req.Header.Set("X-Alarm-Token", "SUBTOKEN")
+ req.Header.Set("Cookie", "maestro_access=operator-secret; maestro_refresh=operator-refresh")
+ req.Header.Set("Authorization", "Bearer operator-session")
+
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("POST through the data plane: %v", err)
+ }
+ defer func() { _ = resp.Body.Close() }()
+ body, _ := io.ReadAll(resp.Body)
+
+ if resp.StatusCode != http.StatusCreated {
+ t.Fatalf("status = %d, want 201 (body: %s)", resp.StatusCode, body)
+ }
+ if string(body) != `{"ok":true,"alert":{"id":"a1"}}` {
+ t.Fatalf("body = %q, want the service's body byte-for-byte", body)
+ }
+ if got := resp.Header.Get("Content-Type"); !strings.Contains(got, "application/json") {
+ t.Fatalf("Content-Type = %q, want the service's", got)
+ }
+
+ calls, method, path, gotQuery, headers, gotBody := rec.snapshot()
+ if calls != 1 {
+ t.Fatalf("service calls = %d, want 1", calls)
+ }
+ if method != http.MethodPost {
+ t.Fatalf("method = %q, want POST", method)
+ }
+ if path != "/action/createAlert" {
+ t.Fatalf("path = %q, want /action/createAlert", path)
+ }
+ if gotQuery != query {
+ t.Fatalf("query = %q, want it forwarded verbatim (%q)", gotQuery, query)
+ }
+ if gotBody != `{"payload":{"alertId":"a1"}}` {
+ t.Fatalf("body = %q, want the caller's body", gotBody)
+ }
+ if got := headers.Get("X-Access-Token"); got != "ACCESSTOKEN" {
+ t.Fatalf("X-Access-Token = %q, want ACCESSTOKEN", got)
+ }
+ if got := headers.Get("X-Alarm-Token"); got != "SUBTOKEN" {
+ t.Fatalf("X-Alarm-Token = %q, want SUBTOKEN", got)
+ }
+ if got := headers.Get("Content-Type"); !strings.Contains(got, "application/json") {
+ t.Fatalf("Content-Type at the service = %q, want application/json", got)
+ }
+ if got := headers.Get("Cookie"); got != "" {
+ t.Fatalf("Cookie = %q, want none — the operator's session must never ride into a service", got)
+ }
+ if got := headers.Get("Authorization"); got != "" {
+ t.Fatalf("Authorization = %q, want none — services authenticate with their own tokens", got)
+ }
+}
+
+// TestAuthRelayed: /api/{service}/auth/{path...} reaches the service's /auth/*
+// namespace with the raw JSON body, which is what the phone clients call directly.
+func TestAuthRelayed(t *testing.T) {
+ rec, upstreamURL := startHTTPService(t, func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusUnauthorized)
+ // A service error body in the service's own shape — must NOT be
+ // reformatted into Maestro's envelope.
+ _, _ = io.WriteString(w, `{"error":{"code":"ERR_INVALID_CREDENTIALS","message":"wrong password"}}`)
+ })
+ srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil))
+
+ resp, err := http.Post(srv.URL+"/api/alarm/auth/login", "application/json",
+ strings.NewReader(`{"username":"admin","password":"bad"}`))
+ if err != nil {
+ t.Fatalf("POST /auth/login: %v", err)
+ }
+ defer func() { _ = resp.Body.Close() }()
+ body, _ := io.ReadAll(resp.Body)
+
+ if resp.StatusCode != http.StatusUnauthorized {
+ t.Fatalf("status = %d, want the service's 401 (body: %s)", resp.StatusCode, body)
+ }
+ if string(body) != `{"error":{"code":"ERR_INVALID_CREDENTIALS","message":"wrong password"}}` {
+ t.Fatalf("body = %q, want the service's body untouched", body)
+ }
+
+ _, method, path, _, _, gotBody := rec.snapshot()
+ if method != http.MethodPost || path != "/auth/login" {
+ t.Fatalf("service saw %s %s, want POST /auth/login", method, path)
+ }
+ if gotBody != `{"username":"admin","password":"bad"}` {
+ t.Fatalf("body at the service = %q, want the raw JSON", gotBody)
+ }
+}
+
+// TestAuthPathDepth keeps the auth namespace a prefix, not a fixed list: a future
+// service auth sub-path needs no Maestro change.
+func TestAuthPathDepth(t *testing.T) {
+ rec, upstreamURL := startHTTPService(t, func(w http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(http.StatusNoContent)
+ })
+ srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil))
+
+ resp, err := http.Post(srv.URL+"/api/alarm/auth/device/pair/confirm", "application/json", nil)
+ if err != nil {
+ t.Fatalf("POST: %v", err)
+ }
+ _ = resp.Body.Close()
+ if resp.StatusCode != http.StatusNoContent {
+ t.Fatalf("status = %d, want 204", resp.StatusCode)
+ }
+ if _, _, path, _, _, _ := rec.snapshot(); path != "/auth/device/pair/confirm" {
+ t.Fatalf("path = %q, want /auth/device/pair/confirm", path)
+ }
+}
+
+// TestMirrorsServiceResponses: whatever the service answers is the caller's
+// answer — status, body and content type, including redirects (never followed)
+// and the service's own error envelope (never re-wrapped).
+func TestMirrorsServiceResponses(t *testing.T) {
+ tests := []struct {
+ name string
+ answer http.HandlerFunc
+ wantStatus int
+ wantBody string
+ wantCT string
+ wantLocation string
+ }{
+ {
+ name: "200 with a service body",
+ answer: func(w http.ResponseWriter, _ *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = io.WriteString(w, `{"alerts":[]}`)
+ },
+ wantStatus: http.StatusOK,
+ wantBody: `{"alerts":[]}`,
+ wantCT: "application/json",
+ },
+ {
+ name: "401 with the service's own envelope",
+ answer: func(w http.ResponseWriter, _ *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusUnauthorized)
+ _, _ = io.WriteString(w, `{"error":{"code":"ERR_UNAUTHORIZED"}}`)
+ },
+ wantStatus: http.StatusUnauthorized,
+ wantBody: `{"error":{"code":"ERR_UNAUTHORIZED"}}`,
+ wantCT: "application/json",
+ },
+ {
+ name: "500 with a plain-text body",
+ answer: func(w http.ResponseWriter, _ *http.Request) {
+ w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+ w.WriteHeader(http.StatusInternalServerError)
+ _, _ = io.WriteString(w, "database is down")
+ },
+ wantStatus: http.StatusInternalServerError,
+ wantBody: "database is down",
+ wantCT: "text/plain",
+ },
+ {
+ name: "302 is mirrored, not followed",
+ answer: func(w http.ResponseWriter, _ *http.Request) {
+ w.Header().Set("Location", "https://elsewhere.example/nowhere")
+ w.WriteHeader(http.StatusFound)
+ },
+ wantStatus: http.StatusFound,
+ wantLocation: "https://elsewhere.example/nowhere",
+ },
+ {
+ name: "204 with no body",
+ answer: func(w http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(http.StatusNoContent)
+ },
+ wantStatus: http.StatusNoContent,
+ },
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ _, upstreamURL := startHTTPService(t, test.answer)
+ srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil))
+
+ client := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error {
+ return http.ErrUseLastResponse
+ }}
+ resp, err := client.Post(srv.URL+"/api/alarm/action/getAlert", "application/json", nil)
+ if err != nil {
+ t.Fatalf("POST: %v", err)
+ }
+ defer func() { _ = resp.Body.Close() }()
+ body, _ := io.ReadAll(resp.Body)
+
+ if resp.StatusCode != test.wantStatus {
+ t.Fatalf("status = %d, want %d (body: %s)", resp.StatusCode, test.wantStatus, body)
+ }
+ if test.wantBody != "" && string(body) != test.wantBody {
+ t.Fatalf("body = %q, want %q", body, test.wantBody)
+ }
+ if test.wantCT != "" {
+ if got := resp.Header.Get("Content-Type"); !strings.Contains(got, test.wantCT) {
+ t.Fatalf("Content-Type = %q, want %q", got, test.wantCT)
+ }
+ }
+ if test.wantLocation != "" {
+ if got := resp.Header.Get("Location"); got != test.wantLocation {
+ t.Fatalf("Location = %q, want %q", got, test.wantLocation)
+ }
+ }
+ })
+ }
+}
+
+// TestSetCookieNotMirrored: a service must not be able to write cookies onto
+// Maestro's own origin, so Set-Cookie never rides back.
+func TestSetCookieNotMirrored(t *testing.T) {
+ _, upstreamURL := startHTTPService(t, func(w http.ResponseWriter, _ *http.Request) {
+ http.SetCookie(w, &http.Cookie{Name: "maestro_access", Value: "forged", Path: "/"})
+ _, _ = io.WriteString(w, "ok")
+ })
+ srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil))
+
+ resp, err := http.Post(srv.URL+"/api/alarm/auth/login", "application/json", nil)
+ if err != nil {
+ t.Fatalf("POST: %v", err)
+ }
+ defer func() { _ = resp.Body.Close() }()
+
+ if cookies := resp.Header.Values("Set-Cookie"); len(cookies) != 0 {
+ t.Fatalf("Set-Cookie = %v, want none — a service must not set cookies on the console origin", cookies)
+ }
+}
+
+// TestResolveErrors: unknown service 404, stopped service 409, Maestro's own
+// plumbing 500 (never a 502), no answer at all 502. Every one carries a message.
+func TestResolveErrors(t *testing.T) {
+ tests := []struct {
+ name string
+ resolve Resolver
+ wantCode string
+ wantBody string
+ }{
+ {
+ name: "unknown service",
+ resolve: resolverAnswer(nil, fmt.Errorf("%w: %q", serviceproxy.ErrNotFound, "alarm")),
+ wantCode: "ERR_NOT_FOUND",
+ wantBody: "unknown service",
+ },
+ {
+ name: "service not running",
+ resolve: resolverAnswer(&serviceproxy.Peer{BaseURL: "http://127.0.0.1:1", Running: false}, nil),
+ wantCode: "ERR_NOT_RUNNING",
+ wantBody: "is not running",
+ },
+ {
+ name: "docker unavailable",
+ resolve: resolverAnswer(nil, fmt.Errorf("%w: cannot connect", serviceproxy.ErrDocker)),
+ wantCode: "ERR_INTERNAL",
+ wantBody: "failed to resolve service",
+ },
+ {
+ name: "no network address",
+ resolve: resolverAnswer(nil, fmt.Errorf("%w: %q", serviceproxy.ErrNoNetwork, "alarm")),
+ wantCode: "ERR_INTERNAL",
+ wantBody: "failed to resolve service",
+ },
+ {
+ name: "upstream unreachable",
+ resolve: resolverAnswer(runningPeer("http://127.0.0.1:1"), nil),
+ wantCode: "ERR_UPSTREAM_UNREACHABLE",
+ wantBody: "failed to reach service",
+ },
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ srv := startProxy(t, test.resolve)
+ resp, err := http.Post(srv.URL+"/api/alarm/action/getAlert", "application/json", nil)
+ if err != nil {
+ t.Fatalf("POST: %v", err)
+ }
+ defer func() { _ = resp.Body.Close() }()
+ body, _ := io.ReadAll(resp.Body)
+
+ if !strings.Contains(string(body), `"code":"`+test.wantCode+`"`) {
+ t.Fatalf("body = %q, want code %s", body, test.wantCode)
+ }
+ if !strings.Contains(string(body), test.wantBody) {
+ t.Fatalf("body = %q, want it to contain %q", body, test.wantBody)
+ }
+ if !strings.Contains(string(body), `"message":"`) {
+ t.Fatalf("body = %q, want a human-readable message", body)
+ }
+ if got := resp.Header.Get("Content-Type"); !strings.Contains(got, "application/json") {
+ t.Fatalf("Content-Type = %q, want application/json", got)
+ }
+ })
+ }
+}
+
+// TestNamespaceAllowlist: only action, auth and ws are reachable. /config,
+// /_manifest and anything else are 404 and never reach the service — this surface
+// is not a tunnel.
+func TestNamespaceAllowlist(t *testing.T) {
+ tests := []string{
+ "/api/alarm/config",
+ "/api/alarm/_manifest",
+ "/api/alarm/manifest",
+ "/api/alarm/actions/getAlert", // plural is the console's route, not a service namespace
+ "/api/alarm",
+ "/api/alarm/",
+ "/api/alarm/action", // no action name
+ "/api/alarm/action/a/b", // an action is one segment
+ "/api/alarm/auth", // no auth path
+ "/api/typo", // no namespace at all
+ "/api/alarm/auth/%2e%2e/config",
+ }
+
+ for _, path := range tests {
+ t.Run(path, func(t *testing.T) {
+ rec, upstreamURL := startHTTPService(t, func(w http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(http.StatusOK)
+ })
+ srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil))
+
+ client := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error {
+ return http.ErrUseLastResponse
+ }}
+ resp, err := client.Get(srv.URL + path)
+ if err != nil {
+ t.Fatalf("GET %s: %v", path, err)
+ }
+ body, _ := io.ReadAll(resp.Body)
+ _ = resp.Body.Close()
+
+ if resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("GET %s = %d (%s), want 404", path, resp.StatusCode, body)
+ }
+ if !strings.Contains(string(body), `"code":"ERR_NOT_FOUND"`) {
+ t.Fatalf("GET %s body = %q, want Maestro's JSON envelope", path, body)
+ }
+ if calls, _, _, _, _, _ := rec.snapshot(); calls != 0 {
+ t.Fatalf("GET %s reached the service %d times, want 0", path, calls)
+ }
+ })
+ }
+}
+
+// TestDotSegmentsNeverReachTheService: a traversal attempt may be redirected by
+// the standard library's path cleaning, but it must never be relayed.
+func TestDotSegmentsNeverReachTheService(t *testing.T) {
+ rec, upstreamURL := startHTTPService(t, func(w http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(http.StatusOK)
+ })
+ srv := startProxy(t, resolverAnswer(runningPeer(upstreamURL), nil))
+
+ client := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error {
+ return http.ErrUseLastResponse
+ }}
+ resp, err := client.Get(srv.URL + "/api/../etc/passwd")
+ if err != nil {
+ t.Fatalf("GET: %v", err)
+ }
+ _ = resp.Body.Close()
+
+ // Either cleaned+404 by the mux or the mux's own redirect — never a relay.
+ if resp.StatusCode == http.StatusOK {
+ t.Fatalf("GET /api/../etc/passwd = 200, want no relay")
+ }
+ if calls, _, path, _, _, _ := rec.snapshot(); calls != 0 {
+ t.Fatalf("service was called (%s) for a traversal attempt", path)
+ }
+}
+
+// TestLegacyActionRouteStaysGated is the model in one test: the legacy console
+// route keeps its owner gate, while the data plane reaches the service with no
+// cookie at all.
+func TestLegacyActionRouteStaysGated(t *testing.T) {
+ rec, upstreamURL := startHTTPService(t, func(w http.ResponseWriter, _ *http.Request) {
+ _, _ = io.WriteString(w, `{"ok":true}`)
+ })
+
+ gate := func(next http.Handler) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ writeAPIError(w, http.StatusUnauthorized, "ERR_UNAUTHENTICATED", "a session is required")
+ })
+ }
+
+ mux := http.NewServeMux()
+ mux.Handle("/api/actions/{containerId}/{actionName}", gate(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ _, _ = io.WriteString(w, "legacy")
+ })))
+ RegisterWith(mux, resolverAnswer(runningPeer(upstreamURL), nil))
+ srv := httptest.NewServer(mux)
+ t.Cleanup(srv.Close)
+
+ // The legacy route still refuses an app.
+ resp, err := http.Post(srv.URL+"/api/actions/alarm/createAlert", "application/json", nil)
+ if err != nil {
+ t.Fatalf("POST legacy: %v", err)
+ }
+ _ = resp.Body.Close()
+ if resp.StatusCode != http.StatusUnauthorized {
+ t.Fatalf("legacy route = %d, want the owner gate's 401", resp.StatusCode)
+ }
+
+ // The data plane does not, and the service receives the action.
+ resp, err = http.Post(srv.URL+"/api/alarm/action/createAlert", "application/json", nil)
+ if err != nil {
+ t.Fatalf("POST data plane: %v", err)
+ }
+ body, _ := io.ReadAll(resp.Body)
+ _ = resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("data plane = %d (%s), want the service's 200", resp.StatusCode, body)
+ }
+ if calls, _, path, _, _, _ := rec.snapshot(); calls != 1 || path != "/action/createAlert" {
+ t.Fatalf("service calls = %d path = %q, want 1 call to /action/createAlert", calls, path)
+ }
+}
+
+// TestResolveIsShared: an action and a socket to the same service must ask the
+// same resolver, so the two edges can never disagree about a service's address.
+func TestResolveIsShared(t *testing.T) {
+ calls := make(chan string, 8)
+ resolve := func(_ context.Context, service string) (*serviceproxy.Peer, error) {
+ calls <- service
+ return nil, fmt.Errorf("%w: %q", serviceproxy.ErrNotFound, service)
+ }
+ mux := http.NewServeMux()
+ RegisterWith(mux, resolve)
+ srv := httptest.NewServer(mux)
+ t.Cleanup(srv.Close)
+
+ for _, path := range []string{"/api/alarm/action/x", "/api/alarm/auth/login"} {
+ resp, err := http.Get(srv.URL + path)
+ if err != nil {
+ t.Fatalf("GET %s: %v", path, err)
+ }
+ _ = resp.Body.Close()
+ }
+ // The socket edge resolves only on a real upgrade, so drive one.
+ if conn, _, err := websocket.DefaultDialer.Dial(
+ "ws"+strings.TrimPrefix(srv.URL, "http")+"/api/alarm/ws", nil); err == nil {
+ _ = conn.Close()
+ }
+
+ for i := 0; i < 3; i++ {
+ select {
+ case got := <-calls:
+ if got != "alarm" {
+ t.Fatalf("resolver saw %q, want alarm", got)
+ }
+ default:
+ t.Fatalf("resolver was not called for every edge")
+ }
+ }
+}
diff --git a/src/handlers/dataplane/ws.go b/src/handlers/dataplane/ws.go
new file mode 100644
index 0000000..0ed2358
--- /dev/null
+++ b/src/handlers/dataplane/ws.go
@@ -0,0 +1,226 @@
+// 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 (
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "time"
+
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/logger"
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/serviceproxy"
+ "github.com/gorilla/websocket"
+)
+
+const (
+ // dialTimeout bounds the service's handshake only. Once the socket is up
+ // there is deliberately NO read/write deadline on either hop: an idle alarm
+ // socket must survive indefinitely (a proxy deadline is how you get a socket
+ // that dies every 60 seconds).
+ dialTimeout = 10 * time.Second
+ // relayWriteWait bounds one relayed control frame (ping/pong/close) so a
+ // dead peer cannot block the frame pump forever.
+ relayWriteWait = 10 * time.Second
+ // maxCloseText keeps a relayed close reason inside a control frame (125 B).
+ maxCloseText = 123
+)
+
+var upgrader = websocket.Upgrader{
+ // Maestro is an instance-local pipe and the caller is an app, not a browser
+ // tab: same permissive origin policy as handlers/containers/ws.go. Access
+ // control is the service's token check, not the Origin header.
+ CheckOrigin: func(*http.Request) bool { return true },
+}
+
+// isUpgrade reports whether this request is a WebSocket handshake. It is the
+// single switch between the two relay modes.
+func isUpgrade(r *http.Request) bool { return websocket.IsWebSocketUpgrade(r) }
+
+// relayWebSocket proxies a WebSocket handshake and then pumps frames both ways.
+func relayWebSocket(resolve Resolver, service, upstreamPath string, w http.ResponseWriter, r *http.Request) {
+ peer, ok := resolvePeer(resolve, service, w, r)
+ if !ok {
+ return
+ }
+
+ // The query string is forwarded verbatim: it carries alertId, token,
+ // deviceId and accessToken, and re-encoding it would corrupt the token.
+ target, err := serviceproxy.Endpoint(peer.BaseURL, upstreamPath, r.URL.RawQuery)
+ if err != nil {
+ writeUnusableAddress(w, service, err)
+ return
+ }
+ target = wsScheme(target)
+
+ // Dial the service BEFORE touching the caller's connection: while nothing has
+ // been written to w we can still answer with a real HTTP status.
+ dialer := *websocket.DefaultDialer
+ dialer.HandshakeTimeout = dialTimeout
+ dialer.Subprotocols = websocket.Subprotocols(r)
+
+ upstream, resp, err := dialer.DialContext(r.Context(), target, forwardableRequestHeaders(r.Header))
+ if err != nil {
+ writeDialError(w, service, resp, err)
+ return
+ }
+ defer func() { _ = upstream.Close() }()
+
+ // Echo the subprotocol the service accepted, if any: a caller that requested
+ // one fails the connection when the answer does not match.
+ var responseHeader http.Header
+ if protocol := upstream.Subprotocol(); protocol != "" {
+ responseHeader = http.Header{"Sec-Websocket-Protocol": {protocol}}
+ }
+
+ client, err := upgrader.Upgrade(w, r, responseHeader)
+ if err != nil {
+ // Upgrade already wrote a plaintext error response.
+ logger.Debug("Client WebSocket upgrade failed for %q: %v", service, err)
+ return
+ }
+ defer func() { _ = client.Close() }()
+
+ // Control frames are RELAYED, not answered. If this proxy answered pings
+ // itself, the service's own 20s ping / 60s pong / 10s write accounting would
+ // be satisfied by Maestro even with the phone switched off — liveness
+ // detection would become a lie. Relay keeps it end-to-end.
+ client.SetPingHandler(func(appData string) error {
+ return relayControl(upstream, websocket.PingMessage, appData)
+ })
+ client.SetPongHandler(func(appData string) error {
+ return relayControl(upstream, websocket.PongMessage, appData)
+ })
+ upstream.SetPingHandler(func(appData string) error {
+ return relayControl(client, websocket.PingMessage, appData)
+ })
+ upstream.SetPongHandler(func(appData string) error {
+ return relayControl(client, websocket.PongMessage, appData)
+ })
+
+ // Two pumps, no barriers: whichever hop ends first ends the session.
+ type pumpEnd struct {
+ err error
+ fromUpstream bool
+ }
+ ends := make(chan pumpEnd, 2)
+ go func() { ends <- pumpEnd{err: pump(upstream, client), fromUpstream: true} }()
+ go func() { ends <- pumpEnd{err: pump(client, upstream), fromUpstream: false} }()
+
+ first := <-ends
+ propagateClose(first.err, first.fromUpstream, client, upstream)
+
+ // Closing both sockets unblocks the other pump; waiting for it keeps every
+ // goroutine inside the handler's lifetime.
+ _ = client.Close()
+ _ = upstream.Close()
+ <-ends
+}
+
+// wsScheme maps the plaintext service base URL onto a WebSocket URL.
+func wsScheme(target string) string {
+ u, err := url.Parse(target)
+ if err != nil {
+ return target
+ }
+ switch u.Scheme {
+ case "http":
+ u.Scheme = "ws"
+ case "https":
+ u.Scheme = "wss"
+ }
+ return u.String()
+}
+
+// writeDialError reports a failed handshake. A service that answered and refused
+// (401/403/404) is mirrored with its own status and body: the service is the gate
+// on this surface, so a rejected token must look like a rejection, not like a
+// broken proxy. No answer at all is a 502.
+func writeDialError(w http.ResponseWriter, service string, resp *http.Response, err error) {
+ if resp != nil {
+ defer func() { _ = resp.Body.Close() }()
+ if resp.StatusCode >= 400 && resp.StatusCode < 500 {
+ if contentType := resp.Header.Get("Content-Type"); contentType != "" {
+ w.Header().Set("Content-Type", contentType)
+ }
+ w.WriteHeader(resp.StatusCode)
+ _, _ = io.Copy(w, resp.Body)
+ return
+ }
+ }
+ logger.Debug("Service WebSocket dial failed for %q: %v", service, err)
+ writeAPIError(w, http.StatusBadGateway, "ERR_UPSTREAM_UNREACHABLE",
+ fmt.Sprintf("failed to reach the socket of service %q: %v", service, err))
+}
+
+// relayControl forwards one control frame to the peer hop.
+func relayControl(dst *websocket.Conn, messageType int, appData string) error {
+ return dst.WriteControl(messageType, []byte(appData), time.Now().Add(relayWriteWait))
+}
+
+// pump copies data messages from src to dst, one frame at a time. Control frames
+// never reach this loop (the per-connection handlers relay them), so the only
+// outcomes are the next message and the terminal error.
+func pump(src, dst *websocket.Conn) error {
+ for {
+ messageType, reader, err := src.NextReader()
+ if err != nil {
+ return err
+ }
+ writer, err := dst.NextWriter(messageType)
+ if err != nil {
+ return err
+ }
+ if _, err := io.Copy(writer, reader); err != nil {
+ _ = writer.Close()
+ return err
+ }
+ if err := writer.Close(); err != nil {
+ return err
+ }
+ }
+}
+
+// propagateClose tells the surviving hop why the session ended. A received close
+// frame is forwarded with its original code and reason, so the client can tell
+// "the admin disarmed the alert" (the service closed cleanly) from "the network
+// died" (abnormal closure). Nothing is invented for a clean code.
+func propagateClose(cause error, fromUpstream bool, client, upstream *websocket.Conn) {
+ dst, ended := upstream, "client"
+ if fromUpstream {
+ dst, ended = client, "upstream"
+ }
+ deadline := time.Now().Add(relayWriteWait)
+
+ var closeErr *websocket.CloseError
+ if errors.As(cause, &closeErr) &&
+ closeErr.Code != websocket.CloseNoStatusReceived &&
+ closeErr.Code != websocket.CloseAbnormalClosure {
+ text := closeErr.Text
+ if len(text) > maxCloseText {
+ text = text[:maxCloseText]
+ }
+ _ = dst.WriteControl(websocket.CloseMessage,
+ websocket.FormatCloseMessage(closeErr.Code, text), deadline)
+ return
+ }
+
+ _ = dst.WriteControl(websocket.CloseMessage,
+ websocket.FormatCloseMessage(websocket.CloseTryAgainLater, "the "+ended+" connection ended"),
+ deadline)
+}
diff --git a/src/handlers/dataplane/ws_test.go b/src/handlers/dataplane/ws_test.go
new file mode 100644
index 0000000..5493190
--- /dev/null
+++ b/src/handlers/dataplane/ws_test.go
@@ -0,0 +1,663 @@
+// 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)
+ }
+}
diff --git a/src/lib/serviceproxy/serviceproxy.go b/src/lib/serviceproxy/serviceproxy.go
new file mode 100644
index 0000000..360228d
--- /dev/null
+++ b/src/lib/serviceproxy/serviceproxy.go
@@ -0,0 +1,167 @@
+// 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 serviceproxy answers the only question a Maestro proxy needs: where is
+// this service on the shared Docker network?
+//
+// It is the single implementation of that question — the service data plane
+// (handlers/dataplane: /api/{service}/action/... , /auth/... and /ws...) and the
+// legacy REST action proxy (/api/actions/{containerId}/{actionName}) both
+// resolve through it, so a service cannot be reachable on one edge and invisible
+// on the other.
+//
+// Resolution: Docker client → docker.ResolveContainer (which matches a container
+// by id, name, image or the org.dcntrlized.service label) → the container's IP on
+// dcntrlized_net → the port declared by the org.dcntrlized.port label, falling
+// back to the first mapped guest port. Rev 9: the service edge is plaintext HTTP
+// on the single :8080 listener, so the base URL is always http://.
+package serviceproxy
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net/url"
+ "strconv"
+ "strings"
+
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/authmgmt"
+ "git.dcntrlized.org/Dcntrlized/Maestro/lib/docker"
+)
+
+// Sentinel errors callers translate into HTTP statuses. Every one of them is
+// wrapped with the offending reference so the message stays human-readable.
+var (
+ // ErrNotFound: the reference matches no container (unknown/uninstalled service).
+ ErrNotFound = errors.New("service not found")
+ // ErrNotRunning: the container exists but is stopped.
+ ErrNotRunning = errors.New("service is not running")
+ // ErrNoNetwork: the container has no usable dcntrlized_net address.
+ ErrNoNetwork = errors.New("service has no network address")
+ // ErrNoPort: the container declares no service port.
+ ErrNoPort = errors.New("service has no configured port")
+ // ErrDockerClient: a Docker client could not be constructed (bad DOCKER_HOST
+ // / certificate config).
+ ErrDockerClient = errors.New("docker client unavailable")
+ // ErrDocker: the Docker daemon did not answer.
+ ErrDocker = errors.New("docker unavailable")
+)
+
+// Peer is a resolved, dialable service.
+type Peer struct {
+ // Service is the container name without Docker's leading slash.
+ Service string
+ // IP is the container's address on dcntrlized_net.
+ IP string
+ // Port is the service's plaintext HTTP port.
+ Port string
+ // Running reports container state at resolution time.
+ Running bool
+ // BaseURL is scheme://ip:port, e.g. http://172.18.0.7:8080.
+ BaseURL string
+}
+
+// Resolve locates a service by id, name, image or org.dcntrlized.service label.
+func Resolve(ctx context.Context, ref string) (*Peer, error) {
+ ref = strings.TrimSpace(ref)
+ if ref == "" {
+ return nil, fmt.Errorf("%w: empty reference", ErrNotFound)
+ }
+
+ dockerClient, err := docker.NewDockerClient()
+ if err != nil {
+ return nil, fmt.Errorf("%w: %v", ErrDockerClient, err)
+ }
+ defer func() { _ = dockerClient.Close() }()
+
+ inspectData, err := docker.ResolveContainer(ctx, dockerClient, ref)
+ if err != nil {
+ // ResolveContainer already searched by id, name, service label and image,
+ // so a miss means "unknown reference" only if the daemon is answering.
+ // Pinging distinguishes an unknown service (404) from a dead daemon (500)
+ // instead of reporting every Docker problem as "not found".
+ if _, pingErr := dockerClient.Client.Ping(ctx); pingErr != nil {
+ return nil, fmt.Errorf("%w: %v", ErrDocker, pingErr)
+ }
+ return nil, fmt.Errorf("%w: %q", ErrNotFound, ref)
+ }
+
+ running := inspectData.State != nil && inspectData.State.Running
+ service := strings.TrimPrefix(inspectData.Name, "/")
+
+ port := ""
+ if inspectData.Config != nil {
+ port = strings.TrimSpace(inspectData.Config.Labels["org.dcntrlized.port"])
+ }
+
+ // ExtractNetworkInfo's bool is "this container uses a standard network", NOT
+ // "we skipped it": it returns false for host/none/container networking. Read
+ // it that way — inverting it here made every normal container look
+ // unreachable (found by running the data plane against the real dev fleet,
+ // where every service answered 500 "uses host/none/container networking").
+ network, standard, err := docker.ExtractNetworkInfo(ctx, dockerClient, inspectData)
+ if err != nil {
+ return nil, fmt.Errorf("failed to extract network info for %q: %w", ref, err)
+ }
+ if !standard {
+ return nil, fmt.Errorf("%w: %q uses host/none/container networking", ErrNoNetwork, ref)
+ }
+ ip := network.IPAddress.Address
+ if ip == "" {
+ return nil, fmt.Errorf("%w: %q has no dcntrlized_net address", ErrNoNetwork, ref)
+ }
+ if port == "" && len(network.Ports) > 0 {
+ port = strconv.Itoa(network.Ports[0].Guest)
+ }
+ if port == "" {
+ return nil, fmt.Errorf("%w: %q", ErrNoPort, ref)
+ }
+
+ return &Peer{
+ Service: service,
+ IP: ip,
+ Port: port,
+ Running: running,
+ BaseURL: fmt.Sprintf("%s://%s:%s", authmgmt.Scheme(), ip, port),
+ }, nil
+}
+
+// Target returns the plaintext base URL of the referenced service, e.g.
+// http://172.18.0.7:8080. It is the convenience form for callers that only need
+// to build one URL.
+func Target(ctx context.Context, ref string) (string, error) {
+ peer, err := Resolve(ctx, ref)
+ if err != nil {
+ return "", err
+ }
+ return peer.BaseURL, nil
+}
+
+// Endpoint joins a resolved base URL with an absolute service path and a raw
+// query string (forwarded verbatim, never parsed and re-encoded).
+//
+// This is what handlers/dataplane uses for every namespace it relays:
+//
+// base, _ := serviceproxy.Target(ctx, service)
+// endpoint, _ := serviceproxy.Endpoint(base, "/action/"+action, r.URL.RawQuery)
+func Endpoint(baseURL, path, rawQuery string) (string, error) {
+ u, err := url.Parse(baseURL)
+ if err != nil {
+ return "", fmt.Errorf("bad service base URL %q: %w", baseURL, err)
+ }
+ u.Path = path
+ u.RawQuery = rawQuery
+ return u.String(), nil
+}