From 8e05f1d5b036c4df8c8392aca8d14d8b9ecbbc13 Mon Sep 17 00:00:00 2001 From: Gary Date: Tue, 22 Sep 2026 10:24:27 +0200 Subject: [PATCH 1/6] build(deps): bump CasaOS-Common to v0.4.27 Picks up InternalAuthorization, which the message bus subscription now needs, and the start-script runner that keeps going after a failure. Nothing else in this service uses the changed APIs. --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 3493b7f..c15a872 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/ReCasaOS/CasaOS-UserService go 1.26.8 require ( - github.com/ReCasaOS/CasaOS-Common v0.4.26 + github.com/ReCasaOS/CasaOS-Common v0.4.27 github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf github.com/deepmap/oapi-codegen v1.12.4 github.com/getkin/kin-openapi v0.149.0 diff --git a/go.sum b/go.sum index 5ada4a8..a46040c 100644 --- a/go.sum +++ b/go.sum @@ -1,6 +1,6 @@ github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk= -github.com/ReCasaOS/CasaOS-Common v0.4.26 h1:aefnl58xQ4z0ga0+vBnQHmkG1/3fnd1C1W/EBFeAgvc= -github.com/ReCasaOS/CasaOS-Common v0.4.26/go.mod h1:HjYCcRAVDa0c/IncJDof1sddCpze2esCTZt941Sk7LI= +github.com/ReCasaOS/CasaOS-Common v0.4.27 h1:7V3rAZj77wII/tqJeu9eqxEvAYln0kpaeK1GTuzTXSk= +github.com/ReCasaOS/CasaOS-Common v0.4.27/go.mod h1:HjYCcRAVDa0c/IncJDof1sddCpze2esCTZt941Sk7LI= github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ= github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk= github.com/benbjohnson/clock v1.3.1 h1:Heo0FGXzOxUHquZbraxt+tT7UXVDhesUQH5ISbsOkCQ= From 12d5bb025e14892b34eba12604eb18c246d58b70 Mon Sep 17 00:00:00 2001 From: Gary Date: Tue, 22 Sep 2026 10:25:53 +0200 Subject: [PATCH 2/6] feat(event): send the internal secret when subscribing to the message bus The message bus now refuses a subscription without the box's internal secret. The websocket handshake is built with websocket.NewConfig and carries Authorization: Internal when the gateway has written one. The secret is read at each dial, since a restarted gateway rewrites it; with no secret file (an older gateway) the handshake goes without it, as before. --- route/event_listen.go | 19 +++++++++++++- route/event_listen_test.go | 51 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 1 deletion(-) create mode 100644 route/event_listen_test.go diff --git a/route/event_listen.go b/route/event_listen.go index 9274454..082bb40 100644 --- a/route/event_listen.go +++ b/route/event_listen.go @@ -27,7 +27,7 @@ func EventListen() { } wsURL := fmt.Sprintf("ws://%s/event/%s", strings.ReplaceAll(messageBusUrl, "http://", ""), "local-storage") - ws, err := websocket.Dial(wsURL, "", "http://localhost") + ws, err := dialMessageBus(wsURL, config.CommonInfo.RuntimePath) if err != nil { logger.Error("connect websocket err"+strconv.Itoa(i), zap.Any("error", err)) time.Sleep(time.Second * 1) @@ -74,3 +74,20 @@ func EventListen() { } logger.Error("error when try to connect to message bus") } + +// dialMessageBus opens the subscription with this boot's internal secret, which +// the bus now requires on its subscription routes. The secret is read at each +// dial because a restarted gateway rewrites it; with no secret file (an older +// gateway) the handshake goes without it, as before. +func dialMessageBus(wsURL, runtimePath string) (*websocket.Conn, error) { + cfg, err := websocket.NewConfig(wsURL, "http://localhost") + if err != nil { + return nil, err + } + + if authorization := external.InternalAuthorization(runtimePath); authorization != "" { + cfg.Header.Set("Authorization", authorization) + } + + return websocket.DialConfig(cfg) +} diff --git a/route/event_listen_test.go b/route/event_listen_test.go new file mode 100644 index 0000000..b401936 --- /dev/null +++ b/route/event_listen_test.go @@ -0,0 +1,51 @@ +package route + +import ( + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/ReCasaOS/CasaOS-Common/external" + "golang.org/x/net/websocket" +) + +func TestDialMessageBusSendsInternalSecretWhenPresent(t *testing.T) { + authorizations := make(chan string, 2) + server := httptest.NewServer(websocket.Server{ + Handshake: func(_ *websocket.Config, r *http.Request) error { + authorizations <- r.Header.Get("Authorization") + return nil + }, + Handler: func(*websocket.Conn) {}, + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/event/local-storage" + runtimePath := t.TempDir() + + dial := func() string { + t.Helper() + ws, err := dialMessageBus(wsURL, runtimePath) + if err != nil { + t.Fatalf("dial: %v", err) + } + ws.Close() + return <-authorizations + } + + // Older gateway: no secret file, no header. + if got := dial(); got != "" { + t.Fatalf("Authorization without a secret file = %q, want none", got) + } + + if err := os.WriteFile(filepath.Join(runtimePath, external.InternalSecretFilename), []byte("s3cret\n"), 0o600); err != nil { + t.Fatal(err) + } + + if got, want := dial(), "Internal s3cret"; got != want { + t.Fatalf("Authorization = %q, want %q", got, want) + } +} From 2699d30b81ab37bc6cff40fffb9e1ece39b843b9 Mon Sep 17 00:00:00 2001 From: Gary Date: Tue, 22 Sep 2026 10:28:27 +0200 Subject: [PATCH 3/6] fix(event): dial the message bus again when the subscription drops When the bus closed the websocket, the read loop kept going with an empty message and dereferenced a nil event UUID, which took the whole service down; the listener only came back through the process restart. A read error now closes the connection and goes back to the dial loop, which reads the internal secret again, so a gateway restart that rewrote it is picked up. A successful subscription resets the retry budget, as the restart used to. Events are read as whole websocket messages instead of 1024-byte chunks, so a larger event is no longer split into two invalid halves, and an event that does not parse or has no UUID is logged and skipped. --- route/event_listen.go | 23 +++++++------ route/event_listen_test.go | 69 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 82 insertions(+), 10 deletions(-) diff --git a/route/event_listen.go b/route/event_listen.go index 082bb40..ca7828c 100644 --- a/route/event_listen.go +++ b/route/event_listen.go @@ -33,21 +33,22 @@ func EventListen() { time.Sleep(time.Second * 1) continue } - defer ws.Close() - logger.Info("subscribed to", zap.Any("url", wsURL)) + // A working subscription starts the retry budget over, as the process + // restart that used to follow a lost subscription did. + i = 0 for { - - msg := make([]byte, 1024) - n, err := ws.Read(msg) - if err != nil { - logger.Error("err", zap.Any("err", err.Error())) + var msg []byte + if err := websocket.Message.Receive(ws, &msg); err != nil { + // The bus went away: dial again, with the secret of now. + logger.Error("message bus subscription lost", zap.Error(err)) + break } var event message_bus.Event - - if err := json.Unmarshal(msg[:n], &event); err != nil { - logger.Error("err", zap.Any("err", err.Error())) + if err := json.Unmarshal(msg, &event); err != nil || event.Uuid == nil { + logger.Error("invalid event from message bus", zap.Any("err", err), zap.ByteString("event", msg)) + continue } propertiesStr, err := json.Marshal(event.Properties) if err != nil { @@ -71,6 +72,8 @@ func EventListen() { // } // logger.Info("info", zap.Any("写入信息", string(output))) } + ws.Close() + time.Sleep(time.Second * 1) } logger.Error("error when try to connect to message bus") } diff --git a/route/event_listen_test.go b/route/event_listen_test.go index b401936..8db4bca 100644 --- a/route/event_listen_test.go +++ b/route/event_listen_test.go @@ -7,11 +7,80 @@ import ( "path/filepath" "strings" "testing" + "time" "github.com/ReCasaOS/CasaOS-Common/external" + "github.com/ReCasaOS/CasaOS-Common/utils/logger" + "github.com/ReCasaOS/CasaOS-UserService/pkg/config" "golang.org/x/net/websocket" ) +// A dropped subscription is dialled again with the secret of the moment, and +// the listener gives up once the bus address is gone. +func TestEventListenReconnectsWithCurrentSecret(t *testing.T) { + logger.LogInitConsoleOnly() + + authorizations := make(chan string) + proceed := make(chan struct{}) + server := httptest.NewServer(websocket.Server{ + Handshake: func(_ *websocket.Config, r *http.Request) error { + authorizations <- r.Header.Get("Authorization") + <-proceed + return nil + }, + Handler: func(*websocket.Conn) {}, // drops the subscription at once + }) + defer server.Close() + + runtimePath := t.TempDir() + writeFile := func(name, content string) { + t.Helper() + if err := os.WriteFile(filepath.Join(runtimePath, name), []byte(content), 0o600); err != nil { + t.Fatal(err) + } + } + writeFile(external.MessageBusAddressFilename, server.URL) + writeFile(external.InternalSecretFilename, "first\n") + + defer func(old string) { config.CommonInfo.RuntimePath = old }(config.CommonInfo.RuntimePath) + config.CommonInfo.RuntimePath = runtimePath + + done := make(chan struct{}) + go func() { EventListen(); close(done) }() + handshake := func() string { + t.Helper() + select { + case got := <-authorizations: + return got + case <-done: + t.Fatal("EventListen returned before dialling") + case <-time.After(10 * time.Second): + t.Fatal("no handshake") + } + return "" + } + + if got := handshake(); got != "Internal first" { + t.Fatalf("first handshake Authorization = %q", got) + } + writeFile(external.InternalSecretFilename, "second\n") // the gateway restarted + proceed <- struct{}{} + + if got := handshake(); got != "Internal second" { + t.Fatalf("second handshake Authorization = %q", got) + } + if err := os.Remove(filepath.Join(runtimePath, external.MessageBusAddressFilename)); err != nil { + t.Fatal(err) + } + proceed <- struct{}{} + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("EventListen did not return once the bus address was gone") + } +} + func TestDialMessageBusSendsInternalSecretWhenPresent(t *testing.T) { authorizations := make(chan string, 2) server := httptest.NewServer(websocket.Server{ From 3d2ddf2eb36f1e6b28388835ea8d8591a6d0b0d3 Mon Sep 17 00:00:00 2001 From: Gary Date: Tue, 22 Sep 2026 10:30:24 +0200 Subject: [PATCH 4/6] test(event): keep a failed reconnect test from hanging on server close The test's handshake waits for the go-ahead; closing that channel first on the way out lets a failed run shut the server down instead of blocking until the test timeout. --- route/event_listen_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/route/event_listen_test.go b/route/event_listen_test.go index 8db4bca..b436457 100644 --- a/route/event_listen_test.go +++ b/route/event_listen_test.go @@ -20,7 +20,7 @@ import ( func TestEventListenReconnectsWithCurrentSecret(t *testing.T) { logger.LogInitConsoleOnly() - authorizations := make(chan string) + authorizations := make(chan string, 4) proceed := make(chan struct{}) server := httptest.NewServer(websocket.Server{ Handshake: func(_ *websocket.Config, r *http.Request) error { @@ -31,6 +31,7 @@ func TestEventListenReconnectsWithCurrentSecret(t *testing.T) { Handler: func(*websocket.Conn) {}, // drops the subscription at once }) defer server.Close() + defer close(proceed) // runs first: a failed test does not leave a handshake blocking Close runtimePath := t.TempDir() writeFile := func(name, content string) { From 4f2a15dd306bb20d82a171bd25fa3a037a95f035 Mon Sep 17 00:00:00 2001 From: Gary Date: Tue, 22 Sep 2026 11:19:35 +0200 Subject: [PATCH 5/6] fix(route): keep tokens out of the access log Websocket routes take the user's token from the query string, since a browser cannot set a header on a websocket, and echo's default access log writes the request URI, query included: users' tokens went to the journal. The access log now writes the path, as the message bus does. --- route/access_log.go | 13 +++++++++++++ route/access_log_test.go | 12 ++++++++++++ route/v1.go | 2 +- route/v2.go | 2 +- 4 files changed, 27 insertions(+), 2 deletions(-) create mode 100644 route/access_log.go create mode 100644 route/access_log_test.go diff --git a/route/access_log.go b/route/access_log.go new file mode 100644 index 0000000..417e136 --- /dev/null +++ b/route/access_log.go @@ -0,0 +1,13 @@ +package route + +import ( + "strings" + + echo_middleware "github.com/labstack/echo/v4/middleware" +) + +// accessLogFormat is echo's default access log with the path in place of the +// request URI: websocket routes take the user's token from the query string +// (a browser cannot set a header on a websocket), and it must not reach the +// journal. +var accessLogFormat = strings.Replace(echo_middleware.DefaultLoggerConfig.Format, `"uri":"${uri}"`, `"path":"${path}"`, 1) diff --git a/route/access_log_test.go b/route/access_log_test.go new file mode 100644 index 0000000..f4a52c7 --- /dev/null +++ b/route/access_log_test.go @@ -0,0 +1,12 @@ +package route + +import ( + "strings" + "testing" +) + +func TestTheAccessLogLeavesTheQueryStringOut(t *testing.T) { + if strings.Contains(accessLogFormat, "${uri}") || !strings.Contains(accessLogFormat, "${path}") { + t.Fatalf("the access log must log the path, not the URI with its query: %s", accessLogFormat) + } +} diff --git a/route/v1.go b/route/v1.go index cabc29f..cc800af 100644 --- a/route/v1.go +++ b/route/v1.go @@ -29,7 +29,7 @@ func InitRouter() http.Handler { e.Use(echo_middleware.Gzip()) - e.Use(echo_middleware.Logger()) + e.Use(echo_middleware.LoggerWithConfig(echo_middleware.LoggerConfig{Format: accessLogFormat})) e.POST("/v1/users/register", v1.PostUserRegister) e.POST("/v1/users/login", v1.PostUserLogin) diff --git a/route/v2.go b/route/v2.go index 5701cff..69cc52a 100644 --- a/route/v2.go +++ b/route/v2.go @@ -61,7 +61,7 @@ func InitV2Router() http.Handler { e.Use(echo_middleware.Gzip()) - e.Use(echo_middleware.Logger()) + e.Use(echo_middleware.LoggerWithConfig(echo_middleware.LoggerConfig{Format: accessLogFormat})) e.Use(echojwt.WithConfig(echojwt.Config{ Skipper: func(c echo.Context) bool { From ab563323d4b418333256104f47ed2146043dc2bb Mon Sep 17 00:00:00 2001 From: Gary Date: Tue, 22 Sep 2026 11:19:35 +0200 Subject: [PATCH 6/6] fix(event): send the secret to a loopback bus only The listener sent the boot's secret to whatever host message-bus.url names. Common's clients only ever send it to loopback addresses; so does this one. --- route/event_listen.go | 13 ++++++++++++- route/event_listen_loopback_test.go | 20 ++++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) create mode 100644 route/event_listen_loopback_test.go diff --git a/route/event_listen.go b/route/event_listen.go index ca7828c..13eaaa8 100644 --- a/route/event_listen.go +++ b/route/event_listen.go @@ -3,6 +3,7 @@ package route import ( "encoding/json" "fmt" + "net" "strconv" "strings" "time" @@ -88,9 +89,19 @@ func dialMessageBus(wsURL, runtimePath string) (*websocket.Conn, error) { return nil, err } - if authorization := external.InternalAuthorization(runtimePath); authorization != "" { + // the secret goes to this box's own bus only, as Common's clients send it + if authorization := external.InternalAuthorization(runtimePath); authorization != "" && isLoopbackHost(cfg.Location.Hostname()) { cfg.Header.Set("Authorization", authorization) } return websocket.DialConfig(cfg) } + +func isLoopbackHost(host string) bool { + if host == "localhost" { + return true + } + ip := net.ParseIP(host) + + return ip != nil && ip.IsLoopback() +} diff --git a/route/event_listen_loopback_test.go b/route/event_listen_loopback_test.go new file mode 100644 index 0000000..819019b --- /dev/null +++ b/route/event_listen_loopback_test.go @@ -0,0 +1,20 @@ +package route + +import "testing" + +// The listener sends the boot's secret with its handshake, so the bus it dials +// must be this box's own: message-bus.url could name any host. +func TestTheSecretGoesToALoopbackBusOnly(t *testing.T) { + for host, want := range map[string]bool{ + "127.0.0.1": true, + "::1": true, + "localhost": true, + "192.168.1.20": false, + "example.com": false, + "": false, + } { + if got := isLoopbackHost(host); got != want { + t.Errorf("isLoopbackHost(%q) = %v, want %v", host, got, want) + } + } +}