mesh-mail: send on the operator's Approve itself, sweep, stale-send status, no header injection (issue 790, review round 1)

modules/mesh-mail/cmd/mesh-mail/imap_test.go

@@ -231,7 +231,10 @@ func TestNothingButTheTestsWritesToAMailbox(t *testing.T) {
 			t.Fatal(err)
 		}
 		for _, m := range writes.FindAllSubmatch(raw, -1) {
-			if string(m[1]) != "os" { // a file of the module's own, never a mailbox
+			// os is a file of the module's own, never a mailbox; stdio is the bus SDK (main.go's hear subscribes to
+			// the router's own decided subject, exactly as mesh-knowledge does, novox/hq issue 790 review round 1)
+			// — its Subscribe has nothing to do with an IMAP mailbox's.
+			if string(m[1]) != "os" && string(m[1]) != "stdio" {
 				t.Errorf("%s calls %s, which writes to a mailbox", f, m[0])
 			}
 		}

modules/mesh-mail/cmd/mesh-mail/main.go

@@ -19,6 +19,7 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"os"
 	"os/signal"
@@ -27,6 +28,7 @@ import (
 	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/asks"
 )
 
 // Module is this module's name.
@@ -178,14 +180,51 @@ func serve(env Env) {
 			return NewS3(facts), nil
 		},
 	}
+	b.Logf = logf
 	p := &Publisher{Index: b.Index, Emit: func(key string, body any) error { return stdio.Emit(key, body) }, Logf: logf}
 	go p.Run(context.Background())
+	go hear(b)
+	go sweepSends(b)
 	if err := stdio.Serve("", b.Tools()); err != nil {
 		logf("%v", err)
 		os.Exit(1)
 	}
 }
 
+// hear subscribes to the router's word on this module's own asks, asked again until the runtime takes it, exactly
+// as mesh-knowledge does. Each warrant heard goes to Heard, the same Decide that mail_send_status uses: an
+// approved send goes out as soon as the operator answers it, never only when something later calls
+// mail_send_status (novox/hq ADR 0361 §2, review round 1 of issue 790).
+func hear(b *Bundle) {
+	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
+		time.Sleep(wait)
+		err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error {
+			var w asks.Warrant
+			if json.Unmarshal(e.Body, &w) == nil {
+				b.Heard(w)
+			}
+			return nil
+		})
+		if err == nil {
+			logf("[mesh-mail] hears the operator's answers")
+			return
+		}
+		logf("[mesh-mail] does not hear the operator's answers yet (%v); asking again in %s", err, wait)
+	}
+}
+
+// sweepSends reads every send still undecided once a minute and asks the router's own record for its answer, so
+// an Approve answered while this module could not hear it directly is still acted on (novox/hq issue 790, review
+// round 1).
+func sweepSends(b *Bundle) {
+	tick := time.NewTicker(SweepEvery)
+	defer tick.Stop()
+	for {
+		b.Sweep(time.Now())
+		<-tick.C
+	}
+}
+
 // Publisher says the events the keeper wrote, oldest first, and removes each once the bus took it. One the bus did not
 // take stays, and is said again on the next round: delivery is at least once, and a consumer reads the uid and the
 // folder to know a repeat.

modules/mesh-mail/cmd/mesh-mail/send.go

