Test the returning socket by a real restart of the stand-in node-engine

modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client.go

@@ -119,6 +119,9 @@ type Client struct {
 	Templates *templateStore
 	// Now dates the reports; nil is time.Now.
 	Now func() time.Time
+	// SocketUp wakes the client when the node-engine's write socket is made again (watchSocket); nil never does.
+	// A module whose write failed for want of the socket is filled again then (issue 1165), and nothing else is.
+	SocketUp <-chan struct{}
 
 	// watchBound watches a provider's bucket of entries for this machine's keys, once per run (consumers.go); set by
 	// Run.
@@ -147,6 +150,8 @@ type state struct {
 	// of the consumers left out of one of its files.
 	providing map[string]bool
 	leftOut   map[string]string
+	// unplaced is a module whose last fill failed for want of the node-engine's write socket (issue 1165).
+	unplaced map[string]bool
 	// bound is each provider's bucket of entries this client watches, by key: this machine's consumers' entries alone;
 	// boundReady is a bucket whose current entries are all read, and boundUsers the modules filled from each.
 	bound      map[string]map[string][]byte
@@ -161,6 +166,8 @@ type update struct {
 	changed []byte
 	// asked is the tool runner asking for a module's value (asks.go). Set, entry is nil.
 	asked *nats.Msg
+	// socketUp is the node-engine's write socket made again (issue 1165). Set, entry is nil.
+	socketUp bool
 }
 
 // Run serves until ctx ends or a watch stops, which it returns: run again, it starts from the current values.
@@ -248,6 +255,23 @@ func (c *Client) Run(ctx context.Context) error {
 		return fmt.Errorf("listening for %s: %w", SecretChangedSubject, err)
 	}
 	defer func() { _ = said.Unsubscribe() }()
+	// The node-engine's write socket made again: what failed for want of it is filled again, in the loop below.
+	if c.SocketUp != nil {
+		go func() {
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-c.SocketUp:
+					select {
+					case updates <- update{socketUp: true}:
+					case <-ctx.Done():
+						return
+					}
+				}
+			}
+		}()
+	}
 	// This machine's tool runner asking for a module's config value (asks.go): answered from this run's state, in
 	// its loop, so an answer is never read from half an update.
 	asks, err := c.Conn.Subscribe(AskSubject(c.Node), func(m *nats.Msg) {
@@ -266,7 +290,7 @@ func (c *Client) Run(ctx context.Context) error {
 		reported: map[string][]byte{}, secrets: map[string]heldSecret{}, items: map[string]map[string]itemValue{},
 		recorded: map[string][]byte{}, assigned: map[string]bool{}, binds: map[string]binding{},
 		providing: map[string]bool{}, leftOut: map[string]string{}, bound: map[string]map[string][]byte{},
-		boundReady: map[string]bool{}, boundUsers: map[string]map[string]bool{}}
+		boundReady: map[string]bool{}, unplaced: map[string]bool{}, boundUsers: map[string]map[string]bool{}}
 	// Every current value first: nothing is filled from half of them.
 	current := map[string]bool{}
 	var early []*nats.Msg
@@ -282,6 +306,9 @@ func (c *Client) Run(ctx context.Context) error {
 				early = append(early, u.asked)
 				continue
 			}
+			if u.socketUp {
+				continue // the first fill below tries every file
+			}
 			if u.changed != nil {
 				// Nothing is held yet: every secret is asked for when it is first filled.
 				continue
@@ -326,6 +353,10 @@ func (c *Client) Run(ctx context.Context) error {
 				switch {
 				case u.asked != nil:
 					c.respond(s, u.asked)
+				case u.socketUp:
+					for m := range s.unplaced {
+						dirty[m] = true
+					}
 				case u.changed != nil:
 					for _, m := range c.secretChanged(s, u.changed) {
 						dirty[m] = true
@@ -550,6 +581,7 @@ func (c *Client) leftModules(s *state) []string {
 // leave removes a module's needs and its report: it is no longer the provisioner's on this node.
 func (c *Client) leave(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, module string) {
 	prefix := c.Node + "." + module + "."
+	delete(s.unplaced, module)
 	removed := 0
 	for _, key := range sortedKeys(s.needs) {
 		if strings.HasPrefix(key, prefix) {
@@ -660,6 +692,7 @@ func (c *Client) reconcile(ctx context.Context, kvs map[string]jetstream.KeyValu
 		c.forgetItems(ctx, kvs, s, module, keep)
 	}
 	moved := false
+	delete(s.unplaced, module)
 	for _, f := range targets {
 		// File by file (issue 1553, ADR 0349 clause 4): a target with a value that waits keeps the file there is.
 		if !allKnown(f.text, values) {
@@ -672,6 +705,9 @@ func (c *Client) reconcile(ctx context.Context, kvs map[string]jetstream.KeyValu
 			err = errFill(a)
 		}
 		if err != nil {
+			if socketAbsent(err) {
+				s.unplaced[module] = true
+			}
 			failed = append(failed, fmt.Sprintf("%s: %v", f.Path, err))
 			continue
 		}

modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go

@@ -200,6 +200,14 @@ type engine struct {
 }
 
 func startEngine(t *testing.T, owners map[string]string) *engine {
+	t.Helper()
+	e := newEngine(t, owners)
+	e.listen(t)
+	return e
+}
+
+// newEngine is an engine whose socket is not made yet: a node-engine not started, or replacing itself.
+func newEngine(t *testing.T, owners map[string]string) *engine {
 	t.Helper()
 	// A unix socket's path is short: under /tmp, not the test's own directory.
 	dir, err := os.MkdirTemp("/tmp", "p5-")
@@ -209,6 +217,12 @@ func startEngine(t *testing.T, owners map[string]string) *engine {
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	e := &engine{root: t.TempDir(), socket: filepath.Join(dir, "fill.sock"), owner: owners,
 		moved: map[string]int{}, asked: map[string]int{}, wrote: map[string]string{}}
+	return e
+}
+
+// listen makes the socket and answers on it until stop, which removes it as a node-engine stopping does.
+func (e *engine) listen(t *testing.T) (stop func()) {
+	t.Helper()
 	l, err := net.Listen("unix", e.socket)
 	if err != nil {
 		t.Fatal(err)
@@ -233,7 +247,7 @@ func startEngine(t *testing.T, owners map[string]string) *engine {
 			}()
 		}
 	}()
-	return e
+	return func() { l.Close(); os.Remove(e.socket) }
 }
 
 func (e *engine) at(path string) string { return filepath.Join(e.root, path) }
@@ -349,10 +363,14 @@ func aClient(t *testing.T, b *aBus, e *engine, state string) (*Client, func()) {
 		t.Fatal(err)
 	}
 	secrets.Within = 2 * time.Second
-	c := &Client{Node: node, JS: js, Conn: nc, Secrets: secrets,
+	ctx, cancel := context.WithCancel(context.Background())
+	up, err := watchSocket(ctx, e.socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	c := &Client{Node: node, JS: js, Conn: nc, Secrets: secrets, SocketUp: up,
 		Fill:      SocketFiller{Socket: e.socket, Busy: 10 * time.Millisecond, Tries: 3},
 		Templates: &templateStore{dir: state, fetch: HTTPFetch}}
-	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan error, 1)
 	go func() { done <- c.Run(ctx) }()
 	stopped := false
@@ -703,3 +721,96 @@ func TestNatsResticsFilesFillFromTheBackupPlacement(t *testing.T) {
 	b.assign(t, "restic", plan("# a\n/srv/a\n", "# a\nstate\n"))
 	eventually(t, "a module taken away refills backups.conf", func() bool { return e.read(t, backups) == head+"# a\n/srv/a\n" })
 }
+
+// Issue 1165: a file whose write failed because the node-engine's socket was gone is written when the socket is made
+// again, with no timer; a file that was placed is not asked again by it.
+func TestNatsAFillThatFailedForWantOfTheSocketIsTriedWhenTheSocketReturns(t *testing.T) {
+	b := startBus(t)
+	heard := b.listen(t)
+	e := newEngine(t, map[string]string{dunstrc: "owned"})
+	b.config(t, "dunst", "font-size", "13")
+	b.assign(t, "dunst", dunst(t))
+	aClient(t, b, e, t.TempDir())
+
+	eventually(t, "the failure is said", func() bool {
+		r, _ := heard.last("dunst")
+		return r.Status == StatusFailed && strings.Contains(r.Reason, "fill.sock")
+	})
+	if got := e.read(t, dunstrc); got != "" {
+		t.Fatalf("written without a socket: %q", got)
+	}
+	e.listen(t)
+	eventually(t, "dunstrc is written once the socket is back", func() bool {
+		return strings.Contains(e.read(t, dunstrc), "Monospace 13")
+	})
+	eventually(t, "and the report says applied", func() bool {
+		r, _ := heard.last("dunst")
+		return r.Status == StatusApplied
+	})
+}
+
+func TestNatsAReturningSocketAsksNothingOfAModuleThatWasPlaced(t *testing.T) {
+	b := startBus(t)
+	e := newEngine(t, map[string]string{dunstrc: "owned"})
+	stop := e.listen(t)
+	b.config(t, "dunst", "font-size", "13")
+	b.assign(t, "dunst", dunst(t))
+	aClient(t, b, e, t.TempDir())
+	eventually(t, "dunstrc is filled", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 13") })
+	_, before := e.counts(dunstrc)
+
+	// The node-engine replaces itself: the socket goes and returns, and a later change proves the loop ran past it.
+	stop()
+	e.listen(t)
+	b.config(t, "dunst", "font-size", "15")
+	eventually(t, "the change is written", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 15") })
+	_, after := e.counts(dunstrc)
+	if after-before != 1 {
+		t.Errorf("dunstrc was asked %d times for one change", after-before)
+	}
+}
+
+func TestWatchSocketWakesWhenTheSocketIsMadeAndWhenItsDirectoryReturns(t *testing.T) {
+	base, err := os.MkdirTemp("/tmp", "ws-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(base) })
+	dir := filepath.Join(base, "engine")
+	if err := os.Mkdir(dir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	socket := filepath.Join(dir, "fill.sock")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	up, err := watchSocket(ctx, socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	heard := func(what string) {
+		t.Helper()
+		select {
+		case <-up:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("no wake for %s", what)
+		}
+	}
+	l, err := net.Listen("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	heard("the socket made")
+	l.Close()
+	// The whole directory goes, as with a runtime directory removed on stop, and returns with the socket.
+	if err := os.RemoveAll(dir); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Mkdir(dir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if l, err = net.Listen("unix", socket); err != nil {
+		t.Fatal(err)
+	}
+	defer l.Close()
+	heard("the directory and socket back")
+}

modules/mesh-provisioner-client/cmd/mesh-provisioner-client/main.go

@@ -133,7 +133,11 @@ func serve(ctx context.Context, credential, socket, stateDir string) error {
 	if err != nil {
 		return err
 	}
-	client := &Client{Node: c.Node, JS: js, Conn: nc, Secrets: secrets,
+	up, err := watchSocket(ctx, socket)
+	if err != nil {
+		return err
+	}
+	client := &Client{Node: c.Node, SocketUp: up, JS: js, Conn: nc, Secrets: secrets,
 		Fill:      SocketFiller{Socket: socket, Busy: time.Second, Tries: 10},
 		Templates: &templateStore{dir: filepath.Join(stateDir, "templates"), fetch: HTTPFetch}}
 	wait := time.Second

modules/mesh-provisioner-client/cmd/mesh-provisioner-client/socketwatch.go

@@ -0,0 +1,103 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"syscall"
+	"unsafe"
+
+	"golang.org/x/sys/unix"
+)
+
+// socketAbsent says whether a write failed because nothing answers at the node-engine's write socket: its file is
+// gone, or it is there with no one listening. That is the node-engine replacing itself (issue 1165); the write is
+// tried again when the socket is back (watchSocket).
+func socketAbsent(err error) bool {
+	return errors.Is(err, syscall.ENOENT) || errors.Is(err, syscall.ECONNREFUSED)
+}
+
+// watchSocket tells, on the channel it returns, each time the file at socket is made, moved in or changed in its
+// attributes — a node-engine that starts makes its write socket and sets its mode — until ctx ends. It is the
+// operating system's own event (inotify), not a timer: the client is on the node-engine's machine. Events that
+// come while one is unread are one. When the socket's directory is gone it watches the directory above for its
+// return. The channel only wakes: what is owed a write is the client's to read.
+func watchSocket(ctx context.Context, socket string) (<-chan struct{}, error) {
+	fd, err := unix.InotifyInit1(unix.IN_CLOEXEC | unix.IN_NONBLOCK)
+	if err != nil {
+		return nil, fmt.Errorf("watching %s: %w", socket, err)
+	}
+	f := os.NewFile(uintptr(fd), "inotify")
+	out := make(chan struct{}, 1)
+	dir := filepath.Dir(socket)
+	// The directory's watch and, while the directory is absent, its parent's.
+	if err := addWatches(fd, dir); err != nil {
+		_ = f.Close()
+		return nil, fmt.Errorf("watching %s: %w", socket, err)
+	}
+	go func() { <-ctx.Done(); _ = f.Close() }()
+	go func() {
+		buf := make([]byte, 16*(unix.SizeofInotifyEvent+unix.NAME_MAX+1))
+		for {
+			n, err := f.Read(buf)
+			if err != nil {
+				return
+			}
+			wake, rewatch := false, false
+			for off := 0; off+unix.SizeofInotifyEvent <= n; {
+				ev := (*unix.InotifyEvent)(unsafe.Pointer(&buf[off]))
+				name := ""
+				if ev.Len > 0 {
+					raw := buf[off+unix.SizeofInotifyEvent : off+unix.SizeofInotifyEvent+int(ev.Len)]
+					for i, b := range raw {
+						if b == 0 {
+							raw = raw[:i]
+							break
+						}
+					}
+					name = string(raw)
+				}
+				off += unix.SizeofInotifyEvent + int(ev.Len)
+				switch {
+				case ev.Mask&(unix.IN_DELETE_SELF|unix.IN_MOVE_SELF|unix.IN_IGNORED) != 0:
+					rewatch = true
+				case name == filepath.Base(socket):
+					wake = true
+				case name == filepath.Base(dir):
+					// The directory came back: watched again, and the socket may already be in it.
+					rewatch, wake = true, true
+				}
+			}
+			if rewatch {
+				if err := addWatches(fd, dir); err != nil {
+					logf("[provisioner-client] watching %s again: %v", socket, err)
+				}
+				if _, err := os.Lstat(socket); err == nil {
+					wake = true
+				}
+			}
+			if wake {
+				select {
+				case out <- struct{}{}:
+				default:
+				}
+			}
+		}
+	}()
+	return out, nil
+}
+
+// addWatches watches dir for what is made in it, and its parent for dir's own return; a watch already there is kept.
+func addWatches(fd int, dir string) error {
+	if _, err := unix.InotifyAddWatch(fd, filepath.Dir(dir), unix.IN_CREATE|unix.IN_MOVED_TO); err != nil {
+		return err
+	}
+	_, err := unix.InotifyAddWatch(fd, dir, unix.IN_CREATE|unix.IN_MOVED_TO|unix.IN_ATTRIB|unix.IN_DELETE_SELF|
+		unix.IN_MOVE_SELF)
+	if errors.Is(err, syscall.ENOENT) {
+		return nil // absent: its parent's watch tells when it is made
+	}
+	return err
+}

modules/mesh-provisioner-client/go.mod

@@ -6,6 +6,7 @@ require (
 	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021114014-66e8fdaa62e2
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
+	golang.org/x/sys v0.48.0
 )
 
 require (
@@ -17,6 +18,5 @@ require (
 	github.com/nats-io/nkeys v0.4.16 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	golang.org/x/crypto v0.57.0 // indirect
-	golang.org/x/sys v0.48.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
 )

modules/minio/client.ts

@@ -256,11 +256,22 @@ export class MinioClient {
     if (argv.some((a) => a.includes(this.rootPassword))) {
       throw new Error("refused to run mc: its command line would carry the root password (novox/hq issue 282)");
     }
-    const { stdout } = await execFileAsync(this.mcBin, argv, {
-      timeout: 30_000,
-      env: { ...process.env, MC_HOST_mesh: mcHost(this.baseUrl, this.rootUser, this.rootPassword) },
-    });
-    return stdout.trim();
+    const mcHostValue = mcHost(this.baseUrl, this.rootUser, this.rootPassword);
+    try {
+      const { stdout } = await execFileAsync(this.mcBin, argv, {
+        timeout: 30_000,
+        env: { ...process.env, MC_HOST_mesh: mcHostValue },
+      });
+      return stdout.trim();
+    } catch (err) {
+      // A fresh error with only the redacted message: execFile's own carries the whole command line, and
+      // its cmd, stdout and stderr properties would carry it on (issue 1766).
+      const known = [mcHostValue, this.rootPassword, encodeURIComponent(this.rootPassword), ...secretArgs(argv)];
+      const e = err as { stderr?: unknown; message?: string };
+      const reason = typeof e.stderr === "string" && e.stderr.trim() !== "" ? e.stderr.trim() : (e.message ?? String(err));
+      const shown = argv.filter((a) => a !== this.mcConfigDir).join(" ");
+      throw new Error(redactSecrets(`mc ${shown} failed: ${reason}`, known));
+    }
   }
 
   // --- SigV4 request plumbing ---------------------------------------------
@@ -360,6 +371,26 @@ function uriEncode(str: string, encodeSlash: boolean): string {
   return out;
 }
 
+/** The values that follow --secret-key and --access-key in an argument list. */
+function secretArgs(argv: string[]): string[] {
+  const out: string[] = [];
+  argv.forEach((a, i) => { if ((a === "--secret-key" || a === "--access-key") && i + 1 < argv.length) out.push(argv[i + 1]); });
+  return out;
+}
+
+/**
+ * The one redaction every `mc` failure goes through (issue 1766): the token after --secret-key or
+ * --access-key (as a separate word or after "="), any MC_HOST_ value, and each known secret, wherever they
+ * appear in the text, become "[redacted]".
+ */
+export function redactSecrets(text: string, known: string[] = []): string {
+  let out = text
+    .replace(/(--(?:secret|access)-key(?:=|\s+))('[^']*'|"[^"]*"|\S+)/g, "$1[redacted]")
+    .replace(/(MC_HOST_\w+=)\S+/g, "$1[redacted]");
+  for (const k of known) if (k !== "") out = out.split(k).join("[redacted]");
+  return out;
+}
+
 /**
  * The MC_HOST_<alias> URL `mc` reads an alias from: the endpoint with the credentials in its userinfo,
  * each percent-encoded so a reserved character in either cannot end it early.

modules/minio/links.ts

@@ -36,6 +36,9 @@ export function linkPolicy(bucket: string): string {
   });
 }
 
+/** A link key's secret: 40 hex characters, minio's limit for a secret key (issue 1766). */
+const LINK_SECRET = /^[0-9a-f]{40}$/;
+
 export function linkSecretPath(dir: string, bucket: string): string {
   return join(dir, `${bucket}.secret`);
 }
@@ -47,9 +50,12 @@ export async function ensureLinkKey(admin: LinkAdmin, dir: string, bucket: strin
   const path = linkSecretPath(dir, bucket);
   mkdirSync(dir, { recursive: true, mode: 0o700 });
   let secret = existsSync(path) ? readFileSync(path, "utf8").trim() : "";
+  // A stored secret of the wrong shape (an earlier version made 64 characters; minio takes at most 40) is
+  // replaced, and its key made again (issue 1766).
+  if (!LINK_SECRET.test(secret)) secret = "";
   if (secret !== "" && (await admin.accessKeyExists(id))) return;
   if (secret === "") {
-    secret = randomBytes(32).toString("hex");
+    secret = randomBytes(20).toString("hex");
     writeFileSync(path, secret + "\n", { mode: 0o600 });
     chmodSync(path, 0o600);
   }

modules/minio/test/links.test.ts

@@ -115,3 +115,58 @@ test("the link key id is the one the object-store tool derives (cmd/object-store
   // 1516/AC1
   assert.equal(linkKeyId("brother"), "link0b4bd77cec705cf5");
 });
+
+// Issue 1766: a link key's secret is minio's length, and no mc failure carries an argument's secret.
+
+import { chmodSync, writeFileSync, mkdirSync } from "node:fs";
+import { MinioClient } from "../client.ts";
+
+test("a link key's secret is 40 characters, and a stored 64-character secret is replaced", async () => {
+  // 1766/AC1
+  const { minio, dir } = setup();
+  await ensureLinkKey(minio, dir, "fresh");
+  assert.equal(minio.keys.get(linkKeyId("fresh"))!.secret.length, 40);
+  assert.match(minio.keys.get(linkKeyId("fresh"))!.secret, /^[0-9a-f]{40}$/);
+
+  mkdirSync(dir, { recursive: true });
+  const old = "ab".repeat(32);
+  writeFileSync(linkSecretPath(dir, "old"), old + "\n");
+  await ensureLinkKey(minio, dir, "old");
+  const made = minio.keys.get(linkKeyId("old"))!.secret;
+  assert.equal(made.length, 40);
+  assert.notEqual(made, old);
+  assert.equal(readFileSync(linkSecretPath(dir, "old"), "utf8").trim(), made);
+});
+
+function failingMc(): { client: MinioClient; root: string } {
+  const d = mkdtempSync(join(process.env.TMPDIR ?? tmpdir(), "minio-mc-"));
+  const bin = join(d, "mc");
+  // Says its arguments and its alias on stderr, then fails, the way a refused command does.
+  writeFileSync(bin, '#!/bin/sh\necho "args: $*" >&2\necho "env: MC_HOST_mesh=$MC_HOST_mesh" >&2\nexit 1\n');
+  chmodSync(bin, 0o755);
+  const root = "RootPass-s3cret/ok";
+  const client = new MinioClient({ endpoint: "http://127.0.0.1:9000", rootUser: "root", rootPassword: root, region: "us-east-1", mcBin: bin, mcConfigDir: d });
+  return { client, root };
+}
+
+test("an mc failure's message holds no --secret-key value and no MC_HOST value", async () => {
+  // 1766/AC2
+  const { client, root } = failingMc();
+  const secret = "SuperSecretValue0123456789";
+  const err: any = await client.createAccessKey("b", "ACCESSKEYID1234567", secret).then(() => null, (e) => e);
+  assert.ok(err, "the failure is raised");
+  for (const text of [String(err.message), String(err.stack ?? ""), JSON.stringify(Object.getOwnPropertyNames(err).map((k) => err[k]))]) {
+    assert.ok(!text.includes(secret), "no secret key");
+    assert.ok(!text.includes("ACCESSKEYID1234567"), "no access key");
+    assert.ok(!text.includes(root) && !text.includes(encodeURIComponent(root)), "no root password");
+    assert.ok(!text.includes("root:"), "no MC_HOST value");
+  }
+  assert.match(err.message, /\[redacted\]/);
+});
+
+test("a consumer key's create passes the minted secret unchanged", async () => {
+  // 1766/AC3
+  const s = setup();
+  await s.h.create(provision);
+  assert.equal(s.minio.keys.get("brother-key")!.secret, "consumer-secret");
+});