feat: Add TLS support to gnet

acceptor_unix.go

@@ -56,6 +56,13 @@ func (el *eventloop) accept0(fd int, _ netpoll.IOEvent, _ netpoll.IOFlags) error
 
 		el := el.engine.eventLoops.next(remoteAddr)
 		c := newTCPConn(nfd, el, sa, el.listeners[fd].addr, remoteAddr)
+
+		if el.engine.opts.TLSconfig != nil {
+			if err = c.UpgradeTLS(el.engine.opts.TLSconfig); err != nil {
+				return err
+			}
+		}
+
 		err = el.poller.Trigger(queue.HighPriority, el.register, c)
 		if err != nil {
 			el.getLogger().Errorf("failed to enqueue the accepted socket fd=%d to poller: %v", c.fd, err)
@@ -93,5 +100,12 @@ func (el *eventloop) accept(fd int, ev netpoll.IOEvent, flags netpoll.IOFlags) e
 	}
 
 	c := newTCPConn(nfd, el, sa, el.listeners[fd].addr, remoteAddr)
+
+	if el.engine.opts.TLSconfig != nil {
+		if err = c.UpgradeTLS(el.engine.opts.TLSconfig); err != nil {
+			return err
+		}
+	}
+
 	return el.register0(c)
 }

connection_unix.go

@@ -35,6 +35,7 @@ import (
 	errorx "github.com/panjf2000/gnet/v2/pkg/errors"
 	"github.com/panjf2000/gnet/v2/pkg/logging"
 	bsPool "github.com/panjf2000/gnet/v2/pkg/pool/byteslice"
+	"github.com/panjf2000/gnet/v2/pkg/tls"
 )
 
 type conn struct {
@@ -51,6 +52,7 @@ type conn struct {
 	buffer         []byte                 // buffer for the latest bytes
 	isDatagram     bool                   // UDP protocol
 	opened         bool                   // connection opened event fired
+	tlsconn        *tls.Conn              // tls connection
 	isEOF          bool                   // whether the connection has reached EOF
 }
 
@@ -134,6 +136,15 @@ func (c *conn) open(buf []byte) error {
 	return nil
 }
 
+func (c *conn) writeTLS(data []byte) (n int, err error) {
+	// use tls to encrypt the data before sending it.
+	// tlsconn will call gnet.WriteTCP() to sent the data directly.
+	// If gnetConn.outboundBuffer is not empty, data will be
+	// buffered in gnetConn.outboundBuffer.
+	n, err = c.tlsconn.Write(data)
+	return
+}
+
 func (c *conn) write(data []byte) (n int, err error) {
 	isET := c.loop.engine.opts.EdgeTriggeredIO
 	n = len(data)
@@ -177,6 +188,28 @@ loop:
 	return
 }
 
+func (c *conn) writevTLS(bs [][]byte) (n int, err error) {
+	for _, b := range bs {
+		n += len(b)
+	}
+
+	// use tls to encrypt the data before sending it.
+	// tlsconn will call gnet.WriteTCP() to sent the data directly.
+	// If gnetConn.outboundBuffer is not empty, data will be
+	// buffered in gnetConn.outboundBuffer.
+	sent := 0
+	var sentN int
+	for _, b := range bs {
+		sentN, err = c.tlsconn.Write(b)
+		if sentN < 0 {
+			// the connection is closed (c.loop.closeConn() is called).
+			return sent, err
+		}
+		sent += sentN
+	}
+	return
+}
+
 func (c *conn) writev(bs [][]byte) (n int, err error) {
 	isET := c.loop.engine.opts.EdgeTriggeredIO
 
@@ -255,7 +288,11 @@ func (c *conn) asyncWrite(itf interface{}) (err error) {
 		return net.ErrClosed
 	}
 
-	_, err = c.write(hook.data)
+	if c.tlsconn != nil {
+		_, err = c.writeTLS(hook.data)
+	} else {
+		_, err = c.write(hook.data)
+	}
 	return
 }
 
@@ -276,7 +313,11 @@ func (c *conn) asyncWritev(itf interface{}) (err error) {
 		return net.ErrClosed
 	}
 
-	_, err = c.writev(hook.data)
+	if c.tlsconn != nil {
+		_, err = c.writevTLS(hook.data)
+	} else {
+		_, err = c.writev(hook.data)
+	}
 	return
 }
 
@@ -396,13 +437,25 @@ func (c *conn) Write(p []byte) (int, error) {
 		}
 		return len(p), nil
 	}
