diff --git a/modules/mesh-mail/cmd/mesh-mail/imap_test.go b/modules/mesh-mail/cmd/mesh-mail/imap_test.go index a17052a3..47d9d53d 100644 --- a/modules/mesh-mail/cmd/mesh-mail/imap_test.go +++ b/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]) } } diff --git a/modules/mesh-mail/cmd/mesh-mail/main.go b/modules/mesh-mail/cmd/mesh-mail/main.go index 373d0020..02d4fffa 100644 --- a/modules/mesh-mail/cmd/mesh-mail/main.go +++ b/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. diff --git a/modules/mesh-mail/cmd/mesh-mail/send.go b/modules/mesh-mail/cmd/mesh-mail/send.go index 213691e0..8135ade8 100644 --- a/modules/mesh-mail/cmd/mesh-mail/send.go +++ b/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) + } + } + } +} diff --git a/modules/mesh-mail/cmd/mesh-mail/send_test.go b/modules/mesh-mail/cmd/mesh-mail/send_test.go index 1793a5f2..3e0b0fe0 100644 --- a/modules/mesh-mail/cmd/mesh-mail/send_test.go +++ b/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 "}, + "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" `) { + t.Fatalf("the To header is not the formatted address:\n%s", got.data) + } +} diff --git a/modules/mesh-mail/cmd/mesh-mail/smtp.go b/modules/mesh-mail/cmd/mesh-mail/smtp.go index 7d797302..bc12d406 100644 --- a/modules/mesh-mail/cmd/mesh-mail/smtp.go +++ b/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") } diff --git a/modules/mesh-mail/cmd/mesh-mail/tools.go b/modules/mesh-mail/cmd/mesh-mail/tools.go index 9d8345fd..3f99c806 100644 --- a/modules/mesh-mail/cmd/mesh-mail/tools.go +++ b/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 {