mirror of
https://blitiri.com.ar/repos/chasquid
synced 2026-10-28 15:02:28 +00:00
smtpsrv: Set a limit on concurrent SMTP connections
If chasquid runs out of file descriptors, that can cause problems accepting connections, writing to the queue, and even reloading configs. Some of those errors are fatal, as they should not happen under normal conditions. Since a remote client can open a lot of connections, this can end up consuming all file descriptors, and resulting in a denial of service. To reduce the impact of that kind of event, this patch makes chasquid limit the number of concurrent SMTP connections to half the maximum number of file descriptors. Connections over that limit will get closed immediately. The server will log (rate-limited) when such event happens, and new metrics make these events visible through monitoring too. This approach complements the previous patch, which made the server retry Accept errors, to reduce the impact of connection overloads. This mainly affects deployments with low file descriptor limits; most modern deployments have very high limits by default, so they are unlikely to experience this problem in the first place. Users who have to be very defensive can deploy OS-level safeguards (such as firewall rules to limit number of connections, etc.) which are more effective and tunable. Thanks to John JEAN <john@wargan.com> for finding and reporting this problem. The detailed report can be found in https://johnjean.io/advisories/chasquid-unauthenticated-remote-dos/.
This commit is contained in:
@@ -11,12 +11,15 @@ import (
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"math"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"path"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"blitiri.com.ar/go/chasquid/internal/aliases"
|
||||
@@ -52,6 +55,11 @@ const (
|
||||
var (
|
||||
acceptErrors = expvarom.NewInt("chasquid/smtpIn/acceptErrors",
|
||||
"count of errors accepting incoming connections")
|
||||
activeConnections = expvarom.NewInt("chasquid/smtpIn/activeConnections",
|
||||
"number of active incoming connections")
|
||||
connectionsOverLimit = expvarom.NewInt(
|
||||
"chasquid/smtpIn/connectionsOverLimit",
|
||||
"count of incoming connections closed due to the connection limit")
|
||||
)
|
||||
|
||||
// Server represents an SMTP server instance.
|
||||
@@ -101,6 +109,14 @@ type Server struct {
|
||||
|
||||
// Path to the hooks.
|
||||
HookPath string
|
||||
|
||||
// Maximum number of concurrent incoming connections (0 = no limit), and
|
||||
// how many there are right now.
|
||||
maxConns int64
|
||||
activeConns atomic.Int64
|
||||
|
||||
// When we last logged about reaching the connection limit (Unix time).
|
||||
connLimitLogged atomic.Int64
|
||||
}
|
||||
|
||||
// NewServer returns a new empty Server.
|
||||
@@ -132,6 +148,7 @@ func NewServer() *Server {
|
||||
|
||||
connTimeout: 20 * time.Minute,
|
||||
commandTimeout: 1 * time.Minute,
|
||||
maxConns: connLimit(),
|
||||
localDomains: &set.String{},
|
||||
authr: authr,
|
||||
aliasesR: aliasesR,
|
||||
@@ -331,6 +348,12 @@ func (s *Server) ListenAndServe() {
|
||||
|
||||
go s.periodicallyReload()
|
||||
|
||||
if s.maxConns > 0 {
|
||||
log.Infof("Incoming connection limit: %d", s.maxConns)
|
||||
} else {
|
||||
log.Infof("No incoming connection limit")
|
||||
}
|
||||
|
||||
for m, addrs := range s.addrs {
|
||||
for _, addr := range addrs {
|
||||
l, err := net.Listen("tcp", addr)
|
||||
@@ -394,6 +417,23 @@ func (s *Server) serve(l net.Listener, mode SocketMode) {
|
||||
}
|
||||
backoff = acceptMinBackoff
|
||||
|
||||
// Close the connection if we're over the limit. Log at most once a
|
||||
// minute (across all listeners) to avoid flooding the logs.
|
||||
if n := s.activeConns.Add(1); s.maxConns > 0 && n > s.maxConns {
|
||||
s.activeConns.Add(-1)
|
||||
connectionsOverLimit.Add(1)
|
||||
conn.Close()
|
||||
|
||||
now, last := time.Now().Unix(), s.connLimitLogged.Load()
|
||||
if now-last >= 60 && s.connLimitLogged.CompareAndSwap(last, now) {
|
||||
log.Errorf("Incoming connection limit (%d) reached, "+
|
||||
"closing new connections (%d closed so far)",
|
||||
s.maxConns, connectionsOverLimit.Value())
|
||||
}
|
||||
continue
|
||||
}
|
||||
activeConnections.Add(1)
|
||||
|
||||
sc := &Conn{
|
||||
hostname: s.Hostname,
|
||||
maxDataSize: s.MaxDataSize,
|
||||
@@ -412,6 +452,32 @@ func (s *Server) serve(l net.Listener, mode SocketMode) {
|
||||
commandTimeout: s.commandTimeout,
|
||||
queue: s.queue,
|
||||
}
|
||||
go sc.Handle()
|
||||
go func() {
|
||||
sc.Handle()
|
||||
s.activeConns.Add(-1)
|
||||
activeConnections.Add(-1)
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// connLimit returns the maximum number of concurrent incoming connections we
|
||||
// should allow, or 0 for no limit.
|
||||
//
|
||||
// It's half of the file descriptor limit, to leave room for everything else
|
||||
// that needs them (queue files, outgoing connections, hooks, etc.), so a large
|
||||
// number of incoming connections doesn't prevent the rest of the server from
|
||||
// working.
|
||||
func connLimit() int64 {
|
||||
var lim syscall.Rlimit
|
||||
if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &lim); err != nil {
|
||||
return 0
|
||||
}
|
||||
|
||||
// The type of lim.Cur varies across platforms (int64 or uint64). Values
|
||||
// out of range (e.g. RLIM_INFINITY) mean there's no meaningful limit.
|
||||
cur := int64(lim.Cur)
|
||||
if cur <= 0 || cur > math.MaxInt32 {
|
||||
return 0
|
||||
}
|
||||
return cur / 2
|
||||
}
|
||||
|
||||
+105
-41
@@ -1,11 +1,11 @@
|
||||
package smtpsrv
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/fs"
|
||||
"net"
|
||||
"net/smtp"
|
||||
@@ -575,75 +575,139 @@ MC4CAQAwBQYDK2VwBCIEID6bjSoiW6g6NJA67RNl0SZ7zpylVOq9w/VGAXF5whnS
|
||||
}
|
||||
}
|
||||
|
||||
// scriptedListener is a net.Listener that returns a predefined sequence of
|
||||
// results from Accept, and net.ErrClosed once they run out.
|
||||
type scriptedListener struct {
|
||||
results []acceptResult
|
||||
}
|
||||
// chanListener is a net.Listener that returns the results sent to it from
|
||||
// Accept, and net.ErrClosed once the channel is closed.
|
||||
type chanListener chan acceptResult
|
||||
|
||||
type acceptResult struct {
|
||||
conn net.Conn
|
||||
err error
|
||||
}
|
||||
|
||||
func (l *scriptedListener) Accept() (net.Conn, error) {
|
||||
if len(l.results) == 0 {
|
||||
func (l chanListener) Accept() (net.Conn, error) {
|
||||
r, ok := <-l
|
||||
if !ok {
|
||||
return nil, net.ErrClosed
|
||||
}
|
||||
r := l.results[0]
|
||||
l.results = l.results[1:]
|
||||
return r.conn, r.err
|
||||
}
|
||||
|
||||
func (l *scriptedListener) Close() error { return nil }
|
||||
func (l *scriptedListener) Addr() net.Addr { return &net.TCPAddr{} }
|
||||
func (l chanListener) Close() error { return nil }
|
||||
func (l chanListener) Addr() net.Addr { return &net.TCPAddr{} }
|
||||
|
||||
func TestAcceptErrors(t *testing.T) {
|
||||
srvConn, cliConn := net.Pipe()
|
||||
defer cliConn.Close()
|
||||
|
||||
emfile := &net.OpError{Op: "accept", Net: "tcp",
|
||||
Err: os.NewSyscallError("accept4", syscall.EMFILE)}
|
||||
l := &scriptedListener{results: []acceptResult{
|
||||
{err: emfile},
|
||||
{err: emfile},
|
||||
{conn: srvConn},
|
||||
{err: emfile},
|
||||
}}
|
||||
|
||||
s := NewServer()
|
||||
s.Hostname = "acceptsrv"
|
||||
prevErrors := acceptErrors.Value()
|
||||
|
||||
// serve should retry on errors, and return once the listener reports
|
||||
// net.ErrClosed.
|
||||
// serveInBackground runs s.serve(l) in a goroutine, and returns a channel
|
||||
// that is closed once it returns.
|
||||
func serveInBackground(s *Server, l net.Listener) chan struct{} {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
s.serve(l, ModeSMTP)
|
||||
close(done)
|
||||
}()
|
||||
return done
|
||||
}
|
||||
|
||||
// The connection we got in between the errors must have been handled.
|
||||
// net.Pipe is synchronous, so this also unblocks the server's write.
|
||||
greeting, err := textproto.NewReader(bufio.NewReader(cliConn)).ReadLine()
|
||||
if err != nil {
|
||||
t.Fatalf("error reading greeting: %v", err)
|
||||
}
|
||||
if greeting != "220 acceptsrv ESMTP chasquid" {
|
||||
t.Errorf("unexpected greeting: %q", greeting)
|
||||
}
|
||||
|
||||
func waitForServe(t *testing.T, done chan struct{}) {
|
||||
t.Helper()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatalf("serve did not return after net.ErrClosed")
|
||||
}
|
||||
}
|
||||
|
||||
// newPipeConn sends a new connection to the listener, and returns the client
|
||||
// side of it.
|
||||
func newPipeConn(l chanListener) *textproto.Conn {
|
||||
srvConn, cliConn := net.Pipe()
|
||||
l <- acceptResult{conn: srvConn}
|
||||
return textproto.NewConn(cliConn)
|
||||
}
|
||||
|
||||
func expectGreeting(t *testing.T, c *textproto.Conn, hostname string) {
|
||||
t.Helper()
|
||||
greeting, err := c.ReadLine()
|
||||
if err != nil {
|
||||
t.Fatalf("error reading greeting: %v", err)
|
||||
}
|
||||
if want := "220 " + hostname + " ESMTP chasquid"; greeting != want {
|
||||
t.Errorf("got greeting %q, expected %q", greeting, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAcceptErrors(t *testing.T) {
|
||||
s := NewServer()
|
||||
s.Hostname = "acceptsrv"
|
||||
prevErrors := acceptErrors.Value()
|
||||
|
||||
emfile := &net.OpError{Op: "accept", Net: "tcp",
|
||||
Err: os.NewSyscallError("accept4", syscall.EMFILE)}
|
||||
srvConn, cliConn := net.Pipe()
|
||||
defer cliConn.Close()
|
||||
|
||||
l := make(chanListener, 4)
|
||||
l <- acceptResult{err: emfile}
|
||||
l <- acceptResult{err: emfile}
|
||||
l <- acceptResult{conn: srvConn}
|
||||
l <- acceptResult{err: emfile}
|
||||
close(l)
|
||||
|
||||
// serve should retry on errors, and return once the listener reports
|
||||
// net.ErrClosed.
|
||||
done := serveInBackground(s, l)
|
||||
|
||||
// The connection we got in between the errors must have been handled.
|
||||
expectGreeting(t, textproto.NewConn(cliConn), "acceptsrv")
|
||||
|
||||
waitForServe(t, done)
|
||||
|
||||
if n := acceptErrors.Value() - prevErrors; n != 3 {
|
||||
t.Errorf("acceptErrors increased by %d, expected 3", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConnectionLimit(t *testing.T) {
|
||||
s := NewServer()
|
||||
s.Hostname = "limitsrv"
|
||||
s.maxConns = 2
|
||||
prevOverLimit := connectionsOverLimit.Value()
|
||||
|
||||
l := make(chanListener)
|
||||
done := serveInBackground(s, l)
|
||||
|
||||
// The first two connections are within the limit.
|
||||
c1 := newPipeConn(l)
|
||||
expectGreeting(t, c1, "limitsrv")
|
||||
c2 := newPipeConn(l)
|
||||
defer c2.Close()
|
||||
expectGreeting(t, c2, "limitsrv")
|
||||
|
||||
// The third one is over the limit, and gets closed without a greeting.
|
||||
c3 := newPipeConn(l)
|
||||
if line, err := c3.ReadLine(); err != io.EOF {
|
||||
t.Errorf("over the limit: got %q, %v; expected EOF", line, err)
|
||||
}
|
||||
if n := connectionsOverLimit.Value() - prevOverLimit; n != 1 {
|
||||
t.Errorf("connectionsOverLimit increased by %d, expected 1", n)
|
||||
}
|
||||
|
||||
// Once a connection finishes, we can accept new ones again.
|
||||
c1.Close()
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for s.activeConns.Load() != 1 {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("activeConns = %d, expected 1", s.activeConns.Load())
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
|
||||
c4 := newPipeConn(l)
|
||||
defer c4.Close()
|
||||
expectGreeting(t, c4, "limitsrv")
|
||||
|
||||
close(l)
|
||||
waitForServe(t, done)
|
||||
}
|
||||
|
||||
//
|
||||
// === Benchmarks ===
|
||||
//
|
||||
|
||||
Reference in New Issue
Block a user