From 0c17c76493f774cafd82394ab78a02f0f35db226 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 00:47:48 +0000 Subject: [PATCH] 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 --- internal/filling/filling.go | 166 ++++++++++++++++++++++-------- internal/filling/filling_test.go | 170 ++++++++++++++++++++++++++++++- 2 files changed, 295 insertions(+), 41 deletions(-) diff --git a/internal/filling/filling.go b/internal/filling/filling.go index 8e13902..e14da2a 100644 --- a/internal/filling/filling.go +++ b/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 — diff --git a/internal/filling/filling_test.go b/internal/filling/filling_test.go index 52a6f95..2c198ad 100644 --- a/internal/filling/filling_test.go +++ b/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()) + } +}