[WIP] Implement pluggable transport interface

pkg/core/README.md

@@ -0,0 +1,15 @@
+# A pluggable transport implementation based on Hysteria
+
+## Hysteria
+[Hysteria](https://github.com/HyNetwork/hysteria) uses a custom version of QUIC protocol ([RFC 9000 - QUIC: A UDP-Based Multiplexed and Secure Transport](https://www.rfc-editor.org/rfc/rfc9000.html)):
+
+* a custom congestion control ([RFC 9002 - QUIC Loss Detection and Congestion Control](https://www.rfc-editor.org/rfc/rfc9002.html))
+* tweaked QUIC parameters
+* an obfuscation layer
+* non-standard transports (e.g. [faketcp](https://github.com/wangyu-/udp2raw))
+
+## Usage
+
+## Implementation
+
+The implementation uses [Pluggable Transport Specification v3.0 - Go Transport API](https://github.com/Pluggable-Transports/Pluggable-Transports-spec/blob/main/releases/PTSpecV3.0/Pluggable%20Transport%20Specification%20v3.0%20-%20Go%20Transport%20API%20v3.0.md)

pkg/core/client.go

@@ -6,18 +6,19 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"math/rand"
+	"net"
+	"strconv"
+	"sync"
+	"time"
+
 	"github.com/lucas-clemente/quic-go"
 	"github.com/lucas-clemente/quic-go/congestion"
 	"github.com/lunixbochs/struc"
 	"github.com/tobyxdd/hysteria/pkg/obfs"
 	"github.com/tobyxdd/hysteria/pkg/pmtud_fix"
 	"github.com/tobyxdd/hysteria/pkg/transport"
 	"github.com/tobyxdd/hysteria/pkg/utils"
-	"math/rand"
-	"net"
-	"strconv"
-	"sync"
-	"time"
 )
 
 var (
@@ -183,6 +184,20 @@ func (c *Client) openStreamWithReconnect() (quic.Connection, quic.Stream, error)
 	return c.quicSession, &wrappedQUICStream{stream}, err
 }
 
+// Implement Pluggable Transport Client interface
+func (c *Client) Dial() (net.Conn, error) {
+	session, stream, err := c.openStreamWithReconnect()
+	if err != nil {
+		return nil, err
+	}
+
+	return &quicConn{
+		Orig:             stream,
+		PseudoLocalAddr:  session.LocalAddr(),
+		PseudoRemoteAddr: session.RemoteAddr(),
+	}, nil
+}
+
 func (c *Client) DialTCP(addr string) (net.Conn, error) {
 	host, port, err := utils.SplitHostPort(addr)
 	if err != nil {

pkg/core/server.go

@@ -5,14 +5,15 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"net"
+
 	"github.com/lucas-clemente/quic-go"
 	"github.com/lunixbochs/struc"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/tobyxdd/hysteria/pkg/acl"
 	"github.com/tobyxdd/hysteria/pkg/obfs"
 	"github.com/tobyxdd/hysteria/pkg/pmtud_fix"
 	"github.com/tobyxdd/hysteria/pkg/transport"
-	"net"
 )
 
 type ConnectFunc func(addr net.Addr, auth []byte, sSend uint64, sRecv uint64) (bool, string)
@@ -42,6 +43,21 @@ type Server struct {
 	listener quic.Listener
 }
 
+type HysteriaTransport struct {
+	addr              string
+	protocol          string
+	tlsConfig         *tls.Config
+	quicConfig        *quic.Config
+	transport         *transport.ServerTransport
+	sendBPS           uint64
+	recvBPS           uint64
+	congestionFactory CongestionFactory
+	disableUDP        bool
+	obfuscator        obfs.Obfuscator
+	connectFunc       ConnectFunc
+	disconnectFunc    DisconnectFunc
+}
+
 func NewServer(addr string, protocol string, tlsConfig *tls.Config, quicConfig *quic.Config, transport *transport.ServerTransport,
 	sendBPS uint64, recvBPS uint64, congestionFactory CongestionFactory, disableUDP bool, aclEngine *acl.Engine,
 	obfuscator obfs.Obfuscator, connectFunc ConnectFunc, disconnectFunc DisconnectFunc,
@@ -92,6 +108,8 @@ func (s *Server) Serve() error {
 	}
 }
 
+// Close closes the listener.
+// Any blocked Accept operations will be unblocked and return errors.
 func (s *Server) Close() error {
 	return s.listener.Close()
 }
@@ -173,3 +191,62 @@ func (s *Server) handleControlStream(cs quic.Connection, stream quic.Stream) ([]
 	}
 	return ch.Auth, ok, vb[0] == protocolVersionV2, nil
 }
+
+// Implement Pluggable Transport Server interface
+func (t *HysteriaTransport) Listen() (net.Listener, error) {
+	listener, err := t.transport.QUICListen(t.protocol, t.addr, t.tlsConfig, t.quicConfig, t.obfuscator)
+	if err != nil {
+		return nil, err
+	}
+	s := &Server{
+		listener:          listener,
+		transport:         t.transport,
+		sendBPS:           t.sendBPS,
+		recvBPS:           t.recvBPS,
+		congestionFactory: t.congestionFactory,
+		disableUDP:        t.disableUDP,
+		connectFunc:       t.connectFunc,
+		disconnectFunc:    t.disconnectFunc,
+	}
+
+	return s, nil
+}
+
+// Addr returns the listener's network address.
+func (s *Server) Addr() net.Addr {
+	return s.listener.Addr()
+}
+
+func (s *Server) Accept() (net.Conn, error) {
+	cs, err := s.listener.Accept(context.Background())
+	if err != nil {
+		return nil, err
+	}
+	// Expect the client to create a control stream to send its own information
+	ctx, ctxCancel := context.WithTimeout(context.Background(), protocolTimeout)
+	stream, err := cs.AcceptStream(ctx)
+	ctxCancel()
+	if err != nil {
+		_ = cs.CloseWithError(closeErrorCodeProtocol, "protocol error")
+		return nil, err
+	}
+	// Handle the control stream
+	_, ok, _, err := s.handleControlStream(cs, stream)
+	if err != nil {
+		_ = cs.CloseWithError(closeErrorCodeProtocol, "protocol error")
+		return nil, err
+	}
+	if !ok {
+		_ = cs.CloseWithError(closeErrorCodeAuth, "auth error")
+		return nil, err
+	}
+
+	// Start accepting streams
+	conn := &quicConn{
+		Orig:             stream,
+		PseudoLocalAddr:  cs.LocalAddr(),
+		PseudoRemoteAddr: cs.RemoteAddr(),
+	}
+
+	return conn, nil
+}