commit 70126177fbdf5b1fa4977f2fc26f624641708098 Author: David Fifield david@bamsoftware.com Date: Tue Jan 28 02:32:02 2020 -0700
Turbo Tunnel client and server.
The client opts into turbotunnel mode by sending a magic token at the beginning of each WebSocket connection (before sending even the ClientID). The token is just a random byte string I generated. The server peeks at the token and, if it matches, uses turbotunnel mode. Otherwise, it unreads the token and continues in the old one-session-per-WebSocket mode. --- client/lib/snowflake.go | 100 +++++++++++++++---- client/lib/turbotunnel.go | 68 +++++++++++++ common/turbotunnel/consts.go | 4 + go.mod | 4 +- go.sum | 21 +++- server/server.go | 231 +++++++++++++++++++++++++++++++++++++++++-- 6 files changed, 399 insertions(+), 29 deletions(-)
diff --git a/client/lib/snowflake.go b/client/lib/snowflake.go index 409ce14..4b7dd4d 100644 --- a/client/lib/snowflake.go +++ b/client/lib/snowflake.go @@ -1,11 +1,16 @@ package lib
import ( + "context" "errors" "io" "log" "net" "time" + + "git.torproject.org/pluggable-transports/snowflake.git/common/turbotunnel" + "github.com/xtaci/kcp-go/v5" + "github.com/xtaci/smux" )
const ( @@ -13,43 +18,98 @@ const ( SnowflakeTimeout = 30 * time.Second )
+type dummyAddr struct{} + +func (addr dummyAddr) Network() string { return "dummy" } +func (addr dummyAddr) String() string { return "dummy" } + // Given an accepted SOCKS connection, establish a WebRTC connection to the // remote peer and exchange traffic. func Handler(socks net.Conn, snowflakes SnowflakeCollector) error { - // Obtain an available WebRTC remote. May block. - snowflake := snowflakes.Pop() - if nil == snowflake { - return errors.New("handler: Received invalid Snowflake") + clientID := turbotunnel.NewClientID() + + // We build a persistent KCP session on a sequence of ephemeral WebRTC + // connections. This dialContext tells RedialPacketConn how to get a new + // WebRTC connection when the previous one dies. Inside each WebRTC + // connection, we use EncapsulationPacketConn to encode packets into a + // stream. + dialContext := func(ctx context.Context) (net.PacketConn, error) { + log.Printf("redialing on same connection") + // Obtain an available WebRTC remote. May block. + conn := snowflakes.Pop() + if conn == nil { + return nil, errors.New("handler: Received invalid Snowflake") + } + log.Println("---- Handler: snowflake assigned ----") + // Send the magic Turbo Tunnel token. + _, err := conn.Write(turbotunnel.Token[:]) + if err != nil { + return nil, err + } + // Send ClientID prefix. + _, err = conn.Write(clientID[:]) + if err != nil { + return nil, err + } + return NewEncapsulationPacketConn(dummyAddr{}, dummyAddr{}, conn), nil } - defer snowflake.Close() - log.Println("---- Handler: snowflake assigned ----") + pconn := turbotunnel.NewRedialPacketConn(dummyAddr{}, dummyAddr{}, dialContext) + defer pconn.Close()
- go func() { - // When WebRTC resets, close the SOCKS connection too. - snowflake.WaitForReset() - socks.Close() - }() + // conn is built on the underlying RedialPacketConn—when one WebRTC + // connection dies, another one will be found to take its place. The + // sequence of packets across multiple WebRTC connections drives the KCP + // engine. + conn, err := kcp.NewConn2(dummyAddr{}, nil, 0, 0, pconn) + if err != nil { + return err + } + defer conn.Close() + // Permit coalescing the payloads of consecutive sends. + conn.SetStreamMode(true) + // Disable the dynamic congestion window (limit only by the + // maximum of local and remote static windows). + conn.SetNoDelay( + 0, // default nodelay + 0, // default interval + 0, // default resend + 1, // nc=1 => congestion window off + ) + // On the KCP connection we overlay an smux session and stream. + smuxConfig := smux.DefaultConfig() + smuxConfig.Version = 2 + smuxConfig.KeepAliveTimeout = 10 * time.Minute + sess, err := smux.Client(conn, smuxConfig) + if err != nil { + return err + } + defer sess.Close() + stream, err := sess.OpenStream() + if err != nil { + return err + } + defer stream.Close()
- // Begin exchanging data. Either WebRTC or localhost SOCKS will close first. - // In eithercase, this closes the handler and induces a new handler. - copyLoop(socks, snowflake) - log.Println("---- Handler: closed ---") + // Begin exchanging data. + log.Printf("---- Handler: begin stream %v ---", stream.ID()) + copyLoop(socks, stream) + log.Printf("---- Handler: closed stream %v ---", stream.ID()) return nil }
// Exchanges bytes between two ReadWriters. -// (In this case, between a SOCKS and WebRTC connection.) -func copyLoop(socks, webRTC io.ReadWriter) { +// (In this case, between a SOCKS connection and smux stream.) +func copyLoop(socks, stream io.ReadWriter) { done := make(chan struct{}, 2) go func() { - if _, err := io.Copy(socks, webRTC); err != nil { + if _, err := io.Copy(socks, stream); err != nil { log.Printf("copying WebRTC to SOCKS resulted in error: %v", err) } done <- struct{}{} }() go func() { - if _, err := io.Copy(webRTC, socks); err != nil { - log.Printf("copying SOCKS to WebRTC resulted in error: %v", err) + if _, err := io.Copy(stream, socks); err != nil { + log.Printf("copying SOCKS to stream resulted in error: %v", err) } done <- struct{}{} }() diff --git a/client/lib/turbotunnel.go b/client/lib/turbotunnel.go new file mode 100644 index 0000000..aad2e6a --- /dev/null +++ b/client/lib/turbotunnel.go @@ -0,0 +1,68 @@ +package lib + +import ( + "bufio" + "errors" + "io" + "net" + "time" + + "git.torproject.org/pluggable-transports/snowflake.git/common/encapsulation" +) + +var errNotImplemented = errors.New("not implemented") + +// EncapsulationPacketConn implements the net.PacketConn interface over an +// io.ReadWriteCloser stream, using the encapsulation package to represent +// packets in a stream. +type EncapsulationPacketConn struct { + io.ReadWriteCloser + localAddr net.Addr + remoteAddr net.Addr + bw *bufio.Writer +} + +// NewEncapsulationPacketConn makes +func NewEncapsulationPacketConn( + localAddr, remoteAddr net.Addr, + conn io.ReadWriteCloser, +) *EncapsulationPacketConn { + return &EncapsulationPacketConn{ + ReadWriteCloser: conn, + localAddr: localAddr, + remoteAddr: remoteAddr, + bw: bufio.NewWriter(conn), + } +} + +// ReadFrom reads an encapsulated packet from the stream. +func (c *EncapsulationPacketConn) ReadFrom(p []byte) (int, net.Addr, error) { + data, err := encapsulation.ReadData(c.ReadWriteCloser) + if err != nil { + return 0, c.remoteAddr, err + } + return copy(p, data), c.remoteAddr, nil +} + +// WriteTo writes an encapsulated packet to the stream. +func (c *EncapsulationPacketConn) WriteTo(p []byte, addr net.Addr) (int, error) { + // addr is ignored. + _, err := encapsulation.WriteData(c.bw, p) + if err == nil { + err = c.bw.Flush() + } + if err != nil { + return 0, err + } + return len(p), nil +} + +// LocalAddr returns the localAddr value that was passed to +// NewEncapsulationPacketConn. +func (c *EncapsulationPacketConn) LocalAddr() net.Addr { + return c.localAddr +} + +func (c *EncapsulationPacketConn) SetDeadline(t time.Time) error { return errNotImplemented } +func (c *EncapsulationPacketConn) SetReadDeadline(t time.Time) error { return errNotImplemented } +func (c *EncapsulationPacketConn) SetWriteDeadline(t time.Time) error { return errNotImplemented } diff --git a/common/turbotunnel/consts.go b/common/turbotunnel/consts.go index 4699d1d..80f70af 100644 --- a/common/turbotunnel/consts.go +++ b/common/turbotunnel/consts.go @@ -6,6 +6,10 @@ package turbotunnel
import "errors"
+// This magic prefix is how a client opts into turbo tunnel mode. It is just a +// randomly generated byte string. +var Token = [8]byte{0x12, 0x93, 0x60, 0x5d, 0x27, 0x81, 0x75, 0xf5} + // The size of receive and send queues. const queueSize = 32
diff --git a/go.mod b/go.mod index 4366d6a..6502651 100644 --- a/go.mod +++ b/go.mod @@ -1,7 +1,5 @@ module git.torproject.org/pluggable-transports/snowflake.git
-go 1.13 - require ( git.torproject.org/pluggable-transports/goptlib.git v1.1.0 github.com/golang/protobuf v1.3.1 // indirect @@ -9,6 +7,8 @@ require ( github.com/pion/sdp/v2 v2.3.4 github.com/pion/webrtc/v2 v2.2.2 github.com/smartystreets/goconvey v1.6.4 + github.com/xtaci/kcp-go/v5 v5.5.12 + github.com/xtaci/smux v1.5.12 golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa golang.org/x/text v0.3.2 // indirect diff --git a/go.sum b/go.sum index 3708fc0..6768e02 100644 --- a/go.sum +++ b/go.sum @@ -21,6 +21,10 @@ github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo= github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= +github.com/klauspost/cpuid v1.2.2 h1:1xAgYebNnsb9LKCdLOvFWtAxGU/33mjJtyOVbmUa0Us= +github.com/klauspost/cpuid v1.2.2/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek= +github.com/klauspost/reedsolomon v1.9.3 h1:N/VzgeMfHmLc+KHMD1UL/tNkfXAt8FnUqlgXGIduwAY= +github.com/klauspost/reedsolomon v1.9.3/go.mod h1:CwCi+NUr9pqSVktrkN+Ondf06rkhYZ/pcNv7fu+8Un4= github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= @@ -82,14 +86,28 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4= github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= +github.com/templexxx/cpu v0.0.1 h1:hY4WdLOgKdc8y13EYklu9OUTXik80BkxHoWvTO6MQQY= +github.com/templexxx/cpu v0.0.1/go.mod h1:w7Tb+7qgcAlIyX4NhLuDKt78AHA5SzPmq0Wj6HiEnnk= +github.com/templexxx/xorsimd v0.4.1 h1:iUZcywbOYDRAZUasAs2eSCUW8eobuZDy0I9FJiORkVg= +github.com/templexxx/xorsimd v0.4.1/go.mod h1:W+ffZz8jJMH2SXwuKu9WhygqBMbFnp14G2fqEr8qaNo= +github.com/tjfoc/gmsm v1.0.1 h1:R11HlqhXkDospckjZEihx9SW/2VW0RgdwrykyWMFOQU= +github.com/tjfoc/gmsm v1.0.1/go.mod h1:XxO4hdhhrzAd+G4CjDqaOkd0hUzmtPR/d3EiBBMn/wc= +github.com/xtaci/kcp-go/v5 v5.5.12 h1:iALGyvti/oBbl1TbVoUpHEUHCorDEb3tEKl1CPY3KXM= +github.com/xtaci/kcp-go/v5 v5.5.12/go.mod h1:H0T/EJ+lPNytnFYsKLH0JHUtiwZjG3KXlTM6c+Q4YUo= +github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae h1:J0GxkO96kL4WF+AIT3M4mfUVinOCPgf2uUWYFUzN0sM= +github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae/go.mod h1:gXtu8J62kEgmN++bm9BVICuT/e8yiLI2KFobd/TRFsE= +github.com/xtaci/smux v1.5.12 h1:n9OGjdqQuVZXLh46+L4IR5tR2wvuUFwRABnN/V55bIY= +github.com/xtaci/smux v1.5.12/go.mod h1:OMlQbT5vcgl2gb49mFkYo6SMf+zP3rcjcwQz7ZU7IGY= golang.org/x/crypto v0.0.0-20190228161510-8dd112bcdc25/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d h1:9FCpayM9Egr1baVnV1SX0H87m+XB0B8S0hAMi99X/3U= golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20191126235420-ef20fe5d7933/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa h1:F+8P+gmewFQYRk6JoLQLwjBCTu3mcIURZfNkVweuRKA= golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f h1:wMNYb4v58l5UBM7MYRLPG6ZhfOqbKu7X5eyFl8ZhKvA= @@ -99,7 +117,8 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h golang.org/x/sys v0.0.0-20190228124157-a34e9553db1e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d h1:+R4KGOnez64A81RvjARKc4UT5/tI9ujCIVX+P5KiHuI= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg= +golang.org/x/sys v0.0.0-20191228213918-04cbcbbfeed8 h1:JA8d3MPx/IToSyXZG/RhwYEtfrKO1Fxrqe8KrkiLXKM= +golang.org/x/sys v0.0.0-20191228213918-04cbcbbfeed8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= diff --git a/server/server.go b/server/server.go index c03e41c..028a24d 100644 --- a/server/server.go +++ b/server/server.go @@ -3,6 +3,8 @@ package main
import ( + "bufio" + "bytes" "crypto/tls" "flag" "fmt" @@ -20,9 +22,13 @@ import ( "time"
pt "git.torproject.org/pluggable-transports/goptlib.git" + "git.torproject.org/pluggable-transports/snowflake.git/common/encapsulation" "git.torproject.org/pluggable-transports/snowflake.git/common/safelog" + "git.torproject.org/pluggable-transports/snowflake.git/common/turbotunnel" "git.torproject.org/pluggable-transports/snowflake.git/common/websocketconn" "github.com/gorilla/websocket" + "github.com/xtaci/kcp-go/v5" + "github.com/xtaci/smux" "golang.org/x/crypto/acme/autocert" "golang.org/x/net/http2" ) @@ -30,6 +36,13 @@ import ( const ptMethodName = "snowflake" const requestTimeout = 10 * time.Second
+// How long to remember outgoing packets for a client, when we don't currently +// have an active WebSocket connection corresponding to that client. Because a +// client session may span multiple WebSocket connections, we keep packets we +// aren't able to send immediately in memory, for a little while but not +// indefinitely. +const clientMapTimeout = 1 * time.Minute + // How long to wait for ListenAndServe or ListenAndServeTLS to return an error // before deciding that it's not going to return. const listenAndServeErrorTimeout = 100 * time.Millisecond @@ -49,8 +62,8 @@ additional HTTP listener on port 80 to work with ACME. flag.PrintDefaults() }
-// Copy from WebSocket to socket and vice versa. -func proxy(local *net.TCPConn, conn *websocketconn.Conn) { +// Copy from one stream to another. +func proxy(local *net.TCPConn, conn net.Conn) { var wg sync.WaitGroup wg.Add(2)
@@ -101,7 +114,23 @@ var upgrader = websocket.Upgrader{ CheckOrigin: func(r *http.Request) bool { return true }, }
-type HTTPHandler struct{} +// overrideReadConn is a net.Conn with an overridden Read method. Compare to +// recordingConn at +// https://dave.cheney.net/2015/05/22/struct-composition-with-go. +type overrideReadConn struct { + net.Conn + io.Reader +} + +func (conn *overrideReadConn) Read(p []byte) (int, error) { + return conn.Reader.Read(p) +} + +type HTTPHandler struct { + // pconn is the adapter layer between stream-oriented WebSocket + // connections and the packet-oriented KCP layer. + pconn *turbotunnel.QueuePacketConn +}
func (handler *HTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { ws, err := upgrader.Upgrade(w, r, nil) @@ -116,15 +145,182 @@ func (handler *HTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // Pass the address of client as the remote address of incoming connection clientIPParam := r.URL.Query().Get("client_ip") addr := clientAddr(clientIPParam) + + var token [len(turbotunnel.Token)]byte + _, err = io.ReadFull(conn, token[:]) + if err != nil { + // Don't bother logging EOF: that happens with an unused + // connection, which clients make frequently as they maintain a + // pool of proxies. + if err != io.EOF { + log.Printf("reading token: %v", err) + } + return + } + + switch { + case bytes.Equal(token[:], turbotunnel.Token[:]): + err = turbotunnelMode(conn, addr, handler.pconn) + default: + // We didn't find a matching token, which means that we are + // dealing with a client that doesn't know about such things. + // "Unread" the token by constructing a new Reader and pass it + // to the old one-session-per-WebSocket mode. + conn2 := &overrideReadConn{Conn: conn, Reader: io.MultiReader(bytes.NewReader(token[:]), conn)} + err = oneshotMode(conn2, addr) + } + if err != nil { + log.Println(err) + return + } +} + +// oneshotMode handles clients that did not send turbotunnel.Token at the start +// of their stream. These clients use the WebSocket as a raw pipe, and expect +// their session to begin and end when this single WebSocket does. +func oneshotMode(conn net.Conn, addr string) error { statsChannel <- addr != "" or, err := pt.DialOr(&ptInfo, addr, ptMethodName) if err != nil { - log.Printf("failed to connect to ORPort: %s", err) - return + return fmt.Errorf("failed to connect to ORPort: %s", err) } defer or.Close()
proxy(or, conn) + + return nil +} + +// turbotunnelMode handles clients that sent turbotunnel.Token at the start of +// their stream. These clients expect to send and receive encapsulated packets, +// with a long-lived session identified by ClientID. +func turbotunnelMode(conn net.Conn, addr string, pconn *turbotunnel.QueuePacketConn) error { + // Read the ClientID prefix. Every packet encapsulated in this WebSocket + // connection pertains to the same ClientID. + var clientID turbotunnel.ClientID + _, err := io.ReadFull(conn, clientID[:]) + if err != nil { + return fmt.Errorf("reading ClientID: %v", err) + } + + // TODO: ClientID-to-client_ip address mapping + // Peek at the first read packet to get the KCP conv ID. + + errCh := make(chan error) + + // The remainder of the WebSocket stream consists of encapsulated + // packets. We read them one by one and feed them into the + // QueuePacketConn on which kcp.ServeConn was set up, which eventually + // leads to KCP-level sessions in the acceptSessions function. + go func() { + for { + p, err := encapsulation.ReadData(conn) + if err != nil { + errCh <- err + break + } + pconn.QueueIncoming(p, clientID) + } + }() + + // At the same time, grab packets addressed to this ClientID and + // encapsulate them into the downstream. + go func() { + // Buffer encapsulation.WriteData operations to keep length + // prefixes in the same send as the data that follows. + bw := bufio.NewWriter(conn) + for p := range pconn.OutgoingQueue(clientID) { + _, err := encapsulation.WriteData(bw, p) + if err == nil { + err = bw.Flush() + } + if err != nil { + errCh <- err + break + } + } + }() + + // Wait until one of the above loops terminates. The closing of the + // WebSocket connection will terminate the other one. + <-errCh + + return nil +} + +// handleStream bidirectionally connects a client stream with the ORPort. +func handleStream(stream net.Conn) error { + // TODO: This is where we need to provide the client IP address. + statsChannel <- false + or, err := pt.DialOr(&ptInfo, "", ptMethodName) + if err != nil { + return fmt.Errorf("connecting to ORPort: %v", err) + } + defer or.Close() + + proxy(or, stream) + + return nil +} + +// acceptStreams layers an smux.Session on the KCP connection and awaits streams +// on it. Passes each stream to handleStream. +func acceptStreams(conn *kcp.UDPSession) error { + smuxConfig := smux.DefaultConfig() + smuxConfig.Version = 2 + smuxConfig.KeepAliveTimeout = 10 * time.Minute + sess, err := smux.Server(conn, smuxConfig) + if err != nil { + return err + } + for { + stream, err := sess.AcceptStream() + if err != nil { + if err, ok := err.(net.Error); ok && err.Temporary() { + continue + } + return err + } + go func() { + defer stream.Close() + err := handleStream(stream) + if err != nil { + log.Printf("handleStream: %v", err) + } + }() + } +} + +// acceptSessions listens for incoming KCP connections and passes them to +// acceptStreams. It is handler.ServeHTTP that provides the network interface +// that drives this function. +func acceptSessions(ln *kcp.Listener) error { + for { + conn, err := ln.AcceptKCP() + if err != nil { + if err, ok := err.(net.Error); ok && err.Temporary() { + continue + } + return err + } + // Permit coalescing the payloads of consecutive sends. + conn.SetStreamMode(true) + // Disable the dynamic congestion window (limit only by the + // maximum of local and remote static windows). + conn.SetNoDelay( + 0, // default nodelay + 0, // default interval + 0, // default resend + 1, // nc=1 => congestion window off + ) + go func() { + defer conn.Close() + err := acceptStreams(conn) + if err != nil { + log.Printf("acceptStreams: %v", err) + } + }() + } }
func initServer(addr *net.TCPAddr, @@ -140,7 +336,12 @@ func initServer(addr *net.TCPAddr, return nil, fmt.Errorf("cannot listen on port %d; configure a port using ServerTransportListenAddr", addr.Port) }
- var handler HTTPHandler + handler := HTTPHandler{ + // pconn is shared among all connections to this server. It + // overlays packet-based client sessions on top of ephemeral + // WebSocket connections. + pconn: turbotunnel.NewQueuePacketConn(addr, clientMapTimeout), + } server := &http.Server{ Addr: addr.String(), Handler: &handler, @@ -176,6 +377,24 @@ func initServer(addr *net.TCPAddr, break }
+ // Start a KCP engine, set up to read and write its packets over the + // WebSocket connections that arrive at the web server. + // handler.ServeHTTP is responsible for encapsulation/decapsulation of + // packets on behalf of KCP. KCP takes those packets and turns them into + // sessions which appear in the acceptSessions function. + ln, err := kcp.ServeConn(nil, 0, 0, handler.pconn) + if err != nil { + server.Close() + return server, err + } + go func() { + defer ln.Close() + err := acceptSessions(ln) + if err != nil { + log.Printf("acceptSessions: %v", err) + } + }() + return server, err }