From 1ccb617a86cbec4bafdae72671d706f4bde1bc82 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 09:57:07 +0200 Subject: [PATCH] mesh-mail: the sender verifies the operator's warrant again from the router's own record and claims a send once before any connection (ADR 0436 clauses 9, 10; issue 1196) --- modules/mesh-mail/cmd/mesh-mail/claim_test.go | 210 ++++++++++++++++++ modules/mesh-mail/cmd/mesh-mail/imap_test.go | 9 +- modules/mesh-mail/cmd/mesh-mail/send.go | 102 +++++++++ modules/mesh-mail/cmd/mesh-mail/sender.go | 38 ++++ 4 files changed, 356 insertions(+), 3 deletions(-) create mode 100644 modules/mesh-mail/cmd/mesh-mail/claim_test.go diff --git a/modules/mesh-mail/cmd/mesh-mail/claim_test.go b/modules/mesh-mail/cmd/mesh-mail/claim_test.go new file mode 100644 index 00000000..a6ef609f --- /dev/null +++ b/modules/mesh-mail/cmd/mesh-mail/claim_test.go @@ -0,0 +1,210 @@ +package main + +// The sender's own claim (novox/hq ADR 0436 clauses 9, 10, issue 1196): it never trusts the queue's message, +// verifying the operator's warrant again from the router's own record before it claims the send by +// compare-and-set — exactly mirroring the tools' own check, so the boundary clause 9 describes (the tool +// runner's account may publish on the queue and write `sends`, but only a warrant that really closed the +// ask moves a send to "sending"). + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// claimTestSend asks and approves one send (through the tools, as issue 1194 built it), leaving it queued, +// and returns its id and the router that holds its warrant. +func claimTestSend(t *testing.T) (id string, router *sendRouter, sends *memSends) { + t.Helper() + b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + queue := newMemQueue() + b.Queue = queue.publish + rt := startBundle(t, b.Tools()) + + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if isErr { + t.Fatalf("%s", text) + } + var out map[string]any + _ = json.Unmarshal([]byte(text), &out) + id = out["ask"].(string) + + w := router.chose(id, OptionSend, judgedAt.Add(time.Minute)) + router.close(w) + if _, err := Decide(sends, b.queuePublish(), w, judgedAt.Add(2*time.Minute)); err != nil { + t.Fatalf("approving: %v", err) + } + req, _, ok, err := getSend(sends, id) + if err != nil || !ok || req.Decided == nil || req.Decided.Outcome != "queued" { + t.Fatalf("not left queued: %v %v %+v", err, ok, req.Decided) + } + return id, router, sends +} + +// rawQueuedSend creates a send directly in the queued state, without going through the tools or the router +// at all, so an AC2 test controls the router's record independently of how the send reached queued: the +// defence in depth clause 9 describes, since the tool runner's account could write `sends` and the queue +// without ever having closed a real warrant. +func rawQueuedSend(t *testing.T, sends *memSends) SendRequest { + t.Helper() + req := SendRequest{ID: newSendID(), Account: "mesh", To: []string{"ops@example.org"}, Subject: "Status", + Body: "the body", Caller: "person.op", At: judgedAt, AskTo: "o… at example org"} + req.Decided = &SendDecision{At: judgedAt.Add(time.Minute), Outcome: "queued"} + if _, err := sends.Create(req.ID, req); err != nil { + t.Fatal(err) + } + return req +} + +// warrantFor is a well-formed warrant answering req's ask with Send, at, and the ask's own digest — the +// same shape sendRouter.chose builds, for a test that never published the ask to a router at all. +func warrantFor(req SendRequest, at time.Time) asks.Warrant { + ask := req.Ask() + opt, _ := ask.Option(OptionSend) + return asks.Warrant{Ask: req.ID, Asker: Module, Outcome: asks.OutcomeChosen, Option: opt.ID, Label: opt.Label, + Level: opt.Level, + By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}, + Channel: "telegram", Proofs: []string{"P1"}, At: at, AskDigest: ask.Digest()} +} + +// 1196/AC1: given a queued send whose router record is closed by the operator's Approve at level approve +// within its expiry and whose recomputed act digest is the option's binds, when the sender takes the +// message, then it claims the send by compare-and-set from queued to sending with the attempt and time +// before any connection (nothing here connects to anything: claimQueuedSend never dials). +func TestClaimQueuedSendClaimsSendingWithTheAttemptBeforeAnyConnection(t *testing.T) { + id, router, sends := claimTestSend(t) + + req, d, outcome, err := claimQueuedSend(sends, router.Record, id, judgedAt.Add(3*time.Minute)) + if err != nil { + t.Fatalf("claiming: %v", err) + } + if outcome != claimToSend { + t.Fatalf("outcome: %v, want claimToSend", outcome) + } + if d.Outcome != "sending" || d.Attempt != 1 || !d.At.Equal(judgedAt.Add(3*time.Minute)) { + t.Fatalf("claimed: %+v", d) + } + if req.ID != id { + t.Fatalf("req: %+v", req) + } + kept, _, ok, err := getSend(sends, id) + if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "sending" { + t.Fatalf("not kept as sending: %v %v %+v", err, ok, kept.Decided) + } +} + +// 1196/AC2: given a router record that is not mesh-mail's, not closed by Approve, approved after its +// expiry, or whose digest differs from the request in sends, when taken, then the sender records refused +// with the reason, terminates the message and connects to no server (claimQueuedSend never dials, so +// "connects to no server" holds for every path by construction; this proves each is refused and the claim +// never advances to sending). +func TestClaimQueuedSendRefusesAMismatchedOrUnclosedWarrant(t *testing.T) { + t.Run("not mesh-mail's ask", func(t *testing.T) { + sends, router := newMemSends(), newSendRouter() + req := rawQueuedSend(t, sends) + w := warrantFor(req, judgedAt.Add(time.Minute)) + w.Asker = "another-module" + router.close(w) + _, d, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(3*time.Minute)) + if err != nil || outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") { + t.Fatalf("outcome %v, decision %+v, err %v", outcome, d, err) + } + }) + t.Run("not closed by approve", func(t *testing.T) { + sends, router := newMemSends(), newSendRouter() + req := rawQueuedSend(t, sends) + // The router keeps no record of this ask at all: never published, or never answered. + _, d, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(3*time.Minute)) + if err != nil || outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") { + t.Fatalf("outcome %v, decision %+v, err %v", outcome, d, err) + } + }) + t.Run("approved after the ask's expiry", func(t *testing.T) { + sends, router := newMemSends(), newSendRouter() + req := rawQueuedSend(t, sends) + w := warrantFor(req, req.Ask().Expires.Add(time.Hour)) + router.close(w) + _, d, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(3*time.Minute)) + if err != nil || outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") { + t.Fatalf("outcome %v, decision %+v, err %v", outcome, d, err) + } + }) + t.Run("a digest the published ask never bound", func(t *testing.T) { + sends, router := newMemSends(), newSendRouter() + req := rawQueuedSend(t, sends) + w := warrantFor(req, judgedAt.Add(time.Minute)) + w.AskDigest = "sha256:" + strings.Repeat("0", 64) + router.close(w) + _, d, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(3*time.Minute)) + if err != nil || outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") { + t.Fatalf("outcome %v, decision %+v, err %v", outcome, d, err) + } + }) +} + +// 1196/AC3: given a send already sending (the sender stopped mid-send), when its message is delivered +// again, then the sender sends nothing, records not proven and terminates. +func TestClaimQueuedSendOnARedeliveryOfAnAlreadySendingClaimRecordsNotProven(t *testing.T) { + id, router, sends := claimTestSend(t) + req, _, outcome, err := claimQueuedSend(sends, router.Record, id, judgedAt.Add(3*time.Minute)) + if err != nil || outcome != claimToSend { + t.Fatalf("first claim: %v %v", outcome, err) + } + _ = req + + // Redelivered: the first claim's handler may have stopped mid-send. + _, d, outcome, err := claimQueuedSend(sends, router.Record, id, judgedAt.Add(4*time.Minute)) + if err != nil { + t.Fatalf("redelivered claim: %v", err) + } + if outcome != claimNotProven || d.Outcome != "not proven" { + t.Fatalf("outcome %v, decision %+v", outcome, d) + } +} + +// 1196/AC4: given a lost compare-and-set (another claim), when taken, then it sends nothing and leaves the +// message to the bus (no decision of this call's own making is kept; it reads back whatever the other +// writer decided). +func TestClaimQueuedSendOnALostRaceSendsNothingAndLeavesTheMessageToTheBus(t *testing.T) { + id, router, sends := claimTestSend(t) + racy := &raceOnFirstUpdate{SendStore: sends, race: func() { + r, rv, ok, err := getSend(sends, id) + if err != nil || !ok { + t.Fatalf("racing claim could not read the send: %v %v", err, ok) + } + if _, _, err := settleSend(sends, r, rv, SendDecision{At: judgedAt.Add(3 * time.Minute), Outcome: "sending", Attempt: 1}); err != nil { + t.Fatalf("racing claim: %v", err) + } + }} + + _, d, outcome, err := claimQueuedSend(racy, router.Record, id, judgedAt.Add(3*time.Minute)) + if err != nil { + t.Fatalf("claiming: %v", err) + } + if outcome != claimLostRace { + t.Fatalf("outcome: %v, want claimLostRace", outcome) + } + if d.Outcome != "sending" || d.Attempt != 1 { + t.Fatalf("the decision read back is not the racing claim's own: %+v", d) + } +} + +// raceOnFirstUpdate performs one competing write the first time Update is called — simulating another +// delivery's claim landing first — so the caller's own compare-and-set loses with stdio.ErrChanged +// (novox/hq issue 1196, AC4). +type raceOnFirstUpdate struct { + SendStore + raced bool + race func() +} + +func (r *raceOnFirstUpdate) Update(key string, value any, revision uint64) (uint64, error) { + if !r.raced { + r.raced = true + r.race() + } + return r.SendStore.Update(key, value, revision) +} diff --git a/modules/mesh-mail/cmd/mesh-mail/imap_test.go b/modules/mesh-mail/cmd/mesh-mail/imap_test.go index 47d9d53d..0159c710 100644 --- a/modules/mesh-mail/cmd/mesh-mail/imap_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/imap_test.go @@ -221,9 +221,12 @@ func TestNothingButTheTestsWritesToAMailbox(t *testing.T) { } writes := regexp.MustCompile(`([A-Za-z_][A-Za-z0-9_]*)\.(Store|Move|Copy|Expunge|UIDExpunge|Append|Create|Delete|Rename|Subscribe|Unsubscribe|SetMetadata|SetQuota)\(`) for _, f := range files { - if strings.HasSuffix(f, "_test.go") || f == "send.go" || f == "smtp.go" { - // Neither touches a mailbox: send.go keeps the `sends` state on the bus, never IMAP; smtp.go speaks SMTP - // to the account's outgoing server, never to a mailbox, and only on the operator's Approve (ADR 0361 §2). + if strings.HasSuffix(f, "_test.go") || f == "send.go" || f == "smtp.go" || f == "sender.go" { + // None touches a mailbox: send.go keeps the `sends` state on the bus, never IMAP; smtp.go speaks SMTP + // to the account's outgoing server, never to a mailbox, and only on the operator's Approve (ADR 0361 + // §2); sender.go is the sender process (novox/hq ADR 0436 §7) — its own bus account's `sends` state + // (process.State.Create, matched here only because its receiver is named s) and router record, never + // IMAP either. continue } raw, err := os.ReadFile(f) diff --git a/modules/mesh-mail/cmd/mesh-mail/send.go b/modules/mesh-mail/cmd/mesh-mail/send.go index 360213a8..7781f146 100644 --- a/modules/mesh-mail/cmd/mesh-mail/send.go +++ b/modules/mesh-mail/cmd/mesh-mail/send.go @@ -186,6 +186,9 @@ type SendDecision struct { Outcome string `json:"outcome"` Warrant string `json:"warrant,omitempty"` Error string `json:"error,omitempty"` + // Attempt is which handover to the server this is, 1 for the first: the sender's own claim from queued + // or retrying to sending counts it (novox/hq ADR 0436 clauses 9, 10, issue 1196). + Attempt int `json:"attempt,omitempty"` } // bodyDigest is the digest an act binds in place of the body itself: never the body, which the act's digest is @@ -694,6 +697,105 @@ func publishQueued(store SendStore, req SendRequest, rev uint64, approved SendDe return final, nil } +// ---- the sender's own claim (novox/hq ADR 0436 clauses 9, 10, issue 1196) -------------------------------------- +// +// The sender never trusts the queue's message: it reads the router's record of mesh-mail's own ask under its +// own account, and mirrors the tools' own check (Decide, above) exactly — w.For, then option.Performs on the +// request's own Act — before it claims the send by compare-and-set and, only then, connects to anything. + +// RouterRecord is how the sender reads the router's record of one ask: a function so a test gives a fake, and +// the real one (sender.go) wraps a process.Conn's Ask(bucket, id) over the operator channel's records. +type RouterRecord func(id string) (*stdio.StateEntry, error) + +// claimOutcome is what became of one claim, so the caller (the message handler) knows the message's fate. +type claimOutcome int + +const ( + // claimToSend: claimed "sending" by this call; the caller may now connect and send. + claimToSend claimOutcome = iota + // claimRefused: recorded refused for a reason this function names; the caller terminates the message. + claimRefused + // claimNotProven: the send was already "sending" (a previous delivery may have handed it to the server); + // recorded not proven; the caller terminates the message, never retrying it automatically. + claimNotProven + // claimLostRace: the compare-and-set to "sending" lost to another writer (another delivery of the same + // message, racing this one); nothing is recorded here and nothing sent; the caller leaves the message to + // the bus entirely — no Ack, no Nak, no Term. + claimLostRace +) + +// claimQueuedSend verifies the operator's warrant for id (mesh-mail's own ask) and claims the send by +// compare-and-set from queued or retrying to sending, with the attempt and time, before any connection +// (novox/hq ADR 0436 clauses 9, 10). It is never trusted from the queue's message alone: the ask is read +// fresh from the router under the sender's own account, closed by the operator's Approve at level approve +// within its expiry (Warrant.For), and its act's digest (the account, recipients, subject and body's digest) +// is the option's binds (Option.Performs) — exactly the tools' own check (Decide, above) mirrored here, +// because the tool runner's account, which every agent session on this machine also reaches the bus as, may +// publish on the queue and write `sends` (this is the boundary clause 9 describes). +func claimQueuedSend(store SendStore, record RouterRecord, id string, now time.Time) (SendRequest, SendDecision, claimOutcome, error) { + req, rev, ok, err := getSend(store, id) + if err != nil { + return SendRequest{}, SendDecision{}, claimLostRace, err + } + if !ok { + return SendRequest{}, SendDecision{}, claimLostRace, fmt.Errorf("no send %q is kept", id) + } + refuse := func(reason string) (SendRequest, SendDecision, claimOutcome, error) { + d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: " + reason}) + return req, d, claimRefused, err + } + switch { + case req.Decided == nil: + return refuse("the send is not yet approved") + case req.Decided.Outcome == "sending": + // A previous delivery may have handed it to the server already; never sent twice, and never + // retried automatically either way (ADR 0436 clause 10). + d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "not proven", Attempt: req.Decided.Attempt}) + return req, d, claimNotProven, err + case req.Decided.Outcome != "queued" && req.Decided.Outcome != "retrying": + return refuse(fmt.Sprintf("the send is %q, neither queued nor retrying", req.Decided.Outcome)) + } + entry, err := record(id) + if err != nil { + return SendRequest{}, SendDecision{}, claimLostRace, err + } + if entry == nil { + return refuse("the router keeps no record of this ask") + } + var rec routerRecord + if json.Unmarshal(entry.Value, &rec) != nil { + return refuse("the router's record of it cannot be read") + } + if rec.State == "" || rec.State == "open" || rec.Warrant == nil { + return refuse("the ask is not closed by the operator's Approve") + } + w := *rec.Warrant + if w.Asker != Module { + return refuse(fmt.Sprintf("the warrant is for %s's ask, not %s's", w.Asker, Module)) + } + ask := req.Ask() + option, err := w.For(Module, ask) + if err == nil { + err = option.Performs(req.Act(option.ID)) + } + if err != nil { + return refuse(err.Error()) + } + if option.ID != OptionSend { + return refuse("the option " + option.ID + " does not authorise a send") + } + claim := SendDecision{At: now, Outcome: "sending", Warrant: w.Says(), Attempt: req.Decided.Attempt + 1} + claimed, newRev, err := updateDecision(store, req, rev, claim) + if err != nil { + return SendRequest{}, SendDecision{}, claimLostRace, err + } + if newRev == 0 { + // Lost the claim race: another delivery of the same message is already handling it. + return req, claimed, claimLostRace, nil + } + return req, claimed, claimToSend, nil +} + // ---- reading the status ----------------------------------------------------------------------------------------- // ReadSendStatus is where a send asked with mail_send stands, by its ask id: waiting for the operator, sent, diff --git a/modules/mesh-mail/cmd/mesh-mail/sender.go b/modules/mesh-mail/cmd/mesh-mail/sender.go index 88ff5880..e60af68f 100644 --- a/modules/mesh-mail/cmd/mesh-mail/sender.go +++ b/modules/mesh-mail/cmd/mesh-mail/sender.go @@ -9,11 +9,13 @@ package main import ( "context" + "encoding/json" "fmt" "os" "os/signal" "syscall" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/process" "mesh-mail/cmd/mesh-mail/internal/secret" @@ -26,6 +28,42 @@ const SendProcessID = "send" // MailSendsStream is the work-queue stream mail-sends carries on the bus (novox/hq ADR 0436 §8). const MailSendsStream = "MAIL-SENDS" +// MessengerAsksBucket is the operator channel's own records bucket, as the controller's own fixtures name +// it (mesh-controller cmd/mesh-controller/process_live_test.go, internal/broker/process_grants_test.go): +// never hard-coded in mesh-sdk go/process itself, since a process may read more than one seat's records. +const MessengerAsksBucket = "messenger_asks" + +// routerRecordOf is the real RouterRecord (send.go), reading the router's record of one of mesh-mail's own +// asks under conn's own account (novox/hq ADR 0436 clause 9) — the same direct-get the tools use +// (stdio.SeatRecord), reached here as the sender's own account instead of the tool runner's. Wired into the +// handler once the claim, the send and its outcome are all in place (issues 1197, 1198). +func routerRecordOf(conn *process.Conn) RouterRecord { + return func(id string) (*stdio.StateEntry, error) { return conn.Ask(MessengerAsksBucket, id) } +} + +// processSends adapts conn's own `sends` state (novox/hq ADR 0436 clause 7's bus.state) to SendStore +// (send.go), the same CAS the tools use, now reached under the sender's own account. Keys is never called +// by the claim or outcome recording (issues 1196, 1198) — only the tools' sweep calls it, over its own +// account — so it is refused by name if ever called by mistake, rather than silently answering empty. +type processSends struct{ state process.State } + +func (p processSends) Get(key string) (json.RawMessage, uint64, bool, error) { + e, err := p.state.Get(key) + if err != nil || e == nil { + return nil, 0, false, err + } + return e.Value, e.Revision, true, nil +} +func (p processSends) Create(key string, value any) (uint64, error) { + return p.state.Create(key, value) +} +func (p processSends) Update(key string, value any, revision uint64) (uint64, error) { + return p.state.Update(key, value, revision) +} +func (p processSends) Keys() ([]string, error) { + return nil, fmt.Errorf("the sender's own `sends` state is never listed whole; only the tools' sweep does that") +} + // send is `mesh-mail send`: it reads its own bus account and the accounts setting, connects, and consumes // mail-sends until stopped. It refuses to start naming what is missing when the account or the accounts // setting cannot be read — never a password, which no error here or anywhere in this process logs