Hammerhead/e2e/mail_test.go

package e2e

import (
	"crypto/tls"
	"errors"
	"fmt"
	"net"
	"net/smtp"
	"net/textproto"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"git.veillax.com/Veillax/blacktip"
	"git.veillax.com/Veillax/reef/model"
)

// ---------- a fake recipient mail server ----------

type mxMsg struct {
	from, ehlo, remote, body string
	rcpts                    []string
	tls                      bool
}

type fakeMX struct {
	addr     string
	mu       sync.Mutex
	msgs     []mxMsg
	rcptCode map[string]int // address -> reply code to give instead of 250
	offerTLS bool
	cert     *tls.Certificate
}

func newFakeMX(t *testing.T, offerTLS bool) *fakeMX {
	l, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { l.Close() })
	m := &fakeMX{addr: l.Addr().String(), rcptCode: map[string]int{}, offerTLS: offerTLS}
	if offerTLS {
		m.cert = selfSigned(t, "mx.example.invalid")
	}
	go func() {
		for {
			c, err := l.Accept()
			if err != nil {
				return
			}
			go m.serve(c)
		}
	}()
	return m
}

func (m *fakeMX) serve(c net.Conn) {
	defer c.Close()
	host, _, _ := net.SplitHostPort(c.RemoteAddr().String())
	tp := textproto.NewConn(c)
	say := func(format string, a ...any) { fmt.Fprintf(c, format+"\r\n", a...) }
	say("220 fake-mx ESMTP")
	cur := mxMsg{remote: host}
	usedTLS := false
	for {
		line, err := tp.ReadLine()
		if err != nil {
			return
		}
		verb, arg, _ := strings.Cut(line, " ")
		switch strings.ToUpper(verb) {
		case "EHLO", "HELO":
			cur.ehlo = arg
			if m.offerTLS && !usedTLS {
				say("250-fake-mx")
				say("250-STARTTLS")
				say("250 8BITMIME")
			} else {
				say("250-fake-mx")
				say("250 8BITMIME")
			}
		case "STARTTLS":
			say("220 go ahead")
			tc := tls.Server(c, &tls.Config{Certificates: []tls.Certificate{*m.cert}})
			if tc.Handshake() != nil {
				return
			}
			c, tp, usedTLS = tc, textproto.NewConn(tc), true
			say = func(format string, a ...any) { fmt.Fprintf(tc, format+"\r\n", a...) }
		case "MAIL":
			if i, j := strings.IndexByte(arg, '<'), strings.IndexByte(arg, '>'); i >= 0 && j > i {
				cur.from = arg[i+1 : j]
			}
			cur.rcpts, cur.tls = nil, usedTLS
			say("250 ok")
		case "RCPT":
			a := strings.Trim(arg[3:], "<> ")
			if code, bad := m.rcptCode[a]; bad {
				say("%d no thanks", code)
				continue
			}
			cur.rcpts = append(cur.rcpts, a)
			say("250 ok")
		case "DATA":
			say("354 go on")
			b := new(strings.Builder)
			r := tp.DotReader()
			buf := make([]byte, 4096)
			for {
				n, err := r.Read(buf)
				b.Write(buf[:n])
				if err != nil {
					break
				}
			}
			cur.body = b.String()
			m.mu.Lock()
			m.msgs = append(m.msgs, cur)
			m.mu.Unlock()
			say("250 2.0.0 queued as FAKE%d", len(m.msgs))
		case "RSET", "NOOP":
			say("250 ok")
		case "QUIT":
			say("221 bye")
			return
		default:
			say("502 nope")
		}
	}
}

func (m *fakeMX) last(t *testing.T) mxMsg {
	t.Helper()
	m.mu.Lock()
	defer m.mu.Unlock()
	if len(m.msgs) == 0 {
		t.Fatal("the mail server received nothing")
	}
	return m.msgs[len(m.msgs)-1]
}

func (m *fakeMX) count() int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return len(m.msgs)
}

