Add r7 using new relayer interface

core/port.go

@@ -17,6 +17,10 @@ const DEFAULT_BUFFER_LIMIT = 64 << 20
 const DEFAULT_UDP_TIMEOUT = 60
 const DEFAULT_UDP_BUFFER_SIZE = 2 << 10
 
+type PortCreator interface {
+	Create(net.Conn) Port
+}
+
 func CloseRead(c net.Conn) error {
 	if c, ok := c.(*net.TCPConn); ok {
 		return c.CloseRead()

proxy/intrinsic/acceptor.go

@@ -0,0 +1,67 @@
+// Copyright (c) 2024 Kai Luo <[email protected]>. All rights reserved.
+
+package intrinsic
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+	"net"
+
+	"github.com/bzEq/bx/core"
+	"github.com/bzEq/bx/core/iovec"
+)
+
+type Acceptor struct {
+	ln *net.TCPListener
+	pc core.PortCreator
+}
+
+func NewAcceptor(ln *net.TCPListener, pc core.PortCreator) *Acceptor {
+	return &Acceptor{ln, pc}
+}
+
+func (self *Acceptor) createPort(c net.Conn) core.Port {
+	return self.pc.Create(c)
+}
+
+func (self *Acceptor) handshake(c net.Conn) (p core.Port, addr net.Addr, err error) {
+	var b iovec.IoVec
+	p = self.createPort(c)
+	err = p.Unpack(&b)
+	if err != nil {
+		return
+	}
+	dec := gob.NewDecoder(&b)
+	var i Intrinsic
+	err = dec.Decode(&i)
+	if err != nil {
+		return
+	}
+	switch i.Func {
+	case RELAY_TCP:
+		var req TCPRequest
+		dec := gob.NewDecoder(bytes.NewBuffer(i.Data))
+		err = dec.Decode(&req)
+		if err != nil {
+			return
+		}
+		addr, err = net.ResolveTCPAddr("tcp", req.Addr)
+	default:
+		err = fmt.Errorf("Unsupported function: %d", i.Func)
+	}
+	return
+}
+
+func (self *Acceptor) Accept() (p core.Port, addr net.Addr, err error) {
+	c, err := self.ln.Accept()
+	if err != nil {
+		return nil, nil, err
+	}
+	p, addr, err = self.handshake(c)
+	if err != nil {
+		c.Close()
+		return
+	}
+	return
+}

proxy/intrinsic/dialer.go

@@ -0,0 +1,64 @@
+// Copyright (c) 2024 Kai Luo <[email protected]>. All rights reserved.
+
+package intrinsic
+
+import (
+	"bytes"
+	"encoding/gob"
+	"net"
+
+	"github.com/bzEq/bx/core"
+	"github.com/bzEq/bx/core/iovec"
+)
+
+type Dialer struct {
+	nextHop *net.TCPAddr
+	pc      core.PortCreator
+}
+
+func NewDialer(raddr *net.TCPAddr, pc core.PortCreator) *Dialer {
+	return &Dialer{raddr, pc}
+}
+
+func (self *Dialer) createPort(c net.Conn) core.Port {
+	return self.pc.Create(c)
+}
+
+func (self *Dialer) handshake(c net.Conn, addr net.Addr) (p core.Port, err error) {
+	p = self.createPort(c)
+	i := Intrinsic{Func: RELAY_TCP}
+	{
+		data := &bytes.Buffer{}
+		req := TCPRequest{Addr: addr.String()}
+		enc := gob.NewEncoder(data)
+		err = enc.Encode(&req)
+		if err != nil {
+			return
+		}
+		i.Data = data.Bytes()
+	}
+	pack := &bytes.Buffer{}
+	enc := gob.NewEncoder(pack)
+	err = enc.Encode(&i)
+	if err != nil {
+		return
+	}
+	err = p.Pack(iovec.FromSlice(pack.Bytes()))
+	if err != nil {
+		return
+	}
+	return
+}
+
+func (self *Dialer) Dial(addr net.Addr) (p core.Port, err error) {
+	c, err := net.Dial(self.nextHop.Network(), self.nextHop.String())
+	if err != nil {
+		return nil, err
+	}
+	p, err = self.handshake(c, addr)
+	if err != nil {
+		c.Close()
+		return
+	}
+	return
+}

proxy/socks5/acceptor.go

@@ -0,0 +1,89 @@
+// Copyright (c) 2024 Kai Luo <[email protected]>. All rights reserved.
+
+package socks5
+
+import (
+	"fmt"
+	"net"
+
+	"github.com/bzEq/bx/core"
+)
+
+type Acceptor struct {
+	ln *net.TCPListener
+	pc core.PortCreator
+}
+
+func NewAcceptor(ln *net.TCPListener, pc core.PortCreator) *Acceptor {
+	return &Acceptor{ln, pc}
+}
+
+func (self *Acceptor) createPort(c net.Conn) core.Port {
+	if self.pc == nil {
+		return core.NewPort(c, nil)
+	}
+	return self.pc.Create(c)
+}
+
+func (self *Acceptor) handshake(c net.Conn) (p core.Port, addr net.Addr, err error) {
+	s := &Socks5Proxy{}
+	err = s.ExchangeMetadata(c)
+	if err != nil {
+		return
+	}
+	var req Request
+	req, err = s.ReceiveRequest(c)
+	if err != nil {
+		return
+	}
+	if req.VER != VER {
+		reply := Reply{
+			VER:      req.VER,
+			REP:      REP_COMMAND_NOT_SUPPORTED,
+			ATYP:     1,
+			BND_ADDR: make([]byte, net.IPv4len),
+		}
+		s.SendReply(c, reply)
+		err = fmt.Errorf("Unsupported SOCKS version: %v", req.VER)
+		return
+	}
+	switch req.CMD {
+	case CMD_CONNECT:
+		reply := Reply{
+			VER:      req.VER,
+			REP:      REP_SUCC,
+			ATYP:     1,
+			BND_ADDR: make([]byte, net.IPv4len),
+		}
+		s.SendReply(c, reply)
+		addr, err = net.ResolveTCPAddr("tcp", s.GetDialAddress(req))
+		if err != nil {
+			return
+		}
+	default:
+		reply := Reply{
+			VER:      req.VER,
+			REP:      REP_COMMAND_NOT_SUPPORTED,
+			ATYP:     1,
+			BND_ADDR: make([]byte, net.IPv4len),
+		}
+		s.SendReply(c, reply)
+		err = fmt.Errorf("Unsupported CMD: %d", req.CMD)
+		return
+	}
+	p = self.createPort(c)
+	return
+}
+
+func (self *Acceptor) Accept() (p core.Port, addr net.Addr, err error) {
+	c, err := self.ln.Accept()
+	if err != nil {
+		return nil, nil, err
+	}
+	p, addr, err = self.handshake(c)
+	if err != nil {
+		c.Close()
+		return
+	}
+	return
+}

r7/main.go

@@ -0,0 +1,64 @@
+// Copyright (c) 2024 Kai Luo <[email protected]>. All rights reserved.
+
+package main
+
+import (
+	crand "crypto/rand"
+	"encoding/binary"
+	"flag"
+	"io/ioutil"
+	"log"
+	"math/rand"
+	"net"
+
+	"github.com/bzEq/bx/proxy/intrinsic"
+	"github.com/bzEq/bx/proxy/socks5"
+	"github.com/bzEq/bx/relayer"
+)
+
+var options relayer.Options
+
+func startRelayer() {
+	pipeline := &relayer.Pipeline{}
+	ln, err := net.Listen("tcp", options.LocalAddr)
+	if err != nil {
+		log.Println(err)
+		return
+	}
+	defer ln.Close()
+	var fe relayer.Frontend
+	var be relayer.Backend
+	if options.NextHop == "" {
+		fe = intrinsic.NewAcceptor(ln.(*net.TCPListener), pipeline)
+		be = &relayer.TCPBE{}
+	} else {
+		raddr, err := net.ResolveTCPAddr("tcp", options.NextHop)
+		if err != nil {
+			log.Println(err)
+			return
+		}
+		fe = socks5.NewAcceptor(ln.(*net.TCPListener), nil)
+		be = intrinsic.NewDialer(raddr, pipeline)
+	}
+	r := relayer.NewRelayer(fe, be)
+	if err := r.Relay(); err != nil {
+		log.Println(err)
+		return
+	}
+}
+
+func main() {
+	var seed int64
+	binary.Read(crand.Reader, binary.BigEndian, &seed)
+	rand.Seed(seed)
+	var debug bool
+	flag.BoolVar(&debug, "debug", false, "Enable debug logging")
+	flag.StringVar(&options.LocalAddr, "l", "localhost:1080", "Listen address of this relayer")
+	flag.StringVar(&options.NextHop, "n", "", "Address of next-hop relayer")
+	flag.Parse()
+	if !debug {
+		log.SetOutput(ioutil.Discard)
+	}
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+	startRelayer()
+}

relayer/pipeline.go

@@ -5,6 +5,7 @@ package relayer
 import (
 	"github.com/bzEq/bx/core"
 	"github.com/bzEq/bx/passes"
+	"net"
 )
 
 func createRandomCodec() (*passes.RandomEncoder, *passes.RandomDecoder) {
@@ -42,3 +43,9 @@ func CreateProtocol(name string) core.Protocol {
 		}
 	}
 }
