From 9ca97298bd115c42d7116b7f93a307f1937e2fd8 Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 24 Jul 2026 11:51:27 -0300 Subject: [PATCH] #7: implement POP3 protocol server (RFC 1939, RFC 2449, STLS) --- .../cmd/lambdamail-protocols/main.go | 11 + .../application/usecase/pop3_session.go | 86 ++++ .../application/usecase/pop3_session_test.go | 76 +++ .../protocols/internal/e2e/pop3_e2e_test.go | 239 ++++++++++ .../internal/presentation/pop3/conn.go | 436 ++++++++++++++++++ .../internal/presentation/pop3/conn_test.go | 337 ++++++++++++++ .../internal/presentation/pop3/server.go | 53 +++ 7 files changed, 1238 insertions(+) create mode 100644 services/protocols/internal/application/usecase/pop3_session.go create mode 100644 services/protocols/internal/application/usecase/pop3_session_test.go create mode 100644 services/protocols/internal/e2e/pop3_e2e_test.go create mode 100644 services/protocols/internal/presentation/pop3/conn.go create mode 100644 services/protocols/internal/presentation/pop3/conn_test.go create mode 100644 services/protocols/internal/presentation/pop3/server.go diff --git a/services/protocols/cmd/lambdamail-protocols/main.go b/services/protocols/cmd/lambdamail-protocols/main.go index 820f6eb..d275304 100644 --- a/services/protocols/cmd/lambdamail-protocols/main.go +++ b/services/protocols/cmd/lambdamail-protocols/main.go @@ -18,6 +18,7 @@ import ( "lambdamail/protocols/internal/infrastructure/postgres" tlsprovider "lambdamail/protocols/internal/infrastructure/tls" imappresentation "lambdamail/protocols/internal/presentation/imap" + pop3presentation "lambdamail/protocols/internal/presentation/pop3" smtppresentation "lambdamail/protocols/internal/presentation/smtp" ) @@ -111,6 +112,16 @@ func main() { } }() + pop3UseCase := usecase.NewPop3SessionUseCase(authRepo, imapFolders, messageQuery, blobReader, expungeRepo) + pop3Server := pop3presentation.NewServer(":110", pop3UseCase, certProvider) + + go func() { + log.Printf("lambdamail-protocols POP3 listening on :110") + if err := pop3Server.ListenAndServe(); err != nil { + log.Fatalf("pop3 serve: %v", err) + } + }() + handler := health.Handler(func() error { return pool.Ping(ctx) }) addr := ":8080" log.Printf("lambdamail-protocols health listening on %s", addr) diff --git a/services/protocols/internal/application/usecase/pop3_session.go b/services/protocols/internal/application/usecase/pop3_session.go new file mode 100644 index 0000000..83b8d36 --- /dev/null +++ b/services/protocols/internal/application/usecase/pop3_session.go @@ -0,0 +1,86 @@ +package usecase + +import ( + "context" + "fmt" + + "github.com/alexedwards/argon2id" + "github.com/google/uuid" + + "lambdamail/protocols/internal/application/port" +) + +type Pop3SessionUseCase struct { + authRepo port.AuthRepository + folderRepo port.ImapFolderRepository + messageQuery port.MessageQueryRepository + blobReader port.BlobReader + expungeRepo port.ExpungeRepository +} + +func NewPop3SessionUseCase( + authRepo port.AuthRepository, + folderRepo port.ImapFolderRepository, + messageQuery port.MessageQueryRepository, + blobReader port.BlobReader, + expungeRepo port.ExpungeRepository, +) *Pop3SessionUseCase { + return &Pop3SessionUseCase{ + authRepo: authRepo, + folderRepo: folderRepo, + messageQuery: messageQuery, + blobReader: blobReader, + expungeRepo: expungeRepo, + } +} + +func (uc *Pop3SessionUseCase) Login(ctx context.Context, username, password string) (string, error) { + rec, err := uc.authRepo.FindByAddress(ctx, username) + if err != nil { + return "", fmt.Errorf("lookup mailbox %q: %w", username, err) + } + if rec == nil { + return "", ErrAuthFailed + } + match, err := argon2id.ComparePasswordAndHash(password, rec.PasswordHash) + if err != nil { + return "", fmt.Errorf("verify password hash: %w", err) + } + if !match { + return "", ErrAuthFailed + } + return rec.ID.String(), nil +} + +func (uc *Pop3SessionUseCase) GetInbox(ctx context.Context, mailboxID string) (*port.ImapFolderRecord, []port.MessageRecord, error) { + folder, err := uc.folderRepo.FindByName(ctx, mailboxID, "INBOX") + if err != nil { + return nil, nil, fmt.Errorf("find INBOX for mailbox %s: %w", mailboxID, err) + } + if folder == nil { + return nil, nil, ErrNoSuchFolder + } + messages, err := uc.messageQuery.ListMessages(ctx, folder.ID) + if err != nil { + return nil, nil, fmt.Errorf("list messages for INBOX folder %s: %w", folder.ID, err) + } + return folder, messages, nil +} + +func (uc *Pop3SessionUseCase) ReadBlob(ctx context.Context, blobID uuid.UUID) ([]byte, error) { + content, err := uc.blobReader.ReadByID(ctx, blobID) + if err != nil { + return nil, fmt.Errorf("read blob %s: %w", blobID, err) + } + return content, nil +} + +func (uc *Pop3SessionUseCase) ExpungeMessages(ctx context.Context, folderID string, uids []uint32) error { + if len(uids) == 0 { + return nil + } + if err := uc.expungeRepo.Expunge(ctx, folderID, uids); err != nil { + return fmt.Errorf("expunge messages in folder %s: %w", folderID, err) + } + return nil +} diff --git a/services/protocols/internal/application/usecase/pop3_session_test.go b/services/protocols/internal/application/usecase/pop3_session_test.go new file mode 100644 index 0000000..8d0b4f3 --- /dev/null +++ b/services/protocols/internal/application/usecase/pop3_session_test.go @@ -0,0 +1,76 @@ +package usecase + +import ( + "context" + "errors" + "testing" + + "github.com/google/uuid" + "lambdamail/protocols/internal/application/port" +) + +func TestPop3SessionUseCase_Login_SucceedsForCorrectPassword(t *testing.T) { + mailboxID := uuid.New() + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, + }} + uc := NewPop3SessionUseCase(auth, nil, nil, nil, nil) + + gotID, err := uc.Login(context.Background(), "user@example.test", testPassword) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if gotID != mailboxID.String() { + t.Errorf("mailboxID = %q, want %q", gotID, mailboxID.String()) + } +} + +func TestPop3SessionUseCase_Login_FailsForWrongPassword(t *testing.T) { + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: uuid.New(), PasswordHash: testPasswordHash}, + }} + uc := NewPop3SessionUseCase(auth, nil, nil, nil, nil) + + _, err := uc.Login(context.Background(), "user@example.test", "wrong-password") + if !errors.Is(err, ErrAuthFailed) { + t.Fatalf("error = %v, want ErrAuthFailed", err) + } +} + +type fakeMessageQueryRepository struct { + messages []port.MessageRecord +} + +func (f *fakeMessageQueryRepository) ListMessages(_ context.Context, _ string) ([]port.MessageRecord, error) { + return f.messages, nil +} + +func TestPop3SessionUseCase_GetInbox_ReturnsFolderAndMessages(t *testing.T) { + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{ + "INBOX": {ID: "folder-1", Name: "INBOX", NumMessages: 1}, + }} + messages := &fakeMessageQueryRepository{messages: []port.MessageRecord{ + {UID: 1, SizeBytes: 100}, + }} + uc := NewPop3SessionUseCase(nil, folders, messages, nil, nil) + + folder, recs, err := uc.GetInbox(context.Background(), "mailbox-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if folder.ID != "folder-1" || len(recs) != 1 || recs[0].UID != 1 { + t.Errorf("unexpected inbox result: folder=%+v, recs=%+v", folder, recs) + } +} + +func TestPop3SessionUseCase_ExpungeMessages_DelegatesToRepo(t *testing.T) { + expunger := &fakeExpungeRepository{} + uc := NewPop3SessionUseCase(nil, nil, nil, nil, expunger) + + if err := uc.ExpungeMessages(context.Background(), "folder-1", []uint32{1, 2}); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(expunger.calls) != 1 || expunger.calls[0].folderID != "folder-1" { + t.Fatalf("unexpected expunger calls: %+v", expunger.calls) + } +} diff --git a/services/protocols/internal/e2e/pop3_e2e_test.go b/services/protocols/internal/e2e/pop3_e2e_test.go new file mode 100644 index 0000000..3d8a87a --- /dev/null +++ b/services/protocols/internal/e2e/pop3_e2e_test.go @@ -0,0 +1,239 @@ +package e2e + +import ( + "bufio" + "bytes" + "context" + "net" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/alexedwards/argon2id" + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + + appusecase "lambdamail/protocols/internal/application/usecase" + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/infrastructure/diskstorage" + "lambdamail/protocols/internal/infrastructure/postgres" + pop3presentation "lambdamail/protocols/internal/presentation/pop3" +) + +func TestPop3EndToEnd(t *testing.T) { + if testing.Short() { + t.Skip("e2e test skipped in -short mode") + } + ctx := context.Background() + + runtimeDir := t.TempDir() + pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig(). + Port(54332). // distinct port for POP3 e2e test execution + RuntimePath(runtimeDir). + StartTimeout(120 * time.Second)) + if err := pg.Start(); err != nil { + t.Fatalf("start embedded postgres: %v", err) + } + defer pg.Stop() + + dbURL := "postgres://postgres:postgres@localhost:54332/postgres?sslmode=disable" + pool, err := postgres.NewPool(ctx, dbURL) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer pool.Close() + + root := repoRoot(t) + sql, err := os.ReadFile(filepath.Join(root, "migrations", "0001_init_schema.up.sql")) + if err != nil { + t.Fatalf("read migration: %v", err) + } + if _, err := pool.Exec(ctx, string(sql)); err != nil { + t.Fatalf("apply migration: %v", err) + } + + // --- Seed domain, mailbox, INBOX ----------------------------------------- + password := "correct horse battery staple" + hash, err := argon2id.CreateHash(password, argon2id.DefaultParams) + if err != nil { + t.Fatalf("hash password: %v", err) + } + domainID := "00000000-0000-0000-0000-000000000001" + mailboxID := "00000000-0000-0000-0000-000000000002" + folderID := "00000000-0000-0000-0000-000000000003" + if _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, 'example.test', 'example.test')`, domainID); err != nil { + t.Fatalf("seed domain: %v", err) + } + if _, err := pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', 'user@example.test', $3)`, mailboxID, domainID, hash); err != nil { + t.Fatalf("seed mailbox: %v", err) + } + if _, err := pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'INBOX', 'inbox')`, folderID, mailboxID); err != nil { + t.Fatalf("seed folder: %v", err) + } + + // Deliver 1 email via SMTP use case + spool := t.TempDir() + mailboxRepo := postgres.NewMailboxRepository(pool) + inboundUC := appusecase.NewProcessInboundEmailUseCase( + mailboxRepo, + diskstorage.NewLocalDiskBlobStorage(pool, spool), + postgres.NewInboundMessageRepository(pool), + ) + + recipientRecord, err := mailboxRepo.FindActiveByAddress(ctx, "user@example.test") + if err != nil || recipientRecord == nil { + t.Fatalf("find recipient record: %v", err) + } + testMessage := "From: sender@remote.example\r\nTo: user@example.test\r\nSubject: pop3 test\r\n\r\n.Dot line test\r\nPOP3 body content.\r\n" + if err := inboundUC.Handle(ctx, appusecase.ProcessInboundEmailInput{ + Sender: "sender@remote.example", + Recipients: []port.MailboxRecord{*recipientRecord}, + RecipientAddresses: []string{"user@example.test"}, + Body: bytes.NewBufferString(testMessage), + }); err != nil { + t.Fatalf("seed message: %v", err) + } + + // Setup POP3 Server + pop3UC := appusecase.NewPop3SessionUseCase( + postgres.NewAuthRepository(pool), + postgres.NewImapFolderRepository(pool), + postgres.NewMessageQueryRepository(pool), + diskstorage.NewLocalDiskBlobReader(pool), + postgres.NewExpungeRepository(pool), + ) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + + srv := pop3presentation.NewServer(ln.Addr().String(), pop3UC, nil) + go srv.Serve(ln) + defer srv.Close() + + // Dial POP3 server + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial POP3: %v", err) + } + defer conn.Close() + + reader := bufio.NewReader(conn) + writer := bufio.NewWriter(conn) + + readLine := func() string { + line, err := reader.ReadString('\n') + if err != nil { + t.Fatalf("read line: %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("expected greeting +OK, got %q", greeting) + } + + // 2. CAPA + sendCmd("CAPA") + if capaHead := readLine(); !strings.HasPrefix(capaHead, "+OK") { + t.Fatalf("CAPA head = %q", capaHead) + } + for { + if l := readLine(); l == "." { + break + } + } + + // 3. USER / PASS + sendCmd("USER user@example.test") + if userResp := readLine(); !strings.HasPrefix(userResp, "+OK") { + t.Fatalf("USER resp = %q", userResp) + } + + sendCmd("PASS correct horse battery staple") + if passResp := readLine(); !strings.HasPrefix(passResp, "+OK") { + t.Fatalf("PASS resp = %q", passResp) + } + + // 4. STAT + sendCmd("STAT") + statResp := readLine() + if !strings.HasPrefix(statResp, "+OK 1 ") { + t.Fatalf("STAT expected 1 message, got %q", statResp) + } + + // 5. LIST + sendCmd("LIST 1") + listResp := readLine() + if !strings.HasPrefix(listResp, "+OK 1 ") { + t.Fatalf("LIST 1 expected +OK 1, got %q", listResp) + } + + // 6. UIDL + sendCmd("UIDL 1") + uidlResp := readLine() + if !strings.HasPrefix(uidlResp, "+OK 1 ") { + t.Fatalf("UIDL 1 expected +OK 1, got %q", uidlResp) + } + + // 7. TOP 1 1 + sendCmd("TOP 1 1") + if topHead := readLine(); !strings.HasPrefix(topHead, "+OK") { + t.Fatalf("TOP 1 1 head = %q", topHead) + } + for { + if l := readLine(); l == "." { + break + } + } + + // 8. RETR 1 + sendCmd("RETR 1") + if retrHead := readLine(); !strings.HasPrefix(retrHead, "+OK") { + t.Fatalf("RETR 1 head = %q", retrHead) + } + var retrBody []string + for { + l := readLine() + if l == "." { + break + } + retrBody = append(retrBody, l) + } + if len(retrBody) == 0 { + t.Fatal("expected non-empty RETR body") + } + + // 9. DELE 1 + sendCmd("DELE 1") + if deleResp := readLine(); !strings.HasPrefix(deleResp, "+OK") { + t.Fatalf("DELE 1 resp = %q", deleResp) + } + + // 10. QUIT (commits expunge to database) + sendCmd("QUIT") + if quitResp := readLine(); !strings.HasPrefix(quitResp, "+OK") { + t.Fatalf("QUIT resp = %q", quitResp) + } + + time.Sleep(100 * time.Millisecond) + + // Verify in Postgres that the message was marked expunged + var count int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM email_messages WHERE folder_id = $1 AND expunged_at IS NOT NULL`, folderID).Scan(&count); err != nil { + t.Fatalf("query expunged_at count: %v", err) + } + if count != 1 { + t.Fatalf("expected 1 expunged message in DB, got %d", count) + } +} diff --git a/services/protocols/internal/presentation/pop3/conn.go b/services/protocols/internal/presentation/pop3/conn.go new file mode 100644 index 0000000..852e1e4 --- /dev/null +++ b/services/protocols/internal/presentation/pop3/conn.go @@ -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") + } +} diff --git a/services/protocols/internal/presentation/pop3/conn_test.go b/services/protocols/internal/presentation/pop3/conn_test.go new file mode 100644 index 0000000..27bf557 --- /dev/null +++ b/services/protocols/internal/presentation/pop3/conn_test.go @@ -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)) + } +} diff --git a/services/protocols/internal/presentation/pop3/server.go b/services/protocols/internal/presentation/pop3/server.go new file mode 100644 index 0000000..0344f4d --- /dev/null +++ b/services/protocols/internal/presentation/pop3/server.go @@ -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 +}