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)
This commit is contained in:
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user