From b665156b0fbfd0ea37d3534588d299baf87d3302 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:54:17 +0100 Subject: [PATCH] A3k finding 2: distribution reads its caller list as mesh-config resolves it, not from an environment nothing renders The node-engine fills ${config:} only in files, unit names, owners and groups, never in a process's environment, so MESH_CONFIG_CALLERS was never set and design 51's defaults always held. The tools bundle now reads `callers` through stdio.Config on each call (holder.ConfiguredCallers, Verbs.WithCallers per call); the program, which has no runtime resolving config, asks its own tools bundle's new registry_callers on each wake of the windows (manifest invokes). The environment variable is gone. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/distribution-tools/contract_test.go | 25 ++++++- .../cmd/distribution-tools/main.go | 69 ++++++++++++++++--- modules/distribution/cmd/distribution/main.go | 6 +- .../distribution/whole_integration_test.go | 17 ++++- .../distribution/cmd/distribution/windows.go | 21 ++++++ .../distribution/internal/holder/callers.go | 31 +++++++++ .../distribution/internal/holder/driver.go | 9 ++- .../internal/holder/holder_test.go | 2 +- modules/distribution/module.json | 5 +- 9 files changed, 164 insertions(+), 21 deletions(-) diff --git a/modules/distribution/cmd/distribution-tools/contract_test.go b/modules/distribution/cmd/distribution-tools/contract_test.go index e76eecf1..0cc8fb8f 100644 --- a/modules/distribution/cmd/distribution-tools/contract_test.go +++ b/modules/distribution/cmd/distribution-tools/contract_test.go @@ -3,6 +3,7 @@ package main import ( "bytes" "encoding/json" + "errors" "reflect" "strings" "testing" @@ -30,7 +31,7 @@ func TestTheVerbsSpeakTheSeatsContract(t *testing.T) { reg.VerbVersions: reg.VersionsInput{}, reg.VerbRemove: reg.RemoveInput{}, reg.VerbHold: reg.HoldInput{}, reg.VerbRelease: reg.ReleaseInput{}, reg.VerbMirror: reg.MirrorInput{}, reg.VerbTags: reg.TagsInput{}} seen := 0 - for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}) { + for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}, func() (holder.Callers, error) { return holder.DefaultCallers(), nil }) { seat, verb, ok := strings.Cut(tool.Name, ".") in, isVerb := inputs[verb] if !ok || !isVerb || (seat != holder.ImageSeat && seat != holder.ArtifactSeat) { @@ -56,6 +57,28 @@ func TestTheVerbsSpeakTheSeatsContract(t *testing.T) { t.Errorf("%s's body %s is not the contract's: %v", event, b, err) } } + // The caller list is read on each call: one that cannot be read answers no verb, and registry_callers says it as read. + unread := func() (holder.Callers, error) { return nil, errors.New("config refused") } + for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}, unread) { + if tool.Name == holder.ImageSeat+"."+holder.VerbVersions { + if _, err := tool.Run(map[string]any{"name": "m/a"}); err == nil || !strings.Contains(err.Error(), "caller list") { + t.Errorf("versions with an unread caller list = %v, want refused naming the caller list", err) + } + } + } + answered := false + for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}, func() (holder.Callers, error) { return holder.Callers{holder.VerbPublish: {"mesh-build-go"}}, nil }) { + if tool.Name == holder.CallersTool { + answered = true + got, err := tool.Run(map[string]any{}) + if b, _ := json.Marshal(got); err != nil || string(b) != `{"callers":{"publish":["mesh-build-go"]}}` { + t.Errorf("%s = %s, %v", holder.CallersTool, b, err) + } + } + } + if !answered { + t.Errorf("no tool %s: the program could not read the caller list", holder.CallersTool) + } if a := publishAnswer(holder.PublishAnswer{Transfer: "t", Missing: []string{"sha256:bb"}}); !reflect.DeepEqual(a.Lacks, []string{"sha256:bb"}) || a.Held { t.Errorf("publish answers %+v", a) } diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index 4c216f21..e33c4ba9 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -238,7 +238,16 @@ func arg[T any](args map[string]any) (T, error) { func str(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } -func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { +// tools are distribution's tools over v; each call that checks a caller reads the caller list on that call (callers: +// the config value as mesh-config resolves it now), so an operator's change of it holds from the next call on. +func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.Callers, error)) []stdio.Tool { + now := func() (*holder.Verbs, error) { + c, err := callers() + if err != nil { + return nil, fmt.Errorf("the caller list cannot be read, so no verb is answered: %w", err) + } + return v.WithCallers(c), nil + } obj := func(props map[string]any, req ...string) map[string]any { return map[string]any{"type": "object", "required": req, "properties": props} } @@ -249,6 +258,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { stdio.Tool{Name: seat + "." + holder.VerbPublish, Description: "Open a transfer of a publish (ADR 0611): its name, version, kind, digest and parts; answers the parts the holder lacks.", Input: obj(map[string]any{"transfer": s("the transfer's id"), "name": s("/"), "version": s("the builder's version"), "kind": s("image, bundle, archive or upstream"), "digest": s("sha256:…"), "parts": map[string]any{"type": "array"}}, "transfer", "name", "kind", "digest", "parts"), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } in, err := arg[reg.PublishInput](args) if err != nil { return nil, err @@ -266,6 +279,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { stdio.Tool{Name: seat + "." + holder.VerbPublishPart, Description: "Send one chunk (at most 1 MiB) of one part of an open transfer, from the caller that opened it.", Input: obj(map[string]any{"transfer": s("its id"), "part": s("the part's digest"), "sequence": map[string]any{"type": "integer"}, "last": map[string]any{"type": "boolean"}, "data": s("base64")}, "transfer", "part", "sequence", "last", "data"), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } in, err := arg[reg.PublishPartInput](args) if err != nil { return nil, err @@ -279,6 +296,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { }}, stdio.Tool{Name: seat + "." + holder.VerbVersions, Description: "The versions the seat holds of a name.", Input: obj(map[string]any{"name": s("/")}, "name"), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } a, err := v.Versions(seat, stdio.CallerOf(args), str(args, "name")) if err != nil { return nil, err @@ -288,6 +309,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { stdio.Tool{Name: seat + "." + holder.VerbRemove, Description: "Remove a tagged version (the builder), or for retention an untagged one by its digest (the artifact sweep); says removed.", Input: obj(map[string]any{"name": s("name"), "version": s("a tagged version"), "tag": s("its tag"), "digest": s("for retention"), "reason": s("why")}, "name", "reason"), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } in, err := arg[reg.RemoveInput](args) if err != nil { return nil, err @@ -301,6 +326,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { stdio.Tool{Name: seat + "." + holder.VerbHold, Description: "Keep one untagged bundle, archive or copied base from retention (the artifact sweep alone); says held.", Input: obj(map[string]any{"name": s("name"), "digest": s("sha256:…"), "reason": s("which of the sweep's rules keeps it")}, "name", "digest", "reason"), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } in, err := arg[reg.HoldInput](args) if err != nil { return nil, err @@ -313,6 +342,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { stdio.Tool{Name: seat + "." + holder.VerbRelease, Description: "Let a hold go (the artifact sweep, or the operator's session through registry_holds); says released.", Input: obj(map[string]any{"name": s("name"), "digest": s("sha256:…"), "reason": s("why it is let go")}, "name", "digest", "reason"), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } in, err := arg[reg.ReleaseInput](args) if err != nil { return nil, err @@ -327,6 +360,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { out = append(out, stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbTags, Description: "The tags of an image.", Input: obj(map[string]any{"name": s("name")}, "name"), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } a, err := v.Tags(holder.ImageSeat, stdio.CallerOf(args), str(args, "name")) if err != nil { return nil, err @@ -336,6 +373,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbMirror, Description: "Copy an outside image, pinned by its digest, into the store as upstream// (every platform, only the parts it lacks): answers at once with the reference the copy will have, and says published (or refused, with why) on the seat's events when the copy ends; wait on that event (mesh-build-image alone).", Input: obj(map[string]any{"source": s("/"), "digest": s("sha256:…")}, "source", "digest"), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } in, err := arg[reg.MirrorInput](args) if err != nil { return nil, err @@ -346,6 +387,15 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { } return reg.MirrorAnswer{Digest: in.Digest, Reference: ref}, nil }}, + stdio.Tool{Name: holder.CallersTool, Description: "Who may call which verb of distribution's seats: the trusted config value callers as mesh-config resolves it on this machine now (design 51's list where unset). The program asks it at each wake of the windows, as it has no config of its own.", + Input: obj(map[string]any{}), + Run: func(map[string]any) (any, error) { + c, err := callers() + if err != nil { + return nil, err + } + return map[string]any{"callers": c}, nil + }}, stdio.Tool{Name: "registry_health", Description: "distribution's health check, run by the runtime each minute: versions of each seat asked through the store's read side (the program's port 5000); fails naming the program when it does not answer.", Input: obj(map[string]any{}), Run: func(map[string]any) (any, error) { @@ -355,6 +405,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { stdio.Tool{Name: "registry_incoming", Description: "Open transfers of incoming and its fill; with drop, drops one by hand (a hand act, why required).", Input: obj(map[string]any{"drop": s("a transfer id to drop"), "why": s("why it is dropped")}), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } idle, _ := v.DropIdle() if id := str(args, "drop"); id != "" { if str(args, "why") == "" { @@ -371,6 +425,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { stdio.Tool{Name: "registry_holds", Description: "What is held, since when and for whom; with release, lets one go through the seat's release as the operator's session (a hand act).", Input: obj(map[string]any{"seat": s("the seat"), "release": s("the name to release"), "digest": s("its digest"), "why": s("why")}), Run: func(args map[string]any) (any, error) { + v, err := now() + if err != nil { + return nil, err + } if name := str(args, "release"); name != "" { if str(args, "why") == "" { return nil, errors.New("a release by hand is a hand act: say why") @@ -414,20 +472,15 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { } func main() { - callers, err := holder.ParseCallers(os.Getenv("MESH_CONFIG_CALLERS")) - if err != nil { - logf("%v", err) - os.Exit(1) - } base, _ := url.Parse("http://127.0.0.1:5001") store := &holder.OCI{Base: base, Registry: os.Getenv("MESH_REGISTRY_REACH"), Upstream: &holder.Upstream{}} - h := &holder.Holder{Callers: callers, Objects: parts.BusObjects(holder.IncomingObjects), Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: store, Say: say, Finding: finding, + h := &holder.Holder{Objects: parts.BusObjects(holder.IncomingObjects), Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: store, Say: say, Finding: finding, Transfers: parts.BusTransfers[holder.Opening](holder.TransfersState)} v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}, Mirrors: mirrors{}} go followMirrors(v) windows := kept[holder.Window]{state: "windows", key: func(w holder.Window) string { return w.Name }} closeBootstrap = bootstrapCloser(windows) - if err := stdio.Serve("", tools(v, windows)); err != nil { + if err := stdio.Serve("", tools(v, windows, holder.ConfiguredCallers)); err != nil { logf("%v", err) os.Exit(1) } diff --git a/modules/distribution/cmd/distribution/main.go b/modules/distribution/cmd/distribution/main.go index cf856c6e..242cc0b4 100644 --- a/modules/distribution/cmd/distribution/main.go +++ b/modules/distribution/cmd/distribution/main.go @@ -64,12 +64,8 @@ func run(ctx context.Context, getenv func(string) string) error { return err } defer calls.Close() - callers, err := holder.ParseCallers(getenv("MESH_CONFIG_CALLERS")) - if err != nil { - return err - } findings := process.OptionalBucket(ctx, js, "distribution_findings", "a window held for seven days is said in the log only", logger.Printf) - go drive(ctx, js, cred, &holder.Driver{Delivery: busDelivery{calls}, State: kvWindows{windows}, Callers: callers, + go drive(ctx, js, cred, &holder.Driver{Delivery: busDelivery{calls}, State: kvWindows{windows}, Callers: busCallers{calls}.Read, Closed: sayClosed(js, cred), Finding: findingsKV(findings, logger.Printf)}, logger) } front := &holder.Front{Store: store, Window: &open, Logf: logger.Printf, diff --git a/modules/distribution/cmd/distribution/whole_integration_test.go b/modules/distribution/cmd/distribution/whole_integration_test.go index 3bfc3be8..c60bb57b 100644 --- a/modules/distribution/cmd/distribution/whole_integration_test.go +++ b/modules/distribution/cmd/distribution/whole_integration_test.go @@ -6,8 +6,9 @@ package main // run() as the node starts it, its broker account on a nats-server of this test alone, the registry container a fake // OCI store on the loopback, the status view's node_haves and the node-engine seat's manifest-reads answered on their // call subjects. An away node holds the old name open; its return (manifest-reads-changed) closes it, said once on -// distribution.events, after which a direct write is refused naming publish; reads stay open; and a waiter on a seat's -// `published` hears none of the windows' events. +// distribution.events, after which a direct write is refused naming publish; reads stay open; a waiter on a seat's +// `published` hears none of the windows' events; and the build modules counted are publish's callers as the tools +// bundle answers the config value, not design 51's defaults. import ( "context" @@ -76,10 +77,22 @@ func TestDistributionWhole(t *testing.T) { answer(m, map[string]any{"keys": map[string]any{ "ace.mesh-build-go": map[string]any{"fp": "fp-new"}, "g14.mesh-build-go": map[string]any{"fp": fp}, "ace.gitea": map[string]any{"fp": "x"}, "g14.gitea": map[string]any{"fp": "x"}, + // A build module the operator took off publish's list: it holds no window (the caller list as configured). + "g14.mesh-build-node": map[string]any{"fp": "fp-node-old"}, }}) }); err != nil { t.Fatal(err) } + // distribution's own tools bundle answers the config value callers as mesh-config resolved it: publish by + // mesh-build-go alone. The program reads it there, never from an environment variable nothing renders. + if _, err := nc.Subscribe("mesh.mod.distribution.>", func(m *nats.Msg) { + if !strings.Contains(m.Subject, holder.CallersTool) { + return + } + answer(m, map[string]any{"callers": map[string]any{"publish": []string{"mesh-build-go"}}}) + }); err != nil { + t.Fatal(err) + } engine := func(node string) { for _, subj := range []string{"mesh.seat.node-engine.call.manifest-reads." + node + ".>", "mesh.seat.node-engine.tool.manifest-reads." + node} { if _, err := nc.Subscribe(subj, func(m *nats.Msg) { diff --git a/modules/distribution/cmd/distribution/windows.go b/modules/distribution/cmd/distribution/windows.go index 1bf2fd98..03734c82 100644 --- a/modules/distribution/cmd/distribution/windows.go +++ b/modules/distribution/cmd/distribution/windows.go @@ -28,6 +28,8 @@ import ( const ( havesAsk = "mesh-delivery-view.node_haves" readsAsk = "seat:node-engine.manifest-reads" + // callersAsk is distribution's own tool answering its config value callers as resolved. + callersAsk = holder.Self + "." + holder.CallersTool ) // caller is a process's call of a tool or a seat verb. @@ -36,6 +38,25 @@ type caller interface { CallOn(node, key string, body any) (json.RawMessage, error) } +// busCallers reads the caller list from distribution's own tools bundle (holder.CallersTool, manifest invokes): a +// program has no runtime that resolves its module's config values, the tools bundle has (stdio.Config), so the program +// asks it on each wake rather than read an environment variable nothing renders. +type busCallers struct{ c caller } + +func (b busCallers) Read() (holder.Callers, error) { + raw, err := b.c.Call(callersAsk, map[string]any{}) + if err != nil { + return nil, fmt.Errorf("%s: %w", callersAsk, err) + } + var a struct { + Callers json.RawMessage `json:"callers"` + } + if err := json.Unmarshal(unwrap(raw), &a); err != nil || len(a.Callers) == 0 { + return nil, fmt.Errorf("%s answered what is not a caller list: %s", callersAsk, unwrap(raw)) + } + return holder.ParseCallers(string(a.Callers)) +} + // busDelivery answers holder.Delivery through the bus. type busDelivery struct{ c caller } diff --git a/modules/distribution/internal/holder/callers.go b/modules/distribution/internal/holder/callers.go index 4c6dfc2f..c2731811 100644 --- a/modules/distribution/internal/holder/callers.go +++ b/modules/distribution/internal/holder/callers.go @@ -89,6 +89,37 @@ func ParseCallers(raw string) (Callers, error) { if err != nil { return nil, err } + return selfNotReleasing(c) +} + +// ConfiguredCallers is the config value `callers` as mesh-config resolved it for this machine, read through the tools +// runtime's stdio.Config on each call (never an environment variable: nothing renders a config value into a +// program's environment); unset is DefaultCallers, and Self on release's list is refused as ParseCallers refuses it. +func ConfiguredCallers() (Callers, error) { + c, err := stdio.ConfiguredVerbCallers(CallersConfig, VerbsOf(ImageSeat), DefaultCallers()) + if err != nil { + return nil, err + } + return selfNotReleasing(c) +} + +// CallersConfig is the config value's key, and CallersTool the tools bundle's tool answering it as resolved: the +// program, which has no tools runtime, asks that tool over the bus (manifest invokes). +const ( + CallersConfig = "callers" + CallersTool = "registry_callers" +) + +// WithCallers is v with the caller list c: a copy for one call, so a list read on that call never changes another's. +func (v *Verbs) WithCallers(c Callers) *Verbs { + h := *v.Holder + h.Callers = c + out := *v + out.Holder = &h + return &out +} + +func selfNotReleasing(c Callers) (Callers, error) { if slices.Contains(c[VerbRelease], Self) { return nil, fmt.Errorf("callers puts %s on release's list: registry_holds would release in its own name", Self) } diff --git a/modules/distribution/internal/holder/driver.go b/modules/distribution/internal/holder/driver.go index 9c29d045..d682923d 100644 --- a/modules/distribution/internal/holder/driver.go +++ b/modules/distribution/internal/holder/driver.go @@ -63,7 +63,8 @@ type WindowState interface { type Driver struct { Delivery Delivery State WindowState - Callers Callers + // Callers is the caller list as it holds on this wake: publish's callers are the build modules counted. + Callers func() (Callers, error) // Closed says a window's closing on distribution.events. Closed func(c Closed) error Finding func(kind, subject, words string, raised bool) @@ -84,7 +85,11 @@ func (d *Driver) Read() ([]NodeRead, error) { if err != nil { return nil, fmt.Errorf("the publishing builds cannot be read, so no window moves: %w", err) } - builds := d.Callers[VerbPublish] + callers, err := d.Callers() + if err != nil { + return nil, fmt.Errorf("the caller list cannot be read, so no window moves: %w", err) + } + builds := callers[VerbPublish] byNode := map[string]*NodeRead{} var order []string for _, h := range haves { diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index 1e740337..130872e0 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/modules/distribution/internal/holder/holder_test.go @@ -532,7 +532,7 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { var closed []string found := map[string]bool{} now := time.Unix(0, 0) - d := &Driver{Delivery: del, State: st, Callers: DefaultCallers(), Now: func() time.Time { return now }, + d := &Driver{Delivery: del, State: st, Callers: func() (Callers, error) { return DefaultCallers(), nil }, Now: func() time.Time { return now }, Closed: func(c Closed) error { closed = append(closed, c.Window+" "+c.Seat); return nil }, Finding: func(kind, subject, _ string, raised bool) { found[kind+" "+subject] = raised }} if _, err := d.Wake("start"); err != nil { diff --git a/modules/distribution/module.json b/modules/distribution/module.json index 7a6f2179..f3e98583 100644 --- a/modules/distribution/module.json +++ b/modules/distribution/module.json @@ -43,6 +43,7 @@ } ], "tools": [ + "registry_callers", "registry_direct_writes", "registry_health", "registry_holds", @@ -50,6 +51,7 @@ "registry_old_name" ], "invokes": [ + "distribution.registry_callers", "mesh-delivery-builder.builder_versions", "mesh-delivery-view.node_haves", "seat:node-engine.manifest-reads" @@ -199,8 +201,7 @@ "env": { "MESH_BROKER_FILE": "/var/lib/mesh/registry/broker", "MESH_REGISTRY_LOOPBACK": "http://127.0.0.1:5001", - "MESH_REGISTRY_LISTEN": ":5000", - "MESH_CONFIG_CALLERS": "${config:callers}" + "MESH_REGISTRY_LISTEN": ":5000" }, "user": "distribution", "health": {