From 6a9dcd0285c08813a2aa3a4a2dfb9c34087eb1a8 Mon Sep 17 00:00:00 2001 From: kevin Date: Wed, 19 Aug 2026 19:56:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(imap):=20=E5=A4=9A=E5=AE=A2=E6=88=B7?= =?UTF-8?q?=E7=AB=AF=E7=8A=B6=E6=80=81=E5=AE=9E=E6=97=B6=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=20+=20=E5=BD=93=E5=89=8D=E8=BF=9E=E6=8E=A5=E3=80=8C=E6=96=AD?= =?UTF-8?q?=E5=BC=80=E5=B9=B6=E5=B0=81=E7=A6=81=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 推送扩展:IMAP STORE(已读/星标/\Deleted)推送 FETCH 标志更新、 EXPUNGE 推送 ExpungeUpdate(删除前序号)、APPEND/COPY/MOVE 推送 新邮件;POP3 QUIT 删除、Web 标已读/删除同样实时同步到 IMAP 客户端 - Pusher 接口统一 SMTP/POP3/Web 的推送入口,IMAP 内部操作经会话 通道直接入队(非阻塞,满则丢弃) - 当前连接页新增「断开并封禁」:connhub 支持断开回调,SMTP/POP3 关底层连接、IMAP 经 ForEachConn 按地址断开;一键封禁 180 天并 断开该 IP 全部在线连接,黑名单页可随时解封 - 修复:POP3 PASS 成功后保留完整邮箱(此前被裸用户名覆盖) - 新增测试:断开/按 IP 断开、flags/expunge 推送内容、POP3 删除推送、 Web 断开封禁处理器;全量 -race 通过 --- README.md | 4 +- internal/connhub/hub.go | 60 ++++++++ internal/connhub/hub_test.go | 62 +++++++++ internal/imap_server/backend.go | 131 +++++++++++++++--- internal/imap_server/notify_test.go | 125 +++++++++++++++-- internal/imap_server/server.go | 89 ++++++++++-- internal/pop3_server/server.go | 23 ++- internal/pop3_server/server_test.go | 87 ++++++++++++ internal/smtp_server/server.go | 20 +-- internal/web/handlers/admin.go | 36 +++++ internal/web/handlers/connections_test.go | 101 ++++++++++++++ internal/web/handlers/mail.go | 48 ++++++- internal/web/server.go | 13 +- internal/web/templates/admin/connections.html | 9 +- main.go | 6 +- 15 files changed, 745 insertions(+), 69 deletions(-) create mode 100644 internal/web/handlers/connections_test.go diff --git a/README.md b/README.md index 0a90132..6214b95 100644 --- a/README.md +++ b/README.md @@ -10,8 +10,8 @@ Web 前端采用 QQ 邮箱风格的布局:顶部导航 + 左侧文件夹栏 + - **Web 邮箱**:QQ 邮箱风格界面,支持收件箱 / 已发送 / 草稿箱、未读角标与搜索过滤、全选 / 批量删除、发件人头像、富文本编辑(Quill.js)、附件上传/下载 - **管理后台**:域名管理、用户管理、DKIM 密钥自动生成、DNS 配置提示、全量邮件查看、外发队列管理、IP 封禁管理、仪表盘统计 - **协议调用日志**:SMTP / IMAP / POP3 每次连接自动记录来源 IP、用户名、成功/失败、失败原因与操作摘要,可按协议/状态/IP/用户名/时间筛选,用于分析密码爆破、中继滥用等攻击行为(默认保留 30 天,自动清理) -- **IMAP 新邮件推送**:本地投递(SMTP/Web 写信)成功后实时推送,挂起 IDLE 的客户端即时收到新邮件通知(无需轮询) -- **当前连接监控**:管理后台实时查看 SMTP/IMAP/POP3 活动连接(来源 IP、用户名、TLS、时长),每 5 秒自动刷新 +- **IMAP 新邮件推送**:本地投递(SMTP/Web 写信)成功后实时推送,挂起 IDLE 的客户端即时收到新邮件通知(无需轮询);其他客户端造成的已读/星标/删除变化也实时同步(IMAP STORE/EXPUNGE、POP3 删除、Web 标已读/删除) +- **当前连接监控**:管理后台实时查看 SMTP/IMAP/POP3 活动连接(来源 IP、用户名、TLS、时长),每 5 秒自动刷新;支持「断开并封禁」一键封禁该 IP 全部在线连接(封禁 180 天,可随时解封) - **外部认证**:OAuth2(Google / GitHub)、LDAP(可选,默认关闭) - **安全机制**:BCrypt 密码哈希、登录失败自动封禁 IP(**阶段性封禁**:前 3 次达到失败阈值只计数,第 4 次起封禁并按档位递增 30分钟 → 3小时 → 3个月 → 半年,成功登录清零)、外发频率限制(防滥用)、非认证禁止中继(防开放中继)、管理员可解封 - **多数据库**:默认 SQLite,可切换 MySQL diff --git a/internal/connhub/hub.go b/internal/connhub/hub.go index affe599..c4521ff 100644 --- a/internal/connhub/hub.go +++ b/internal/connhub/hub.go @@ -20,6 +20,9 @@ type Conn struct { LastActive time.Time hub *Hub + // disconnect 强制断开底层连接的回调(由各协议服务器注册)。 + // 关闭底层 socket 后协议服务器会正常走收尾清理(Logout/注销)。 + disconnect func() } // Hub 管理所有活动连接(同一把锁保护注册表与连接字段)。 @@ -88,6 +91,16 @@ func (c *Conn) Touch() { c.hub.mu.Unlock() } +// SetDisconnect 注册强制断开底层连接的回调(管理后台「断开并封禁」用)。 +func (c *Conn) SetDisconnect(fn func()) { + if c == nil || c.hub == nil { + return + } + c.hub.mu.Lock() + c.disconnect = fn + c.hub.mu.Unlock() +} + // Close 从注册中心移除该连接。 func (c *Conn) Close() { if c == nil || c.hub == nil { @@ -98,6 +111,53 @@ func (c *Conn) Close() { c.hub.mu.Unlock() } +// Get 按 ID 查找活动连接。 +func (h *Hub) Get(id uint64) (*Conn, bool) { + if h == nil { + return nil, false + } + h.mu.Lock() + defer h.mu.Unlock() + c, ok := h.conns[id] + return c, ok +} + +// Disconnect 强制断开指定连接(关闭底层连接并注销)。 +func (h *Hub) Disconnect(id uint64) bool { + c, ok := h.Get(id) + if !ok { + return false + } + h.mu.Lock() + fn := c.disconnect + h.mu.Unlock() + if fn != nil { + fn() + } + return true +} + +// DisconnectByIP 强制断开该 IP 的全部连接,返回断开的连接数。 +// 用于封禁 IP 后立即踢掉其所有在线会话。 +func (h *Hub) DisconnectByIP(ip string) int { + if h == nil || ip == "" { + return 0 + } + h.mu.Lock() + var fns []func() + for _, c := range h.conns { + if c.IP == ip && c.disconnect != nil { + fns = append(fns, c.disconnect) + } + } + h.mu.Unlock() + + for _, fn := range fns { + fn() + } + return len(fns) +} + // List 返回当前所有活动连接(拷贝),按连接时间升序。 func (h *Hub) List() []Conn { if h == nil { diff --git a/internal/connhub/hub_test.go b/internal/connhub/hub_test.go index ac10aa7..fea271c 100644 --- a/internal/connhub/hub_test.go +++ b/internal/connhub/hub_test.go @@ -2,6 +2,7 @@ package connhub import ( "sync" + "sync/atomic" "testing" "time" ) @@ -113,3 +114,64 @@ func TestListOrderedByID(t *testing.T) { } } } + +// TestDisconnect 验证强制断开回调被调用。 +func TestDisconnect(t *testing.T) { + h := New() + var closed atomic.Int32 + + c1 := h.Register("smtp", "10.0.0.1", 25, false) + c1.SetDisconnect(func() { closed.Add(1) }) + c2 := h.Register("imap", "10.0.0.2", 993, true) + c2.SetDisconnect(func() { closed.Add(1) }) + + if !h.Disconnect(c1.ID) { + t.Fatal("Disconnect should report success") + } + if closed.Load() != 1 { + t.Fatalf("closed = %d, want 1", closed.Load()) + } + // 已断开(未注销)仍可查到 + if _, ok := h.Get(c1.ID); !ok { + t.Fatal("conn should still be registered until Close") + } + // 不存在的 ID + if h.Disconnect(99999) { + t.Fatal("Disconnect of unknown id must fail") + } +} + +// TestDisconnectByIP 验证封禁时断开该 IP 全部连接。 +func TestDisconnectByIP(t *testing.T) { + h := New() + var closed atomic.Int32 + + // 同一 IP 三个协议连接 + for _, proto := range []string{"smtp", "imap", "pop3"} { + c := h.Register(proto, "203.0.113.5", 25, false) + c.SetDisconnect(func() { closed.Add(1) }) + } + // 另一 IP 不受影响 + other := h.Register("smtp", "203.0.113.6", 25, false) + other.SetDisconnect(func() { closed.Add(1) }) + + n := h.DisconnectByIP("203.0.113.5") + if n != 3 { + t.Fatalf("disconnected = %d, want 3", n) + } + if closed.Load() != 3 { + t.Fatalf("closed = %d, want 3", closed.Load()) + } +} + +// TestDisconnectNoCallback 验证未注册断开回调的连接安全跳过。 +func TestDisconnectNoCallback(t *testing.T) { + h := New() + c := h.Register("pop3", "10.0.0.9", 110, false) + if !h.Disconnect(c.ID) { + t.Fatal("Disconnect should report success even without callback") + } + if n := h.DisconnectByIP("10.0.0.9"); n != 0 { + t.Fatalf("disconnected = %d, want 0", n) + } +} diff --git a/internal/imap_server/backend.go b/internal/imap_server/backend.go index 99b3527..6a2efcc 100644 --- a/internal/imap_server/backend.go +++ b/internal/imap_server/backend.go @@ -34,6 +34,8 @@ type imapBackend struct { // updates 承载新邮件等后端更新,由 go-imap 服务器广播给相关客户端。 updates chan backend.Update + // disconnectAddr 强制断开指定远端地址的连接(管理后台断开封禁用)。 + disconnectAddr func(addr string) } // Updates 实现 backend.BackendUpdater:新邮件推送通道(广播按用户名与 @@ -42,28 +44,20 @@ func (b *imapBackend) Updates() <-chan backend.Update { return b.updates } -// buildNewMessageUpdate 为一条新投递到 INBOX 的邮件构造 IMAP 更新。 -// seq 取该邮件在 INBOX 中的序号(通知时机在入库之后,取当前 INBOX 长度)。 -func buildNewMessageUpdate(stores *store.Stores, userEmail string, msg *db.Message) *backend.MessageUpdate { - if stores == nil || msg == nil || userEmail == "" { +// buildNewMessageUpdate 为一条新投递到 mailbox 的邮件构造 IMAP 更新。 +// seq 取该邮件在邮箱中的序号(通知时机在入库之后,取当前列表长度)。 +func buildNewMessageUpdate(stores *store.Stores, userEmail, mailbox string, msg *db.Message) *backend.MessageUpdate { + if stores == nil || msg == nil || userEmail == "" || mailbox == "" { return nil } seq := uint32(1) - if msgs, err := stores.Mails.ListAllByUserAndFolder(msg.UserID, "INBOX"); err == nil { + if msgs, err := stores.Mails.ListAllByUserAndFolder(msg.UserID, mailbox); err == nil { seq = uint32(len(msgs)) } - flags := make([]string, 0, 2) - if msg.IsRead { - flags = append(flags, "\\Seen") - } - if msg.IsFlagged { - flags = append(flags, "\\Flagged") - } - imapMsg := imap.NewMessage(seq, []imap.FetchItem{imap.FetchUid, imap.FetchFlags, imap.FetchInternalDate, imap.FetchRFC822Size, imap.FetchEnvelope}) imapMsg.Uid = uint32(msg.ID) - imapMsg.Flags = flags + imapMsg.Flags = flagsOf(msg.IsRead, msg.IsFlagged, false) imapMsg.InternalDate = msg.Date imapMsg.Size = uint32(len(msg.RawData)) imapMsg.Envelope = &imap.Envelope{ @@ -78,11 +72,68 @@ func buildNewMessageUpdate(stores *store.Stores, userEmail string, msg *db.Messa } return &backend.MessageUpdate{ - Update: backend.NewUpdate(userEmail, "INBOX"), + Update: backend.NewUpdate(userEmail, mailbox), Message: imapMsg, } } +// buildFlagsUpdate 为一条消息的标志变化构造 IMAP 更新(已读/星标/删除标记)。 +// deleted 为会话内 \Deleted 标记(IMAP STORE 会话状态,不入库)。 +func buildFlagsUpdate(stores *store.Stores, userEmail, mailbox string, msg *db.Message, deleted bool) *backend.MessageUpdate { + if stores == nil || msg == nil || userEmail == "" || mailbox == "" { + return nil + } + imapMsg := imap.NewMessage(seqOf(stores, msg.UserID, mailbox, msg.ID), + []imap.FetchItem{imap.FetchUid, imap.FetchFlags}) + imapMsg.Uid = uint32(msg.ID) + imapMsg.Flags = flagsOf(msg.IsRead, msg.IsFlagged, deleted) + return &backend.MessageUpdate{ + Update: backend.NewUpdate(userEmail, mailbox), + Message: imapMsg, + } +} + +// seqOf 返回消息在文件夹中的序号(1 基),未找到返回 0。 +func seqOf(stores *store.Stores, userID uint, mailbox string, msgID uint) uint32 { + msgs, err := stores.Mails.ListAllByUserAndFolder(userID, mailbox) + if err != nil { + return 0 + } + for i := range msgs { + if msgs[i].ID == msgID { + return uint32(i + 1) + } + } + return 0 +} + +// flagsOf 按数据库状态生成 IMAP 标志列表(deleted 为会话内 \Deleted 标记)。 +func flagsOf(read, flagged, deleted bool) []string { + flags := make([]string, 0, 3) + if read { + flags = append(flags, "\\Seen") + } + if flagged { + flags = append(flags, "\\Flagged") + } + if deleted { + flags = append(flags, "\\Deleted") + } + return flags +} + +// pushUpdate 非阻塞地把一条后端更新送入推送通道(满则丢弃并记日志)。 +func pushUpdate(ch chan backend.Update, u backend.Update) { + if ch == nil { + return + } + select { + case ch <- u: + default: + log.Printf("IMAP: 推送通道已满,丢弃更新") + } +} + // Login authenticates a user by email and password. func (b *imapBackend) Login(connInfo *imap.ConnInfo, username, password string) (backend.User, error) { clientIP := store.ClientIPFromAddr(connInfo.RemoteAddr) @@ -110,10 +161,19 @@ func (b *imapBackend) Login(connInfo *imap.ConnInfo, username, password string) logID := b.recordLogin(clientIP, username, true, "", "LOGIN 成功", 0, now) - // 连接追踪:注册到当前连接中心,Logout 时注销 + // 连接追踪:注册到当前连接中心,Logout 时注销; + // 注册强制断开回调(按远端地址匹配,断开时由 go-imap 走正常收尾)。 conn := b.hub.Register("imap", clientIP, b.port, connInfo.TLS != nil) if conn != nil { conn.SetUser(email) + remoteAddr := "" + if connInfo.RemoteAddr != nil { + remoteAddr = connInfo.RemoteAddr.String() + } + if b.disconnectAddr != nil && remoteAddr != "" { + addr := remoteAddr + conn.SetDisconnect(func() { b.disconnectAddr(addr) }) + } } return &imapUser{ @@ -124,6 +184,7 @@ func (b *imapBackend) Login(connInfo *imap.ConnInfo, username, password string) clientIP: clientIP, startedAt: now, conn: conn, + updates: b.updates, }, nil } @@ -158,6 +219,8 @@ type imapUser struct { clientIP string startedAt time.Time conn *connhub.Conn + // updates 所在 backend 的推送通道(STORE/EXPUNGE 等实时同步用)。 + updates chan backend.Update } // Username returns the user's email address. @@ -671,6 +734,9 @@ func (m *imapMailbox) CreateMessage(flags []string, date time.Time, body imap.Li return fmt.Errorf("failed to create message: %w", err) } + // 新邮件(IMAP APPEND)→ 推送给同用户其他已选中该邮箱的客户端 + pushUpdate(m.user.updates, buildNewMessageUpdate(m.stores, m.user.email, m.name, msg)) + return nil } @@ -730,6 +796,15 @@ func (m *imapMailbox) UpdateMessagesFlags(uid bool, seqset *imap.SeqSet, op imap applyFlag(flag, false) } } + + // 标志变化(已读/星标/删除)→ 推送给同用户其他客户端 + // (重新读库取最新状态,\Deleted 取会话内状态) + fresh, err := m.stores.Mails.GetByID(dbMsg.ID) + if err != nil { + continue + } + deleted := m.deleted != nil && m.deleted[dbMsg.ID] + pushUpdate(m.user.updates, buildFlagsUpdate(m.stores, m.user.email, m.name, fresh, deleted)) } return nil @@ -776,7 +851,10 @@ func (m *imapMailbox) CopyMessages(uid bool, seqset *imap.SeqSet, dest string) e } if err := m.stores.Mails.Create(copyMsg); err != nil { log.Printf("IMAP: failed to copy message %d to %s: %v", dbMsg.ID, dest, err) + continue } + // 目标邮箱新增 → 推送给同用户其他客户端 + pushUpdate(m.user.updates, buildNewMessageUpdate(m.stores, m.user.email, dest, copyMsg)) } return nil @@ -806,6 +884,11 @@ func (m *imapMailbox) MoveMessages(uid bool, seqset *imap.SeqSet, dest string) e } if err := m.stores.Mails.MoveToFolder(dbMsg.ID, dest); err != nil { log.Printf("IMAP: failed to move message %d to %s: %v", dbMsg.ID, dest, err) + continue + } + // 目标邮箱新增(移动后消息在 dest)→ 推送给同用户其他客户端 + if moved, err := m.stores.Mails.GetByID(dbMsg.ID); err == nil { + pushUpdate(m.user.updates, buildNewMessageUpdate(m.stores, m.user.email, dest, moved)) } } return nil @@ -817,12 +900,28 @@ func (m *imapMailbox) Expunge() error { return nil } + // 删除前计算各消息的序号(Expunge 响应序号为删除前状态下的序号) + var seqs []uint32 + for msgID := range m.deleted { + if seq := seqOf(m.stores, m.user.id, m.name, msgID); seq > 0 { + seqs = append(seqs, seq) + } + } + for msgID := range m.deleted { if err := m.stores.Mails.Delete(msgID); err != nil { log.Printf("IMAP: failed to expunge message %d: %v", msgID, err) } } m.deleted = make(map[uint]bool) + + // 删除 → 推送给同用户其他客户端(每条序号一个 ExpungeUpdate) + for _, seq := range seqs { + pushUpdate(m.user.updates, &backend.ExpungeUpdate{ + Update: backend.NewUpdate(m.user.email, m.name), + SeqNum: seq, + }) + } return nil } diff --git a/internal/imap_server/notify_test.go b/internal/imap_server/notify_test.go index 992f1c2..b6e0081 100644 --- a/internal/imap_server/notify_test.go +++ b/internal/imap_server/notify_test.go @@ -15,8 +15,8 @@ import ( "gorm.io/gorm" ) -// TestNotifyNewMessage 验证本地投递成功后推送的 MessageUpdate 内容正确。 -func TestNotifyNewMessage(t *testing.T) { +// TestPushNewMessage 验证本地投递成功后推送的 MessageUpdate 内容正确。 +func TestPushNewMessage(t *testing.T) { gdb, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "test.db")), &gorm.Config{}) if err != nil { t.Fatalf("open sqlite: %v", err) @@ -61,7 +61,7 @@ func TestNotifyNewMessage(t *testing.T) { // 模拟明文 + TLS 两个监听器(生产环境由 Start/StartTLS 注册) srv.newServer("127.0.0.1:143", nil) srv.newServer("127.0.0.1:993", nil) - srv.NotifyNewMessage(email, inboxMsg) + srv.PushNewMessage(email, inboxMsg) // 两个监听器(明文/TLS)各有一个 backend 通道,都应收到同一更新 srv.beMu.Lock() @@ -99,8 +99,8 @@ func TestNotifyNewMessage(t *testing.T) { } } -// TestNotifyNewMessageChannelFull 验证通道满时推送不阻塞(非阻塞丢弃)。 -func TestNotifyNewMessageChannelFull(t *testing.T) { +// TestPushNewMessageChannelFull 验证通道满时推送不阻塞(非阻塞丢弃)。 +func TestPushNewMessageChannelFull(t *testing.T) { gdb, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "test.db")), &gorm.Config{}) if err != nil { t.Fatalf("open sqlite: %v", err) @@ -127,22 +127,123 @@ func TestNotifyNewMessageChannelFull(t *testing.T) { b.updates <- backend.NewUpdate("a@b", "INBOX") } } - srv.NotifyNewMessage("a@b", msg) + srv.PushNewMessage("a@b", msg) close(done) }() select { case <-done: case <-time.After(time.Second): - t.Fatal("NotifyNewMessage blocked on full channel") + t.Fatal("PushNewMessage blocked on full channel") } } -// TestNotifyNewMessageNilSafe 验证空参数/空指针安全。 -func TestNotifyNewMessageNilSafe(t *testing.T) { +// TestPushNewMessageNilSafe 验证空参数/空指针安全。 +func TestPushNewMessageNilSafe(t *testing.T) { var srv *IMAPServer - srv.NotifyNewMessage("a@b", &db.Message{ID: 1}) // 不应 panic + srv.PushNewMessage("a@b", &db.Message{ID: 1}) // 不应 panic srv = NewIMAPServer(config.IMAPConfig{}, nil, nil, config.BanConfig{}, nil) - srv.NotifyNewMessage("", &db.Message{ID: 1}) // 空邮箱 - srv.NotifyNewMessage("a@b", nil) // 空消息 + srv.PushNewMessage("", &db.Message{ID: 1}) // 空邮箱 + srv.PushNewMessage("a@b", nil) // 空消息 +} + +// TestPushFlagsChanged 验证标志变化(已读/星标)推送内容正确。 +func TestPushFlagsChanged(t *testing.T) { + gdb, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "test.db")), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + if err := gdb.AutoMigrate(&db.User{}, &db.Domain{}, &db.Message{}); err != nil { + t.Fatalf("migrate: %v", err) + } + stores := store.NewStores(gdb) + + domain := &db.Domain{Name: "example.com"} + if err := stores.Domains.Create(domain); err != nil { + t.Fatalf("create domain: %v", err) + } + user := &db.User{Username: "alice", DomainID: domain.ID, IsActive: true} + if err := stores.Users.Create(user); err != nil { + t.Fatalf("create user: %v", err) + } + + msg := &db.Message{UserID: user.ID, Folder: "INBOX", FromAddr: "x@y", Subject: "s", Date: time.Now()} + if err := stores.Mails.Create(msg); err != nil { + t.Fatalf("create message: %v", err) + } + msg.IsRead = true + msg.IsFlagged = true + + hub := connhub.New() + srv := NewIMAPServer(config.IMAPConfig{}, stores, nil, config.BanConfig{}, hub) + srv.newServer("127.0.0.1:143", nil) + srv.PushFlagsChanged("alice@example.com", "INBOX", msg) + + srv.beMu.Lock() + b := srv.bes[0] + srv.beMu.Unlock() + + select { + case upd := <-b.updates: + mu, ok := upd.(*backend.MessageUpdate) + if !ok { + t.Fatalf("update type = %T, want *MessageUpdate", upd) + } + if mu.Username() != "alice@example.com" || mu.Mailbox() != "INBOX" { + t.Fatalf("update targeting = %s/%s", mu.Username(), mu.Mailbox()) + } + if mu.Message.Uid != uint32(msg.ID) { + t.Fatalf("uid = %d, want %d", mu.Message.Uid, msg.ID) + } + got := make(map[string]bool) + for _, f := range mu.Message.Flags { + got[f] = true + } + if !got["\\Seen"] || !got["\\Flagged"] { + t.Fatalf("flags = %v, want \\Seen and \\Flagged", mu.Message.Flags) + } + case <-time.After(time.Second): + t.Fatal("no flags update received") + } +} + +// TestPushExpunged 验证删除推送:每条序号一个 ExpungeUpdate。 +func TestPushExpunged(t *testing.T) { + gdb, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "test.db")), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + if err := gdb.AutoMigrate(&db.Message{}); err != nil { + t.Fatalf("migrate: %v", err) + } + stores := store.NewStores(gdb) + + hub := connhub.New() + srv := NewIMAPServer(config.IMAPConfig{}, stores, nil, config.BanConfig{}, hub) + srv.newServer("127.0.0.1:143", nil) + srv.PushExpunged("alice@example.com", "INBOX", []uint32{2, 5}) + + srv.beMu.Lock() + b := srv.bes[0] + srv.beMu.Unlock() + + var seqs []uint32 + for i := 0; i < 2; i++ { + select { + case upd := <-b.updates: + eu, ok := upd.(*backend.ExpungeUpdate) + if !ok { + t.Fatalf("update type = %T, want *ExpungeUpdate", upd) + } + if eu.Username() != "alice@example.com" || eu.Mailbox() != "INBOX" { + t.Fatalf("update targeting = %s/%s", eu.Username(), eu.Mailbox()) + } + seqs = append(seqs, eu.SeqNum) + case <-time.After(time.Second): + t.Fatal("no expunge update received") + } + } + if seqs[0] != 2 || seqs[1] != 5 { + t.Fatalf("seqs = %v, want [2 5]", seqs) + } } diff --git a/internal/imap_server/server.go b/internal/imap_server/server.go index 2e60eee..8bef425 100644 --- a/internal/imap_server/server.go +++ b/internal/imap_server/server.go @@ -18,6 +18,17 @@ import ( imapserver "github.com/emersion/go-imap/server" ) +// Pusher 是 IMAP 实时推送接口:SMTP/POP3/Web 在邮件状态变化后调用, +// 由 go-imap 广播给相关客户端(按用户名+邮箱过滤,IDLE 时即时送达)。 +type Pusher interface { + // PushNewMessage 推送新邮件(本地投递成功)。 + PushNewMessage(userEmail string, msg *db.Message) + // PushFlagsChanged 推送已读/星标等标志变化(MessageUpdate)。 + PushFlagsChanged(userEmail, mailbox string, msg *db.Message) + // PushExpunged 推送邮件被删除(ExpungeUpdate,seqNums 为删除前序号)。 + PushExpunged(userEmail, mailbox string, seqNums []uint32) +} + // IMAPServer wraps a go-imap Server and provides mailbox access capability. type IMAPServer struct { stores *store.Stores @@ -27,7 +38,8 @@ type IMAPServer struct { hub *connhub.Hub beMu sync.Mutex - bes []*imapBackend // 各监听器(明文/TLS)的 backend,用于新邮件推送 + bes []*imapBackend // 各监听器(明文/TLS)的 backend,用于新邮件推送 + srvs []*imapserver.Server // 各监听器实例,用于强制断开连接 } // NewIMAPServer creates a new IMAP server instance. tlsLoader may be nil @@ -45,15 +57,45 @@ func NewIMAPServer(cfg config.IMAPConfig, stores *store.Stores, tlsLoader *tlsut // NotifyNewMessage 向所有 IMAP 监听器推送新邮件通知(go-imap 广播时按 // 用户名+邮箱过滤,只送达已选中 INBOX 的客户端,IDLE 挂起时实时收到 // FETCH 响应)。由 SMTP/Web 本地投递成功时调用;channel 满时非阻塞丢弃。 -func (s *IMAPServer) NotifyNewMessage(userEmail string, msg *db.Message) { +func (s *IMAPServer) PushNewMessage(userEmail string, msg *db.Message) { if s == nil || userEmail == "" || msg == nil { return } - update := buildNewMessageUpdate(s.stores, userEmail, msg) + update := buildNewMessageUpdate(s.stores, userEmail, "INBOX", msg) if update == nil { return } + s.broadcastUpdate(update, userEmail, msg.ID) +} +// PushFlagsChanged 推送邮件标志(已读/星标等)变化给同用户的其他客户端。 +func (s *IMAPServer) PushFlagsChanged(userEmail, mailbox string, msg *db.Message) { + if s == nil || userEmail == "" || mailbox == "" || msg == nil { + return + } + update := buildFlagsUpdate(s.stores, userEmail, mailbox, msg, false) + if update == nil { + return + } + s.broadcastUpdate(update, userEmail, msg.ID) +} + +// PushExpunged 推送邮件被删除(每条序号一个 ExpungeUpdate)。 +func (s *IMAPServer) PushExpunged(userEmail, mailbox string, seqNums []uint32) { + if s == nil || userEmail == "" || mailbox == "" || len(seqNums) == 0 { + return + } + for _, seq := range seqNums { + update := &backend.ExpungeUpdate{ + Update: backend.NewUpdate(userEmail, mailbox), + SeqNum: seq, + } + s.broadcastUpdate(update, userEmail, 0) + } +} + +// broadcastUpdate 把一条更新非阻塞地投递到所有监听器的推送通道。 +func (s *IMAPServer) broadcastUpdate(update backend.Update, userEmail string, msgID uint) { s.beMu.Lock() bes := append([]*imapBackend(nil), s.bes...) s.beMu.Unlock() @@ -62,7 +104,7 @@ func (s *IMAPServer) NotifyNewMessage(userEmail string, msg *db.Message) { select { case b.updates <- update: default: - log.Printf("IMAP: 新邮件推送通道已满,丢弃 %s 的更新 (msg=%d)", userEmail, msg.ID) + log.Printf("IMAP: 推送通道已满,丢弃 %s 的更新 (msg=%d)", userEmail, msgID) } } } @@ -74,6 +116,33 @@ func (s *IMAPServer) registerBackend(be *imapBackend) { s.beMu.Unlock() } +// registerServer 记录监听器实例(用于强制断开连接)。 +func (s *IMAPServer) registerServer(srv *imapserver.Server) { + s.beMu.Lock() + s.srvs = append(s.srvs, srv) + s.beMu.Unlock() +} + +// DisconnectByAddr 强制断开指定远端地址的连接(管理后台「断开并封禁」)。 +// 关闭连接会触发 go-imap 的收尾流程(user.Logout、协议日志回填、hub 注销)。 +func (s *IMAPServer) DisconnectByAddr(remoteAddr string) { + if s == nil || remoteAddr == "" { + return + } + s.beMu.Lock() + srvs := append([]*imapserver.Server(nil), s.srvs...) + s.beMu.Unlock() + + for _, srv := range srvs { + srv.ForEachConn(func(conn imapserver.Conn) { + info := conn.Info() + if info != nil && info.RemoteAddr != nil && info.RemoteAddr.String() == remoteAddr { + _ = conn.Close() + } + }) + } +} + func (s *IMAPServer) tlsConfig() (*tls.Config, error) { if s.tlsLoader == nil { return nil, fmt.Errorf("IMAP TLS certificate or key not configured") @@ -85,17 +154,19 @@ func (s *IMAPServer) tlsConfig() (*tls.Config, error) { // newServer creates a configured imapserver.Server with the given address. func (s *IMAPServer) newServer(addr string, tlsConfig *tls.Config) *imapserver.Server { be := &imapBackend{ - stores: s.stores, - banCfg: s.banCfg, - port: portOf(addr), - hub: s.hub, - updates: make(chan backend.Update, 256), + stores: s.stores, + banCfg: s.banCfg, + port: portOf(addr), + hub: s.hub, + updates: make(chan backend.Update, 256), + disconnectAddr: s.DisconnectByAddr, } s.registerBackend(be) srv := imapserver.New(be) srv.Addr = addr srv.TLSConfig = tlsConfig srv.AllowInsecureAuth = tlsConfig == nil + s.registerServer(srv) return srv } diff --git a/internal/pop3_server/server.go b/internal/pop3_server/server.go index 4b18d91..5ea1091 100644 --- a/internal/pop3_server/server.go +++ b/internal/pop3_server/server.go @@ -14,6 +14,7 @@ import ( "mail_go/config" "mail_go/internal/connhub" "mail_go/internal/db" + "mail_go/internal/imap_server" "mail_go/internal/store" "mail_go/internal/tlsutil" ) @@ -26,13 +27,14 @@ type POP3Server struct { banCfg config.BanConfig tlsLoader *tlsutil.Loader hub *connhub.Hub + pusher imap_server.Pusher // 邮件删除推送(IMAP 客户端同步),可空 wg sync.WaitGroup } // NewPOP3Server creates a new POP3 server instance. tlsLoader may be nil // when TLS is not configured. -func NewPOP3Server(cfg config.POP3Config, stores *store.Stores, tlsLoader *tlsutil.Loader, banCfg config.BanConfig, hub *connhub.Hub) *POP3Server { - return &POP3Server{stores: stores, cfg: cfg, banCfg: banCfg, tlsLoader: tlsLoader, hub: hub} +func NewPOP3Server(cfg config.POP3Config, stores *store.Stores, tlsLoader *tlsutil.Loader, banCfg config.BanConfig, hub *connhub.Hub, pusher imap_server.Pusher) *POP3Server { + return &POP3Server{stores: stores, cfg: cfg, banCfg: banCfg, tlsLoader: tlsLoader, hub: hub, pusher: pusher} } func (s *POP3Server) tlsConfig() (*tls.Config, error) { @@ -135,8 +137,12 @@ func (s *POP3Server) handleConn(conn net.Conn, port int) { return } - // 连接追踪:注册到当前连接中心,连接结束时注销 + // 连接追踪:注册到当前连接中心,连接结束时注销; + // 强制断开:关闭底层连接(STLS 后 conn 变量已指向 tlsConn,同样生效)。 activeConn := s.hub.Register("pop3", clientIP, port, false) + if activeConn != nil { + activeConn.SetDisconnect(func() { _ = conn.Close() }) + } // 会话状态(供协议日志汇总) var ( @@ -397,6 +403,9 @@ func (s *POP3Server) handlePASS(conn net.Conn, password string, user *db.User) ( return nil, nil, nil } + // 保留完整邮箱作为登录标识(与 handleUSER 一致),便于推送/日志使用 + authUser.Username = user.Username + messages := s.loadMessages(authUser) deleted := make(map[int]bool) sendResponse(conn, fmt.Sprintf("+OK authenticated, %d messages", len(messages))) @@ -532,12 +541,14 @@ func (s *POP3Server) handleUIDL(conn net.Conn, arg string, messages []pop3Messag } // expungeDeleted actually deletes messages that were marked for deletion, -// returning the number of messages deleted. +// returning the number of messages deleted. 删除成功后向 IMAP 客户端推送 +// Expunge 通知(序号为删除前在 INBOX 中的位置)。 func (s *POP3Server) expungeDeleted(messages []pop3Message, deleted map[int]bool, user *db.User) int { if deleted == nil || user == nil || user.ID == 0 { return 0 } count := 0 + var seqs []uint32 for seqNum, msgDeleted := range deleted { if msgDeleted && seqNum >= 1 && seqNum <= len(messages) { if err := s.stores.Mails.Delete(messages[seqNum-1].id); err != nil { @@ -545,8 +556,12 @@ func (s *POP3Server) expungeDeleted(messages []pop3Message, deleted map[int]bool continue } count++ + seqs = append(seqs, uint32(seqNum)) } } + if s.pusher != nil && len(seqs) > 0 && user.Username != "" { + s.pusher.PushExpunged(user.Username, "INBOX", seqs) + } return count } diff --git a/internal/pop3_server/server_test.go b/internal/pop3_server/server_test.go index 6239179..86647e4 100644 --- a/internal/pop3_server/server_test.go +++ b/internal/pop3_server/server_test.go @@ -160,3 +160,90 @@ func TestPop3CommandDetail(t *testing.T) { t.Fatalf("empty detail = %q", empty) } } + +// mockPusher 记录推送调用的测试桩。 +type mockPusher struct { + expunged []struct { + Email string + Mailbox string + Seqs []uint32 + } +} + +func (m *mockPusher) PushNewMessage(string, *db.Message) {} +func (m *mockPusher) PushFlagsChanged(string, string, *db.Message) {} +func (m *mockPusher) PushExpunged(email, mailbox string, seqs []uint32) { + m.expunged = append(m.expunged, struct { + Email string + Mailbox string + Seqs []uint32 + }{email, mailbox, seqs}) +} + +// TestExpungePushesIMAPUpdate 验证 POP3 删除邮件后向 IMAP 推送 Expunge。 +func TestExpungePushesIMAPUpdate(t *testing.T) { + s := newTestServer(t) + pusher := &mockPusher{} + s.pusher = pusher + + domain := &db.Domain{Name: "example.com"} + if err := s.stores.Domains.Create(domain); err != nil { + t.Fatalf("create domain: %v", err) + } + user := &db.User{Username: "alice", DomainID: domain.ID, IsActive: true} + hashed, _ := bcrypt.GenerateFromPassword([]byte("secret123"), bcrypt.DefaultCost) + user.PasswordHash = string(hashed) + if err := s.stores.Users.Create(user); err != nil { + t.Fatalf("create user: %v", err) + } + for i := 0; i < 3; i++ { + msg := &db.Message{UserID: user.ID, Folder: "INBOX", FromAddr: "x@y", Subject: "m", Date: time.Now()} + if err := s.stores.Mails.Create(msg); err != nil { + t.Fatalf("create message %d: %v", i, err) + } + } + + server, client := net.Pipe() + defer server.Close() + defer client.Close() + + done := make(chan struct{}) + go func() { + defer close(done) + s.handleConn(server, 110) + }() + + br := bufio.NewReader(client) + br.ReadString('\n') + client.Write([]byte("USER alice@example.com\r\n")) + br.ReadString('\n') + client.Write([]byte("PASS secret123\r\n")) + br.ReadString('\n') + // 删除第 1 封后退出 + client.Write([]byte("DELE 1\r\n")) + br.ReadString('\n') + client.Write([]byte("QUIT\r\n")) + br.ReadString('\n') + client.Close() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("handleConn did not return") + } + + if len(pusher.expunged) != 1 { + t.Fatalf("expunge pushes = %d, want 1", len(pusher.expunged)) + } + p := pusher.expunged[0] + if p.Email != "alice@example.com" || p.Mailbox != "INBOX" { + t.Fatalf("push target = %s/%s", p.Email, p.Mailbox) + } + if len(p.Seqs) != 1 || p.Seqs[0] != 1 { + t.Fatalf("seqs = %v, want [1]", p.Seqs) + } + // 邮件确实已删除 + if n, _ := s.stores.Mails.CountByUserAndFolder(user.ID, "INBOX"); n != 2 { + t.Fatalf("inbox count = %d, want 2", n) + } +} diff --git a/internal/smtp_server/server.go b/internal/smtp_server/server.go index dbfa4df..179e450 100644 --- a/internal/smtp_server/server.go +++ b/internal/smtp_server/server.go @@ -14,6 +14,7 @@ import ( "mail_go/config" "mail_go/internal/connhub" "mail_go/internal/db" + "mail_go/internal/imap_server" "mail_go/internal/mailutil" "mail_go/internal/outbound" "mail_go/internal/storage" @@ -33,10 +34,6 @@ const ( smtpModeImplicitTLS ) -// NewMailNotify 是本地新邮件投递完成后的通知回调(IMAP 推送用), -// userEmail 为收件人完整邮箱,msg 为已入库的邮件。 -type NewMailNotify func(userEmail string, msg *db.Message) - // SMTPServer wraps go-smtp servers and provides local mail delivery. type SMTPServer struct { stores *store.Stores @@ -46,13 +43,13 @@ type SMTPServer struct { banCfg config.BanConfig tlsLoader *tlsutil.Loader hub *connhub.Hub - notify NewMailNotify // 本地投递成功通知(IMAP 新邮件推送),可空 + pusher imap_server.Pusher // 本地投递成功推送(IMAP 新邮件),可空 } // NewSMTPServer creates a new SMTP server instance. tlsLoader may be nil // when TLS is not configured. -func NewSMTPServer(cfg config.SMTPConfig, stores *store.Stores, attStorage *storage.AttachmentStorage, ob *outbound.Manager, tlsLoader *tlsutil.Loader, banCfg config.BanConfig, hub *connhub.Hub, notify NewMailNotify) *SMTPServer { - return &SMTPServer{stores: stores, storage: attStorage, outbound: ob, cfg: cfg, banCfg: banCfg, tlsLoader: tlsLoader, hub: hub, notify: notify} +func NewSMTPServer(cfg config.SMTPConfig, stores *store.Stores, attStorage *storage.AttachmentStorage, ob *outbound.Manager, tlsLoader *tlsutil.Loader, banCfg config.BanConfig, hub *connhub.Hub, pusher imap_server.Pusher) *SMTPServer { + return &SMTPServer{stores: stores, storage: attStorage, outbound: ob, cfg: cfg, banCfg: banCfg, tlsLoader: tlsLoader, hub: hub, pusher: pusher} } func (s *SMTPServer) tlsConfig() (*tls.Config, error) { @@ -119,6 +116,11 @@ type smtpBackend struct { func (be *smtpBackend) NewSession(c *smtp.Conn) (smtp.Session, error) { clientIP := store.ClientIPFromAddr(c.Conn().RemoteAddr()) conn := be.server.hub.Register("smtp", clientIP, be.server.sessionPort(be.mode), be.server.tlsActive(c)) + if conn != nil { + // 强制断开:关闭底层连接后 go-smtp 读到 EOF,正常走 Logout 收尾 + raw := c.Conn() + conn.SetDisconnect(func() { _ = raw.Close() }) + } return &smtpSession{ backend: be, mode: be.mode, @@ -350,8 +352,8 @@ func (s *smtpSession) Data(r io.Reader) error { log.Printf("SMTP: message delivered to %s", rcpt) localDelivered++ // 本地投递成功 → IMAP 新邮件推送(IDLE 客户端实时收到通知) - if notify := s.backend.server.notify; notify != nil && msg != nil { - notify(user.Username+"@"+user.Domain.Name, msg) + if pusher := s.backend.server.pusher; pusher != nil && msg != nil { + pusher.PushNewMessage(user.Username+"@"+user.Domain.Name, msg) } } s.msgCount += localDelivered diff --git a/internal/web/handlers/admin.go b/internal/web/handlers/admin.go index c946c38..0201b44 100644 --- a/internal/web/handlers/admin.go +++ b/internal/web/handlers/admin.go @@ -43,6 +43,42 @@ func NewAdminHandler(stores *store.Stores, attStorage *storage.AttachmentStorage return &AdminHandler{stores: stores, storage: attStorage, tlsDir: tlsDir, caddyDataDir: caddyDataDir, outbound: ob, protocolLogKeepDays: protocolLogKeepDays, hub: hub} } +// manualBanDuration 管理员手动封禁时长(180 天,与自动封禁档位上限制一致)。 +const manualBanDuration = 180 * 24 * time.Hour + +// DisconnectConnection 强制断开指定连接并封禁其 IP(管理后台「断开并封禁」)。 +// 封禁后该 IP 的所有在线连接一并断开。 +func (h *AdminHandler) DisconnectConnection(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.String(http.StatusBadRequest, "无效的连接ID") + return + } + + conn, ok := h.hub.Get(id) + if !ok { + c.String(http.StatusNotFound, "连接不存在或已断开") + return + } + + // 加入黑名单:180 天封禁(管理员可随时解封) + if err := h.stores.Bans.Create(&db.BanEntry{ + IPAddress: conn.IP, + Reason: "管理员手动封禁(连接断开)", + FailCount: 0, + BanCount: 0, + ExpiresAt: time.Now().Add(manualBanDuration), + }); err != nil { + c.String(http.StatusInternalServerError, "封禁失败: %v", err) + return + } + + // 断开该 IP 的全部连接(含本连接与其他协议连接) + n := h.hub.DisconnectByIP(conn.IP) + log.Printf("admin: 已封禁并断开 IP %s 的 %d 个连接", conn.IP, n) + c.Redirect(http.StatusFound, "/admin/connections") +} + // ListConnections 渲染当前协议连接页面(SMTP/IMAP/POP3 实时连接)。 func (h *AdminHandler) ListConnections(c *gin.Context) { conns := h.hub.List() diff --git a/internal/web/handlers/connections_test.go b/internal/web/handlers/connections_test.go new file mode 100644 index 0000000..dcf1c19 --- /dev/null +++ b/internal/web/handlers/connections_test.go @@ -0,0 +1,101 @@ +package handlers + +import ( + "net/http/httptest" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "mail_go/internal/connhub" + "mail_go/internal/db" + "mail_go/internal/store" + + "github.com/gin-gonic/gin" + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +// performPost 发送 POST 请求并返回响应(用于处理器测试)。 +func performPost(r *gin.Engine, path string) *httptest.ResponseRecorder { + req := httptest.NewRequest("POST", path, nil) + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + return rec +} + +// TestDisconnectConnection 验证「断开并封禁」:创建黑名单记录并断开该 IP 全部连接。 +func TestDisconnectConnection(t *testing.T) { + gdb, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "test.db")), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + if err := gdb.AutoMigrate(&db.BanEntry{}); err != nil { + t.Fatalf("migrate: %v", err) + } + stores := store.NewStores(gdb) + hub := connhub.New() + + var closed atomic.Int32 + // 目标 IP 两个连接(模拟多协议在线) + c1 := hub.Register("smtp", "203.0.113.77", 25, false) + c1.SetDisconnect(func() { closed.Add(1) }) + c2 := hub.Register("imap", "203.0.113.77", 993, true) + c2.SetDisconnect(func() { closed.Add(1) }) + // 其他 IP 不应受影响 + c3 := hub.Register("pop3", "203.0.113.78", 110, false) + c3.SetDisconnect(func() { closed.Add(1) }) + + h := &AdminHandler{stores: stores, hub: hub} + + gin.SetMode(gin.TestMode) + r := gin.New() + r.POST("/admin/connections/:id/disconnect", h.DisconnectConnection) + + rec := performPost(r, "/admin/connections/1/disconnect") + if rec.Code != 302 { + t.Fatalf("status = %d, want 302", rec.Code) + } + + // 该 IP 的两个连接都被断开,其他连接不受影响 + if closed.Load() != 2 { + t.Fatalf("closed = %d, want 2", closed.Load()) + } + if n := hub.Counts()["pop3"]; n != 1 { + t.Fatalf("pop3 count = %d, want 1 (unaffected)", n) + } + + // 黑名单记录:180 天封禁 + banned, entry := stores.Bans.IsBanned("203.0.113.77") + if !banned { + t.Fatal("IP should be banned") + } + if entry.Reason != "管理员手动封禁(连接断开)" { + t.Fatalf("reason = %q", entry.Reason) + } + wantExpiry := time.Now().Add(180 * 24 * time.Hour) + if entry.ExpiresAt.Before(wantExpiry.Add(-time.Minute)) || entry.ExpiresAt.After(wantExpiry.Add(time.Minute)) { + t.Fatalf("expiry = %v, want ~180 days", entry.ExpiresAt) + } +} + +// TestDisconnectConnectionNotFound 验证不存在的连接返回 404。 +func TestDisconnectConnectionNotFound(t *testing.T) { + gdb, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "test.db")), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + if err := gdb.AutoMigrate(&db.BanEntry{}); err != nil { + t.Fatalf("migrate: %v", err) + } + h := &AdminHandler{stores: store.NewStores(gdb), hub: connhub.New()} + + gin.SetMode(gin.TestMode) + r := gin.New() + r.POST("/admin/connections/:id/disconnect", h.DisconnectConnection) + + rec := performPost(r, "/admin/connections/999/disconnect") + if rec.Code != 404 { + t.Fatalf("status = %d, want 404", rec.Code) + } +} diff --git a/internal/web/handlers/mail.go b/internal/web/handlers/mail.go index a9ede96..f95c12f 100644 --- a/internal/web/handlers/mail.go +++ b/internal/web/handlers/mail.go @@ -12,8 +12,8 @@ import ( "time" "mail_go/internal/db" + "mail_go/internal/imap_server" "mail_go/internal/outbound" - "mail_go/internal/smtp_server" "mail_go/internal/storage" "mail_go/internal/store" @@ -50,14 +50,14 @@ type MailHandler struct { stores *store.Stores storage *storage.AttachmentStorage outbound *outbound.Manager - // notify 本地投递成功通知(IMAP 新邮件推送),可空 - notify smtp_server.NewMailNotify + // pusher 邮件状态变化推送(IMAP 客户端实时同步),可空 + pusher imap_server.Pusher } // NewMailHandler creates a new MailHandler with the given stores, attachment // storage and outbound delivery manager. -func NewMailHandler(stores *store.Stores, attStorage *storage.AttachmentStorage, ob *outbound.Manager, notify smtp_server.NewMailNotify) *MailHandler { - return &MailHandler{stores: stores, storage: attStorage, outbound: ob, notify: notify} +func NewMailHandler(stores *store.Stores, attStorage *storage.AttachmentStorage, ob *outbound.Manager, pusher imap_server.Pusher) *MailHandler { + return &MailHandler{stores: stores, storage: attStorage, outbound: ob, pusher: pusher} } // folderCounts returns sidebar badge counts for the current user. @@ -386,8 +386,8 @@ func (h *MailHandler) DoSend(c *gin.Context) { return } // 本地投递成功 → IMAP 新邮件推送(IDLE 客户端实时收到通知) - if h.notify != nil { - h.notify(rcptUser.Username+"@"+rcptUser.Domain.Name, inboxMsg) + if h.pusher != nil { + h.pusher.PushNewMessage(rcptUser.Username+"@"+rcptUser.Domain.Name, inboxMsg) } } @@ -638,8 +638,30 @@ func (h *MailHandler) Delete(c *gin.Context) { _ = h.stores.Users.UpdateUsedBytes(userID, -att.FileSize) } _ = h.stores.Attachments.DeleteByMessage(uint(id)) + + // 删除前计算消息在所属文件夹中的序号(用于 Expunge 推送) + var seq uint32 + if msgs, err := h.stores.Mails.ListAllByUserAndFolder(userID, msg.Folder); err == nil { + for i := range msgs { + if msgs[i].ID == uint(id) { + seq = uint32(i + 1) + break + } + } + } _ = h.stores.Mails.Delete(uint(id)) + // 删除 → 推送给该用户的其他 IMAP 客户端 + if h.pusher != nil && seq > 0 { + userEmail := "" + if cu, ok := c.Get("currentUser"); ok { + if u, ok := cu.(*db.User); ok { + userEmail = u.Username + "@" + u.Domain.Name + } + } + h.pusher.PushExpunged(userEmail, msg.Folder, []uint32{seq}) + } + // Redirect back based on the folder(仅同站相对路径,防开放重定向) referer := safeRedirectPath(c.GetHeader("Referer")) if referer == "" { @@ -665,6 +687,18 @@ func (h *MailHandler) MarkRead(c *gin.Context) { _ = h.stores.Mails.MarkRead(uint(id)) + // 已读变化 → 推送给该用户的其他 IMAP 客户端 + if h.pusher != nil { + msg.IsRead = true + userEmail := "" + if cu, ok := c.Get("currentUser"); ok { + if u, ok := cu.(*db.User); ok { + userEmail = u.Username + "@" + u.Domain.Name + } + } + h.pusher.PushFlagsChanged(userEmail, msg.Folder, msg) + } + // Redirect back based on the folder(仅同站相对路径,防开放重定向) referer := safeRedirectPath(c.GetHeader("Referer")) if referer == "" { diff --git a/internal/web/server.go b/internal/web/server.go index 3839beb..4e57936 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -15,9 +15,9 @@ import ( "mail_go/config" "mail_go/internal/connhub" + "mail_go/internal/imap_server" "mail_go/internal/mailutil" "mail_go/internal/outbound" - "mail_go/internal/smtp_server" "mail_go/internal/storage" "mail_go/internal/store" "mail_go/internal/web/handlers" @@ -54,8 +54,8 @@ type WebServer struct { caddyDataDir string outbound *outbound.Manager hub *connhub.Hub - // notify 本地投递成功通知(IMAP 新邮件推送),可空 - notify smtp_server.NewMailNotify + // pusher 邮件状态变化推送(IMAP 客户端实时同步),可空 + pusher imap_server.Pusher } // templateFuncs returns custom template functions for rendering. @@ -179,7 +179,7 @@ func avatarStyle(s string) string { // NewWebServer creates a new WebServer, initializes the Gin engine, // configures sessions, middleware, and registers all routes. -func NewWebServer(cfg config.WebConfig, stores *store.Stores, attStorage *storage.AttachmentStorage, storageCfg config.StorageConfig, authCfg config.AuthConfig, banCfg config.BanConfig, caddyCfg config.CaddyConfig, ob *outbound.Manager, hub *connhub.Hub, notify smtp_server.NewMailNotify) (*WebServer, error) { +func NewWebServer(cfg config.WebConfig, stores *store.Stores, attStorage *storage.AttachmentStorage, storageCfg config.StorageConfig, authCfg config.AuthConfig, banCfg config.BanConfig, caddyCfg config.CaddyConfig, ob *outbound.Manager, hub *connhub.Hub, pusher imap_server.Pusher) (*WebServer, error) { if err := config.ValidateSecretKey(cfg.SecretKey); err != nil { return nil, err } @@ -226,7 +226,7 @@ func NewWebServer(cfg config.WebConfig, stores *store.Stores, attStorage *storag caddyDataDir: caddyCfg.DataDir, outbound: ob, hub: hub, - notify: notify, + pusher: pusher, } ws.registerRoutes() @@ -236,7 +236,7 @@ func NewWebServer(cfg config.WebConfig, stores *store.Stores, attStorage *storag // registerRoutes sets up all HTTP routes with their handlers and middleware. func (ws *WebServer) registerRoutes() { authHandler := handlers.NewAuthHandler(ws.stores, ws.authCfg, ws.banCfg) - mailHandler := handlers.NewMailHandler(ws.stores, ws.storage, ws.outbound, ws.notify) + mailHandler := handlers.NewMailHandler(ws.stores, ws.storage, ws.outbound, ws.pusher) adminHandler := handlers.NewAdminHandler(ws.stores, ws.storage, filepath.Join(ws.storageCfg.BaseDir, "tls", "domains"), ws.caddyDataDir, ws.outbound, ws.cfg.ProtocolLogKeepDays, ws.hub) // Apply BanMiddleware globally before public routes @@ -308,6 +308,7 @@ func (ws *WebServer) registerRoutes() { admin.GET("/protocol-logs", adminHandler.ListProtocolLogs) admin.POST("/protocol-logs/cleanup", adminHandler.CleanupProtocolLogs) admin.GET("/connections", adminHandler.ListConnections) + admin.POST("/connections/:id/disconnect", adminHandler.DisconnectConnection) } } diff --git a/internal/web/templates/admin/connections.html b/internal/web/templates/admin/connections.html index 832985d..14dca67 100644 --- a/internal/web/templates/admin/connections.html +++ b/internal/web/templates/admin/connections.html @@ -61,6 +61,7 @@ 连接时间 时长 最后活跃 + 操作 @@ -79,9 +80,15 @@ {{.Connected.Format "2006-01-02 15:04:05"}} {{durationSeconds ($.now.Sub .Connected)}}s {{.LastActive.Format "2006-01-02 15:04:05"}} + +
+ +
+ {{else}} - 当前没有活动连接 + 当前没有活动连接 {{end}} diff --git a/main.go b/main.go index 7b87dcc..f149f38 100644 --- a/main.go +++ b/main.go @@ -258,7 +258,7 @@ func main() { } // 8. Start SMTP server(本地投递成功后触发 IMAP 新邮件推送) - smtpSrv := smtp_server.NewSMTPServer(cfg.SMTP, stores, attStorage, outboundMgr, smtpTLS, cfg.Ban, connHub, imapSrv.NotifyNewMessage) + smtpSrv := smtp_server.NewSMTPServer(cfg.SMTP, stores, attStorage, outboundMgr, smtpTLS, cfg.Ban, connHub, imapSrv) go func() { if err := smtpSrv.Start(); err != nil { log.Printf("SMTP 服务启动失败: %v", err) @@ -279,7 +279,7 @@ func main() { } // 9. Start POP3 server - pop3Srv := pop3_server.NewPOP3Server(cfg.POP3, stores, pop3TLS, cfg.Ban, connHub) + pop3Srv := pop3_server.NewPOP3Server(cfg.POP3, stores, pop3TLS, cfg.Ban, connHub, imapSrv) go func() { if err := pop3Srv.Start(); err != nil { log.Printf("POP3 服务启动失败: %v", err) @@ -295,7 +295,7 @@ func main() { } // 10. Start Web server(本地写信投递成功后同样触发 IMAP 新邮件推送) - webServer, err := web.NewWebServer(cfg.Web, stores, attStorage, cfg.Storage, cfg.Auth, cfg.Ban, cfg.Caddy, outboundMgr, connHub, imapSrv.NotifyNewMessage) + webServer, err := web.NewWebServer(cfg.Web, stores, attStorage, cfg.Storage, cfg.Auth, cfg.Ban, cfg.Caddy, outboundMgr, connHub, imapSrv) if err != nil { log.Fatalf("Web 服务初始化失败: %v", err) }