Judge a caller before it holds the socket, and keep accepting after a failed accept

Review 11959 of mesh-host#111: any account could hold the four slots for a second each and starve the provisioner's client, and one failed accept ended the socket until the engine restarted. A caller is judged from the kernel's word before it holds a slot and, refused, is answered and closed at once; busy refusals that wait are bounded as mesh-cli's are; a failed accept is said, backed off and retried. The request refuses unknown fields, and the socket's directory is given its mode. Issue 668.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/filling/filling.go

@@ -192,23 +192,42 @@ var (
 	sameNamespaces = meshcli.SameNamespaces
 )
 
-// within bounds how long a connection may take to say its request, and refusedWithin how long a refused one is
-// read and dropped before its refusal is written (so the client is not answered with a broken pipe). Variables so a
-// test need not wait.
+// within bounds how long an accepted connection may take to say its request, and refusedWithin how long a busy one
+// is read and dropped before its refusal is written (so the client is not answered with a broken pipe). Variables so
+// a test need not wait.
 var (
 	within        = 30 * time.Second
 	refusedWithin = 1 * time.Second
 )
 
-// atOnce bounds the connections served at the same time: one client, writing one file at a time. One over is
-// refused busy at once.
-var atOnce = 4
+// atOnce bounds the accepted connections served at the same time: one client, writing one file at a time. One over
+// is refused busy. busyWaiting bounds how many busy refusals wait for their request at once; over it a busy refusal
+// is written at once, as mesh-cli's socket does. Variables so a test can say.
+var (
+	atOnce        = 4
+	busyWaiting   = make(chan struct{}, 8)
+	judgingAtOnce = 16
+)
+
+// acceptBackoff is how long the socket waits after an accept that failed before it accepts again, at first and at
+// most: a socket that stopped at the first such failure (too many open files, say) would stop the client's writes
+// until the engine restarted. Variables so a test need not wait.
+var (
+	acceptBackoff    = 50 * time.Millisecond
+	acceptBackoffMax = 5 * time.Second
+)
+
+// acceptFrom accepts one connection. A variable so a test can make it fail.
+var acceptFrom = func(l *net.UnixListener) (*net.UnixConn, error) { return l.AcceptUnix() }
 
 // Serve listens on socket until ctx ends, and answers each connection: one request, one answer.
 //
 // The directory is made 0755 and the socket 0666, as mesh-cli's: who may write is judged per connection from the
