From 38a634723b9a9e6da4acf61fb090129b230d30c5 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 13:49:15 +0100 Subject: [PATCH] Stamp a ledger operation with the caller the runner proved for the call in flight it names, refusing another module's call or one not in flight; serve the ledger calls on the pipe (issue 1966 T5) --- .../internal/launch/event_answer_test.go | 1 + node-tools/internal/launch/launch.go | 24 ++- node-tools/internal/ledger/ledger.go | 6 +- node-tools/internal/runtime/busstate.go | 5 + node-tools/internal/runtime/caller_test.go | 7 + node-tools/internal/runtime/ledgers.go | 148 +++++++++++++++++- node-tools/internal/runtime/ledgers_test.go | 103 ++++++++++++ node-tools/internal/runtime/runtime.go | 2 + node-tools/test/fixtures/caller-echo.py | 5 +- 9 files changed, 290 insertions(+), 11 deletions(-) diff --git a/node-tools/internal/launch/event_answer_test.go b/node-tools/internal/launch/event_answer_test.go index 6746b36..0824263 100644 --- a/node-tools/internal/launch/event_answer_test.go +++ b/node-tools/internal/launch/event_answer_test.go @@ -52,6 +52,7 @@ func (b *subscribing) BusState(string, json.RawMessage) (json.RawMessage, error) func (b *subscribing) BusStateWatch(string, json.RawMessage, func(json.RawMessage) error) (func(), error) { return func() {}, nil } +func (b *subscribing) Begin(string) (string, func()) { return "", func() {} } func (b *subscribing) Subscribe(d func(json.RawMessage) error) error { b.mu.Lock() defer b.mu.Unlock() diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 17d95d4..f663c82 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -52,12 +52,21 @@ type Tool struct { // stdio.CallerOf, and says Unproven when it is absent. const MetaCaller = "mesh/caller" -// callParams is a `tools/call`'s params: the tool, its arguments as the caller sent them, and — only when the -// subject named one — the caller, in `_meta`, which the runtime writes and no caller can. -func callParams(name, caller string, args json.RawMessage) map[string]any { +// callParams is a `tools/call`'s params: the tool, its arguments as the caller sent them, and in `_meta`, which the +// runtime writes and no caller can, the caller — only when the subject named one — and the id of the call in flight +// (mesh-sdk buscalls.MetaCall), which a ledger's apply names so the runtime stamps the caller it proved (novox/hq ADR +// 0597 §7). +func callParams(name, caller, call string, args json.RawMessage) map[string]any { params := map[string]any{"name": name, "arguments": args} + meta := map[string]any{} if caller != "" { - params["_meta"] = map[string]any{MetaCaller: caller} + meta[MetaCaller] = caller + } + if call != "" { + meta[buscalls.MetaCall] = call + } + if len(meta) > 0 { + params["_meta"] = meta } return params } @@ -101,6 +110,9 @@ type Bus interface { BusState(method string, params json.RawMessage) (json.RawMessage, error) // BusStateWatch hands deliver each change of an objects or a ledger watch, the current ones first. BusStateWatch(method string, params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error) + // Begin says a tool call of the module is in flight for the caller the runtime proved, and answers its id, which the + // bundle is handed in the call's `_meta`, and the end to call once the call is answered (novox/hq ADR 0597 §7). + Begin(caller string) (id string, end func()) } // busStateCalls are the methods of mesh-sdk buscalls a bundle asks and BusState answers. @@ -659,7 +671,9 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an if len(args) == 0 || string(args) == "null" { args = json.RawMessage("{}") } - res, err := asking("tools/call", callParams(full, caller, args), CallTimeout) + call, end := mesh.Begin(caller) + defer end() + res, err := asking("tools/call", callParams(full, caller, call, args), CallTimeout) if err != nil { return nil, err } diff --git a/node-tools/internal/ledger/ledger.go b/node-tools/internal/ledger/ledger.go index f39b889..3088a6f 100644 --- a/node-tools/internal/ledger/ledger.go +++ b/node-tools/internal/ledger/ledger.go @@ -81,8 +81,10 @@ type Snapshots interface { // Ledger is one ledger of one module: its types and where it is kept. type Ledger struct { - Module string - Name string + Module string + Name string + // Account is the owning module's authenticated account, which every operation it applies is stamped with. + Account string Types []catalogue.LedgerType History History State State diff --git a/node-tools/internal/runtime/busstate.go b/node-tools/internal/runtime/busstate.go index 905d3f0..bd72a15 100644 --- a/node-tools/internal/runtime/busstate.go +++ b/node-tools/internal/runtime/busstate.go @@ -65,6 +65,9 @@ func (b *moduleBus) BusState(method string, params json.RawMessage) (json.RawMes if err = decode(method, params, &r); err == nil { answer, err = conn.WorkQueueDead(b.module, r) } + case buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerQuery, buscalls.LedgerRebuild, + buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag: + answer, err = b.ledgerCall(method, params) default: return nil, fmt.Errorf("the runtime answers no %s", method) } @@ -89,6 +92,8 @@ func (b *moduleBus) BusStateWatch(method string, params json.RawMessage, deliver } return deliver(raw) }) + case buscalls.LedgerWatch: + return b.ledgerWatch(params, deliver) } return nil, fmt.Errorf("the runtime answers no %s", method) } diff --git a/node-tools/internal/runtime/caller_test.go b/node-tools/internal/runtime/caller_test.go index 8228231..7958b39 100644 --- a/node-tools/internal/runtime/caller_test.go +++ b/node-tools/internal/runtime/caller_test.go @@ -13,6 +13,7 @@ import ( type saidCaller struct { Caller string `json:"caller"` + Call string `json:"call"` Arguments map[string]any `json:"arguments"` } @@ -36,6 +37,7 @@ func TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller(t *testing.T) asker.Logf = func(string, ...any) {} defer asker.Close() + calls := map[string]bool{} for _, key := range []string{"ledger.who", "ledger.who@anchor", "seat:node-desk.who@anchor"} { var got saidCaller mt.Until(t, func() error { @@ -49,6 +51,11 @@ func TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller(t *testing.T) if got.Caller != "shanks.node-tools" { t.Errorf("%s reached the module with caller %q, want the asker's bus user", key, got.Caller) } + // And the id of the call in flight, one per call, which a ledger's apply names (novox/hq ADR 0597 §7). + if got.Call == "" || calls[got.Call] { + t.Errorf("%s reached the module with call id %q, want one of its own", key, got.Call) + } + calls[got.Call] = true if got.Arguments["caller"] != "novox.node-tools" { t.Errorf("%s: the arguments did not reach the module as sent: %v", key, got.Arguments) } diff --git a/node-tools/internal/runtime/ledgers.go b/node-tools/internal/runtime/ledgers.go index 3b5b053..7c15a1b 100644 --- a/node-tools/internal/runtime/ledgers.go +++ b/node-tools/internal/runtime/ledgers.go @@ -1,9 +1,12 @@ package runtime import ( + "crypto/rand" + "encoding/hex" "encoding/json" "errors" "fmt" + "sync" "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -35,6 +38,59 @@ type ledgerSource struct { writer func(module string) (*bus.Conn, error) } +// callsInFlight are the tool calls the runtime is serving, each with the module whose bundle serves it and the caller +// the runtime proved from the subject the call arrived on (bus.CalledSubject, novox/hq issue 365). +type callsInFlight struct { + mu sync.Mutex + of map[string]callInFlight +} + +type callInFlight struct{ module, caller string } + +// begin keeps a call of module in flight for caller, under an id no bundle can guess, until end. +func (f *callsInFlight) begin(module, caller string) (string, func()) { + var raw [16]byte + _, _ = rand.Read(raw[:]) + id := hex.EncodeToString(raw[:]) + f.mu.Lock() + if f.of == nil { + f.of = map[string]callInFlight{} + } + f.of[id] = callInFlight{module, caller} + f.mu.Unlock() + return id, func() { + f.mu.Lock() + delete(f.of, id) + f.mu.Unlock() + } +} + +// authorOf is the author an apply of module naming call is stamped with (ADR 0597 §7): the module's account, and the +// caller the runtime proved for that call; naming no call, the account alone. A call not in flight, or another +// module's, is refused, naming it. Nothing the bundle writes is taken as the author. +func (f *callsInFlight) authorOf(module, account, call string) (buscalls.Author, error) { + author := buscalls.Author{Account: account} + if call == "" { + return author, nil + } + f.mu.Lock() + inFlight, ok := f.of[call] + f.mu.Unlock() + switch { + case !ok: + return author, fmt.Errorf("the apply names the call %s, which is not a tool call in flight here; an apply names the "+ + "call it serves as its tools/call's _meta %s gave it, or none (novox/hq ADR 0597 §7)", call, buscalls.MetaCall) + case inFlight.module != module: + return author, fmt.Errorf("the apply names the call %s, which is another module's tool call; %s names only its own "+ + "(novox/hq ADR 0597 §7)", call, module) + } + author.Caller = inFlight.caller + return author, nil +} + +// Begin keeps one of the module's tool calls in flight for the caller the runtime proved (launch.Bus). +func (b *moduleBus) Begin(caller string) (string, func()) { return b.all.calls.begin(b.module, caller) } + // ledger opens one of a module's ledgers by its local name. func (c *consumers) ledger(module, local string) (*ledger.Ledger, error) { b, err := c.conn.Binding(module, local, catalogue.StorageLedger) @@ -56,8 +112,96 @@ func (c *consumers) ledger(module, local string) (*ledger.Ledger, error) { } } history, state, snapshots := c.conn.LedgerStores(b, writer) - return &ledger.Ledger{Module: module, Name: local, Types: ts, History: history, State: state, Snapshots: snapshots, - Indexes: c.conn.LedgerIndexes(b), Logf: c.logf}, nil + // The owning module's authenticated account: the user of the connection its history is published on. + account := module + if writer != nil && writer.Credential().User != "" { + account = writer.Credential().User + } + return &ledger.Ledger{Module: module, Name: local, Account: account, Types: ts, History: history, State: state, + Snapshots: snapshots, Indexes: c.conn.LedgerIndexes(b), Logf: c.logf}, nil +} + +// ledgerCall answers a bundle's ledger call, as the module whose bundle asks. +func (b *moduleBus) ledgerCall(method string, params json.RawMessage) (any, error) { + var named struct { + Ledger string `json:"ledger"` + } + if err := decode(method, params, &named); err != nil { + return nil, err + } + if method == buscalls.LedgerApply { + var r buscalls.ApplyRequest + if err := decode(method, params, &r); err != nil { + return nil, err + } + // The call is judged first: a forged one is refused before anything of the ledger is touched. + if _, err := b.all.calls.authorOf(b.module, "", r.Call); err != nil { + return nil, err + } + l, err := b.all.ledger(b.module, r.Ledger) + if err != nil { + return nil, err + } + author, err := b.all.calls.authorOf(b.module, l.Account, r.Call) + if err != nil { + return nil, err + } + return l.Apply(r, author) + } + l, err := b.all.ledger(b.module, named.Ledger) + if err != nil { + return nil, err + } + switch method { + case buscalls.LedgerGet: + var r buscalls.InstanceRequest + _ = json.Unmarshal(params, &r) + return l.Get(r) + case buscalls.LedgerHistory: + var r buscalls.HistoryRequest + _ = json.Unmarshal(params, &r) + return l.ReadHistory(r) + case buscalls.LedgerQuery: + var r buscalls.QueryRequest + _ = json.Unmarshal(params, &r) + return l.Query(r) + case buscalls.LedgerRebuild: + var r buscalls.RebuildRequest + _ = json.Unmarshal(params, &r) + return l.Rebuild(r) + case buscalls.LedgerReindex: + var r buscalls.ReindexRequest + _ = json.Unmarshal(params, &r) + return l.Reindex(r) + case buscalls.LedgerSnapshot: + var r buscalls.InstanceRequest + _ = json.Unmarshal(params, &r) + return l.Snapshot(r) + case buscalls.LedgerLag: + var r buscalls.LedgerRequest + _ = json.Unmarshal(params, &r) + return l.Lag(r) + } + return nil, fmt.Errorf("the runtime answers no %s", method) +} + +// ledgerWatch hands a bundle each change of its ledger's states. +func (b *moduleBus) ledgerWatch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) { + var r buscalls.LedgerWatchRequest + if err := decode(buscalls.LedgerWatch, params, &r); err != nil { + return nil, err + } + l, err := b.all.ledger(b.module, r.Ledger) + if err != nil { + return nil, err + } + return l.Watch(r, func(change buscalls.InstanceChange) error { + raw, err := json.Marshal(change) + if err != nil { + return err + } + return deliver(raw) + }) } // hasLedger is whether a module's membership binds a ledger. diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go index 15ba59f..789cece 100644 --- a/node-tools/internal/runtime/ledgers_test.go +++ b/node-tools/internal/runtime/ledgers_test.go @@ -1,15 +1,20 @@ package runtime import ( + "context" "encoding/json" "errors" "sort" "strings" "testing" + "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/membership" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-tools/node-tools/internal/bus" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) @@ -72,3 +77,101 @@ func TestTheLedgerRepairsAreToolsOfEveryLedgerOwner(t *testing.T) { }) }) } + +// aLedger raises one ledger of module m on the bus as the bus module would create it, binds it in m's membership, and +// answers the runtime's buses for m and for other, its types given as a test gives them. +func aLedger(t *testing.T) (m, other *moduleBus, mesh *mt.Mesh) { + t.Helper() + mesh = mt.New(t) + nc, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + ctx := context.Background() + for _, cfg := range []jetstream.StreamConfig{ + {Name: "LEDGER_m_records", Subjects: []string{"mesh.ledger.m.records.>"}}, + {Name: "SNAPSHOT_m_records", Subjects: []string{"mesh.ledger-snapshot.m.records.>"}, MaxMsgsPerSubject: 1}, + } { + if _, err := js.CreateStream(ctx, cfg); err != nil { + t.Fatal(err) + } + } + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "m_records"}); err != nil { + t.Fatal(err) + } + mem := mt.MembershipOf("m", "anchor", false, nil) + mem.Bindings = []membership.BusBinding{{Local: "records", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger, + History: "LEDGER_m_records", Subject: "mesh.ledger.m.records", State: "m_records", Snapshot: "mesh.ledger-snapshot.m.records"}} + mesh.Issue(t, mem) + mesh.Issue(t, mt.MembershipOf("other", "anchor", false, nil)) + conn := connect(t, "node-tools", "anchor") + conn.Follow("m") + conn.Follow("other") + mt.Until(t, func() error { + if conn.Membership("m") == nil || conn.Membership("other") == nil { + return errors.New("not issued yet") + } + return nil + }) + all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}, ledgers: ledgerSource{ + types: func(string, string) ([]catalogue.LedgerType, error) { + return []catalogue.LedgerType{{Name: "decision", Schema: map[string]catalogue.LedgerField{"title": {Type: "string"}}, + Operations: []catalogue.LedgerOperation{{Name: "title", Effect: catalogue.EffectSetFields, Fields: []string{"title"}, Author: true}}}}, nil + }, + writer: func(string) (*bus.Conn, error) { return conn, nil }, + }} + return all.forModule("m").(*moduleBus), all.forModule("other").(*moduleBus), mesh +} + +// 1966/AC17, 1966/AC18: the author an apply is stamped with is the caller the runtime proved for the call in flight it +// names, whatever author the bundle wrote; another module's call, or one not in flight, is refused naming it (ADR 0597 §7). +func TestAnApplyIsStampedWithTheCallerTheRuntimeProved(t *testing.T) { + m, other, _ := aLedger(t) + call, end := m.Begin("op@shanks") + defer end() + theirs, endTheirs := other.Begin("op@shanks") + defer endTheirs() + forged := `"body":{"title":"T","author":{"account":"x","caller":"forged"}}` + for _, c := range []struct{ name, params, caller, refusal string }{ + {"1966/AC17 a call in flight stamps its proven caller", `{"ledger":"records","instance":"decision/1","operation":"title",` + forged + `,"call":"` + call + `","author":"forged"}`, "op@shanks", ""}, + {"naming no call stamps the account alone", `{"ledger":"records","instance":"decision/2","operation":"title",` + forged + `}`, "", ""}, + {"1966/AC18 another module's call is refused, naming it", `{"ledger":"records","instance":"decision/3","operation":"title",` + forged + `,"call":"` + theirs + `"}`, "", "the call " + theirs + ", which is another module's"}, + {"a call not in flight is refused, naming it", `{"ledger":"records","instance":"decision/3","operation":"title",` + forged + `,"call":"c0ffee"}`, "", "the call c0ffee, which is not a tool call in flight"}, + } { + t.Run(c.name, func(t *testing.T) { + var req struct { + Instance string `json:"instance"` + } + _ = json.Unmarshal([]byte(c.params), &req) + _, err := m.BusState(buscalls.LedgerApply, json.RawMessage(c.params)) + if c.refusal != "" { + if err == nil || !strings.Contains(err.Error(), c.refusal) { + t.Fatalf("applied with %v, want refused with %q", err, c.refusal) + } + return + } + if err != nil { + t.Fatal(err) + } + raw, err := m.BusState(buscalls.LedgerHistory, json.RawMessage(`{"ledger":"records","instance":"`+req.Instance+`"}`)) + if err != nil { + t.Fatal(err) + } + var page buscalls.HistoryPage + _ = json.Unmarshal(raw, &page) + if len(page.Operations) != 1 || page.Operations[0].Author != (buscalls.Author{Account: "m", Caller: c.caller}) { + t.Fatalf("the history holds %+v, want the author {m %s}", page.Operations, c.caller) + } + state, _ := m.BusState(buscalls.LedgerGet, json.RawMessage(`{"ledger":"records","instance":"`+req.Instance+`"}`)) + if !strings.Contains(string(state), `"author":{"account":"m"`) || strings.Contains(string(state), "forged") { + t.Fatalf("the state took the bundle's author: %s", state) + } + }) + } + end() + if _, err := m.BusState(buscalls.LedgerApply, json.RawMessage(`{"ledger":"records","instance":"decision/4","operation":"title","call":"`+call+`"}`)); err == nil { + t.Fatal("a call that ended was still taken as in flight") + } +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 8df61b5..b7d6e41 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -744,6 +744,8 @@ type consumers struct { provisioned provisionWatches // ledgers is how a module's ledger is opened (ledgers.go). ledgers ledgerSource + // calls are the tool calls in flight, by the id each bundle was handed (ledgers.go, novox/hq ADR 0597 §7). + calls callsInFlight } type moduleEvents struct { diff --git a/node-tools/test/fixtures/caller-echo.py b/node-tools/test/fixtures/caller-echo.py index 16706cd..a7f7c3e 100755 --- a/node-tools/test/fixtures/caller-echo.py +++ b/node-tools/test/fixtures/caller-echo.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 # A tools bundle that says who the runtime said called it (novox/hq issue 365): MCP over stdio, no SDK. Its tool -# and its seat verb answer the caller the call's `_meta` named — "" when it named none — and the arguments as given. +# and its seat verb answer the caller the call's `_meta` named — "" when it named none —, the id of the call in flight +# it named (novox/hq ADR 0597 §7), and the arguments as given. import json, sys def say(m): @@ -18,7 +19,7 @@ for line in sys.stdin: say({"id": rid, "result": {"tools": TOOLS}}) elif m == "tools/call": meta = p.get("_meta") or {} - said = {"caller": meta.get("mesh/caller", ""), "arguments": p.get("arguments") or {}} + said = {"caller": meta.get("mesh/caller", ""), "call": meta.get("mesh/call", ""), "arguments": p.get("arguments") or {}} say({"id": rid, "result": {"content": [{"type": "text", "text": json.dumps(said)}]}}) elif rid is not None: say({"id": rid, "result": {}})