package bamboo import ( "bufio" "context" "crypto/subtle" "encoding/base64" "encoding/json" "errors" "fmt" "io" "log" "net" "net/textproto" "strconv" "strings" "time" "git.veillax.com/Veillax/reef/model" "git.veillax.com/Veillax/reef/tunnel" ) // Mail egress: a small SMTP relay on this machine. Point your mail server's // relayhost at it (e.g. Postfix: relayhost = [127.0.0.1]:2525). Each message is // handed through the WireGuard tunnel to the edge, which delivers it to the // recipient's server from the edge's public IP and passes the replies back. // // One transaction is relayed to one destination domain. If a message has // recipients in several domains, the extra ones get "452 too many recipients" // and your MTA resends them in a separate transaction, which is standard behavior. type mailRun struct { spec model.MailSpec ln net.Listener } const maxRecipients = 100 // syncMail starts, restarts and stops relays to match specs. func (b *Bamboo) syncMail(specs []model.MailSpec) { b.mu.Lock() defer b.mu.Unlock() want := map[string]model.MailSpec{} for _, sp := range specs { want[sp.ID] = sp } for id, run := range b.mail { if sp, ok := want[id]; !ok || sp != run.spec { run.ln.Close() delete(b.mail, id) delete(b.mailErr, id) } } for id, sp := range want { if _, running := b.mail[id]; running { continue } ln, err := net.Listen("tcp", sp.Listen) if err != nil { b.mailErr[id] = err.Error() log.Printf("mail egress %s: %v", sp.Listen, err) continue } run := &mailRun{spec: sp, ln: ln} b.mail[id] = run b.mailErr[id] = "" log.Printf("mail egress relay listening on %s (via edge %s)", sp.Listen, sp.EdgeID) go b.serveMail(run) } } func (b *Bamboo) mailStatus() map[string]string { b.mu.Lock() defer b.mu.Unlock() out := map[string]string{} for id, e := range b.mailErr { out[id] = e } return out } func (b *Bamboo) serveMail(r *mailRun) { for { c, err := r.ln.Accept() if err != nil { return } go b.handleSMTP(c, r.spec) } } type smtpSession struct { b *Bamboo c net.Conn tp *textproto.Conn spec model.MailSpec authed bool hasFrom bool from string domain string rcpts int stream net.Conn sr *bufio.Reader } func (s *smtpSession) say(code int, text string) { text = strings.NewReplacer("\r", " ", "\n", " ").Replace(text) fmt.Fprintf(s.c, "%d %s\r\n", code, text) } func (s *smtpSession) sayMulti(code int, lines ...string) { for i, l := range lines { sep := "-" if i == len(lines)-1 { sep = " " } fmt.Fprintf(s.c, "%d%s%s\r\n", code, sep, l) } } // openStream connects to the edge's mail port through the tunnel. func (s *smtpSession) openStream() error { b := s.b b.mu.Lock() tun, edgeIP := b.tun, "" for _, e := range b.edges { if e.ID == s.spec.EdgeID { edgeIP = e.IP } } b.mu.Unlock() if tun == nil || edgeIP == "" { return errors.New("the tunnel to the edge isn't up") } ctx, cancel := context.WithTimeout(b.ctx, 12*time.Second) defer cancel() c, err := tun.DialContext(ctx, "tcp", net.JoinHostPort(edgeIP, strconv.Itoa(tunnel.MailTCPPort))) if err != nil { return errors.New("can't reach the edge through the tunnel") } s.stream, s.sr = c, bufio.NewReader(c) return nil } func (s *smtpSession) closeStream() { if s.stream != nil { s.stream.Close() s.stream, s.sr = nil, nil } } func (s *smtpSession) resetTxn() { s.closeStream() s.hasFrom, s.from, s.domain, s.rcpts = false, "", "", 0 } // ask sends one request to the edge and reads its reply. func (s *smtpSession) ask(m tunnel.MailMsg, body []byte) (tunnel.MailReply, error) { var rep tunnel.MailReply line, _ := json.Marshal(m) s.stream.SetDeadline(time.Now().Add(4 * time.Minute)) if _, err := s.stream.Write(append(line, '\n')); err != nil { return rep, err } if body != nil { if _, err := s.stream.Write(body); err != nil { return rep, err } } resp, err := s.sr.ReadBytes('\n') if err != nil { return rep, err } return rep, json.Unmarshal(resp, &rep) } // angle extracts the address from "FROM: SIZE=1" or "TO:". func angle(arg, key string) (addr, params string, ok bool) { if len(arg) < len(key) || !strings.EqualFold(arg[:len(key)], key) { return "", "", false } rest := strings.TrimSpace(arg[len(key):]) if !strings.HasPrefix(rest, "<") { return "", "", false } end := strings.IndexByte(rest, '>') if end < 0 { return "", "", false } return rest[1:end], strings.TrimSpace(rest[end+1:]), true } func (b *Bamboo) handleSMTP(c net.Conn, spec model.MailSpec) { s := &smtpSession{b: b, c: c, tp: textproto.NewConn(c), spec: spec, authed: spec.Username == ""} defer func() { recover() s.closeStream() c.Close() }() s.say(220, "mail-egress ESMTP ready") badAuth := 0 for { c.SetReadDeadline(time.Now().Add(5 * time.Minute)) line, err := s.tp.ReadLine() if err != nil { return } cmd, arg, _ := strings.Cut(line, " ") arg = strings.TrimSpace(arg) switch strings.ToUpper(cmd) { case "EHLO", "HELO": lines := []string{"mail-egress at your service", "SIZE " + strconv.Itoa(tunnel.MaxMessage), "8BITMIME", "SMTPUTF8", "ENHANCEDSTATUSCODES"} if spec.Username != "" { lines = append(lines, "AUTH PLAIN LOGIN") } if strings.ToUpper(cmd) == "HELO" { s.say(250, "mail-egress") } else { s.sayMulti(250, lines...) } case "AUTH": if spec.Username == "" { s.say(503, "5.5.1 authentication isn't needed here") continue } if !s.auth(arg) { s.say(535, "5.7.8 authentication failed") if badAuth++; badAuth >= 3 { return } continue } s.authed = true s.say(235, "2.7.0 authenticated") case "MAIL": if !s.authed { s.say(530, "5.7.0 authentication required") continue } if s.hasFrom { s.say(503, "5.5.1 a transaction is already in progress") continue } addr, params, ok := angle(arg, "FROM:") if !ok { s.say(501, "5.1.7 syntax: MAIL FROM:
") continue } if strings.Contains(strings.ToUpper(params), "SIZE=") { for _, p := range strings.Fields(params) { if strings.HasPrefix(strings.ToUpper(p), "SIZE=") { if n, _ := strconv.Atoi(p[5:]); n > tunnel.MaxMessage { s.say(552, "5.3.4 message too large") addr = "\x00" } } } if addr == "\x00" { continue } } s.hasFrom, s.from = true, addr s.say(250, "2.1.0 ok") case "RCPT": if !s.hasFrom { s.say(503, "5.5.1 send MAIL FROM first") continue } addr, _, ok := angle(arg, "TO:") if !ok || addr == "" { s.say(501, "5.1.3 syntax: RCPT TO:
") continue } s.rcpt(addr) case "DATA": s.data() case "RSET": if s.stream != nil { s.ask(tunnel.MailMsg{Op: "rset"}, nil) } s.resetTxn() s.say(250, "2.0.0 ok") case "NOOP": s.say(250, "2.0.0 ok") case "QUIT": s.say(221, "2.0.0 bye") return case "VRFY", "EXPN": s.say(252, "2.5.2 cannot verify, but will accept and attempt delivery") case "STARTTLS": s.say(502, "5.5.1 TLS isn't offered: this relay is meant to be reached on this machine") default: s.say(500, "5.5.2 command not recognized") } } } func (s *smtpSession) auth(arg string) bool { mech, initial, _ := strings.Cut(arg, " ") read := func() string { // one base64 line from the client l, err := s.tp.ReadLine() if err != nil { return "" } b, _ := base64.StdEncoding.DecodeString(l) return string(b) } var user, pass string switch strings.ToUpper(mech) { case "PLAIN": raw := "" if initial != "" { b, _ := base64.StdEncoding.DecodeString(initial) raw = string(b) } else { s.say(334, "") raw = read() } parts := strings.Split(raw, "\x00") // authzid \0 user \0 password if len(parts) != 3 { return false } user, pass = parts[1], parts[2] case "LOGIN": s.say(334, base64.StdEncoding.EncodeToString([]byte("Username:"))) user = read() s.say(334, base64.StdEncoding.EncodeToString([]byte("Password:"))) pass = read() default: return false } okU := subtle.ConstantTimeCompare([]byte(user), []byte(s.spec.Username)) == 1 okP := subtle.ConstantTimeCompare([]byte(pass), []byte(s.spec.Password)) == 1 return okU && okP } func (s *smtpSession) rcpt(addr string) { at := strings.LastIndexByte(addr, '@') if at < 1 || at == len(addr)-1 { s.say(501, "5.1.3 bad recipient address") return } domain := strings.ToLower(addr[at+1:]) if s.rcpts > 0 && domain != s.domain { s.say(452, "4.5.3 one destination domain per transaction; send the others separately") return } if s.rcpts >= maxRecipients { s.say(452, "4.5.3 too many recipients") return } first := s.stream == nil if first { if err := s.openStream(); err != nil { s.say(451, "4.4.1 "+err.Error()) return } } rep, err := s.ask(tunnel.MailMsg{Op: "rcpt", Egress: s.spec.ID, From: s.from, To: addr}, nil) if err != nil { s.closeStream() s.say(451, "4.4.2 lost contact with the edge") return } if rep.Code != 250 { if first { s.closeStream() } s.say(rep.Code, rep.Text) return } s.rcpts++ s.domain = domain s.say(250, "2.1.5 ok") } func (s *smtpSession) data() { if s.rcpts == 0 || s.stream == nil { s.say(503, "5.5.1 need a recipient first") return } rep, err := s.ask(tunnel.MailMsg{Op: "data"}, nil) if err != nil { s.resetTxn() s.say(451, "4.4.2 lost contact with the edge") return } if rep.Code != 354 { s.say(rep.Code, rep.Text) s.resetTxn() return } s.say(354, "end data with .") s.c.SetReadDeadline(time.Now().Add(10 * time.Minute)) dr := s.tp.DotReader() body, err := io.ReadAll(io.LimitReader(dr, tunnel.MaxMessage+1)) if err != nil { s.resetTxn() return } if len(body) > tunnel.MaxMessage { io.Copy(io.Discard, dr) s.ask(tunnel.MailMsg{Op: "rset"}, nil) s.resetTxn() s.say(552, "5.3.4 message too large") return } rep, err = s.ask(tunnel.MailMsg{Op: "body", Len: len(body)}, body) s.resetTxn() if err != nil { // we can't know whether the far side accepted it; a temporary failure // makes the MTA retry, which risks a duplicate but never loses mail s.say(451, "4.4.2 lost contact with the edge while sending") return } s.say(rep.Code, rep.Text) }