From ec27f12524011d1c1c9fd5e4c9841a48ebe67b8a Mon Sep 17 00:00:00 2001 From: jochens Date: Fri, 13 Nov 2026 01:04:41 +0100 Subject: [PATCH] A-37 the keeper: the module's one process bus account, a socket only root opens, and the root processes' bus through it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 42 zfs-owns-zfs §18, §20, §26, §27, A-6, A-20, A-37 --- modules/zfs/cmd/zfs-tools/bus.go | 50 +++- modules/zfs/cmd/zfs-tools/keeper.go | 312 +++++++++++++++++++++++ modules/zfs/cmd/zfs-tools/keeper_test.go | 124 +++++++++ modules/zfs/cmd/zfs-tools/main.go | 5 + modules/zfs/go.mod | 2 +- 5 files changed, 480 insertions(+), 13 deletions(-) create mode 100644 modules/zfs/cmd/zfs-tools/keeper.go create mode 100644 modules/zfs/cmd/zfs-tools/keeper_test.go diff --git a/modules/zfs/cmd/zfs-tools/bus.go b/modules/zfs/cmd/zfs-tools/bus.go index 8a8f4ba8f..347e0d57e 100644 --- a/modules/zfs/cmd/zfs-tools/bus.go +++ b/modules/zfs/cmd/zfs-tools/bus.go @@ -1,7 +1,9 @@ package main import ( + "context" "encoding/json" + "fmt" "os" "strings" "sync" @@ -33,25 +35,49 @@ func (b busStates) Get(state, key string) (json.RawMessage, bool, error) { return e.Value, true, nil } -// JoinBus gives the machine this process's events and states, as its own bus account says. Without the -// account or the bus the machine keeps none, and what it changes is said as a warning (§18: with the bus down -// the apply converges from the config file it last got). It answers how to leave the bus. +// JoinBus gives a root process the module's events and states through the keeper's socket (A-37): the process holds no +// bus account of its own. Without the socket the machine keeps none, and what it changes is said as a warning (§18: +// with the bus down the apply converges from the config file it last got). It answers how to leave the bus. func (m *Machine) JoinBus() func() { + sock := strings.TrimSpace(os.Getenv(KeeperSocketEnv)) + if sock == "" { + m.Warnings = append(m.Warnings, KeeperSocketEnv+" is not set, so the keeper is not reached and the bus is taken as down") + return func() {} + } + c := keeperClient{sock} + // The keeper answering is what the bus being up means: one call says it before the process relies on it. + if _, err := c.Keys(StateSchedule); err != nil { + m.Warnings = append(m.Warnings, err.Error()+": the bus is taken as down") + return func() {} + } + m.Emit = c.emit + m.States = c + m.Findings = meshconditionsclient.FindingWriter{State: findingState{c}} + m.Call = c.call + return func() {} +} + +// RunKeeper is the process `zfs-keeper`: the module's one process bus account, serving the socket to root. +func RunKeeper(ctx context.Context) error { + sock := strings.TrimSpace(os.Getenv(KeeperSocketEnv)) + if sock == "" { + return fmt.Errorf("%s is not set", KeeperSocketEnv) + } acct, err := process.ReadBrokerAccount(os.Getenv) if err != nil { - m.Warnings = append(m.Warnings, err.Error()) - return func() {} + return err } c, err := process.Connect(acct) if err != nil { - m.Warnings = append(m.Warnings, err.Error()) - return func() {} + return err } - m.Emit = func(kind string, body any) error { return c.Emit(kind, body, stdio.EmitOptions{}) } - m.States = busStates{c} - m.Findings = meshconditionsclient.FindingWriter{State: c.State(StateFindings)} - m.Call = c.Call - return c.Close + defer c.Close() + l, err := Listen(sock) + if err != nil { + return err + } + k := &Keeper{Emit: func(kind string, body any) error { return c.Emit(kind, body, stdio.EmitOptions{}) }, States: busStates{c}, Call: c.Call} + return k.Serve(ctx, l) } // toolStates is the module's states as the tools bundle reaches them, through the tool runner. diff --git a/modules/zfs/cmd/zfs-tools/keeper.go b/modules/zfs/cmd/zfs-tools/keeper.go new file mode 100644 index 000000000..fc63abf7e --- /dev/null +++ b/modules/zfs/cmd/zfs-tools/keeper.go @@ -0,0 +1,312 @@ +package main + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "net" + "os" + "path/filepath" + "strings" + "time" + + "golang.org/x/sys/unix" +) + +// The keeper (design 42 zfs-owns-zfs A-37, §18, §20, §26, §27; ADR 0436 §1, §2). +// +// A process's bus account is a user the module makes, never root, so the three root processes (the apply, the run and +// the event follower) declare no `bus`. The process `zfs-keeper`, running as the module's user `zfs-keeper`, holds the +// module's one process bus account and serves a unix socket in the module's state directory that only root may open +// (the peer's uid is checked on every connection). Its calls are a closed list: emit an event of the module, put or +// delete a key of its states `applied`, `schedule` and `findings`, and invoke one of three (`zfs.zfs_snapshots_record`, +// `zfs.zfs_snapshots_schedule` and `seat:node-service-manager.start`, of the two units of the module only). The root +// processes say everything they say on the bus through it; with the keeper not answering they act as with the bus down +// (A-8), and warn. +// +// Reading a key or the keys of those states (`get`, `keys`) is also served: the run reads the hand acts waiting in the +// state `schedule` and the apply reads the state `applied` (A-17), and a root process has no other way to the bus. + +// KeeperSocketEnv names the variable holding the socket's path, in every process of the module that uses it. +const KeeperSocketEnv = "ZFS_KEEPER_SOCKET" + +// KeeperTimeout bounds one call through the keeper. +const KeeperTimeout = 30 * time.Second + +// The calls of the keeper. +const ( + OpEmit = "emit" + OpPut = "put" + OpDelete = "delete" + OpGet = "get" + OpKeys = "keys" + OpInvoke = "invoke" +) + +// KeeperRequest is one call: one JSON line. +type KeeperRequest struct { + Op string `json:"op"` + Kind string `json:"kind,omitempty"` + State string `json:"state,omitempty"` + Key string `json:"key,omitempty"` + Body json.RawMessage `json:"body,omitempty"` +} + +// KeeperAnswer is its answer: one JSON line. +type KeeperAnswer struct { + Error string `json:"error,omitempty"` + Found bool `json:"found,omitempty"` + Value json.RawMessage `json:"value,omitempty"` + Keys []string `json:"keys,omitempty"` +} + +var keeperStates = map[string]bool{StateApplied: true, StateSchedule: true, StateFindings: true} + +// Keeper serves the calls over the bus account of its own process. +type Keeper struct { + Emit func(kind string, body any) error + States StateStore + Call func(key string, body any) (json.RawMessage, error) + // PeerOf is the uid at the other end of a connection; nil reads it from the socket (SO_PEERCRED). + PeerOf func(net.Conn) (int, error) +} + +// invokable says whether a call key is one of the three, and for the service manager whether it starts one of the +// module's two units. +func invokable(key string, body json.RawMessage) error { + switch { + case key == ToolRecord || key == ToolSchedule: + return nil + case strings.HasPrefix(key, "seat:node-service-manager.start@"): + var a map[string]any + if json.Unmarshal(body, &a) != nil || a["scope"] != "system" || (a["unit"] != RunService && a["unit"] != ApplyUnit) || len(a) != 2 { + return fmt.Errorf("the keeper starts only %s or %s, in the system scope", RunService, ApplyUnit) + } + return nil + } + return fmt.Errorf("%q is not one of the three calls the keeper invokes", key) +} + +// Do answers one request. +func (k *Keeper) Do(r KeeperRequest) KeeperAnswer { + fail := func(err error) KeeperAnswer { return KeeperAnswer{Error: err.Error()} } + switch r.Op { + case OpEmit: + known := false + for _, e := range EventKinds { + known = known || (e == r.Kind && e != EventScrubRequested) + } + if !known { + return fail(fmt.Errorf("%q is not an event the module's processes say", r.Kind)) + } + var body any + if len(r.Body) > 0 { + if err := json.Unmarshal(r.Body, &body); err != nil { + return fail(err) + } + } + if err := k.Emit(r.Kind, body); err != nil { + return fail(err) + } + return KeeperAnswer{} + case OpPut, OpDelete, OpGet, OpKeys: + if !keeperStates[r.State] { + return fail(fmt.Errorf("%q is not a state of the module's processes (applied, schedule, findings)", r.State)) + } + switch r.Op { + case OpPut: + var v any + if err := json.Unmarshal(r.Body, &v); err != nil { + return fail(err) + } + if err := k.States.Put(r.State, r.Key, v); err != nil { + return fail(err) + } + case OpDelete: + if err := k.States.Delete(r.State, r.Key); err != nil { + return fail(err) + } + case OpGet: + v, ok, err := k.States.Get(r.State, r.Key) + if err != nil { + return fail(err) + } + return KeeperAnswer{Found: ok, Value: v} + case OpKeys: + keys, err := k.States.Keys(r.State) + if err != nil { + return fail(err) + } + return KeeperAnswer{Keys: keys} + } + return KeeperAnswer{} + case OpInvoke: + if err := invokable(r.Key, r.Body); err != nil { + return fail(err) + } + var body any + _ = json.Unmarshal(r.Body, &body) + out, err := k.Call(r.Key, body) + if err != nil { + return fail(err) + } + return KeeperAnswer{Value: out} + } + return fail(fmt.Errorf("%q is not a call of the keeper (emit, put, delete, get, keys, invoke)", r.Op)) +} + +// peerUID is the account at the other end of a unix socket. +func peerUID(c net.Conn) (int, error) { + uc, ok := c.(*net.UnixConn) + if !ok { + return -1, errors.New("not a unix socket") + } + raw, err := uc.SyscallConn() + if err != nil { + return -1, err + } + uid := -1 + var gerr error + if err := raw.Control(func(fd uintptr) { + cred, err := unix.GetsockoptUcred(int(fd), unix.SOL_SOCKET, unix.SO_PEERCRED) + if err != nil { + gerr = err + return + } + uid = int(cred.Uid) + }); err != nil { + return -1, err + } + return uid, gerr +} + +// Serve answers each connection on l, one request each, until ctx ends: a peer that is not root is told so and +// nothing it sent is read. +func (k *Keeper) Serve(ctx context.Context, l net.Listener) error { + go func() { <-ctx.Done(); l.Close() }() + peerOf := k.PeerOf + if peerOf == nil { + peerOf = peerUID + } + for { + c, err := l.Accept() + if err != nil { + if ctx.Err() != nil { + return nil + } + return err + } + go func() { + defer c.Close() + _ = c.SetDeadline(time.Now().Add(KeeperTimeout)) + enc := json.NewEncoder(c) + if uid, err := peerOf(c); err != nil || uid != 0 { + _ = enc.Encode(KeeperAnswer{Error: "only root opens the keeper's socket"}) + return + } + line, err := bufio.NewReader(c).ReadBytes('\n') + if err != nil { + return + } + var r KeeperRequest + if err := json.Unmarshal(line, &r); err != nil { + _ = enc.Encode(KeeperAnswer{Error: "the request is not JSON: " + err.Error()}) + return + } + _ = enc.Encode(k.Do(r)) + }() + } +} + +// Listen opens the socket: in a folder of the keeper's own, as the keeper's account, mode 0600 so that only the +// account itself and root open it; any older socket is replaced. +func Listen(sock string) (net.Listener, error) { + if err := os.MkdirAll(filepath.Dir(sock), 0o711); err != nil { + return nil, err + } + _ = os.Remove(sock) + l, err := net.Listen("unix", sock) + if err != nil { + return nil, err + } + if err := os.Chmod(sock, 0o600); err != nil { + l.Close() + return nil, err + } + return l, nil +} + +// keeperClient is a root process's way to the bus: calls on the keeper's socket. +type keeperClient struct{ sock string } + +func (c keeperClient) do(r KeeperRequest) (KeeperAnswer, error) { + conn, err := net.DialTimeout("unix", c.sock, 5*time.Second) + if err != nil { + return KeeperAnswer{}, fmt.Errorf("the keeper does not answer on %s: %v", c.sock, err) + } + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(KeeperTimeout)) + if err := json.NewEncoder(conn).Encode(r); err != nil { + return KeeperAnswer{}, err + } + var a KeeperAnswer + if err := json.NewDecoder(conn).Decode(&a); err != nil { + return KeeperAnswer{}, fmt.Errorf("the keeper's answer: %v", err) + } + if a.Error != "" { + return a, errors.New(a.Error) + } + return a, nil +} + +func (c keeperClient) emit(kind string, body any) error { + raw, err := json.Marshal(body) + if err != nil { + return err + } + _, err = c.do(KeeperRequest{Op: OpEmit, Kind: kind, Body: raw}) + return err +} + +func (c keeperClient) Put(state, key string, v any) error { + raw, err := json.Marshal(v) + if err != nil { + return err + } + _, err = c.do(KeeperRequest{Op: OpPut, State: state, Key: key, Body: raw}) + return err +} + +func (c keeperClient) Delete(state, key string) error { + _, err := c.do(KeeperRequest{Op: OpDelete, State: state, Key: key}) + return err +} + +func (c keeperClient) Keys(state string) ([]string, error) { + a, err := c.do(KeeperRequest{Op: OpKeys, State: state}) + return a.Keys, err +} + +func (c keeperClient) Get(state, key string) (json.RawMessage, bool, error) { + a, err := c.do(KeeperRequest{Op: OpGet, State: state, Key: key}) + return a.Value, a.Found, err +} + +func (c keeperClient) call(key string, body any) (json.RawMessage, error) { + raw, err := json.Marshal(body) + if err != nil { + return nil, err + } + a, err := c.do(KeeperRequest{Op: OpInvoke, Key: key, Body: raw}) + return a.Value, err +} + +// findingState is the state `findings` as the finding writer reaches it, through the keeper. +type findingState struct{ c keeperClient } + +func (f findingState) Put(key string, value any) (uint64, error) { + return 0, f.c.Put(StateFindings, key, value) +} +func (f findingState) Delete(key string) error { return f.c.Delete(StateFindings, key) } diff --git a/modules/zfs/cmd/zfs-tools/keeper_test.go b/modules/zfs/cmd/zfs-tools/keeper_test.go new file mode 100644 index 000000000..9c5cbb2cd --- /dev/null +++ b/modules/zfs/cmd/zfs-tools/keeper_test.go @@ -0,0 +1,124 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "net" + "os" + "path/filepath" + "strings" + "testing" +) + +// A-37: the keeper answers a closed list of calls: an event of the module (never scrub.requested), a key of its three +// states, and one of three invokes (the service manager only for the module's two units); anything else is refused. +func TestTheKeeperAnswersOnlyItsClosedListOfCalls(t *testing.T) { + var emitted []string + var called []string + st := memStates{} + k := &Keeper{ + Emit: func(kind string, body any) error { emitted = append(emitted, kind); return nil }, + States: st, + Call: func(key string, body any) (json.RawMessage, error) { + called = append(called, key) + return json.RawMessage(`{"ok":true}`), nil + }, + } + req := func(op, kind, state, key, body string) KeeperRequest { + return KeeperRequest{Op: op, Kind: kind, State: state, Key: key, Body: json.RawMessage(body)} + } + cases := []struct { + name string + r KeeperRequest + refused string + }{ + {"emit a kind", req(OpEmit, EventApplied, "", "", `{"key":"x"}`), ""}, + {"emit the scrub's kind", req(OpEmit, EventScrubRequested, "", "", `{}`), "not an event"}, + {"emit another module's kind", req(OpEmit, "package.changed", "", "", `{}`), "not an event"}, + {"put applied", req(OpPut, "", StateApplied, "hostid", `{"outcome":"applied"}`), ""}, + {"put schedule", req(OpPut, "", StateSchedule, "paused", `{}`), ""}, + {"put another state", req(OpPut, "", "holds", "x", `{}`), "not a state"}, + {"get findings", req(OpGet, "", StateFindings, "k", ``), ""}, + {"keys of another state", req(OpKeys, "", "mirrors", "", ``), "not a state"}, + {"delete schedule", req(OpDelete, "", StateSchedule, "paused", ``), ""}, + {"invoke record", req(OpInvoke, "", "", ToolRecord, `{}`), ""}, + {"invoke schedule", req(OpInvoke, "", "", ToolSchedule, `{}`), ""}, + {"start the run", req(OpInvoke, "", "", "seat:node-service-manager.start@ace", `{"scope":"system","unit":"zfs-mesh-run.service"}`), ""}, + {"start the apply", req(OpInvoke, "", "", "seat:node-service-manager.start@ace", `{"scope":"system","unit":"zfs-config-apply.service"}`), ""}, + {"start another unit", req(OpInvoke, "", "", "seat:node-service-manager.start@ace", `{"scope":"system","unit":"sshd.service"}`), "starts only"}, + {"stop a unit", req(OpInvoke, "", "", "seat:node-service-manager.stop@ace", `{"scope":"system","unit":"zfs-mesh-run.service"}`), "not one of the three"}, + {"invoke a tool of the module", req(OpInvoke, "", "", "zfs.zfs_snapshot", `{}`), "not one of the three"}, + {"another op", req("shell", "", "", "", ``), "not a call"}, + } + for _, c := range cases { + a := k.Do(c.r) + if (c.refused == "") != (a.Error == "") || (c.refused != "" && !strings.Contains(a.Error, c.refused)) { + t.Errorf("%s: %+v", c.name, a) + } + } + if len(emitted) != 1 || len(called) != 4 { + t.Errorf("emitted %v, called %v", emitted, called) + } + if _, left := st["applied/hostid"]; !left { + t.Errorf("the put was not kept") + } +} + +// A-37: only root opens the socket; a root process says and reads through it as through a bus of its own, and with no +// keeper answering it acts as with the bus down, with a warning (A-8). +func TestOnlyRootOpensTheKeepersSocketAndANeedlessKeeperIsTheBusDown(t *testing.T) { + st := memStates{} + uid := 0 + k := &Keeper{Emit: func(string, any) error { return nil }, States: st, + Call: func(string, any) (json.RawMessage, error) { return nil, errors.New("down") }, + PeerOf: func(net.Conn) (int, error) { return uid, nil }} + dir, err := os.MkdirTemp("/var/tmp", "zk") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + sock := filepath.Join(dir, "k", "k.sock") + l, err := Listen(sock) + if err != nil { + t.Fatal(err) + } + ctx, stop := context.WithCancel(context.Background()) + defer stop() + go k.Serve(ctx, l) + c := keeperClient{sock} + if err := c.Put(StateSchedule, "snapshots", map[string]string{"by": "x"}); err != nil { + t.Fatal(err) + } + if keys, err := c.Keys(StateSchedule); err != nil || len(keys) != 1 || keys[0] != "snapshots" { + t.Errorf("%v %v", keys, err) + } + if v, ok, err := c.Get(StateSchedule, "snapshots"); err != nil || !ok || !strings.Contains(string(v), `"by":"x"`) { + t.Errorf("%s %v %v", v, ok, err) + } + if _, err := c.call(ToolSchedule, map[string]any{}); err == nil || !strings.Contains(err.Error(), "down") { + t.Errorf("an invoke that failed: %v", err) + } + uid = 1000 + if err := c.Put(StateSchedule, "x", 1); err == nil || !strings.Contains(err.Error(), "only root") { + t.Errorf("a peer that is not root: %v", err) + } + if _, left := st["schedule/x"]; left { + t.Errorf("a refused peer's call was read") + } + // No keeper: the machine keeps no bus and says so. + t.Setenv(KeeperSocketEnv, filepath.Join(dir, "none.sock")) + m := newFake(t).machine() + m.Emit = nil + m.JoinBus() + if m.Emit != nil || m.States != nil || len(m.Warnings) != 1 || !strings.Contains(m.Warnings[0], "bus is taken as down") { + t.Errorf("no keeper: %+v", m.Warnings) + } + t.Setenv(KeeperSocketEnv, sock) + uid = 0 + m2 := newFake(t).machine() + m2.JoinBus() + if m2.Emit == nil || m2.States == nil || m2.Call == nil || m2.Findings == nil { + t.Errorf("a keeper answering: the bus is up") + } +} diff --git a/modules/zfs/cmd/zfs-tools/main.go b/modules/zfs/cmd/zfs-tools/main.go index eff072e8f..b10224215 100644 --- a/modules/zfs/cmd/zfs-tools/main.go +++ b/modules/zfs/cmd/zfs-tools/main.go @@ -53,6 +53,11 @@ func main() { os.Exit(1) } fmt.Println(said) + case "keeper": + if err := RunKeeper(ctx); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } case "events": if err := Follower(ctx, os.Stdout); err != nil { fmt.Fprintln(os.Stderr, err) diff --git a/modules/zfs/go.mod b/modules/zfs/go.mod index 804a627cf..8b1859664 100644 --- a/modules/zfs/go.mod +++ b/modules/zfs/go.mod @@ -15,5 +15,5 @@ require ( github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect - golang.org/x/sys v0.48.0 // indirect + golang.org/x/sys v0.48.0 )