+
+type Pipeline struct{}
+
+func (self *Pipeline) Create(c net.Conn) core.Port {
+	return core.NewPort(c, CreateProtocol(""))
+}

relayer/raw.go

@@ -0,0 +1,19 @@
+// Copyright (c) 2024 Kai Luo <[email protected]>. All rights reserved.
+
+package relayer
+
+import (
+	"net"
+
+	"github.com/bzEq/bx/core"
+)
+
+type TCPBE struct{}
+
+func (self *TCPBE) Dial(addr net.Addr) (core.Port, error) {
+	c, err := net.Dial(addr.Network(), addr.String())
+	if err != nil {
+		return nil, err
+	}
+	return core.NewPort(c, nil), nil
+}

relayer/relayer.go

@@ -11,14 +11,18 @@ import (
 
 type Options struct {
 	LocalAddr string
-	NextHops  []string
+	NextHop   string
 }
 
 type Relayer struct {
 	fe Frontend
 	be Backend
 }
 
+func NewRelayer(fe Frontend, be Backend) *Relayer {
+	return &Relayer{fe, be}
+}
+
 type Frontend interface {
 	Accept() (core.Port, net.Addr, error)
 }
@@ -31,7 +35,8 @@ func (self *Relayer) Relay() error {
 	for {
 		fp, addr, err := self.fe.Accept()
 		if err != nil {
-			return err
+			log.Println(err)
+			continue
 		}
 		go func(fp core.Port, addr net.Addr) {
 			defer fp.Close()
@@ -44,4 +49,5 @@ func (self *Relayer) Relay() error {
 			core.RunSimpleSwitch(fp, bp)
 		}(fp, addr)
 	}
+	return nil
 }