diff --git a/node-tools/go.mod b/node-tools/go.mod index 54e4d8b..e9db0f0 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -7,6 +7,8 @@ require ( git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 diff --git a/node-tools/internal/approval/act.go b/node-tools/internal/approval/act.go new file mode 100644 index 0000000..7f9445d --- /dev/null +++ b/node-tools/internal/approval/act.go @@ -0,0 +1,291 @@ +package approval + +import ( + "bytes" + "encoding/json" + "fmt" + "math" + "sort" + "strconv" + "strings" + "unicode/utf8" + + "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + "git.novox.be/novox/mesh-sdk/go/outward" +) + +// Call is one call the runner serves, as the gate judges it: a module's tool, or a seat verb its module serves. +type Call struct { + // Address is the address as called: `.` or `.`. + Address string + // Module is the module whose bundle serves it; Seat the seat of a verb, empty for a tool; Name the tool or the verb. + Module, Seat, Name string + // Fingerprint is the build the bundle serves, from its composed environment (§5). + Fingerprint string + // Caller is the caller the bus proved. + Caller string + // Args are the call's arguments, a JSON object. + Args json.RawMessage + // Description is the tool's or verb's own description, whose first line the ask explains it by. + Description string +} + +// MostArguments is the most a marked call's arguments may hold: more is refused, asking nothing (§9). +const MostArguments = 16 * 1024 + +// arguments are a call's arguments by name; a body that is not an object is refused. +func arguments(raw json.RawMessage) (map[string]json.RawMessage, error) { + args := map[string]json.RawMessage{} + if len(bytes.TrimSpace(raw)) == 0 || string(bytes.TrimSpace(raw)) == "null" { + return args, nil + } + if err := json.Unmarshal(raw, &args); err != nil { + return nil, fmt.Errorf("its arguments are not a JSON object: %v", err) + } + return args, nil +} + +// refusedUnasked says why a marked call is refused at once, asking nothing (§9): its arguments hold a secret's shape, or are +// more than MostArguments. +func refusedUnasked(raw json.RawMessage) string { + if len(raw) > MostArguments { + return fmt.Sprintf("its arguments are %d bytes, more than the %d a call that needs the operator's approval may carry", len(raw), MostArguments) + } + if r, ok := outward.CheckSecret(string(raw)); !ok { + return fmt.Sprintf("its arguments hold %s, which no ask may carry", r.What) + } + return "" +} + +// ActOf is the act a marked call binds (§8): `address`, `node`, `module` and `tool` or `seat` and `verb`, `build`, and +// `arg.` for each argument given, a string as itself and any other value as its JSON in RFC 8785's canonical form. +func ActOf(c Call, node string) (operatorchannelclient.Act, error) { + act := operatorchannelclient.Act{"address": c.Address, "node": node, "build": c.Fingerprint} + if c.Seat == "" { + act["module"], act["tool"] = c.Module, c.Name + } else { + act["seat"], act["verb"] = c.Seat, c.Name + } + args, err := arguments(c.Args) + if err != nil { + return nil, err + } + for name, raw := range args { + var s string + if json.Unmarshal(raw, &s) == nil { + act["arg."+name] = s + continue + } + canon, err := Canonical(raw) + if err != nil { + return nil, fmt.Errorf("the argument %s cannot be put in canonical form: %v", name, err) + } + act["arg."+name] = canon + } + return act, nil +} + +// Canonical is a JSON value in RFC 8785's canonical form: no white space, an object's members sorted by their names' UTF-16 +// code units, a number as ECMAScript writes a double, a string with only the escapes the RFC names. +func Canonical(raw json.RawMessage) (string, error) { + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() + var v any + if err := dec.Decode(&v); err != nil { + return "", err + } + var b strings.Builder + if err := canonicalValue(&b, v); err != nil { + return "", err + } + return b.String(), nil +} + +func canonicalValue(b *strings.Builder, v any) error { + switch x := v.(type) { + case nil: + b.WriteString("null") + case bool: + b.WriteString(strconv.FormatBool(x)) + case json.Number: + f, err := strconv.ParseFloat(string(x), 64) + if err != nil || math.IsInf(f, 0) || math.IsNaN(f) { + return fmt.Errorf("the number %s is not a double", x) + } + b.WriteString(es6Number(f)) + case string: + canonicalString(b, x) + case []any: + b.WriteByte('[') + for i, e := range x { + if i > 0 { + b.WriteByte(',') + } + if err := canonicalValue(b, e); err != nil { + return err + } + } + b.WriteByte(']') + case map[string]any: + names := make([]string, 0, len(x)) + for k := range x { + names = append(names, k) + } + sort.Slice(names, func(i, j int) bool { return utf16Less(names[i], names[j]) }) + b.WriteByte('{') + for i, k := range names { + if i > 0 { + b.WriteByte(',') + } + canonicalString(b, k) + b.WriteByte(':') + if err := canonicalValue(b, x[k]); err != nil { + return err + } + } + b.WriteByte('}') + default: + return fmt.Errorf("a value of type %T", v) + } + return nil +} + +// es6Number is a double as ECMAScript's Number.prototype.toString writes it (RFC 8785 §3.2.2.3). +func es6Number(f float64) string { + if f == 0 { + return "0" + } + abs := math.Abs(f) + if abs >= 1e-6 && abs < 1e21 { + return strconv.FormatFloat(f, 'f', -1, 64) + } + s := strconv.FormatFloat(f, 'e', -1, 64) // 1e+21, 1.5e-07 + mant, exp, _ := strings.Cut(s, "e") + sign, digits := exp[:1], strings.TrimLeft(exp[1:], "0") + return mant + "e" + sign + digits +} + +func canonicalString(b *strings.Builder, s string) { + b.WriteByte('"') + for _, r := range s { + switch { + case r == '"': + b.WriteString(`\"`) + case r == '\\': + b.WriteString(`\\`) + case r == '\b': + b.WriteString(`\b`) + case r == '\f': + b.WriteString(`\f`) + case r == '\n': + b.WriteString(`\n`) + case r == '\r': + b.WriteString(`\r`) + case r == '\t': + b.WriteString(`\t`) + case r < 0x20: + fmt.Fprintf(b, `\u%04x`, r) + default: + b.WriteRune(r) + } + } + b.WriteByte('"') +} + +func utf16Less(a, b string) bool { + ua, ub := utf16Of(a), utf16Of(b) + for i := 0; i < len(ua) && i < len(ub); i++ { + if ua[i] != ub[i] { + return ua[i] < ub[i] + } + } + return len(ua) < len(ub) +} + +func utf16Of(s string) []uint16 { + out := make([]uint16, 0, utf8.RuneCountInString(s)) + for _, r := range s { + if r >= 0x10000 { + r -= 0x10000 + out = append(out, uint16(0xD800+(r>>10)), uint16(0xDC00+(r&0x3FF))) + } else { + out = append(out, uint16(r)) + } + } + return out +} + +// Fingerprint is how an act is shown: its digest's first 24 hexadecimal digits in groups of four (§10). +func Fingerprint(digest string) string { + hex := strings.TrimPrefix(digest, "sha256:") + if len(hex) > 24 { + hex = hex[:24] + } + var groups []string + for len(hex) > 4 { + groups, hex = append(groups, hex[:4]), hex[4:] + } + return strings.Join(append(groups, hex), " ") +} + +// The bounds of the arguments shown in an ask (§11). +const ( + ShownBytes, ShownLines = 2000, 40 + PartBytes, PartLines = 1200, 20 +) + +// shown is a call's arguments as an ask shows them (§11), one per line as "name: value", and whether all were shown whole: +// an argument whole where outward.Check passes it, else its shape, ‹path›, ‹address› or ‹withheld›; the whole when it is at +// most ShownBytes and ShownLines, else its first PartLines lines and PartBytes bytes, marked NOT SHOWN IN FULL. +func shown(raw json.RawMessage) (string, bool) { + args, _ := arguments(raw) + names := make([]string, 0, len(args)) + for n := range args { + names = append(names, n) + } + sort.Strings(names) + whole := true + var lines []string + for _, n := range names { + var value string + if json.Unmarshal(args[n], &value) != nil { + value = string(args[n]) + } + if r, ok := outward.Check(value); !ok { + value, whole = shapeOf(r), false + } + lines = append(lines, n+": "+value) + } + text := strings.Join(lines, "\n") + if all := strings.Split(text, "\n"); len(text) > ShownBytes || len(all) > ShownLines { + if len(all) > PartLines { + all = all[:PartLines] + } + text = strings.Join(all, "\n") + if len(text) > PartBytes { + text = cutBytes(text, PartBytes) + } + text += "\nNOT SHOWN IN FULL" + whole = false + } + return text, whole +} + +// shapeOf is what an argument outward.Check refuses is shown as. +func shapeOf(r outward.Refusal) string { + switch r.Class { + case "path": + return "‹path›" + case "address": + return "‹address›" + } + return "‹withheld›" +} + +// cutBytes is s at most n bytes, cut at a character's end. +func cutBytes(s string, n int) string { + for n > 0 && !utf8.RuneStart(s[n]) { + n-- + } + return s[:n] +} diff --git a/node-tools/internal/approval/approval_test.go b/node-tools/internal/approval/approval_test.go new file mode 100644 index 0000000..42b25ff --- /dev/null +++ b/node-tools/internal/approval/approval_test.go @@ -0,0 +1,276 @@ +package approval + +import ( + "encoding/json" + "sort" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + seat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// mem is the runner's state approvals as a fixture; fail makes every read and write fail, as a bus refusing them. +type mem struct { + mu sync.Mutex + vals map[string]json.RawMessage + revs map[string]uint64 + n uint64 + fail error +} + +func newMem() *mem { return &mem{vals: map[string]json.RawMessage{}, revs: map[string]uint64{}} } + +func (m *mem) Get(k string) (*stdio.StateEntry, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.fail != nil { + return nil, m.fail + } + v, ok := m.vals[k] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: k, Value: v, Revision: m.revs[k]}, nil +} +func (m *mem) Create(k string, v any) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.fail != nil { + return 0, m.fail + } + if _, ok := m.vals[k]; ok { + return 0, stdio.ErrChanged + } + return m.put(k, v) +} +func (m *mem) Update(k string, v any, rev uint64) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.fail != nil { + return 0, m.fail + } + if rev == 0 || m.revs[k] != rev { + return 0, stdio.ErrChanged + } + return m.put(k, v) +} +func (m *mem) Put(k string, v any) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.fail != nil { + return 0, m.fail + } + return m.put(k, v) +} +func (m *mem) Keys() ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.fail != nil { + return nil, m.fail + } + var out []string + for k := range m.vals { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} +func (m *mem) put(k string, v any) (uint64, error) { + raw, ok := v.(json.RawMessage) + if !ok { + raw, _ = json.Marshal(v) + } + m.n++ + m.vals[k], m.revs[k] = raw, m.n + return m.n, nil +} + +// world is a gate on a fixture mesh: the catalog half, the router's records as the router writes them (mesh-catalog's +// messenger, asks.go), the asks published, the clock. +type world struct { + g *Gate + state *mem + cat *fakeCatalog + records map[string]seat.AsksByAskerIDValue // by `.` + takes bool // the router takes an ask as it is published + asked []seat.Ask + now time.Time + runs int + findings *findings +} + +func newWorld() *world { + shell := entry("src2:1", []string{"node-login-shell.execute"}, map[string]any{"needs-approval": []any{"wipe"}}) + w := &world{state: newMem(), records: map[string]seat.AsksByAskerIDValue{}, takes: true, findings: &findings{}, + now: time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC), + cat: &fakeCatalog{revision: "7", byFp: map[string]catalogseat.CatalogEntry{"src2:1": shell}, + newest: map[string]catalogseat.CatalogEntry{"shell": shell}, + defs: map[string]seatdef.Definition{"node-login-shell": {Seat: "node-login-shell", + Verbs: []seatdef.Verb{{Name: "execute", NeedsApproval: true}}}}}} + w.g = New(Deps{Node: "n1", Runner: "node-tools", Catalog: w.cat, State: w.state, Findings: w.findings, + Now: func() time.Time { return w.now }, + After: func(d time.Duration) <-chan time.Time { + w.now = w.now.Add(d) + c := make(chan time.Time, 1) + c <- w.now + return c + }, + Publish: func(module, subject string, body any, id string) error { + if subject != seat.AcceptSubjectAsk(module) { + return nil + } + a := body.(seat.Ask) + w.asked = append(w.asked, a) + if w.takes { + w.records[module+"."+a.ID] = seat.AsksByAskerIDValue{Asker: module, Key: module + "." + a.ID, State: "open", + Headline: a.Headline, Expires: a.Expires, Opened: operatorchannelclient.FormatTime(w.now), OfferedOn: []string{"telegram"}} + } + return nil + }, + Record: func(module, id string) (*stdio.StateEntry, error) { + rec, ok := w.records[module+"."+id] + if !ok { + return nil, nil + } + raw, _ := json.Marshal(rec) + return &stdio.StateEntry{Key: module + "." + id, Value: raw}, nil + }}) + _ = w.g.Unserved("shell", "src2:1", []string{"wipe", "tidy"}) + return w +} + +// call is a fixture call of the marked tool wipe, or of another tool of the module. +func call(name, args string) Call { + return Call{Address: "shell." + name, Module: "shell", Name: name, Fingerprint: "src2:1", Caller: "agent", Args: json.RawMessage(args), + Description: "Wipes a scratch directory.\nMore words."} +} + +// theActAndItsAsk is §8's and §11's rows: the act's fields, and how an ask shows a call's arguments. +func theActAndItsAsk(t *testing.T) { + base := call("wipe", `{"dir":"scratch","n":1}`) + digest := func(c Call) string { + act, err := ActOf(c, "n1") + if err != nil { + t.Fatal(err) + } + d, _ := operatorchannelclient.ActDigest(act) + return d + } + act, _ := ActOf(base, "n1") + want := operatorchannelclient.Act{"address": "shell.wipe", "node": "n1", "module": "shell", "tool": "wipe", "build": "src2:1", + "arg.dir": "scratch", "arg.n": "1"} + if len(act) != len(want) { + t.Fatalf("act %v", act) + } + for k, v := range want { + if act[k] != v { + t.Errorf("act[%s] = %q, want %q", k, act[k], v) + } + } + seen := map[string]string{digest(base): "the call"} + for name, c := range map[string]Call{ + "an omitted argument": call("wipe", `{"dir":"scratch"}`), + "another number": call("wipe", `{"dir":"scratch","n":2}`), + "an object": call("wipe", `{"dir":{"name":"scratch"},"n":1}`), + "a changed build": func() Call { c := base; c.Fingerprint = "src2:2"; return c }(), + "another node's seat": {Address: "node-login-shell.execute", Module: "shell", Seat: "node-login-shell", Name: "execute", Fingerprint: "src2:1"}, + } { + d := digest(c) + if other, dup := seen[d]; dup { + t.Errorf("%s gives the digest of %s", name, other) + } + seen[d] = name + } + if got, _ := Canonical(json.RawMessage(`{"b":[1.0,2.5e-7,1e21],"a":"é<"}`)); got != `{"a":"é<","b":[1,2.5e-7,1e+21]}` { + t.Errorf("canonical form %s", got) + } + // §11: a path, an address and a long argument are shown in their shape and cut. + text, whole := shown(json.RawMessage(`{"dir":"/home/someone/scratch","host":"10.0.0.7","ok":"yes"}`)) + if whole || text != "dir: ‹path›\nhost: ‹address›\nok: yes" { + t.Errorf("shown %q whole %v", text, whole) + } + long, _ := json.Marshal(map[string]string{"text": strings.Repeat("word ", 500)}) + if text, whole := shown(long); whole || !strings.HasSuffix(text, "NOT SHOWN IN FULL") || len(text) > PartBytes+len("\nNOT SHOWN IN FULL") { + t.Errorf("a long argument shown whole %v, %d bytes", whole, len(text)) + } + // §10: the headline, cut with … to 60 characters, ending Part shown. + for _, tc := range []struct{ name, node, want string }{ + {"wipe", "n1", "Run wipe on n1?"}, + {strings.Repeat("x", 50), "n1", "Run " + strings.Repeat("x", 50) + "?"}, + {strings.Repeat("y", 80), "n1", "Run " + strings.Repeat("y", 54) + "…?"}, + } { + if got := headlineOf(tc.name, tc.node, true); got != tc.want || len([]rune(got)) > 60 { + t.Errorf("headline %q, want %q", got, tc.want) + } + } + if got := headlineOf("wipe", "n1", false); got != "Run wipe on n1? Part shown" { + t.Errorf("headline %q", got) + } +} + +// Design 46 7-approval-mark §22: the refusals and answers of a marked call, on the fixture mesh, the router's record and +// the warrant written as the router writes them. +func TestTheRefusalsAndAnswers(t *testing.T) { + t.Run("§8, §10, §11: the act's fields give each its digest, and the ask shows the arguments", theActAndItsAsk) + for _, tc := range []struct { + name string + run func(t *testing.T, w *world) + }{ + {"§9: a secret in the arguments is refused asking nothing", func(t *testing.T, w *world) { + _, err := w.g.firstCall(call("wipe", `{"token":"ghp_0123456789abcdefghijABCDEFGHIJ012345"}`)) + if err == nil || !strings.Contains(err.Error(), "which no ask may carry") || len(w.asked) != 0 { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + }}, + {"§9: more than 16 KiB of arguments is refused asking nothing", func(t *testing.T, w *world) { + big, _ := json.Marshal(map[string]string{"text": strings.Repeat("a b ", 5000)}) + _, err := w.g.firstCall(call("wipe", string(big))) + if err == nil || !strings.Contains(err.Error(), "more than the 16384") || len(w.asked) != 0 { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + }}, + {"§10, §12: the published ask's fields, and Not run with the router's record", func(t *testing.T, w *world) { + _, err := w.g.firstCall(call("wipe", `{"dir":"scratch"}`)) + if err == nil || !strings.HasPrefix(err.Error(), "Not run: this needs the operator's approval. Asked on telegram at ") || + !strings.Contains(err.Error(), "make the same call again within 30 minutes of the approval; the ask expires at ") { + t.Fatalf("answer %v", err) + } + if len(w.asked) != 1 { + t.Fatalf("asked %d", len(w.asked)) + } + a := w.asked[0] + act, _ := ActOf(call("wipe", `{"dir":"scratch"}`), "n1") + d, _ := operatorchannelclient.ActDigest(act) + if a.About != d || a.Who != "operator" || a.OnExpiry != OnExpiry || a.Headline != "Run wipe on n1?" || + a.Expires != operatorchannelclient.FormatTime(time.Date(2026, 11, 12, 11, 0, 0, 0, time.UTC)) || len(a.Options) != 2 || + a.Options[0].ID != "approve" || a.Options[1].ID != "decline" || a.Options[0].Binds != d || a.Options[1].Binds != d || + a.Options[1].Level != "approve" { + t.Errorf("ask %+v", a) + } + if !strings.HasPrefix(a.Explanation, "Wipes a scratch directory.\nOn n1, called through the mesh's tools, which cannot tell who called.\ndir: scratch\nFingerprint ") { + t.Errorf("explanation %q", a.Explanation) + } + var k operatorchannelclient.Kept + _ = json.Unmarshal(w.state.vals["n1.ask."+a.ID], &k) + if k.Act["arg.dir"] != "scratch" || string(k.Arguments) != `{"dir":"scratch"}` || string(w.state.vals["n1.act."+d]) != `"`+a.ID+`"` { + t.Errorf("kept %+v, act key %s", k, w.state.vals["n1.act."+d]) + } + }}, + {"§12: with no record within 5 seconds, Not run, asked; not yet taken by the router", func(t *testing.T, w *world) { + w.takes = false + start := w.now + _, err := w.g.firstCall(call("wipe", `{}`)) + if err == nil || !strings.HasPrefix(err.Error(), "Not run, asked; not yet taken by the router (ask ") || w.now.Sub(start) > RecordWithin { + t.Fatalf("answer %v after %s", err, w.now.Sub(start)) + } + }}, + } { + t.Run(tc.name, func(t *testing.T) { tc.run(t, newWorld()) }) + } +} diff --git a/node-tools/internal/approval/ask.go b/node-tools/internal/approval/ask.go new file mode 100644 index 0000000..a183f3f --- /dev/null +++ b/node-tools/internal/approval/ask.go @@ -0,0 +1,230 @@ +package approval + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + seat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// StateName is the runner's own state its asks and claims are kept in, per machine (§13). +const StateName = "approvals" + +// Store is the runner's state `approvals` (§13): a key's value, a create that refuses a key that exists and an update by +// compare-and-set, each failing with stdio.ErrChanged when another write came first; a put, and its keys. +type Store interface { + Get(key string) (*stdio.StateEntry, error) + Create(key string, value any) (uint64, error) + Update(key string, value any, revision uint64) (uint64, error) + Put(key string, value any) (uint64, error) + Keys() ([]string, error) +} + +// The bounds of an ask (§10, §12, §13, §14, §17). +const ( + AskLasts = 60 * time.Minute + RecordWithin = 5 * time.Second + RunWithin = 30 * time.Minute + DeclinedFor = 10 * time.Minute + MostKeptValue = 256 * 1024 + OnExpiry = "nothing runs; call again to ask anew" +) + +// The options every ask offers (§10), both at level approve, both binding the act's digest. +const ( + OptionApprove = "approve" + OptionDecline = "decline" +) + +// client is operator-channel's client asking under the marked module's name, its asks kept in the runner's state under +// this machine's keys, its publish and the router's record on the runner's own account (§7, §10). +func (g *Gate) client(module string) *operatorchannelclient.Client { + return &operatorchannelclient.Client{Module: module, State: g.d.State, KeyPrefix: g.d.Node + ".", Now: g.d.Now, + Publish: func(subject string, body any, id string) error { return g.d.Publish(module, subject, body, id) }, + Record: func(id string) (*stdio.StateEntry, error) { return g.d.Record(module, id) }} +} + +// actKey is the key naming the newest ask of an act (§13). +func (g *Gate) actKey(digest string) string { return g.d.Node + ".act." + digest } + +// askKey is the key the client keeps an ask under (§13). +func (g *Gate) askKey(id string) string { return g.d.Node + ".ask." + id } + +// newID is an ask's id: 128 random bits. +func newID() string { + var b [16]byte + _, _ = rand.Read(b[:]) + return hex.EncodeToString(b[:]) +} + +// askFor is the ask a marked call raises (§10, §11). +func (g *Gate) askFor(c Call, digest, id string) seat.Ask { + args, whole := shown(c.Args) + name := c.Name + if c.Seat != "" { + name = c.Seat + "." + c.Name + } + headline := headlineOf(name, g.d.Node, whole) + first, _, _ := strings.Cut(strings.TrimSpace(c.Description), "\n") + if first == "" { + first = name + } + lines := []string{first, "On " + g.d.Node + ", called through the mesh's tools, which cannot tell who called."} + if args != "" { + lines = append(lines, args) + } + if !whole { + lines = append(lines, "The whole is read with mesh_approvals on "+g.d.Node+", whose fingerprint must be the one on the phone.") + } + lines = append(lines, "Fingerprint "+Fingerprint(digest)) + return seat.Ask{ID: id, About: digest, Who: operatorchannelclient.Operator, Headline: headline, + Explanation: strings.Join(lines, "\n"), Expires: operatorchannelclient.FormatTime(g.d.Now().Add(AskLasts)), OnExpiry: OnExpiry, + Options: []seat.Option{ + {ID: OptionApprove, Label: "Approve", Does: "the same call, made again within 30 minutes, runs once", + Level: operatorchannelclient.Approve, Binds: digest}, + {ID: OptionDecline, Label: "Decline", Does: "nothing runs; the same call is not asked again for 10 minutes", + Level: operatorchannelclient.Approve, Binds: digest}}} +} + +// headlineOf is "Run on ?", "Run ?" when that is longer than the router allows, and the names cut with +// "…" when that too is longer; ending "Part shown" whenever an argument is not shown whole (§10, Q-HL). +func headlineOf(name, node string, whole bool) string { + suffix := "" + if !whole { + suffix = " Part shown" + } + max := operatorchannelclient.HeadlineLength + fits := func(s string) bool { return len([]rune(s)) <= max } + if h := "Run " + name + " on " + node + "?" + suffix; fits(h) { + return h + } + if h := "Run " + name + "?" + suffix; fits(h) { + return h + } + room := max - len([]rune("Run …?"+suffix)) + r := []rune(name) + if room < 1 { + room = 1 + } + if len(r) > room { + r = r[:room] + } + return "Run " + string(r) + "…?" + suffix +} + +// firstCall is a marked call whose act has no ask open: refused at once when its arguments may not be asked (§9), else +// asked (§10, §12). +func (g *Gate) firstCall(c Call) (any, error) { + if why := refusedUnasked(c.Args); why != "" { + return nil, g.refused(c, "", why) + } + act, err := ActOf(c, g.d.Node) + if err != nil { + return nil, g.refused(c, "", err.Error()) + } + digest, err := operatorchannelclient.ActDigest(act) + if err != nil { + return nil, g.refused(c, "", err.Error()) + } + return nil, g.raise(c, act, digest, "") +} + +// raise raises the ask of a marked call whose act has none open (§10): kept in the runner's state with the act and the whole +// arguments, published under the module's name, its act's key naming it; then the 5 seconds for the router's record and +// the first answer (§12), always an error, never a result. +func (g *Gate) raise(c Call, act operatorchannelclient.Act, digest, markedSince string) error { + id := newID() + a := g.askFor(c, digest, id) + cl := g.client(c.Module) + if size := keptSize(a, act, c.Args); size > MostKeptValue { + return g.refused(c, digest, fmt.Sprintf("its ask would keep %d bytes, more than the %d a value of the state %s holds", size, MostKeptValue, StateName)) + } + if err := cl.Ask(a); err != nil { + return g.refused(c, digest, "the ask could not be raised: "+err.Error()) + } + if err := g.keepCall(id, act, c.Args); err != nil { + return g.refused(c, digest, "the ask was raised and its call could not be kept: "+err.Error()) + } + if _, err := g.d.State.Put(g.actKey(digest), id); err != nil { + return g.refused(c, digest, "the ask was raised and its act could not be kept: "+err.Error()) + } + return g.firstAnswer(c, id, digest, markedSince) +} + +// keptSize is the bytes the kept ask would take. +func keptSize(a seat.Ask, act operatorchannelclient.Act, args json.RawMessage) int { + raw, _ := json.Marshal(operatorchannelclient.Kept{Ask: a, Act: act, Arguments: args}) + return len(raw) +} + +// keepCall writes the act and the whole arguments beside the ask the client kept (§13). +func (g *Gate) keepCall(id string, act operatorchannelclient.Act, args json.RawMessage) error { + e, err := g.d.State.Get(g.askKey(id)) + if err != nil || e == nil { + return fmt.Errorf("the kept ask %s cannot be read: %v", id, err) + } + var k operatorchannelclient.Kept + if err := json.Unmarshal(e.Value, &k); err != nil { + return err + } + k.Act, k.Arguments = act, args + _, err = g.d.State.Update(g.askKey(id), k, e.Revision) + return err +} + +// firstAnswer waits up to RecordWithin for the router's record of the ask and says the call was not run (§12). +func (g *Gate) firstAnswer(c Call, id, digest, markedSince string) error { + rec, _ := g.awaitRecord(c.Module, id) + since := "" + if markedSince != "" { + since = "; marked since revision " + markedSince + } + if rec == nil { + return fmt.Errorf("Not run, asked; not yet taken by the router (ask %s, fingerprint %s); make the same call again later%s", + id, Fingerprint(digest), since) + } + return fmt.Errorf("Not run: this needs the operator's approval. Asked on %s at %s (ask %s, fingerprint %s); make the same call "+ + "again within 30 minutes of the approval; the ask expires at %s%s", orNone(strings.Join(rec.OfferedOn, ", ")), rec.Opened, + id, Fingerprint(digest), rec.Expires, since) +} + +// record is the router's record of an ask under the module's name, nil when it holds none. +func (g *Gate) record(module, id string) (*seat.AsksByAskerIDValue, error) { + e, err := g.d.Record(module, id) + if err != nil || e == nil { + return nil, err + } + var rec seat.AsksByAskerIDValue + if err := json.Unmarshal(e.Value, &rec); err != nil { + return nil, fmt.Errorf("the router's record of %s cannot be read: %w", id, err) + } + return &rec, nil +} + +// awaitRecord reads the router's record of a new ask until it is there, within RecordWithin: the seat answers a key's +// value and says nothing when the router takes an ask, so it is read again after a growing pause (§12). +func (g *Gate) awaitRecord(module, id string) (*seat.AsksByAskerIDValue, error) { + until := g.d.Now().Add(RecordWithin) + pause := 50 * time.Millisecond + for { + rec, err := g.record(module, id) + if rec != nil || g.d.Now().Add(pause).After(until) { + return rec, err + } + <-g.d.After(pause) + pause *= 2 + } +} + +// refused is a marked call refused, run nothing, written in the runner's log with the ask's fingerprint and never its +// arguments (§19). +func (g *Gate) refused(c Call, digest, why string) error { + g.d.Logf("[mesh-tools] %s on %s is refused (fingerprint %s): %s", c.Address, g.d.Node, orNone(Fingerprint(digest)), why) + return fmt.Errorf("Not run: %s needs the operator's approval and %s", c.Address, why) +} diff --git a/node-tools/internal/approval/gate.go b/node-tools/internal/approval/gate.go index 02f7d07..0aa5e32 100644 --- a/node-tools/internal/approval/gate.go +++ b/node-tools/internal/approval/gate.go @@ -9,6 +9,7 @@ import ( "time" condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/keys" ) @@ -38,6 +39,14 @@ type Deps struct { Findings FindingWriter Logf func(string, ...any) Now func() time.Time + // After is the clock's pause; nil is time.After. + After func(time.Duration) <-chan time.Time + // State is the runner's state approvals; nil when its membership binds none, so every marked call is refused (§13). + State Store + // Publish says an accept on a seat under the module's name, on the runner's own account (§7). + Publish func(module, subject string, body any, id string) error + // Record reads the router's record of an ask, `.` in the seat's state asks, on the runner's own account (§14). + Record func(module, id string) (*stdio.StateEntry, error) } // Gate is the runner's approval gate: every call of a module it serves goes through it before the bundle (§20). @@ -58,6 +67,9 @@ func New(d Deps) *Gate { if d.Now == nil { d.Now = time.Now } + if d.After == nil { + d.After = time.After + } return &Gate{d: d, registered: map[string]map[string]bool{}, unserved: map[string][]string{}} }