The event listener authenticates with the message bus

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

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=

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)

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)
+	}
+}

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()
+}

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)
+		}
+	}
+}

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)
+	}
+}

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)

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 {