feat: transport & obfs refactoring

cmd/client.go

@@ -85,7 +85,7 @@ func client(config *clientConfig) {
 		auth = []byte(config.AuthString)
 	}
 	// Obfuscator
-	var obfuscator core.Obfuscator
+	var obfuscator obfs.Obfuscator
 	if len(config.Obfs) > 0 {
 		obfuscator = obfs.NewXPlusObfuscator([]byte(config.Obfs))
 	}

cmd/server.go

@@ -128,7 +128,7 @@ func server(config *serverConfig) {
 		logrus.WithField("mode", config.Auth.Mode).Fatal("Unsupported authentication mode")
 	}
 	// Obfuscator
-	var obfuscator core.Obfuscator
+	var obfuscator obfs.Obfuscator
 	if len(config.Obfs) > 0 {
 		obfuscator = obfs.NewXPlusObfuscator([]byte(config.Obfs))
 	}

pkg/conns/faketcp/obfs.go

@@ -0,0 +1,94 @@
+package faketcp
+
+import (
+	"github.com/tobyxdd/hysteria/pkg/obfs"
+	"net"
+	"sync"
+	"syscall"
+	"time"
+)
+
+const udpBufferSize = 65535
+
+type ObfsFakeTCPConn struct {
+	orig *TCPConn
+	obfs obfs.Obfuscator
+
+	readBuf    []byte
+	readMutex  sync.Mutex
+	writeBuf   []byte
+	writeMutex sync.Mutex
+}
+
+func NewObfsFakeTCPConn(orig *TCPConn, obfs obfs.Obfuscator) *ObfsFakeTCPConn {
+	return &ObfsFakeTCPConn{
+		orig:     orig,
+		obfs:     obfs,
+		readBuf:  make([]byte, udpBufferSize),
+		writeBuf: make([]byte, udpBufferSize),
+	}
+}
+
+func (c *ObfsFakeTCPConn) ReadFrom(p []byte) (int, net.Addr, error) {
+	for {
+		c.readMutex.Lock()
+		n, addr, err := c.orig.ReadFrom(c.readBuf)
+		if n <= 0 {
+			c.readMutex.Unlock()
+			return 0, addr, err
+		}
+		newN := c.obfs.Deobfuscate(c.readBuf[:n], p)
+		c.readMutex.Unlock()
+		if newN > 0 {
+			// Valid packet
+			return newN, addr, err
+		} else if err != nil {
+			// Not valid and orig.ReadFrom had some error
+			return 0, addr, err
+		}
+	}
+}
+
+func (c *ObfsFakeTCPConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
+	c.writeMutex.Lock()
+	bn := c.obfs.Obfuscate(p, c.writeBuf)
+	_, err = c.orig.WriteTo(c.writeBuf[:bn], addr)
+	c.writeMutex.Unlock()
+	if err != nil {
+		return 0, err
+	} else {
+		return len(p), nil
+	}
+}
+
+func (c *ObfsFakeTCPConn) Close() error {
+	return c.orig.Close()
+}
+
+func (c *ObfsFakeTCPConn) LocalAddr() net.Addr {
+	return c.orig.LocalAddr()
+}
+
+func (c *ObfsFakeTCPConn) SetDeadline(t time.Time) error {
+	return c.orig.SetDeadline(t)
+}
+
+func (c *ObfsFakeTCPConn) SetReadDeadline(t time.Time) error {
+	return c.orig.SetReadDeadline(t)
+}
+
+func (c *ObfsFakeTCPConn) SetWriteDeadline(t time.Time) error {
+	return c.orig.SetWriteDeadline(t)
+}
+
+func (c *ObfsFakeTCPConn) SetReadBuffer(bytes int) error {
+	return c.orig.SetReadBuffer(bytes)
+}
+
+func (c *ObfsFakeTCPConn) SetWriteBuffer(bytes int) error {
+	return c.orig.SetWriteBuffer(bytes)
+}
+
+func (c *ObfsFakeTCPConn) SyscallConn() (syscall.RawConn, error) {
+	return c.orig.SyscallConn()
+}

pkg/conns/udp/obfs.go

