ptls add multiplex support

cmd/gost/route.go

@@ -205,6 +205,7 @@ func parseChainNode(ns string) (nodes []gost.Node, err error) {
 		opts.Host = node.Get("host")
 		opts.BrowserSig = node.Get("browser_sig")
 		opts.Key = node.Get("key")
+		opts.EnableMultiplex = node.GetBool("mux")
 		tr = gost.PTLSTransporter(opts)
 	default:
 		tr = gost.TCPTransporter()

go.mod

@@ -40,6 +40,7 @@ require (
 	github.com/templexxx/cpufeat v0.0.0-20180724012125-cef66df7f161 // indirect
 	github.com/templexxx/xor v0.0.0-20181023030647-4e92f724b73b // indirect
 	github.com/tjfoc/gmsm v1.0.1 // indirect
+	github.com/xtaci/smux v1.5.12
 	github.com/xtaci/tcpraw v1.2.25
 	golang.org/x/crypto v0.0.0-20200221231518-2aa609cf4a9d
 	golang.org/x/net v0.0.0-20190923162816-aa69164e4478

go.sum

@@ -28,16 +28,6 @@ github.com/ginuerzh/gosocks4 v0.0.1 h1:ojDKUyz+uaEeRm2usY1cyQiXTqJqrKxfeE6SVBXq4
 github.com/ginuerzh/gosocks4 v0.0.1/go.mod h1:8SdwBMKjfJ9+BfP2vDJM1jcrgWUbWV6qxBPHHVrwptY=
 github.com/ginuerzh/gosocks5 v0.2.0 h1:K0Ua23U9LU3BZrf3XpGDcs0mP8DiEpa6PJE4TA/MU3s=
 github.com/ginuerzh/gosocks5 v0.2.0/go.mod h1:qp22mr6tH/prEoaN0pFukq76LlScIE+F2rP2ZP5ZHno=
-github.com/ginuerzh/tls-dissector v0.0.1 h1:yF6fIt78TO4CdjiLLn6R8r0XajQJE1Lbnuq6rP8mGW8=
-github.com/ginuerzh/tls-dissector v0.0.1/go.mod h1:u/kbBOqIOgJv39gywuUb3VwyzdZG5DKquOqfToKE6lk=
-github.com/ginuerzh/tls-dissector v0.0.2-0.20200223041816-c0cb3da7ea91 h1:bFBTbZglO4xNVWSLwDEcVKBIurTXGL2sNKi9UuQima4=
-github.com/ginuerzh/tls-dissector v0.0.2-0.20200223041816-c0cb3da7ea91/go.mod h1:YyzP8PQrGwDH/XsfHJXwqdHLwWvBYxu77YVKm0+68f0=
-github.com/ginuerzh/tls-dissector v0.0.2-0.20200223072427-83db9c3e4eb5 h1:pmGmno31njvF5xncoDcDuM8mE1984cxrQ0DeVD4lVfA=
-github.com/ginuerzh/tls-dissector v0.0.2-0.20200223072427-83db9c3e4eb5/go.mod h1:YyzP8PQrGwDH/XsfHJXwqdHLwWvBYxu77YVKm0+68f0=
-github.com/ginuerzh/tls-dissector v0.0.2-0.20200223110639-e9c10af0eb19 h1:t/AZCq8FiVNN+Mx6UmIv7bXj3+OVThg070G8ajZ3wJw=
-github.com/ginuerzh/tls-dissector v0.0.2-0.20200223110639-e9c10af0eb19/go.mod h1:YyzP8PQrGwDH/XsfHJXwqdHLwWvBYxu77YVKm0+68f0=
-github.com/ginuerzh/tls-dissector v0.0.2-0.20200223121713-a8bf02a99d69 h1:h9lREy0OWSTrjweGxduikppA2tCjGPoUj32SVHI3dr0=
-github.com/ginuerzh/tls-dissector v0.0.2-0.20200223121713-a8bf02a99d69/go.mod h1:YyzP8PQrGwDH/XsfHJXwqdHLwWvBYxu77YVKm0+68f0=
 github.com/ginuerzh/tls-dissector v0.0.2-0.20200224064855-24ab2b3a3796 h1:VPXbYRvZUzTemsI7u0FzOnEuHeHwQuMTPXApAu8aeX4=
 github.com/ginuerzh/tls-dissector v0.0.2-0.20200224064855-24ab2b3a3796/go.mod h1:YyzP8PQrGwDH/XsfHJXwqdHLwWvBYxu77YVKm0+68f0=
 github.com/go-gost/relay v0.1.0 h1:UOf2YwAzzaUjY5mdpMuLfSw0vz62iIFYk7oJQkuhlGw=
@@ -83,8 +73,6 @@ github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/ryanuber/go-glob v0.0.0-20170128012129-256dc444b735 h1:7YvPJVmEeFHR1Tj9sZEYsmarJEQfMVYpd/Vyy/A8dqE=
 github.com/ryanuber/go-glob v0.0.0-20170128012129-256dc444b735/go.mod h1:807d1WSdnB0XRJzKNil9Om6lcp/3a0v4qIHxIXzX/Yc=
-github.com/shadowsocks/go-shadowsocks2 v0.0.12-0.20191211020244-a57bc393e43a h1:cxYYZwo6iuuJ/5f8x1mHnya7xvSF3cDrOh8Pqh7RZ/w=
-github.com/shadowsocks/go-shadowsocks2 v0.0.12-0.20191211020244-a57bc393e43a/go.mod h1:/0aFGbhK8mtOX4J/6kTJsPLZlEs9KnzKoWCOCvjd7vk=
 github.com/shadowsocks/go-shadowsocks2 v0.1.0 h1:jQhkjAmMuOTQ7B04bnrRJ5IAoZEwoaXXkKspE7rQ6ck=
 github.com/shadowsocks/go-shadowsocks2 v0.1.0/go.mod h1:/0aFGbhK8mtOX4J/6kTJsPLZlEs9KnzKoWCOCvjd7vk=
 github.com/shadowsocks/shadowsocks-go v0.0.0-20170121203516-97a5c71f80ba h1:tJgNXb3S+RkB4kNPi6N5OmEWe3m+Y3Qs6LUMiNDAONM=
@@ -97,6 +85,8 @@ github.com/templexxx/xor v0.0.0-20181023030647-4e92f724b73b h1:mnG1fcsIB1d/3vbkB
 github.com/templexxx/xor v0.0.0-20181023030647-4e92f724b73b/go.mod h1:5XA7W9S6mni3h5uvOC75dA3m9CCCaS83lltmc0ukdi4=
 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/smux v1.5.12 h1:n9OGjdqQuVZXLh46+L4IR5tR2wvuUFwRABnN/V55bIY=
+github.com/xtaci/smux v1.5.12/go.mod h1:OMlQbT5vcgl2gb49mFkYo6SMf+zP3rcjcwQz7ZU7IGY=
 github.com/xtaci/tcpraw v1.2.25 h1:VDlqo0op17JeXBM6e2G9ocCNLOJcw9mZbobMbJjo0vk=
 github.com/xtaci/tcpraw v1.2.25/go.mod h1:dKyZ2V75s0cZ7cbgJYdxPvms7af0joIeOyx1GgJQbLk=
 golang.org/x/crypto v0.0.0-20181015023909-0c41d7ab0a0e/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=

ptls.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/go-log/log"
 	ss_core "github.com/shadowsocks/go-shadowsocks2/core"
+	"github.com/xtaci/smux"
 )
 
 const TIMESTAMP_TOLERANCE = 180 * time.Second
@@ -30,6 +31,7 @@ type PTLSOptions struct {
 	Key             string
 	Host            string
 	BrowserSig      string
+	EnableMultiplex bool
 }
 
 type ptlsTransporter struct {
@@ -37,10 +39,11 @@ type ptlsTransporter struct {
 	host string
 	browser
 	key             [16]byte
+	enableMultiplex bool
 	connCipher      ss_core.StreamConnCipher
 }
 
-func PTLSTransporter(opts PTLSOptions) Transporter {
+func PTLSTransporter(opts PTLSOptions) (tr Transporter) {
 	var browser browser
 	switch opts.BrowserSig {
 	case "chrome":
@@ -60,13 +63,24 @@ func PTLSTransporter(opts PTLSOptions) Transporter {
 	if err != nil {
 		panic(err)
 	}
-	return &ptlsTransporter{host: host, browser: browser, key: key, connCipher: cipher}
+	tr = &ptlsTransporter{
+		host:            host,
+		browser:         browser,
+		key:             key,
+		enableMultiplex: opts.EnableMultiplex,
+		connCipher:      cipher,
+	}
+	if opts.EnableMultiplex {
+		tr = newMuxTransport(tr)
+	}
+	return tr
 }
 
 func (tr *ptlsTransporter) Handshake(conn net.Conn, options ...HandshakeOption) (net.Conn, error) {
 	var random [32]byte
 	cryptoRandRead(random[:])
-	authPayload := makeAuthenticationPayload(tr.key, random)
+	ai := authenticationInfo{EnableMultiplex: tr.enableMultiplex}
+	authPayload := makeAuthenticationPayload(tr.key, random, ai)
 	ch := tr.browser.composeClientHello(clientHelloFields{
 		random:         random[:],
 		sessionId:      authPayload.ciphertextWithTag[:32],
@@ -106,6 +120,132 @@ func (tr *ptlsTransporter) Handshake(conn net.Conn, options ...HandshakeOption)
 	return tr.connCipher.StreamConn(&ptlsConn{Conn: conn}), nil
 }
 
+type muxTransport struct {
+	originTr   Transporter
+	sessionsMu sync.Mutex
+	sessions   map[string]*muxTransportSession
+}
+
+func newMuxTransport(tr Transporter) *muxTransport {
+	if tr.Multiplex() {
+		panic("cannot multiplex transport that has already support multiplex")
+	}
+	return &muxTransport{
+		originTr: tr,
+		sessions: make(map[string]*muxTransportSession),
+	}
+}
+
+func (tr *muxTransport) Multiplex() bool {
+	return true
+}
+
+func (tr *muxTransport) Dial(addr string, options ...DialOption) (conn net.Conn, err error) {
+	tr.sessionsMu.Lock()
+	defer tr.sessionsMu.Unlock()
+
+	session, ok := tr.sessions[addr]
+	if session != nil && session.IsClosed() {
+		delete(tr.sessions, addr)
+		ok = false // session is dead
+	}
+	if !ok {
+		conn, err = tr.originTr.Dial(addr, options...)
+		if err != nil {
+			return
+		}
+		session = &muxTransportSession{conn: conn}
+		tr.sessions[addr] = session
+	}
+	return session.conn, nil
+}
+
+func (tr *muxTransport) Handshake(conn net.Conn, options ...HandshakeOption) (net.Conn, error) {
+	opts := &HandshakeOptions{}
+	for _, option := range options {
+		option(opts)
+	}
+
+	timeout := opts.Timeout
+	if timeout <= 0 {
+		timeout = HandshakeTimeout
+	}
+
+	tr.sessionsMu.Lock()
+	defer tr.sessionsMu.Unlock()
+
+	conn.SetDeadline(time.Now().Add(timeout))
+	defer conn.SetDeadline(time.Time{})
+
+	session, ok := tr.sessions[opts.Addr]
+	if !ok || session.session == nil {
+		s, err := tr.initSession(opts.Addr, conn, options...)
+		if err != nil {
+			conn.Close()
+			delete(tr.sessions, opts.Addr)
+			return nil, err
+		}
+		session = s
+		tr.sessions[opts.Addr] = session
+	}
+	cc, err := session.GetConn()
+	if err != nil {
+		session.Close()
+		delete(tr.sessions, opts.Addr)
+		return nil, err
+	}
+
+	return cc, nil
+}
+
+func (tr *muxTransport) initSession(addr string, conn net.Conn, options ...HandshakeOption) (*muxTransportSession, error) {
+	prepared, err := tr.originTr.Handshake(conn, options...)
+	if err != nil {
+		return nil, err
+	}
+
+	// stream multiplex
+	smuxConfig := smux.DefaultConfig()
+	smuxConfig.Version = 2
+	session, err := smux.Client(prepared, smuxConfig)
+	if err != nil {
+		return nil, err
+	}
+	return &muxTransportSession{conn: prepared, session: session}, nil
+}
+
+type muxTransportSession struct {
+	conn    net.Conn
+	session *smux.Session
+}
+
+func (s *muxTransportSession) Close() error {
+	if s.session == nil {
+		return nil
+	}
+	return s.session.Close()
+}
+
+func (s *muxTransportSession) IsClosed() bool {
+	if s.session == nil {
+		return true
+	}
+	return s.session.IsClosed()
+}
+
+func (session *muxTransportSession) GetConn() (net.Conn, error) {
+	stream, err := session.session.OpenStream()
+	if err != nil {
+		return nil, err
+	}
+	return &muxTransportStreamConn{conn: session.conn, Stream: stream}, nil
+}
+
+type muxTransportStreamConn struct {
+	conn net.Conn
+	*smux.Stream
+}
+
 type ptlsListener struct {
 	net.Listener
 	opts       PTLSOptions
@@ -147,60 +287,72 @@ func PTLSListener(addr string, opts PTLSOptions) (Listener, error) {
 	return ln, nil
 }
 
-func (ln *ptlsListener) handshake(conn net.Conn) (net.Conn, error) {
+func (ln *ptlsListener) handshake(conn net.Conn) (prepared net.Conn, ai authenticationInfo, err error) {
 	firstPacket, err := readFirstPacket(conn)
 	if err != nil {
 		conn.Close()
-		return nil, fmt.Errorf("failed to read first packet: %s", err)
+		err = fmt.Errorf("failed to read first packet: %s", err)
+		return
 	}
 
 	ch, err := parseClientHello(firstPacket)
 	if err != nil {
 		go ln.redirect(conn, firstPacket)
-		return nil, fmt.Errorf("non (or malformed) ClientHello: %s", err)
+		err = fmt.Errorf("non (or malformed) ClientHello: %s", err)
+		return
 	}
 
-	if err := ln.auth(ch, time.Now()); err != nil {
+	ai, err = ln.auth(ch, time.Now())
+	if err != nil {
 		go ln.redirect(conn, firstPacket)
-		return nil, err
+		return
 	}
 
 	reply, err := composeReply(ch, ln.key[:])
 	if err != nil {
-		return nil, fmt.Errorf("failed to compose TLS reply: %s", err)
+		err = fmt.Errorf("failed to compose TLS reply: %s", err)
+		return
 	}
 	conn.Write(reply)
 	if err != nil {
 		conn.Close()
-		return nil, fmt.Errorf("failed to write TLS reploy: %s", err)
+		err = fmt.Errorf("failed to write TLS reploy: %s", err)
+		return
 	}
 	if Debug {
 		log.Logf("[ptls] handshake completed")
 	}
-	return ln.connCipher.StreamConn(&ptlsConn{Conn: conn}), nil
+	prepared = ln.connCipher.StreamConn(&ptlsConn{Conn: conn})
+	return
 }
 
-func (ln *ptlsListener) auth(ch *ClientHello, serverTime time.Time) error {
+func (ln *ptlsListener) auth(ch *ClientHello, serverTime time.Time) (ai authenticationInfo, err error) {
 	authPayload, err := unmarshalClientHello(ch, ln.key)
 	if err != nil {
-		return fmt.Errorf("failed to unmarshal ClientHello into authenticationPayload")
+		err = fmt.Errorf("failed to unmarshal ClientHello into authenticationPayload")
+		return
 	}
 
 	_, loaded := ln.usedRandom.LoadOrStore(authPayload.random, time.Now().Unix())
 	if loaded {
-		return fmt.Errorf("duplicate random")
+		err = fmt.Errorf("duplicate random")
+		return
 	}
 
-	plaintext, err := aesGCMDecrypt(authPayload.random[0:12], ln.key[:], authPayload.ciphertextWithTag[:])
+	var plaintext []byte
+	plaintext, err = aesGCMDecrypt(authPayload.random[0:12], ln.key[:], authPayload.ciphertextWithTag[:])
 	if err != nil {
-		return err
+		return
 	}
 	timestamp := int64(binary.BigEndian.Uint64(plaintext[1:9]))
 	clientTime := time.Unix(timestamp, 0)
 	if !(clientTime.After(serverTime.Truncate(TIMESTAMP_TOLERANCE)) && clientTime.Before(serverTime.Add(TIMESTAMP_TOLERANCE))) {
-		return fmt.Errorf("timestamp is outside of the accepting window: received timestamp %d", timestamp)
+		err = fmt.Errorf("timestamp is outside of the accepting window: received timestamp %d", timestamp)
+		return
 	}
-	return nil
+	flags := plaintext[10]
+	ai.EnableMultiplex = flags&EnableMultiplex == EnableMultiplex
+	return
 }
 
 func (ln *ptlsListener) cleanupRandoms(deadline time.Time) {
@@ -263,18 +415,52 @@ func (ln *ptlsListener) acceptLoop() {
 			}
 			return
 		}
+		log.Logf("[ptls] %s - %s", conn.RemoteAddr(), ln.Listener.Addr())
 		go func(conn net.Conn) {
-			preparedConn, err := ln.handshake(conn)
+			preparedConn, ai, err := ln.handshake(conn)
 			if err != nil {
 				log.Logf("[ptls] failed to handshake conn from %s: %s", conn.RemoteAddr().String(), err)
 				return
 			}
+			if ai.EnableMultiplex {
+				conf := smux.DefaultConfig()
+				conf.Version = 2
+				sess, err := smux.Server(preparedConn, conf)
+				if err != nil {
+					panic(err)
+				}
+				defer sess.Close()
+				if Debug {
+					log.Logf("[ptls] new smux session created for %s", preparedConn.RemoteAddr())
+				}
+				log.Logf("[ptls] %s <-> %s", conn.RemoteAddr(), ln.Listener.Addr())
+				defer log.Logf("[ptls] %s >-< %s", conn.RemoteAddr(), ln.Listener.Addr())
+				for {
+					if Debug {
+						log.Logf("[ptls] accepting streams")
+					}
+					stream, err := sess.AcceptStream()
+					if err != nil {
+						if err != io.EOF {
+							log.Logf("[ptls] failed to accept stream from %s: %s", preparedConn.RemoteAddr(), err)
+						}
+						return
+					}
+					if Debug {
+						log.Logf("[ptls] accpet stream  from %s", stream.RemoteAddr())
+					}
+					if atomic.LoadInt32(&ln.closed) != 0 {
+						return
+					}
+					ln.conns <- stream
+				}
+			} else {
 				if atomic.LoadInt32(&ln.closed) != 0 {
 					return
 				}
 				ln.conns <- preparedConn
+			}
 		}(conn)
-
 	}
 }
 
@@ -339,22 +525,35 @@ type authenticationPayload struct {
 	ciphertextWithTag [64]byte
 }
 
-func makeAuthenticationPayload(key [16]byte, random [32]byte) (payload authenticationPayload) {
+const (
+	EnableMultiplex = 0x1
+)
+
+func makeAuthenticationPayload(key [16]byte, random [32]byte, ai authenticationInfo) (payload authenticationPayload) {
 	/*
 		Version: 0
 
 		Authentication data (48 bytes):
-		+--------------+-------------+------------+
+		+--------------+-------------+---------+------------+
-		| _Version_    | _Timestamp_ | _reserved_ |
+		| _Version_    | _Timestamp_ | _Flags_ | _reserved_ |
-		+--------------+-------------+------------+
+		+--------------+-------------+---------+------------+
-		| 1 byte (0x0) | 8 bytes     | 39 bytes   |
+		| 1 byte (0x0) | 8 bytes     | 1 byte  | 38 bytes   |
-		+--------------+-------------+------------+
+		+--------------+-------------+---------+------------+
+
+		Flags:
+
+		Currently only EnableMultiplex are specified, that's 0x1
 	*/
 	timestamp := uint64(time.Now().Unix())
 
 	plaintext := make([]byte, 48)
-	plaintext[0] = 0x0
+	plaintext[0] = 0x0                                    // Version
-	binary.BigEndian.PutUint64(plaintext[1:9], timestamp)
+	binary.BigEndian.PutUint64(plaintext[1:9], timestamp) // Timestamp
+	flags := byte(0x0)
+	if ai.EnableMultiplex {
+		flags |= EnableMultiplex
+	}
+	plaintext[10] = flags
 	ciphertextWithTag, _ := aesGCMEncrypt(random[:12], key[:], plaintext)
 	copy(payload.ciphertextWithTag[:], ciphertextWithTag[:])
 	return
@@ -383,6 +582,10 @@ type clientHelloFields struct {
 	sni            []byte
 }
 
+type authenticationInfo struct {
+	EnableMultiplex bool
+}
+
 type browser interface {
 	composeClientHello(clientHelloFields) []byte
 }