// ---------- helpers ----------

type mailEnv struct {
	*world
	mx    *fakeMX
	mu    sync.Mutex
	hosts map[string][]blacktip.MailHost
}

func setupMail(t *testing.T, allowPrivate bool, mx *fakeMX, extra ...func(*blacktip.Config)) *mailEnv {
	e := &mailEnv{mx: mx, hosts: map[string][]blacktip.MailHost{}}
	host, port, _ := net.SplitHostPort(mx.addr)
	p, _ := strconv.Atoi(port)
	e.hosts["dest.example"] = []blacktip.MailHost{{Host: host, Port: p}}
	opts := append([]func(*blacktip.Config){func(c *blacktip.Config) {
		c.MailAllowPrivate = allowPrivate
		c.MailProbe = mx.addr
		c.MailLookup = func(domain string) ([]blacktip.MailHost, error) {
			e.mu.Lock()
			defer e.mu.Unlock()
			if h, ok := e.hosts[domain]; ok {
				return h, nil
			}
			return nil, errors.New("no such domain in this test")
		}
	}}, extra...)
	e.world = setup(t, blacktip.TraefikConfig{}, opts...)
	return e
}

func (e *mailEnv) add(m map[string]any) (id, listen string) {
	e.t.Helper()
	listen = "127.0.0.1:" + strconv.Itoa(freePort(e.t))
	defaults := map[string]any{"listen": listen, "sourceIp": "127.0.0.2", "heloName": "mail.example.invalid",
		"siteId": e.site, "edgeId": e.edge, "enabled": true}
	for k, v := range defaults {
		if _, ok := m[k]; !ok {
			m[k] = v
		}
	}
	var out struct{ ID string }
	e.admin.must("POST", "/api/orgs/"+e.org+"/mail-egress", m, &out)
	return out.ID, m["listen"].(string)
}

// send delivers one message through the relay at addr, retrying while the tunnel comes up.
func send(t *testing.T, addr string, auth smtp.Auth, from string, to []string, body string) error {
	t.Helper()
	var err error
	for i := 0; i < 50; i++ {
		err = trySend(addr, auth, from, to, body)
		var te *textproto.Error
		// retry only while the relay or the tunnel is still coming up
		transient := err != nil && (!errors.As(err, &te) || (te.Code == 451 && strings.Contains(te.Msg, "edge")))
		if !transient {
			return err
		}
		time.Sleep(200 * time.Millisecond)
	}
	return err
}

func trySend(addr string, auth smtp.Auth, from string, to []string, body string) error {
	c, err := smtp.Dial(addr)
	if err != nil {
		return err
	}
	defer c.Close()
	if err := c.Hello("client.local"); err != nil {
		return err
	}
	if auth != nil {
		if err := c.Auth(auth); err != nil {
			return err
		}
	}
	if err := c.Mail(from); err != nil {
		return err
	}
	for _, r := range to {
		if err := c.Rcpt(r); err != nil {
			return err
		}
	}
	w, err := c.Data()
	if err != nil {
		return err
	}
	if _, err := w.Write([]byte(body)); err != nil {
		return err
	}
	if err := w.Close(); err != nil {
		return err
	}
	return c.Quit()
}

func smtpCode(err error) int {
	var te *textproto.Error
	if errors.As(err, &te) {
		return te.Code
	}
	return 0
}

const letter = "Subject: hi\r\nFrom: me@home.example\r\nTo: you@dest.example\r\n\r\nHello.\r\n.line that starts with a dot\r\nBye.\r\n"

// ---------- the tests ----------