@@ -0,0 +1,99 @@
+package udp
+
+import (
+	"github.com/tobyxdd/hysteria/pkg/obfs"
+	"net"
+	"os"
+	"sync"
+	"syscall"
+	"time"
+)
+
+const udpBufferSize = 65535
+
+type ObfsUDPConn struct {
+	orig *net.UDPConn
+	obfs obfs.Obfuscator
+
+	readBuf    []byte
+	readMutex  sync.Mutex
+	writeBuf   []byte
+	writeMutex sync.Mutex
+}
+
+func NewObfsUDPConn(orig *net.UDPConn, obfs obfs.Obfuscator) *ObfsUDPConn {
+	return &ObfsUDPConn{
+		orig:     orig,
+		obfs:     obfs,
+		readBuf:  make([]byte, udpBufferSize),
+		writeBuf: make([]byte, udpBufferSize),
+	}
+}
+
+func (c *ObfsUDPConn) ReadFrom(p []byte) (int, net.Addr, error) {
+	for {
+		c.readMutex.Lock()
+		n, addr, err := c.orig.ReadFrom(c.readBuf)
+		if n <= 0 {
+			c.readMutex.Unlock()
+			return 0, addr, err
+		}
+		newN := c.obfs.Deobfuscate(c.readBuf[:n], p)
+		c.readMutex.Unlock()
+		if newN > 0 {
+			// Valid packet
+			return newN, addr, err
+		} else if err != nil {
+			// Not valid and orig.ReadFrom had some error
+			return 0, addr, err
+		}
+	}
+}
+
+func (c *ObfsUDPConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
+	c.writeMutex.Lock()
+	bn := c.obfs.Obfuscate(p, c.writeBuf)
+	_, err = c.orig.WriteTo(c.writeBuf[:bn], addr)
+	c.writeMutex.Unlock()
+	if err != nil {
+		return 0, err
+	} else {
+		return len(p), nil
+	}
+}
+
+func (c *ObfsUDPConn) Close() error {
+	return c.orig.Close()
+}
+
+func (c *ObfsUDPConn) LocalAddr() net.Addr {
+	return c.orig.LocalAddr()
+}
+
+func (c *ObfsUDPConn) SetDeadline(t time.Time) error {
+	return c.orig.SetDeadline(t)
+}
+
+func (c *ObfsUDPConn) SetReadDeadline(t time.Time) error {
+	return c.orig.SetReadDeadline(t)
+}
+
+func (c *ObfsUDPConn) SetWriteDeadline(t time.Time) error {
+	return c.orig.SetWriteDeadline(t)
+}
+
+func (c *ObfsUDPConn) SetReadBuffer(bytes int) error {
+	return c.orig.SetReadBuffer(bytes)
+}
+
+func (c *ObfsUDPConn) SetWriteBuffer(bytes int) error {
+	return c.orig.SetWriteBuffer(bytes)
+}
+
+func (c *ObfsUDPConn) SyscallConn() (syscall.RawConn, error) {
+	return c.orig.SyscallConn()
+}
+
+func (c *ObfsUDPConn) File() (f *os.File, err error) {
+	return c.orig.File()
+}

pkg/core/client.go