@@ -42,6 +42,16 @@ const (
 // SendAskLasts is how long the operator has to answer: a day less a margin, under the router's bound for an approval.
 const SendAskLasts = asks.ApproveLasts - 10*time.Minute
 
+// StaleSendAfter is how long a send may read "sending" before mail_send_status stops calling it "waiting": past
+// this, the claim before the SMTP call is old enough that the send may or may not have gone out, and is never
+// retried automatically (novox/hq issue 790, review round 1).
+const StaleSendAfter = 10 * time.Minute
+
+// SweepEvery is how often the sweep reads every send still undecided and asks the router's own record for its
+// answer, so an Approve answered while this module did not hear it directly (down, or not yet subscribed) is
+// still acted on (novox/hq issue 790, review round 1).
+const SweepEvery = time.Minute
+
 // The bounds of what mail_send takes.
 const (
 	MostRecipients = 20
@@ -149,6 +159,7 @@ type SendStore interface {
 	Get(key string) (json.RawMessage, uint64, bool, error)
 	Create(key string, value any) (uint64, error)
 	Update(key string, value any, revision uint64) (uint64, error)
+	Keys() ([]string, error)
 }
 
 // keptSends is the `sends` state on the bus.
@@ -165,6 +176,7 @@ func (k keptSends) Create(key string, value any) (uint64, error) { return k.s.Cr
 func (k keptSends) Update(key string, value any, revision uint64) (uint64, error) {
 	return k.s.Update(key, value, revision)
 }
+func (k keptSends) Keys() ([]string, error) { return k.s.Keys() }
 
 func getSend(store SendStore, id string) (SendRequest, uint64, bool, error) {
 	raw, rev, ok, err := store.Get(id)
@@ -207,6 +219,17 @@ func cleanAddresses(in []string) []string {
 	return out
 }
 
+// hasControlChar is true when s holds a byte a mail header must never carry: CR, LF, any other ASCII control
+// character below 0x20, or DEL (0x7f) — exactly what a header injection needs (novox/hq issue 790, review round 1).
+func hasControlChar(s string) bool {
+	for i := 0; i < len(s); i++ {
+		if s[i] < 0x20 || s[i] == 0x7f {
+			return true
+		}
+	}
+	return false
+}
+
 // AskMailSend raises an ask for one send and keeps it, under the module's own name. It never sends: the mail
 // reaches the server only once the operator's Approve is read back and held to this request (Decide). It answers
 // at once, with the ask's id; mail_send_status says where it stands.
@@ -220,6 +243,9 @@ func AskMailSend(in SendInput, store SendStore, bus AskBus, now time.Time) (map[
 		return refuse(fmt.Sprintf("to names %d recipients, more than %d", len(to), MostRecipients))
 	}
 	for _, addr := range to {
+		if hasControlChar(addr) {
+			return refuse(fmt.Sprintf("%q is not a mail address: it holds a control character", addr))
+		}
 		if _, err := mail.ParseAddress(addr); err != nil {
 			return refuse(fmt.Sprintf("%q is not a mail address", addr))
 		}
@@ -228,6 +254,8 @@ func AskMailSend(in SendInput, store SendStore, bus AskBus, now time.Time) (map[
 	switch {
 	case subject == "":
 		return refuse("subject is empty")
+	case hasControlChar(subject):
+		return refuse("the subject holds a control character")
 	case len([]rune(subject)) > MostSubject:
 		return refuse(fmt.Sprintf("the subject is %d characters, more than %d", len([]rune(subject)), MostSubject))
 	}
@@ -391,7 +419,7 @@ func ReadSendStatus(store SendStore, bus AskBus, lookup AccountLookup, send Send
 	out := map[string]any{"ask": req.ID, "account": req.Account, "to": req.To, "subject": req.Subject,
 		"asked": req.At.Format(time.RFC3339)}
 	if req.Decided != nil {
-		reportDecision(out, *req.Decided)
+		reportDecision(out, *req.Decided, now)
 		return out, nil
 	}
 	entry, err := bus.Record(req.ID)
@@ -416,19 +444,25 @@ func ReadSendStatus(store SendStore, bus AskBus, lookup AccountLookup, send Send
 		out["status"], out["words"] = "not proven", err.Error()
 		return out, nil
 	}
-	reportDecision(out, d)
+	reportDecision(out, d, now)
 	return out, nil
 }
 
 // reportDecision turns a kept decision into what mail_send_status answers: a status word a caller can branch on,
-// and the words that explain it.
-func reportDecision(out map[string]any, d SendDecision) {
+// and the words that explain it. now is when it is asked, not when the send was decided: a claim still "sending"
+// longer than StaleSendAfter with no final write is "not proven" (the send may or may not have gone out), never
+// "waiting" forever — and it is never retried automatically either way (novox/hq issue 790, review round 1).
+func reportDecision(out map[string]any, d SendDecision, now time.Time) {
 	out["at"] = d.At.Format(time.RFC3339)
 	switch {
 	case d.Outcome == "sent":
 		out["status"], out["words"] = "sent", "sent on the operator's Approve ("+d.Warrant+")"
 	case d.Outcome == "declined":
 		out["status"], out["words"] = "declined", "nothing was sent: the operator declined ("+d.Warrant+")"
+	case d.Outcome == "sending" && now.Sub(d.At) > StaleSendAfter:
+		out["status"] = "not proven"
+		out["words"] = "not proven: the send may or may not have gone out; the operator's Approve was being acted " +
+			"on at " + d.At.Format(time.RFC3339) + ", with no final word since"
 	case d.Outcome == "sending":
 		out["status"], out["words"] = "waiting", "the operator's Approve is being acted on; ask again shortly"
 	case strings.HasPrefix(d.Outcome, "failed: "):
@@ -441,3 +475,42 @@ func reportDecision(out map[string]any, d SendDecision) {
 		out["status"], out["words"] = "ended", d.Outcome
 	}
 }
+
+// ---- heard directly, and swept -------------------------------------------------------------------------------
+
+// Heard acts on one warrant the router's own subject delivered to this module directly (main.go's hear), through
+// the same Decide that mail_send_status uses: an approved send goes out as soon as the operator answers, never
+// only when something later happens to call mail_send_status (novox/hq ADR 0361 §2, review round 1 of issue 790).
+func (b *Bundle) Heard(w asks.Warrant) {
+	if _, err := Decide(b.sends(), b.lookupAccount, b.sendActor(), w, b.now()); err != nil {
+		if b.Logf != nil {
+			b.Logf("[mesh-mail] the warrant for the send %s could not be acted on: %v", w.Ask, err)
+		}
+	}
+}
+
+// Sweep reads every send this bundle still keeps undecided and asks the router's own record for its answer, the
+// same path mail_send_status takes (ReadSendStatus): an Approve recorded while this module did not hear it
+// directly (down, or not yet subscribed) is still acted on, once (novox/hq ADR 0361 §2, review round 1 of issue
+// 790).
+func (b *Bundle) Sweep(now time.Time) {
+	store := b.sends()
+	keys, err := store.Keys()
+	if err != nil {
+		if b.Logf != nil {
+			b.Logf("[mesh-mail] the sends kept cannot be listed: %v", err)
+		}
+		return
+	}
+	for _, id := range keys {
+		req, _, ok, err := getSend(store, id)
+		if err != nil || !ok || req.Decided != nil {
+			continue
+		}
+		if _, err := ReadSendStatus(store, b.askBus(), b.lookupAccount, b.sendActor(), id, now); err != nil {
+			if b.Logf != nil {
+				b.Logf("[mesh-mail] sweeping the send %s: %v", id, err)
+			}
+		}
+	}
+}

modules/mesh-mail/cmd/mesh-mail/send_test.go

@@ -67,6 +67,15 @@ func (m *memSends) put(key string, value any) (uint64, error) {
 	m.vals[key], m.revs[key] = raw, m.n
 	return m.n, nil
 }
+func (m *memSends) Keys() ([]string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	out := make([]string, 0, len(m.vals))
+	for k := range m.vals {
+		out = append(out, k)
+	}
+	return out, nil
+}
 
 // sendRouter is the operator channel's holder as this module reaches it, kept by a test: what was asked, and the
 // router's own record of each ask.
@@ -430,3 +439,149 @@ func TestMailSendRefusesAnAccountWithNoOutgoingServer(t *testing.T) {
 		t.Fatalf("an ask was raised for an account with no outgoing server: %d", n)
 	}
 }
+
+// (5) An approved send goes out once the warrant reaches the module through the hear handler alone — never
+// through mail_send_status (novox/hq ADR 0361 §2, review round 1 of issue 790).
+func TestAnApprovedSendGoesOutOnceWithNoStatusCall(t *testing.T) {
+	fake := newFakeSMTPServer(t)
+	b, router, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
+	b.Dial = fake.dialer()
+	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)
+
+	b.Heard(w) // the warrant delivered through the hear handler only
+	if n := fake.count(); n != 1 {
+		t.Fatalf("the fake server received %d message(s) after the warrant was heard once", n)
+	}
+	b.Heard(w) // the same warrant delivered a second time
+	if n := fake.count(); n != 1 {
+		t.Fatalf("the fake server received %d message(s) after the warrant was heard twice", n)
+	}
+}
+
+// (6) The sweep acts on a send the operator approved while this module was never told directly: it reads the
+// router's own record, the same path mail_send_status takes (novox/hq issue 790, review round 1).
+func TestSweepActsOnAnApprovedSendItWasNotToldAboutDirectly(t *testing.T) {
+	fake := newFakeSMTPServer(t)
+	b, router, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
+	b.Dial = fake.dialer()
+	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) // only the router's record holds the answer; mesh-mail is never told directly
+
+	b.Sweep(judgedAt.Add(2 * time.Minute))
+	if n := fake.count(); n != 1 {
+		t.Fatalf("the sweep did not act on the approved send: %d message(s)", n)
+	}
+	b.Sweep(judgedAt.Add(3 * time.Minute))
+	if n := fake.count(); n != 1 {
+		t.Fatalf("a second sweep sent again: %d message(s)", n)
+	}
+}
+
+// (7) A send claimed "sending" longer than StaleSendAfter with no final write reads as "not proven", never
+// "waiting" forever (novox/hq issue 790, review round 1).
+func TestAStaleSendingClaimIsNotProvenNotWaiting(t *testing.T) {
+	b, _, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
+	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)
+
+	req, rev, ok, err := getSend(sends, id)
+	if err != nil || !ok {
+		t.Fatalf("%v %v", err, ok)
+	}
+	claim := SendDecision{At: judgedAt.Add(-11 * time.Minute), Outcome: "sending", Warrant: "W1"}
+	req.Decided = &claim
+	if _, err := sends.Update(req.ID, req, rev); err != nil {
+		t.Fatalf("%v", err)
+	}
+
+	text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
+	if isErr {
+		t.Fatalf("%s", text)
+	}
+	if strings.Contains(text, `"status":"waiting"`) {
+		t.Fatalf("still reports waiting past the stale bound: %s", text)
+	}
+	if !strings.Contains(text, `"status":"not proven"`) {
+		t.Fatalf("does not report not proven once the claim is stale: %s", text)
+	}
+}
+
+// (8) The subject and every address are refused when they hold a control character: no header injection
+// (novox/hq issue 790, review round 1).
+func TestMailSendRefusesAControlCharacterInTheSubjectOrAnAddress(t *testing.T) {
+	b, _, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
+	rt := startBundle(t, b.Tools())
+
+	bySubject := map[string]any{"account": "mesh", "to": []any{"ops@example.org"},
+		"subject": "Report\r\nBcc: attacker@example.org", "body": "body"}
+	text, isErr := rt.call(SendTool, "novox.triage", bySubject)
+	if !isErr || !strings.Contains(text, "control character") {
+		t.Fatalf("a subject with a control character was not refused: %v %s", isErr, text)
+	}
+
+	byAddress := map[string]any{"account": "mesh", "to": []any{"ops@example.org\r\nBcc: attacker@example.org"},
+		"subject": "Report", "body": "body"}
+	text, isErr = rt.call(SendTool, "novox.triage", byAddress)
+	if !isErr || !strings.Contains(text, "control character") {
+		t.Fatalf("an address with a control character was not refused: %v %s", isErr, text)
+	}
+}
+
+// (9) Rcpt takes the parsed address alone; the To header takes the formatted address (novox/hq issue 790, review
+// round 1).
+func TestSendUsesTheParsedAddressForRcptAndTheFormattedAddressForTheToHeader(t *testing.T) {
+	fake := newFakeSMTPServer(t)
+	b, router, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
+	b.Dial = fake.dialer()
+	rt := startBundle(t, b.Tools())
+
+	args := map[string]any{"account": "mesh", "to": []any{"Ops Team <ops@example.org>"},
+		"subject": "Status", "body": "body"}
+	text, isErr := rt.call(SendTool, "novox.triage", args)
+	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)
+	b.Heard(w)
+
+	got := fake.last()
+	if len(got.to) != 1 || got.to[0] != "ops@example.org" {
+		t.Fatalf("RCPT TO got %v, want the parsed address alone", got.to)
+	}
+	if !strings.Contains(got.data, `To: "Ops Team" <ops@example.org>`) {
+		t.Fatalf("the To header is not the formatted address:\n%s", got.data)
+	}
+}

modules/mesh-mail/cmd/mesh-mail/smtp.go

@@ -10,6 +10,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"mime"
+	"net/mail"
 	"net/smtp"
 	"strings"
 	"time"
@@ -61,6 +62,18 @@ func SendMail(dial SMTPDialer, a Account, password string, to []string, subject,
 	if a.Outgoing == nil {
 		return fmt.Errorf("the account %s has no outgoing server", a.Name)
 	}
+	// Rcpt and the To header take the parsed address, never the raw string a caller gave: the address alone for
+	// Rcpt, the formatted address for the header (novox/hq issue 790, review round 1; no header injection).
+	rcpts := make([]string, 0, len(to))
+	headers := make([]string, 0, len(to))
+	for _, raw := range to {
+		addr, err := mail.ParseAddress(raw)
+		if err != nil {
+			return fmt.Errorf("the recipient %q is not a mail address: %w", raw, err)
+		}
+		rcpts = append(rcpts, addr.Address)
+		headers = append(headers, addr.String())
+	}
 	client, err := dial(*a.Outgoing)
 	if err != nil {
 		return err
@@ -74,7 +87,7 @@ func SendMail(dial SMTPDialer, a Account, password string, to []string, subject,
 	if err := client.Mail(from); err != nil {
 		return fmt.Errorf("the outgoing server refused the sender: %w", err)
 	}
-	for _, rcpt := range to {
+	for _, rcpt := range rcpts {
 		if err := client.Rcpt(rcpt); err != nil {
 			return fmt.Errorf("the outgoing server refused the recipient %s: %w", rcpt, err)
 		}
@@ -83,7 +96,7 @@ func SendMail(dial SMTPDialer, a Account, password string, to []string, subject,
 	if err != nil {
 		return fmt.Errorf("the outgoing server refused the message: %w", err)
 	}
-	if _, err := w.Write(composeMessage(from, to, subject, body, now)); err != nil {
+	if _, err := w.Write(composeMessage(from, headers, subject, body, now)); err != nil {
 		return fmt.Errorf("the message could not be written: %w", err)
 	}
 	if err := w.Close(); err != nil {
@@ -92,7 +105,8 @@ func SendMail(dial SMTPDialer, a Account, password string, to []string, subject,
 	return client.Quit()
 }
 
-// composeMessage is a plain-text message, its lines CRLF-terminated as SMTP takes them.
+// composeMessage is a plain-text message, its lines CRLF-terminated as SMTP takes them. to is the recipients
+// already formatted (mail.Address.String()), the same ones Rcpt was given, never the raw strings a caller gave.
 func composeMessage(from string, to []string, subject, body string, now time.Time) []byte {
 	var b strings.Builder
 	header := func(k, v string) { b.WriteString(k + ": " + v + "\r\n") }

modules/mesh-mail/cmd/mesh-mail/tools.go

@@ -48,6 +48,7 @@ type Bundle struct {
 	Asks  AskBus
 	Sends SendStore
 	Dial  SMTPDialer
+	Logf  func(format string, a ...any)
 }
 
 func (b *Bundle) now() time.Time {