func TestMailEgressDeliversFromTheEdgeIP(t *testing.T) {
	mx := newFakeMX(t, true) // offers STARTTLS (self-signed)
	e := setupMail(t, true, mx)
	_, listen := e.add(map[string]any{"name": "mail out"})

	if err := send(t, listen, nil, "me@home.example", []string{"you@dest.example", "other@dest.example"}, letter); err != nil {
		t.Fatalf("send: %v", err)
	}
	got := mx.last(t)
	if got.remote != "127.0.0.2" {
		t.Errorf("mail left from %s, want the edge source IP 127.0.0.2", got.remote)
	}
	if got.ehlo != "mail.example.invalid" {
		t.Errorf("announced itself as %q, want the configured HELO name", got.ehlo)
	}
	if got.from != "me@home.example" || strings.Join(got.rcpts, ",") != "you@dest.example,other@dest.example" {
		t.Errorf("envelope wrong: from=%q rcpts=%v", got.from, got.rcpts)
	}
	if !got.tls {
		t.Error("opportunistic STARTTLS was not used although the server offered it")
	}
	if !strings.Contains(got.body, "\n.line that starts with a dot\n") && !strings.Contains(got.body, "\r\n.line that starts with a dot\r\n") {
		t.Errorf("a line starting with a dot was mangled: %q", got.body)
	}
	if !strings.Contains(got.body, "Hello.") || !strings.Contains(got.body, "Bye.") {
		t.Errorf("body incomplete: %q", got.body)
	}

	// a 1 MB message goes through intact
	big := "Subject: big\r\n\r\n" + strings.Repeat("0123456789abcdef", 64*1024) + "\r\n"
	if err := send(t, listen, nil, "me@home.example", []string{"you@dest.example"}, big); err != nil {
		t.Fatalf("big message: %v", err)
	}
	if n := len(mx.last(t).body); n < len(big)-200 {
		t.Errorf("large message truncated: got %d bytes of %d", n, len(big))
	}

	// the null sender (bounces) is allowed
	if err := send(t, listen, nil, "", []string{"you@dest.example"}, letter); err != nil {
		t.Errorf("null sender: %v", err)
	}
	if mx.last(t).from != "" {
		t.Errorf("null sender became %q", mx.last(t).from)
	}

	// the delivery shows up in the access log with the recipient domain and the SMTP status
	deadline := time.Now().Add(10 * time.Second)
	for {
		var logs []model.LogEntry
		e.admin.must("GET", "/api/orgs/"+e.org+"/logs?kind=access&limit=100", nil, &logs)
		found := false
		for _, l := range logs {
			if l.Result == "mail" && l.Host == "dest.example" && l.Status == 250 {
				found = true
			}
		}
		if found {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("no mail entry in the access log: %+v", logs)
		}
		time.Sleep(300 * time.Millisecond)
	}
}

func TestMailEgressReplyHandling(t *testing.T) {
	mx := newFakeMX(t, false)
	mx.rcptCode["nobody@dest.example"] = 550
	mx.rcptCode["full@dest.example"] = 452
	e := setupMail(t, true, mx)
	_, listen := e.add(map[string]any{})

	// the receiving server's refusal reaches the sender with its own code
	err := send(t, listen, nil, "me@home.example", []string{"nobody@dest.example"}, letter)
	if smtpCode(err) != 550 {
		t.Errorf("a 550 from the recipient's server came back as %v", err)
	}
	// ...and a refused first recipient doesn't poison the next transaction
	if err := send(t, listen, nil, "me@home.example", []string{"you@dest.example"}, letter); err != nil {
		t.Errorf("a good message after a refusal: %v", err)
	}
	// a refused extra recipient is reported for that recipient only
	c, err := smtp.Dial(listen)
	if err != nil {
		t.Fatal(err)
	}
	defer c.Close()
	c.Hello("client.local")
	c.Mail("me@home.example")
	if err := c.Rcpt("you@dest.example"); err != nil {
		t.Fatalf("rcpt 1: %v", err)
	}
	if err := c.Rcpt("full@dest.example"); smtpCode(err) != 452 {
		t.Errorf("second recipient: %v", err)
	}
	// recipients in a different domain are deferred to another transaction
	e.mu.Lock()
	e.hosts["elsewhere.example"] = e.hosts["dest.example"]
	e.mu.Unlock()
	if err := c.Rcpt("x@elsewhere.example"); smtpCode(err) != 452 {
		t.Errorf("a second domain in one transaction: %v", err)
	}
	c.Reset()
	if err := c.Mail("me@home.example"); err != nil {
		t.Errorf("RSET then a new transaction: %v", err)
	}

	// a mail server that can't be reached is a temporary failure: the sender will retry
	e.mu.Lock()
	e.hosts["down.example"] = []blacktip.MailHost{{Host: "127.0.0.1", Port: freePort(t)}}
	e.mu.Unlock()
	if err := send(t, listen, nil, "me@home.example", []string{"a@down.example"}, letter); smtpCode(err) != 451 {
		t.Errorf("unreachable server: %v", err)
	}
	// an unknown domain with no records is temporary here too (the test resolver errors)
	if err := send(t, listen, nil, "me@home.example", []string{"a@nowhere.example"}, letter); smtpCode(err) < 400 {
		t.Errorf("unknown domain: %v", err)
	}
}