+	if c.tlsconn != nil {
+		return c.writeTLS(p)
+	}
+	return c.write(p)
+}
+
+// Expose the plaintext write API which should only be used
+// by tlsconn.Write().
+func (c *conn) WriteTCP(p []byte) (int, error) {
 	return c.write(p)
 }
 
 func (c *conn) Writev(bs [][]byte) (int, error) {
 	if c.isDatagram {
 		return 0, errorx.ErrUnsupportedOp
 	}
+	if c.tlsconn != nil {
+		return c.writevTLS(bs)
+	}
 	return c.writev(bs)
 }
 
@@ -522,3 +575,28 @@ func (*conn) SetReadDeadline(_ time.Time) error {
 func (*conn) SetWriteDeadline(_ time.Time) error {
 	return errorx.ErrUnsupportedOp
 }
+
+func (c *conn) UpgradeTLS(config *tls.Config) (err error) {
+	// TODO: create a sync.pool to manage the TLS connection
+	c.tlsconn = tls.Server(c, config.Clone())
+
+	// It is very likely that the handshake packet was sent before UpgradeTls.
+	// So, the remaining data in the inboundBuffer is treated as handshake data here
+	if c.inboundBuffer.Len() > 0 {
+		head, tail := c.inboundBuffer.Peek(-1)
+		c.tlsconn.RawInputSet(head) //nolint:errcheck
+		c.tlsconn.RawInputSet(tail) //nolint:errcheck
+		c.inboundBuffer.Reset()
+		if err := c.tlsconn.Handshake(); err != nil {
+			return err
+		}
+	}
+
+	// handshake is failed
+	time.AfterFunc(time.Second*5, func() {
+		if c.opened && (c.tlsconn == nil || !c.tlsconn.HandshakeComplete()) {
+			c.Close()
+		}
+	})
+	return err
+}

eventloop_unix.go

@@ -114,6 +114,47 @@ func (el *eventloop) open(c *conn) error {
 	return el.handleAction(c, action)
 }
 
+func (el *eventloop) readTLS(c *conn) error {
+	// Todo: align this method with func (el *eventloop) read(c *conn) error
+	// Since the el.Buffer may contain multiple TLS record,
+	// we process one TLS record in each iteration until no more
+	// TLS records are available
+	for {
+		if err := c.tlsconn.ReadFrame(); err != nil {
+			// Receive error unix.EAGAIN, wait for the next round
+			if err == unix.EAGAIN {
+				c.tlsconn.DataDone()
+				return nil
+			}
+			// If err is io.EOF, it can either the data is drained,
+			// receives a close notify from the client.
+			return el.close(c, os.NewSyscallError("TLS read", err))
+		}
+
+		// load all decrypted data and make it ready for gnet to use
+		c.buffer = c.tlsconn.Data()
+
+		action := el.eventHandler.OnTraffic(c)
+		switch action {
+		case None:
+		case Close:
+			// tls data will be cleaned up in el.closeConn()
+			return el.close(c, nil)
+		case Shutdown:
+			c.tlsconn.DataDone()
+			return errorx.ErrEngineShutdown
+		}
+		_, _ = c.inboundBuffer.Write(c.buffer)
+		c.buffer = c.buffer[:0]
+
+		// all available TLS records are processed
+		if !c.tlsconn.IsRecordCompleted(c.tlsconn.RawInputData()) {
+			c.tlsconn.DataDone()
+			return nil
+		}
+	}
+}
+
 func (el *eventloop) read0(itf interface{}) error {
 	return el.read(itf.(*conn))
 }
@@ -125,6 +166,17 @@ func (el *eventloop) read(c *conn) error {
 		return nil
 	}
 
+	// detected whether kernel TLS RX is enabled
+	// This only happens after TLS handshake is completed.
+	// Therefore, no need to call c.tlsconn.HandshakeComplete().
+	if c.tlsconn != nil && c.tlsconn.IsKTLSRXEnabled() {
+		// attach the gnet eventloop.buffer to tlsconn.rawInput.
+		// So, KTLS can decrypt the data directly to the buffer without memory allocation.
+		// Since data is read through KTLS, there is no need to call unix.read(c.fd, el.buffer)
+		c.tlsconn.RawInputSet(el.buffer) //nolint:errcheck
+		return el.readTLS(c)
+	}
+
 	var recv int
 	isET := el.engine.opts.EdgeTriggeredIO
 loop:
@@ -140,6 +192,29 @@ loop:
 	}
 	recv += n
 
