#7: implement POP3 protocol server (RFC 1939, RFC 2449, STLS)

This commit is contained in:
Lucas Rayan committed 2026-07-24 11:51:27 -03:00
1 parent 883c78e48c
commit 9ca97298bd
7 files changed
+1238

No files matched your search

@@ -0,0 +1,436 @@
package pop3presentation
import (
"bufio"
"bytes"
"context"
"crypto/tls"
"fmt"
"net"
"strconv"
"strings"
"lambdamail/protocols/internal/application/port"
appusecase "lambdamail/protocols/internal/application/usecase"
)
type state int
const (
stateAuth state = iota
stateTransaction
stateUpdate
)
type pop3Msg struct {
seqNum int
record port.MessageRecord
deleted bool
}
type conn struct {
rawConn net.Conn
reader *bufio.Reader
writer *bufio.Writer
tlsProvider port.CertProvider
useCase *appusecase.Pop3SessionUseCase
state state
username string
mailboxID string
folderID string
messages []pop3Msg
isTLS bool
}
func newConn(c net.Conn, useCase *appusecase.Pop3SessionUseCase, tlsProvider port.CertProvider) *conn {
return &conn{
rawConn: c,
reader: bufio.NewReader(c),
writer: bufio.NewWriter(c),
useCase: useCase,
tlsProvider: tlsProvider,
state: stateAuth,
}
}
func (c *conn) serve() {
defer c.rawConn.Close()
if err := c.writeLine("+OK LambdaMail POP3 server ready"); err != nil {
return
}
for {
line, err := c.reader.ReadString('\n')
if err != nil {
return
}
line = strings.TrimRight(line, "\r\n")
if line == "" {
continue
}
parts := strings.SplitN(line, " ", 2)
cmd := strings.ToUpper(parts[0])
arg := ""
if len(parts) > 1 {
arg = parts[1]
}
switch cmd {
case "CAPA":
c.handleCapa()
case "STLS":
if c.handleStls() {
// STLS succeeded, TLS handshake took over
continue
}
case "USER":
c.handleUser(arg)
case "PASS":
c.handlePass(arg)
case "STAT":
c.handleStat()
case "LIST":
c.handleList(arg)
case "UIDL":
c.handleUidl(arg)
case "RETR":
c.handleRetr(arg)
case "TOP":
c.handleTop(arg)
case "DELE":
c.handleDele(arg)
case "RSET":
c.handleRset()
case "NOOP":
c.handleNoop()
case "QUIT":
c.handleQuit()
return
default:
c.writeLine("-ERR Unknown or unsupported command")
}
}
}
func (c *conn) writeLine(msg string) error {
if _, err := c.writer.WriteString(msg + "\r\n"); err != nil {
return err
}
return c.writer.Flush()
}
func (c *conn) handleCapa() {
c.writer.WriteString("+OK Capability list follows\r\n")
c.writer.WriteString("USER\r\n")
c.writer.WriteString("TOP\r\n")
c.writer.WriteString("UIDL\r\n")
c.writer.WriteString("RESP-CODES\r\n")
c.writer.WriteString("PIPELINING\r\n")
if !c.isTLS && c.tlsProvider != nil {
c.writer.WriteString("STLS\r\n")
}
c.writer.WriteString("IMPLEMENTATION LambdaMail POP3\r\n")
c.writer.WriteString(".\r\n")
c.writer.Flush()
}
func (c *conn) handleStls() bool {
if c.isTLS {
c.writeLine("-ERR TLS already active")
return false
}
if c.tlsProvider == nil {
c.writeLine("-ERR STLS not supported")
return false
}
if err := c.writeLine("+OK Begin TLS negotiation"); err != nil {
return false
}
tlsConfig := &tls.Config{
GetCertificate: c.tlsProvider.GetCertificate,
MinVersion: tls.VersionTLS12,
}
tlsConn := tls.Server(c.rawConn, tlsConfig)
if err := tlsConn.Handshake(); err != nil {
return false
}
c.rawConn = tlsConn
c.reader = bufio.NewReader(tlsConn)
c.writer = bufio.NewWriter(tlsConn)
c.isTLS = true
return true
}
func (c *conn) handleUser(arg string) {
if c.state != stateAuth {
c.writeLine("-ERR Invalid command state")
return
}
if arg == "" {
c.writeLine("-ERR Missing username argument")
return
}
c.username = arg
c.writeLine("+OK User accepted")
}
func (c *conn) handlePass(password string) {
if c.state != stateAuth || c.username == "" {
c.writeLine("-ERR Invalid state or missing USER")
return
}
ctx := context.Background()
mailboxID, err := c.useCase.Login(ctx, c.username, password)
if err != nil {
c.writeLine("-ERR Invalid credentials")
return
}
folder, recs, err := c.useCase.GetInbox(ctx, mailboxID)
if err != nil {
c.writeLine("-ERR Unable to access INBOX")
return
}
c.mailboxID = mailboxID
c.folderID = folder.ID
c.messages = make([]pop3Msg, len(recs))
var totalBytes int64
for i, rec := range recs {
c.messages[i] = pop3Msg{
seqNum: i + 1,
record: rec,
deleted: false,
}
totalBytes += rec.SizeBytes
}
c.state = stateTransaction
c.writeLine(fmt.Sprintf("+OK Logged in, %d message(s) (%d octets)", len(c.messages), totalBytes))
}
func (c *conn) handleStat() {
if c.state != stateTransaction {
c.writeLine("-ERR Invalid command state")
return
}
count, totalBytes := c.statActive()
c.writeLine(fmt.Sprintf("+OK %d %d", count, totalBytes))
}
func (c *conn) statActive() (int, int64) {
var count int
var totalBytes int64
for _, msg := range c.messages {
if !msg.deleted {
count++
totalBytes += msg.record.SizeBytes
}
}
return count, totalBytes
}
func (c *conn) handleList(arg string) {
if c.state != stateTransaction {
c.writeLine("-ERR Invalid command state")
return
}
if arg != "" {
seq, err := strconv.Atoi(arg)
if err != nil || seq < 1 || seq > len(c.messages) || c.messages[seq-1].deleted {
c.writeLine("-ERR No such message")
return
}
c.writeLine(fmt.Sprintf("+OK %d %d", seq, c.messages[seq-1].record.SizeBytes))
return
}
count, totalBytes := c.statActive()
c.writer.WriteString(fmt.Sprintf("+OK %d messages (%d octets)\r\n", count, totalBytes))
for _, msg := range c.messages {
if !msg.deleted {
c.writer.WriteString(fmt.Sprintf("%d %d\r\n", msg.seqNum, msg.record.SizeBytes))
}
}
c.writer.WriteString(".\r\n")
c.writer.Flush()
}
func (c *conn) handleUidl(arg string) {
if c.state != stateTransaction {
c.writeLine("-ERR Invalid command state")
return
}
if arg != "" {
seq, err := strconv.Atoi(arg)
if err != nil || seq < 1 || seq > len(c.messages) || c.messages[seq-1].deleted {
c.writeLine("-ERR No such message")
return
}
c.writeLine(fmt.Sprintf("+OK %d %d", seq, c.messages[seq-1].record.UID))
return
}
c.writer.WriteString("+OK Unique-ID listing follows\r\n")
for _, msg := range c.messages {
if !msg.deleted {
c.writer.WriteString(fmt.Sprintf("%d %d\r\n", msg.seqNum, msg.record.UID))
}
}
c.writer.WriteString(".\r\n")
c.writer.Flush()
}
func (c *conn) handleRetr(arg string) {
if c.state != stateTransaction {
c.writeLine("-ERR Invalid command state")
return
}
seq, err := strconv.Atoi(arg)
if err != nil || seq < 1 || seq > len(c.messages) || c.messages[seq-1].deleted {
c.writeLine("-ERR No such message")
return
}
msg := &c.messages[seq-1]
ctx := context.Background()
content, err := c.useCase.ReadBlob(ctx, msg.record.BlobID)
if err != nil {
c.writeLine("-ERR Unable to read message content")
return
}
c.writer.WriteString(fmt.Sprintf("+OK %d octets\r\n", len(content)))
c.writeDotStuffedContent(content)
c.writer.WriteString(".\r\n")
c.writer.Flush()
}
func (c *conn) handleTop(arg string) {
if c.state != stateTransaction {
c.writeLine("-ERR Invalid command state")
return
}
parts := strings.Fields(arg)
if len(parts) < 2 {
c.writeLine("-ERR Missing parameters for TOP")
return
}
seq, err := strconv.Atoi(parts[0])
numLines, err2 := strconv.Atoi(parts[1])
if err != nil || err2 != nil || seq < 1 || seq > len(c.messages) || c.messages[seq-1].deleted || numLines < 0 {
c.writeLine("-ERR Invalid message number or line count")
return
}
msg := &c.messages[seq-1]
ctx := context.Background()
content, err := c.useCase.ReadBlob(ctx, msg.record.BlobID)
if err != nil {
c.writeLine("-ERR Unable to read message content")
return
}
c.writer.WriteString("+OK Top of message follows\r\n")
topContent := extractTopLines(content, numLines)
c.writeDotStuffedContent(topContent)
c.writer.WriteString(".\r\n")
c.writer.Flush()
}
func extractTopLines(content []byte, n int) []byte {
// Separate header and body by \r\n\r\n or \n\n
sepIdx := bytes.Index(content, []byte("\r\n\r\n"))
sepLen := 4
if sepIdx == -1 {
sepIdx = bytes.Index(content, []byte("\n\n"))
sepLen = 2
}
if sepIdx == -1 {
// Entire content treated as header
return content
}
header := content[:sepIdx+sepLen]
body := content[sepIdx+sepLen:]
var buf bytes.Buffer
buf.Write(header)
bodyScanner := bufio.NewScanner(bytes.NewReader(body))
linesWritten := 0
for bodyScanner.Scan() && linesWritten < n {
buf.WriteString(bodyScanner.Text() + "\r\n")
linesWritten++
}
return buf.Bytes()
}
func (c *conn) writeDotStuffedContent(content []byte) {
s := bufio.NewScanner(bytes.NewReader(content))
for s.Scan() {
line := s.Text()
if strings.HasPrefix(line, ".") {
c.writer.WriteString("." + line + "\r\n")
} else {
c.writer.WriteString(line + "\r\n")
}
}
}
func (c *conn) handleDele(arg string) {
if c.state != stateTransaction {
c.writeLine("-ERR Invalid command state")
return
}
seq, err := strconv.Atoi(arg)
if err != nil || seq < 1 || seq > len(c.messages) || c.messages[seq-1].deleted {
c.writeLine("-ERR Message already deleted or no such message")
return
}
c.messages[seq-1].deleted = true
c.writeLine(fmt.Sprintf("+OK Message %d marked for deletion", seq))
}
func (c *conn) handleRset() {
if c.state != stateTransaction {
c.writeLine("-ERR Invalid command state")
return
}
for i := range c.messages {
c.messages[i].deleted = false
}
c.writeLine("+OK Mailbox reset")
}
func (c *conn) handleNoop() {
c.writeLine("+OK")
}
func (c *conn) handleQuit() {
if c.state == stateAuth {
c.writeLine("+OK POP3 server signing off")
return
}
if c.state == stateTransaction {
c.state = stateUpdate
var toDelete []uint32
for _, msg := range c.messages {
if msg.deleted {
toDelete = append(toDelete, msg.record.UID)
}
}
if len(toDelete) > 0 {
ctx := context.Background()
_ = c.useCase.ExpungeMessages(ctx, c.folderID, toDelete)
}
c.writeLine("+OK POP3 server signing off")
}
}
@@ -0,0 +1,337 @@
package pop3presentation
import (
"bufio"
"context"
"net"
"strings"
"testing"
"time"
"github.com/alexedwards/argon2id"
"github.com/google/uuid"
"lambdamail/protocols/internal/application/port"
appusecase "lambdamail/protocols/internal/application/usecase"
)
type fakeAuthRepository struct {
byAddress map[string]port.MailboxAuth
}
func (f *fakeAuthRepository) FindByAddress(_ context.Context, address string) (*port.MailboxAuth, error) {
rec, ok := f.byAddress[address]
if !ok {
return nil, nil
}
return &rec, nil
}
type fakeImapFolderRepository struct {
byName map[string]port.ImapFolderRecord
}
func (f *fakeImapFolderRepository) ListFolders(_ context.Context, _ string) ([]port.ImapFolderRecord, error) {
var out []port.ImapFolderRecord
for _, rec := range f.byName {
out = append(out, rec)
}
return out, nil
}
func (f *fakeImapFolderRepository) FindByName(_ context.Context, _ string, name string) (*port.ImapFolderRecord, error) {
rec, ok := f.byName[name]
if !ok {
return nil, nil
}
return &rec, nil
}
type fakeMessageQueryRepository struct {
messages []port.MessageRecord
}
func (f *fakeMessageQueryRepository) ListMessages(_ context.Context, _ string) ([]port.MessageRecord, error) {
return f.messages, nil
}
type fakeBlobReader struct {
content map[uuid.UUID][]byte
}
func (f *fakeBlobReader) ReadByID(_ context.Context, blobID uuid.UUID) ([]byte, error) {
return f.content[blobID], nil
}
type fakeExpungeRepository struct {
calls []struct {
folderID string
uids []uint32
}
}
func (f *fakeExpungeRepository) Expunge(_ context.Context, folderID string, uids []uint32) error {
f.calls = append(f.calls, struct {
folderID string
uids []uint32
}{folderID, uids})
return nil
}
func setupPop3Server(t *testing.T, messages []port.MessageRecord, blobs map[uuid.UUID][]byte, expunger *fakeExpungeRepository) (net.Listener, func()) {
t.Helper()
password := "correct horse battery staple"
hash, err := argon2id.CreateHash(password, argon2id.DefaultParams)
if err != nil {
t.Fatalf("hash: %v", err)
}
mailboxID := uuid.New()
auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{
"user@example.test": {ID: mailboxID, PasswordHash: hash},
}}
folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{
"INBOX": {ID: "folder-1", Name: "INBOX", NumMessages: uint32(len(messages))},
}}
query := &fakeMessageQueryRepository{messages: messages}
reader := &fakeBlobReader{content: blobs}
uc := appusecase.NewPop3SessionUseCase(auth, folders, query, reader, expunger)
srv := NewServer("127.0.0.1:0", uc, nil)
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
go srv.Serve(ln)
return ln, func() {
ln.Close()
}
}
func TestPop3Server_Capa_User_Pass_Stat_List_Retr_Dele_Quit(t *testing.T) {
blobID := uuid.New()
msgContent := []byte("From: alice@example.test\r\nSubject: Test\r\n\r\n.Dot-leading line\r\nHello World\r\n")
messages := []port.MessageRecord{
{UID: 101, BlobID: blobID, SizeBytes: int64(len(msgContent))},
}
blobs := map[uuid.UUID][]byte{blobID: msgContent}
expunger := &fakeExpungeRepository{}
ln, cleanup := setupPop3Server(t, messages, blobs, expunger)
defer cleanup()
c, err := net.Dial("tcp", ln.Addr().String())
if err != nil {
t.Fatalf("dial: %v", err)
}
defer c.Close()
reader := bufio.NewReader(c)
writer := bufio.NewWriter(c)
readLine := func() string {
line, err := reader.ReadString('\n')
if err != nil {
t.Fatalf("read: %v", err)
}
return strings.TrimRight(line, "\r\n")
}
sendCmd := func(cmd string) {
writer.WriteString(cmd + "\r\n")
writer.Flush()
}
// 1. Greeting
greeting := readLine()
if !strings.HasPrefix(greeting, "+OK") {
t.Fatalf("greeting = %q, want +OK", greeting)
}
// 2. CAPA
sendCmd("CAPA")
capaLine := readLine()
if !strings.HasPrefix(capaLine, "+OK") {
t.Fatalf("CAPA resp = %q", capaLine)
}
for {
l := readLine()
if l == "." {
break
}
}
// 3. USER & PASS
sendCmd("USER user@example.test")
if l := readLine(); !strings.HasPrefix(l, "+OK") {
t.Fatalf("USER resp = %q", l)
}
sendCmd("PASS correct horse battery staple")
if l := readLine(); !strings.HasPrefix(l, "+OK") {
t.Fatalf("PASS resp = %q", l)
}
// 4. STAT
sendCmd("STAT")
statResp := readLine()
if !strings.HasPrefix(statResp, "+OK 1 ") {
t.Fatalf("STAT resp = %q, want 1 message", statResp)
}
// 5. LIST
sendCmd("LIST")
listHead := readLine()
if !strings.HasPrefix(listHead, "+OK") {
t.Fatalf("LIST resp = %q", listHead)
}
msg1List := readLine()
if !strings.HasPrefix(msg1List, "1 ") {
t.Fatalf("LIST item = %q", msg1List)
}
if dot := readLine(); dot != "." {
t.Fatalf("LIST term = %q", dot)
}
// 6. UIDL
sendCmd("UIDL")
uidlHead := readLine()
if !strings.HasPrefix(uidlHead, "+OK") {
t.Fatalf("UIDL resp = %q", uidlHead)
}
uidlItem := readLine()
if uidlItem != "1 101" {
t.Fatalf("UIDL item = %q, want '1 101'", uidlItem)
}
if dot := readLine(); dot != "." {
t.Fatalf("UIDL term = %q", dot)
}
// 7. RETR 1 (verifying dot-stuffing ..Dot-leading line)
sendCmd("RETR 1")
retrHead := readLine()
if !strings.HasPrefix(retrHead, "+OK") {
t.Fatalf("RETR resp = %q", retrHead)
}
var retrLines []string
for {
l := readLine()
if l == "." {
break
}
retrLines = append(retrLines, l)
}
if len(retrLines) < 4 {
t.Fatalf("retrLines = %v", retrLines)
}
if retrLines[3] != "..Dot-leading line" {
t.Errorf("dot-stuffed line = %q, want '..Dot-leading line'", retrLines[3])
}
// 8. DELE 1
sendCmd("DELE 1")
deleResp := readLine()
if !strings.HasPrefix(deleResp, "+OK") {
t.Fatalf("DELE resp = %q", deleResp)
}
// 9. QUIT (commits expunge)
sendCmd("QUIT")
quitResp := readLine()
if !strings.HasPrefix(quitResp, "+OK") {
t.Fatalf("QUIT resp = %q", quitResp)
}
time.Sleep(50 * time.Millisecond)
if len(expunger.calls) != 1 || len(expunger.calls[0].uids) != 1 || expunger.calls[0].uids[0] != 101 {
t.Errorf("expunged calls = %+v, want UID 101", expunger.calls)
}
}
func TestPop3Server_Top_Rset_AuthFailures(t *testing.T) {
blobID := uuid.New()
msgContent := []byte("From: bob@example.test\r\nSubject: Top test\r\n\r\nLine 1\r\nLine 2\r\nLine 3\r\n")
messages := []port.MessageRecord{
{UID: 202, BlobID: blobID, SizeBytes: int64(len(msgContent))},
}
blobs := map[uuid.UUID][]byte{blobID: msgContent}
expunger := &fakeExpungeRepository{}
ln, cleanup := setupPop3Server(t, messages, blobs, expunger)
defer cleanup()
c, err := net.Dial("tcp", ln.Addr().String())
if err != nil {
t.Fatalf("dial: %v", err)
}
defer c.Close()
reader := bufio.NewReader(c)
writer := bufio.NewWriter(c)
readLine := func() string {
line, err := reader.ReadString('\n')
if err != nil {
t.Fatalf("read: %v", err)
}
return strings.TrimRight(line, "\r\n")
}
sendCmd := func(cmd string) {
writer.WriteString(cmd + "\r\n")
writer.Flush()
}
readLine() // Greeting
// Test wrong password
sendCmd("USER user@example.test")
readLine()
sendCmd("PASS wrong-password")
if l := readLine(); !strings.HasPrefix(l, "-ERR") {
t.Fatalf("wrong pass resp = %q, want -ERR", l)
}
// Login correctly
sendCmd("USER user@example.test")
readLine()
sendCmd("PASS correct horse battery staple")
readLine()
// TOP 1 2 (fetch headers + 2 body lines)
sendCmd("TOP 1 2")
topHead := readLine()
if !strings.HasPrefix(topHead, "+OK") {
t.Fatalf("TOP resp = %q", topHead)
}
var topLines []string
for {
l := readLine()
if l == "." {
break
}
topLines = append(topLines, l)
}
if len(topLines) < 4 {
t.Fatalf("topLines = %v", topLines)
}
// DELE then RSET
sendCmd("DELE 1")
readLine()
sendCmd("RSET")
if rsetResp := readLine(); !strings.HasPrefix(rsetResp, "+OK") {
t.Fatalf("RSET resp = %q", rsetResp)
}
// QUIT without deleting anything
sendCmd("QUIT")
readLine()
time.Sleep(50 * time.Millisecond)
if len(expunger.calls) != 0 {
t.Errorf("expected 0 expunge calls after RSET, got %d", len(expunger.calls))
}
}
@@ -0,0 +1,53 @@
package pop3presentation
import (
"net"
"lambdamail/protocols/internal/application/port"
appusecase "lambdamail/protocols/internal/application/usecase"
)
type Server struct {
addr string
useCase *appusecase.Pop3SessionUseCase
tlsProvider port.CertProvider
listener net.Listener
}
func NewServer(addr string, useCase *appusecase.Pop3SessionUseCase, tlsProvider port.CertProvider) *Server {
return &Server{
addr: addr,
useCase: useCase,
tlsProvider: tlsProvider,
}
}
func (s *Server) ListenAndServe() error {
ln, err := net.Listen("tcp", s.addr)
if err != nil {
return err
}
s.listener = ln
return s.Serve(ln)
}
func (s *Server) Serve(ln net.Listener) error {
s.listener = ln
for {
c, err := ln.Accept()
if err != nil {
return err
}
go func(rawConn net.Conn) {
connHandler := newConn(rawConn, s.useCase, s.tlsProvider)
connHandler.serve()
}(c)
}
}
func (s *Server) Close() error {
if s.listener != nil {
return s.listener.Close()
}
return nil
}