-// kernel's word, against the current declaration, so the socket needs no account that may not exist yet. A file at
-// the socket's path that is not a socket is never removed: the engine says so and serves nothing.
+// kernel's word, against the current declaration, so the socket needs no account that may not exist yet. **The
+// caller is judged before it holds anything**: one the kernel does not say is the provisioner client's account is
+// answered and closed at once, and only the client's connections are served or wait as busy, so no other account can
+// hold the socket from it. A file at the socket's path that is not a socket is never removed: the engine says so and
+// serves nothing.
 func Serve(ctx context.Context, socket string, w Writer, say func(string)) error {
 	if say == nil {
 		say = func(string) {}
@@ -220,6 +239,9 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error
 	if problem := meshcli.NotOwnDirectory(dir); problem != "" {
 		return fmt.Errorf("filled files are not written on request on this machine: %s", problem)
 	}
+	if err := os.Chmod(dir, 0o755); err != nil {
+		return fmt.Errorf("the filled files' socket directory: %w", err)
+	}
 	if info, err := os.Lstat(socket); err == nil {
 		if info.Mode()&os.ModeSocket == 0 {
 			return fmt.Errorf("%s is there and is not a socket, and the node-engine did not put it there; filled "+
@@ -242,59 +264,117 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error
 		l.Close()
 	}()
 	slots := make(chan struct{}, atOnce)
+	// Judging reads the declaration: at most judging callers are judged at once, and the rest wait their turn — a
+	// few milliseconds each — rather than being refused, so a flood delays the client's write and never stops it.
+	judging := make(chan struct{}, judgingAtOnce)
+	refusals := &refusalWords{say: say, last: map[uint32]time.Time{}}
 	var wg sync.WaitGroup
 	defer wg.Wait()
+	backoff := time.Duration(0)
 	for {
-		conn, err := l.AcceptUnix()
+		conn, err := acceptFrom(l)
 		if err != nil {
 			if ctx.Err() != nil {
 				return nil
 			}
-			var ne net.Error
-			if errors.As(err, &ne) && ne.Timeout() {
-				continue
+			if errors.Is(err, net.ErrClosed) {
+				return fmt.Errorf("the filled files' socket was closed: %w", err)
 			}
-			return fmt.Errorf("the filled files' socket: %w", err)
+			// Said, waited for, and accepted again: the failure is the machine's for a moment, not the socket's.
+			backoff = min(max(2*backoff, acceptBackoff), acceptBackoffMax)
+			say(fmt.Sprintf("the filled files' socket could not accept a connection (%v); accepting again in %s",
+				err, backoff))
+			select {
+			case <-ctx.Done():
+				return nil
+			case <-time.After(backoff):
+			}
+			continue
 		}
-		readBy := time.Now().Add(within)
-		_ = conn.SetDeadline(readBy)
+		backoff = 0
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			// Judged before it holds a slot or is waited for.
+			judging <- struct{}{}
+			uid, ok := judgePeer(conn, w, refusals)
+			<-judging
+			if !ok {
+				return
+			}
+			readBy := time.Now().Add(within)
+			_ = conn.SetDeadline(readBy)
 			select {
 			case slots <- struct{}{}:
 				defer func() { <-slots }()
-				answer(ctx, conn, readBy, w, say)
+				answer(ctx, conn, uid, w, say)
 			default:
-				refuse(conn, readBy, Refusal("", "busy: other writes are being served; ask again"), say)
+				busy(conn, readBy, say)
 			}
 		}()
 	}
 }
 
-// answer serves one connection.
-func answer(ctx context.Context, conn *net.UnixConn, readBy time.Time, w Writer, say func(string)) {
+// judgePeer reads the asking account from the kernel and judges it. A caller that is not the provisioner client's
+// account is answered with why and closed at once, its request never read: false.
+func judgePeer(conn *net.UnixConn, w Writer, refusals *refusalWords) (uint32, bool) {
+	closeWith := func(uid uint32, why string) (uint32, bool) {
+		refusals.said(uid, why)
+		_ = conn.SetWriteDeadline(time.Now().Add(refusedWithin))
+		_ = json.NewEncoder(conn).Encode(Refusal("", why))
+		conn.Close()
+		return uid, false
+	}
 	uid, pid, err := peerOf(conn)
 	if err != nil {
-		say("a filled file's write was refused: the asking account could not be read: " + err.Error())
-		refuse(conn, readBy, Refusal("", "the node-engine could not read which account is asking, so nothing was "+
-			"written: "+err.Error()), say)
-		return
+		return closeWith(^uint32(0), "the node-engine could not read which account is asking, so nothing was "+
+			"written: "+err.Error())
 	}
 	if err := sameNamespaces(pid); err != nil {
-		say(fmt.Sprintf("a filled file's write from user id %d was refused: not in this machine's namespaces: %v",
-			uid, err))
-		refuse(conn, readBy, Refusal("", "the asking process does not run in this machine's own namespaces, so "+
-			"nothing was written: "+err.Error()), say)
-		return
+		return closeWith(uid, "the asking process does not run in this machine's own namespaces, so nothing was "+
+			"written: "+err.Error())
 	}
 	if err := w.Accepts(uid); err != nil {
-		say(fmt.Sprintf("a filled file's write from user id %d was refused: %v", uid, err))
-		refuse(conn, readBy, Refusal("", err.Error()), say)
+		return closeWith(uid, err.Error())
+	}
+	return uid, true
+}
+
+// refusalWords says a refused caller in the journal at most once a minute for each user id, with how many more
+// were refused meanwhile: a flood from another account must not flood the journal either.
+type refusalWords struct {
+	mu     sync.Mutex
+	say    func(string)
+	last   map[uint32]time.Time
+	unsaid map[uint32]int
+}
+
+func (r *refusalWords) said(uid uint32, why string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.unsaid == nil {
+		r.unsaid = map[uint32]int{}
+	}
+	if at, ok := r.last[uid]; ok && time.Since(at) < time.Minute {
+		r.unsaid[uid]++
 		return
 	}
+	more := ""
+	if n := r.unsaid[uid]; n > 0 {
+		more = fmt.Sprintf(" (and %d more from it since it was last said)", n)
+	}
+	r.last[uid], r.unsaid[uid] = time.Now(), 0
+	r.say(fmt.Sprintf("a filled file's write from user id %d was refused%s: %s", uid, more, why))
+}
+
+// answer serves one connection from the provisioner client's account.
+func answer(ctx context.Context, conn *net.UnixConn, uid uint32, w Writer, say func(string)) {
 	var req Request
-	if err := json.NewDecoder(io.LimitReader(conn, MaxRequest)).Decode(&req); err != nil {
+	dec := json.NewDecoder(io.LimitReader(conn, MaxRequest))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&req); err != nil {
+		// Answered even when the deadline to say it passed: the deadline covered the write too.
+		_ = conn.SetWriteDeadline(time.Now().Add(refusedWithin))
 		reply(conn, Refusal("", "the request could not be read, so nothing was written: "+err.Error()), say)
 		return
 	}
@@ -314,16 +394,22 @@ func answer(ctx context.Context, conn *net.UnixConn, readBy time.Time, w Writer,
 	reply(conn, a, say)
 }
 
-// refuse reads and drops what a refused client wrote, briefly, and answers it: nothing it sent is kept.
-func refuse(conn *net.UnixConn, readBy time.Time, a Answer, say func(string)) {
-	by := time.Now().Add(refusedWithin)
-	if readBy.Before(by) {
-		by = readBy
+// busy answers the client's connection over atOnce: its request read and dropped first, briefly, so the client reads
+// the answer rather than a broken pipe — while no more than busyWaiting such refusals wait at once; over that, at once.
+func busy(conn *net.UnixConn, readBy time.Time, say func(string)) {
+	select {
+	case busyWaiting <- struct{}{}:
+		by := time.Now().Add(refusedWithin)
+		if readBy.Before(by) {
+			by = readBy
+		}
+		_ = conn.SetReadDeadline(by)
+		drain(conn)
+		<-busyWaiting
+	default:
 	}
-	_ = conn.SetReadDeadline(by)
-	drain(conn)
-	_ = conn.SetWriteDeadline(time.Now().Add(within))
-	reply(conn, a, say)
+	_ = conn.SetWriteDeadline(time.Now().Add(refusedWithin))
+	reply(conn, Refusal("", "busy: other writes are being served; ask again"), say)
 }
 
 // drain reads and drops a request up to its end — the newline after it, the end of the stream or the deadline —

internal/filling/filling_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net"
 	"os"
 	"path/filepath"
@@ -251,7 +252,17 @@ func TestTheSocketWritesForTheAccountTheKernelSays(t *testing.T) {
 func TestTheSocketRefusesAnotherAccountBeforeReadingItsRequest(t *testing.T) {
 	w := &fixed{uid: uint32(os.Getuid()) + 1}
 	socket, said := serve(t, w)
-	a := ask(t, socket, Request{Path: "/etc/m/m.conf", Content: []byte("secret value")})
+	conn, err := net.Dial("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	// Closed at once, its request never read: the write may meet a closed socket, and the refusal is there to read.
+	_ = json.NewEncoder(conn).Encode(Request{Path: "/etc/m/m.conf", Content: []byte("secret value")})
+	var a Answer
+	if err := json.NewDecoder(conn).Decode(&a); err != nil {
+		t.Fatal(err)
+	}
 	if a.Refused == "" || len(w.writes) != 0 {
 		t.Fatalf("another account was written for: %+v", a)
 	}
@@ -261,3 +272,160 @@ func TestTheSocketRefusesAnotherAccountBeforeReadingItsRequest(t *testing.T) {
 		}
 	}
 }
+
+// floodersAreAnother has the kernel's word say another account for every connection a test binds to a name holding
+// "flood": the test runs as one account, and the socket must tell the client from everyone else.
+func floodersAreAnother(t *testing.T) {
+	t.Helper()
+	was := peerOf
+	peerOf = func(conn *net.UnixConn) (uint32, int32, error) {
+		uid, pid, err := was(conn)
+		if a, ok := conn.RemoteAddr().(*net.UnixAddr); ok && a != nil && strings.Contains(a.Name, "flood") {
+			uid++
+		}
+		return uid, pid, err
+	}
+	t.Cleanup(func() { peerOf = was })
+}
+
+// A flood of connections from another account, each saying nothing and held open, does not stop the client's write:
+// each is refused and closed before it holds anything.
+func TestAFloodFromAnotherAccountDoesNotStopTheClient(t *testing.T) {
+	floodersAreAnother(t)
+	w := &fixed{uid: uint32(os.Getuid())}
+	socket, _ := serve(t, w)
+	from := t.TempDir()
+	var flood []*net.UnixConn
+	t.Cleanup(func() {
+		for _, c := range flood {
+			c.Close()
+		}
+	})
+	for i := 0; i < 3*atOnce+busyWaitingCap(); i++ {
+		c, err := net.DialUnix("unix", &net.UnixAddr{Name: filepath.Join(from, fmt.Sprintf("flood-%d", i)), Net: "unix"},
+			&net.UnixAddr{Name: socket, Net: "unix"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		flood = append(flood, c)
+	}
+	start := time.Now()
+	a := ask(t, socket, Request{Path: "/etc/m/m.conf", Content: []byte("x")})
+	if a.Refused != "" || a.Action != "updated" {
+		t.Fatalf("the client's write during a flood: %+v", a)
+	}
+	if took := time.Since(start); took > refusedWithin/2 {
+		t.Errorf("the client's write waited %s behind the flood", took)
+	}
+	// And every flooder was closed: it reads its refusal and then the end.
+	for _, c := range flood {
+		_ = c.SetReadDeadline(time.Now().Add(2 * time.Second))
+		var got Answer
+		if err := json.NewDecoder(c).Decode(&got); err != nil || got.Refused == "" {
+			t.Fatalf("a flooder was not refused at once: %+v %v", got, err)
+		}
+	}
+}
+
+func busyWaitingCap() int { return cap(busyWaiting) }
+
+// A client that connects and says nothing holds one slot only until within, and another write goes on meanwhile.
+func TestASilentClientDoesNotHoldTheSocket(t *testing.T) {
+	was := within
+	within = 300 * time.Millisecond
+	t.Cleanup(func() { within = was })
+	w := &fixed{uid: uint32(os.Getuid())}
+	socket, _ := serve(t, w)
+	silent, err := net.Dial("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer silent.Close()
+	if a := ask(t, socket, Request{Path: "/etc/m/m.conf", Content: []byte("x")}); a.Action != "updated" {
+		t.Fatalf("a write beside a silent client: %+v", a)
+	}
+	_ = silent.SetReadDeadline(time.Now().Add(3 * time.Second))
+	var got Answer
+	if err := json.NewDecoder(silent).Decode(&got); err != nil || got.Refused == "" {
+		t.Errorf("the silent client was not let go after within: %+v %v", got, err)
+	}
+}
+
+// A request carrying more than the path and the content is refused: it says what the declaration says.
+func TestARequestWithAnUnknownFieldIsRefused(t *testing.T) {
+	w := &fixed{uid: uint32(os.Getuid())}
+	socket, _ := serve(t, w)
+	conn, err := net.Dial("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	if _, err := conn.Write([]byte(`{"path":"/etc/m/m.conf","content":"eA==","mode":"0777"}` + "\n")); err != nil {
+		t.Fatal(err)
+	}
+	var a Answer
+	if err := json.NewDecoder(conn).Decode(&a); err != nil {
+		t.Fatal(err)
+	}
+	if a.Refused == "" || !strings.Contains(a.Refused, "mode") || len(w.writes) != 0 {
+		t.Errorf("a request with an unknown field: %+v, writes %v", a, w.writes)
+	}
+}
+
+// An accept that fails is said, waited for and tried again: the socket goes on serving.
+func TestTheSocketGoesOnAfterAnAcceptFails(t *testing.T) {
+	wasAccept, wasBackoff := acceptFrom, acceptBackoff
+	failures := 3
+	var mu sync.Mutex
+	acceptFrom = func(l *net.UnixListener) (*net.UnixConn, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		if failures > 0 {
+			failures--
+			return nil, errors.New("accept: too many open files")
+		}
+		return l.AcceptUnix()
+	}
+	acceptBackoff = time.Millisecond
+	t.Cleanup(func() { acceptFrom, acceptBackoff = wasAccept, wasBackoff })
+	w := &fixed{uid: uint32(os.Getuid())}
+	socket, said := serve(t, w)
+	if a := ask(t, socket, Request{Path: "/etc/m/m.conf", Content: []byte("x")}); a.Action != "updated" {
+		t.Fatalf("after failed accepts: %+v", a)
+	}
+	if !strings.Contains(strings.Join(said(), "\n"), "too many open files") {
+		t.Errorf("the failed accepts were not said: %v", said())
+	}
+}
+
+// The socket's directory is given the mode the socket needs, as mesh-cli's is.
+func TestTheSocketsDirectoryIsOpenedToEveryAccount(t *testing.T) {
+	dir, err := os.MkdirTemp("", "fill")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	if err := os.Chmod(dir, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	stopped := make(chan error)
+	go func() { stopped <- Serve(ctx, filepath.Join(dir, "fill.sock"), &fixed{}, nil) }()
+	for i := 0; i < 100; i++ {
+		if _, err := os.Stat(filepath.Join(dir, "fill.sock")); err == nil {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	if err := <-stopped; err != nil {
+		t.Fatal(err)
+	}
+	info, err := os.Stat(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info.Mode().Perm() != 0o755 {
+		t.Errorf("the socket's directory is %o", info.Mode().Perm())
+	}
+}