+	if c.tlsconn != nil {
+		// attach the gnet eventloop.buffer to tlsconn.rawInput.
+		c.tlsconn.RawInputSet(el.buffer[:n]) //nolint:errcheck
+		if !c.tlsconn.HandshakeComplete() {
+			// check whether the buffer data is sufficient for a complete TLS record
+			data := c.tlsconn.RawInputData()
+			if !c.tlsconn.IsRecordCompleted(data) {
+				c.tlsconn.DataDone()
+				return nil
+			}
+			if err = c.tlsconn.Handshake(); err != nil {
+				// close will cleanup the TLS data at the end,
+				// so no need to call tlsconn.DataDone()
+				return el.close(c, os.NewSyscallError("TLS handshake", err))
+			}
+			if !c.tlsconn.HandshakeComplete() || len(c.tlsconn.RawInputData()) == 0 { // 握手没成功，或者握手成功，但是没有数据黏包了
+				c.tlsconn.DataDone()
+				return nil
+			}
+		}
+		return el.readTLS(c)
+	}
+
 	c.buffer = el.buffer[:n]
 	action := el.eventHandler.OnTraffic(c)
 	switch action {
@@ -244,6 +319,16 @@ func (el *eventloop) close(c *conn, err error) (rerr error) {
 		return // ignore stale connections
 	}
 
+	// close the TLS connection by sending the alert
+	if c.tlsconn != nil {
+		// close the TLS connection, which will send a close notify to the client
+		c.tlsconn.Close()
+		// Make sure all memory requested from the pool is returned.
+		c.tlsconn.DataCleanUpAfterClose()
+		c.tlsconn = nil
+		// TODO: create a sync.pool to manage the TLS connection
+	}
+
 	// Send residual data in buffer back to the remote before actually closing the connection.
 	for !c.outboundBuffer.IsEmpty() {
 		iov, _ := c.outboundBuffer.Peek(0)

go.mod

@@ -1,6 +1,7 @@
 module github.com/panjf2000/gnet/v2
 
 require (
+	github.com/0-haha/gnet-tls-go1-20 v1.20.2-rc.1
 	github.com/panjf2000/ants/v2 v2.9.0
 	github.com/stretchr/testify v1.8.4
 	github.com/valyala/bytebufferpool v1.0.0
@@ -15,7 +16,8 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
+	golang.org/x/crypto v0.7.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
-go 1.17
+go 1.20

go.sum

@@ -1,3 +1,7 @@
+github.com/0-haha/gnet-tls-go1-20 v1.20.2-rc.1 h1:cYbILYopJYpNQej70cMiYUi2vOE+ZEflt0hypVa3gOw=
+github.com/0-haha/gnet-tls-go1-20 v1.20.2-rc.1/go.mod h1:ZipvhkWAVMtaMixCaqOh9NgQ+u4u1OCzDOohXR33VfU=
+github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
+github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -37,6 +41,8 @@ go.uber.org/zap v1.21.0 h1:WefMeulhovoZ2sYXz7st6K0sLj7bBhpiFaud4r4zST8=
 go.uber.org/zap v1.21.0/go.mod h1:wjWOCqI0f2ZZrJF/UufIOkiC8ii6tm1iqIsLo76RfJw=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.7.0 h1:AvwMYaRytfdeVt3u6mLaxYtErKYjxA2OXjJ1HHq6t3A=
+golang.org/x/crypto v0.7.0/go.mod h1:pYwdfH91IfpZVANVyUOhSIPZaFoJGxTFbZhFTx+dXZU=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=

options.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/panjf2000/gnet/v2/pkg/logging"
+	"github.com/panjf2000/gnet/v2/pkg/tls"
 )
 
 // Option is a function that will set up option.
@@ -109,6 +110,9 @@ type Options struct {
 	// SocketSendBuffer sets the maximum socket send buffer in bytes.
 	SocketSendBuffer int
 
+	// TLSconfig sets the configuration of a TLS connection
+	TLSconfig *tls.Config
+
 	// LogPath the local path where logs will be written, this is the easiest way to set up logging,
 	// gnet instantiates a default uber-go/zap logger with this given log path, you are also allowed to employ
 	// you own logger during the lifetime by implementing the following log.Logger interface.
@@ -255,6 +259,13 @@ func WithMulticastInterfaceIndex(idx int) Option {
 	}
 }
 
+// WithTLS sets the tls configuration which includes the cert, the key, the cipher suite, the rotocol version, and etc.
+func WithTLS(tlsconfig *tls.Config) Option {
+	return func(opts *Options) {
+		opts.TLSconfig = tlsconfig
+	}
+}
+
 // WithEdgeTriggeredIO enables the edge-triggered I/O for the underlying epoll/kqueue event-loop.
 func WithEdgeTriggeredIO(et bool) Option {
 	return func(opts *Options) {