From fddcddfaf720da89c399c10d725420574c058653 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 00:33:48 +0000 Subject: [PATCH 1/7] Name the provisioner client's account in the declaration, and share the apply's record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The local write of a filled file (ADR 0349 §4) accepts only the provisioner client's account, which the declaration names; the reader takes provisioner_account before any controller sends it (contract 10). The apply's record and its rule for what moved become helpers the write uses too. Issue 668. Co-Authored-By: Claude Opus 5.5 --- cmd/mesh-host/main.go | 8 +++- cmd/mesh-host/units_test.go | 9 ++-- internal/apply/apply.go | 72 +++++++++++++++++------------ internal/declaration/declaration.go | 26 ++++++++++- internal/link/messages.go | 6 +++ validate/validate_test.go | 6 +++ 6 files changed, 91 insertions(+), 36 deletions(-) diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 0837e9a..f7d7dc1 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1280,6 +1280,9 @@ func runLink(ctx context.Context, opts options) error { // And mesh-cli is answered on this machine: the account from the kernel, the line asked of the controller // (novox/hq ADR 0272). go serveMeshCLI(aside, queue, say) + // And a filled file is written on the provisioner client's request, one at a time with the apply (novox/hq ADR + // 0349 §4). + go serveFilling(aside, opts, mine.Membership.Signer, say) held := link.HoldRoused(aside, membership, queue, say, opts.timeout, rousedBySignal(ctx)) // The act in hand finishes and is kept before this process exits: an apply that stood aside with @@ -1770,8 +1773,9 @@ func healthAsReported(st liveness.Statement, ns *network.Statement) *link.Health // LibraryContract: this engine reads a resource's declared `health` and judges it (ADR 0240 Phase B), judges a // user's declared `root` (novox/hq ADR 0266), and reads a declaration's identities, groups and libraries // (novox/hq ADR 0328), a container's busy (ADR 0335) and a file's owner-at-runtime and the filled files (ADR - // 0349 §5), which is what tells the controller it may be sent any of those fields. - h := &link.Health{Contract: link.OwnershipContract, At: st.At.UTC(), Resources: []link.ResourceHealth{}} + // 0349 §5) and the provisioner client's account (ADR 0349 §4), which is what tells the controller it may be sent + // any of those fields. + h := &link.Health{Contract: link.ProvisionerContract, At: st.At.UTC(), Resources: []link.ResourceHealth{}} for _, r := range st.Resources { h.Resources = append(h.Resources, link.ResourceHealth{Module: r.Module, Resource: r.ID, Kind: r.Kind, Target: r.Target, State: r.State, Reason: r.Reason, Since: r.Since.UTC(), Streak: r.Streak, diff --git a/cmd/mesh-host/units_test.go b/cmd/mesh-host/units_test.go index a81df0e..423c286 100644 --- a/cmd/mesh-host/units_test.go +++ b/cmd/mesh-host/units_test.go @@ -87,10 +87,11 @@ func TestTheStatementSaysAnAccountJudgedForRoot(t *testing.T) { Groups: []string{"openrazer"}}, State: accounts.Healthy, Since: at}, }} h := withAccounts(healthAsReported(liveness.Statement{At: at}, nil), as) - // At RootContract or later: an engine that also reads a file's owner-at-runtime and the filled files says - // OwnershipContract (novox/hq ADR 0349 §5), one above ADR 0335's BusyContract. - if h.Contract < link.RootContract || link.RootContract != 3 || h.Contract != link.OwnershipContract || - link.OwnershipContract != link.BusyContract+1 || link.BusyContract != link.OwnGroupContract+1 { + // At RootContract or later: an engine that also reads the provisioner client's account says ProvisionerContract + // (novox/hq ADR 0349 §4), one above ADR 0349 §5's OwnershipContract. + if h.Contract < link.RootContract || link.RootContract != 3 || h.Contract != link.ProvisionerContract || + link.ProvisionerContract != link.OwnershipContract+1 || link.OwnershipContract != link.BusyContract+1 || + link.BusyContract != link.OwnGroupContract+1 { t.Fatalf("contract %d", h.Contract) } got := map[string]string{} diff --git a/internal/apply/apply.go b/internal/apply/apply.go index 0305047..9b98d0b 100644 --- a/internal/apply/apply.go +++ b/internal/apply/apply.go @@ -999,34 +999,7 @@ func ApplyMindingWindows( kept, keptMode, keptOwner = from.Kept, from.KeptMode, from.KeptOwner } // Only now. The record follows the fact, never leads it. - known.Record(store.Applied{ - Origin: origin, - ID: resource.Identity(), Type: string(resource.Kind()), - Target: outcome.Target, AppliedAt: time.Now().UTC(), - Wrote: outcome.wrote, - Into: outcome.into, - Kept: kept, - KeptMode: keptMode, - KeptOwner: keptOwner, - Reads: outcome.reads, - BusySince: outcome.busySince, - Stateless: outcome.stateless, - Scope: outcome.scope, - User: outcome.user, - Found: outcome.found, - DeclaredUnit: outcome.declared, - Shell: outcome.shell, - Linger: outcome.linger, - Groups: outcome.groups, - Unpacked: outcome.unpacked, - AsFound: outcome.asFound, - Declared: outcome.asFoundDeclared, - FoundAt: outcome.asFoundAt, - Holds: holds(resource), - }) - if outcome.found != nil { - known.DropFound(resource.Identity()) - } + recordApplied(&known, origin, resource, outcome, kept, keptMode, keptOwner) // Its module has been taken, and what was held for it is now the mesh's. A file written // into, or a service whose lifecycle is the machine's, replaced nothing that was found, so // its outcome says what the apply did, not that a cutover happened; a hold from when it was @@ -1075,7 +1048,7 @@ func ApplyMindingWindows( if outcome.Action == heldStill || outcome.Action == waitsBusy { // Not changed: nothing about it moved, so nothing that restarts on it restarts. log(fmt.Sprintf(" %s %s (%s): %s", outcome.Action, outcome.ID, outcome.Target, outcome.Detail)) - } else if outcome.Action != "unchanged" { + } else if moved(outcome) { changed[resource.Identity()] = true // With the detail, when there is one: "updated app" says a container was replaced; // which file made that happen is what somebody reading the log at the time needs @@ -1168,6 +1141,47 @@ func ApplyMindingWindows( return report, known, nil } +// recordApplied records what applying a resource did, once it is done: the record follows the fact, never leads it. +// kept, keptMode and keptOwner are where the original of what it replaced was kept, and how it was found. What an +// earlier apply found its unit as goes with it. +func recordApplied(known *store.State, origin string, resource declaration.Resource, outcome Outcome, + kept, keptMode, keptOwner string) { + known.Record(store.Applied{ + Origin: origin, + ID: resource.Identity(), Type: string(resource.Kind()), + Target: outcome.Target, AppliedAt: time.Now().UTC(), + Wrote: outcome.wrote, + Into: outcome.into, + Kept: kept, + KeptMode: keptMode, + KeptOwner: keptOwner, + Reads: outcome.reads, + BusySince: outcome.busySince, + Stateless: outcome.stateless, + Scope: outcome.scope, + User: outcome.user, + Found: outcome.found, + DeclaredUnit: outcome.declared, + Shell: outcome.shell, + Linger: outcome.linger, + Groups: outcome.groups, + Unpacked: outcome.unpacked, + AsFound: outcome.asFound, + Declared: outcome.asFoundDeclared, + FoundAt: outcome.asFoundAt, + Holds: holds(resource), + }) + if outcome.found != nil { + known.DropFound(resource.Identity()) + } +} + +// moved reports whether applying a resource moved it, so what restarts or reloads on it does: anything but +// unchanged, and not a container held still or waiting for its busy one to say idle. +func moved(outcome Outcome) bool { + return outcome.Action != "unchanged" && outcome.Action != heldStill && outcome.Action != waitsBusy +} + // afterWhatTheyRead is the resources in declared order, except that **a service comes after every // resource it names under restart-on or reload-on** (novox/hq issue 260). // diff --git a/internal/declaration/declaration.go b/internal/declaration/declaration.go index 0c165c3..6ef97ec 100644 --- a/internal/declaration/declaration.go +++ b/internal/declaration/declaration.go @@ -1645,6 +1645,25 @@ type Declaration struct { // node-engine write only these. Absent is a controller that hands no file to the provisioner, and every // file is written from the declaration as before. Filled []string + + // Provisioner is the account the provisioner's client runs as on this machine, by name (novox/hq ADR 0349 §4): + // the only account whose write of a filled file the node-engine's local write socket accepts. Absent is a + // controller that names none, and every such write is refused. + Provisioner string +} + +// checkProvisioner refuses a provisioner client's account that is no account name, or is root: the account whose +// writes the node-engine accepts must be one that could not write the file without it. +func checkProvisioner(name string) []string { + switch { + case name == "": + return nil + case !accountName.MatchString(name): + return []string{fmt.Sprintf("provisioner_account %q is not an account name", name)} + case name == "root": + return []string{"provisioner_account is root; the provisioner's client runs as an account of its own"} + } + return nil } // Fills reports whether the provisioner fills the file at path on this machine. @@ -2027,6 +2046,9 @@ type envelope struct { // Filled is optional too (novox/hq ADR 0349 §5), as is a file's `owner-at-runtime`. **An older host refuses // these keys**; a controller sends them only to a host whose health says link.OwnershipContract. Filled []string `json:"filled_files,omitempty"` + // Provisioner is optional too (novox/hq ADR 0349 §4). **An older host refuses this key**; a controller sends it + // only to a host whose health says link.ProvisionerContract. + Provisioner string `json:"provisioner_account,omitempty"` } func parse(raw []byte, allowActions bool) (*Declaration, error) { @@ -2045,8 +2067,10 @@ func parse(raw []byte, allowActions bool) (*Declaration, error) { d := &Declaration{Version: env.Version, For: env.For, Adoption: env.Adoption, Sequence: env.Sequence, Epoch: env.Epoch, LeftOut: env.LeftOut, Generation: env.Generation, PutBack: env.PutBack, - Identities: env.Identities, Groups: env.Groups, Libraries: env.Libraries, Filled: env.Filled} + Identities: env.Identities, Groups: env.Groups, Libraries: env.Libraries, Filled: env.Filled, + Provisioner: env.Provisioner} problems := checkIdentities(env.Identities, env.Groups) + problems = append(problems, checkProvisioner(env.Provisioner)...) problems = append(problems, checkLibraries(env.Groups, env.Libraries)...) if env.Generation < 0 { // Below zero would read as "none claimed" and pass every refusal of an older generation. diff --git a/internal/link/messages.go b/internal/link/messages.go index 5bc1828..3a849a1 100644 --- a/internal/link/messages.go +++ b/internal/link/messages.go @@ -302,6 +302,12 @@ const BusyContract = 8 // fields; an older host refuses the whole declaration for them. const OwnershipContract = 9 +// ProvisionerContract is the statement of a host that also reads a declaration's `provisioner_account` (novox/hq ADR +// 0349 §4) and serves the local write socket: it accepts a write of a filled file from that account alone. Its +// presence is what tells the controller this host may be sent the field; an older host refuses the whole +// declaration for it. +const ProvisionerContract = 10 + // Health is one statement of every long-running resource's state on this machine (to-be 48 §4). Said in // every report, as an event on each change, and again every minute while anything is not healthy — so a // lost statement is not a lost fault. diff --git a/validate/validate_test.go b/validate/validate_test.go index 433850c..51ade73 100644 --- a/validate/validate_test.go +++ b/validate/validate_test.go @@ -24,7 +24,13 @@ func TestTheValidatorIsTheHosts(t *testing.T) { if problems := Declaration([]byte(owned)); problems != nil { t.Fatalf("a declaration saying who owns a file at runtime was refused: %v", problems) } + // And one naming the provisioner client's account (novox/hq ADR 0349 §4). + named := `{"declaration": 1, "provisioner_account": "mesh-node-provisioner", "filled_files": ["/etc/x"], "resources": [{"id": "f", "type": "file", "path": "/etc/x", "content": "a", "owner-at-runtime": "owned"}]}` + if problems := Declaration([]byte(named)); problems != nil { + t.Fatalf("a declaration naming the provisioner client's account was refused: %v", problems) + } for _, bad := range []string{ + `{"declaration": 1, "provisioner_account": "root", "resources": [{"id": "f", "type": "file", "path": "/etc/x", "content": "a"}]}`, `{"declaration": 1, "filled_files": ["/etc/none"], "resources": [{"id": "f", "type": "file", "path": "/etc/x", "content": "a", "owner-at-runtime": "owned"}]}`, `{"declaration": 1, "resources": []}`, `{"declaration": 99, "resources": [{"id": "d", "type": "directory", "path": "/x"}]}`, From eb2ec23679609abdcac55a71f4ffd7ec79a1ca47 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 00:34:54 +0000 Subject: [PATCH 2/7] Write a filled file on the provisioner client's request, and nothing else MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The provisioner's client (ADR 0349 §4) fills a module's templates and needs the node-engine to write them: it must not hold the right to write the machine's files itself. A second local socket accepts a write only from the account the declaration names as the provisioner client's (SO_PEERCRED, same namespaces), only for a clean path the declaration names as filled, under the apply's own locks, and writes it through the apply as the declared file would be, firing restart-on, reload-on and container reads. Issue 668. Co-Authored-By: Claude Opus 5.5 --- cmd/mesh-host/filling.go | 74 ++++++++++++++++ cmd/mesh-host/filling_test.go | 105 +++++++++++++++++++++++ internal/declaration/provisioner_test.go | 37 ++++++++ internal/meshcli/peer.go | 19 ++++ 4 files changed, 235 insertions(+) create mode 100644 cmd/mesh-host/filling.go create mode 100644 cmd/mesh-host/filling_test.go create mode 100644 internal/declaration/provisioner_test.go create mode 100644 internal/meshcli/peer.go diff --git a/cmd/mesh-host/filling.go b/cmd/mesh-host/filling.go new file mode 100644 index 0000000..22f6dcc --- /dev/null +++ b/cmd/mesh-host/filling.go @@ -0,0 +1,74 @@ +package main + +import ( + "context" + "crypto/ed25519" + "errors" + "path/filepath" + + "github.com/novox/mesh-host/internal/apply" + "github.com/novox/mesh-host/internal/declaration" + "github.com/novox/mesh-host/internal/filling" + "github.com/novox/mesh-host/internal/store" + "github.com/novox/mesh-host/internal/system" +) + +// serveFilling writes filled files on the provisioner client's request for as long as this engine serves (novox/hq +// ADR 0349 §4). A socket that cannot be served is said once, and this engine goes on applying. +func serveFilling(ctx context.Context, opts options, signer ed25519.PublicKey, say func(string)) { + if err := filling.Serve(ctx, filling.Socket, fillingEngine(opts, signer, say), say); err != nil { + say("filled files are not written on request on this machine: " + err.Error()) + } +} + +// fillingEngine is the write of a filled file as this engine does it: the declaration it keeps, verified, read under +// the apply's lock — the same two locks every apply takes, the process's and the one beside the state — and the file +// written through the apply (apply.WriteFilled), its outcome recorded in the node's state as an apply's is. +func fillingEngine(opts options, signer ed25519.PublicKey, say func(string)) *filling.Engine { + return &filling.Engine{ + Declared: func() (*declaration.Declaration, error) { + raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer) + if err != nil { + return nil, err + } + return declaration.Parse(raw) + }, + UIDOf: filling.UIDOf, + Lock: func() (func(), error) { + applying.Lock() + unlock, err := store.Lock(opts.state, nil) + if err != nil { + applying.Unlock() + return nil, err + } + return func() { unlock(); applying.Unlock() }, nil + }, + Place: func(ctx context.Context, d *declaration.Declaration, path string, content []byte) (string, string, error) { + built, err := system.For(builtFor) + if err != nil { + return "", "", err + } + known, err := store.Load(opts.state) + if err != nil { + return "", "", err + } + report, updated, writeErr := apply.WriteFilled(ctx, built, d, known, path, content, apply.ExecRunner, + announceOr(say), apply.KeepIn(filepath.Dir(opts.state)), apply.WindowsIn(filepath.Dir(opts.state))) + // Saved whichever way it went, as an apply's: what was written is on the machine either way. + if err := store.Save(opts.state, updated); err != nil { + return "", "", errors.Join(writeErr, errors.New("written, and the node's state could not be saved: "+ + err.Error())) + } + action, digest := "", "" + for _, o := range report.Outcomes { + if o.Target == path && o.Type == string(declaration.TypeFile) { + action = o.Action + } + } + if rec, ok := updated.At(string(declaration.TypeFile), path); ok { + digest = rec.Wrote + } + return action, digest, writeErr + }, + } +} diff --git a/cmd/mesh-host/filling_test.go b/cmd/mesh-host/filling_test.go new file mode 100644 index 0000000..828682a --- /dev/null +++ b/cmd/mesh-host/filling_test.go @@ -0,0 +1,105 @@ +package main + +import ( + "context" + "crypto/ed25519" + "encoding/json" + "os" + "os/user" + "path/filepath" + "testing" + "time" + + "github.com/novox/mesh-host/internal/filling" + "github.com/novox/mesh-host/internal/store" +) + +// Defends novox/hq ADR 0349 §4 (issue 668): a write of a filled file and an apply are one at a time — the write +// takes the lock the apply holds, and writes only once the apply let it go — and the write is recorded in the +// node's state as an apply's outcome is. +func TestAWriteAndAnApplyOfTheSameFileSerialise(t *testing.T) { + was := builtFor + builtFor = "arch" + t.Cleanup(func() { builtFor = was }) + me, err := user.Current() + if err != nil { + t.Fatal(err) + } + if me.Uid == "0" { + t.Skip("run as root, whose account is never the provisioner's") + } + dir := t.TempDir() + opts := options{state: filepath.Join(dir, "state.json")} + public, private, err := ed25519.GenerateKey(nil) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(dir, "m.conf") + raw, err := json.Marshal(map[string]any{"declaration": 1, "provisioner_account": me.Username, + "filled_files": []string{path}, "resources": []any{map[string]any{"id": "m.conf", "type": "file", + "path": path, "content": "from the declaration\n", "owner-at-runtime": "owned", "owner": me.Username}}}) + if err != nil { + t.Fatal(err) + } + if err := store.SaveDeclared(store.DeclaredPath(opts.state), + store.Declared{Declaration: raw, Signature: ed25519.Sign(private, raw)}); err != nil { + t.Fatal(err) + } + e := fillingEngine(opts, public, func(string) {}) + uid := uint32(os.Getuid()) + if err := e.Accepts(uid); err != nil { + t.Fatal(err) + } + + // An apply holds the lock. + applying.Lock() + done := make(chan filling.Answer) + go func() { + done <- e.Write(context.Background(), uid, filling.Request{Path: path, Content: []byte("filled\n")}) + }() + select { + case a := <-done: + applying.Unlock() + t.Fatalf("written while an apply held the lock: %+v", a) + case <-time.After(200 * time.Millisecond): + } + if _, err := os.Stat(path); err == nil { + applying.Unlock() + t.Fatal("the file was written while an apply held the lock") + } + applying.Unlock() + a := <-done + if a.Refused != "" || a.Failed != "" || a.Action != "created" { + t.Fatalf("the write after the apply: %+v", a) + } + got, err := os.ReadFile(path) + if err != nil || string(got) != "filled\n" { + t.Fatalf("the file holds %q (%v)", got, err) + } + known, err := store.Load(opts.state) + if err != nil { + t.Fatal(err) + } + if rec, ok := known.Find("m.conf"); !ok || rec.Wrote != a.Digest || rec.Target != path { + t.Errorf("the write is not recorded in the node's state: %+v, answered %q", rec, a.Digest) + } + + // And the cross-process lock beside the state too: a write waits for a `reconcile` run by hand. + unlock, err := store.Lock(opts.state, nil) + if err != nil { + t.Fatal(err) + } + go func() { + done <- e.Write(context.Background(), uid, filling.Request{Path: path, Content: []byte("again\n")}) + }() + select { + case a := <-done: + unlock() + t.Fatalf("written while another process held the apply's lock: %+v", a) + case <-time.After(200 * time.Millisecond): + } + unlock() + if a := <-done; a.Action != "updated" { + t.Fatalf("the second write: %+v", a) + } +} diff --git a/internal/declaration/provisioner_test.go b/internal/declaration/provisioner_test.go new file mode 100644 index 0000000..233aa40 --- /dev/null +++ b/internal/declaration/provisioner_test.go @@ -0,0 +1,37 @@ +package declaration + +import ( + "strings" + "testing" +) + +// Defends novox/hq ADR 0349 §4 (issue 668): the declaration names the account the provisioner's client runs as, +// the only one whose write of a filled file the node-engine accepts. Read here before any controller sends it. + +func TestTheDeclarationNamesTheProvisionersAccount(t *testing.T) { + d, err := Parse([]byte(`{"declaration":1,"provisioner_account":"mesh-node-provisioner","filled_files":["/etc/a"], + "resources":[{"id":"a","type":"file","path":"/etc/a","content":"a","owner-at-runtime":"owned"}]}`)) + if err != nil { + t.Fatal(err) + } + if d.Provisioner != "mesh-node-provisioner" { + t.Errorf("the provisioner's account read as %q", d.Provisioner) + } + d, err = Parse([]byte(`{"declaration":1,"resources":[{"id":"a","type":"file","path":"/etc/a","content":"a"}]}`)) + if err != nil { + t.Fatal(err) + } + if d.Provisioner != "" { + t.Errorf("a declaration naming no account names %q", d.Provisioner) + } +} + +func TestAProvisionersAccountThatIsNoAccountIsRefused(t *testing.T) { + for name, account := range map[string]string{"root": "root", "not a name": "a b", "a path": "../x"} { + _, err := Parse([]byte(`{"declaration":1,"provisioner_account":"` + account + `", + "resources":[{"id":"a","type":"file","path":"/etc/a","content":"a"}]}`)) + if err == nil || !strings.Contains(err.Error(), "provisioner_account") { + t.Errorf("%s: want a refusal naming provisioner_account, got %v", name, err) + } + } +} diff --git a/internal/meshcli/peer.go b/internal/meshcli/peer.go new file mode 100644 index 0000000..88397cd --- /dev/null +++ b/internal/meshcli/peer.go @@ -0,0 +1,19 @@ +package meshcli + +import "net" + +// The kernel's word on who is at the other end of a socket, shared with the node-engine's other local socket, the +// write of a filled file (internal/filling, novox/hq ADR 0349 §4): one reading of it, so the two never judge a peer +// differently. + +// PeerCredentials is the user id and process id of the process at the other end, as the kernel recorded them when +// it connected (SO_PEERCRED). Nothing the process says is read. +func PeerCredentials(conn *net.UnixConn) (uint32, int32, error) { return peerCredentials(conn) } + +// SameNamespaces refuses a process whose mount or user namespace is not the engine's: a user id is this machine's +// account only in this machine's namespaces. +func SameNamespaces(pid int32) error { return namespacesOf(pid) } + +// NotOwnDirectory says what is wrong with a socket's directory, or nothing: it must be a directory, not a link, +// owned by the account the engine runs as. +func NotOwnDirectory(dir string) string { return notOwnDirectory(dir) } From 1d57bcd6f94a68cb380993e7aacffdb1eee18936 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 00:36:04 +0000 Subject: [PATCH 3/7] Write a filled file through the apply, and move what reads it A filled file is the declared file with the client's content: applied through applyFile, applyInto or applyBlock by who owns it at runtime, with the apply's guards, Keep and record, then restart-on, reload-on and container reads applied as an apply does. Issue 668. Co-Authored-By: Claude Opus 5.5 --- internal/apply/filled.go | 203 ++++++++++++++++++++++++++++ internal/apply/filled_test.go | 247 ++++++++++++++++++++++++++++++++++ 2 files changed, 450 insertions(+) create mode 100644 internal/apply/filled.go create mode 100644 internal/apply/filled_test.go diff --git a/internal/apply/filled.go b/internal/apply/filled.go new file mode 100644 index 0000000..a0d750f --- /dev/null +++ b/internal/apply/filled.go @@ -0,0 +1,203 @@ +package apply + +import ( + "context" + "fmt" + "slices" + + "github.com/novox/mesh-host/internal/declaration" + "github.com/novox/mesh-host/internal/store" + "github.com/novox/mesh-host/internal/system" +) + +// WriteFilled writes what the provisioner's client filled into one file the declaration names as filled (novox/hq +// ADR 0349 §4), and makes what reads the file take the change, exactly as an apply that changed it would. +// +// **The declared file says everything but the content.** Its path, mode, owner and who owns it at runtime are the +// declaration's; the client's request carries the content alone. So the file is applied as the declared one with +// that content in place of the declaration's: an owned file whole through applyFile, a seed only when absent, a +// managed file's keys through applyInto or its block through applyBlock — the same guards on where nothing is +// placed, the same Keep of an original found there, the same record. What the declaration seals or fills with +// `${secret:}` the client has already filled, so the content is written as it came; a file that carried a secret +// keeps the mode a secret is written with when the declaration gives none. +// +// **Then what reads it, in the apply's order** (afterWhatTheyRead): a service naming it under restart-on is +// restarted and under reload-on reloaded, a process or container naming it under restart-on is restarted, and a +// running container that reads it at creation — an env-file, a file mounted into it — is recreated, because what +// it read is recorded by digest and this write changed the digest. Each of those is applied and recorded as an +// apply does, and what moves goes on moving what names it. Nothing else of the declaration is touched. +// +// The caller holds the apply's lock: a write and an apply never write one file at once. log is said the path, the +// action and the digest, never the content. +func WriteFilled( + ctx context.Context, + sys system.System, + d *declaration.Declaration, + known store.State, + path string, + content []byte, + run Runner, + log func(string), + keep Keep, + windows *Windows, +) (report Report, _ store.State, _ error) { + if log == nil { + log = func(string) {} + } + if !d.Fills(path) { + return report, known, &Error{Resource: path, Err: fmt.Errorf("this node's declaration does not name %s "+ + "as a file the provisioner fills, so nothing was written", path)} + } + var declared []*declaration.File + for _, r := range d.Resources { + if f, ok := r.(*declaration.File); ok && f.Path == path { + declared = append(declared, f) + } + } + if len(declared) != 1 { + // The parser refuses a filled path no file declares; two at one path would need the request to say whose + // keys or block it carries, and it carries the content only. + return report, known, &Error{Resource: path, Err: fmt.Errorf("%d resources declare %s, and a filled file "+ + "is written for exactly one, so nothing was written", len(declared), path)} + } + file := filledAs(declared[0], content) + if held, isHeld := known.HeldAt(file.ID); isHeld { + return report, known, &Error{Resource: file.ID, Err: fmt.Errorf("%s is held as found for a module not "+ + "yet taken (%s), so nothing was written", path, held.Why)} + } + + in := inputs{declares: map[string]string{}, known: &known, windows: windows, away: newStoreAway(windows, log), + groups: groupsWanted(d.Resources), handed: Handed{}} + refreshRuntimeRoots(ctx) + in.refused = refusedPlaces(d.Resources) + in.agentHomes = rootNeverHomes(d) + in.dirsBefore = map[string]bool{} + for _, r := range d.Resources { + in.declares[r.Identity()] = declaredDigest(r) + } + changed := map[string]bool{} + + // The file, as the apply loop applies it. + was, _ := known.Find(file.ID) + var keepFound Keep + if (was.ID == "" || was.Target != file.Path) && !known.Recorded(string(declaration.TypeFile), file.Path) { + keepFound = keep + } + outcome, err := applyOne(ctx, sys, file, run, changed, in, was, nil, keepFound) + if err != nil { + log(fmt.Sprintf(" failed %s (%s): %v", file.ID, file.Path, err)) + return report, known, &Error{Resource: file.ID, Err: err, Done: report} + } + recordKeeping(&known, file, outcome, was) + report.Outcomes = append(report.Outcomes, outcome) + if !moved(outcome) { + return report, known, nil + } + changed[file.ID] = true + line := fmt.Sprintf(" %s %s (%s), filled: %s", outcome.Action, file.ID, file.Path, outcome.wrote) + if outcome.Detail != "" { + line += ": " + outcome.Detail + } + log(line) + + // What reads it, in the apply's order, each moving what names it in turn. + var failures []*Error + for _, r := range afterWhatTheyRead(d.Resources) { + if r.Identity() == file.ID || !readsChange(r, path, changed) { + continue + } + if _, isHeld := known.HeldAt(r.Identity()); isHeld { + log(fmt.Sprintf(" held %s (%s): held as found for a module not yet taken, so it was not restarted "+ + "for %s", r.Identity(), r.Target(), file.ID)) + continue + } + was, _ := known.Find(r.Identity()) + o, err := applyOne(ctx, sys, r, run, changed, in, was, nil, nil) + if err != nil { + if o.found != nil { + known.KeepFound(r.Identity(), store.OriginDeclared, *o.found) + } + failures = append(failures, &Error{Resource: r.Identity(), Err: err}) + log(fmt.Sprintf(" failed %s (%s): %v", r.Identity(), o.Target, err)) + continue + } + if o.Action == "waiting" { + if o.found != nil { + known.KeepFound(r.Identity(), store.OriginDeclared, *o.found) + } + report.Outcomes = append(report.Outcomes, o) + log(fmt.Sprintf(" waiting %s (%s): %s", o.ID, o.Target, o.Detail)) + continue + } + recordKeeping(&known, r, o, was) + report.Outcomes = append(report.Outcomes, o) + if o.busy != nil { + report.Busy = append(report.Busy, *o.busy) + } + if moved(o) { + changed[r.Identity()] = true + line := fmt.Sprintf(" %s %s (%s)", o.Action, o.ID, o.Target) + if o.Detail != "" { + line += ": " + o.Detail + } + log(line) + } else if o.Action != "unchanged" { + log(fmt.Sprintf(" %s %s (%s): %s", o.Action, o.ID, o.Target, o.Detail)) + } + } + if len(failures) > 0 { + first := failures[0] + first.Done = report + first.Others = len(failures) - 1 + return report, known, first + } + return report, known, nil +} + +// filledAs is the declared file with the filled content in place of the declaration's. +func filledAs(declared *declaration.File, content []byte) *declaration.File { + f := *declared + f.Content, f.Sealed, f.Bytes, f.Secrets = string(content), "", "", nil + if f.Mode == "" && (declared.Secret() || len(declared.Secrets) > 0) { + // Written as the declaration would have written it: a file carrying a credential is not world-readable. + f.Mode = "0600" + } + return &f +} + +// readsChange reports whether applying r is what makes it take a change of what changed: a service's restart-on or +// reload-on, a process's or container's restart-on, or a running container that reads path at creation. +func readsChange(r declaration.Resource, path string, changed map[string]bool) bool { + switch res := r.(type) { + case *declaration.Service: + return len(restartedBy(res.RestartOn, changed)) > 0 || len(restartedBy(res.ReloadOn, changed)) > 0 + case *declaration.Process: + return len(restartedBy(res.RestartOn, changed)) > 0 + case *declaration.Container: + if len(restartedBy(res.RestartOn, changed)) > 0 { + return true + } + if res.RunOnce || res.Schedule != "" { + return false + } + if slices.Contains(res.EnvFile, path) { + return true + } + for _, v := range res.Volumes { + if mountSource(v) == path { + return true + } + } + } + return false +} + +// recordKeeping records r as applied, carrying where the original it replaced was kept from its record when this +// write kept none — for as long as the resource is recorded at the same target, as the apply loop carries it. +func recordKeeping(known *store.State, r declaration.Resource, o Outcome, was store.Applied) { + kept, keptMode, keptOwner := o.kept, o.keptMode, o.keptOwner + if kept == "" && was.Target == o.Target { + kept, keptMode, keptOwner = was.Kept, was.KeptMode, was.KeptOwner + } + recordApplied(known, store.OriginDeclared, r, o, kept, keptMode, keptOwner) +} diff --git a/internal/apply/filled_test.go b/internal/apply/filled_test.go new file mode 100644 index 0000000..b2e9003 --- /dev/null +++ b/internal/apply/filled_test.go @@ -0,0 +1,247 @@ +package apply + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/novox/mesh-host/internal/declaration" + "github.com/novox/mesh-host/internal/store" +) + +// Defends novox/hq ADR 0349 §4 (issue 668): the provisioner's client has the node-engine write a file it filled, +// and the node-engine writes it as an apply of the declared file with that content would — through applyFile, +// applyInto or applyBlock by who owns it at runtime — and makes what reads it take the change as an apply does. + +// filledDecl is a declaration naming path as filled, with the file's fields after id, type and path, and more +// resources after it. +func filledDecl(t *testing.T, path, fields, more string) *declaration.Declaration { + t.Helper() + if more != "" { + more = "," + more + } + return parseTrusted(t, fmt.Sprintf(`{"declaration":1,"provisioner_account":"mesh-node-provisioner", + "filled_files":[%q],"resources":[{"id":"m.conf","type":"file","path":%q,%s}%s]}`, path, path, fields, more)) +} + +func TestAWriteOfAPathNotFilledIsRefused(t *testing.T) { + dir := t.TempDir() + filled := filepath.Join(dir, "filled.conf") + other := filepath.Join(dir, "other.conf") + d := parseTrusted(t, fmt.Sprintf(`{"declaration":1,"filled_files":[%q],"resources":[ + {"id":"m.conf","type":"file","path":%q,"content":"x\n","owner-at-runtime":"owned"}, + {"id":"m.other","type":"file","path":%q,"content":"y\n"} + ]}`, filled, filled, other)) + for _, path := range []string{other, filepath.Join(dir, "nothing.conf")} { + _, state, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("evil\n"), + noServices, nil, nil, nil) + if err == nil || !strings.Contains(err.Error(), "does not name") { + t.Errorf("%s: a write of a path the declaration does not fill was not refused: %v", path, err) + } + if _, err := os.Stat(path); err == nil && path != other { + t.Errorf("%s was written", path) + } + if len(state.Resources) != 0 { + t.Errorf("a refused write recorded something: %+v", state.Resources) + } + } + if _, err := os.Stat(other); err == nil { + t.Errorf("%s was written", other) + } +} + +// Each ownership is written by the write verb exactly as an apply of the declared file with the same content: the +// same action, the same bytes on disk, the same mode. +func TestAFilledFileIsWrittenAsItsOwnershipSays(t *testing.T) { + for _, c := range []struct { + name, fields, content, before, action string + }{ + {"owned, absent", `"content":"","mode":"0640","owner-at-runtime":"owned"`, "port = 8080\n", "", "created"}, + {"owned, over what is there", `"content":"","mode":"0640","owner-at-runtime":"owned"`, "port = 8080\n", + "port = 1\n", "updated"}, + {"seed, absent", `"content":"","mode":"0600","owner-at-runtime":"seed"`, "user default on\n", "", "created"}, + {"seed, present", `"content":"","mode":"0600","owner-at-runtime":"seed"`, "user default on\n", + "user default on\nuser app-one on\n", "kept"}, + {"managed json", `"into":"json","content":"{}","owner-at-runtime":"managed"`, + `{"insecure-registries":["10.42.0.1:5000"]}`, machinesOwn, "updated"}, + {"managed block", `"into":"block","content":"x\n","owner-at-runtime":"managed"`, "10.42.0.1 ace\n", + workstationHosts, "updated"}, + } { + t.Run(c.name, func(t *testing.T) { + at := func() string { + path := filepath.Join(t.TempDir(), "the-file") + if c.before != "" { + if err := os.WriteFile(path, []byte(c.before), 0o644); err != nil { + t.Fatal(err) + } + } + return path + } + written := at() + report, state, err := WriteFilled(context.Background(), archHost(t), filledDecl(t, written, c.fields, ""), + store.State{}, written, []byte(c.content), noServices, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + applied := at() + withContent := strings.Replace(c.fields, fmt.Sprintf(`"content":%q`, contentOf(t, c.fields)), + fmt.Sprintf(`"content":%q`, c.content), 1) + want, _, err := Apply(context.Background(), archHost(t), filledDecl(t, applied, withContent, ""), + store.State{}, store.OriginDeclared, noServices, nil, nil) + if err != nil { + t.Fatal(err) + } + if got := report.Outcomes[0].Action; got != c.action || got != want.Outcomes[0].Action { + t.Errorf("written %s; an apply of the same content %s; want %s", got, want.Outcomes[0].Action, c.action) + } + if got, apply := readText(t, written), readText(t, applied); got != apply { + t.Errorf("written, the file holds\n%q\nan apply of the same content leaves\n%q", got, apply) + } + if a, b := modeAt(t, written), modeAt(t, applied); a != b { + t.Errorf("written mode %o, an apply's %o", a, b) + } + if rec, ok := state.Find("m.conf"); !ok || rec.Target != written || rec.Origin != store.OriginDeclared { + t.Errorf("the write is not recorded as the file's: %+v", rec) + } + }) + } +} + +func contentOf(t *testing.T, fields string) string { + t.Helper() + var holder struct { + Content string `json:"content"` + } + if err := json.Unmarshal([]byte("{"+fields+"}"), &holder); err != nil { + t.Fatal(err) + } + return holder.Content +} + +func modeAt(t *testing.T, path string) os.FileMode { + t.Helper() + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + return info.Mode().Perm() +} + +// A file that carried a secret keeps the mode a secret is written with: the client filled the value, and the file +// is no more readable for it. +func TestAFilledFileThatCarriedASecretIsNotWorldReadable(t *testing.T) { + path := filepath.Join(t.TempDir(), "token.env") + d := filledDecl(t, path, `"content":"TOKEN=${secret:token}\n","secrets":{"token":"c2VhbGVk"},"owner-at-runtime":"owned"`, "") + if _, _, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("TOKEN=abc\n"), + noServices, nil, nil, nil); err != nil { + t.Fatal(err) + } + if m := modeAt(t, path); m != 0o600 { + t.Errorf("a filled file that carried a secret is mode %o", m) + } +} + +func TestAWriteUnderReloadOnReloadsItsProgramOnce(t *testing.T) { + path := filepath.Join(t.TempDir(), "daemon.conf") + d := filledDecl(t, path, `"content":"","owner-at-runtime":"owned"`, + `{"id":"m.daemon","type":"service","unit":"m.service","state":"running","reload-on":["m.conf"]}, + {"id":"m.other","type":"service","unit":"other.service","state":"running"}`) + var commands []string + var said []string + _, state, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("level = 1\n"), + recordingServices(&commands), func(line string) { said = append(said, line) }, nil, nil) + if err != nil { + t.Fatal(err) + } + if n := count(commands, "systemctl reload m.service"); n != 1 { + t.Errorf("the program was reloaded %d times; commands were %v", n, commands) + } + joined := strings.Join(commands, "\n") + if strings.Contains(joined, "restart m.service") || strings.Contains(joined, "stop m.service") { + t.Errorf("the program was restarted, not reloaded: %v", commands) + } + if strings.Contains(joined, "other.service") { + t.Errorf("a service that does not read the file was touched: %v", commands) + } + if _, ok := state.Find("m.daemon"); !ok { + t.Errorf("the reloaded service was not recorded") + } + // The log says the path and the digest, never the content. + for _, line := range said { + if strings.Contains(line, "level = 1") { + t.Errorf("the content was said: %q", line) + } + } + if !strings.Contains(strings.Join(said, "\n"), digestOf("level = 1\n")) { + t.Errorf("the digest was not said: %q", said) + } + + // The same content again moves nothing, and reloads nothing. + commands = nil + report, _, err := WriteFilled(context.Background(), archHost(t), d, state, path, []byte("level = 1\n"), + recordingServices(&commands), nil, nil, nil) + if err != nil { + t.Fatal(err) + } + if report.Outcomes[0].Action != "unchanged" || count(commands, "reload") != 0 { + t.Errorf("an unchanged write reloaded: %v %+v", commands, report.Outcomes) + } +} + +func TestAWriteUnderRestartOnRestartsItsProgram(t *testing.T) { + path := filepath.Join(t.TempDir(), "daemon.conf") + d := filledDecl(t, path, `"content":"","owner-at-runtime":"owned"`, + `{"id":"m.daemon","type":"service","unit":"m.service","state":"running","restart-on":["m.conf"]}`) + var commands []string + if _, _, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("level = 1\n"), + recordingServices(&commands), nil, nil, nil); err != nil { + t.Fatal(err) + } + // A restart is a stop and a start, once. + stops, starts := count(commands, "systemctl stop m.service"), count(commands, "systemctl start m.service") + if stops != 1 || starts != 1 { + t.Errorf("the program was stopped %d and started %d times; commands were %v", stops, starts, commands) + } +} + +// A running container that reads the file at creation is recreated, as an apply recreates it for a changed file. +func TestAWriteOfAFileAContainerReadsRecreatesIt(t *testing.T) { + env := filepath.Join(t.TempDir(), "forge.env") + d := filledDecl(t, env, `"content":"","mode":"0600","owner-at-runtime":"owned"`, + `{"id":"forge.server","type":"container","name":"forge","image":"`+pinned+`","env-file":["`+env+`"]}`) + m := &machine{containers: map[string]*fakeContainer{}} + _, state, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, env, []byte("PORT=5432\n"), + m.run, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + if !m.did("docker run") { + t.Fatalf("the container was not created: %v", m.asked) + } + m.asked = nil + report, _, err := WriteFilled(context.Background(), archHost(t), d, state, env, []byte("PORT=5433\n"), m.run, + nil, nil, nil) + if err != nil { + t.Fatal(err) + } + if !m.removed("forge") || !m.did("docker run") { + t.Fatalf("the container kept what it read before the write: %v", m.asked) + } + if o := outcomeOf(report, "forge.server"); o.Action != "updated" || !strings.Contains(o.Detail, env) { + t.Errorf("the recreate did not say which file changed: %+v", o) + } +} + +func count(lines []string, prefix string) int { + n := 0 + for _, l := range lines { + if strings.Contains(l, prefix) { + n++ + } + } + return n +} From 499a687149beabb115c596e1d77eb1e30f6b2864 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 00:37:29 +0000 Subject: [PATCH 4/7] Serve the local write of a filled file on its own socket A second unix socket beside mesh-cli's: the kernel's word on the asking account (SO_PEERCRED, same namespaces), that account judged against the declaration before its request is read, the path judged clean, absolute and filled, and the write done under the apply's lock. The journal has the path and digest, never the content. Issue 668. Co-Authored-By: Claude Opus 5.5 --- internal/filling/filling.go | 347 +++++++++++++++++++++++++++++++ internal/filling/filling_test.go | 263 +++++++++++++++++++++++ 2 files changed, 610 insertions(+) create mode 100644 internal/filling/filling.go create mode 100644 internal/filling/filling_test.go diff --git a/internal/filling/filling.go b/internal/filling/filling.go new file mode 100644 index 0000000..8e13902 --- /dev/null +++ b/internal/filling/filling.go @@ -0,0 +1,347 @@ +// Package filling is the node-engine's narrow local write verb (novox/hq ADR 0349 §4, design 57 P2): a socket on +// this machine through which the provisioner's client has a file it filled written, and nothing else. +// +// **The client fills; the node-engine writes.** The client renders a module's template with the values each domain +// owns and hands the node-engine the content; the node-engine writes it as the declared file, with the declared +// file's path, owner, mode and ownership at runtime, and makes what reads it take the change. So the one process that +// writes the machine's files is still the node-engine, and the client needs no right to write any of them. +// +// **Narrow by three checks, each from something the client cannot say for itself.** The account asking, from the +// kernel (SO_PEERCRED), in this machine's own namespaces, must be the account the node's current declaration names +// as the provisioner's client's; the path, clean and absolute, must be one the declaration names as filled; and the +// write takes the apply's lock, so a write and an apply never write one file at once. While the declaration names no +// account — every declaration before the provisioner's client has one — every write is refused. +// +// **No listening control surface** (novox/hq ADR 0004): the socket has no network address. And the content is never +// said: the journal has the path, the account and the content's digest. +package filling + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "os" + "os/user" + "path/filepath" + "strconv" + "sync" + "time" + + "github.com/novox/mesh-host/internal/declaration" + "github.com/novox/mesh-host/internal/meshcli" +) + +// Socket is where the node-engine accepts a write of a filled file, beside mesh-cli's socket. +const Socket = "/run/nox-mesh-host/fill.sock" + +// MaxRequest bounds one request, its content encoded in it: a configuration file, never an archive. +const MaxRequest = 8 << 20 + +// Request is one write, sent as one JSON object on one line: the path of a filled file and what the client filled +// into it (base64 in the JSON). The content only: the declaration says everything else about the file. +type Request struct { + Path string `json:"path"` + Content []byte `json:"content"` +} + +// Answer is what came of one request. Refused is a write that was not attempted, with why; Failed one that was +// attempted and did not complete, with why — the file, or something that reads it. Otherwise Action is what the +// write did to the file (created, updated, corrected, unchanged, kept) and Digest what the file holds now, as the +// node-engine records it. +type Answer struct { + Path string `json:"path,omitempty"` + Action string `json:"action,omitempty"` + Digest string `json:"digest,omitempty"` + Refused string `json:"refused,omitempty"` + Failed string `json:"failed,omitempty"` +} + +// Refusal is an answer that nothing was written, and why. +func Refusal(path, why string) Answer { return Answer{Path: path, Refused: why} } + +// Writer is the node-engine's side of the socket. +type Writer interface { + // Accepts judges the asking account before its request is read: nil when the current declaration names it as + // the provisioner's client's. + Accepts(uid uint32) error + // Write judges the request and writes it, holding the apply's lock. + Write(ctx context.Context, uid uint32, req Request) Answer +} + +// Engine is the Writer the node-engine serves: the declaration it applies now, read again under the apply's lock. +type Engine struct { + // Declared reads the declaration this node applies now, verified. + Declared func() (*declaration.Declaration, error) + // UIDOf is the user id of an account, by name, from this machine's user database. + UIDOf func(account string) (uint32, error) + // Lock takes the apply's lock and returns what releases it: the same the apply takes, so a write and an apply + // are one at a time. + Lock func() (func(), error) + // Place writes the content into the filled file at path, as the declared file, and makes what reads it take the + // change. Called holding the lock, with the declaration read under it. + Place func(ctx context.Context, d *declaration.Declaration, path string, content []byte) (action, digest string, + err error) +} + +// Accepts reads the declaration and judges the account. +func (e *Engine) Accepts(uid uint32) error { + d, err := e.Declared() + if err != nil { + return fmt.Errorf("this node's declaration cannot be read, so no write is accepted: %w", err) + } + return JudgeAccount(d, uid, e.UIDOf) +} + +// Write takes the apply's lock, reads the declaration under it, judges the account and the path again — the +// declaration may have changed since the request was accepted — and writes. +func (e *Engine) Write(ctx context.Context, uid uint32, req Request) Answer { + if err := JudgeShape(req.Path); err != nil { + return Refusal(req.Path, err.Error()) + } + unlock, err := e.Lock() + if err != nil { + return Refusal(req.Path, "the apply's lock could not be taken, so nothing was written: "+err.Error()) + } + defer unlock() + d, err := e.Declared() + if err != nil { + return Refusal(req.Path, "this node's declaration cannot be read, so nothing was written: "+err.Error()) + } + if err := JudgeAccount(d, uid, e.UIDOf); err != nil { + return Refusal(req.Path, err.Error()) + } + if err := JudgePath(d, req.Path); err != nil { + return Refusal(req.Path, err.Error()) + } + action, digest, err := e.Place(ctx, d, req.Path, req.Content) + if err != nil { + return Answer{Path: req.Path, Action: action, Digest: digest, Failed: err.Error()} + } + return Answer{Path: req.Path, Action: action, Digest: digest} +} + +// JudgeAccount refuses every account but the one the declaration names as the provisioner's client's, by its user +// id here — and every account while the declaration names none. +func JudgeAccount(d *declaration.Declaration, uid uint32, uidOf func(string) (uint32, error)) error { + if d.Provisioner == "" { + return errors.New("this node's declaration names no provisioner client's account, so no write is accepted") + } + want, err := uidOf(d.Provisioner) + if err != nil { + return fmt.Errorf("the provisioner client's account %s is not on this machine, so no write is accepted: %w", + d.Provisioner, err) + } + if want == 0 { + return fmt.Errorf("the provisioner client's account %s is root here, so no write is accepted", d.Provisioner) + } + if uid != want { + return fmt.Errorf("user id %d is not the provisioner client's account %s (user id %d), so nothing was written", + uid, d.Provisioner, want) + } + return nil +} + +// JudgeShape refuses a path that is not clean and absolute: a relative path, `..`, a doubled or trailing slash. +// Compared as written with the declaration's, a path that is not clean could name a file the declaration does not. +func JudgeShape(path string) error { + switch { + case path == "": + return errors.New("the request names no path, so nothing was written") + case !filepath.IsAbs(path): + return fmt.Errorf("%q is not an absolute path, so nothing was written", path) + case filepath.Clean(path) != path: + return fmt.Errorf("%q is not a clean path (it is %q cleaned), so nothing was written", path, + filepath.Clean(path)) + } + return nil +} + +// JudgePath refuses a path that is not clean and absolute, or that the declaration does not name as filled. Where it +// leads — a link on the way, a place nothing is written — is judged by the write, by the apply's own rules. +func JudgePath(d *declaration.Declaration, path string) error { + if err := JudgeShape(path); err != nil { + return err + } + if !d.Fills(path) { + return fmt.Errorf("%s is not a file this node's declaration names as filled, so nothing was written", path) + } + return nil +} + +// UIDOf is the user id of an account by name, from this machine's user database. +func UIDOf(account string) (uint32, error) { + u, err := user.Lookup(account) + if err != nil { + return 0, err + } + id, err := strconv.ParseUint(u.Uid, 10, 32) + if err != nil { + return 0, fmt.Errorf("user id %q: %w", u.Uid, err) + } + return uint32(id), nil +} + +// peerOf and sameNamespaces are the kernel's word on the asking process, as mesh-cli's socket reads it. Variables so +// a test can say. +var ( + peerOf = meshcli.PeerCredentials + 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. +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 + +// 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. +func Serve(ctx context.Context, socket string, w Writer, say func(string)) error { + if say == nil { + say = func(string) {} + } + dir := filepath.Dir(socket) + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("the filled files' socket directory: %w", err) + } + if problem := meshcli.NotOwnDirectory(dir); problem != "" { + return fmt.Errorf("filled files are not written on request on this machine: %s", problem) + } + 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 "+ + "files are not written on request until it is moved", socket) + } + if err := os.Remove(socket); err != nil { + return fmt.Errorf("the socket a previous engine left at %s: %w", socket, err) + } + } + l, err := net.ListenUnix("unix", &net.UnixAddr{Name: socket, Net: "unix"}) + if err != nil { + return fmt.Errorf("listening for filled files at %s: %w", socket, err) + } + if err := os.Chmod(socket, 0o666); err != nil { + l.Close() + return fmt.Errorf("opening the filled files' socket: %w", err) + } + go func() { + <-ctx.Done() + l.Close() + }() + slots := make(chan struct{}, atOnce) + var wg sync.WaitGroup + defer wg.Wait() + for { + conn, err := l.AcceptUnix() + if err != nil { + if ctx.Err() != nil { + return nil + } + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + continue + } + return fmt.Errorf("the filled files' socket: %w", err) + } + readBy := time.Now().Add(within) + _ = conn.SetDeadline(readBy) + wg.Add(1) + go func() { + defer wg.Done() + select { + case slots <- struct{}{}: + defer func() { <-slots }() + answer(ctx, conn, readBy, w, say) + default: + refuse(conn, readBy, Refusal("", "busy: other writes are being served; ask again"), say) + } + }() + } +} + +// answer serves one connection. +func answer(ctx context.Context, conn *net.UnixConn, readBy time.Time, w Writer, say func(string)) { + 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 + } + 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 + } + 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 + } + var req Request + if err := json.NewDecoder(io.LimitReader(conn, MaxRequest)).Decode(&req); err != nil { + reply(conn, Refusal("", "the request could not be read, so nothing was written: "+err.Error()), say) + return + } + // The write waits for the apply's lock as long as an apply takes, and the client waits with it. + _ = conn.SetDeadline(time.Time{}) + a := w.Write(ctx, uid, req) + switch { + case a.Refused != "": + say(fmt.Sprintf("a write of %s from user id %d was refused: %s", req.Path, uid, a.Refused)) + case a.Failed != "": + say(fmt.Sprintf("a write of %s from user id %d failed: %s", req.Path, uid, a.Failed)) + default: + say(fmt.Sprintf("wrote %s for the provisioner's client (user id %d): %s, %s", req.Path, uid, a.Action, + a.Digest)) + } + _ = conn.SetWriteDeadline(time.Now().Add(within)) + 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 + } + _ = conn.SetReadDeadline(by) + drain(conn) + _ = conn.SetWriteDeadline(time.Now().Add(within)) + reply(conn, a, say) +} + +// drain reads and drops a request up to its end — the newline after it, the end of the stream or the deadline — +// and at most MaxRequest bytes, keeping none of it. +func drain(r io.Reader) { + var buf [4096]byte + for read := 0; read < MaxRequest; { + n, err := r.Read(buf[:]) + read += n + if bytes.IndexByte(buf[:n], '\n') >= 0 || err != nil { + return + } + } +} + +func reply(conn *net.UnixConn, a Answer, say func(string)) { + defer conn.Close() + if err := json.NewEncoder(conn).Encode(a); err != nil { + say("a filled file's answer could not be written back: " + err.Error()) + } +} diff --git a/internal/filling/filling_test.go b/internal/filling/filling_test.go new file mode 100644 index 0000000..52a6f95 --- /dev/null +++ b/internal/filling/filling_test.go @@ -0,0 +1,263 @@ +package filling + +import ( + "context" + "encoding/json" + "errors" + "net" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/novox/mesh-host/internal/declaration" +) + +// Defends novox/hq ADR 0349 §4 (issue 668): the node-engine writes a filled file on request only for the provisioner +// client's account, only for a clean path the current declaration names as filled, and one at a time with the apply. + +const provisionerUID = 4242 + +func declared(t *testing.T, raw string) *declaration.Declaration { + t.Helper() + d, err := declaration.Parse([]byte(raw)) + if err != nil { + t.Fatal(err) + } + return d +} + +const withAccount = `{"declaration":1,"provisioner_account":"mesh-node-provisioner","filled_files":["/etc/m/m.conf"], + "resources":[{"id":"m.conf","type":"file","path":"/etc/m/m.conf","content":"x","owner-at-runtime":"owned"}, + {"id":"m.other","type":"file","path":"/etc/m/other.conf","content":"y"}]}` + +const withoutAccount = `{"declaration":1,"filled_files":["/etc/m/m.conf"], + "resources":[{"id":"m.conf","type":"file","path":"/etc/m/m.conf","content":"x","owner-at-runtime":"owned"}]}` + +// engine is an Engine over a fixed declaration, recording what it was asked to place. +type engine struct { + Engine + mu sync.Mutex + placed []string +} + +func newEngine(t *testing.T, raw string, lock sync.Locker) *engine { + d := declared(t, raw) + e := &engine{} + e.Engine = Engine{ + Declared: func() (*declaration.Declaration, error) { return d, nil }, + UIDOf: func(name string) (uint32, error) { + if name == "mesh-node-provisioner" { + return provisionerUID, nil + } + return 0, errors.New("no such account") + }, + Lock: func() (func(), error) { lock.Lock(); return lock.Unlock, nil }, + Place: func(_ context.Context, _ *declaration.Declaration, path string, content []byte) (string, string, error) { + e.mu.Lock() + defer e.mu.Unlock() + e.placed = append(e.placed, path) + return "updated", "digest", nil + }, + } + return e +} + +func TestAWriteFromTheProvisionersAccountOfAFilledPathIsPlaced(t *testing.T) { + e := newEngine(t, withAccount, &sync.Mutex{}) + if err := e.Accepts(provisionerUID); err != nil { + t.Fatal(err) + } + a := e.Write(context.Background(), provisionerUID, Request{Path: "/etc/m/m.conf", Content: []byte("x")}) + if a.Refused != "" || a.Failed != "" || a.Action != "updated" || len(e.placed) != 1 { + t.Fatalf("answer %+v, placed %v", a, e.placed) + } +} + +func TestAnUndeclaredPathIsRefused(t *testing.T) { + e := newEngine(t, withAccount, &sync.Mutex{}) + for _, path := range []string{"/etc/m/other.conf", "/etc/m/none.conf", "/etc/passwd"} { + a := e.Write(context.Background(), provisionerUID, Request{Path: path, Content: []byte("evil")}) + if a.Refused == "" || !strings.Contains(a.Refused, "not a file this node's declaration names as filled") { + t.Errorf("%s: %+v", path, a) + } + } + if len(e.placed) != 0 { + t.Errorf("placed %v", e.placed) + } +} + +func TestAnotherAccountIsRefused(t *testing.T) { + e := newEngine(t, withAccount, &sync.Mutex{}) + for _, uid := range []uint32{0, 1000, provisionerUID + 1} { + if err := e.Accepts(uid); err == nil { + t.Errorf("user id %d was accepted", uid) + } + a := e.Write(context.Background(), uid, Request{Path: "/etc/m/m.conf", Content: []byte("evil")}) + if a.Refused == "" || !strings.Contains(a.Refused, "is not the provisioner client's account") { + t.Errorf("user id %d: %+v", uid, a) + } + } + if len(e.placed) != 0 { + t.Errorf("placed %v", e.placed) + } +} + +func TestEveryWriteIsRefusedWhileTheDeclarationNamesNoAccount(t *testing.T) { + e := newEngine(t, withoutAccount, &sync.Mutex{}) + if err := e.Accepts(provisionerUID); err == nil || !strings.Contains(err.Error(), "names no provisioner") { + t.Errorf("accepted with no account named: %v", err) + } + a := e.Write(context.Background(), provisionerUID, Request{Path: "/etc/m/m.conf", Content: []byte("x")}) + if a.Refused == "" || len(e.placed) != 0 { + t.Errorf("written with no account named: %+v", a) + } +} + +func TestAnAccountThatIsRootHereIsRefused(t *testing.T) { + e := newEngine(t, withAccount, &sync.Mutex{}) + e.UIDOf = func(string) (uint32, error) { return 0, nil } + if err := e.Accepts(0); err == nil || !strings.Contains(err.Error(), "root") { + t.Errorf("an account that is root here was accepted: %v", err) + } +} + +func TestAnUncleanPathIsRefused(t *testing.T) { + e := newEngine(t, withAccount, &sync.Mutex{}) + for _, path := range []string{"", "etc/m/m.conf", "m.conf", "/etc//m/m.conf", "/etc/m/../m/m.conf", + "/etc/m/./m.conf", "/etc/m/m.conf/"} { + a := e.Write(context.Background(), provisionerUID, Request{Path: path, Content: []byte("x")}) + if a.Refused == "" { + t.Errorf("%q was not refused: %+v", path, a) + } + } + if len(e.placed) != 0 { + t.Errorf("placed %v", e.placed) + } +} + +// A write waits for an apply holding the apply's lock, and writes only once it is let go. +func TestAWriteWaitsForTheApply(t *testing.T) { + var applying sync.Mutex + e := newEngine(t, withAccount, &applying) + applying.Lock() + done := make(chan Answer) + go func() { + done <- e.Write(context.Background(), provisionerUID, Request{Path: "/etc/m/m.conf", Content: []byte("x")}) + }() + select { + case a := <-done: + t.Fatalf("written while the apply held the lock: %+v", a) + case <-time.After(100 * time.Millisecond): + } + e.mu.Lock() + n := len(e.placed) + e.mu.Unlock() + if n != 0 { + t.Fatalf("placed while the apply held the lock") + } + applying.Unlock() + if a := <-done; a.Action != "updated" { + t.Fatalf("after the apply: %+v", a) + } +} + +// fixed is a Writer that accepts one user id and answers every write. +type fixed struct { + uid uint32 + mu sync.Mutex + writes []Request +} + +func (f *fixed) Accepts(uid uint32) error { + if uid != f.uid { + return errors.New("not the provisioner client's account") + } + return nil +} + +func (f *fixed) Write(_ context.Context, _ uint32, req Request) Answer { + f.mu.Lock() + defer f.mu.Unlock() + f.writes = append(f.writes, req) + return Answer{Path: req.Path, Action: "updated", Digest: "d"} +} + +// serve serves w on a socket in a fresh directory and returns its path, and what was said. +func serve(t *testing.T, w Writer) (string, func() []string) { + t.Helper() + dir, err := os.MkdirTemp("", "fill") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + socket := filepath.Join(dir, "fill.sock") + ctx, cancel := context.WithCancel(context.Background()) + var mu sync.Mutex + var said []string + stopped := make(chan error) + go func() { + stopped <- Serve(ctx, socket, w, func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }) + }() + t.Cleanup(func() { + cancel() + if err := <-stopped; err != nil { + t.Error(err) + } + }) + for i := 0; i < 100; i++ { + if _, err := os.Stat(socket); err == nil { + break + } + time.Sleep(10 * time.Millisecond) + } + return socket, func() []string { mu.Lock(); defer mu.Unlock(); return append([]string(nil), said...) } +} + +func ask(t *testing.T, socket string, req Request) Answer { + t.Helper() + conn, err := net.Dial("unix", socket) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if err := json.NewEncoder(conn).Encode(req); err != nil { + t.Fatal(err) + } + var a Answer + if err := json.NewDecoder(conn).Decode(&a); err != nil { + t.Fatal(err) + } + return a +} + +// Over a real socket, the kernel says who is asking: this test's own account is the provisioner's here. +func TestTheSocketWritesForTheAccountTheKernelSays(t *testing.T) { + w := &fixed{uid: uint32(os.Getuid())} + socket, said := serve(t, w) + a := ask(t, socket, Request{Path: "/etc/m/m.conf", Content: []byte("secret value")}) + if a.Refused != "" || a.Action != "updated" || len(w.writes) != 1 || string(w.writes[0].Content) != "secret value" { + t.Fatalf("answer %+v, writes %+v", a, w.writes) + } + for _, line := range said() { + if strings.Contains(line, "secret value") { + t.Errorf("the content was said: %q", line) + } + } +} + +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")}) + if a.Refused == "" || len(w.writes) != 0 { + t.Fatalf("another account was written for: %+v", a) + } + for _, line := range said() { + if strings.Contains(line, "secret value") { + t.Errorf("the content was said: %q", line) + } + } +} From 76dc93c421bf9738a130dd3b87dd7b49366e9878 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 00:46:39 +0000 Subject: [PATCH 5/7] Hold a library's containers through the write verb, as a full apply does MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review 11959 of mesh-host#111: a write through the verb recreated a container that reads the filled file over a library whose disk was missing (ADR 0328 §9). Also refuses a filled path two resources declare, with a test. Issue 668. Co-Authored-By: Claude Opus 5.5 --- internal/apply/filled.go | 3 ++ internal/apply/filled_library_test.go | 57 +++++++++++++++++++++++++++ internal/apply/filled_test.go | 21 ++++++++++ internal/apply/library.go | 18 ++++++--- 4 files changed, 93 insertions(+), 6 deletions(-) create mode 100644 internal/apply/filled_library_test.go diff --git a/internal/apply/filled.go b/internal/apply/filled.go index a0d750f..f3f4800 100644 --- a/internal/apply/filled.go +++ b/internal/apply/filled.go @@ -72,6 +72,9 @@ func WriteFilled( in.refused = refusedPlaces(d.Resources) in.agentHomes = rootNeverHomes(d) in.dirsBefore = map[string]bool{} + // A container that reads the file and mounts a library whose disk or mount is not there is held, as a full apply + // holds it: not created over the missing disk (novox/hq ADR 0328 §9). + in.held = heldLibraries(ctx, d.Libraries) for _, r := range d.Resources { in.declares[r.Identity()] = declaredDigest(r) } diff --git a/internal/apply/filled_library_test.go b/internal/apply/filled_library_test.go new file mode 100644 index 0000000..a378c93 --- /dev/null +++ b/internal/apply/filled_library_test.go @@ -0,0 +1,57 @@ +//go:build linux + +package apply + +import ( + "context" + "fmt" + "path/filepath" + "strings" + "testing" + + "github.com/novox/mesh-host/internal/declaration" + "github.com/novox/mesh-host/internal/store" +) + +// A container that reads a filled file and mounts a library whose disk is not there is held by the write verb as a +// full apply holds it (novox/hq ADR 0328 §9, issue 668): never created or recreated over the missing disk. +func TestAWriteDoesNotRecreateAContainerHeldForItsLibrary(t *testing.T) { + // No folder is given the group: the library's root is absent, and nothing is made. + gid := 61000 + libraries(t, nil, nil) + base := t.TempDir() + root := filepath.Join(base, "gone", "media") + env := filepath.Join(base, "radarr.env") + raw := fmt.Sprintf(`{"declaration":1,"provisioner_account":"mesh-node-provisioner","filled_files":[%q], + "groups":[{"name":"mesh-media","gid":%d}],"libraries":[%s],"resources":[ + {"id":"radarr.env","type":"file","path":%q,"content":"","mode":"0600","owner-at-runtime":"owned"}, + {"id":"radarr.app","type":"container","name":"radarr","image":%q,"env-file":[%q],"volumes":[%q]}]}`, + env, gid, libJSON("media", root, gid, true, false, "movies"), env, pinned, env, + filepath.Join(root, "movies")+":/movies") + d, err := declaration.Parse([]byte(raw)) + if err != nil { + t.Fatal(err) + } + var ran []string + run := func(_ context.Context, name string, args ...string) (string, error) { + ran = append(ran, name+" "+strings.Join(args, " ")) + return "", nil + } + var state store.State + for _, content := range []string{"PORT=1\n", "PORT=2\n"} { + var report Report + report, state, err = WriteFilled(context.Background(), archHost(t), d, state, env, []byte(content), run, + nil, nil, nil) + if err != nil { + t.Fatal(err) + } + if o := outcomeOf(report, "radarr.app"); o.Action != "waiting" { + t.Errorf("after writing %q the container is %+v, not held for its library", content, o) + } + } + for _, line := range ran { + if strings.HasPrefix(line, "docker run") || strings.HasPrefix(line, "docker rm") { + t.Errorf("the container was created or removed over a missing disk: %v", ran) + } + } +} diff --git a/internal/apply/filled_test.go b/internal/apply/filled_test.go index b2e9003..6166d31 100644 --- a/internal/apply/filled_test.go +++ b/internal/apply/filled_test.go @@ -245,3 +245,24 @@ func count(lines []string, prefix string) int { } return n } + +// Two resources declaring one filled path — two modules' blocks in one file — are refused: the request carries the +// content only, so it cannot say whose block it is, and nothing is written. +func TestAFilledPathTwoResourcesDeclareIsRefused(t *testing.T) { + path := filepath.Join(t.TempDir(), "hosts") + if err := os.WriteFile(path, []byte(workstationHosts), 0o644); err != nil { + t.Fatal(err) + } + d := parseTrusted(t, fmt.Sprintf(`{"declaration":1,"filled_files":[%q],"resources":[ + {"id":"a.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.1 ace\n","owner-at-runtime":"managed"}, + {"id":"b.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.2 g14\n","owner-at-runtime":"managed"} + ]}`, path, path, path)) + _, state, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("6.6.6.6 evil\n"), + noServices, nil, nil, nil) + if err == nil || !strings.Contains(err.Error(), "2 resources declare") { + t.Fatalf("a path two resources declare was not refused: %v", err) + } + if got := readText(t, path); got != workstationHosts || len(state.Resources) != 0 { + t.Errorf("something was written or recorded: %q %+v", got, state.Resources) + } +} diff --git a/internal/apply/library.go b/internal/apply/library.go index a198813..36a53b3 100644 --- a/internal/apply/library.go +++ b/internal/apply/library.go @@ -318,6 +318,17 @@ func heldRoots(verdicts map[string]libraryVerdict, libs []declaration.Library) m return out } +// heldLibraries is the root of every library whose disk or mount is not there, with its name, judged now and read +// only: what a full apply holds its containers for (ADR 0328 §9), for one that judges without applying libraries. +func heldLibraries(ctx context.Context, libs []declaration.Library) map[string]string { + verdicts := map[string]libraryVerdict{} + for _, l := range libs { + rec, _ := libraryRecord(l.Library) + verdicts[l.Library] = assessLibrary(ctx, l, rec) + } + return heldRoots(verdicts, libs) +} + // mountsHeld is the library a container waits for: one whose root it mounts, or mounts below, is held. func mountsHeld(c *declaration.Container, held map[string]string) string { for _, v := range c.Volumes { @@ -341,12 +352,7 @@ func WaitingOnLibrary(d *declaration.Declaration) map[string]bool { if d == nil || len(d.Libraries) == 0 { return out } - verdicts := map[string]libraryVerdict{} - for _, l := range d.Libraries { - rec, _ := libraryRecord(l.Library) - verdicts[l.Library] = assessLibrary(context.Background(), l, rec) - } - held := heldRoots(verdicts, d.Libraries) + held := heldLibraries(context.Background(), d.Libraries) for _, r := range d.Resources { if c, ok := r.(*declaration.Container); ok && mountsHeld(c, held) != "" { out[c.ID] = true From 0c17c76493f774cafd82394ab78a02f0f35db226 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 00:47:48 +0000 Subject: [PATCH 6/7] 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()) + } +} From 2767278228bc3812004c16a3b55dd08fa5a7f791 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 01:06:47 +0000 Subject: [PATCH 7/7] Say every refusal kind at once and never lose a count; bound the waiting and the busy per socket Review 12004 of mesh-host#111: refusals counted after a said one reached the journal only if the same account was refused again a minute later, and a refusal of another kind hid behind it. The first of each (account, kind) is said at once, and what is counted is said on a timer and when the socket stops. Connections waiting to be judged are bounded below the open-file limit, the busy refusals belong to one Serve, and the tests' ask has a deadline. Issue 668. Co-Authored-By: Claude Opus 5.5 --- internal/filling/filling.go | 137 ++++++++++++++++++++++++------- internal/filling/filling_test.go | 132 ++++++++++++++++++++++++++++- 2 files changed, 239 insertions(+), 30 deletions(-) diff --git a/internal/filling/filling.go b/internal/filling/filling.go index e14da2a..fc26c5e 100644 --- a/internal/filling/filling.go +++ b/internal/filling/filling.go @@ -27,8 +27,10 @@ import ( "os" "os/user" "path/filepath" + "sort" "strconv" "sync" + "sync/atomic" "time" "github.com/novox/mesh-host/internal/declaration" @@ -201,14 +203,22 @@ var ( ) // 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. +// is refused busy. busyWaitingAtOnce bounds how many busy refusals wait for their request at once, per Serve; over it +// a busy refusal is written at once, as mesh-cli's socket does. judgingAtOnce bounds the callers judged at once, and +// judgeWaiting the connections waiting to be judged — each an open file, so far below the engine's open-file limit; +// one over is closed at once, unread. Variables so a test can say. var ( - atOnce = 4 - busyWaiting = make(chan struct{}, 8) - judgingAtOnce = 16 + atOnce = 4 + busyWaitingAtOnce = 8 + judgingAtOnce = 16 + judgeWaiting = 256 ) +// refusalsSaidEvery is how often the refusals counted and not yet said are said: each kind of refusal of each account +// is said at once the first time, and what follows it within this time is counted and said at its end, or when the +// socket stops. A variable so a test need not wait. +var refusalsSaidEvery = time.Minute + // 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. @@ -267,9 +277,27 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error // 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 waiting atomic.Int64 + busyWaiting := make(chan struct{}, busyWaitingAtOnce) + refusals := &refusalWords{say: say} + // Said once everything has stopped: what was counted is never lost with the socket. + defer refusals.flush() var wg sync.WaitGroup defer wg.Wait() + wg.Add(1) + go func() { + defer wg.Done() + tick := time.NewTicker(refusalsSaidEvery) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + refusals.flush() + } + } + }() backoff := time.Duration(0) for { conn, err := acceptFrom(l) @@ -292,6 +320,13 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error continue } backoff = 0 + if waiting.Add(1) > int64(judgeWaiting) { + // Too many waiting to be judged already: this one is closed unread, and counted. + waiting.Add(-1) + refusals.said(^uint32(0), "overflow", "too many connections are waiting to be judged; closed unread") + conn.Close() + continue + } wg.Add(1) go func() { defer wg.Done() @@ -299,6 +334,7 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error judging <- struct{}{} uid, ok := judgePeer(conn, w, refusals) <-judging + waiting.Add(-1) if !ok { return } @@ -309,7 +345,7 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error defer func() { <-slots }() answer(ctx, conn, uid, w, say) default: - busy(conn, readBy, say) + busy(conn, readBy, busyWaiting, say) } }() } @@ -318,8 +354,8 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error // 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) + closeWith := func(uid uint32, kind, why string) (uint32, bool) { + refusals.said(uid, kind, why) _ = conn.SetWriteDeadline(time.Now().Add(refusedWithin)) _ = json.NewEncoder(conn).Encode(Refusal("", why)) conn.Close() @@ -327,44 +363,87 @@ func judgePeer(conn *net.UnixConn, w Writer, refusals *refusalWords) (uint32, bo } uid, pid, err := peerOf(conn) if err != nil { - return closeWith(^uint32(0), "the node-engine could not read which account is asking, so nothing was "+ + return closeWith(^uint32(0), "unread", "the node-engine could not read which account is asking, so nothing was "+ "written: "+err.Error()) } if err := sameNamespaces(pid); err != nil { - return closeWith(uid, "the asking process does not run in this machine's own namespaces, so nothing was "+ + return closeWith(uid, "namespace", "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 { - return closeWith(uid, err.Error()) + return closeWith(uid, "account", 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. +// refusalWords says refused callers in the journal without letting a flood flood it: the first refusal of each kind +// for each account at once, and those that follow it counted and said by flush — on a timer and when the socket stops +// — so an account that stops still has its count said, and a refusal of another kind never hides behind one said. type refusalWords struct { - mu sync.Mutex - say func(string) - last map[uint32]time.Time - unsaid map[uint32]int + mu sync.Mutex + say func(string) + pending map[refusalKey]*refusalCount } -func (r *refusalWords) said(uid uint32, why string) { +type refusalKey struct { + uid uint32 + kind string +} + +type refusalCount struct { + n int + why string +} + +func (r *refusalWords) said(uid uint32, kind, why string) { r.mu.Lock() defer r.mu.Unlock() - if r.unsaid == nil { - r.unsaid = map[uint32]int{} + if r.pending == nil { + r.pending = map[refusalKey]*refusalCount{} } - if at, ok := r.last[uid]; ok && time.Since(at) < time.Minute { - r.unsaid[uid]++ + k := refusalKey{uid, kind} + if c, ok := r.pending[k]; ok { + c.n++ + c.why = why return } - more := "" - if n := r.unsaid[uid]; n > 0 { - more = fmt.Sprintf(" (and %d more from it since it was last said)", n) + r.pending[k] = &refusalCount{} + r.say(fmt.Sprintf("a filled file's write from %s was refused (%s): %s", who(k), kind, why)) +} + +// flush says what was counted since each was last said, and forgets every key, so the next refusal of each is said at +// once again. +func (r *refusalWords) flush() { + r.mu.Lock() + defer r.mu.Unlock() + keys := make([]refusalKey, 0, len(r.pending)) + for k := range r.pending { + keys = append(keys, k) } - 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)) + sort.Slice(keys, func(i, j int) bool { + if keys[i].uid != keys[j].uid { + return keys[i].uid < keys[j].uid + } + return keys[i].kind < keys[j].kind + }) + for _, k := range keys { + if c := r.pending[k]; c.n > 0 { + r.say(fmt.Sprintf("%d more writes of a filled file from %s were refused (%s) since it was last said; the last: %s", + c.n, who(k), k.kind, c.why)) + } + } + r.pending = nil +} + +// who names a caller by its user id, or says why it has none. +func who(k refusalKey) string { + switch { + case k.kind == "overflow": + return "a caller not yet judged" + case k.uid == ^uint32(0): + return "an account that could not be read" + } + return fmt.Sprintf("user id %d", k.uid) } // answer serves one connection from the provisioner client's account. @@ -396,7 +475,7 @@ func answer(ctx context.Context, conn *net.UnixConn, uid uint32, w Writer, say f // 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)) { +func busy(conn *net.UnixConn, readBy time.Time, busyWaiting chan struct{}, say func(string)) { select { case busyWaiting <- struct{}{}: by := time.Now().Add(refusedWithin) diff --git a/internal/filling/filling_test.go b/internal/filling/filling_test.go index 2c198ad..0e7cff3 100644 --- a/internal/filling/filling_test.go +++ b/internal/filling/filling_test.go @@ -224,6 +224,8 @@ func ask(t *testing.T, socket string, req Request) Answer { t.Fatal(err) } defer conn.Close() + // Bounded: a socket that never answers fails the test instead of hanging it. + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) if err := json.NewEncoder(conn).Encode(req); err != nil { t.Fatal(err) } @@ -327,7 +329,7 @@ func TestAFloodFromAnotherAccountDoesNotStopTheClient(t *testing.T) { } } -func busyWaitingCap() int { return cap(busyWaiting) } +func busyWaitingCap() int { return busyWaitingAtOnce } // A client that connects and says nothing holds one slot only until within, and another write goes on meanwhile. func TestASilentClientDoesNotHoldTheSocket(t *testing.T) { @@ -429,3 +431,131 @@ func TestTheSocketsDirectoryIsOpenedToEveryAccount(t *testing.T) { t.Errorf("the socket's directory is %o", info.Mode().Perm()) } } + +// The first refusal of each kind for each account is said at once; what follows within the time is counted, and the +// count is said when the time is up, even when the account has stopped. +func TestRefusalsAreSaidOnceAndTheirCountIsNeverLost(t *testing.T) { + var mu sync.Mutex + var said []string + r := &refusalWords{say: func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }} + for i := 0; i < 3; i++ { + r.said(1000, "account", "not the provisioner client's account") + } + // Another kind from the same account is said at once, not hidden behind the first. + r.said(1000, "namespace", "another namespace") + if len(said) != 2 || !strings.Contains(said[0], "(account)") || !strings.Contains(said[1], "(namespace)") { + t.Fatalf("said at once: %q", said) + } + // The account stops; the time is up; the two counted are said. + r.flush() + if len(said) != 3 || !strings.Contains(said[2], "2 more") || !strings.Contains(said[2], "user id 1000") { + t.Fatalf("after the time: %q", said) + } + // And the next refusal is said at once again. + r.said(1000, "account", "not the provisioner client's account") + if len(said) != 4 { + t.Fatalf("after the count was said: %q", said) + } +} + +// refusingSocket serves a socket that refuses this test's account, with counted refusals said every every. It returns +// how to be refused n times, everything said so far, and how to stop it. +func refusingSocket(t *testing.T, every time.Duration) (func(int), func() string, func()) { + t.Helper() + was := refusalsSaidEvery + refusalsSaidEvery = every + dir, err := os.MkdirTemp("", "fill") + if err != nil { + t.Fatal(err) + } + socket := filepath.Join(dir, "fill.sock") + ctx, cancel := context.WithCancel(context.Background()) + var mu sync.Mutex + var said []string + all := func() string { mu.Lock(); defer mu.Unlock(); return strings.Join(said, "\n") } + stopped := make(chan error) + go func() { + stopped <- Serve(ctx, socket, &fixed{uid: uint32(os.Getuid()) + 1}, + func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }) + }() + for i := 0; i < 100; i++ { + if _, err := os.Stat(socket); err == nil { + break + } + time.Sleep(10 * time.Millisecond) + } + refuse := func(n int) { + for i := 0; i < n; i++ { + conn, err := net.Dial("unix", socket) + if err != nil { + t.Fatal(err) + } + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) + var a Answer + _ = json.NewDecoder(conn).Decode(&a) + conn.Close() + } + } + var once sync.Once + stop := func() { + once.Do(func() { + cancel() + if err := <-stopped; err != nil { + t.Error(err) + } + refusalsSaidEvery = was + os.RemoveAll(dir) + }) + } + t.Cleanup(stop) + return refuse, all, stop +} + +// Refusals counted while the socket serves are said on its timer, though the account has stopped asking. +func TestTheSocketSaysItsCountedRefusalsOnATimer(t *testing.T) { + refuse, all, _ := refusingSocket(t, 200*time.Millisecond) + refuse(3) + // "more writes", not a number: a tick may fall between two refusals and split the count. + deadline := time.Now().Add(3 * time.Second) + for !strings.Contains(all(), "more writes") && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if !strings.Contains(all(), "more writes") { + t.Fatalf("the counted refusals were not said on the timer: %s", all()) + } +} + +// Refusals counted when the socket stops are said as it stops. +func TestTheSocketSaysItsCountedRefusalsWhenItStops(t *testing.T) { + refuse, all, stop := refusingSocket(t, time.Hour) + refuse(5) + if strings.Contains(all(), "4 more") { + t.Fatalf("said before the socket stopped: %s", all()) + } + stop() + if !strings.Contains(all(), "4 more") { + t.Errorf("the refusals counted when the socket stopped were lost: %s", all()) + } +} + +// A connection over the bound of those waiting to be judged is closed unread, and said. +func TestAConnectionOverTheWaitingBoundIsClosedUnread(t *testing.T) { + was := judgeWaiting + judgeWaiting = 0 + t.Cleanup(func() { judgeWaiting = was }) + w := &fixed{uid: uint32(os.Getuid())} + socket, said := serve(t, w) + conn, err := net.Dial("unix", socket) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) + var a Answer + if err := json.NewDecoder(conn).Decode(&a); err == nil { + t.Fatalf("answered over the bound: %+v", a) + } + if len(w.writes) != 0 || !strings.Contains(strings.Join(said(), "\n"), "(overflow)") { + t.Errorf("writes %v, said %q", w.writes, said()) + } +}