@@ -9,6 +9,7 @@ import (
 	"github.com/lucas-clemente/quic-go"
 	"github.com/lucas-clemente/quic-go/congestion"
 	"github.com/lunixbochs/struc"
+	"github.com/tobyxdd/hysteria/pkg/obfs"
 	transport2 "github.com/tobyxdd/hysteria/pkg/transport"
 	"github.com/tobyxdd/hysteria/pkg/utils"
 	"net"
@@ -30,7 +31,7 @@ type Client struct {
 	sendBPS, recvBPS  uint64
 	auth              []byte
 	congestionFactory CongestionFactory
-	obfuscator        Obfuscator
+	obfuscator        obfs.Obfuscator
 
 	tlsConfig  *tls.Config
 	quicConfig *quic.Config
@@ -45,7 +46,7 @@ type Client struct {
 
 func NewClient(serverAddr string, protocol string, auth []byte, tlsConfig *tls.Config, quicConfig *quic.Config,
 	transport transport2.Transport, sendBPS uint64, recvBPS uint64, congestionFactory CongestionFactory,
-	obfuscator Obfuscator) (*Client, error) {
+	obfuscator obfs.Obfuscator) (*Client, error) {
 	c := &Client{
 		transport:         transport,
 		serverAddr:        serverAddr,
@@ -69,29 +70,9 @@ func (c *Client) connectToServer() error {
 	if err != nil {
 		return err
 	}
-	var pktConn net.PacketConn
-	if len(c.protocol) == 0 || c.protocol == "udp" {
-		udpConn, err := c.transport.QUICListenUDP(nil)
-		if err != nil {
-			return err
-		}
-		if c.obfuscator != nil {
-			pktConn = newObfsUDPConn(udpConn, c.obfuscator)
-		} else {
-			pktConn = udpConn
-		}
-	} else if c.protocol == "faketcp" {
-		ftcpConn, err := c.transport.QUICDialFakeTCP(c.serverAddr)
-		if err != nil {
-			return err
-		}
-		if c.obfuscator != nil {
-			pktConn = newObfsFakeTCPConn(ftcpConn, c.obfuscator)
-		} else {
-			pktConn = ftcpConn
-		}
-	} else {
-		return fmt.Errorf("unsupported protocol: %s", c.protocol)
+	pktConn, err := c.transport.QUICPacketConn(c.protocol, false, "", c.serverAddr, c.obfuscator)
+	if err != nil {
+		return err
 	}
 	qs, err := quic.Dial(pktConn, serverUDPAddr, c.serverAddr, c.tlsConfig, c.quicConfig)
 	if err != nil {

pkg/core/obfs.go

@@ -1,185 +0,0 @@
-package core
-
-import (
-	"github.com/tobyxdd/hysteria/pkg/faketcp"
-	"net"
-	"os"
-	"sync"
-	"syscall"
-	"time"
-)
-
-type Obfuscator interface {
-	Deobfuscate(in []byte, out []byte) int
-	Obfuscate(in []byte, out []byte) int
-}
-
-type obfsUDPConn struct {
-	orig *net.UDPConn
-	obfs Obfuscator
-
-	readBuf    []byte
-	readMutex  sync.Mutex
-	writeBuf   []byte
-	writeMutex sync.Mutex
-}
-
-func newObfsUDPConn(orig *net.UDPConn, obfs Obfuscator) *obfsUDPConn {
-	return &obfsUDPConn{
-		orig:     orig,
-		obfs:     obfs,
-		readBuf:  make([]byte, udpBufferSize),
-		writeBuf: make([]byte, udpBufferSize),
-	}
-}
-
-func (c *obfsUDPConn) ReadFrom(p []byte) (int, net.Addr, error) {
-	for {
-		c.readMutex.Lock()
-		n, addr, err := c.orig.ReadFrom(c.readBuf)
-		if n <= 0 {
-			c.readMutex.Unlock()
-			return 0, addr, err
-		}
-		newN := c.obfs.Deobfuscate(c.readBuf[:n], p)
-		c.readMutex.Unlock()
-		if newN > 0 {
-			// Valid packet
-			return newN, addr, err
-		} else if err != nil {
-			// Not valid and orig.ReadFrom had some error
-			return 0, addr, err
-		}
-	}
-}
-
-func (c *obfsUDPConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
-	c.writeMutex.Lock()
-	bn := c.obfs.Obfuscate(p, c.writeBuf)
-	_, err = c.orig.WriteTo(c.writeBuf[:bn], addr)
-	c.writeMutex.Unlock()
-	if err != nil {
-		return 0, err
-	} else {
-		return len(p), nil
-	}
-}
-
-func (c *obfsUDPConn) Close() error {
-	return c.orig.Close()
-}
-
-func (c *obfsUDPConn) LocalAddr() net.Addr {
-	return c.orig.LocalAddr()
-}
-
-func (c *obfsUDPConn) SetDeadline(t time.Time) error {
-	return c.orig.SetDeadline(t)
-}
-
-func (c *obfsUDPConn) SetReadDeadline(t time.Time) error {
-	return c.orig.SetReadDeadline(t)
-}
-
-func (c *obfsUDPConn) SetWriteDeadline(t time.Time) error {
-	return c.orig.SetWriteDeadline(t)
-}
-
-func (c *obfsUDPConn) SetReadBuffer(bytes int) error {
-	return c.orig.SetReadBuffer(bytes)
-}
-
-func (c *obfsUDPConn) SetWriteBuffer(bytes int) error {
-	return c.orig.SetWriteBuffer(bytes)
-}
-
-func (c *obfsUDPConn) SyscallConn() (syscall.RawConn, error) {
-	return c.orig.SyscallConn()
-}
-
-func (c *obfsUDPConn) File() (f *os.File, err error) {
-	return c.orig.File()
-}
-
-type obfsFakeTCPConn struct {
-	orig *faketcp.TCPConn
-	obfs Obfuscator
-
-	readBuf    []byte
-	readMutex  sync.Mutex
-	writeBuf   []byte
-	writeMutex sync.Mutex
-}
-
-func newObfsFakeTCPConn(orig *faketcp.TCPConn, obfs Obfuscator) *obfsFakeTCPConn {
-	return &obfsFakeTCPConn{
-		orig:     orig,
-		obfs:     obfs,
-		readBuf:  make([]byte, udpBufferSize),
-		writeBuf: make([]byte, udpBufferSize),
-	}
-}
-
-func (c *obfsFakeTCPConn) ReadFrom(p []byte) (int, net.Addr, error) {
-	for {
-		c.readMutex.Lock()
-		n, addr, err := c.orig.ReadFrom(c.readBuf)
-		if n <= 0 {
-			c.readMutex.Unlock()
-			return 0, addr, err
-		}
-		newN := c.obfs.Deobfuscate(c.readBuf[:n], p)
-		c.readMutex.Unlock()
-		if newN > 0 {
-			// Valid packet
-			return newN, addr, err
-		} else if err != nil {
-			// Not valid and orig.ReadFrom had some error
-			return 0, addr, err
-		}
-	}
-}
-
-func (c *obfsFakeTCPConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
-	c.writeMutex.Lock()
-	bn := c.obfs.Obfuscate(p, c.writeBuf)
-	_, err = c.orig.WriteTo(c.writeBuf[:bn], addr)
-	c.writeMutex.Unlock()
-	if err != nil {
-		return 0, err
-	} else {
-		return len(p), nil
-	}
-}
-
-func (c *obfsFakeTCPConn) Close() error {
-	return c.orig.Close()
-}
-
-func (c *obfsFakeTCPConn) LocalAddr() net.Addr {
-	return c.orig.LocalAddr()
-}
-
-func (c *obfsFakeTCPConn) SetDeadline(t time.Time) error {
-	return c.orig.SetDeadline(t)
-}
-
-func (c *obfsFakeTCPConn) SetReadDeadline(t time.Time) error {
-	return c.orig.SetReadDeadline(t)
-}
-
-func (c *obfsFakeTCPConn) SetWriteDeadline(t time.Time) error {
-	return c.orig.SetWriteDeadline(t)
-}
-
-func (c *obfsFakeTCPConn) SetReadBuffer(bytes int) error {
-	return c.orig.SetReadBuffer(bytes)
-}
-
-func (c *obfsFakeTCPConn) SetWriteBuffer(bytes int) error {
-	return c.orig.SetWriteBuffer(bytes)
-}
-
-func (c *obfsFakeTCPConn) SyscallConn() (syscall.RawConn, error) {
-	return c.orig.SyscallConn()
-}

pkg/core/server.go

@@ -9,6 +9,7 @@ import (
 	"github.com/lunixbochs/struc"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/tobyxdd/hysteria/pkg/acl"
+	"github.com/tobyxdd/hysteria/pkg/obfs"
 	transport2 "github.com/tobyxdd/hysteria/pkg/transport"
 	"net"
 )
@@ -40,35 +41,11 @@ type Server struct {
 
 func NewServer(addr string, protocol string, tlsConfig *tls.Config, quicConfig *quic.Config, transport transport2.Transport,
 	sendBPS uint64, recvBPS uint64, congestionFactory CongestionFactory, disableUDP bool, aclEngine *acl.Engine,
-	obfuscator Obfuscator, authFunc AuthFunc, tcpRequestFunc TCPRequestFunc, tcpErrorFunc TCPErrorFunc,
+	obfuscator obfs.Obfuscator, authFunc AuthFunc, tcpRequestFunc TCPRequestFunc, tcpErrorFunc TCPErrorFunc,
 	udpRequestFunc UDPRequestFunc, udpErrorFunc UDPErrorFunc, promRegistry *prometheus.Registry) (*Server, error) {
-	var pktConn net.PacketConn
-	if len(protocol) == 0 || protocol == "udp" {
-		udpAddr, err := transport.QUICResolveUDPAddr(addr)
-		if err != nil {
-			return nil, err
-		}
-		udpConn, err := transport.QUICListenUDP(udpAddr)
-		if err != nil {
-			return nil, err
-		}
-		if obfuscator != nil {
-			pktConn = newObfsUDPConn(udpConn, obfuscator)
-		} else {
-			pktConn = udpConn
-		}
-	} else if protocol == "faketcp" {
-		ftcpConn, err := transport.QUICListenFakeTCP(addr)
-		if err != nil {
-			return nil, err
-		}
-		if obfuscator != nil {
-			pktConn = newObfsFakeTCPConn(ftcpConn, obfuscator)
-		} else {
-			pktConn = ftcpConn
-		}
-	} else {
-		return nil, fmt.Errorf("unsupported protocol: %s", protocol)
+	pktConn, err := transport.QUICPacketConn(protocol, true, addr, "", obfuscator)
+	if err != nil {
+		return nil, err
 	}
 	listener, err := quic.Listen(pktConn, tlsConfig, quicConfig)
 	if err != nil {

pkg/obfs/obfs.go

@@ -0,0 +1,6 @@
+package obfs
+
+type Obfuscator interface {
+	Deobfuscate(in []byte, out []byte) int
+	Obfuscate(in []byte, out []byte) int
+}

pkg/transport/transport.go

@@ -1,16 +1,17 @@
 package transport
 
 import (
-	"github.com/tobyxdd/hysteria/pkg/faketcp"
+	"fmt"
+	"github.com/tobyxdd/hysteria/pkg/conns/faketcp"
+	"github.com/tobyxdd/hysteria/pkg/conns/udp"
+	"github.com/tobyxdd/hysteria/pkg/obfs"
 	"net"
 	"time"
 )
 
 type Transport interface {
 	QUICResolveUDPAddr(address string) (*net.UDPAddr, error)
-	QUICListenUDP(laddr *net.UDPAddr) (*net.UDPConn, error)
-	QUICListenFakeTCP(address string) (*faketcp.TCPConn, error)
-	QUICDialFakeTCP(address string) (*faketcp.TCPConn, error)
+	QUICPacketConn(proto string, server bool, laddr, raddr string, obfs obfs.Obfuscator) (net.PacketConn, error)
 
 	LocalResolveIPAddr(address string) (*net.IPAddr, error)
 	LocalResolveTCPAddr(address string) (*net.TCPAddr, error)
@@ -39,16 +40,49 @@ func (t *defaultTransport) QUICResolveUDPAddr(address string) (*net.UDPAddr, err
 	return net.ResolveUDPAddr("udp", address)
 }
 
-func (t *defaultTransport) QUICListenUDP(laddr *net.UDPAddr) (*net.UDPConn, error) {
-	return net.ListenUDP("udp", laddr)
-}
-
-func (t *defaultTransport) QUICListenFakeTCP(address string) (*faketcp.TCPConn, error) {
-	return faketcp.Listen("tcp", address)
-}
-
-func (t *defaultTransport) QUICDialFakeTCP(address string) (*faketcp.TCPConn, error) {
-	return faketcp.Dial("tcp", address)
+func (t *defaultTransport) QUICPacketConn(proto string, server bool, laddr, raddr string, obfs obfs.Obfuscator) (net.PacketConn, error) {
+	if len(proto) == 0 || proto == "udp" {
+		var laddrU *net.UDPAddr
+		if len(laddr) > 0 {
+			var err error
+			laddrU, err = t.QUICResolveUDPAddr(laddr)
+			if err != nil {
+				return nil, err
+			}
+		}
+		conn, err := net.ListenUDP("udp", laddrU)
+		if err != nil {
+			return nil, err
+		}
+		if obfs != nil {
+			oc := udp.NewObfsUDPConn(conn, obfs)
+			return oc, nil
+		} else {
+			return conn, nil
+		}
+	} else if proto == "faketcp" {
+		var conn *faketcp.TCPConn
+		var err error
+		if server {
+			conn, err = faketcp.Listen("tcp", laddr)
+			if err != nil {
+				return nil, err
+			}
+		} else {
+			conn, err = faketcp.Dial("tcp", raddr)
+			if err != nil {
+				return nil, err
+			}
+		}
+		if obfs != nil {
+			oc := faketcp.NewObfsFakeTCPConn(conn, obfs)
+			return oc, nil
+		} else {
+			return conn, nil
+		}
+	} else {
+		return nil, fmt.Errorf("unsupported protocol: %s", proto)
+	}
 }
 
 func (t *defaultTransport) LocalResolveIPAddr(address string) (*net.IPAddr, error) {