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= 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/event_listen.go b/route/event_listen.go index 9274454..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" @@ -27,27 +28,28 @@ 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) 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 +73,35 @@ 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") } + +// 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 + } + + // 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) + } + } +} diff --git a/route/event_listen_test.go b/route/event_listen_test.go new file mode 100644 index 0000000..b436457 --- /dev/null +++ b/route/event_listen_test.go @@ -0,0 +1,121 @@ +package route + +import ( + "net/http" + "net/http/httptest" + "os" + "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, 4) + 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() + defer close(proceed) // runs first: a failed test does not leave a handshake blocking 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{ + 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) + } +} 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 {