9a44de69f1 · krz/gitbay

cmd/gitbayd/main.go +3

		@@ -213,6 +213,9 @@ func serveCmd() *cobra.Command {
213	213	if _, err := in.Password(); err != nil {
214	214	return err
215	215	}
	216	if in.TrustedAuthservID == "" {
	217	slog.Warn("mail reply: [mail.inbound] trusted_authserv_id is unset, so a reply's From is not checked against the mail host's DMARC and DKIM results; set it on any instance reachable from the internet")
	218	}
216	219	go (&mailin.Poller{P: &mailin.Processor{St: st, Cfg: cfg}, In: in}).Run(whCtx)
217	220	}
218	221	if cfg.Push.Enabled {

internal/config/config.go +8

		@@ -297,6 +297,11 @@ type MailInbound struct {
297	297	// from: reply@example.org becomes reply+<token>@example.org, so the
298	298	// mailbox must receive plus-addressed mail for it (or a catch-all).
299	299	ReplyAddress string `toml:"reply_address"`
	300	// TrustedAuthservID is the authserv-id the mail host writes in its
	301	// Authentication-Results header. When set, a reply must carry DMARC
	302	// pass, or an aligned DKIM pass, in the topmost such header. Only
	303	// safe when the mail host removes incoming headers claiming its id.
	304	TrustedAuthservID string `toml:"trusted_authserv_id"`
300	305	}
301	306
302	307	// DefaultInboundPoll is the poll interval when poll_interval is unset.
		@@ -378,6 +383,9 @@ func (m MailInbound) validate() []error {
378	383	errs = append(errs, fmt.Errorf("mail.inbound.poll_interval %q must be a duration of at least 10s", m.PollInterval))
379	384	}
380	385	}
	386	if id := m.TrustedAuthservID; id != "" && strings.ContainsAny(id, " \t;()\"\r\n") {
	387	errs = append(errs, fmt.Errorf("mail.inbound.trusted_authserv_id %q must be a bare host name such as mx.google.com", id))
	388	}
