Merge pull request 'Add IMAP SOCKS5 proxy dialing support' (#8) from feat/imap-proxy-support into main
Reviewed-on: #8 Reviewed-by: codex <39+codex@noreply.localhost>
This commit was merged in pull request #8.
This commit is contained in:
@@ -14,3 +14,6 @@ data/*.db-shm
|
||||
.idea/
|
||||
.vscode/
|
||||
*.swp
|
||||
|
||||
# Local worktrees
|
||||
.worktrees/
|
||||
|
||||
+89
-3
@@ -2,9 +2,12 @@ package mail
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -13,6 +16,7 @@ import (
|
||||
"github.com/emersion/go-imap/v2"
|
||||
"github.com/emersion/go-imap/v2/imapclient"
|
||||
"github.com/google/uuid"
|
||||
"golang.org/x/net/proxy"
|
||||
"thuanle.me/claw-email-bridge/internal/config"
|
||||
"thuanle.me/claw-email-bridge/internal/database"
|
||||
"thuanle.me/claw-email-bridge/internal/logging"
|
||||
@@ -25,14 +29,40 @@ type IMAPWatcher struct {
|
||||
db *gorm.DB
|
||||
client *imapclient.Client
|
||||
|
||||
dialIMAP func(addr string, options *imapclient.Options) (*imapclient.Client, error)
|
||||
dialIMAPViaProxy func(addr, proxyURL string, options *imapclient.Options) (*imapclient.Client, error)
|
||||
|
||||
// onReceived is called after a task is saved as RECEIVED.
|
||||
// This will be wired to the OpenClaw dispatch in a future step.
|
||||
OnReceived func(task *database.Task)
|
||||
}
|
||||
|
||||
const imapDialTimeout = 30 * time.Second
|
||||
|
||||
type timeoutContextDialer struct {
|
||||
timeout time.Duration
|
||||
}
|
||||
|
||||
func (d timeoutContextDialer) Dial(network, address string) (net.Conn, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), d.timeout)
|
||||
defer cancel()
|
||||
return d.DialContext(ctx, network, address)
|
||||
}
|
||||
|
||||
func (d timeoutContextDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
|
||||
return (&net.Dialer{Timeout: d.timeout}).DialContext(ctx, network, address)
|
||||
}
|
||||
|
||||
var socks5DialerFactory = proxy.SOCKS5
|
||||
|
||||
// NewIMAPWatcher creates a new IMAP watcher.
|
||||
func NewIMAPWatcher(cfg *config.Config, db *gorm.DB) *IMAPWatcher {
|
||||
return &IMAPWatcher{cfg: cfg, db: db}
|
||||
return &IMAPWatcher{
|
||||
cfg: cfg,
|
||||
db: db,
|
||||
dialIMAP: imapclient.DialTLS,
|
||||
dialIMAPViaProxy: dialTLSViaSOCKS5,
|
||||
}
|
||||
}
|
||||
|
||||
// Run connects to IMAP, selects INBOX, and enters the IDLE loop.
|
||||
@@ -71,9 +101,15 @@ func (w *IMAPWatcher) connectAndWatch(ctx context.Context) error {
|
||||
addr := fmt.Sprintf("%s:%s", w.cfg.IMAPHost, w.cfg.IMAPPort)
|
||||
slog.Info("imap: connecting", "addr", addr)
|
||||
|
||||
c, err := imapclient.DialTLS(addr, options)
|
||||
var c *imapclient.Client
|
||||
var err error
|
||||
if strings.TrimSpace(w.cfg.IMAPProxyURL) == "" {
|
||||
c, err = w.dialIMAP(addr, options)
|
||||
} else {
|
||||
c, err = w.dialIMAPViaProxy(addr, w.cfg.IMAPProxyURL, options)
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("dial TLS: %w", err)
|
||||
return fmt.Errorf("dial IMAP: %w", err)
|
||||
}
|
||||
w.client = c
|
||||
defer func() {
|
||||
@@ -144,6 +180,56 @@ func (w *IMAPWatcher) connectAndWatch(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
|
||||
func dialTLSViaSOCKS5(addr, proxyURL string, options *imapclient.Options) (*imapclient.Client, error) {
|
||||
u, err := url.Parse(proxyURL)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse proxy url: %w", err)
|
||||
}
|
||||
if u.Scheme != "socks5" {
|
||||
return nil, fmt.Errorf("unsupported proxy scheme: %s", u.Scheme)
|
||||
}
|
||||
|
||||
var auth *proxy.Auth
|
||||
if u.User != nil {
|
||||
pw, _ := u.User.Password()
|
||||
auth = &proxy.Auth{User: u.User.Username(), Password: pw}
|
||||
}
|
||||
|
||||
dialer, err := socks5DialerFactory("tcp", u.Host, auth, timeoutContextDialer{timeout: imapDialTimeout})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create socks5 dialer: %w", err)
|
||||
}
|
||||
|
||||
dialCtx, dialCancel := context.WithTimeout(context.Background(), imapDialTimeout)
|
||||
defer dialCancel()
|
||||
|
||||
var conn net.Conn
|
||||
if cd, ok := dialer.(proxy.ContextDialer); ok {
|
||||
conn, err = cd.DialContext(dialCtx, "tcp", addr)
|
||||
} else {
|
||||
conn, err = dialer.Dial("tcp", addr)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("proxy dial: %w", err)
|
||||
}
|
||||
|
||||
host, _, splitErr := net.SplitHostPort(addr)
|
||||
if splitErr != nil || host == "" {
|
||||
host = addr
|
||||
}
|
||||
tlsConn := tls.Client(conn, &tls.Config{ServerName: host})
|
||||
_ = tlsConn.SetDeadline(time.Now().Add(imapDialTimeout))
|
||||
handshakeCtx, handshakeCancel := context.WithTimeout(context.Background(), imapDialTimeout)
|
||||
defer handshakeCancel()
|
||||
if err := tlsConn.HandshakeContext(handshakeCtx); err != nil {
|
||||
_ = tlsConn.Close()
|
||||
return nil, fmt.Errorf("tls handshake: %w", err)
|
||||
}
|
||||
_ = tlsConn.SetDeadline(time.Time{})
|
||||
|
||||
return imapclient.New(tlsConn, options), nil
|
||||
}
|
||||
|
||||
// fetchUnseen searches for UNSEEN messages and processes each one.
|
||||
func (w *IMAPWatcher) fetchUnseen(c *imapclient.Client) error {
|
||||
criteria := &imap.SearchCriteria{
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
package mail
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/emersion/go-imap/v2/imapclient"
|
||||
"golang.org/x/net/proxy"
|
||||
"thuanle.me/claw-email-bridge/internal/config"
|
||||
)
|
||||
|
||||
func TestConnectAndWatch_UsesDirectDial_WhenProxyUnset(t *testing.T) {
|
||||
watcher := &IMAPWatcher{
|
||||
cfg: &config.Config{
|
||||
IMAPHost: "imap.example.com",
|
||||
IMAPPort: "993",
|
||||
IMAPUser: "u",
|
||||
IMAPPass: "p",
|
||||
},
|
||||
}
|
||||
|
||||
var directCalled bool
|
||||
watcher.dialIMAP = func(addr string, _ *imapclient.Options) (*imapclient.Client, error) {
|
||||
directCalled = true
|
||||
return nil, errors.New("stop")
|
||||
}
|
||||
watcher.dialIMAPViaProxy = func(addr, proxyURL string, _ *imapclient.Options) (*imapclient.Client, error) {
|
||||
t.Fatalf("did not expect proxy dial, got addr=%s proxy=%s", addr, proxyURL)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
err := watcher.connectAndWatch(context.Background())
|
||||
if err == nil || !strings.Contains(err.Error(), "stop") {
|
||||
t.Fatalf("expected stop error, got %v", err)
|
||||
}
|
||||
if !directCalled {
|
||||
t.Fatal("expected direct dial path")
|
||||
}
|
||||
}
|
||||
|
||||
func TestConnectAndWatch_UsesProxyDial_WhenProxySet(t *testing.T) {
|
||||
watcher := &IMAPWatcher{
|
||||
cfg: &config.Config{
|
||||
IMAPHost: "imap.example.com",
|
||||
IMAPPort: "993",
|
||||
IMAPUser: "u",
|
||||
IMAPPass: "p",
|
||||
IMAPProxyURL: "socks5://127.0.0.1:1080",
|
||||
},
|
||||
}
|
||||
|
||||
var proxyCalled bool
|
||||
watcher.dialIMAP = func(addr string, _ *imapclient.Options) (*imapclient.Client, error) {
|
||||
t.Fatalf("did not expect direct dial, got addr=%s", addr)
|
||||
return nil, nil
|
||||
}
|
||||
watcher.dialIMAPViaProxy = func(addr, proxyURL string, _ *imapclient.Options) (*imapclient.Client, error) {
|
||||
proxyCalled = true
|
||||
return nil, errors.New("stop")
|
||||
}
|
||||
|
||||
err := watcher.connectAndWatch(context.Background())
|
||||
if err == nil || !strings.Contains(err.Error(), "stop") {
|
||||
t.Fatalf("expected stop error, got %v", err)
|
||||
}
|
||||
if !proxyCalled {
|
||||
t.Fatal("expected proxy dial path")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDialTLSViaSOCKS5_RejectsNonSocks5Scheme(t *testing.T) {
|
||||
_, err := dialTLSViaSOCKS5("imap.example.com:993", "http://proxy:8080", nil)
|
||||
if err == nil || !strings.Contains(err.Error(), "unsupported proxy scheme") {
|
||||
t.Fatalf("expected unsupported scheme error, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
type fakeContextDialer struct{}
|
||||
|
||||
func (fakeContextDialer) Dial(network, address string) (net.Conn, error) {
|
||||
return nil, errors.New("used dial")
|
||||
}
|
||||
|
||||
func (fakeContextDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
|
||||
if _, ok := ctx.Deadline(); !ok {
|
||||
return nil, errors.New("missing deadline")
|
||||
}
|
||||
return nil, errors.New("used dialcontext")
|
||||
}
|
||||
|
||||
func TestDialTLSViaSOCKS5_UsesDialContextWithTimeout(t *testing.T) {
|
||||
orig := socks5DialerFactory
|
||||
t.Cleanup(func() { socks5DialerFactory = orig })
|
||||
|
||||
socks5DialerFactory = func(network, address string, auth *proxy.Auth, forward proxy.Dialer) (proxy.Dialer, error) {
|
||||
if _, ok := forward.(proxy.ContextDialer); !ok {
|
||||
t.Fatalf("expected ContextDialer, got %T", forward)
|
||||
}
|
||||
return fakeContextDialer{}, nil
|
||||
}
|
||||
|
||||
_, err := dialTLSViaSOCKS5("imap.example.com:993", "socks5://127.0.0.1:1080", nil)
|
||||
if err == nil || !strings.Contains(err.Error(), "proxy dial: used dialcontext") {
|
||||
t.Fatalf("expected DialContext path, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
type trackingConn struct {
|
||||
deadline time.Time
|
||||
}
|
||||
|
||||
func (c *trackingConn) Read(b []byte) (int, error) { return 0, io.EOF }
|
||||
func (c *trackingConn) Write(b []byte) (int, error) { return 0, io.EOF }
|
||||
func (c *trackingConn) Close() error { return nil }
|
||||
func (c *trackingConn) LocalAddr() net.Addr { return dummyAddr("local") }
|
||||
func (c *trackingConn) RemoteAddr() net.Addr { return dummyAddr("remote") }
|
||||
func (c *trackingConn) SetDeadline(t time.Time) error {
|
||||
c.deadline = t
|
||||
return nil
|
||||
}
|
||||
func (c *trackingConn) SetReadDeadline(t time.Time) error { return nil }
|
||||
func (c *trackingConn) SetWriteDeadline(t time.Time) error { return nil }
|
||||
|
||||
type dummyAddr string
|
||||
|
||||
func (a dummyAddr) Network() string { return "tcp" }
|
||||
func (a dummyAddr) String() string { return string(a) }
|
||||
|
||||
type contextConnDialer struct {
|
||||
conn *trackingConn
|
||||
}
|
||||
|
||||
func (d contextConnDialer) Dial(network, address string) (net.Conn, error) {
|
||||
return nil, errors.New("used dial")
|
||||
}
|
||||
|
||||
func (d contextConnDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
|
||||
return d.conn, nil
|
||||
}
|
||||
|
||||
func TestDialTLSViaSOCKS5_SetsDeadlineForTLSHandshake(t *testing.T) {
|
||||
orig := socks5DialerFactory
|
||||
t.Cleanup(func() { socks5DialerFactory = orig })
|
||||
|
||||
conn := &trackingConn{}
|
||||
socks5DialerFactory = func(network, address string, auth *proxy.Auth, forward proxy.Dialer) (proxy.Dialer, error) {
|
||||
return contextConnDialer{conn: conn}, nil
|
||||
}
|
||||
|
||||
_, err := dialTLSViaSOCKS5("imap.example.com:993", "socks5://127.0.0.1:1080", nil)
|
||||
if err == nil || !strings.Contains(err.Error(), "tls handshake") {
|
||||
t.Fatalf("expected tls handshake error, got %v", err)
|
||||
}
|
||||
if conn.deadline.IsZero() {
|
||||
t.Fatal("expected TLS handshake deadline to be set")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user