Start scripts that run, log and time out; the message bus socket leaves /tmp

external/internal_auth.go

@@ -93,14 +93,20 @@ func isLoopbackIP(ip string) bool {
 // request to a loopback address. The file is read at each request, so a secret
 // rewritten by a restarted gateway is picked up without anybody restarting.
 func useInternalSecret(runtimePath string) {
-	http2.SetInternalAuthorization(func() string {
-		secret, err := readInternalSecret(runtimePath)
-		if err != nil {
-			return ""
-		}
+	http2.SetInternalAuthorization(func() string { return InternalAuthorization(runtimePath) })
+}
+
+// InternalAuthorization returns the Authorization value that makes a request
+// internal, "Internal <secret>", or "" when the secret cannot be read. It is for
+// what a request editor does not fit, such as the Config.Header of a websocket
+// handshake. Send it to loopback addresses only.
+func InternalAuthorization(runtimePath string) string {
+	secret, err := readInternalSecret(runtimePath)
+	if err != nil {
+		return ""
+	}
 
-		return internalScheme + secret
-	})
+	return internalScheme + secret
 }
 
 // InternalRequestEditor is for the API clients oapi-codegen generates, which do
@@ -114,8 +120,8 @@ func InternalRequestEditor(runtimePath string) func(ctx context.Context, req *ht
 		if req.Header.Get("Authorization") != "" || !isLoopbackHost(req.URL.Hostname()) {
 			return nil
 		}
-		if secret, err := readInternalSecret(runtimePath); err == nil {
-			req.Header.Set("Authorization", internalScheme+secret)
+		if authorization := InternalAuthorization(runtimePath); authorization != "" {
+			req.Header.Set("Authorization", authorization)
 		}
 
 		return nil

external/internal_auth_test.go

@@ -141,3 +141,22 @@ func TestTheRequestEditorSendsTheSecretToLoopbackOnly(t *testing.T) {
 		t.Fatalf("with no secret file the request goes without one, got %q (%v)", missing.Header.Get("Authorization"), err)
 	}
 }
+
+func TestInternalAuthorizationIsTheHeaderValueOrNothing(t *testing.T) {
+	dir, secret := writeSecretFor(t)
+	if got := InternalAuthorization(dir); got != internalScheme+secret {
+		t.Fatalf("got %q, want %q", got, internalScheme+secret)
+	}
+
+	if got := InternalAuthorization(t.TempDir()); got != "" {
+		t.Fatalf("with no secret file, got %q", got)
+	}
+
+	empty := t.TempDir()
+	if err := os.WriteFile(filepath.Join(empty, InternalSecretFilename), []byte("\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if got := InternalAuthorization(empty); got != "" {
+		t.Fatalf("with an empty secret file, got %q", got)
+	}
+}

external/message_bus.go

@@ -16,8 +16,19 @@ import (
 const (
 	MessageBusAddressFilename = "message-bus.url"
 	APIMessageBus             = "/v2/message_bus"
+
+	// MessageBusSocketFilename is the unix socket, in the runtime path, where
+	// the message bus takes events.
+	MessageBusSocketFilename = "message-bus.sock"
 )
 
+// MessageBusSocketPath is where the message bus listens for events over a unix
+// socket: in the runtime path, which only root can write. It used to be
+// /tmp/message-bus.sock, a name any local user could take first.
+func MessageBusSocketPath(runtimePath string) string {
+	return filepath.Join(runtimePath, MessageBusSocketFilename)
+}
+
 type EventType struct {
 	Name             string
 	SourceID         string
@@ -70,13 +81,17 @@ func GetMessageBusAddress(runtimePath string) (string, error) {
 	return strings.TrimRight(address, "/") + APIMessageBus, nil
 }
 
-func PublishEventInSocket(ctx context.Context, SourceID string, Name string, properties map[string]string) (*http.Response, error) {
-	socketPath := "/tmp/message-bus.sock"
+// PublishEventInSocket posts an event to the message bus through its unix
+// socket in runtimePath. The response comes back with its body closed.
+func PublishEventInSocket(ctx context.Context, runtimePath, sourceID, name string, properties map[string]string) (*http.Response, error) {
+	socketPath := MessageBusSocketPath(runtimePath)
 	httpClient := &http.Client{
 		Transport: &http.Transport{
 			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
-				return net.Dial("unix", socketPath)
+				return (&net.Dialer{}).DialContext(ctx, "unix", socketPath)
 			},
+			// A transport per call: an idle connection kept alive would be leaked.
+			DisableKeepAlives: true,
 		},
 	}
 
@@ -85,8 +100,8 @@ func PublishEventInSocket(ctx context.Context, SourceID string, Name string, pro
 		return nil, err
 	}
 
-	req, err := http.NewRequest("POST",
-		fmt.Sprintf("http://unix/v2/message_bus/event/%s/%s", SourceID, Name),
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost,
+		fmt.Sprintf("http://unix/v2/message_bus/event/%s/%s", sourceID, name),
 		bytes.NewBuffer(body),
 	)
 	if err != nil {

external/message_bus_test.go

@@ -0,0 +1,65 @@
+package external
+
+import (
+	"context"
+	"encoding/json"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestTheMessageBusSocketIsInTheRuntimePath(t *testing.T) {
+	if got, want := MessageBusSocketPath("/var/run/casaos"), filepath.Join("/var/run/casaos", "message-bus.sock"); got != want {
+		t.Fatalf("got %q, want %q", got, want)
+	}
+}
+
+func TestPublishEventInSocketPostsToTheRuntimeSocket(t *testing.T) {
+	// Short: a unix socket path is limited to about a hundred bytes.
+	runtimePath, err := os.MkdirTemp("", "mb")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(runtimePath) })
+
+	listener, err := net.Listen("unix", MessageBusSocketPath(runtimePath))
+	if err != nil {
+		t.Skipf("no unix socket here: %v", err)
+	}
+
+	type request struct {
+		method, path, contentType string
+		properties                map[string]string
+	}
+	received := make(chan request, 1)
+	server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		got := request{method: r.Method, path: r.URL.Path, contentType: r.Header.Get("Content-Type")}
+		json.NewDecoder(r.Body).Decode(&got.properties)
+		received <- got
+		w.WriteHeader(http.StatusAccepted)
+	})}
+	go server.Serve(listener)
+	t.Cleanup(func() { server.Close() })
+
+	response, err := PublishEventInSocket(context.Background(), runtimePath, "app-management", "app:install-end", map[string]string{"app:name": "nextcloud"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if response.StatusCode != http.StatusAccepted {
+		t.Fatalf("status %d, want %d", response.StatusCode, http.StatusAccepted)
+	}
+
+	got := <-received
+	if got.method != http.MethodPost || got.path != "/v2/message_bus/event/app-management/app:install-end" || got.contentType != "application/json" {
+		t.Errorf("got %s %s (%s)", got.method, got.path, got.contentType)
+	}
+	if got.properties["app:name"] != "nextcloud" {
+		t.Errorf("got properties %v", got.properties)
+	}
+
+	if _, err := PublishEventInSocket(context.Background(), t.TempDir(), "app-management", "app:install-end", nil); err == nil {
+		t.Fatal("a runtime path without the message bus must fail")
+	}
+}