func TestMailEgressAuthTLSSmarthostAndRate(t *testing.T) {
	plain := newFakeMX(t, false)
	e := setupMail(t, true, plain)

	// passwords
	_, authAddr := e.add(map[string]any{"name": "auth", "username": "relay", "password": "hunter2!"})
	if err := send(t, authAddr, smtp.PlainAuth("", "relay", "hunter2!", "127.0.0.1"), "me@home.example", []string{"you@dest.example"}, letter); err != nil {
		t.Fatalf("authenticated send: %v", err)
	}
	if err := trySend(authAddr, nil, "me@home.example", []string{"you@dest.example"}, letter); smtpCode(err) != 530 {
		t.Errorf("sending without credentials: %v", err)
	}
	if err := trySend(authAddr, smtp.PlainAuth("", "relay", "wrong", "127.0.0.1"), "me@home.example", []string{"you@dest.example"}, letter); smtpCode(err) != 535 {
		t.Errorf("wrong password: %v", err)
	}

	// require-TLS refuses a server with no STARTTLS, and one whose certificate can't be verified
	_, tlsAddr := e.add(map[string]any{"name": "strict", "requireTls": true})
	err := send(t, tlsAddr, nil, "me@home.example", []string{"you@dest.example"}, letter)
	if smtpCode(err) != 451 || !strings.Contains(err.Error(), "TLS") {
		t.Errorf("require-TLS against a plain server: %v", err)
	}
	selfSignedMX := newFakeMX(t, true)
	h, p, _ := net.SplitHostPort(selfSignedMX.addr)
	pn, _ := strconv.Atoi(p)
	e.mu.Lock()
	e.hosts["tls.example"] = []blacktip.MailHost{{Host: h, Port: pn}}
	e.mu.Unlock()
	err = send(t, tlsAddr, nil, "me@home.example", []string{"you@tls.example"}, letter)
	if smtpCode(err) != 451 || selfSignedMX.count() != 0 {
		t.Errorf("require-TLS with an untrusted certificate: err=%v delivered=%d", err, selfSignedMX.count())
	}

	// a smarthost receives everything, whatever the domain
	smart := newFakeMX(t, false)
	_, smartAddr := e.add(map[string]any{"name": "smart", "smarthost": smart.addr})
	if err := send(t, smartAddr, nil, "me@home.example", []string{"anyone@unknown.example"}, letter); err != nil {
		t.Fatalf("smarthost: %v", err)
	}
	if got := smart.last(t); got.remote != "127.0.0.2" || strings.Join(got.rcpts, ",") != "anyone@unknown.example" {
		t.Errorf("smarthost got %+v", got)
	}

	// the per-minute limit protects the IP's reputation
	_, rateAddr := e.add(map[string]any{"name": "slow", "ratePerMinute": 2})
	for i := 0; i < 2; i++ {
		if err := send(t, rateAddr, nil, "me@home.example", []string{"you@dest.example"}, letter); err != nil {
			t.Fatalf("message %d: %v", i+1, err)
		}
	}
	if err := send(t, rateAddr, nil, "me@home.example", []string{"you@dest.example"}, letter); smtpCode(err) != 451 || !strings.Contains(err.Error(), "rate") {
		t.Errorf("third message in a minute: %v", err)
	}
}