381	389	if a := m.ReplyAddress; a != "" {
382	390	local, domain, ok := strings.Cut(a, "@")
383	391	if !ok \|\| local == "" \|\| domain == "" \|\| strings.ContainsAny(a, "+ <>\"\r\n") \|\| strings.Contains(domain, "@") {

internal/config/config_test.go +1

		@@ -434,6 +434,7 @@ func TestMailInbound(t *testing.T) {
434	434	minimal + smtp + "[mail.inbound]\nenabled = true\n": "mail.inbound.password_file is required",
435	435	minimal + smtp + strings.Replace(inbound, "reply@gitbay.example", "reply+x@gitbay.example", 1): "no + in it",
436	436	minimal + smtp + strings.Replace(inbound, "reply@gitbay.example", "gitbay.example", 1): "bare address",
	437	minimal + smtp + inbound + "trusted_authserv_id = \"mx; x\"\n": "trusted_authserv_id",
437	438	minimal + smtp + inbound + "password = \"x\"\n": "unknown config key",
438	439	} {
439	440	if _, err := Load(writeConfig(t, body)); err == nil \|\| !strings.Contains(err.Error(), want) {

internal/control/adminmail.go +7

		@@ -17,6 +17,8 @@ func init() {
17	17	ReadOnly: true, Run: runAdminMailInboundCheck})
18	18	}
19	19
	20	const unauthenticatedWarning = "trusted_authserv_id is unset: a reply's From is not checked against the mail host's DMARC and DKIM results"
	21
20	22	// runAdminMailInboundCheck logs in to the [mail.inbound] mailbox and
21	23	// opens it with EXAMINE, which changes no flag, so a check never marks a
22	24	// reply seen before the poller reads it.
		@@ -34,6 +36,7 @@ func runAdminMailInboundCheck(c *Ctx, args []string) int {
34	36	Mailbox string `json:"mailbox,omitempty"`
35	37	Messages int `json:"messages"`
36	38	Unseen int `json:"unseen"`
	39	Warning string `json:"warning,omitempty"`
37	40	}
38	41	if !in.Enabled {
39	42	return c.emit(out{}, func(w io.Writer) {
		@@ -50,6 +53,10 @@ func runAdminMailInboundCheck(c *Ctx, args []string) int {
50	53	return c.fail(protocol.ExitFailure, "%s: %v", in.Addr(), err)
51	54	}
52	55	d := out{Enabled: true, Server: in.Addr(), Mailbox: in.MailboxName(), Messages: n, Unseen: len(unseen)}
	56	if in.TrustedAuthservID == "" {
	57	d.Warning = unauthenticatedWarning
	58	fmt.Fprintln(c.Stderr, "warning: "+d.Warning)
	59	}
53	60	return c.emit(d, func(w io.Writer) {
54	61	c.view(w).fields(
55	62	"server", d.Server,

internal/control/notifications_test.go +2 −2

		@@ -373,8 +373,8 @@ func TestNotifyReplyTo(t *testing.T) {
373	373	}
374	374	secrets, _ := ring.Derive(mailreply.Purpose)
375	375	target, err := mailreply.Verify(secrets, tok, time.Now())
376		want := mailreply.Target{UserID: bob, RepoID: repo.ID, Kind: tc.kind, Number: 1}
377		if err != nil \|\| target != want {
	376	want := mailreply.Target{UserID: bob, RepoID: repo.ID, Kind: tc.kind, Number: 1, Expires: target.Expires}
	377	if err != nil \|\| target != want \|\| time.Since(target.Issued()) > time.Minute {
378	378	t.Fatalf("token names %+v, %v; want %+v", target, err, want)
379	379	}
380	380	})

internal/imapc/imapc.go +91 −36

		@@ -18,14 +18,27 @@ import (
18	18	"time"
19	19	)
20	20
21		// MaxMessage is the largest message Fetch returns. A larger one is read
22		// and discarded, and Fetch returns ErrTooLarge.
	21	// MaxMessage is the largest message Fetch returns; a larger one is
	22	// refused by its RFC822.SIZE before its body is fetched (ErrTooLarge).
23	23	const MaxMessage = 10 << 20
24	24
25		// maxLine bounds one response line outside literals.
26		const maxLine = 1 << 20
	25	// Limits on what one command may make the client read. A server that
	26	// exceeds one has its connection closed and the command returns
	27	// ErrLimit.
	28	const (
	29	cmdBudget = MaxMessage + 1<<20 // bytes read for one command
	30	maxUntagged = 1000 // untagged responses to one command
	31	maxLine = 1 << 20 // one response line outside literals
	32	)
	33
	34	// MaxUnseen is the most UIDs Unseen returns; the rest wait for the next
	35	// poll.
	36	const MaxUnseen = 10000
27	37
28		var ErrTooLarge = errors.New("message larger than the fetch limit")
	38	var (
	39	ErrTooLarge = errors.New("message larger than the fetch limit")
	40	ErrLimit = errors.New("IMAP server exceeded a response limit; connection closed")
	41	)
29	42
30	43	// rootCAs verifies the server's certificate; nil is the system pool.
31	44	// Tests set it.
		@@ -36,6 +49,7 @@ type Client struct {
36	49	conn net.Conn
37	50	r *bufio.Reader
38	51	tag int
	52	left int64 // bytes the current command may still read
39	53	}
40	54
41	55	// Dial connects to addr (host:port) and reads the greeting. With
		@@ -88,7 +102,9 @@ func New(conn net.Conn) *Client {
88	102	func (c *Client) SetDeadline(t time.Time) error { return c.conn.SetDeadline(t) }
89	103
90	104	func (c *Client) greeting() error {
91		line, _, err := c.readResponse()
	105	c.left = cmdBudget
	106	resp, err := c.readResponse()
	107	line := resp.line
92	108	if err != nil {
93	109	return err
94	110	}
		@@ -157,15 +173,34 @@ func (c *Client) Unseen() ([]uint32, error) {
157	173	if err != nil {
158	174	return nil, fmt.Errorf("UID SEARCH: bad uid %q", clip(s))
159	175	}
160		uids = append(uids, uint32(n))
	176	if len(uids) < MaxUnseen {
	177	uids = append(uids, uint32(n))
	178	}
161	179	}
162	180	}
163	181	return uids, nil
164	182	}
165	183
166		// Fetch returns the whole message without setting \Seen.
	184	// Fetch returns the whole message without setting \Seen. A message
	185	// over MaxMessage is refused by its size first. A server answering
	186	// BODY[] with NIL or "" returns an empty message.
167	187	func (c *Client) Fetch(uid uint32) ([]byte, error) {
168		untagged, err := c.cmd(fmt.Sprintf("UID FETCH %d BODY.PEEK[]", uid))
	188	untagged, err := c.cmd(fmt.Sprintf("UID FETCH %d RFC822.SIZE", uid))
	189	if err != nil {
	190	return nil, fmt.Errorf("UID FETCH: %w", err)
	191	}
	192	for _, u := range untagged {
	193	up := strings.ToUpper(u.line)
	194	if i := strings.Index(up, "RFC822.SIZE "); i >= 0 && strings.Contains(up, " FETCH ") {
	195	f := strings.Fields(strings.TrimRight(up[i+len("RFC822.SIZE "):], ")"))
	196	if len(f) > 0 {
	197	if n, err := strconv.ParseInt(strings.TrimRight(f[0], ")"), 10, 64); err == nil && n > MaxMessage {
	198	return nil, ErrTooLarge
	199	}
	200	}
	201	}
	202	}
	203	untagged, err = c.cmd(fmt.Sprintf("UID FETCH %d BODY.PEEK[]", uid))
169	204	if err != nil {
170	205	return nil, fmt.Errorf("UID FETCH: %w", err)
171	206	}
		@@ -174,20 +209,16 @@ func (c *Client) Fetch(uid uint32) ([]byte, error) {
174	209	if len(f) < 3 \|\| !strings.EqualFold(f[2], "FETCH") {
175	210	continue
176	211	}
177		// The literal is the one after BODY[]; a server may send other
178		// items (FLAGS, UID) around it.
179		i := strings.Index(strings.ToUpper(u.line), "BODY[] {")
180		if i < 0 {
181		continue
	212	if u.tooLarge {
	213	return nil, ErrTooLarge
182	214	}
183		idx := strings.Count(u.line[:i], "{")
184		if idx >= len(u.literals) {
185		continue
	215	if u.body != nil {
	216	return u.body, nil
186	217	}
187		if u.literals[idx] == nil {
188		return nil, ErrTooLarge
	218	up := strings.ToUpper(u.line)
	219	if strings.Contains(up, "BODY[] NIL") \|\| strings.Contains(up, `BODY[] ""`) {
	220	return []byte{}, nil
189	221	}
190		return u.literals[idx], nil
191	222	}
192	223	return nil, fmt.Errorf("UID FETCH %d: no message body in the response", uid)
193	224	}
		@@ -207,24 +238,27 @@ func (c *Client) Close() error {
207	238	}
208	239
209	240	type response struct {
210		line string // the response with each literal's bytes left out
211		literals [][]byte // nil for a literal over MaxMessage
	241	line string // the response with each literal's bytes left out
	242	body []byte // the BODY[] literal, when the response carried one
	243	tooLarge bool // the BODY[] literal was over MaxMessage and not kept
212	244	}
213	245
214	246	// cmd sends one tagged command and collects the untagged responses up to
215	247	// its completion. A NO or BAD completion is an error.
216	248	func (c *Client) cmd(command string) ([]response, error) {
217	249	c.tag++
	250	c.left = cmdBudget
218	251	tag := "g" + strconv.Itoa(c.tag)
219	252	if _, err := io.WriteString(c.conn, tag+" "+command+"\r\n"); err != nil {
220	253	return nil, err
221	254	}
222	255	var untagged []response
223	256	for {
224		line, lits, err := c.readResponse()
	257	resp, err := c.readResponse()
225	258	if err != nil {
226	259	return nil, err
227	260	}
	261	line := resp.line
228	262	if rest, ok := strings.CutPrefix(line, tag+" "); ok {
229	263	status, _, _ := strings.Cut(rest, " ")
230	264	if strings.EqualFold(status, "OK") {
		@@ -236,41 +270,61 @@ func (c *Client) cmd(command string) ([]response, error) {
236	270	return nil, fmt.Errorf("server closed the session: %s", clip(line))
237	271	}
238	272	if strings.HasPrefix(line, "*") {
239		untagged = append(untagged, response{line, lits})
	273	if len(untagged) >= maxUntagged {
	274	return nil, c.limit()
	275	}
	276	untagged = append(untagged, resp)
240	277	}
241	278	// A "+" continuation is not expected: no command here sends a
242	279	// literal.
243	280	}
244	281	}
245	282
	283	// limit closes a connection whose server exceeded a limit.
	284	func (c *Client) limit() error {
	285	c.conn.Close()
	286	return ErrLimit
	287	}
	288
246	289	// readResponse reads one response: a line, and for each literal it
247	290	// announces ("{n}" at the end of a line) the n bytes and the rest of the
248		// response after them.
249		func (c *Client) readResponse() (string, [][]byte, error) {
	291	// response after them. Only the literal after "BODY[]" is kept; any
	292	// other is read and discarded. Everything read counts against the
	293	// command's budget.
	294	func (c *Client) readResponse() (response, error) {
250	295	var b strings.Builder
251		var lits [][]byte
	296	var resp response
252	297	for {
253	298	line, err := c.readLine()
254	299	if err != nil {
255		return "", nil, err
	300	return response{}, err
256	301	}
257	302	b.WriteString(line)
258	303	n, ok := literalSize(line)
259	304	if !ok {
260		return b.String(), lits, nil
	305	resp.line = b.String()
	306	return resp, nil
261	307	}
262		if n > MaxMessage {
	308	if n > c.left {
	309	return response{}, c.limit()
	310	}
	311	c.left -= n
	312	prefix := strings.TrimRight(line[:strings.LastIndexByte(line, '{')], " ")
	313	keep := resp.body == nil && !resp.tooLarge && strings.HasSuffix(strings.ToUpper(prefix), "BODY[]")
	314	if !keep \|\| n > MaxMessage {
263	315	if _, err := io.CopyN(io.Discard, c.r, n); err != nil {
264		return "", nil, err
	316	return response{}, err
	317	}
	318	if keep {
	319	resp.tooLarge = true
265	320	}
266		lits = append(lits, nil)
267	321	continue
268	322	}
269	323	buf := make([]byte, n)
270	324	if _, err := io.ReadFull(c.r, buf); err != nil {
271		return "", nil, err
	325	return response{}, err
272	326	}
273		lits = append(lits, buf)
	327	resp.body = buf
274	328	}
275	329	}
276	330
		@@ -282,8 +336,9 @@ func (c *Client) readLine() (string, error) {
282	336	return "", err
283	337	}
284	338	b = append(b, chunk...)
285		if len(b) > maxLine {
286		return "", errors.New("response line too long")
	339	c.left -= int64(len(chunk)) + 2
	340	if len(b) > maxLine \|\| c.left < 0 {
	341	return "", c.limit()
287	342	}
288	343	if !isPrefix {
289	344	return string(b), nil

internal/imapc/imapc_test.go +136

		@@ -4,6 +4,7 @@ import (
4	4	"bufio"
5	5	"crypto/tls"
6	6	"crypto/x509"
	7	"errors"
7	8	"fmt"
8	9	"net"
9	10	"net/http"
		@@ -19,6 +20,9 @@ type fakeServer struct {
19	20	msgs map[uint32]string
20	21	seen map[uint32]bool
21	22	cmds []string
	23	// raw, when set, answers a command in place of the default: it
	24	// writes whatever it likes and reports whether it handled it.
	25	raw func(conn net.Conn, tag, cmd string) bool
22	26	}
23	27
24	28	func (f *fakeServer) serve(conn net.Conn) {
		@@ -34,6 +38,9 @@ func (f *fakeServer) serve(conn net.Conn) {
34	38	tag, cmd, _ := strings.Cut(line, " ")
35	39	f.cmds = append(f.cmds, cmd)
36	40	up := strings.ToUpper(cmd)
	41	if f.raw != nil && f.raw(conn, tag, cmd) {
	42	continue
	43	}
37	44	switch {
38	45	case strings.HasPrefix(up, "STARTTLS"):
39	46	fmt.Fprintf(conn, "%s OK begin\r\n", tag)
		@@ -58,6 +65,10 @@ func (f *fakeServer) serve(conn net.Conn) {
58	65	}
59	66	}
60	67	fmt.Fprintf(conn, "* SEARCH %s\r\n%s OK done\r\n", strings.Join(ids, " "), tag)
	68	case strings.HasPrefix(up, "UID FETCH") && strings.HasSuffix(up, "RFC822.SIZE"):
	69	var uid uint32
	70	fmt.Sscanf(cmd, "UID FETCH %d", &uid)
	71	fmt.Fprintf(conn, "* 1 FETCH (UID %d RFC822.SIZE %d)\r\n%s OK done\r\n", uid, len(f.msgs[uid]), tag)
61	72	case strings.HasPrefix(up, "UID FETCH"):
62	73	var uid uint32
63	74	fmt.Sscanf(cmd, "UID FETCH %d", &uid)
		@@ -195,3 +206,128 @@ func TestLiteralSize(t *testing.T) {
195	206	}
196	207	}
197	208	}
	209
	210	// session dials f over implicit TLS and logs in.
	211	func session(t testing.T, f fakeServer) *Client {
	212	t.Helper()
	213	setupTLS(t)
	214	if f.msgs == nil {
	215	f.msgs, f.seen = map[uint32]string{}, map[uint32]bool{}
	216	}
	217	c, err := Dial(listen(t, f, true), false, 10*time.Second)
	218	if err != nil {
	219	t.Fatal(err)
	220	}
	221	t.Cleanup(func() { c.Close() })
	222	if err := c.Login("u", `p"w`); err != nil {
	223	t.Fatal(err)
	224	}
	225	return c
	226	}
	227
	228	// A server that answers one FETCH with literal after literal is cut off
	229	// at the command's byte budget, however it labels them.
	230	func TestHostileRepeatedLiterals(t *testing.T) {
	231	chunk := strings.Repeat("x", 10<<20)
	232	f := &fakeServer{raw: func(conn net.Conn, tag, cmd string) bool {
	233	if !strings.HasPrefix(cmd, "UID FETCH 1 BODY") {
	234	return false
	235	}
	236	for i := 0; i < 5; i++ {
	237	if _, err := fmt.Fprintf(conn, "* 1 FETCH (FLAGS {%d}\r\n%s)\r\n", len(chunk), chunk); err != nil {
	238	return true
	239	}
	240	}
	241	fmt.Fprintf(conn, "%s OK done\r\n", tag)
	242	return true
	243	}}
	244	c := session(t, f)
	245	f.msgs[1] = "small"
	246	if _, err := c.Fetch(1); !errors.Is(err, ErrLimit) {
	247	t.Fatalf("Fetch = %v, want ErrLimit", err)
	248	}
	249	}
	250
	251	// A body literal over MaxMessage is refused even when RFC822.SIZE lied.
	252	func TestLyingSize(t *testing.T) {
	253	big := strings.Repeat("x", MaxMessage+10)
	254	f := &fakeServer{raw: func(conn net.Conn, tag, cmd string) bool {
	255	if !strings.HasPrefix(cmd, "UID FETCH 1 BODY") {
	256	return false
	257	}
	258	fmt.Fprintf(conn, "* 1 FETCH (BODY[] {%d}\r\n%s)\r\n%s OK done\r\n", len(big), big, tag)
	259	return true
	260	}}
	261	c := session(t, f)
	262	f.msgs[1] = "small"
	263	if _, err := c.Fetch(1); !errors.Is(err, ErrTooLarge) {
	264	t.Fatalf("Fetch = %v, want ErrTooLarge", err)
	265	}
	266	}
	267
	268	func TestSizeRefusedBeforeFetch(t *testing.T) {
	269	f := &fakeServer{}
	270	c := session(t, f)
	271	f.msgs[1] = strings.Repeat("x", MaxMessage+1)
	272	if _, err := c.Fetch(1); !errors.Is(err, ErrTooLarge) {
	273	t.Fatalf("Fetch = %v, want ErrTooLarge", err)
	274	}
	275	for _, cmd := range f.cmds {
	276	if strings.Contains(cmd, "BODY.PEEK") {
	277	t.Fatal("fetched the body of an oversized message")
	278	}
	279	}
	280	}
	281
	282	func TestHugeSearch(t *testing.T) {
	283	var b strings.Builder
	284	for i := 1; i <= 20000; i++ {
	285	fmt.Fprintf(&b, " %d", i)
	286	}
	287	f := &fakeServer{raw: func(conn net.Conn, tag, cmd string) bool {
	288	if cmd != "UID SEARCH UNSEEN" {
	289	return false
	290	}
	291	fmt.Fprintf(conn, "* SEARCH%s\r\n%s OK done\r\n", b.String(), tag)
	292	return true
	293	}}
	294	c := session(t, f)
	295	uids, err := c.Unseen()
	296	if err != nil \|\| len(uids) != MaxUnseen {
	297	t.Fatalf("Unseen = %d uids, %v", len(uids), err)
	298	}
	299	}
	300
	301	func TestTooManyUntagged(t *testing.T) {
	302	f := &fakeServer{raw: func(conn net.Conn, tag, cmd string) bool {
	303	if cmd != "UID SEARCH UNSEEN" {
	304	return false
	305	}
	306	for i := 0; i < maxUntagged+10; i++ {
	307	fmt.Fprint(conn, "* OK noise\r\n")
	308	}
	309	fmt.Fprintf(conn, "%s OK done\r\n", tag)
	310	return true
	311	}}
	312	c := session(t, f)
	313	if _, err := c.Unseen(); !errors.Is(err, ErrLimit) {
	314	t.Fatalf("Unseen = %v, want ErrLimit", err)
	315	}
	316	}
	317
	318	func TestEmptyBody(t *testing.T) {
	319	for _, form := range []string{"NIL", `""`} {
	320	f := &fakeServer{raw: func(conn net.Conn, tag, cmd string) bool {
	321	if !strings.HasPrefix(cmd, "UID FETCH 1 BODY") {
	322	return false
	323	}
	324	fmt.Fprintf(conn, "* 1 FETCH (UID 1 BODY[] %s)\r\n%s OK done\r\n", form, tag)
	325	return true
	326	}}
	327	c := session(t, f)
	328	f.msgs[1] = ""
	329	if b, err := c.Fetch(1); err != nil \|\| len(b) != 0 {
	330	t.Fatalf("%s: Fetch = %q, %v", form, b, err)
	331	}
	332	}
	333	}

internal/mailin/authres.go added +105

		@@ -0,0 +1,105 @@
	1	package mailin
	2
	3	import (
	4	"net/mail"
	5	"strings"
	6	)
	7
	8	// authenticated checks the sender against the mail host's own verdict:
	9	// the topmost Authentication-Results header (RFC 8601) whose authserv-id
	10	// is authserv. The mail host adds its header above any the message
	11	// arrived with, so a lower header claiming the same id is the sender's
	12	// and is not read. It returns "" when the header shows dmarc=pass for
	13	// the From domain, or dkim=pass with a signing domain aligned with it,
	14	// and the refusal's reason otherwise.
	15	func authenticated(h mail.Header, authserv, from string) string {
	16	_, fromDomain, ok := strings.Cut(strings.ToLower(from), "@")
	17	if !ok \|\| fromDomain == "" {
	18	return "no From domain"
	19	}
	20	for _, v := range h["Authentication-Results"] {
	21	parts := strings.Split(stripComments(v), ";")
	22	f := strings.Fields(parts[0])
	23	if len(f) == 0 \|\| !strings.EqualFold(f[0], authserv) {
	24	continue
	25	}
	26	for _, r := range parts[1:] {
	27	method, result, props := resinfo(r)
	28	if result != "pass" {
	29	continue
	30	}
	31	switch method {
	32	case "dmarc":
	33	if strings.EqualFold(props["header.from"], fromDomain) {
	34	return ""
	35	}
	36	case "dkim":
	37	d := props["header.d"]
	38	if d == "" {
	39	_, d, _ = strings.Cut(props["header.i"], "@")
	40	}
	41	if aligned(strings.ToLower(d), fromDomain) {
	42	return ""
	43	}
	44	}
	45	}
	46	return "sender not authenticated by " + authserv + " (no DMARC pass or aligned DKIM pass)"
	47	}
	48	return "no Authentication-Results from " + authserv
	49	}
	50
	51	// resinfo splits "method[/version]=result prop=value ..." into its
	52	// method, result and properties, all lower case.
	53	func resinfo(s string) (string, string, map[string]string) {
	54	props := map[string]string{}
	55	f := strings.Fields(s)
	56	if len(f) == 0 {
	57	return "", "", props
	58	}
	59	method, result, _ := strings.Cut(strings.ToLower(f[0]), "=")
	60	method, _, _ = strings.Cut(method, "/")
	61	for _, kv := range f[1:] {
	62	k, v, ok := strings.Cut(kv, "=")
	63	if ok {
	64	props[strings.ToLower(k)] = strings.ToLower(strings.Trim(v, `"`))
	65	}
	66	}
	67	return method, result, props
	68	}
	69
	70	// aligned reports relaxed alignment, kept simple: the signing domain is
	71	// the From domain, or one is a subdomain of the other.
	72	func aligned(d, from string) bool {
	73	if !strings.Contains(d, ".") {
	74	return false
	75	}
	76	return d == from \|\| strings.HasSuffix(from, "."+d) \|\| strings.HasSuffix(d, "."+from)
	77	}
	78
	79	// stripComments removes RFC 5322 comments, "(...)", nested or not,
	80	// outside quoted strings.
	81	func stripComments(s string) string {
	82	var b strings.Builder
	83	depth, quoted := 0, false
	84	for i := 0; i < len(s); i++ {
	85	c := s[i]
	86	switch {
	87	case c == '\\' && i+1 < len(s):
	88	if depth == 0 {
	89	b.WriteByte(c)
	90	b.WriteByte(s[i+1])
	91	}
	92	i++
	93	case c == '"' && depth == 0:
	94	quoted = !quoted
	95	b.WriteByte(c)
	96	case c == '(' && !quoted:
	97	depth++
	98	case c == ')' && !quoted && depth > 0:
	99	depth--
	100	case depth == 0:
	101	b.WriteByte(c)
	102	}
	103	}
	104	return b.String()
	105	}

internal/mailin/mailin.go +53 −4

		@@ -81,14 +81,29 @@ func (p *Processor) Drain(mb Mailbox) error {
81	81	return err
82	82	}
83	83	for _, uid := range uids {
	84	// A message that failed in earlier polls before it could be
	85	// handled (a fetch the server cut off) is given up on unread.
	86	if p.tries[uid] >= maxTries {
	87	p.audit(0, "", "gave up after "+strconv.Itoa(maxTries)+" tries")
	88	delete(p.tries, uid)
	89	if err := mb.MarkSeen(uid); err != nil {
	90	return err
	91	}
	92	continue
	93	}
84	94	raw, err := mb.Fetch(uid)
85	95	var res Result
86	96	switch {
87	97	case errors.Is(err, imapc.ErrTooLarge):
88	98	res = refused("message larger than %d bytes", imapc.MaxMessage)
89	99	p.audit(0, "", res.Reason)
90		case err != nil:
	100	case errors.Is(err, imapc.ErrLimit):
	101	// The connection is closed; the message counts a try and
	102	// the poll ends.
	103	p.tries[uid]++
91	104	return err
	105	case err != nil:
	106	res = Result{Retry: true, Reason: "fetch: " + err.Error()}
92	107	default:
93	108	res = p.Handle(raw)
94	109	}
		@@ -111,6 +126,9 @@ func (p *Processor) Drain(mb Mailbox) error {
111	126	// Handle checks one message and posts it when every check passes.
112	127	// Refusals are audited here.
113	128	func (p *Processor) Handle(raw []byte) Result {
	129	if len(bytes.TrimSpace(raw)) == 0 {
	130	return p.refuse(0, "", "empty message")
	131	}
114	132	msg, err := mail.ReadMessage(bytes.NewReader(raw))
115	133	if err != nil {
116	134	return p.refuse(0, "", "unreadable message")
		@@ -154,6 +172,11 @@ func (p *Processor) Handle(raw []byte) Result {
154	172	case u.Pending:
155	173	return p.refuse(u.ID, msgID, "account not active")
156	174	}
	175	// Ids are reused after a hard delete: an account created after the
	176	// token was minted is not the one it named.
	177	if code := p.createdAfter("users", u.ID, target, msgID, "account"); code != nil {
	178	return *code
	179	}
157	180	// The token alone is not enough: the reply must come from one of
158	181	// the account's verified addresses.
159	182	from, err := msg.Header.AddressList("From")
		@@ -167,6 +190,11 @@ func (p *Processor) Handle(raw []byte) Result {
167	190	if !ok {
168	191	return p.refuse(u.ID, msgID, "From is not a verified address of the account")
169	192	}
	193	if id := p.Cfg.Mail.Inbound.TrustedAuthservID; id != "" {
	194	if reason := authenticated(msg.Header, id, from[0].Address); reason != "" {
	195	return p.refuse(u.ID, msgID, reason)
	196	}
	197	}
170	198	if on, err := p.St.ReplyEnabled(u.ID); err != nil {
171	199	return Result{Retry: true, Reason: err.Error()}
172	200	} else if !on {
		@@ -197,12 +225,18 @@ func (p *Processor) Handle(raw []byte) Result {
197	225	case err != nil:
198	226	return Result{Retry: true, Reason: err.Error()}
199	227	}
	228	if code := p.createdAfter("repos", repo.ID, target, msgID, "repository"); code != nil {
	229	return *code
	230	}
200	231
201		key := msgID
202		if key == "" {
	232	// The claim names the thread and the account as well as the
	233	// message, so one account's Message-ID cannot suppress another's.
	234	id := msgID
	235	if id == "" {
203	236	sum := sha256.Sum256(raw)
204		key = "sha256:" + hex.EncodeToString(sum[:])
	237	id = "sha256:" + hex.EncodeToString(sum[:])
205	238	}
	239	key := fmt.Sprintf("%d/%s/%d/%d/%s", u.ID, target.Kind, target.RepoID, target.Number, id)
206	240	claimed, err := p.St.ClaimMailReply(key)
207	241	if err != nil {
208	242	return Result{Retry: true, Reason: err.Error()}
		@@ -230,6 +264,21 @@ func (p *Processor) Handle(raw []byte) Result {
230	264	return p.refuse(u.ID, msgID, "comment refused: "+reason)
231	265	}
232	266
	267	// createdAfter refuses when the row was created after the token was
	268	// minted: a later account or repository that took a freed id. Created
	269	// times are compared to the second, the token's precision.
	270	func (p Processor) createdAfter(table string, id int64, target mailreply.Target, msgID, what string) Result {
	271	created, err := p.St.CreatedAt(table, id)
	272	if err != nil {
	273	return &Result{Retry: true, Reason: err.Error()}
	274	}
	275	if created.Truncate(time.Second).After(target.Issued()) {
	276	r := p.refuse(0, msgID, what+" created after the reply token was issued")
	277	return &r
	278	}
	279	return nil
	280	}
	281
233	282	func (p *Processor) now() time.Time {
234	283	if p.Now != nil {
235	284	return p.Now()

internal/mailin/mailin_test.go +169 −4

		@@ -1,6 +1,7 @@
1	1	package mailin
2	2
3	3	import (
	4	"errors"
4	5	"fmt"
5	6	"strings"
6	7	"testing"
		@@ -22,6 +23,7 @@ type fixture struct {
22	23	issueID int64
23	24	bob int64
24	25	secrets [][]byte
	26	issued time.Time // when the fixture's tokens are minted
25	27	}
26	28
27	29	// setup is alice's public repository alice/app with issue #1, and bob,
		@@ -84,13 +86,13 @@ func setup(t testing.T) fixture {
84	86	t.Fatal(err)
85	87	}
86	88	return &fixture{p: &Processor{St: st, Cfg: cfg}, st: st, repo: repo,
87		issueID: issueID, bob: bob, secrets: secrets}
	89	issueID: issueID, bob: bob, secrets: secrets, issued: time.Now()}
88	90	}
89	91
90	92	func (f fixture) token(t testing.T, user int64) string {
91	93	t.Helper()
92	94	tok, err := mailreply.Mint(f.secrets, mailreply.Target{UserID: user, RepoID: f.repo.ID, Kind: "issue", Number: 1},
93		time.Now().Add(mailreply.Lifetime))
	95	f.issued.Add(mailreply.Lifetime))
94	96	if err != nil {
95	97	t.Fatal(err)
96	98	}
		@@ -102,8 +104,15 @@ var msgSeq int
102	104	func (f fixture) message(t testing.T, from, body string) string {
103	105	t.Helper()
104	106	msgSeq++
105		return fmt.Sprintf("From: Bob <%s>\r\nTo: gitbay <%s>\r\nSubject: Re: [alice/app] #1: title\r\nMessage-ID: <m%d@example.test>\r\n"+
106		"Content-Type: text/plain; charset=utf-8\r\n\r\n%s\r\n", from, mailreply.Address(replyBase, f.token(t, f.bob)), msgSeq, body)
	107	return f.messageAs(t, f.bob, from, fmt.Sprintf("<m%d@example.test>", msgSeq), "", body)
	108	}
	109
	110	// messageAs is a reply from user's token with the given Message-ID and
	111	// extra header lines (each ending in CRLF).
	112	func (f fixture) messageAs(t testing.T, user int64, from, msgID, headers, body string) string {
	113	t.Helper()
	114	return fmt.Sprintf("%sFrom: Someone <%s>\r\nTo: gitbay <%s>\r\nSubject: Re: [alice/app] #1: title\r\nMessage-ID: %s\r\n"+
	115	"Content-Type: text/plain; charset=utf-8\r\n\r\n%s\r\n", headers, from, mailreply.Address(replyBase, f.token(t, user)), msgID, body)
107	116	}
108	117
109	118	func (f fixture) comments(t testing.T) []store.IssueComment {
		@@ -253,6 +262,7 @@ func TestRefusalLeavesNoClaim(t *testing.T) {
253	262	type fakeMailbox struct {
254	263	msgs map[uint32][]byte
255	264	seen map[uint32]bool
	265	errs map[uint32]error
256	266	}
257	267
258	268	func (m *fakeMailbox) Unseen() ([]uint32, error) {
		@@ -266,6 +276,9 @@ func (m *fakeMailbox) Unseen() ([]uint32, error) {
266	276	}
267	277
268	278	func (m *fakeMailbox) Fetch(uid uint32) ([]byte, error) {
	279	if err := m.errs[uid]; err != nil {
	280	return nil, err
	281	}
269	282	if m.msgs[uid] == nil {
270	283	return nil, imapc.ErrTooLarge
271	284	}
		@@ -319,3 +332,155 @@ func TestDrainRetriesTransientFailure(t *testing.T) {
319	332	t.Fatal("not given up on")
320	333	}
321	334	}
	335
	336	// A repository id freed by a delete and taken by a later repository does
	337	// not accept replies meant for the old one.
	338	func TestReusedRepositoryID(t *testing.T) {
	339	f := setup(t)
	340	m := []byte(f.message(t, "bob@example.test", "hi"))
	341	if err := f.st.DeleteRepo(f.repo.ID); err != nil {
	342	t.Fatal(err)
	343	}
	344	alice, _ := f.st.UserByUsername("alice")
	345	id, err := f.st.CreateRepo("user", alice.ID, "other", "public")
	346	if err != nil \|\| id != f.repo.ID {
	347	t.Fatalf("new repository has id %d (%v), want the freed %d", id, err, f.repo.ID)
	348	}
	349	f.st.CreateIssue(id, alice.ID, "t", "", "md")
	350	// Created after the token, as it would be outside a fast test.
	351	f.st.DB.Exec("UPDATE repos SET created_at = ? WHERE id = ?",
	352	time.Now().Add(5*time.Second).UTC().Format("2006-01-02T15:04:05.000Z"), id)
	353	res := f.p.Handle(m)
	354	if res.Posted \|\| !strings.Contains(res.Reason, "repository created after the reply token") {
	355	t.Fatalf("result %+v", res)
	356	}
	357	if !strings.Contains(f.refusalReasons(t), "repository created after") {
	358	t.Fatal("refusal not audited")
	359	}
	360	}
	361
	362	func TestReusedUserID(t *testing.T) {
	363	f := setup(t)
	364	f.st.DB.Exec("UPDATE users SET created_at = ? WHERE id = ?",
	365	time.Now().Add(5*time.Second).UTC().Format("2006-01-02T15:04:05.000Z"), f.bob)
	366	res := f.p.Handle([]byte(f.message(t, "bob@example.test", "hi")))
	367	if res.Posted \|\| !strings.Contains(res.Reason, "account created after the reply token") {
	368	t.Fatalf("result %+v", res)
	369	}
	370	}
	371
	372	// One account's Message-ID does not suppress another account's reply.
	373	func TestDedupePerAccount(t *testing.T) {
	374	f := setup(t)
	375	carol, err := f.st.CreateUser("carol", false)
	376	if err != nil {
	377	t.Fatal(err)
	378	}
	379	f.st.AddEmail(carol, "carol@example.test", "admin", true)
	380	f.st.SetReplyEnabled(carol, true)
	381	if res := f.p.Handle([]byte(f.messageAs(t, f.bob, "bob@example.test", "<same@x>", "", "from bob"))); !res.Posted {
	382	t.Fatalf("bob: %+v", res)
	383	}
	384	if res := f.p.Handle([]byte(f.messageAs(t, carol, "carol@example.test", "<same@x>", "", "from carol"))); !res.Posted {
	385	t.Fatalf("carol: %+v", res)
	386	}
	387	if n := len(f.comments(t)); n != 2 {
	388	t.Fatalf("%d comments", n)
	389	}
	390	}
	391
	392	func TestEmptyMessage(t *testing.T) {
	393	f := setup(t)
	394	if res := f.p.Handle([]byte{}); res.Posted \|\| res.Reason != "empty message" {
	395	t.Fatalf("result %+v", res)
	396	}
	397	}
	398
	399	// A fetch that keeps failing counts tries for that message alone; the
	400	// rest of the mailbox is handled, and after maxTries the failing one is
	401	// marked seen and audited.
	402	func TestDrainFetchErrors(t *testing.T) {
	403	f := setup(t)
	404	mb := &fakeMailbox{seen: map[uint32]bool{},
	405	msgs: map[uint32][]byte{1: []byte("x"), 2: []byte(f.message(t, "bob@example.test", "hi"))},
	406	errs: map[uint32]error{1: errors.New("NO [UNAVAILABLE] try later")}}
	407	for i := 1; i < maxTries; i++ {
	408	if err := f.p.Drain(mb); err != nil {
	409	t.Fatal(err)
	410	}
	411	if mb.seen[1] {
	412	t.Fatalf("marked seen after %d tries", i)
	413	}
	414	if !mb.seen[2] {
	415	t.Fatal("the next message was not handled")
	416	}
	417	}
	418	f.p.Drain(mb)
	419	if !mb.seen[1] \|\| !strings.Contains(f.refusalReasons(t), "gave up after") {
	420	t.Fatal("not given up on and audited")
	421	}
	422	}
	423
	424	// A fetch the server cut off ends the poll; the message is given up on
	425	// unread once it has cost maxTries polls.
	426	func TestDrainLimitEndsPoll(t *testing.T) {
	427	f := setup(t)
	428	mb := &fakeMailbox{seen: map[uint32]bool{}, msgs: map[uint32][]byte{1: []byte("x")},
	429	errs: map[uint32]error{1: imapc.ErrLimit}}
	430	for i := 0; i < maxTries; i++ {
	431	if err := f.p.Drain(mb); !errors.Is(err, imapc.ErrLimit) {
	432	t.Fatalf("poll %d: %v", i, err)
	433	}
	434	}
	435	if err := f.p.Drain(mb); err != nil \|\| !mb.seen[1] {
	436	t.Fatalf("not given up on: %v", err)
	437	}
	438	}
	439
	440	func TestAuthenticationResults(t *testing.T) {
	441	const id = "mx.example.net"
	442	for _, tc := range []struct {
	443	name, headers, from, reason string
	444	}{
	445	{"dmarc pass",
	446	"Authentication-Results: mx.example.net; spf=pass smtp.mailfrom=example.test; dmarc=pass (p=REJECT) header.from=example.test\r\n",
	447	"bob@example.test", ""},
	448	{"aligned dkim pass, gmail header.i",
	449	"Authentication-Results: mx.example.net;\r\n dkim=pass header.i=@mail.example.test header.s=s1 header.b=abc\r\n",
	450	"bob@example.test", ""},
	451	{"dmarc fail",
	452	"Authentication-Results: mx.example.net; dkim=fail header.d=example.test; dmarc=fail header.from=example.test\r\n",
	453	"bob@example.test", "sender not authenticated"},
	454	{"missing header", "", "bob@example.test", "no Authentication-Results from mx.example.net"},
	455	{"spoofed lower header with the same id",
	456	"Authentication-Results: mx.example.net; dmarc=fail header.from=example.test\r\nAuthentication-Results: mx.example.net; dmarc=pass header.from=example.test\r\n",
	457	"bob@example.test", "sender not authenticated"},
	458	{"other authserv only",
	459	"Authentication-Results: evil.example; dmarc=pass header.from=example.test\r\n",
	460	"bob@example.test", "no Authentication-Results from mx.example.net"},
	461	{"misaligned dkim domain",
	462	"Authentication-Results: mx.example.net; dkim=pass header.d=attacker.example; dmarc=none header.from=example.test\r\n",
	463	"bob@example.test", "sender not authenticated"},
	464	{"dmarc pass for another domain",
	465	"Authentication-Results: mx.example.net; dmarc=pass header.from=attacker.example\r\n",
	466	"bob@example.test", "sender not authenticated"},
	467	} {
	468	t.Run(tc.name, func(t *testing.T) {
	469	f := setup(t)
	470	f.p.Cfg.Mail.Inbound.TrustedAuthservID = id
	471	res := f.p.Handle([]byte(f.messageAs(t, f.bob, tc.from, "<a@x>", tc.headers, "hi")))
	472	if tc.reason == "" {
	473	if !res.Posted {
	474	t.Fatalf("not posted: %+v", res)
	475	}
	476	return
	477	}
	478	if res.Posted \|\| !strings.Contains(res.Reason, tc.reason) {
	479	t.Fatalf("result %+v, want %q", res, tc.reason)
	480	}
	481	if !strings.Contains(f.refusalReasons(t), tc.reason) {
	482	t.Fatal("refusal not audited")
	483	}
	484	})
	485	}
	486	}

internal/mailreply/mailreply.go +11 −3

		@@ -33,8 +33,15 @@ type Target struct {
33	33	RepoID int64
34	34	Kind string // "issue" or "mr"
35	35	Number int64
	36	// Expires is set by Verify; Mint takes the expiry separately.
	37	Expires time.Time
36	38	}
37	39
	40	// Issued is when the token was minted: every token is minted to expire
	41	// Lifetime later. An account or repository created after it is not the
	42	// one the token named, but a later row that took a freed id.
	43	func (t Target) Issued() time.Time { return t.Expires.Add(-Lifetime) }
	44
38	45	var (
39	46	ErrMalformed = errors.New("malformed reply token")
40	47	ErrBadMAC = errors.New("reply token does not verify")
		@@ -46,7 +53,7 @@ var (
46	53	var enc = base32.StdEncoding.WithPadding(base32.NoPadding)
47	54
48	55	// Mint returns the token for t, valid until expires, authenticated under
49		// keys[0].
	56	// keys[0]. expires is the mint time plus Lifetime (Target.Issued).
50	57	func Mint(keys [][]byte, t Target, expires time.Time) (string, error) {
51	58	if len(keys) == 0 {
52	59	return "", errors.New("no key to mint a reply token under")
		@@ -67,7 +74,7 @@ func Mint(keys [][]byte, t Target, expires time.Time) (string, error) {
67	74	p = binary.AppendUvarint(p, uint64(t.UserID))
68	75	p = binary.AppendUvarint(p, uint64(t.RepoID))
69	76	p = binary.AppendUvarint(p, uint64(t.Number))
70		p = binary.AppendUvarint(p, uint64(expires.Unix()/3600))
	77	p = binary.AppendUvarint(p, uint64(expires.Unix()))
71	78	p = append(p, mac(keys[0], p)...)
72	79	return strings.ToLower(enc.EncodeToString(p)), nil
73	80	}
		@@ -118,7 +125,8 @@ func Verify(keys [][]byte, token string, now time.Time) (Target, error) {
118	125	return Target{}, ErrMalformed
119	126	}
120	127	t.UserID, t.RepoID, t.Number = int64(v[0]), int64(v[1]), int64(v[2])
121		if !now.Before(time.Unix(int64(v[3])*3600, 0)) {
	128	t.Expires = time.Unix(int64(v[3]), 0).UTC()
	129	if !now.Before(t.Expires) {
122	130	return t, ErrExpired
123	131	}
124	132	return t, nil

internal/mailreply/mailreply_test.go +5

		@@ -36,9 +36,13 @@ func TestRoundTrip(t *testing.T) {
36	36	t.Errorf("local part is %d octets", l)
37	37	}
38	38	got, err := Verify([][]byte{keyA}, tok, now)
	39	tg.Expires = now.Add(Lifetime)
39	40	if err != nil \|\| got != tg {
40	41	t.Errorf("Verify = %+v, %v; want %+v", got, err, tg)
41	42	}
	43	if !got.Issued().Equal(now) {
	44	t.Errorf("Issued = %v, want %v", got.Issued(), now)
	45	}
42	46	// A mail system that upper-cases the local part does not break it.
43	47	if got, err := Verify([][]byte{keyA}, strings.ToUpper(tok), now); err != nil \|\| got != tg {
44	48	t.Errorf("upper-case Verify = %+v, %v", got, err)
		@@ -84,6 +88,7 @@ func TestExpiry(t *testing.T) {
84	88	t.Fatalf("before expiry: %v", err)
85	89	}
86	90	got, err := Verify([][]byte{keyA}, tok, now.Add(Lifetime+time.Hour))
	91	tg.Expires = now.Add(Lifetime)
87	92	if !errors.Is(err, ErrExpired) \|\| got != tg {
88	93	t.Fatalf("after expiry: %+v, %v", got, err)
89	94	}

internal/store/mailreply.go +19

		@@ -1,10 +1,29 @@
1	1	package store
2	2
3	3	import (
	4	"database/sql"
	5	"errors"
4	6	"strings"
5	7	"time"
6	8	)
7	9
	10	// CreatedAt is when the account (table "users") or repository ("repos")
	11	// with id was created.
	12	func (s *Store) CreatedAt(table string, id int64) (time.Time, error) {
	13	if table != "users" && table != "repos" {
	14	return time.Time{}, errors.New("CreatedAt: unknown table " + table)
	15	}
	16	var v string
	17	err := s.DB.QueryRow("SELECT created_at FROM "+table+" WHERE id = ?", id).Scan(&v)
	18	if errors.Is(err, sql.ErrNoRows) {
	19	return time.Time{}, ErrNotFound
	20	}
	21	if err != nil {
	22	return time.Time{}, err
	23	}
	24	return time.Parse("2006-01-02T15:04:05.000Z", v)
	25	}
	26
8	27	// ClaimMailReply records that the reply identified by key is being
9	28	// posted. False means an earlier fetch of the same message already
10	29	// claimed it.

internal/store/migrations/0073_mail_reply.up.sql +4

		@@ -1,9 +1,10 @@
1	1	-- Reply by mail (#295). notify_reply puts a Reply-To carrying a reply
2	2	-- token on the account's issue and merge request mail when the instance
3	3	-- polls a mailbox for replies. notifications.reply_to is that address,
	4	-- per queued message, blanked once it is sent. mail_replies records each
	5	-- reply that posted a comment, keyed by account, thread and Message-ID
	6	-- (or a hash of the message when it has none), so a message fetched
	7	-- twice posts once.
4	8	ALTER TABLE users ADD COLUMN notify_reply INTEGER NOT NULL DEFAULT 0;
5	9	ALTER TABLE notifications ADD COLUMN reply_to TEXT NOT NULL DEFAULT '';
6	10	CREATE TABLE mail_replies (

internal/store/notify.go +5 −3

		@@ -15,7 +15,9 @@ func (s *Store) EnqueueMail(recipient, subject, body string) error {
15	15	return s.EnqueueMailReplyTo(recipient, "", subject, body)
16	16	}
17	17
18		// EnqueueMailReplyTo queues mail with a Reply-To address.
	18	// EnqueueMailReplyTo queues mail with a Reply-To address. The address
	19	// carries a reply token, so it is blanked once the row is sent or
	20	// dead-lettered.
19	21	func (s *Store) EnqueueMailReplyTo(recipient, replyTo, subject, body string) error {
20	22	_, err := s.DB.Exec(
21	23	"INSERT INTO notifications (recipient, reply_to, subject, body) VALUES (?, ?, ?, ?)",
		@@ -46,14 +48,14 @@ func (s *Store) DueMail(limit int) ([]QueuedMail, error) {
46	48
47	49	func (s *Store) MarkMailSent(id int64) error {
48	50	_, err := s.DB.Exec(
49		"UPDATE notifications SET sent_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), attempts = attempts + 1 WHERE id = ?", id)
	51	"UPDATE notifications SET sent_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), attempts = attempts + 1, reply_to = '' WHERE id = ?", id)
50	52	return err
51	53	}
52	54
53	55	func (s Store) MarkMailFailed(id int64, errMsg string, nextAt time.Time) error {
54	56	if nextAt == nil {
55	57	_, err := s.DB.Exec(
56		"UPDATE notifications SET failed_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), attempts = attempts + 1, last_error = ? WHERE id = ?",
	58	"UPDATE notifications SET failed_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), attempts = attempts + 1, last_error = ?, reply_to = '' WHERE id = ?",
57	59	errMsg, id)
58	60	return err
59	61	}

Commit 9a44de69f1

Commit `9a44de69f1`