Merge pull request 'Mail and bus hardening: mesh-mail sends on Approve, desk prompt says what for, duplicate PR delivery, cpu_top (issues 790, 789, 783, 741, 811)' (#316) from feat/mail-hardening into main

This commit was merged in pull request #316.
This commit is contained in:
2026-10-15 14:59:25 +00:00
20 changed files with 2500 additions and 56 deletions
+5 -2
View File
@@ -3,7 +3,9 @@
How loaded a machine's storage is, read from `/proc` and `/sys` and nothing else: the share of CPU time
spent waiting on I/O, I/O pressure stall, each disk's busy share, throughput, IOPS, wait and queue
depth, the filesystems' space, and the units, containers and processes doing the I/O (novox/hq issue
312). It changes nothing and declares no resources, so it can be assigned to every machine.
312). It also reads a machine's CPU use by systemd unit and by container, from each cgroup's cpu.stat,
so an overload outside every container can be named (novox/hq issue 741). It changes nothing and
declares no resources, so it can be assigned to every machine.
## Why it exists, and why here
@@ -22,7 +24,7 @@ machine's resources a seat, these tools become its verbs.
## Tools
All four only read.
All five only read.
| tool | what | replaces |
|---|---|---|
@@ -30,6 +32,7 @@ All four only read.
| `disk_top` | Over a window. `by=unit` (the default) ranks systemd units and containers by bytes moved, from their cgroups. `by=container` ranks containers by name, and `by=process` ranks processes | `iotop` |
| `disk_filesystems` | Each filesystem that stores data, said once however many places it is mounted: size, used and free GiB, used % as `df` counts it, inodes used % | `df -h`, `df -i` |
| `disk_devices` | Each disk without sampling: model, size, whether it spins, scheduler, queue limit, what it is built on, and since boot the GiB read and written, the busy share and the wait per read and per write | `lsblk`, `cat /proc/diskstats` |
| `cpu_top` | Samples a window (`seconds`, default 3, at most 10). It answers the machine's CPU in use (% of all CPUs and as CPUs busy, from `/proc/stat`), then `by=unit` (the default) ranks systemd units and containers by CPU, from their cgroups' cpu.stat, busiest first (`limit`, default 15), with the busy time no unit accounts for; `by=container` ranks containers by name | `top`, `systemd-cgtop` |
### How each number is read
+15 -6
View File
@@ -68,9 +68,10 @@ func (m *Machine) bottomDevices() map[string]bool {
return out
}
// unitIO is every unit's storage I/O, by its cgroup's path: the innermost .service or .scope, because
// io.stat counts a cgroup's descendants too and an outer unit would count them twice.
func (m *Machine) unitIO(counted map[string]bool) map[string]cgroupIO {
// unitPaths is every systemd unit's cgroup path under /sys/fs/cgroup: the innermost .service or
// .scope, because a cgroup's own stat file counts its descendants too and an outer unit would count
// them twice. Shared by every reader that walks the cgroup tree (unitIO here, unitCPU in cpu.go).
func (m *Machine) unitPaths() []string {
base := m.path("/sys/fs/cgroup")
var units []string
filepath.WalkDir(base, func(p string, d fs.DirEntry, err error) error {
@@ -89,11 +90,19 @@ func (m *Machine) unitIO(counted map[string]bool) map[string]cgroupIO {
outer[dir] = true
}
}
out := map[string]cgroupIO{}
out := make([]string, 0, len(units))
for _, u := range units {
if outer[u] {
continue
if !outer[u] {
out = append(out, u)
}
}
return out
}
// unitIO is every unit's storage I/O, by its cgroup's path.
func (m *Machine) unitIO(counted map[string]bool) map[string]cgroupIO {
out := map[string]cgroupIO{}
for _, u := range m.unitPaths() {
text := m.read("/sys/fs/cgroup/" + u + "/io.stat")
if text == "" {
continue
+176
View File
@@ -0,0 +1,176 @@
package main
import (
"path/filepath"
"sort"
"strconv"
"strings"
"time"
)
// cgroupCPU is a cgroup's processor time, from its cpu.stat.
type cgroupCPU struct {
usageUsec uint64
}
// parseCPUStat reads a cgroup's cpu.stat ("usage_usec 123\nuser_usec …\nsystem_usec …"), readable by
// every account, unlike /proc/<pid>/stat for another account's process.
func parseCPUStat(s string) cgroupCPU {
var c cgroupCPU
for _, line := range strings.Split(s, "\n") {
k, v, ok := strings.Cut(line, " ")
if !ok {
continue
}
if k == "usage_usec" {
c.usageUsec, _ = strconv.ParseUint(strings.TrimSpace(v), 10, 64)
}
}
return c
}
// unitCPU is every unit's processor time, by its cgroup's path. A cgroup whose cpu.stat cannot be
// read is left out, never an error for the whole answer.
func (m *Machine) unitCPU() map[string]cgroupCPU {
out := map[string]cgroupCPU{}
for _, u := range m.unitPaths() {
text := m.read("/sys/fs/cgroup/" + u + "/cpu.stat")
if text == "" {
continue
}
out[u] = parseCPUStat(text)
}
return out
}
// MachineCPU is the machine's CPU in use over the window, from /proc/stat's cpu line. CPUsBusy is
// rounded only for display; raw, unexported, is the same count unrounded, for topCPU's share and
// unaccounted math (review round 1 of novox/hq issue 741: a small busy time that rounds CPUsBusy to
// 0.0 must not zero out every unit's share along with it).
type MachineCPU struct {
CPUs int `json:"cpus"`
BusyPct float64 `json:"busy_pct"` // % of all CPUs together
CPUsBusy float64 `json:"cpus_busy"` // as a count of CPUs, e.g. 7.5 of 12, rounded for display
raw float64
}
// machineCPUOver is the machine's CPU in use between two readings of /proc/stat: everything but idle
// and iowait, counted against the window's total ticks over every CPU.
func machineCPUOver(a, b cpuTimes) MachineCPU {
total := float64(b.total() - a.total())
if total <= 0 {
return MachineCPU{CPUs: b.cpus}
}
idle := sub(b.idle, a.idle) + sub(b.iowait, a.iowait)
busy := total - idle
if busy < 0 {
busy = 0
}
raw := busy / total * float64(b.cpus)
return MachineCPU{CPUs: b.cpus, BusyPct: round1(busy / total * 100), CPUsBusy: round1(raw), raw: raw}
}
// CPUUsage is a unit's or a container's share of CPU between two readings of its cgroup's cpu.stat.
// Cgroup is its cgroup's path, relative to /sys/fs/cgroup, said beside Unit always (review round 1 of
// novox/hq issue 741): two cgroups can share a leaf name (an outer "init.scope" and a user session's
// "user.slice/user@1000.service/init.scope"), and the path is how they are told apart.
type CPUUsage struct {
Unit string `json:"unit,omitempty"`
Cgroup string `json:"cgroup,omitempty"`
Container string `json:"container,omitempty"`
CPUPct float64 `json:"cpu_pct"` // % of one CPU: 100 = one CPU busy the whole window
SharePct float64 `json:"share_pct"` // this unit's share of the machine's busy time
raw float64
}
// topCPU ranks the units, or the containers, with the most CPU time between two readings of
// cpu.stat, busiest first. unaccountedPct is the machine's busy time no unit accounts for, as a
// share of it (never below 0): the sum is always taken over every unit, whichever by ranks by, so a
// limit on the list never changes it.
func (m *Machine) topCPU(a, b map[string]cgroupCPU, window time.Duration, by string, limit int, cpusBusy float64) ([]CPUUsage, float64) {
windowUsec := window.Seconds() * 1e6
type entry struct {
path, unit, container string
pct float64
}
var entries []entry
var names map[string]string
var sumPct float64
for path, end := range b {
start, ok := a[path]
if !ok {
continue
}
d := sub(end.usageUsec, start.usageUsec)
if d == 0 {
continue
}
pct := d / windowUsec * 100
sumPct += pct
unit := filepath.Base(path)
var container string
if id := containerID.FindStringSubmatch("/" + path); id != nil {
if names == nil {
names = m.containerNames()
}
container = id[1][:12]
if n, ok := names[id[1]]; ok {
container = n
}
}
entries = append(entries, entry{path: path, unit: unit, container: container, pct: pct})
}
// Units are keyed by their cgroup's own path, never by its leaf name (filepath.Base): two cgroups can
// share a leaf name (an outer "init.scope" and a user session's nested one), and keying by the name
// alone would merge them into one row (review round 1 of novox/hq issue 741). Containers stay keyed
// by their name, as before.
groups := map[string]*CPUUsage{}
for _, e := range entries {
key := e.path
if by == "container" {
if e.container == "" {
continue
}
key = e.container
}
g, ok := groups[key]
if !ok {
g = &CPUUsage{Unit: e.unit, Cgroup: e.path, Container: e.container}
if by == "container" {
g.Unit, g.Cgroup = "", ""
}
groups[key] = g
}
g.raw += e.pct
}
out := make([]CPUUsage, 0, len(groups))
for _, g := range groups {
out = append(out, *g)
}
sort.Slice(out, func(i, j int) bool {
if out[i].raw != out[j].raw {
return out[i].raw > out[j].raw
}
return out[i].Unit+out[i].Cgroup+out[i].Container < out[j].Unit+out[j].Cgroup+out[j].Container
})
if len(out) > limit {
out = out[:limit]
}
for i := range out {
out[i].CPUPct = round1(out[i].raw)
if cpusBusy > 0 {
out[i].SharePct = round1(out[i].raw / (cpusBusy * 100) * 100)
}
}
unaccounted := cpusBusy*100 - sumPct
if unaccounted < 0 {
unaccounted = 0
}
unaccountedPct := 0.0
if cpusBusy > 0 {
unaccountedPct = round1(unaccounted / (cpusBusy * 100) * 100)
}
return out, unaccountedPct
}
+182
View File
@@ -0,0 +1,182 @@
package main
import (
"strings"
"testing"
"time"
)
// aCPUHog is a machine whose CPU is 75 % busy (1.5 of 2 CPUs) over a 2 s window: a unit, another unit,
// and a container, each with a known usage_usec delta, leaving part of the machine's busy time
// unaccounted — the shape of issue 739, where 750 % of 12 CPUs went to processes outside containers.
func aCPUHog(t *testing.T) *fake {
f := newFake(t)
f.file("/proc/stat", "cpu 0 0 0 0 0 0 0 0 0 0\ncpu0 0 0 0 0 0 0 0 0 0 0\ncpu1 0 0 0 0 0 0 0 0 0 0\nintr 1\n")
f.then("/proc/stat", "cpu 100 0 50 50 0 0 0 0 0 0\ncpu0 50 0 25 25 0 0 0 0 0 0\ncpu1 50 0 25 25 0 0 0 0 0 0\nintr 2\n")
zero := "usage_usec 0\nuser_usec 0\nsystem_usec 0\n"
cg := func(path string, usec int) {
f.file("/sys/fs/cgroup/"+path+"/cpu.stat", zero)
f.then("/sys/fs/cgroup/"+path+"/cpu.stat", "usage_usec "+itoa(usec)+"\nuser_usec "+itoa(usec)+"\nsystem_usec 0\n")
}
cg("system.slice/alpha.service", 1000000) // 50 % of one CPU over 2 s
cg("system.slice/beta.service", 400000) // 20 %
cg("system.slice/docker-"+store+".scope", 600000) // 30 %, named "cpu-hog" by the runtime
f.answers["docker ps --no-trunc --format {{.ID}}\t{{.Names}}"] = store + "\tcpu-hog\n"
return f
}
func TestCPUTopRanksUnitsAndContainersByCPUOverTheWindow(t *testing.T) {
f := aCPUHog(t)
out, err := tool(t, f.machine(), "cpu_top").Run(map[string]any{"seconds": 2.0})
if err != nil {
t.Fatal(err)
}
m := out.(map[string]any)
if m["by"] != "unit" || f.waited != 2*time.Second {
t.Errorf("the default: %+v, waited %s", m, f.waited)
}
cpu := m["cpu"].(MachineCPU)
if cpu.CPUs != 2 || cpu.BusyPct != 75 || cpu.CPUsBusy != 1.5 {
t.Errorf("machine cpu %+v", cpu)
}
top := m["top"].([]CPUUsage)
if len(top) != 3 {
t.Fatalf("three units moved: %+v", top)
}
if top[0].Unit != "alpha.service" || top[0].CPUPct != 50 || top[0].SharePct != 33.3 {
t.Errorf("busiest first: %+v", top[0])
}
if top[1].Container != "cpu-hog" || top[1].CPUPct != 30 || top[1].SharePct != 20 {
t.Errorf("the container, named by the runtime: %+v", top[1])
}
if top[2].Unit != "beta.service" || top[2].CPUPct != 20 || top[2].SharePct != 13.3 {
t.Errorf("the least busy: %+v", top[2])
}
if m["unaccounted_pct"] != 33.3 {
t.Errorf("unaccounted: %+v (machine busy 150%% of one CPU, units sum to 100%%, 50%% of 150%% left over)", m["unaccounted_pct"])
}
}
func TestCPUTopByContainerNamesOnlyContainers(t *testing.T) {
out, err := tool(t, aCPUHog(t).machine(), "cpu_top").Run(map[string]any{"by": "container", "seconds": 2.0})
if err != nil {
t.Fatal(err)
}
top := out.(map[string]any)["top"].([]CPUUsage)
if len(top) != 1 || top[0].Container != "cpu-hog" || top[0].Unit != "" || top[0].CPUPct != 30 {
t.Errorf("by container: %+v", top)
}
if _, err := tool(t, aCPUHog(t).machine(), "cpu_top").Run(map[string]any{"by": "process"}); err == nil {
t.Error("by=process was accepted; cpu_top only ranks units and containers")
}
}
func TestCPUTopLimitsHowManyAreNamed(t *testing.T) {
out, err := tool(t, aCPUHog(t).machine(), "cpu_top").Run(map[string]any{"limit": 1.0, "seconds": 2.0})
if err != nil {
t.Fatal(err)
}
m := out.(map[string]any)
top := m["top"].([]CPUUsage)
if len(top) != 1 || top[0].Unit != "alpha.service" {
t.Errorf("limit 1: %+v", top)
}
// The limit narrows the list, never the unaccounted share, which is taken over every unit.
if m["unaccounted_pct"] != 33.3 {
t.Errorf("unaccounted with limit 1: %+v", m["unaccounted_pct"])
}
}
func TestCPUTopLeavesOutACgroupThatAppearsOrVanishesWithinTheWindow(t *testing.T) {
f := aCPUHog(t)
f.then("/sys/fs/cgroup/system.slice/short-lived.service/cpu.stat", "usage_usec 500000\n")
out, err := tool(t, f.machine(), "cpu_top").Run(map[string]any{"seconds": 2.0})
if err != nil {
t.Fatal(err)
}
top := out.(map[string]any)["top"].([]CPUUsage)
for _, u := range top {
if u.Unit == "short-lived.service" {
t.Errorf("a unit only in the second reading was counted: %+v", top)
}
}
if !strings.Contains(strings.Join(f.calls, ","), "docker ps") {
t.Errorf("the container runtime was never asked: %v", f.calls)
}
}
// aTinyLoad is a machine with 12 CPUs, only 0.04 of them busy over a 2 s window (busy 1 of 300 total
// ticks): rounded to one decimal for display, CPUsBusy reads 0.0, but a unit using half of that busy
// time still has a non-zero share — review round 1's decision that share_pct and unaccounted_pct are
// computed from the unrounded busy time, never from the rounded CPUsBusy.
func aTinyLoad(t *testing.T) *fake {
f := newFake(t)
cpus := "cpu0 0 0 0 0 0 0 0 0 0 0\ncpu1 0 0 0 0 0 0 0 0 0 0\ncpu2 0 0 0 0 0 0 0 0 0 0\ncpu3 0 0 0 0 0 0 0 0 0 0\n" +
"cpu4 0 0 0 0 0 0 0 0 0 0\ncpu5 0 0 0 0 0 0 0 0 0 0\ncpu6 0 0 0 0 0 0 0 0 0 0\ncpu7 0 0 0 0 0 0 0 0 0 0\n" +
"cpu8 0 0 0 0 0 0 0 0 0 0\ncpu9 0 0 0 0 0 0 0 0 0 0\ncpu10 0 0 0 0 0 0 0 0 0 0\ncpu11 0 0 0 0 0 0 0 0 0 0\n"
f.file("/proc/stat", "cpu 0 0 0 0 0 0 0 0 0 0\n"+cpus)
f.then("/proc/stat", "cpu 1 0 0 299 0 0 0 0 0 0\n"+cpus)
f.file("/sys/fs/cgroup/system.slice/tiny.service/cpu.stat", "usage_usec 0\n")
f.then("/sys/fs/cgroup/system.slice/tiny.service/cpu.stat", "usage_usec 60000\n") // 3 % of one CPU over 2 s
return f
}
func TestCPUTopSharesAreNonZeroEvenWhenRoundedCPUsBusyIsZero(t *testing.T) {
out, err := tool(t, aTinyLoad(t).machine(), "cpu_top").Run(map[string]any{"seconds": 2.0})
if err != nil {
t.Fatal(err)
}
m := out.(map[string]any)
cpu := m["cpu"].(MachineCPU)
if cpu.CPUs != 12 || cpu.CPUsBusy != 0 {
t.Fatalf("the machine 0.04 CPUs busy, rounded for display: %+v", cpu)
}
top := m["top"].([]CPUUsage)
if len(top) != 1 || top[0].Unit != "tiny.service" || top[0].CPUPct != 3 || top[0].SharePct != 75 {
t.Fatalf("a non-zero share from the unrounded busy time: %+v", top)
}
if m["unaccounted_pct"] != 25.0 {
t.Fatalf("unaccounted from the unrounded busy time: %+v (want 25, so shares plus unaccounted add to 100)",
m["unaccounted_pct"])
}
}
// TestCPUTopKeepsCgroupsWithTheSameUnitNameAsSeparateRows is review round 1's other decision: units are keyed
// by their cgroup path, never by filepath.Base, so two cgroups that both end in "init.scope" — one at the
// root, one under a user's session — are two rows, not one merged by their shared leaf name.
func TestCPUTopKeepsCgroupsWithTheSameUnitNameAsSeparateRows(t *testing.T) {
f := aCPUHog(t)
cg := func(path string, usec int) {
f.file("/sys/fs/cgroup/"+path+"/cpu.stat", "usage_usec 0\n")
f.then("/sys/fs/cgroup/"+path+"/cpu.stat", "usage_usec "+itoa(usec)+"\n")
}
cg("init.scope", 1000000) // the root init, 50 % of one CPU over 2 s
cg("user.slice/user@1000.service/init.scope", 400000) // a user session's init, 20 %
out, err := tool(t, f.machine(), "cpu_top").Run(map[string]any{"seconds": 2.0, "limit": 10.0})
if err != nil {
t.Fatal(err)
}
top := out.(map[string]any)["top"].([]CPUUsage)
var inits []CPUUsage
for _, u := range top {
if u.Unit == "init.scope" {
inits = append(inits, u)
}
}
if len(inits) != 2 {
t.Fatalf("two cgroups named init.scope, merged into %d row(s): %+v", len(inits), top)
}
byCgroup := map[string]CPUUsage{}
for _, u := range inits {
byCgroup[u.Cgroup] = u
}
if u, ok := byCgroup["init.scope"]; !ok || u.CPUPct != 50 {
t.Errorf("the root init.scope: %+v", byCgroup)
}
if u, ok := byCgroup["user.slice/user@1000.service/init.scope"]; !ok || u.CPUPct != 20 {
t.Errorf("the user session's init.scope: %+v", byCgroup)
}
}
+51
View File
@@ -85,6 +85,57 @@ func Tools(m *Machine) []stdio.Tool {
return out, nil
},
},
{
Name: "cpu_top",
Description: "What used the most CPU over a few seconds, from each unit's cgroup cpu.stat, which needs no root: the machine's " +
"CPU in use (as % of all CPUs and as CPUs busy, from /proc/stat), then by=unit (the default) each systemd unit and " +
"container ranked by CPU, busiest first, with the busy time no unit accounts for; by=container only the containers, " +
"by name. Replaces top, systemd-cgtop. (r)",
Input: map[string]any{
"seconds": map[string]any{"type": "integer", "description": "how long to sample, in seconds (default 3, at most 10)"},
"limit": map[string]any{"type": "integer", "description": "how many units or containers to name (default 15, at most 50)"},
"by": map[string]any{"type": "string", "enum": []string{"unit", "container"}, "description": "what to rank (default unit)"},
},
Run: func(args map[string]any) (any, error) {
n, err := count(args, "seconds", 3, 1, 10)
if err != nil {
return nil, err
}
w := time.Duration(n) * time.Second
limit, err := count(args, "limit", 15, 1, 50)
if err != nil {
return nil, err
}
by := text(args, "by")
switch by {
case "":
by = "unit"
case "unit", "container":
default:
return nil, fmt.Errorf("by is unit or container, not %q", by)
}
ac, err := parseCPU(m.read("/proc/stat"))
if err != nil {
return nil, err
}
a := m.unitCPU()
m.Wait(w)
bc, err := parseCPU(m.read("/proc/stat"))
if err != nil {
return nil, err
}
b := m.unitCPU()
machine := machineCPUOver(ac, bc)
top, unaccounted := m.topCPU(a, b, w, by, limit, machine.raw)
return map[string]any{
"window_s": w.Seconds(),
"by": by,
"cpu": machine,
"top": top,
"unaccounted_pct": unaccounted,
}, nil
},
},
{
Name: "disk_filesystems",
Description: "The space of every filesystem that stores data (not proc, tmpfs or overlay layers): size, used and free in GiB, " +
+2 -1
View File
@@ -5,7 +5,8 @@
"disk_load",
"disk_top",
"disk_filesystems",
"disk_devices"
"disk_devices",
"cpu_top"
],
"build": {
"artifacts": [
@@ -312,19 +312,25 @@ func (h *Holder) PullUpdated(p PullEvent) {
defer h.mu.Unlock()
id := IDOf(p.repository(), p.HeadSHA)
now := h.Now()
startedOver := ""
if d := h.deliveries[id]; d != nil {
// Announced again: nothing new — but for a delivery whose verdict is out of date, the announcement is its
// check asked again against the trunk's head (the merge tool's ask, novox/hq ADR 0323 §5), and its bound
// runs from now (§7). The base head it was announced with is kept, for a check this owner asks after it.
if p.BaseSHA != "" && d.BaseSHA != p.BaseSHA {
d.BaseSHA = p.BaseSHA
h.keep(d)
if startedOver = startedOverFor(d, p.Number); startedOver == "" {
// Announced again: nothing new — but for a delivery whose verdict is out of date, the announcement is its
// check asked again against the trunk's head (the merge tool's ask, novox/hq ADR 0323 §5), and its bound
// runs from now (§7). The base head it was announced with is kept, for a check this owner asks after it.
if p.BaseSHA != "" && d.BaseSHA != p.BaseSHA {
d.BaseSHA = p.BaseSHA
h.keep(d)
}
if d.State == Proposed && d.outOfDate() != "" && d.AskedAgain == nil {
d.AskedAgain = &now
h.keep(d)
}
return
}
if d.State == Proposed && d.outOfDate() != "" && d.AskedAgain == nil {
d.AskedAgain = &now
h.keep(d)
}
return
// A delivery stopped only because its pull request closed unmerged does not end its commit's delivery for
// another pull request with the same head (novox/hq issue 783): started over below, exactly as for a first
// announcement, under the same id — never a transition out of Stopped.
}
replaced := ""
for _, old := range h.deliveries {
@@ -338,11 +344,7 @@ func (h *Holder) PullUpdated(p PullEvent) {
}
}
}
d := &Delivery{ID: id, Repository: p.repository(), Commit: p.HeadSHA, Number: p.Number, Title: p.Title,
Base: p.Base, Branch: p.Head, HTMLURL: p.HTMLURL, CloneURL: p.CloneURL, Paths: p.Paths,
PathsTruncated: p.PathsTruncated, Removed: p.Removed, ModuleDirs: p.ModuleDirs, ModuleDirsSaid: p.ModuleDirsSaid,
After: AfterIn(p.Body), Created: now, BaseSHA: p.BaseSHA}
t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: fmt.Sprintf("%s#%d's head announced", d.Repository, d.Number)})
d, t, err := h.freshAnnouncement(p, now, startedOver)
if err != nil {
h.refuse(err)
return
@@ -360,6 +362,34 @@ func (h *Holder) PullUpdated(p PullEvent) {
}
}
// startedOverFor is why a delivery already kept at a head's id is started over for a pull request's number rather
// than taken as announced again: only when it was stopped solely by an earlier, closed duplicate pull request with
// the same head — still Stopped, ClosedUnmerged, never merged, and under another number (novox/hq issue 783).
// Empty for the ordinary 'announced again' path, or for any other stop (a person's, a group's, its own close).
func startedOverFor(d *Delivery, number int) string {
if d != nil && d.State == Stopped && d.ClosedUnmerged && d.MergedAs == "" && d.Number != number {
return fmt.Sprintf("#%d with the same head closed unmerged", d.Number)
}
return ""
}
// freshAnnouncement makes a delivery for p's head exactly as a first announcement would: its own id, proposed.
// startedOver, when not empty, is folded into the transition's Why, naming the pull request this replaces
// (novox/hq issue 783); the delivery it replaces is swapped out by the caller keeping the new one under the same
// id (keep), never by a transition out of its final state.
func (h *Holder) freshAnnouncement(p PullEvent, now time.Time, startedOver string) (*Delivery, Transition, error) {
d := &Delivery{ID: IDOf(p.repository(), p.HeadSHA), Repository: p.repository(), Commit: p.HeadSHA, Number: p.Number,
Title: p.Title, Base: p.Base, Branch: p.Head, HTMLURL: p.HTMLURL, CloneURL: p.CloneURL, Paths: p.Paths,
PathsTruncated: p.PathsTruncated, Removed: p.Removed, ModuleDirs: p.ModuleDirs, ModuleDirsSaid: p.ModuleDirsSaid,
After: AfterIn(p.Body), Created: now, BaseSHA: p.BaseSHA}
why := fmt.Sprintf("%s#%d's head announced", d.Repository, d.Number)
if startedOver != "" {
why = fmt.Sprintf("%s; %s", why, startedOver)
}
t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: why})
return d, t, err
}
// takePlaceOf puts a member's new head — a rebase, a push — in its group in the place of the head it superseded,
// when the group is closed: a member of it already reached the trunk, and a closed group takes no new member. The
// new head has its own check, and the group's composed check runs again for the new heads rather than holding the
@@ -491,7 +521,29 @@ func (h *Holder) PullMerged(p PullEvent) {
}
now := h.Now()
var ts []Transition
if d == nil {
if over := startedOverFor(d, p.Number); over != "" {
// A delivery stopped only by an earlier, closed duplicate pull request with the same head is not lost when
// its merge is heard without the open announcement either: it is started over for the merging pull request
// the same way an announcement would, before the merge is taken as usual (novox/hq issue 783). Heard this
// way — its own merge, with no open announcement of this pull request — it is made exactly as a first-heard
// merge in every field but its id: FromForge, and its check the forge's word on the head, so a long-past
// merge replayed from the bus's history still falls under the history rule (issue 313) and is not held for
// a person (review round 1).
fresh, t, err := h.freshAnnouncement(p, now, over)
if err != nil {
h.refuse(err)
return
}
fresh.FromForge = true
if s := p.HeadChecks["mesh/merge-gate"]; s != "" {
fresh.Check = &Verdict{At: now, Gate: verdictOfState(s), Summary: "the head's mesh/merge-gate, as the forge holds it",
Repo: verdictOfState(p.HeadChecks["mesh/repo-check"])}
}
d, ts = fresh, []Transition{t}
if d.Check != nil {
ts = append(ts, settle(d, Facts{Now: now})...)
}
} else if d == nil {
head := p.HeadSHA
if head == "" {
head = p.MergeCommit
@@ -0,0 +1,151 @@
package main
import (
"strings"
"testing"
"time"
)
// novox/hq issue 783: mesh-catalog commit b5417c28 was pushed while #305 (from branch adr-359-mesh-mail-733) was
// open, so this owner recorded its delivery as #305's. #305 then closed unmerged (its base branch had been
// deleted), and #315, opened from the same branch and head, merged it as 0914a33f. The delivery stayed stopped
// ("the pull request closed unmerged"), though it recorded merged_as 0914a33f: a rule did not hold — a delivery
// keyed by commit follows the pull request that merges that commit, and a closed duplicate does not end it — and
// the walk that carried the merge waited 64 minutes for this owner's word until a person's `plans go` released it.
//
// A delivery stopped only because its pull request closed unmerged, with no merge recorded, does not end its
// commit's delivery for another pull request with the same head: it is started over, under the same id, for the
// pull request that announces or merges it next.
const (
head783, merge783 = "d305d305d305d305", "d315d315d315d315"
head783b, merge783b = "d306d306d306d306", "d316d316d316d316"
head783c = "d307d307d307d307"
head783d, merge783d = "d308d308d308d308", "d318d318d318d318"
catalog783 = "novox/mesh-catalog"
branch783 = "adr-359-mesh-mail-733"
)
// TestADeliveryStoppedByAClosedDuplicateFollowsThePullRequestThatMergesItsCommit is the issue's own shape: #305
// announced, closed unmerged (stopped); #315 announced from the same head, checked and readied; #315 merges it,
// and the walk that answers that merge is let go — the delivery follows #315, never stuck on #305's stop.
func TestADeliveryStoppedByAClosedDuplicateFollowsThePullRequestThatMergesItsCommit(t *testing.T) {
w := newWorld(t)
id := IDOf(catalog783, head783)
w.h.PullUpdated(pr(catalog783, 305, head783, branch783, "src/a.go"))
w.h.PullClosed(pr(catalog783, 305, head783, branch783))
if w.state(id) != Stopped {
t.Fatalf("#305 closed unmerged left it %s; want stopped", w.state(id))
}
w.h.PullUpdated(pr(catalog783, 315, head783, branch783, "src/a.go"))
if d := w.delivery(id); d.State == Stopped || d.Number != 315 {
t.Fatalf("#315, the same head, left it %s, Number %d; want started over for #315, not stopped", d.State, d.Number)
}
if d := w.delivery(id); !strings.Contains(d.Transitions[len(d.Transitions)-1].Why, "#305 with the same head closed unmerged") {
t.Fatalf("the start-over's transition names what it replaces: %q", d.Transitions[len(d.Transitions)-1].Why)
}
w.h.Checked(verdict(catalog783, 315, head783, "pass"))
if d := w.delivery(id); d.State != Ready {
t.Fatalf("a passing verdict for #315 left it %s", d.State)
}
walk := w.ctl.put(aWalk("plan-783", catalog783, merge783, true))
w.h.WalkMoved(walk)
w.h.PullMerged(merged(catalog783, 315, head783, merge783))
if d := w.delivery(id); d.State != Published || d.Number != 315 {
t.Fatalf("#315's merge left it %s, Number %d; want published under #315", d.State, d.Number)
}
w.settleAll()
if len(w.ctl.delivers) != 1 || w.ctl.delivers[0] != "plan-783" {
t.Fatalf("the walk that answers #315's merge was not let go: %v", w.ctl.delivers)
}
if d := w.delivery(id); d.State != Delivering {
t.Fatalf("let go, it is %s; want delivering", d.State)
}
}
// TestAMergeWithoutAnAnnouncementStartsOverAStoppedDuplicateToo is decision 4: the merging pull request is never
// announced on its own (heard only as `pull.merged`) — the start-over happens in PullMerged itself, not only in
// PullUpdated, so a merge heard without the open announcement is not lost either.
func TestAMergeWithoutAnAnnouncementStartsOverAStoppedDuplicateToo(t *testing.T) {
w := newWorld(t)
id := IDOf(catalog783, head783b)
w.h.PullUpdated(pr(catalog783, 306, head783b, branch783+"-b", "src/b.go"))
w.h.PullClosed(pr(catalog783, 306, head783b, branch783+"-b"))
if w.state(id) != Stopped {
t.Fatalf("#306 closed unmerged left it %s; want stopped", w.state(id))
}
walk := w.ctl.put(aWalk("plan-783b", catalog783, merge783b, true))
w.h.WalkMoved(walk)
m := merged(catalog783, 316, head783b, merge783b)
m.HeadChecks = map[string]string{"mesh/merge-gate": "success", "mesh/repo-check": "success"}
w.h.PullMerged(m)
if d := w.delivery(id); d.State == Stopped || d.Number != 316 {
t.Fatalf("#316's merge, heard with no announcement of its own, left it %s, Number %d; want started over for #316",
d.State, d.Number)
}
w.settleAll()
if len(w.ctl.delivers) != 1 || w.ctl.delivers[0] != "plan-783b" {
t.Fatalf("the walk that answers #316's merge was not let go: %v", w.ctl.delivers)
}
}
// TestAStoppedDeliveryAnnouncedAgainByItsOwnPullRequestStaysStopped guards decision 3's "any other stopped
// delivery ... is untouched": the same pull request announced again after its own close is the ordinary
// 'announced again' path, never a start-over, and stays stopped.
func TestAStoppedDeliveryAnnouncedAgainByItsOwnPullRequestStaysStopped(t *testing.T) {
w := newWorld(t)
id := IDOf(catalog783, head783c)
w.h.PullUpdated(pr(catalog783, 307, head783c, branch783+"-c", "src/c.go"))
w.h.PullClosed(pr(catalog783, 307, head783c, branch783+"-c"))
if w.state(id) != Stopped {
t.Fatalf("#307 closed unmerged left it %s; want stopped", w.state(id))
}
w.h.PullUpdated(pr(catalog783, 307, head783c, branch783+"-c", "src/c.go"))
if d := w.delivery(id); d.State != Stopped || d.Number != 307 {
t.Fatalf("#307 announced again by itself left it %s, Number %d; want it to stay stopped", d.State, d.Number)
}
}
// TestAMergeHeardLongAfterStartingOverAStoppedDuplicateIsHistoryLikeAnyOtherReplayedMerge is review round 1's
// decision: a delivery started over from a merge heard without its open announcement (as in decision 4, above)
// takes FromForge exactly as a first-heard merge does, so one replayed from the bus's history long after it
// happened still falls under the history rule (novox/hq issue 313) and is never held for a person.
func TestAMergeHeardLongAfterStartingOverAStoppedDuplicateIsHistoryLikeAnyOtherReplayedMerge(t *testing.T) {
w := newWorld(t)
id := IDOf(catalog783, head783d)
w.h.PullUpdated(pr(catalog783, 308, head783d, branch783+"-d", "src/d.go"))
w.h.PullClosed(pr(catalog783, 308, head783d, branch783+"-d"))
if w.state(id) != Stopped {
t.Fatalf("#308 closed unmerged left it %s; want stopped", w.state(id))
}
m := merged(catalog783, 318, head783d, merge783d)
m.MergedAt = w.now.Add(-historyLate - time.Minute).UTC().Format(time.RFC3339)
w.h.PullMerged(m)
d := w.delivery(id)
if d.State == Stopped || d.Number != 318 {
t.Fatalf("#318's merge, heard long after it happened with no announcement of its own, left it %s, Number %d; "+
"want started over for #318", d.State, d.Number)
}
if !d.FromForge {
t.Fatal("a merge heard without its own announcement is made from the forge's word: FromForge is false")
}
if d.History == "" {
t.Fatalf("a merge heard %s after it happened is history, never held for a person; History is empty", historyLate+time.Minute)
}
if d.State == Held {
t.Fatalf("history is never held for a person; state is %s, HeldWhy %q", d.State, d.HeldWhy)
}
}
+14 -7
View File
@@ -2,7 +2,9 @@
One module holds the operator's mail accounts, watches them, and says each change of a message on the bus for the
modules the operator grants it to (novox/hq ADR 0359, research 049 document 02). One instance, on the control-node.
This is the first stage: accounts of the kind `imap`, read-only. Nothing here flags, moves, deletes, drafts or sends.
Reading is the first stage: accounts of the kind `imap`, read-only — nothing here flags, moves, deletes or drafts.
Sending is `mail_send` alone (novox/hq ADR 0361 §2): it sends nothing itself, and reaches an account's outgoing
server only once the operator approves it on a channel that proves who answered.
## What it does
@@ -32,12 +34,13 @@ This is the first stage: accounts of the kind `imap`, read-only. Nothing here fl
"folders": ["optional: the folders watched besides the inbox; every folder when left out"]}}
```
`outgoing` is read and never used in this stage. A second `imap` account on another server is one more entry and one
more password; no code changes. `user` signs in under another name than the address.
`outgoing` is the server `mail_send` sends through: host, port, and `security` `tls` (465, implicit) or `starttls`
(587). A second `imap` account on another server is one more entry and one more password; no code changes. `user`
signs in under another name than the address.
`grants` — per account, who may read its mail: a module's name (on every machine), `<machine>.<module>`, or
`person.<name>`. Nobody by default. It restricts the stream `MAIL` on the bus (a person's entry grants nothing there)
and the tools `mail_search` and `mail_read`.
`grants` — per account, who may read its mail, and who may send from it: a module's name (on every machine),
`<machine>.<module>`, or `person.<name>`. Nobody by default. It restricts the stream `MAIL` on the bus (a person's
entry grants nothing there) and the tools `mail_search`, `mail_read` and `mail_send` — the same check for all three.
The password of an account is the own secret `mail-credential-<account>`, given at the desk; the mesh never makes one.
@@ -49,8 +52,12 @@ The password of an account is the own secret `mail-credential-<account>`, given
| `mail_search` | messages of the accounts the caller is granted, by every word of a query, newest first, no body |
| `mail_read` | one message's metadata and text, or one attachment's bytes, to a caller granted its account |
| `mail_health` | healthy; unhealthy naming the account and why (connection down, or no answer for five minutes); waits naming `mail-credential-<account>` while its password is not given |
| `mail_send` | raises an ask on the operator channel (the account, the recipients, the subject, the first lines) for a caller granted the account; answers at once with the ask's id. Sends nothing itself |
| `mail_send_status` | where a send stands, by its ask id: waiting, sent, declined, ended without a choice, or not proven |
The caller is the runtime's word, never an argument.
The caller is the runtime's word, never an argument. `mail_send` sends through the account's outgoing server only
once the operator's Approve comes back on a channel that proves who answered, and once: a warrant delivered twice
never sends twice (novox/hq ADR 0361 §2).
## Its processes
@@ -28,8 +28,8 @@ func TestTheFirstStagesAccountIsReadAndItsSecretNamed(t *testing.T) {
}
// A second account of the kind imap on another server — Gmail with an app password, another member of the family —
// is configuration only: no code changes for it. Each account may also say its outgoing server, which this stage
// reads and never uses (it sends nothing).
// is configuration only: no code changes for it. Each account may also say its outgoing server, which mail_send
// uses to send through it (novox/hq ADR 0361 §1).
func TestASecondIMAPAccountIsConfigurationOnly(t *testing.T) {
got, err := ParseAccounts([]byte(`{"accounts": {
"mesh": {"kind": "imap", "address": "jschoubben@novox.be", "incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"}},
+13 -5
View File
@@ -209,8 +209,11 @@ func TestTheKeeperWaitsForThePasswordAndSaysARefusal(t *testing.T) {
})
}
// It only reads (ADR 0359 §5): no code of the module, but its tests, issues a command that flags, moves, copies,
// deletes, appends, creates, renames or sends, and every folder is opened read-only and every body peeked.
// It only reads a mailbox (ADR 0359 §5): no code of the module, but its tests, issues a command that flags, moves,
// copies, deletes, appends, creates, renames or otherwise writes to one over IMAP, and every folder is opened
// read-only and every body peeked. mail_send is the one exception the operator's own Approve carves out (ADR 0361
// §2): it never touches a mailbox either — it speaks SMTP, in smtp.go alone, to the account's outgoing server, and
// only once the operator approved it (send.go).
func TestNothingButTheTestsWritesToAMailbox(t *testing.T) {
files, err := filepath.Glob("*.go")
if err != nil {
@@ -218,7 +221,9 @@ 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") {
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).
continue
}
raw, err := os.ReadFile(f)
@@ -226,12 +231,15 @@ 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])
}
}
if strings.Contains(string(raw), "net/smtp") {
t.Errorf("%s speaks SMTP: this stage sends nothing", f)
t.Errorf("%s speaks SMTP: only smtp.go does, and only on the operator's Approve (ADR 0361 §2)", f)
}
}
src, _ := os.ReadFile("imap.go")
+44 -3
View File
@@ -1,8 +1,10 @@
// mesh-mail: the operator's mail accounts, watched, and each mail said on the bus (novox/hq ADR 0359). One instance,
// on the control-node. The first stage holds the mesh's own server, read-only: no flag, move, delete, draft or send.
// on the control-node. Reading is read-only: no flag, move, delete or draft. mail_send is the one way out (novox/hq
// ADR 0361 §2): it sends nothing itself, and reaches an account's outgoing server only on the operator's Approve.
//
// mesh-mail the bundle the node's runtime launches: the tools mail_accounts, mail_search, mail_read and
// mail_health, and the events the keeper wrote, said on the bus as the module
// mesh-mail the bundle the node's runtime launches: the tools mail_accounts, mail_search, mail_read,
// mail_health, mail_send and mail_send_status, and the events the keeper wrote, said on the
// bus as the module
// mesh-mail prepare the preparation step: the index's schema brought to this version's
// mesh-mail keep acts: watches every account (IDLE on the inbox, a look at the other folders every five
// minutes), keeps the index and the attachment store, and writes each change's event beside it
@@ -17,6 +19,7 @@ package main
import (
"context"
"encoding/json"
"fmt"
"os"
"os/signal"
@@ -25,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.
@@ -176,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.
@@ -78,7 +78,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) {
for _, tool := range (&Bundle{}).Tools() {
names = append(names, tool.Name)
}
if !reflect.DeepEqual(m.Tools, names) || !reflect.DeepEqual(names, []string{"mail_accounts", "mail_search", "mail_read", "mail_health"}) {
if !reflect.DeepEqual(m.Tools, names) || !reflect.DeepEqual(names, []string{"mail_accounts", "mail_search", "mail_read", "mail_health", "mail_send", "mail_send_status"}) {
t.Fatalf("tools %v, the code serves %v", m.Tools, names)
}
// Its events go to its own stream MAIL, kept a week, each account's part heard only by the consumers the trusted
+643
View File
@@ -0,0 +1,643 @@
package main
// mail_send (novox/hq ADR 0361 §2): the tool that replaces the no-send rule of ADR 0359 §5, for sending alone. It
// sends nothing itself. A caller granted the account (the same check as mail_read, ADR 0359 §4) asks for one send;
// this raises an ask on the operator channel (novox/hq ADR 0259) showing the account, the recipients, the subject
// and the first lines of the body, and the tool returns at once with the ask's id. The mail reaches the account's
// outgoing server only on the operator's Approve, on a channel that proves who answered, and once: the id is kept
// with the decision it was given, so a warrant delivered twice is acted on once (asks.Warrant.For and
// asks.Option.Performs hold the decision to exactly the request that was asked). A Decline, or no answer before it
// expires, sends nothing, which mail_send_status says. The password is read only to sign in, and the body only to
// send; neither is ever logged.
//
// The pattern — publish an ask under this module's own name, compose it again from what was kept to hold the
// router's warrant, act once on a compare-and-set claim — follows mesh-knowledge's own approval of a record
// (cmd/mesh-knowledge/operator.go) and claude-code's proposals and questions (cmd/claude-code/proposals.go,
// questions.go).
import (
"crypto/rand"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"net/mail"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/asks"
)
// SendsState is the module's state the sends are kept in.
const SendsState = "sends"
// The two answers, both at the level approve, so only a channel that proves who answered can carry either.
const (
OptionSend = "send"
OptionDecline = "decline"
)
// 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
// undecidedIndexKey is a reserved key in the `sends` state (never a send id, which always starts "send-") that
// keeps the ids of the sends still undecided, so the sweep can list only those instead of every send ever kept
// (novox/hq issue 811). It stays in the same declared state rather than a state of its own, which a manifest
// change would need.
const undecidedIndexKey = "undecided-index"
// undecidedIndex is the undecidedIndexKey document: the ids of the sends still undecided.
type undecidedIndex struct {
IDs []string `json:"ids"`
}
// addUndecided adds id to the undecided index, retrying its compare-and-set write on a race. A stored index that
// cannot be parsed is never overwritten (which would silently drop every id it already named): it is reported as
// an error naming the key instead (novox/hq issue 811, review round 3).
func addUndecided(store SendStore, id string) error {
for {
raw, rev, ok, err := store.Get(undecidedIndexKey)
if err != nil {
return err
}
var idx undecidedIndex
if ok {
if err := json.Unmarshal(raw, &idx); err != nil {
return fmt.Errorf("the undecided index %q cannot be parsed: %w", undecidedIndexKey, err)
}
for _, existing := range idx.IDs {
if existing == id {
return nil
}
}
}
idx.IDs = append(idx.IDs, id)
if !ok {
if _, err := store.Create(undecidedIndexKey, idx); err != nil {
if errors.Is(err, stdio.ErrChanged) {
continue
}
return err
}
return nil
}
if _, err := store.Update(undecidedIndexKey, idx, rev); err != nil {
if errors.Is(err, stdio.ErrChanged) {
continue
}
return err
}
return nil
}
}
// removeUndecided drops id from the undecided index, retrying its compare-and-set write on a race. A send never
// read by the index (already absent) is left as it is. A stored index that cannot be parsed is never overwritten:
// it is reported as an error naming the key instead (novox/hq issue 811, review round 3).
func removeUndecided(store SendStore, id string) error {
for {
raw, rev, ok, err := store.Get(undecidedIndexKey)
if err != nil || !ok {
return err
}
var idx undecidedIndex
if err := json.Unmarshal(raw, &idx); err != nil {
return fmt.Errorf("the undecided index %q cannot be parsed: %w", undecidedIndexKey, err)
}
at := -1
for i, existing := range idx.IDs {
if existing == id {
at = i
break
}
}
if at == -1 {
return nil
}
idx.IDs = append(idx.IDs[:at], idx.IDs[at+1:]...)
if _, err := store.Update(undecidedIndexKey, idx, rev); err != nil {
if errors.Is(err, stdio.ErrChanged) {
continue
}
return err
}
return nil
}
}
// undecidedIDs is every id the undecided index currently names, for the sweep to read. An index that cannot be
// parsed is reported as an error naming the key, not silently read as empty (novox/hq issue 811, review round 3).
func undecidedIDs(store SendStore) ([]string, error) {
raw, _, ok, err := store.Get(undecidedIndexKey)
if err != nil || !ok {
return nil, err
}
var idx undecidedIndex
if err := json.Unmarshal(raw, &idx); err != nil {
return nil, fmt.Errorf("the undecided index %q cannot be parsed: %w", undecidedIndexKey, err)
}
return idx.IDs, nil
}
// The bounds of what mail_send takes.
const (
MostRecipients = 20
MostSubject = 200
MostSendBody = 1 << 20 // 1 MiB; only the first ShownBodyChars reach the operator's ask
ShownBodyChars = 300
)
// SendRequest is one send asked of the operator, as the `sends` state keeps it. Every field the ask composes from
// is here, so the decision is held to exactly what was asked.
type SendRequest struct {
ID string `json:"id"`
Account string `json:"account"`
To []string `json:"to"`
Subject string `json:"subject"`
Body string `json:"body"`
Caller string `json:"caller"`
At time.Time `json:"at"`
// Decided is set once, and never changed again: the ask is acted on once (ADR 0361 §2).
Decided *SendDecision `json:"decided,omitempty"`
}
// SendDecision is what became of one send, kept once and read back by mail_send_status.
type SendDecision struct {
At time.Time `json:"at"`
Outcome string `json:"outcome"`
Warrant string `json:"warrant,omitempty"`
Error string `json:"error,omitempty"`
}
// bodyDigest is the digest an act binds in place of the body itself: never the body, which the act's digest is
// published and read back on the bus.
func bodyDigest(body string) string {
sum := sha256.Sum256([]byte(body))
return "sha256:" + hex.EncodeToString(sum[:])
}
// Act is what an answer binds: this request, whole, and the option chosen.
func (s SendRequest) Act(option string) asks.Act {
return asks.Act{"verb": Module + ".send", "option": option, "send": s.ID, "account": s.Account,
"to": strings.Join(s.To, ","), "subject": s.Subject, "body-digest": bodyDigest(s.Body),
"caller": s.Caller, "at": s.At.UTC().Format(time.RFC3339Nano)}
}
// firstChars is at most n runes of s, marked when it was cut.
func firstChars(s string, n int) string {
r := []rune(s)
if len(r) <= n {
return s
}
return string(r[:n]) + " […cut]"
}
// Ask is the ask for this request, composed from it alone: the account, the recipients, the subject and the first
// lines of the body (novox/hq ADR 0361 §2).
func (s SendRequest) Ask() asks.Ask {
to := strings.Join(s.To, ", ")
headline := fmt.Sprintf("Send mail from %s?", s.Account)
if len([]rune(headline)) > asks.HeadlineLength {
headline = "Send mail?"
}
explanation := fmt.Sprintf("From %s to %s: %q.", s.Account, to, s.Subject)
if len([]rune(explanation)) > asks.SummaryLength {
explanation = fmt.Sprintf("From %s: %q.", s.Account, s.Subject)
}
details := fmt.Sprintf("Asked by %s through the mesh-mail module, at %s UTC.\nAccount: %s\nTo: %s\nSubject: %s\n\n%s",
s.Caller, s.At.UTC().Format("15:04 on 2 Jan"), s.Account, to, s.Subject, firstChars(s.Body, ShownBodyChars))
a := asks.Ask{ID: s.ID, Headline: headline, Explanation: explanation, Details: details,
Who: asks.Operator, Expires: s.At.Add(SendAskLasts), OnExpiry: "nothing is sent"}
for _, o := range []struct{ id, label, does string }{
{OptionSend, "Send", "the mail is sent through " + s.Account + "'s outgoing server"},
{OptionDecline, "Decline", "nothing is sent"},
} {
binds, _ := asks.ActDigest(s.Act(o.id))
a.Options = append(a.Options, asks.Option{ID: o.id, Label: o.label, Does: o.does, Level: asks.Approve, Binds: binds})
}
return a
}
func newSendID() string {
var b [8]byte
_, _ = rand.Read(b[:])
return "send-" + hex.EncodeToString(b[:])
}
// ---- the bus, as a send reaches it ----------------------------------------------------------------------------
// AskBus is the operator channel as this module reaches it: publish an ask under its own name, and read the
// router's own record of one of its asks.
type AskBus interface {
Publish(subject string, body any, id string) error
Record(id string) (*stdio.StateEntry, error)
}
type sdkAskBus struct{}
func (sdkAskBus) Publish(subject string, body any, id string) error {
return stdio.SeatPublish(subject, body, id)
}
func (sdkAskBus) Record(id string) (*stdio.StateEntry, error) { return stdio.SeatRecord("", id) }
// SendStore is the `sends` state: a value with its revision, created only when absent, updated only at the revision
// read — so of two writers only the first writes (the compare-and-set a decision is claimed by).
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.
type keptSends struct{ s stdio.KeptState }
func (k keptSends) Get(key string) (json.RawMessage, uint64, bool, error) {
e, err := k.s.Get(key)
if err != nil || e == nil {
return nil, 0, false, err
}
return e.Value, e.Revision, true, nil
}
func (k keptSends) Create(key string, value any) (uint64, error) { return k.s.Create(key, value) }
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)
if err != nil || !ok {
return SendRequest{}, 0, false, err
}
var req SendRequest
if json.Unmarshal(raw, &req) != nil || req.ID != id {
return SendRequest{}, 0, false, nil
}
return req, rev, true, nil
}
// routerRecord is the router's record of one ask, as far as this module reads it.
type routerRecord struct {
State string `json:"state"`
Warrant *asks.Warrant `json:"warrant"`
}
// ---- asking ----------------------------------------------------------------------------------------------------
// SendInput is one send a caller asks for: the account is already the caller's to use (the grants check, the same
// as mail_read's, is the tool's to make before this is called).
type SendInput struct {
Account Account
To []string
Subject string
Body string
Caller string
}
// cleanAddresses trims and drops empty entries.
func cleanAddresses(in []string) []string {
var out []string
for _, a := range in {
if a = strings.TrimSpace(a); a != "" {
out = append(out, a)
}
}
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.
func AskMailSend(in SendInput, store SendStore, bus AskBus, now time.Time) (map[string]any, error) {
refuse := func(why string) (map[string]any, error) { return nil, errors.New(why + ". Nothing was sent") }
to := cleanAddresses(in.To)
switch {
case len(to) == 0:
return refuse("to names no recipient")
case len(to) > MostRecipients:
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))
}
}
subject := strings.TrimSpace(in.Subject)
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))
}
switch {
case strings.TrimSpace(in.Body) == "":
return refuse("body is empty")
case len(in.Body) > MostSendBody:
return refuse(fmt.Sprintf("the body is %d bytes, more than %d", len(in.Body), MostSendBody))
}
if in.Account.Outgoing == nil {
return refuse(fmt.Sprintf("the account %s has no outgoing server configured", in.Account.Name))
}
req := SendRequest{ID: newSendID(), Account: in.Account.Name, To: to, Subject: subject, Body: in.Body,
Caller: in.Caller, At: now.UTC()}
ask := req.Ask()
if err := ask.Check(now); err != nil {
return refuse(err.Error())
}
if _, err := store.Create(req.ID, req); err != nil {
return nil, fmt.Errorf("the send could not be kept, so the operator was not asked: %w. Nothing was sent", err)
}
if err := addUndecided(store, req.ID); err != nil {
// Nothing was published yet, so refusing here costs nothing: an ask the sweep could never see is never
// sent (novox/hq issue 811, review round 3).
return nil, fmt.Errorf("the send could not be made visible to the sweep, so the operator was not asked: %w. Nothing was sent", err)
}
if err := bus.Publish(asks.AskSubject(Module), ask, ask.ID); err != nil {
return nil, fmt.Errorf("the operator could not be asked (%v). Nothing was sent; mail_send_status %s says so", err, req.ID)
}
return map[string]any{
"ask": req.ID,
"status": "waiting",
"asked": "\"" + ask.Headline + "\", to " + to[0] + onwards(to),
"from": in.Account.Name,
"expires": ask.Expires.Format(time.RFC3339),
"next": "mail_send_status with this ask id says where it stands; nothing is sent until the operator's Approve, " +
"and nothing on Decline or at expiry",
}, nil
}
func onwards(to []string) string {
if len(to) <= 1 {
return ""
}
return fmt.Sprintf(" and %d more", len(to)-1)
}
// ---- deciding ---------------------------------------------------------------------------------------------------
// AccountLookup resolves an account by name, as the settings currently hold it (not as it stood when it was asked:
// the account's server may have changed since, and its password is read fresh), or found false.
type AccountLookup func(name string) (Account, bool)
// SendActor performs the one send a chosen option authorises, resolving the account's own secret itself. It is
// called at most once per request (Decide claims the request first).
type SendActor func(a Account, to []string, subject, body string, now time.Time) error
// settleSend writes a decision onto a kept send (claiming it, or ending it outright) and, once the write lands,
// drops it from the undecided index: an index entry is never left behind once the record it points to carries a
// decision (novox/hq issue 811). On a race lost to another writer, it reads back whatever that writer decided.
func settleSend(store SendStore, req SendRequest, rev uint64, d SendDecision) (SendDecision, uint64, error) {
next := req
next.Decided = &d
newRev, err := store.Update(req.ID, next, rev)
if err != nil {
if errors.Is(err, stdio.ErrChanged) {
if again, _, ok, err2 := getSend(store, req.ID); err2 == nil && ok && again.Decided != nil {
return *again.Decided, 0, nil
}
}
return d, 0, err
}
_ = removeUndecided(store, req.ID) // best-effort (see removeUndecided)
return d, newRev, nil
}
// Decide holds one closed ask's warrant to the request it answers (asks.Warrant.For, asks.Option.Performs), and —
// on Approve, exactly once — performs the send. A request already decided is never decided again: decide answers
// its kept decision at once, and two warrants for the same request (a redelivery, or two readers racing) have only
// the first claim the request by a compare-and-set write; the second finds it already claimed or decided and sends
// nothing.
func Decide(store SendStore, lookup AccountLookup, send SendActor, w asks.Warrant, now time.Time) (SendDecision, error) {
req, rev, ok, err := getSend(store, w.Ask)
if err != nil {
return SendDecision{}, err
}
if !ok {
return SendDecision{}, fmt.Errorf("no send %q is kept", w.Ask)
}
if req.Decided != nil {
return *req.Decided, nil
}
if w.Asker != Module {
return SendDecision{}, fmt.Errorf("the warrant is for %s's ask %s, not %s's", w.Asker, w.Ask, Module)
}
settle := func(d SendDecision) (SendDecision, error) {
final, _, err := settleSend(store, req, rev, d)
return final, err
}
if w.Outcome != asks.OutcomeChosen {
outcome := "ended: " + string(w.Outcome)
if w.Words != "" {
outcome += " (" + w.Words + ")"
}
return settle(SendDecision{At: now, Outcome: outcome})
}
ask := req.Ask()
option, err := w.For(Module, ask)
if err == nil {
err = option.Performs(req.Act(option.ID))
}
if err != nil {
return settle(SendDecision{At: now, Outcome: "refused: " + err.Error()})
}
switch option.ID {
case OptionDecline:
return settle(SendDecision{At: now, Outcome: "declined", Warrant: w.Says()})
case OptionSend:
return decideSend(store, lookup, send, req, rev, w, now)
default:
return settle(SendDecision{At: now, Outcome: "refused: the option " + option.ID + " is not one mail_send offers"})
}
}
// decideSend claims the request by a compare-and-set write before it sends, so a redelivered warrant finds it
// already claimed (or decided) and acts on nothing: a person's Approve is performed once, never retried on a
// transient failure (the failure is kept in the decision, not hidden).
func decideSend(store SendStore, lookup AccountLookup, send SendActor, req SendRequest, rev uint64, w asks.Warrant, now time.Time) (SendDecision, error) {
claimed := req
claim := SendDecision{At: now, Outcome: "sending", Warrant: w.Says()}
claimedDecision, newRev, err := settleSend(store, req, rev, claim)
if err != nil {
return SendDecision{}, err
}
if newRev == 0 {
// Lost the claim race: another writer already decided this send.
return claimedDecision, nil
}
claimed.Decided = &claim
acc, found := lookup(req.Account)
var final SendDecision
switch {
case !found:
final = SendDecision{At: now, Outcome: "refused: the account no longer exists", Warrant: w.Says()}
default:
if sendErr := send(acc, req.To, req.Subject, req.Body, now); sendErr != nil {
final = SendDecision{At: now, Outcome: "failed: " + sendErr.Error(), Warrant: w.Says(), Error: sendErr.Error()}
} else {
final = SendDecision{At: now, Outcome: "sent", Warrant: w.Says()}
}
}
claimed.Decided = &final
if _, err := store.Update(req.ID, claimed, newRev); err != nil {
// The claim stands either way: a person's Approve is acted on once. The caller logs this; it is not retried.
return final, fmt.Errorf("the send %s was %s, and that could not be kept (%w)", req.ID, final.Outcome, err)
}
return final, nil
}
// ---- reading the status -----------------------------------------------------------------------------------------
// ReadSendStatus is where a send asked with mail_send stands, by its ask id: waiting for the operator, sent,
// declined, ended without a choice (expired, refused, cancelled), or not proven. Read from the router's own
// record; when the router's word is a close one that was not decided yet, this is also where a send is first
// decided (and so sent), exactly once (Decide).
func ReadSendStatus(store SendStore, bus AskBus, lookup AccountLookup, send SendActor, id string, now time.Time) (map[string]any, error) {
trimmed := strings.TrimSpace(id)
req, rev, ok, err := getSend(store, trimmed)
if err != nil {
return nil, fmt.Errorf("the send %q cannot be read now (%v); ask again", trimmed, err)
}
if !ok {
return nil, fmt.Errorf("no send %q is kept: the id is the one mail_send gave", trimmed)
}
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, now)
return out, nil
}
entry, err := bus.Record(req.ID)
if err != nil {
return nil, fmt.Errorf("the router's record of the ask cannot be read now (%v); ask again", err)
}
if entry == nil {
// The router never recorded this ask at all (its Publish failed, or it was never heard): once its asked
// time plus SendAskLasts has passed, it never will be, so it is settled as ended rather than left
// "waiting" forever (novox/hq issue 811).
if now.Sub(req.At) > SendAskLasts {
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "ended: never heard"})
if err != nil {
return nil, fmt.Errorf("the send %q could not be settled as ended (%v); ask again", trimmed, err)
}
reportDecision(out, d, now)
return out, nil
}
out["status"], out["words"] = "waiting", "not yet recorded by the router; nothing is sent until the operator answers"
return out, nil
}
var rec routerRecord
if json.Unmarshal(entry.Value, &rec) != nil {
out["status"], out["words"] = "not proven", "the router's record of it cannot be read"
return out, nil
}
if rec.State == "" || rec.State == "open" || rec.Warrant == nil {
out["status"], out["words"] = "waiting", "the operator has not answered yet"
return out, nil
}
d, err := Decide(store, lookup, send, *rec.Warrant, now)
if err != nil {
out["status"], out["words"] = "not proven", err.Error()
return out, nil
}
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. 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 d.Outcome == "ended: never heard":
out["status"], out["words"] = "ended", "its ask to the operator was never heard, so it was not sent"
case strings.HasPrefix(d.Outcome, "failed: "):
out["status"], out["words"] = "failed", "nothing reached the server: "+strings.TrimPrefix(d.Outcome, "failed: ")
case strings.HasPrefix(d.Outcome, "ended: "):
out["status"], out["words"] = "ended", "nothing was sent: "+strings.TrimPrefix(d.Outcome, "ended: ")
case strings.HasPrefix(d.Outcome, "refused: "):
out["status"], out["words"] = "not proven", "nothing was sent: "+strings.TrimPrefix(d.Outcome, "refused: ")
default:
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 only the sends the undecided index still names, and asks the router's own record for each one's
// 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. Unlike reading every send ever kept, its
// work grows only with the sends still undecided, never with every one a module has ever sent (novox/hq ADR 0361
// §2, review round 1 of issue 790; the index, issue 811).
func (b *Bundle) Sweep(now time.Time) {
store := b.sends()
ids, err := undecidedIDs(store)
if err != nil {
if b.Logf != nil {
b.Logf("[mesh-mail] the undecided index cannot be read: %v", err)
}
return
}
for _, id := range ids {
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)
}
}
}
}
@@ -0,0 +1,745 @@
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)
}
// ---- 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.
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", "ops@example.org", "lead@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)
}
}
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)
}
}
}
// (3) On Approve, the fake SMTP server receives exactly one message, even if the warrant is delivered twice.
func TestOnApproveTheFakeSMTPServerReceivesExactlyOneMessageEvenDeliveredTwice(t *testing.T) {
fake := newFakeSMTPServer(t)
b, router, sends := 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)
lookup := b.lookupAccount
actor := b.sendActor()
if _, err := Decide(sends, lookup, actor, w, judgedAt.Add(2*time.Minute)); err != nil {
t.Fatalf("deciding: %v", err)
}
if n := fake.count(); n != 1 {
t.Fatalf("the fake server received %d message(s) after one Approve", n)
}
// The warrant delivered a second time: still exactly one message.
if _, err := Decide(sends, lookup, actor, w, judgedAt.Add(3*time.Minute)); err != nil {
t.Fatalf("deciding again: %v", err)
}
if n := fake.count(); n != 1 {
t.Fatalf("the fake server received %d message(s) after the warrant was delivered twice", n)
}
got := fake.last()
if got.from != "jschoubben@novox.be" || len(got.to) != 2 || got.to[0] != "ops@example.org" || got.to[1] != "lead@example.org" {
t.Fatalf("received %+v", got)
}
if !strings.Contains(got.data, "The nightly gate is red") || !strings.Contains(got.data, "The merge gate failed on the resolver change") {
t.Fatalf("the message sent: %q", got.data)
}
// mail_send_status reports it sent, and reports the same thing again without sending a third time.
text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
if isErr || !strings.Contains(text, `"status":"sent"`) {
t.Fatalf("%v %s", isErr, text)
}
if n := fake.count(); n != 1 {
t.Fatalf("mail_send_status sent again: %d", n)
}
}
// (4) On Decline, nothing is sent, and the status says declined.
func TestOnDeclineNothingIsSentAndTheStatusSaysDeclined(t *testing.T) {
fake := newFakeSMTPServer(t)
b, router, sends := 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, OptionDecline, judgedAt.Add(time.Minute))
router.close(w)
if _, err := Decide(sends, b.lookupAccount, b.sendActor(), w, judgedAt.Add(2*time.Minute)); err != nil {
t.Fatalf("deciding: %v", err)
}
if n := fake.count(); n != 0 {
t.Fatalf("the fake server 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)
}
}
// (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)
}
}
// (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":"ended"`) {
t.Fatalf("does not read ended 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) {
fake := newFakeSMTPServer(t)
b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
b.Dial = fake.dialer()
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.lookupAccount, b.sendActor(), 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])
}
}
+139
View File
@@ -0,0 +1,139 @@
package main
// How mail_send sends, once the operator approved it (novox/hq ADR 0361 §1): through the account's outgoing
// server, implicit TLS from the first byte on 465 or upgraded with STARTTLS on 587, as the account's outgoing
// `security` says — the same two the module already reads for an account's incoming server. The password is the
// own secret mail-credential-<account>, the one IMAP already signs in with; it is read once, used once, and never
// logged, and neither is the body.
import (
"crypto/tls"
"fmt"
"mime"
"net/mail"
"net/smtp"
"strings"
"time"
"unicode"
)
// SMTPDialer opens a connection to an outgoing server, signed in over nothing: implicit TLS on the first byte, or
// plain and then upgraded with STARTTLS, as the server's security says. A test gives a fake dialing a server of its
// own; DialSMTP is the mesh's.
type SMTPDialer func(server Server) (*smtp.Client, error)
// DialSMTP is the dialer for the mesh (novox/hq ADR 0361 §1).
func DialSMTP(server Server) (*smtp.Client, error) {
switch server.Security {
case "tls":
conn, err := tls.Dial("tcp", server.Address(), &tls.Config{ServerName: server.Host, MinVersion: tls.VersionTLS12})
if err != nil {
return nil, fmt.Errorf("the outgoing server %s cannot be reached over TLS: %w", server.Address(), err)
}
c, err := smtp.NewClient(conn, server.Host)
if err != nil {
_ = conn.Close()
return nil, fmt.Errorf("the outgoing server %s did not answer as SMTP: %w", server.Address(), err)
}
return c, nil
case "starttls":
c, err := smtp.Dial(server.Address())
if err != nil {
return nil, fmt.Errorf("the outgoing server %s cannot be reached: %w", server.Address(), err)
}
if ok, _ := c.Extension("STARTTLS"); !ok {
_ = c.Close()
return nil, fmt.Errorf("the outgoing server %s does not offer STARTTLS", server.Address())
}
if err := c.StartTLS(&tls.Config{ServerName: server.Host, MinVersion: tls.VersionTLS12}); err != nil {
_ = c.Close()
return nil, fmt.Errorf("STARTTLS to %s failed: %w", server.Address(), err)
}
return c, nil
default:
return nil, fmt.Errorf("the outgoing security %q sends over neither tls nor starttls", server.Security)
}
}
// SendMail sends one message through the account's outgoing server (novox/hq ADR 0361 §1, §2): signs in with the
// password given, states every recipient, and writes the message once. Never logs the password or the body; an
// error names the server and what it said, never either.
func SendMail(dial SMTPDialer, a Account, password string, to []string, subject, body string, now time.Time) error {
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
}
defer client.Close()
from := a.SignInAs()
auth := smtp.PlainAuth("", from, password, a.Outgoing.Host)
if err := client.Auth(auth); err != nil {
return fmt.Errorf("signing in to %s failed: %w", a.Outgoing.Address(), err)
}
if err := client.Mail(from); err != nil {
return fmt.Errorf("the outgoing server refused the sender: %w", err)
}
for _, rcpt := range rcpts {
if err := client.Rcpt(rcpt); err != nil {
return fmt.Errorf("the outgoing server refused the recipient %s: %w", rcpt, err)
}
}
w, err := client.Data()
if err != nil {
return fmt.Errorf("the outgoing server refused the message: %w", err)
}
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 {
return fmt.Errorf("the outgoing server did not accept the message: %w", err)
}
return client.Quit()
}
// 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") }
header("From", from)
header("To", strings.Join(to, ", "))
header("Subject", encodeHeader(subject))
header("Date", now.Format(time.RFC1123Z))
header("MIME-Version", "1.0")
header("Content-Type", "text/plain; charset=utf-8")
header("Content-Transfer-Encoding", "8bit")
b.WriteString("\r\n")
b.WriteString(crlf(body))
return []byte(b.String())
}
// encodeHeader word-encodes a header's value when it holds more than ASCII; a header a phone never needs to show.
func encodeHeader(s string) string {
for _, r := range s {
if r > unicode.MaxASCII {
return mime.QEncoding.Encode("UTF-8", s)
}
}
return s
}
// crlf gives every line break in s a CR, once, the ending SMTP's DATA takes.
func crlf(s string) string {
s = strings.ReplaceAll(s, "\r\n", "\n")
return strings.ReplaceAll(s, "\n", "\r\n")
}
+146 -7
View File
@@ -1,8 +1,10 @@
package main
// The module's tools (novox/hq ADR 0359 §5): mail_accounts, mail_search, mail_read and mail_health. Who calls is the
// runtime's word (stdio.CallerOf), never an argument; a caller reads an account's mail — its search, a body, an
// attachment — only when the trusted setting `grants` names it for that account (§4). Nothing here writes to a mailbox.
// The module's tools (novox/hq ADR 0359 §5, ADR 0361 §2): mail_accounts, mail_search, mail_read, mail_health,
// mail_send and mail_send_status. Who calls is the runtime's word (stdio.CallerOf), never an argument; a caller
// reads an account's mail, or sends from it, only when the trusted setting `grants` names it for that account (§4).
// Nothing here writes to a mailbox; mail_send writes nothing itself either (send.go) — it only ever reaches the
// account's outgoing server on the operator's own Approve.
import (
"context"
@@ -17,9 +19,11 @@ import (
// Tool names, as the manifest declares them.
const (
AccountsTool = "mail_accounts"
SearchTool = "mail_search"
ReadTool = "mail_read"
AccountsTool = "mail_accounts"
SearchTool = "mail_search"
ReadTool = "mail_read"
SendTool = "mail_send"
SendStatusTool = "mail_send_status"
)
// Bounds of what a tool answers.
@@ -30,13 +34,21 @@ const (
MostAttachment = 8 << 20
)
// Bundle is what the tools reach: the settings, the index and the attachment store.
// Bundle is what the tools reach: the settings, the index, the attachment store, and — for mail_send — the
// operator channel, the `sends` state and the outgoing dialer. The last three are nil outside a test: the real
// bus and dialer (sdkAskBus, keptSends, DialSMTP) are used lazily, so building a Bundle for the read-only tools
// never needs them.
type Bundle struct {
Env Env
Index func(ctx context.Context) (Store, error)
Objects func() (Objects, error)
Caller func(args map[string]any) string
Now func() time.Time
Asks AskBus
Sends SendStore
Dial SMTPDialer
Logf func(format string, a ...any)
}
func (b *Bundle) now() time.Time {
@@ -120,6 +132,27 @@ func (b *Bundle) Tools() []stdio.Tool {
"waits, naming the account's password, while the operator has not given it at the desk.",
Input: schema(nil, map[string]any{}),
Run: func(map[string]any) (any, error) { return ReadStatus(b.Env.Status, b.now()), nil }},
{Name: SendTool,
Description: "Send mail from an account you are granted (the same check as mail_read, novox/hq ADR 0361 §2). It " +
"sends nothing itself: it raises an ask on the operator channel — the account, the recipients, the subject and " +
"the first lines of the body — and answers at once with the ask's id. The mail reaches the account's outgoing " +
"server only once the operator's Approve comes back on a channel that proves who answered, and once; on " +
"Decline, or no answer before it expires, nothing is sent. Refused for an account you are not granted, or one " +
"with no outgoing server. mail_send_status, with this id, says where it stands.",
Input: schema([]string{"account", "to", "subject", "body"}, map[string]any{
"account": text("the account to send from, by its name in mail_accounts"),
"to": map[string]any{"type": "array", "items": map[string]any{"type": "string"},
"description": "the recipients' mail addresses, at most 20"},
"subject": text("the message's subject, at most 200 characters"),
"body": text("the message's plain-text body"),
}),
Run: b.sendTool},
{Name: SendStatusTool,
Description: "Where a send asked with mail_send stands, by its ask id: waiting for the operator, sent, declined, " +
"ended without a choice (expired, refused, cancelled), or not proven. Read from the router's own record, held " +
"to the request mail_send asked (so a warrant for another request, or one changed since, sends nothing here).",
Input: schema([]string{"id"}, map[string]any{"id": text("the ask id mail_send gave")}),
Run: b.sendStatusTool},
}
}
@@ -304,6 +337,112 @@ func refused(caller, account string) error {
return fmt.Errorf("refused: %s is not granted the mail account %s; the operator grants it through the setting grants", caller, account)
}
// ---- mail_send, mail_send_status ---------------------------------------------------------------------------------
// askBus is the operator channel this bundle reaches: the bus's own, unless a test gave one.
func (b *Bundle) askBus() AskBus {
if b.Asks != nil {
return b.Asks
}
return sdkAskBus{}
}
// sends is the `sends` state this bundle keeps: the bus's own, unless a test gave one.
func (b *Bundle) sends() SendStore {
if b.Sends != nil {
return b.Sends
}
return keptSends{stdio.State(SendsState)}
}
// lookupAccount resolves an account by name as the settings currently hold it — read fresh at decide time, never
// the account as it stood when it was asked.
func (b *Bundle) lookupAccount(name string) (Account, bool) {
accounts, err := ReadAccounts(b.Env.Accounts)
if err != nil {
return Account{}, false
}
return accountByName(accounts, name)
}
func accountByName(accounts []Account, name string) (Account, bool) {
for _, a := range accounts {
if a.Name == name {
return a, true
}
}
return Account{}, false
}
// sendActor performs one send on the operator's Approve: it reads the account's own secret itself, so Decide and
// the `sends` state never hold a password.
func (b *Bundle) sendActor() SendActor {
dial := b.Dial
if dial == nil {
dial = DialSMTP
}
return func(a Account, to []string, subject, body string, now time.Time) error {
pw, given, err := ReadSecret(b.Env.Secrets, a)
if err != nil {
return fmt.Errorf("the account %s's password cannot be read: %w", a.Name, err)
}
if !given {
return fmt.Errorf("the account %s has no password yet; the operator gives it at the desk", a.Name)
}
return SendMail(dial, a, pw, to, subject, body, now)
}
}
func stringList(v any) []string {
list, _ := v.([]any)
out := make([]string, 0, len(list))
for _, item := range list {
if s, ok := item.(string); ok {
out = append(out, s)
}
}
return out
}
func (b *Bundle) sendTool(args map[string]any) (any, error) {
if err := only(args, "account", "to", "subject", "body"); err != nil {
return nil, err
}
account, _ := args["account"].(string)
subject, _ := args["subject"].(string)
body, _ := args["body"].(string)
to := stringList(args["to"])
if account == "" {
return nil, fmt.Errorf("%s needs account, by its name in mail_accounts", SendTool)
}
accounts, grants, err := b.settings()
if err != nil {
return nil, err
}
caller := b.caller(args)
// The same grants check as mail_read (novox/hq ADR 0359 §4, ADR 0361 §2): an ungranted caller is refused, and
// nothing is asked.
if !grants.Allows(caller, account) {
return nil, grants.refuse(caller, account)
}
acc, found := accountByName(accounts, account)
if !found {
return nil, fmt.Errorf("no account %q is held; mail_accounts lists the accounts", account)
}
return AskMailSend(SendInput{Account: acc, To: to, Subject: subject, Body: body, Caller: caller}, b.sends(), b.askBus(), b.now())
}
func (b *Bundle) sendStatusTool(args map[string]any) (any, error) {
if err := only(args, "id"); err != nil {
return nil, err
}
id, _ := args["id"].(string)
if id == "" {
return nil, fmt.Errorf("%s needs id, the ask %s gave", SendStatusTool, SendTool)
}
return ReadSendStatus(b.sends(), b.askBus(), b.lookupAccount, b.sendActor(), id, b.now())
}
// ---- arguments ----------------------------------------------------------------------------------------------------
// only refuses an argument the tool does not take: never ignored, and none can say who calls.
+11 -2
View File
@@ -22,14 +22,23 @@
"own-secrets": {
"mail-credential-*": {
"path": "${dir:state}/mail-credential-*.secret",
"issued-by": "outside"
"issued-by": "outside",
"prompt": "the password of the mail account {member}"
}
},
"tools": [
"mail_accounts",
"mail_search",
"mail_read",
"mail_health"
"mail_health",
"mail_send",
"mail_send_status"
],
"uses": [
"operator-channel"
],
"state": [
"sends"
],
"streams": [
{
+33 -3
View File
@@ -46,6 +46,13 @@ var aName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,62}$`)
// break, no markup, no punctuation a sentence is written with — so the prompt's words stay this holder's.
var aCaller = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/@ -]{0,79}$`)
// escapeMarkup is `what` shown as text, never markup (novox/hq issue 789): the window's message is read with
// pango markup on, so a catalogue's words reach it with the characters pango gives meaning to turned to their
// entities first — never interpreted, whatever a module's manifest holds.
var escapeMarkupReplacer = strings.NewReplacer("&", "&amp;", "<", "&lt;", ">", "&gt;")
func escapeMarkup(s string) string { return escapeMarkupReplacer.Replace(s) }
// secretQuestion reads the call. The prompt is this holder's words, written from the module, the secret and the
// machine (the review of 2026-10-09, M4): it says the mesh's controller asks, which the bus makes true — it
// denies these subjects to every principal but the controller — and a caller's own words are refused, so
@@ -81,14 +88,37 @@ func secretQuestion(args map[string]any) (SecretQuestion, error) {
}
askedBy = v
}
// What the secret is for, in the catalogue's own words (novox/hq issue 789): the controller's own line, read
// from the secret's declared `prompt` and never a word the caller chose. One line, so it is shown under the
// module and secret names as its own; its characters are never markup, whatever the catalogue holds.
what := ""
if raw, there := args["what"]; there {
v, err := text(args, "what", true)
if err != nil {
return q, err
}
if strings.ContainsAny(v, "\n\r") {
return q, fmt.Errorf("what %q holds more than one line", raw)
}
what = v
}
q.Prompt = names["secret"] + " for " + names["module"]
q.Message = fmt.Sprintf("The mesh's controller asks for %s, the own secret of %s on %s.", names["secret"],
names["module"], names["node"])
if askedBy != "" {
q.Message += fmt.Sprintf(" It was asked for by %s.", askedBy)
if what != "" {
q.Message += "\nFor: " + escapeMarkup(what)
}
q.Message += " What you type is not shown, and is sealed before it leaves this machine. Type it only if you " +
tail := ""
if askedBy != "" {
tail += fmt.Sprintf(" It was asked for by %s.", askedBy)
}
tail += " What you type is not shown, and is sealed before it leaves this machine. Type it only if you " +
"expect this ask."
if what != "" {
q.Message += "\n" + strings.TrimSpace(tail)
} else {
q.Message += tail
}
key, err := text(args, "seal_to", true)
if err != nil {
return q, err
@@ -150,3 +150,61 @@ func TestThePromptIsWrittenHereFromNamesAndSaysTheControllerAsks(t *testing.T) {
}
}
}
// What the secret is for shows as its own line under the module and secret names, in the catalogue's own
// words (novox/hq issue 789): read from the secret's declared `prompt`, never a word the caller chose, and
// never shown without a declared one — an old module, or one that declares none, prompts as it does today.
func TestWhatShowsAsItsOwnLineUnderTheNamesWhenDeclaredAndNothingWhenNot(t *testing.T) {
key := base64.StdEncoding.EncodeToString(make([]byte, 32))
base := map[string]any{"module": "mesh-mail", "secret": "mail-credential-gmail", "node": "anchor", "seal_to": key}
q, err := secretQuestion(base)
if err != nil {
t.Fatal(err)
}
if strings.Contains(q.Message, "For:") {
t.Errorf("a \"For:\" line with no what declared: %q", q.Message)
}
withWhat := map[string]any{}
for k, v := range base {
withWhat[k] = v
}
withWhat["what"] = "the password of the mail account gmail"
q, err = secretQuestion(withWhat)
if err != nil {
t.Fatal(err)
}
if !strings.Contains(q.Message, "\nFor: the password of the mail account gmail") {
t.Errorf("no \"For:\" line of its own: %q", q.Message)
}
if !strings.Contains(q.Message, "What you type is not shown") {
t.Errorf("the rest of the message was lost: %q", q.Message)
}
}
// What the secret is for is shown as text, never markup (novox/hq issue 789): the prompt's window reads its
// message with markup on, so what the catalogue holds reaches it with the characters markup gives meaning to
// turned to their entities first, and a line holding more than one is refused before anybody sees it.
func TestWhatIsShownAsTextNeverMarkupAndIsOneLine(t *testing.T) {
key := base64.StdEncoding.EncodeToString(make([]byte, 32))
base := map[string]any{"module": "mesh-mail", "secret": "mail-credential-gmail", "node": "anchor", "seal_to": key,
"what": "the password of <b>the</b> mail & account"}
q, err := secretQuestion(base)
if err != nil {
t.Fatal(err)
}
if !strings.Contains(q.Message, "For: the password of &lt;b&gt;the&lt;/b&gt; mail &amp; account") {
t.Errorf("markup was not escaped: %q", q.Message)
}
for _, bad := range []string{"a\nb", "a\rb"} {
multi := map[string]any{}
for k, v := range base {
multi[k] = v
}
multi["what"] = bad
if _, err := secretQuestion(multi); err == nil {
t.Errorf("what=%q holding more than one line was taken", bad)
}
}
}