func TestMailEgressGuardsAndValidation(t *testing.T) {
	mx := newFakeMX(t, false)
	e := setupMail(t, false, mx) // loopback destinations are NOT allowed here
	_, listen := e.add(map[string]any{})

	// a mail server on an internal address is refused: the edge can't be aimed at its own network
	err := send(t, listen, nil, "me@home.example", []string{"you@dest.example"}, letter)
	if smtpCode(err) != 554 || !strings.Contains(err.Error(), "not allowed") || mx.count() != 0 {
		t.Errorf("internal destination: err=%v delivered=%d", err, mx.count())
	}

	base := "/api/orgs/" + e.org + "/mail-egress"
	good := func(over map[string]any) map[string]any {
		m := map[string]any{"siteId": e.site, "edgeId": e.edge, "sourceIp": "127.0.0.2", "heloName": "mail.example.invalid",
			"listen": "127.0.0.1:" + strconv.Itoa(freePort(t)), "enabled": true}
		for k, v := range over {
			m[k] = v
		}
		return m
	}
	e.admin.expect(400, "POST", base, good(map[string]any{"siteId": e.local}))                                                                // the local site needs no relay
	e.admin.expect(400, "POST", base, good(map[string]any{"sourceIp": ""}))                                                                   // the exit IP must be chosen
	e.admin.expect(400, "POST", base, good(map[string]any{"sourceIp": "nope"}))                                                               //
	e.admin.expect(400, "POST", base, good(map[string]any{"heloName": "not a host"}))                                                         //
	e.admin.expect(400, "POST", base, good(map[string]any{"heloName": ""}))                                                                   //
	e.admin.expect(400, "POST", base, good(map[string]any{"smarthost": "no-port"}))                                                           //
	e.admin.expect(400, "POST", base, good(map[string]any{"ratePerMinute": 99999}))                                                           //
	e.admin.expect(400, "POST", base, good(map[string]any{"username": "x"}))                                                                  // user without password
	e.admin.expect(400, "POST", base, good(map[string]any{"listen": "0.0.0.0:2525"}))                                                         // open to the LAN without a password
	e.admin.expect(400, "POST", base, good(map[string]any{"listen": listen}))                                                                 // already taken
	e.admin.must("POST", base, good(map[string]any{"listen": "0.0.0.0:" + strconv.Itoa(freePort(t)), "username": "u", "password": "p"}), nil) // LAN is fine with a password

	// the state shows the egress without its password, plus the edge's deliverability report
	deadline := time.Now().Add(15 * time.Second)
	for {
		var st struct {
			MailEgress []struct {
				ID, Name    string
				PasswordSet bool
				Password    string
				Check       *model.MailCheck
			}
		}
		e.admin.must("GET", "/api/orgs/"+e.org+"/state", nil, &st)
		ready := len(st.MailEgress) > 0
		for _, m := range st.MailEgress {
			if m.Password != "" {
				t.Fatalf("state leaks the password of %s", m.Name)
			}
			if m.Check == nil || m.Check.Port25 == "" {
				ready = false
			}
		}
		if ready {
			for _, m := range st.MailEgress {
				if m.Check.Port25 != "open" {
					t.Errorf("port 25 probe = %q, want open (the probe target is the fake server)", m.Check.Port25)
				}
				if m.Check.PTRMatchesHelo || m.Check.HeloResolves {
					t.Errorf("a made-up HELO name can't match reverse DNS: %+v", m.Check)
				}
			}
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("no deliverability report arrived: %+v", st.MailEgress)
		}
		time.Sleep(300 * time.Millisecond)
	}
}