feat: flag Ubuntu phased updates on PackageUpdate

go.mod

@@ -5,6 +5,7 @@ go 1.26
 require (
 	github.com/google/uuid v1.6.0
 	github.com/hyperboloide/lk v0.0.0-20251220053519-b291812e3216
+	github.com/yuin/goldmark v1.8.6
 	go.mongodb.org/mongo-driver/v2 v2.8.0
 	golang.org/x/crypto v0.54.0
 	google.golang.org/grpc v1.64.0

go.sum

@@ -21,6 +21,8 @@ github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gi
 github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM=
 github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+github.com/yuin/goldmark v1.8.6 h1:d0VcaP1sx9GkFVkoW+KtggpGi2KZ965i14b0+bDQST4=
+github.com/yuin/goldmark v1.8.6/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg=
 go.mongodb.org/mongo-driver/v2 v2.8.0 h1:CxWDGQYY8QQwNjAl/aq2sfWakdnWZynnqJ9F4DhHbP8=
 go.mongodb.org/mongo-driver/v2 v2.8.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=

grpc/pb/patch_test.go

@@ -0,0 +1,46 @@
+package pb
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+// An empty ApplyUpdatesCmd must stay an empty object on the wire, so an old
+// agent and a new server, or a new agent and an old server, agree that it
+// means "install everything, no reboot, no deadline".
+func TestApplyUpdatesCmdEmptyIsEmptyObject(t *testing.T) {
+	b, err := json.Marshal(ApplyUpdatesCmd{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(b) != "{}" {
+		t.Fatalf("got %s, want {}", b)
+	}
+}
+
+func TestPatchResultRoundTrip(t *testing.T) {
+	in := AgentMessage{PatchResult: &PatchResult{
+		CommandId: "c1", Status: PatchStatusOK, OutputTail: "done",
+		PendingAfter: 0, RebootRequired: true, Rebooting: true,
+	}}
+	b, err := json.Marshal(in)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var out AgentMessage
+	if err := json.Unmarshal(b, &out); err != nil {
+		t.Fatal(err)
+	}
+	if out.PatchResult == nil || *out.PatchResult != *in.PatchResult {
+		t.Fatalf("round trip lost data: %+v", out.PatchResult)
+	}
+}
+
+func TestInventoryBootTimeOnWire(t *testing.T) {
+	b, _ := json.Marshal(InventoryReport{BootTimeUnix: 1757800000})
+	var m map[string]any
+	_ = json.Unmarshal(b, &m)
+	if m["boot_time_unix"] != float64(1757800000) {
+		t.Fatalf("boot_time_unix missing: %s", b)
+	}
+}

grpc/pb/phased_test.go

@@ -0,0 +1,23 @@
+package pb
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+// A phased update is flagged on the wire; an ordinary one carries no field at
+// all, so an old server or agent sees exactly the shape it always did.
+func TestPackageUpdatePhasedOnWire(t *testing.T) {
+	b, err := json.Marshal(PackageUpdate{Name: "libkrb5-3", NewVersion: "1.20", Phased: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(b), `"phased":true`) {
+		t.Fatalf("phased missing: %s", b)
+	}
+	b, _ = json.Marshal(PackageUpdate{Name: "curl", NewVersion: "8"})
+	if strings.Contains(string(b), "phased") {
+		t.Fatalf("an ordinary update must not carry phased: %s", b)
+	}
+}

grpc/pb/vantage.pb.go

@@ -85,6 +85,10 @@ type PackageUpdate struct {
 	Name           string `json:"name"`
 	CurrentVersion string `json:"current_version,omitempty"`
 	NewVersion     string `json:"new_version"`
+	// Phased marks an Ubuntu phased update this host is not yet selected for:
+	// apt lists it as upgradable, but an upgrade defers it until the host's
+	// phase comes up. It is pending, not installable, so counts leave it out.
+	Phased bool `json:"phased,omitempty"`
 }
 
 type ReportUpdatesRequest struct {
@@ -123,6 +127,7 @@ type InventoryReport struct {
 	Partitions     []PartitionReport `json:"partitions,omitempty"`
 	Kernel         string            `json:"kernel,omitempty"`
 	RebootRequired bool              `json:"reboot_required,omitempty"`
+	BootTimeUnix   int64             `json:"boot_time_unix,omitempty"` // every report; proves a reboot happened
 }
 type InventoryReportResponse struct{}
 
@@ -161,7 +166,40 @@ type ReportChecksRequest struct {
 }
 type ReportChecksResponse struct{}
 
-type ApplyUpdatesCmd struct{}
+// ApplyUpdatesCmd installs pending OS updates. The zero value means what the
+// command always meant: every pending update, no reboot, no deadline. That is
+// what keeps old servers and new agents, and new servers and old agents,
+// compatible - but only in that direction for Scope: an agent that predates
+// these fields installs everything even when asked for security only, which
+// is why the control plane gates on agent version before sending a scope.
+type ApplyUpdatesCmd struct {
+	Scope            string `json:"scope,omitempty"`              // "" or PatchScopeAll | PatchScopeSecurity
+	RebootIfRequired bool   `json:"reboot_if_required,omitempty"` // reboot only if the OS reports one is owed
+	DeadlineUnix     int64  `json:"deadline_unix,omitempty"`      // 0 = none; the agent caps the upgrade at 2h
+}
+
+const (
+	PatchScopeAll      = "all"
+	PatchScopeSecurity = "security"
+
+	PatchStatusOK          = "ok"
+	PatchStatusFailed      = "failed"
+	PatchStatusUnsupported = "unsupported"
+	PatchStatusBusy        = "busy"
+)
+
+// PatchResult answers an ApplyUpdatesCmd. Rebooting is sent immediately before
+// the agent restarts the host, so the control plane knows to wait for a
+// post-boot inventory report rather than a second result.
+type PatchResult struct {
+	CommandId      string `json:"command_id"`
+	Status         string `json:"status"`
+	Message        string `json:"message,omitempty"`
+	OutputTail     string `json:"output_tail,omitempty"` // at most 64KB, newest bytes
+	PendingAfter   int32  `json:"pending_after"`         // -1 when the post-apply check failed
+	RebootRequired bool   `json:"reboot_required,omitempty"`
+	Rebooting      bool   `json:"rebooting,omitempty"`
+}
 
 type OpenProxyCmd struct {
 	ProxyId string `json:"proxy_id"`
@@ -240,6 +278,7 @@ type AgentMessage struct {
 	StepOutput *StepOutputChunk `json:"step_output,omitempty"`
 
 	WorkloadLogsResult *WorkloadLogsResult `json:"workload_logs_result,omitempty"`
+	PatchResult        *PatchResult        `json:"patch_result,omitempty"`
 }
 
 type AgentReady struct{}

mail/announcement.go

@@ -0,0 +1,80 @@
+package mail
+
+import (
+	"bytes"
+	"fmt"
+	htmltmpl "html/template"
+	"strings"
+
+	"github.com/yuin/goldmark"
+)
+
+// Announcement is one product email to one person: a feature launch, a guide,
+// an offer. vantage-admin's announce package builds it per recipient, because
+// the unsubscribe links carry that person's token.
+type Announcement struct {
+	CategoryLabel  string // "New features"
+	Subject        string
+	BodyMarkdown   string
+	UnsubscribeURL string // RFC 8058 one-click target for this category
+	PreferencesURL string // the no-login preferences page
+	PostalAddress  string // optional footer line
+}
+
+type announcementData struct {
+	Announcement
+	BodyHTML htmltmpl.HTML
+}
+
+// md renders with goldmark's defaults, which omit raw HTML: a pasted <script>
+// never reaches an inbox.
+var md = goldmark.New()
+
+func announcementMessage(a Announcement, publicURL string) (message, error) {
+	var buf bytes.Buffer
+	if err := md.Convert([]byte(a.BodyMarkdown), &buf); err != nil {
+		return message{}, fmt.Errorf("mail: markdown: %w", err)
+	}
+	return render("announcement", announcementData{Announcement: a, BodyHTML: htmltmpl.HTML(buf.String())}, publicURL)
+}
+
+// RenderAnnouncement is the staff preview: exactly what a recipient gets.
+func RenderAnnouncement(a Announcement, publicURL string) (subject, html, text string, err error) {
+	m, err := announcementMessage(a, publicURL)
+	if err != nil {
+		return "", "", "", err
+	}
+	return m.Subject, m.HTML, m.Text, nil
+}
+
+// SendAnnouncement sends one announcement to one address, with the
+// List-Unsubscribe pair Gmail and Yahoo require of bulk senders.
+func (s Sender) SendAnnouncement(to string, a Announcement) error {
+	m, err := s.announcementFor(to, a)
+	if err != nil {
+		return err
+	}
+	return s.send(m)
+}
+
+// announcementFor builds the one-recipient message SendAnnouncement puts on
+// the wire. Every refusal is a *SendError at PhasePrepare: it is about this
+// one message, never the mail server, so a caller moves on to the next row.
+func (s Sender) announcementFor(to string, a Announcement) (message, error) {
+	if strings.Contains(to, ",") {
+		return message{}, &SendError{Phase: PhasePrepare, Err: fmt.Errorf("mail: an announcement goes to exactly one address")}
+	}
+	if a.UnsubscribeURL == "" {
+		return message{}, &SendError{Phase: PhasePrepare, Err: fmt.Errorf("mail: announcement without an unsubscribe URL")}
+	}
+	m, err := announcementMessage(a, s.PublicURL)
+	if err != nil {
+		return message{}, &SendError{Phase: PhasePrepare, Err: err}
+	}
+	m.To = to
+	m.Headers = map[string]string{
+		"List-Unsubscribe":      "<" + a.UnsubscribeURL + ">",
+		"List-Unsubscribe-Post": "List-Unsubscribe=One-Click",
+	}
+	return m, nil
+}

mail/announcement_test.go

@@ -0,0 +1,96 @@
+package mail
+
+import (
+	"bytes"
+	"errors"
+	"net/mail"
+	"strings"
+	"testing"
+)
+
+func sampleAnnouncement() Announcement {
+	return Announcement{
+		CategoryLabel:  "New features",
+		Subject:        "New in Vantage: scheduled workflows",
+		BodyMarkdown:   "Workflows can now run **on a schedule**.\n\n<script>alert(1)</script>\n\n- cron syntax\n- time zones",
+		UnsubscribeURL: "https://api.example/email/unsubscribe?c=features&t=abc",
+		PreferencesURL: "https://hq.example/email-preferences?t=abc",
+		PostalAddress:  "Hostxtra Ltd, 1 High Street, Leeds",
+	}
+}
+
+func TestRenderAnnouncement(t *testing.T) {
+	subject, html, text, err := RenderAnnouncement(sampleAnnouncement(), "https://hq.example")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if subject != "New in Vantage: scheduled workflows" {
+		t.Errorf("subject = %q", subject)
+	}
+	if !strings.Contains(html, "<strong>on a schedule</strong>") {
+		t.Error("markdown not rendered to html")
+	}
+	if strings.Contains(html, "<script>") {
+		t.Error("raw html from markdown reached the email")
+	}
+	for label, body := range map[string]string{"html": html, "text": text} {
+		for _, want := range []string{"email/unsubscribe?c=features", "email-preferences?t=abc", "New features", "Hostxtra Ltd"} {
+			if !strings.Contains(body, want) {
+				t.Errorf("%s body missing %q", label, want)
+			}
+		}
+		if strings.Contains(body, "\u2014") {
+			t.Errorf("%s body contains an em dash", label)
+		}
+	}
+	if !strings.Contains(text, "**on a schedule**") {
+		t.Error("text part should carry the markdown source")
+	}
+}
+
+func TestRenderAnnouncementWithoutPostalAddress(t *testing.T) {
+	a := sampleAnnouncement()
+	a.PostalAddress = ""
+	_, html, _, err := RenderAnnouncement(a, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(html, "<no value>") {
+		t.Error("empty postal address rendered as <no value>")
+	}
+}
+
+func TestSendAnnouncementRefusesListsAndMissingLink(t *testing.T) {
+	s := Sender{Host: "smtp.invalid", From: "updates@example.com"}
+	prepare := func(what string, err error) {
+		t.Helper()
+		var se *SendError
+		if !errors.As(err, &se) || se.Phase != PhasePrepare {
+			t.Errorf("%s: err = %v, want a *SendError at PhasePrepare", what, err)
+		}
+	}
+	prepare("a comma-separated To", s.SendAnnouncement("a@example.com, b@example.com", sampleAnnouncement()))
+	a := sampleAnnouncement()
+	a.UnsubscribeURL = ""
+	prepare("no unsubscribe URL", s.SendAnnouncement("a@example.com", a))
+}
+
+// Gmail and Yahoo require both headers of the RFC 8058 pair on bulk mail.
+func TestSendAnnouncementPutsListUnsubscribeOnTheWire(t *testing.T) {
+	port, got := fakeSMTPData(t, nil)
+	s := Sender{Host: "127.0.0.1", Port: port, From: "updates@example.com"}
+	a := sampleAnnouncement()
+	if err := s.SendAnnouncement("a@example.com", a); err != nil {
+		t.Fatal(err)
+	}
+	msg, err := mail.ReadMessage(bytes.NewReader(<-got))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if h := msg.Header.Get("List-Unsubscribe"); h != "<"+a.UnsubscribeURL+">" {
+		t.Errorf("List-Unsubscribe = %q", h)
+	}
+	if h := msg.Header.Get("List-Unsubscribe-Post"); h != "List-Unsubscribe=One-Click" {
+		t.Errorf("List-Unsubscribe-Post = %q", h)
+	}
+}

mail/render_test.go

@@ -20,18 +20,18 @@ func cases() map[string]any {
 		"invite":       s.inviteData("sam@example.com", "Northwind Ops", "8a1b2c3d4e"),
 		"license": licenseData("northwind-prod",
 			"eyJ2IjoxLCJpbnN0YW5jZSI6Im5vcnRod2luZC1wcm9kIiwidGllciI6ImZyZWUiLCJleHAiOjE3OTk5OTk5OTl9.MEUCIQDk3v1rX0sX7kQ2m9WQ1o0q8pWm6lq3vRjz8yN0yX4bGQIgYk2Fh7s"),
-		"instanceready":           instanceReadyData("northwind-prod", "https://northwind-prod.vantage.hostxtra.co.uk", now.Add(30*day)),
-		"instanceready/nologin":   instanceReadyData("northwind-prod", "", now.Add(30*day)),
-		"renewed":                 renewedData("northwind-prod", now.Add(30*day)),
-		"expiring":                expiringData("northwind-prod", s.PublicURL, now.Add(7*day)),
-		"expired":                 expiredData("northwind-prod", s.PublicURL, now.Add(14*day)),
-		"expired/noreaper":        expiredData("northwind-prod", s.PublicURL, time.Time{}),
-		"deletionwarning":         deletionWarningData("northwind-prod", s.PublicURL, now.Add(7*day), 7),
-		"deletionwarning/1day":    deletionWarningData("northwind-prod", s.PublicURL, now.Add(day), 1),
-		"cancelled":               s.billingData("northwind-prod"),
-		"pastdue":                 s.billingData("northwind-prod"),
-		"monitoralert":            MonitorEvent{MonitorName: "api.northwind.io", Type: "http", OldStatus: "up", NewStatus: "down", Message: "GET /healthz returned 503 Service Unavailable after 2.4s", Time: now, Down: true},
-		"monitoralert/recovered":  MonitorEvent{MonitorName: "api.northwind.io", Type: "http", OldStatus: "down", NewStatus: "up", Message: "200 OK in 182ms", Time: now},
+		"instanceready":          instanceReadyData("northwind-prod", "https://northwind-prod.vantage.hostxtra.co.uk", now.Add(30*day)),
+		"instanceready/nologin":  instanceReadyData("northwind-prod", "", now.Add(30*day)),
+		"renewed":                renewedData("northwind-prod", now.Add(30*day)),
+		"expiring":               expiringData("northwind-prod", s.PublicURL, now.Add(7*day)),
+		"expired":                expiredData("northwind-prod", s.PublicURL, now.Add(14*day)),
+		"expired/noreaper":       expiredData("northwind-prod", s.PublicURL, time.Time{}),
+		"deletionwarning":        deletionWarningData("northwind-prod", s.PublicURL, now.Add(7*day), 7),
+		"deletionwarning/1day":   deletionWarningData("northwind-prod", s.PublicURL, now.Add(day), 1),
+		"cancelled":              s.billingData("northwind-prod"),
+		"pastdue":                s.billingData("northwind-prod"),
+		"monitoralert":           MonitorEvent{MonitorName: "api.northwind.io", Type: "http", OldStatus: "up", NewStatus: "down", Message: "GET /healthz returned 503 Service Unavailable after 2.4s", Time: now, Down: true},
+		"monitoralert/recovered": MonitorEvent{MonitorName: "api.northwind.io", Type: "http", OldStatus: "down", NewStatus: "up", Message: "200 OK in 182ms", Time: now},
 		"contact": struct {
 			Enquiry
 			Received string
@@ -47,6 +47,10 @@ func cases() map[string]any {
 				{CVEID: "CVE-2025-48112", Severity: "low", PackageName: "less", ServerName: "worker-03"},
 			},
 			More: 9, DBAge: "2 days"},
+		"announcement": announcementData{
+			Announcement: sampleAnnouncement(),
+			BodyHTML:     "<p>Workflows can now run <strong>on a schedule</strong>.</p>",
+		},
 		"accountlocked":     sampleNotice(),
 		"disputereminder":   sampleNotice(),
 		"accountterminated": sampleNotice(),

mail/sender.go

@@ -20,6 +20,7 @@ import (
 	"net/smtp"
 	"net/textproto"
 	"os"
+	"sort"
 	"strings"
 	"time"
 )
@@ -29,6 +30,29 @@ import (
 // client gives up - and admin's signup rollback runs on that request's context.
 const timeout = 15 * time.Second
 
+// SMTP send phases. A caller that retries per recipient needs to know which
+// one failed: a recipient or data reply names one person's problem, while a
+// connect or quit failure says nothing about the message itself and a data
+// failure with no reply means the outcome of that one send is unknown.
+const (
+	PhasePrepare   = "prepare"
+	PhaseConnect   = "connect"
+	PhaseRecipient = "recipient"
+	PhaseData      = "data"
+	PhaseQuit      = "quit"
+)
+
+// SendError names which phase of the SMTP conversation failed. Err is kept as
+// the original wrapped error, so errors.As(err, &textprotoErr) still reaches
+// a *textproto.Error through Unwrap when the server sent one.
+type SendError struct {
+	Phase string
+	Err   error
+}
+
+func (e *SendError) Error() string { return "smtp " + e.Phase + ": " + e.Err.Error() }
+func (e *SendError) Unwrap() error { return e.Err }
+
 // Sender is a configured SMTP destination. It is a value, not a singleton:
 // server/internal/notify builds one per notification channel from data in
 // Mongo, while admin and sitesvc build one at boot.
@@ -76,6 +100,11 @@ type message struct {
 	Subject string
 	HTML    string
 	Text    string
+
+	// Headers are extra header lines for this one message, such as an
+	// announcement's List-Unsubscribe pair. Keys and values are CR/LF-stripped
+	// like every other header.
+	Headers map[string]string
 }
 
 // sendTemplate renders name against data and delivers the result.
@@ -98,17 +127,17 @@ func (s Sender) sendTemplate(to, replyTo, name string, data any) error {
 // every admin email from being delivered once already.
 func (s Sender) send(m message) error {
 	if !s.Enabled() {
-		return fmt.Errorf("smtp: not configured")
+		return &SendError{Phase: PhasePrepare, Err: fmt.Errorf("smtp: not configured")}
 	}
 	rcpts := recipients(m.To)
 	if len(rcpts) == 0 {
-		return fmt.Errorf("smtp: no recipient")
+		return &SendError{Phase: PhasePrepare, Err: fmt.Errorf("smtp: no recipient")}
 	}
 
 	addr := net.JoinHostPort(s.Host, s.Port)
 	conn, err := net.DialTimeout("tcp", addr, timeout)
 	if err != nil {
-		return fmt.Errorf("smtp: dial %s: %w", addr, err)
+		return &SendError{Phase: PhaseConnect, Err: fmt.Errorf("smtp: dial %s: %w", addr, err)}
 	}
 	_ = conn.SetDeadline(time.Now().Add(timeout))
 
@@ -119,49 +148,52 @@ func (s Sender) send(m message) error {
 	client, err := smtp.NewClient(conn, s.Host)
 	if err != nil {
 		conn.Close()
-		return fmt.Errorf("smtp: client: %w", err)
+		return &SendError{Phase: PhaseConnect, Err: fmt.Errorf("smtp: client: %w", err)}
 	}
 	defer client.Close()
 
 	if s.Port != "465" {
 		if ok, _ := client.Extension("STARTTLS"); ok {
 			if err := client.StartTLS(&tls.Config{ServerName: s.Host}); err != nil {
-				return fmt.Errorf("smtp: starttls: %w", err)
+				return &SendError{Phase: PhaseConnect, Err: fmt.Errorf("smtp: starttls: %w", err)}
 			}
 		}
 	}
 
 	if s.Username != "" {
 		if err := client.Auth(smtp.PlainAuth("", s.Username, s.Password, s.Host)); err != nil {
-			return fmt.Errorf("smtp: auth: %w", err)
+			return &SendError{Phase: PhaseConnect, Err: fmt.Errorf("smtp: auth: %w", err)}
 		}
 	}
 
 	if err := client.Mail(addrSpec(s.From)); err != nil {
-		return fmt.Errorf("smtp: mail from: %w", err)
+		return &SendError{Phase: PhaseConnect, Err: fmt.Errorf("smtp: mail from: %w", err)}
 	}
 	for _, rcpt := range rcpts {
 		if err := client.Rcpt(addrSpec(rcpt)); err != nil {
-			return fmt.Errorf("smtp: rcpt %s: %w", rcpt, err)
+			return &SendError{Phase: PhaseRecipient, Err: fmt.Errorf("smtp: rcpt %s: %w", rcpt, err)}
 		}
 	}
 
 	body, err := s.envelope(m)
 	if err != nil {
-		return fmt.Errorf("smtp: build message: %w", err)
+		return &SendError{Phase: PhasePrepare, Err: fmt.Errorf("smtp: build message: %w", err)}
 	}
 
 	w, err := client.Data()
 	if err != nil {
-		return fmt.Errorf("smtp: data: %w", err)
+		return &SendError{Phase: PhaseData, Err: fmt.Errorf("smtp: data: %w", err)}
 	}
 	if _, err := w.Write(body); err != nil {
-		return fmt.Errorf("smtp: write: %w", err)
+		return &SendError{Phase: PhaseData, Err: fmt.Errorf("smtp: write: %w", err)}
 	}
 	if err := w.Close(); err != nil {
-		return fmt.Errorf("smtp: close data: %w", err)
+		return &SendError{Phase: PhaseData, Err: fmt.Errorf("smtp: close data: %w", err)}
 	}
-	return client.Quit()
+	if err := client.Quit(); err != nil {
+		return &SendError{Phase: PhaseQuit, Err: fmt.Errorf("smtp: quit: %w", err)}
+	}
+	return nil
 }
 
 // addrSpec is the bare address for the SMTP envelope. SMTP_FROM is usually
@@ -226,6 +258,14 @@ func (s Sender) envelope(m message) ([]byte, error) {
 	b.WriteString("Message-ID: " + messageID(s.From) + "\r\n")
 	b.WriteString("Subject: " + mime.QEncoding.Encode("utf-8", sanitizeHeader(m.Subject)) + "\r\n")
 	b.WriteString("MIME-Version: 1.0\r\n")
+	keys := make([]string, 0, len(m.Headers))
+	for k := range m.Headers {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for _, k := range keys {
+		b.WriteString(sanitizeHeader(k) + ": " + sanitizeHeader(m.Headers[k]) + "\r\n")
+	}
 	b.WriteString("Content-Type: multipart/alternative; boundary=" + w.Boundary() + "\r\n")
 	b.WriteString("\r\n")
 	b.WriteString(parts.String())

mail/sender_test.go

@@ -1,16 +1,234 @@
 package mail
 
 import (
+	"bufio"
 	"bytes"
+	"errors"
 	"io"
 	"mime"
 	"mime/multipart"
 	"mime/quotedprintable"
+	"net"
 	"net/mail"
+	"net/textproto"
 	"strings"
 	"testing"
 )
 
+// fakeSMTP is a tiny scripted SMTP server. script maps an uppercased command
+// verb (or "DATA_BODY" for the reply after the body's terminating dot, or
+// "CLOSE_AFTER_BODY" to drop the connection with no reply at all) to the
+// line(s) to write back. Anything not listed gets a plain 250 OK.
+func fakeSMTP(t *testing.T, script map[string]string) string {
+	t.Helper()
+	port, _ := fakeSMTPData(t, script)
+	return port
+}
+
+// fakeSMTPData is fakeSMTP that also hands back the raw DATA the client sent,
+// once the body's terminating dot arrives.
+func fakeSMTPData(t *testing.T, script map[string]string) (string, <-chan []byte) {
+	t.Helper()
+	got := make(chan []byte, 1)
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { ln.Close() })
+
+	go func() {
+		conn, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		w := bufio.NewWriter(conn)
+		r := bufio.NewReader(conn)
+		writeLine := func(s string) {
+			w.WriteString(s + "\r\n")
+			w.Flush()
+		}
+		writeLine("220 fake.example ESMTP")
+		inData := false
+		var data bytes.Buffer
+		for {
+			line, err := r.ReadString('\n')
+			if err != nil {
+				return
+			}
+			raw := line
+			line = strings.TrimRight(line, "\r\n")
+			if inData {
+				if line != "." {
+					data.WriteString(raw)
+				}
+				if line == "." {
+					inData = false
+					got <- data.Bytes()
+					if reply, ok := script["DATA_BODY"]; ok {
+						if reply == "CLOSE" {
+							return
+						}
+						writeLine(reply)
+					} else {
+						writeLine("250 OK")
+					}
+				}
+				continue
+			}
+			verb := strings.ToUpper(strings.Fields(line)[0])
+			if verb == "EHLO" || verb == "HELO" {
+				writeLine("250 fake.example")
+				continue
+			}
+			if verb == "DATA" {
+				if reply, ok := script["DATA"]; ok {
+					if reply == "CLOSE" {
+						return
+					}
+					writeLine(reply)
+					continue
+				}
+				inData = true
+				writeLine("354 go ahead")
+				continue
+			}
+			if verb == "QUIT" {
+				if reply, ok := script["QUIT"]; ok {
+					if reply == "CLOSE" {
+						return
+					}
+					writeLine(reply)
+				} else {
+					writeLine("221 bye")
+				}
+				return
+			}
+			if reply, ok := script[verb]; ok {
+				if reply == "CLOSE" {
+					return
+				}
+				writeLine(reply)
+				continue
+			}
+			writeLine("250 OK")
+		}
+	}()
+
+	host, port, _ := net.SplitHostPort(ln.Addr().String())
+	_ = host
+	return port, got
+}
+
+func testMsg() message {
+	return message{To: "rcpt@example.com", Subject: "s", Text: "t", HTML: "<p>h</p>"}
+}
+
+func TestSendPhaseOnRecipientFailure(t *testing.T) {
+	port := fakeSMTP(t, map[string]string{"RCPT": "550 no such user"})
+	s := Sender{Host: "127.0.0.1", Port: port, From: "updates@example.com"}
+	err := s.send(testMsg())
+	var se *SendError
+	if !errors.As(err, &se) {
+		t.Fatalf("err = %v, want *SendError", err)
+	}
+	if se.Phase != PhaseRecipient {
+		t.Fatalf("phase = %q, want %q", se.Phase, PhaseRecipient)
+	}
+	var tp *textproto.Error
+	if !errors.As(err, &tp) || tp.Code != 550 {
+		t.Fatalf("textproto reply = %+v", tp)
+	}
+}
+
+func TestSendPhaseOnMailFromFailure(t *testing.T) {
+	// No AUTH configured, so a MAIL-stage 550 exercises the connect phase
+	// without needing to script a real AUTH challenge/response.
+	port := fakeSMTP(t, map[string]string{"MAIL": "550 relay denied"})
+	s := Sender{Host: "127.0.0.1", Port: port, From: "updates@example.com"}
+	err := s.send(testMsg())
+	var se *SendError
+	if !errors.As(err, &se) {
+		t.Fatalf("err = %v, want *SendError", err)
+	}
+	if se.Phase != PhaseConnect {
+		t.Fatalf("phase = %q, want %q", se.Phase, PhaseConnect)
+	}
+	var tp *textproto.Error
+	if !errors.As(err, &tp) || tp.Code != 550 {
+		t.Fatalf("textproto reply = %+v", tp)
+	}
+}
+
+func TestSendPhaseOnDataCommandFailure(t *testing.T) {
+	port := fakeSMTP(t, map[string]string{"DATA": "554 no thanks"})
+	s := Sender{Host: "127.0.0.1", Port: port, From: "updates@example.com"}
+	err := s.send(testMsg())
+	var se *SendError
+	if !errors.As(err, &se) {
+		t.Fatalf("err = %v, want *SendError", err)
+	}
+	if se.Phase != PhaseData {
+		t.Fatalf("phase = %q, want %q", se.Phase, PhaseData)
+	}
+	var tp *textproto.Error
+	if !errors.As(err, &tp) || tp.Code != 554 {
+		t.Fatalf("textproto reply = %+v", tp)
+	}
+}
+
+// The connection drops after the body's terminating dot but before any reply
+// is read: the data phase, but with no textproto error, since the server
+// never spoke back at all.
+func TestSendPhaseOnDataNoReply(t *testing.T) {
+	port := fakeSMTP(t, map[string]string{"DATA_BODY": "CLOSE"})
+	s := Sender{Host: "127.0.0.1", Port: port, From: "updates@example.com"}
+	err := s.send(testMsg())
+	var se *SendError
+	if !errors.As(err, &se) {
+		t.Fatalf("err = %v, want *SendError", err)
+	}
+	if se.Phase != PhaseData {
+		t.Fatalf("phase = %q, want %q", se.Phase, PhaseData)
+	}
+	var tp *textproto.Error
+	if errors.As(err, &tp) {
+		t.Fatalf("expected no textproto reply, got %+v", tp)
+	}
+}
+
+// A 250 on DATA_BODY (the message is accepted) followed by a dropped
+// connection on QUIT: the message was already delivered, so this must be the
+// quit phase, not data or connect.
+func TestSendPhaseOnQuitFailure(t *testing.T) {
+	port := fakeSMTP(t, map[string]string{"QUIT": "CLOSE"})
+	s := Sender{Host: "127.0.0.1", Port: port, From: "updates@example.com"}
+	err := s.send(testMsg())
+	var se *SendError
+	if !errors.As(err, &se) {
+		t.Fatalf("err = %v, want *SendError", err)
+	}
+	if se.Phase != PhaseQuit {
+		t.Fatalf("phase = %q, want %q", se.Phase, PhaseQuit)
+	}
+}
+
+func TestSendPrepareErrors(t *testing.T) {
+	s := Sender{}
+	err := s.send(testMsg())
+	var se *SendError
+	if !errors.As(err, &se) || se.Phase != PhasePrepare {
+		t.Fatalf("not configured: err = %v", err)
+	}
+
+	s2 := Sender{Host: "127.0.0.1", Port: "0", From: "updates@example.com"}
+	err2 := s2.send(message{Subject: "s", Text: "t", HTML: "<p>h</p>"})
+	var se2 *SendError
+	if !errors.As(err2, &se2) || se2.Phase != PhasePrepare {
+		t.Fatalf("no recipient: err = %v", err2)
+	}
+}
+
 // The envelope carries the bare address. "Vantage <x@y>" as MAIL FROM left
 // rspamd with no envelope sender, and mailcow skipped DKIM signing.
 func TestAddrSpecStripsDisplayName(t *testing.T) {
@@ -83,3 +301,30 @@ func TestEnvelopePartsAreQuotedPrintable(t *testing.T) {
 		}
 	}
 }
+
+func TestEnvelopeWritesExtraHeadersSanitised(t *testing.T) {
+	s := Sender{From: "Vantage <updates@example.com>"}
+	raw, err := s.envelope(message{
+		To: "a@example.com", Subject: "s", Text: "t", HTML: "<p>h</p>",
+		Headers: map[string]string{
+			"List-Unsubscribe-Post": "List-Unsubscribe=One-Click",
+			"List-Unsubscribe":      "<https://x.example/u?t=1>\r\nBcc: evil@example.com",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	msg, err := mail.ReadMessage(bytes.NewReader(raw))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := msg.Header.Get("List-Unsubscribe-Post"); got != "List-Unsubscribe=One-Click" {
+		t.Fatalf("List-Unsubscribe-Post = %q", got)
+	}
+	if got := msg.Header.Get("Bcc"); got != "" {
+		t.Fatalf("header injection: Bcc = %q", got)
+	}
+	if !strings.HasPrefix(msg.Header.Get("List-Unsubscribe"), "<https://x.example/u?t=1>") {
+		t.Fatalf("List-Unsubscribe = %q", msg.Header.Get("List-Unsubscribe"))
+	}
+}

mail/templates/announcement.html.tmpl

@@ -0,0 +1,9 @@
+{{define "tone"}}accent{{end}}
+{{define "category"}}{{.CategoryLabel}}{{end}}
+{{define "pill"}}{{template "chip" (dict "label" .CategoryLabel "tone" "accent")}}{{end}}
+{{define "title"}}{{.Subject}}{{end}}
+{{define "body"}}
+<div style="color:#e4ecf6;font-size:15px;line-height:1.6;">{{.BodyHTML}}</div>
+<p style="margin:24px 0 0;color:#71879f;font-size:12px;line-height:1.6;"><a href="{{.UnsubscribeURL}}" style="color:#9fb3ca;">Unsubscribe from {{.CategoryLabel}}</a> &middot; <a href="{{.PreferencesURL}}" style="color:#9fb3ca;">Email preferences</a>{{if .PostalAddress}}<br>{{.PostalAddress}}{{end}}</p>
+{{end}}
+{{define "why"}}you have a Vantage account and are subscribed to {{.CategoryLabel}}. Account emails such as sign-in, licences and billing are sent whatever you choose here.{{end}}

mail/templates/announcement.txt.tmpl

@@ -0,0 +1,13 @@
+{{define "subject"}}{{.Subject}}{{end}}
+{{define "tone"}}accent{{end}}
+{{define "category"}}{{.CategoryLabel}}{{end}}
+{{define "pill"}}{{template "chip" (dict "label" .CategoryLabel "tone" "accent")}}{{end}}
+{{define "title"}}{{.Subject}}{{end}}
+{{define "body"}}
+{{.BodyMarkdown}}
+
+Unsubscribe from {{.CategoryLabel}}: {{.UnsubscribeURL}}
+Email preferences: {{.PreferencesURL}}
+{{if .PostalAddress}}{{.PostalAddress}}{{end}}
+{{end}}
+{{define "why"}}you have a Vantage account and are subscribed to {{.CategoryLabel}}. Account emails such as sign-in, licences and billing are sent whatever you choose here.{{end}}

proto/vantage/v1/vantage.proto

@@ -109,6 +109,7 @@ message AgentMessage {
     StepResult step_result = 5;
     StepOutputChunk step_output = 6;
     WorkloadLogsResult workload_logs_result = 7;
+    PatchResult patch_result = 8;
   }
 }
 
@@ -126,6 +127,7 @@ message PackageUpdate {
   string name            = 1;
   string current_version = 2;
   string new_version     = 3;
+  bool   phased          = 4; // Ubuntu phased update this host is not yet selected for; deferred by apt
 }
 
 message ReportUpdatesRequest {
@@ -168,9 +170,9 @@ message InventoryReport {
   uint64 swap_used      = 7;
   repeated PartitionReport partitions = 8;
   string kernel         = 9;
-  // Set on static snapshots only. The agent never reboots; it reports that one
-  // is owed and leaves the decision to a person or a workflow.
+  // Set on static snapshots only. The agent reboots a host only when an ApplyUpdatesCmd asks it to and the OS reports a reboot is owed.
   bool   reboot_required = 10;
+  int64  boot_time_unix = 11;  // every report; proves a reboot happened
 }
 
 message InventoryReportResponse {
@@ -220,7 +222,19 @@ message ReportChecksResponse {
 }
 
 message ApplyUpdatesCmd {
+  string scope = 1;              // "" or "all" | "security"
+  bool   reboot_if_required = 2; // reboot only if the OS reports one is owed
+  int64  deadline_unix = 3;      // 0 = none; the agent caps the upgrade at 2h
+}
 
+message PatchResult {
+  string command_id = 1;
+  string status = 2;             // ok | failed | unsupported | busy
+  string message = 3;
+  string output_tail = 4;        // at most 64KB, newest bytes kept
+  int32  pending_after = 5;      // -1 when the post-apply check failed
+  bool   reboot_required = 6;
+  bool   rebooting = 7;          // sent just before the agent reboots itself
 }
 
 message ServerCommand {