1024 lines
36 KiB
Go
1024 lines
36 KiB
Go
package main
|
|
|
|
// mail_send (novox/hq ADR 0361 §2): an ungranted caller is refused before anything is asked; a granted caller's ask
|
|
// holds the account, the recipients, the subject and the first lines; the operator's Approve sends through an
|
|
// in-process fake SMTP server exactly once, even delivered twice; a Decline sends nothing.
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net"
|
|
"net/smtp"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
)
|
|
|
|
// ---- a fake `sends` state and a fake operator channel, as a test keeps them --------------------------------------
|
|
|
|
type memSends struct {
|
|
mu sync.Mutex
|
|
vals map[string]json.RawMessage
|
|
revs map[string]uint64
|
|
n uint64
|
|
|
|
// failKey/failErr: when failKey matches, Create and Update report failErr instead of writing, for a test to
|
|
// simulate a store whose index write fails (novox/hq issue 811, review round 3).
|
|
failKey string
|
|
failErr error
|
|
}
|
|
|
|
func newMemSends() *memSends {
|
|
return &memSends{vals: map[string]json.RawMessage{}, revs: map[string]uint64{}}
|
|
}
|
|
|
|
func (m *memSends) Get(key string) (json.RawMessage, uint64, bool, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
v, ok := m.vals[key]
|
|
if !ok {
|
|
return nil, 0, false, nil
|
|
}
|
|
return v, m.revs[key], true, nil
|
|
}
|
|
func (m *memSends) Create(key string, value any) (uint64, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
if m.failErr != nil && key == m.failKey {
|
|
return 0, m.failErr
|
|
}
|
|
if _, ok := m.vals[key]; ok {
|
|
return 0, stdio.ErrChanged
|
|
}
|
|
return m.put(key, value)
|
|
}
|
|
func (m *memSends) Update(key string, value any, revision uint64) (uint64, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
if m.failErr != nil && key == m.failKey {
|
|
return 0, m.failErr
|
|
}
|
|
if revision == 0 || m.revs[key] != revision {
|
|
return 0, stdio.ErrChanged
|
|
}
|
|
return m.put(key, value)
|
|
}
|
|
func (m *memSends) put(key string, value any) (uint64, error) {
|
|
raw, err := json.Marshal(value)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
m.n++
|
|
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.
|
|
type sendRouter struct {
|
|
mu sync.Mutex
|
|
published map[string]asks.Ask
|
|
records map[string]json.RawMessage
|
|
publishErr error
|
|
recordCalls map[string]int
|
|
}
|
|
|
|
func newSendRouter() *sendRouter {
|
|
return &sendRouter{published: map[string]asks.Ask{}, records: map[string]json.RawMessage{}, recordCalls: map[string]int{}}
|
|
}
|
|
|
|
func (r *sendRouter) Publish(subject string, body any, id string) error {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if r.publishErr != nil {
|
|
return r.publishErr
|
|
}
|
|
raw, _ := json.Marshal(body)
|
|
var a asks.Ask
|
|
_ = json.Unmarshal(raw, &a)
|
|
r.published[a.ID] = a
|
|
r.records[a.ID], _ = json.Marshal(map[string]any{"state": "open"})
|
|
return nil
|
|
}
|
|
|
|
// recordCallsFor is how many times Record was asked about id since the last resetRecordCalls, for a test to check
|
|
// the sweep read only the sends still undecided (novox/hq issue 811).
|
|
func (r *sendRouter) recordCallsFor(id string) int {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return r.recordCalls[id]
|
|
}
|
|
|
|
func (r *sendRouter) resetRecordCalls() {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.recordCalls = map[string]int{}
|
|
}
|
|
|
|
func (r *sendRouter) Record(id string) (*stdio.StateEntry, error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.recordCalls[id]++
|
|
raw, ok := r.records[id]
|
|
if !ok {
|
|
return nil, nil
|
|
}
|
|
return &stdio.StateEntry{Key: id, Value: raw}, nil
|
|
}
|
|
|
|
// close ends an ask with a warrant, as the router records it.
|
|
func (r *sendRouter) close(w asks.Warrant) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.records[w.Ask], _ = json.Marshal(map[string]any{"state": string(w.Outcome), "warrant": w})
|
|
}
|
|
|
|
// chose is the warrant for a person's choice of an option of the ask as the router took it.
|
|
func (r *sendRouter) chose(id, option string, at time.Time) asks.Warrant {
|
|
r.mu.Lock()
|
|
a := r.published[id]
|
|
r.mu.Unlock()
|
|
o, _ := a.Option(option)
|
|
return asks.Warrant{Ask: id, Asker: Module, Outcome: asks.OutcomeChosen, Option: o.ID, Label: o.Label, Level: o.Level,
|
|
By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"},
|
|
Channel: "telegram", Proofs: []string{"P1"}, At: at, AskDigest: a.Digest()}
|
|
}
|
|
|
|
func (r *sendRouter) published1() int {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return len(r.published)
|
|
}
|
|
|
|
// ---- a fake work-queue publish, as the tools reach mail-sends (novox/hq ADR 0436 clause 8) ----------------------
|
|
|
|
type queuedMsg struct {
|
|
id string
|
|
body QueuedSend
|
|
}
|
|
|
|
type memQueue struct {
|
|
mu sync.Mutex
|
|
published []queuedMsg
|
|
failErr error
|
|
}
|
|
|
|
func newMemQueue() *memQueue { return &memQueue{} }
|
|
|
|
func (q *memQueue) publish(id string, body any) error {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
if q.failErr != nil {
|
|
return q.failErr
|
|
}
|
|
b, _ := body.(QueuedSend)
|
|
q.published = append(q.published, queuedMsg{id: id, body: b})
|
|
return nil
|
|
}
|
|
|
|
func (q *memQueue) count() int {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
return len(q.published)
|
|
}
|
|
|
|
func (q *memQueue) last() queuedMsg {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
if len(q.published) == 0 {
|
|
return queuedMsg{}
|
|
}
|
|
return q.published[len(q.published)-1]
|
|
}
|
|
|
|
// ---- an in-process fake SMTP server, speaking the protocol over a real socket -------------------------------------
|
|
|
|
type receivedMail struct {
|
|
from string
|
|
to []string
|
|
data string
|
|
}
|
|
|
|
type fakeSMTPServer struct {
|
|
ln net.Listener
|
|
|
|
mu sync.Mutex
|
|
received []receivedMail
|
|
refuse bool // when true, DATA is refused: a transient server error
|
|
}
|
|
|
|
func newFakeSMTPServer(t *testing.T) *fakeSMTPServer {
|
|
t.Helper()
|
|
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
f := &fakeSMTPServer{ln: ln}
|
|
go f.serve()
|
|
t.Cleanup(func() { _ = ln.Close() })
|
|
return f
|
|
}
|
|
|
|
func (f *fakeSMTPServer) addr() string { return f.ln.Addr().String() }
|
|
|
|
// dialer reaches the fake server directly, plain: only a test skips the TLS the mesh's own dialer always uses.
|
|
func (f *fakeSMTPServer) dialer() SMTPDialer {
|
|
addr := f.addr()
|
|
return func(Server) (*smtp.Client, error) { return smtp.Dial(addr) }
|
|
}
|
|
|
|
func (f *fakeSMTPServer) count() int {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return len(f.received)
|
|
}
|
|
|
|
func (f *fakeSMTPServer) last() receivedMail {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if len(f.received) == 0 {
|
|
return receivedMail{}
|
|
}
|
|
return f.received[len(f.received)-1]
|
|
}
|
|
|
|
func (f *fakeSMTPServer) serve() {
|
|
for {
|
|
conn, err := f.ln.Accept()
|
|
if err != nil {
|
|
return
|
|
}
|
|
go f.handle(conn)
|
|
}
|
|
}
|
|
|
|
func (f *fakeSMTPServer) handle(conn net.Conn) {
|
|
defer conn.Close()
|
|
r := bufio.NewReader(conn)
|
|
reply := func(code int, msg string) { fmt.Fprintf(conn, "%d %s\r\n", code, msg) }
|
|
reply(220, "fake.local ESMTP")
|
|
var from string
|
|
var to []string
|
|
for {
|
|
line, err := r.ReadString('\n')
|
|
if err != nil {
|
|
return
|
|
}
|
|
line = strings.TrimRight(line, "\r\n")
|
|
upper := strings.ToUpper(line)
|
|
switch {
|
|
case strings.HasPrefix(upper, "EHLO"), strings.HasPrefix(upper, "HELO"):
|
|
reply(250, "fake.local")
|
|
case strings.HasPrefix(upper, "AUTH"):
|
|
reply(235, "authenticated")
|
|
case strings.HasPrefix(upper, "MAIL FROM:"):
|
|
from = addrIn(line)
|
|
reply(250, "OK")
|
|
case strings.HasPrefix(upper, "RCPT TO:"):
|
|
to = append(to, addrIn(line))
|
|
reply(250, "OK")
|
|
case upper == "DATA":
|
|
f.mu.Lock()
|
|
refuse := f.refuse
|
|
f.mu.Unlock()
|
|
if refuse {
|
|
reply(451, "temporary failure")
|
|
continue
|
|
}
|
|
reply(354, "End data with <CR><LF>.<CR><LF>")
|
|
var b strings.Builder
|
|
for {
|
|
dl, err := r.ReadString('\n')
|
|
if err != nil {
|
|
return
|
|
}
|
|
if dl == ".\r\n" || dl == ".\n" {
|
|
break
|
|
}
|
|
if strings.HasPrefix(dl, "..") {
|
|
dl = dl[1:]
|
|
}
|
|
b.WriteString(dl)
|
|
}
|
|
f.mu.Lock()
|
|
f.received = append(f.received, receivedMail{from: from, to: append([]string{}, to...), data: b.String()})
|
|
f.mu.Unlock()
|
|
reply(250, "OK: queued")
|
|
from, to = "", nil
|
|
case upper == "RSET":
|
|
reply(250, "OK")
|
|
from, to = "", nil
|
|
case upper == "QUIT":
|
|
reply(221, "bye")
|
|
return
|
|
default:
|
|
reply(500, "unrecognized command")
|
|
}
|
|
}
|
|
}
|
|
|
|
func addrIn(line string) string {
|
|
start, end := strings.Index(line, "<"), strings.Index(line, ">")
|
|
if start >= 0 && end > start {
|
|
return line[start+1 : end]
|
|
}
|
|
return line
|
|
}
|
|
|
|
// ---- a bundle over an account with an outgoing server ---------------------------------------------------------
|
|
|
|
// sendTestBundle is a bundle over one account, "mesh", granted to "triage" alone, with an outgoing server of the
|
|
// caller's choosing (a fake SMTP server's, in most tests) and its password already given.
|
|
func sendTestBundle(t *testing.T, outgoing Server) (*Bundle, *sendRouter, *memSends) {
|
|
t.Helper()
|
|
dir := t.TempDir()
|
|
env := Env{Accounts: filepath.Join(dir, "accounts.json"), Grants: filepath.Join(dir, "grants.json"), Secrets: dir}
|
|
writeFile(t, env.Accounts, fmt.Sprintf(`{"accounts": {"mesh": {"kind": "imap", "address": "jschoubben@novox.be",
|
|
"incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"},
|
|
"outgoing": {"host": %q, "port": %d, "security": %q}}}}`, outgoing.Host, outgoing.Port, outgoing.Security))
|
|
writeFile(t, env.Grants, `{"grants": {"mesh": ["triage"]}}`)
|
|
giveSecret(t, dir, Account{Name: "mesh"}, "the-password")
|
|
router, sends := newSendRouter(), newMemSends()
|
|
b := &Bundle{Env: env, Now: func() time.Time { return judgedAt }, Asks: router, Sends: sends}
|
|
return b, router, sends
|
|
}
|
|
|
|
var sendArgs = map[string]any{"account": "mesh", "to": []any{"ops@example.org", "lead@example.org"},
|
|
"subject": "The nightly gate is red", "body": "The merge gate failed on the resolver change.\nSee the job log."}
|
|
|
|
// (1) An ungranted caller is refused, and no ask is raised.
|
|
func TestMailSendRefusesAnUngrantedCallerAndRaisesNoAsk(t *testing.T) {
|
|
b, router, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
|
|
rt := startBundle(t, b.Tools())
|
|
text, isErr := rt.call(SendTool, "novox.other", sendArgs)
|
|
if !isErr || !strings.Contains(text, "refused") || !strings.Contains(text, "mesh") {
|
|
t.Fatalf("%v %s", isErr, text)
|
|
}
|
|
if n := router.published1(); n != 0 {
|
|
t.Fatalf("an ask was raised for an ungranted caller: %d", n)
|
|
}
|
|
}
|
|
|
|
// (2) A granted caller raises one ask whose text holds the account, the recipients, the subject and the first
|
|
// lines — the recipients without their addresses, in the form the operator channel's content rule (ADR 0253 §3)
|
|
// lets through (novox/hq issue 868; ask_test.go judges the real rule on asks like this one).
|
|
func TestMailSendAsksWithTheAccountRecipientsSubjectAndFirstLines(t *testing.T) {
|
|
b, router, _ := 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
|
|
if json.Unmarshal([]byte(text), &out) != nil {
|
|
t.Fatalf("not JSON: %s", text)
|
|
}
|
|
id, _ := out["ask"].(string)
|
|
if id == "" || out["status"] != "waiting" {
|
|
t.Fatalf("%s", text)
|
|
}
|
|
if n := router.published1(); n != 1 {
|
|
t.Fatalf("published %d asks, not 1", n)
|
|
}
|
|
a := router.published[id]
|
|
if err := a.Check(judgedAt); err != nil {
|
|
t.Errorf("the ask does not pass the asks contract: %v", err)
|
|
}
|
|
whole := a.Headline + "\n" + a.Explanation + "\n" + a.Details
|
|
for _, want := range []string{"mesh", "o… at example org", "l… at example org", "The nightly gate is red", "The merge gate failed on the resolver change"} {
|
|
if !strings.Contains(whole, want) {
|
|
t.Errorf("the ask does not say %q:\n%s", want, whole)
|
|
}
|
|
}
|
|
if strings.Contains(whole, "ops@example.org") || strings.Contains(whole, "lead@example.org") {
|
|
t.Errorf("the ask carries a recipient's mail address, which the operator channel's content rule refuses:\n%s", whole)
|
|
}
|
|
for _, o := range a.Options {
|
|
if o.Level != asks.Approve || o.Binds == "" {
|
|
t.Errorf("the option %s is %s binding %q: both answers are an approval bound to this send", o.ID, o.Level, o.Binds)
|
|
}
|
|
}
|
|
}
|
|
|
|
// 1194/AC1: On Approve, the send is queued for the sender exactly once, even if the warrant is delivered twice
|
|
// (novox/hq ADR 0436 clause 8). Since issue 1194 the tools never dial an outgoing server: they record approved
|
|
// and queue the send for the sender process, which alone performs it (issue 1196, 1198).
|
|
func TestOnApproveTheSendIsQueuedExactlyOnceEvenDeliveredTwice(t *testing.T) {
|
|
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)
|
|
|
|
publish := b.queuePublish()
|
|
if _, err := Decide(sends, publish, w, judgedAt.Add(2*time.Minute)); err != nil {
|
|
t.Fatalf("deciding: %v", err)
|
|
}
|
|
if n := queue.count(); n != 1 {
|
|
t.Fatalf("the queue received %d message(s) after one Approve", n)
|
|
}
|
|
// The warrant delivered a second time: still queued exactly once.
|
|
if _, err := Decide(sends, publish, w, judgedAt.Add(3*time.Minute)); err != nil {
|
|
t.Fatalf("deciding again: %v", err)
|
|
}
|
|
if n := queue.count(); n != 1 {
|
|
t.Fatalf("the queue received %d message(s) after the warrant was delivered twice", n)
|
|
}
|
|
got := queue.last()
|
|
if got.id != id || got.body.Send != id || got.body.Account != "mesh" || got.body.Ask != id {
|
|
t.Fatalf("queued %+v, want the send id %s as the message id, account and ask", got, id)
|
|
}
|
|
|
|
// mail_send_status reports it waiting on the sender, and reports the same thing again without queueing a
|
|
// third time.
|
|
text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
|
|
if isErr || !strings.Contains(text, `"status":"waiting"`) {
|
|
t.Fatalf("%v %s", isErr, text)
|
|
}
|
|
if n := queue.count(); n != 1 {
|
|
t.Fatalf("mail_send_status queued again: %d", n)
|
|
}
|
|
}
|
|
|
|
// 1194/AC3: On Decline, nothing is queued, and the status says declined.
|
|
func TestOnDeclineNothingIsQueuedAndTheStatusSaysDeclined(t *testing.T) {
|
|
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, OptionDecline, judgedAt.Add(time.Minute))
|
|
router.close(w)
|
|
if _, err := Decide(sends, b.queuePublish(), w, judgedAt.Add(2*time.Minute)); err != nil {
|
|
t.Fatalf("deciding: %v", err)
|
|
}
|
|
if n := queue.count(); n != 0 {
|
|
t.Fatalf("the queue received %d message(s) on Decline", n)
|
|
}
|
|
|
|
text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
|
|
if isErr || !strings.Contains(text, `"status":"declined"`) {
|
|
t.Fatalf("%v %s", isErr, text)
|
|
}
|
|
}
|
|
|
|
// An account with no outgoing server is refused before anything is asked.
|
|
func TestMailSendRefusesAnAccountWithNoOutgoingServer(t *testing.T) {
|
|
dir := t.TempDir()
|
|
env := Env{Accounts: filepath.Join(dir, "accounts.json"), Grants: filepath.Join(dir, "grants.json"), Secrets: dir}
|
|
writeFile(t, env.Accounts, stageOne) // no outgoing server
|
|
writeFile(t, env.Grants, `{"grants": {"mesh": ["triage"]}}`)
|
|
router := newSendRouter()
|
|
b := &Bundle{Env: env, Now: func() time.Time { return judgedAt }, Asks: router, Sends: newMemSends()}
|
|
rt := startBundle(t, b.Tools())
|
|
text, isErr := rt.call(SendTool, "novox.triage", sendArgs)
|
|
if !isErr || !strings.Contains(text, "no outgoing server") {
|
|
t.Fatalf("%v %s", isErr, text)
|
|
}
|
|
if n := router.published1(); n != 0 {
|
|
t.Fatalf("an ask was raised for an account with no outgoing server: %d", n)
|
|
}
|
|
}
|
|
|
|
// 1194/AC1: An approved send is queued 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 TestAnApprovedSendIsQueuedOnceWithNoStatusCall(t *testing.T) {
|
|
b, router, _ := 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)
|
|
|
|
b.Heard(w) // the warrant delivered through the hear handler only
|
|
if n := queue.count(); n != 1 {
|
|
t.Fatalf("the queue received %d message(s) after the warrant was heard once", n)
|
|
}
|
|
b.Heard(w) // the same warrant delivered a second time
|
|
if n := queue.count(); n != 1 {
|
|
t.Fatalf("the queue received %d message(s) after the warrant was heard twice", n)
|
|
}
|
|
}
|
|
|
|
// 1194/AC1: 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) {
|
|
b, router, _ := 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) // only the router's record holds the answer; mesh-mail is never told directly
|
|
|
|
b.Sweep(judgedAt.Add(2 * time.Minute))
|
|
if n := queue.count(); n != 1 {
|
|
t.Fatalf("the sweep did not act on the approved send: %d message(s)", n)
|
|
}
|
|
// A second sweep, still within the just-queued record's own grace, does not publish it again (past the
|
|
// grace, it would: TestSweepPublishesAgainAQueuedRecordPastItsGrace, novox/hq review round 1).
|
|
b.Sweep(judgedAt.Add(2*time.Minute + QueuedGrace/2))
|
|
if n := queue.count(); n != 1 {
|
|
t.Fatalf("a second sweep within the grace queued again: %d message(s)", n)
|
|
}
|
|
}
|
|
|
|
// 1194/AC2 (as decided to replace the two-state approved/queued design, review round 1): the sweep
|
|
// republishes a queued, unclaimed record whose queued time is older than QueuedGrace, the send id still
|
|
// its message id — the stream's duplicate window and the sender's own claim keep it one send either way
|
|
// (proved end to end over a real bus by the module-whole acceptance test, issue 1198 AC4).
|
|
func TestSweepPublishesAgainAQueuedRecordPastItsGrace(t *testing.T) {
|
|
b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
|
|
queue := newMemQueue()
|
|
queue.failErr = errors.New("the bus is down")
|
|
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)
|
|
decidedAt := judgedAt.Add(2 * time.Minute)
|
|
if _, err := Decide(sends, b.queuePublish(), w, decidedAt); err != nil {
|
|
t.Fatalf("deciding: %v", err)
|
|
}
|
|
if n := queue.count(); n != 0 {
|
|
t.Fatalf("the queue received %d message(s) despite failing", n)
|
|
}
|
|
// There is no "approved" state: the record is already "queued", its own publish just failed.
|
|
req, _, ok, err := getSend(sends, id)
|
|
if err != nil || !ok || req.Decided == nil || req.Decided.Outcome != "queued" {
|
|
t.Fatalf("left queued despite the publish failing: %v %v %+v", err, ok, req.Decided)
|
|
}
|
|
|
|
queue.failErr = nil
|
|
// Within its grace, the sweep leaves a freshly queued record alone.
|
|
b.Sweep(decidedAt.Add(QueuedGrace / 2))
|
|
if n := queue.count(); n != 0 {
|
|
t.Fatalf("the sweep republished within the grace: %d message(s)", n)
|
|
}
|
|
|
|
// Past its grace, the sweep publishes the still-queued send again.
|
|
b.Sweep(decidedAt.Add(QueuedGrace + time.Second))
|
|
if n := queue.count(); n != 1 {
|
|
t.Fatalf("the sweep did not republish the queued send past its grace: %d message(s)", n)
|
|
}
|
|
got := queue.last()
|
|
if got.id != id || got.body.Send != id {
|
|
t.Fatalf("republished %+v, want the send id as the message id", got)
|
|
}
|
|
}
|
|
|
|
// (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). Tested directly against SendMail (smtp.go): since issue 1194 the tools never dial an outgoing
|
|
// server themselves — Decide only records approved and queues the send for the sender (issue 1196, 1198), which
|
|
// is where SendMail is actually called from.
|
|
func TestSendUsesTheParsedAddressForRcptAndTheFormattedAddressForTheToHeader(t *testing.T) {
|
|
fake := newFakeSMTPServer(t)
|
|
acc := Account{Name: "mesh", Address: "jschoubben@novox.be", Outgoing: &Server{Host: "127.0.0.1", Port: 25, Security: "tls"}}
|
|
if err := SendMail(fake.dialer(), acc, "the-password", []string{"Ops Team <ops@example.org>"}, "Status", "body", judgedAt, "send-abc123"); err != nil {
|
|
t.Fatalf("SendMail: %v", err)
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
// (10) A send whose ask was never published (so the router never recorded it) is settled as ended, with its own
|
|
// words, once its asked time plus SendAskLasts has passed — never left "waiting" forever (novox/hq issue 811).
|
|
func TestASendWhoseAskNeverPublishedEndsAfterTheAskWouldHaveLasted(t *testing.T) {
|
|
b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
|
|
router.publishErr = errors.New("the bus is down")
|
|
rt := startBundle(t, b.Tools())
|
|
|
|
_, isErr := rt.call(SendTool, "novox.triage", sendArgs)
|
|
if !isErr {
|
|
t.Fatalf("mail_send did not report the publish failure")
|
|
}
|
|
|
|
// The send was kept despite the publish failure (so a later sweep can still find it).
|
|
keys, err := sends.Keys()
|
|
if err != nil {
|
|
t.Fatalf("%v", err)
|
|
}
|
|
var id string
|
|
for _, k := range keys {
|
|
if strings.HasPrefix(k, "send-") {
|
|
id = k
|
|
}
|
|
}
|
|
if id == "" {
|
|
t.Fatalf("no send was kept despite the publish failure: %v", keys)
|
|
}
|
|
|
|
// Before the ask would have lasted, it still reads waiting: the router may yet come up and record it.
|
|
text, isErr := rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
|
|
if isErr || !strings.Contains(text, `"status":"waiting"`) {
|
|
t.Fatalf("before expiry: %v %s", isErr, text)
|
|
}
|
|
|
|
// Once a sweep runs after the ask would have lasted, it settles as ended, and nothing was sent.
|
|
b.Sweep(judgedAt.Add(SendAskLasts + time.Minute))
|
|
|
|
text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
|
|
if isErr {
|
|
t.Fatalf("%s", text)
|
|
}
|
|
if !strings.Contains(text, `"status":"expired"`) {
|
|
t.Fatalf("does not read expired once the ask would have lasted: %s", text)
|
|
}
|
|
if !strings.Contains(text, "its ask to the operator was never heard, so it was not sent") {
|
|
t.Fatalf("does not say why: %s", text)
|
|
}
|
|
}
|
|
|
|
// (11) The sweep reads only the sends still undecided, by the undecided index, not every send ever kept: a send
|
|
// already decided is never asked about again (novox/hq issue 811).
|
|
func TestSweepReadsOnlyUndecidedSends(t *testing.T) {
|
|
b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
|
|
rt := startBundle(t, b.Tools())
|
|
|
|
// One send, declined right away: decided, so it leaves the undecided index.
|
|
text, isErr := rt.call(SendTool, "novox.triage", sendArgs)
|
|
if isErr {
|
|
t.Fatalf("%s", text)
|
|
}
|
|
var out map[string]any
|
|
_ = json.Unmarshal([]byte(text), &out)
|
|
declinedID := out["ask"].(string)
|
|
w := router.chose(declinedID, OptionDecline, judgedAt.Add(time.Minute))
|
|
router.close(w)
|
|
if _, err := Decide(sends, b.queuePublish(), w, judgedAt.Add(2*time.Minute)); err != nil {
|
|
t.Fatalf("deciding: %v", err)
|
|
}
|
|
|
|
// A second send, left undecided: still waiting on the operator.
|
|
text, isErr = rt.call(SendTool, "novox.triage", sendArgs)
|
|
if isErr {
|
|
t.Fatalf("%s", text)
|
|
}
|
|
_ = json.Unmarshal([]byte(text), &out)
|
|
waitingID := out["ask"].(string)
|
|
|
|
router.resetRecordCalls()
|
|
b.Sweep(judgedAt.Add(3 * time.Minute))
|
|
|
|
if n := router.recordCallsFor(declinedID); n != 0 {
|
|
t.Fatalf("the sweep read the decided send's router record %d time(s): it already left the undecided index", n)
|
|
}
|
|
if n := router.recordCallsFor(waitingID); n == 0 {
|
|
t.Fatalf("the sweep never read the undecided send's router record")
|
|
}
|
|
}
|
|
|
|
// (12) When the undecided index cannot be written, mail_send refuses the send and publishes no ask: nothing was
|
|
// published yet, so refusing costs nothing, and an ask the sweep could never see is never sent (novox/hq issue
|
|
// 811, review round 3).
|
|
func TestMailSendRefusesWhenTheUndecidedIndexCannotBeWritten(t *testing.T) {
|
|
b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
|
|
sends.failKey, sends.failErr = undecidedIndexKey, errors.New("the store is down")
|
|
rt := startBundle(t, b.Tools())
|
|
|
|
text, isErr := rt.call(SendTool, "novox.triage", sendArgs)
|
|
if !isErr {
|
|
t.Fatalf("mail_send did not report the undecided index failing to write: %s", text)
|
|
}
|
|
if n := router.published1(); n != 0 {
|
|
t.Fatalf("an ask was published despite the undecided index failing to write: %d", n)
|
|
}
|
|
}
|
|
|
|
// (13) An undecided index that cannot be parsed is never overwritten: addUndecided and removeUndecided fail,
|
|
// naming the key, and leave the stored bytes as they were, instead of resetting the index to empty (novox/hq
|
|
// issue 811, review round 3).
|
|
func TestAddAndRemoveUndecidedFailOnACorruptIndexRatherThanResettingIt(t *testing.T) {
|
|
sends := newMemSends()
|
|
sends.vals[undecidedIndexKey] = json.RawMessage(`not json`)
|
|
sends.revs[undecidedIndexKey] = 1
|
|
|
|
if err := addUndecided(sends, "send-x"); err == nil || !strings.Contains(err.Error(), undecidedIndexKey) {
|
|
t.Fatalf("addUndecided did not fail naming the key on a corrupt index: %v", err)
|
|
}
|
|
if string(sends.vals[undecidedIndexKey]) != "not json" {
|
|
t.Fatalf("addUndecided overwrote the corrupt index instead of leaving it: %s", sends.vals[undecidedIndexKey])
|
|
}
|
|
|
|
if err := removeUndecided(sends, "send-x"); err == nil || !strings.Contains(err.Error(), undecidedIndexKey) {
|
|
t.Fatalf("removeUndecided did not fail naming the key on a corrupt index: %v", err)
|
|
}
|
|
if string(sends.vals[undecidedIndexKey]) != "not json" {
|
|
t.Fatalf("removeUndecided overwrote the corrupt index instead of leaving it: %s", sends.vals[undecidedIndexKey])
|
|
}
|
|
}
|
|
|
|
// 1194/AC3: a warrant that does not bind the request (its digest is not the ask's) is refused, as today, and
|
|
// nothing is published on mail-sends.
|
|
func TestADigestMismatchIsRefusedAndNothingIsQueued(t *testing.T) {
|
|
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))
|
|
w.AskDigest = "sha256:" + strings.Repeat("0", 64) // not the digest of the ask actually published
|
|
router.close(w)
|
|
|
|
d, err := Decide(sends, b.queuePublish(), w, judgedAt.Add(2*time.Minute))
|
|
if err != nil {
|
|
t.Fatalf("deciding: %v", err)
|
|
}
|
|
if !strings.HasPrefix(d.Outcome, "refused: ") {
|
|
t.Fatalf("outcome: %q, want refused: ...", d.Outcome)
|
|
}
|
|
if n := queue.count(); n != 0 {
|
|
t.Fatalf("the queue received %d message(s) despite the digest mismatch", n)
|
|
}
|
|
}
|
|
|
|
// ---- mail_send_status answers only its caller (novox/hq ADR 0440 §7, issue 1261) ---------------------------------
|
|
|
|
// 1261/AC1: the caller that made a send reads its own status in full, with its account, recipients and subject
|
|
// as before.
|
|
func TestSendStatusAnswersTheCallersOwnSendInFull(t *testing.T) {
|
|
b, _, _ := 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)
|
|
|
|
text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
|
|
if isErr {
|
|
t.Fatalf("%s", text)
|
|
}
|
|
var status map[string]any
|
|
if err := json.Unmarshal([]byte(text), &status); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if status["status"] != "waiting" || status["account"] != "mesh" || status["subject"] != sendArgs["subject"] {
|
|
t.Fatalf("the caller's own send: %v", status)
|
|
}
|
|
to, _ := status["to"].([]any)
|
|
if len(to) != 2 {
|
|
t.Fatalf("to: %v", status["to"])
|
|
}
|
|
}
|
|
|
|
// 1261/AC2: a caller other than the one that made the send reads status unknown and nothing else of it — no
|
|
// account, recipients or subject, and no error either.
|
|
func TestSendStatusAnswersUnknownAndNothingElseToAnotherCaller(t *testing.T) {
|
|
b, _, _ := 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)
|
|
|
|
text, isErr = rt.call(SendStatusTool, "novox.someone-else", map[string]any{"id": id})
|
|
if isErr {
|
|
t.Fatalf("another caller's read of the send was an error: %s", text)
|
|
}
|
|
var status map[string]any
|
|
if err := json.Unmarshal([]byte(text), &status); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(status) != 1 || status["status"] != "unknown" {
|
|
t.Fatalf("another caller's read of the send: %v, want only status unknown", status)
|
|
}
|
|
}
|
|
|
|
// 1261/AC3: an id no send is kept under answers status unknown and no error, for any caller.
|
|
func TestSendStatusAnswersUnknownForAnUnknownIDWithoutAnError(t *testing.T) {
|
|
b, _, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
|
|
rt := startBundle(t, b.Tools())
|
|
|
|
text, isErr := rt.call(SendStatusTool, "novox.triage", map[string]any{"id": "send-0000000000000000"})
|
|
if isErr {
|
|
t.Fatalf("an unknown id was an error: %s", text)
|
|
}
|
|
var status map[string]any
|
|
if err := json.Unmarshal([]byte(text), &status); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(status) != 1 || status["status"] != "unknown" {
|
|
t.Fatalf("%v", status)
|
|
}
|
|
}
|
|
|
|
// 1261/AC4: every kept decision's status word is one of the fixed vocabulary (waiting, sent, failed, refused,
|
|
// not proven, declined, expired), answered to the send's own caller.
|
|
func TestSendStatusWordsAreTheFixedVocabularyForTheCallersOwnSend(t *testing.T) {
|
|
sends := newMemSends()
|
|
const caller = "novox.triage"
|
|
cases := []struct {
|
|
name, outcome, want string
|
|
}{
|
|
{"queued", "queued", "waiting"},
|
|
{"sending", "sending", "waiting"},
|
|
{"retrying", "retrying", "waiting"},
|
|
{"sent", "sent", "sent"},
|
|
{"declined", "declined", "declined"},
|
|
{"not-proven", "not proven", "not proven"},
|
|
{"failed", "failed: boom", "failed"},
|
|
{"refused", "refused: nope", "refused"},
|
|
{"ended-never-heard", "ended: never heard", "expired"},
|
|
{"ended-other", "ended: cancelled", "expired"},
|
|
}
|
|
for _, c := range cases {
|
|
id := "send-" + c.name
|
|
req := SendRequest{ID: id, Account: "mesh", To: []string{"ops@example.org"}, Subject: "s", Caller: caller,
|
|
At: judgedAt, Decided: &SendDecision{At: judgedAt, Outcome: c.outcome}}
|
|
if _, err := sends.Create(id, req); err != nil {
|
|
t.Fatalf("%s: keeping: %v", c.name, err)
|
|
}
|
|
out, err := ReadSendStatus(sends, nil, nil, id, caller, judgedAt)
|
|
if err != nil {
|
|
t.Fatalf("%s: %v", c.name, err)
|
|
}
|
|
if out["status"] != c.want {
|
|
t.Fatalf("%s: status %v, want %v", c.name, out["status"], c.want)
|
|
}
|
|
}
|
|
}
|
|
|
|
// 1261/AC5: the seat verb mail-store.send-status answers exactly as mail_send_status, for the same caller.
|
|
func TestSeatSendStatusAnswersExactlyAsMailSendStatus(t *testing.T) {
|
|
b, _, _ := 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)
|
|
|
|
toolText, toolErr := rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
|
|
seatText, seatErr := rt.call(SeatSendStatusTool, "novox.triage", map[string]any{"id": id})
|
|
if toolErr != seatErr || toolText != seatText {
|
|
t.Fatalf("mail_send_status: %v %s; mail-store.send-status: %v %s", toolErr, toolText, seatErr, seatText)
|
|
}
|
|
}
|
|
|
|
// 1261/AC6: the sweep settles a send left undecided no matter who asked for it — never refused by the caller
|
|
// check mail_send_status itself applies.
|
|
func TestSweepSettlesASendRegardlessOfCaller(t *testing.T) {
|
|
b, router, _ := 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) // only the router's record holds the answer
|
|
|
|
// The sweep itself names no caller at all; it must still settle "novox.triage"'s send.
|
|
b.Sweep(judgedAt.Add(2 * time.Minute))
|
|
if n := queue.count(); n != 1 {
|
|
t.Fatalf("the sweep did not settle the send regardless of caller: %d message(s)", n)
|
|
}
|
|
}
|