This commit is contained in:
parent
d5de31eda7
commit
2e73689762
@ -41,5 +41,5 @@ func NewTunnel(cfg *config.ClientConfig) (*tunnel.Tunnel, error) {
|
||||
return nil, fmt.Errorf("failed to connect: %v", err)
|
||||
}
|
||||
|
||||
return tunnel.NewClientTunnel(cfg.TunnelName, cfg.TunnelTarget, serverConn), nil
|
||||
return tunnel.NewClientTunnel(cfg.TunnelName, cfg.TunnelTarget, serverConn)
|
||||
}
|
||||
|
@ -22,9 +22,6 @@ var tunnelCmd = &cobra.Command{
|
||||
if err != nil {
|
||||
log.Fatal("failed to create tunnel:", err)
|
||||
}
|
||||
|
||||
// Start Tunnel
|
||||
log.Infof("creating TCP tunnel: %s -> %s", cfg.TunnelName, cfg.TunnelTarget)
|
||||
tunnel.Start()
|
||||
},
|
||||
}
|
||||
|
104
tunnel/http.go
Normal file
104
tunnel/http.go
Normal file
@ -0,0 +1,104 @@
|
||||
package tunnel
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httputil"
|
||||
"net/url"
|
||||
"sync"
|
||||
)
|
||||
|
||||
func HTTPConnectionBuilder(targetURL *url.URL) (ConnBuilder, error) {
|
||||
multiConnListener := newMultiConnListener()
|
||||
|
||||
// Create Reverse Proxy
|
||||
proxy := &httputil.ReverseProxy{
|
||||
Director: func(req *http.Request) {
|
||||
req.Host = targetURL.Host
|
||||
req.URL.Host = targetURL.Host
|
||||
req.URL.Scheme = targetURL.Scheme
|
||||
},
|
||||
ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
|
||||
http.Error(w, fmt.Sprintf("Proxy error: %v", err), http.StatusBadGateway)
|
||||
},
|
||||
}
|
||||
|
||||
// Start HTTP Proxy
|
||||
go func() {
|
||||
defer multiConnListener.Close()
|
||||
_ = http.Serve(multiConnListener, proxy)
|
||||
}()
|
||||
|
||||
// Return Connection Builder
|
||||
return func() (conn io.ReadWriteCloser, err error) {
|
||||
clientConn, serverConn := net.Pipe()
|
||||
|
||||
if err := multiConnListener.addConn(serverConn); err != nil {
|
||||
_ = clientConn.Close()
|
||||
_ = serverConn.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return clientConn, nil
|
||||
}, nil
|
||||
}
|
||||
|
||||
type multiConnListener struct {
|
||||
connCh chan net.Conn
|
||||
closed chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func newMultiConnListener() *multiConnListener {
|
||||
return &multiConnListener{
|
||||
connCh: make(chan net.Conn, 100),
|
||||
closed: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (l *multiConnListener) Accept() (net.Conn, error) {
|
||||
select {
|
||||
case conn := <-l.connCh:
|
||||
if conn == nil {
|
||||
return nil, fmt.Errorf("listener closed")
|
||||
}
|
||||
return conn, nil
|
||||
case <-l.closed:
|
||||
return nil, fmt.Errorf("listener closed")
|
||||
}
|
||||
}
|
||||
|
||||
func (l *multiConnListener) Close() error {
|
||||
l.once.Do(func() {
|
||||
close(l.closed)
|
||||
// Drain any remaining connections
|
||||
go func() {
|
||||
for conn := range l.connCh {
|
||||
if conn != nil {
|
||||
conn.Close()
|
||||
}
|
||||
}
|
||||
}()
|
||||
close(l.connCh)
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *multiConnListener) Addr() net.Addr {
|
||||
return &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
|
||||
}
|
||||
|
||||
func (l *multiConnListener) addConn(conn net.Conn) error {
|
||||
select {
|
||||
case l.connCh <- conn:
|
||||
return nil
|
||||
case <-l.closed:
|
||||
conn.Close()
|
||||
return fmt.Errorf("listener is closed")
|
||||
default:
|
||||
conn.Close()
|
||||
return fmt.Errorf("connection queue full")
|
||||
}
|
||||
}
|
@ -4,6 +4,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/url"
|
||||
"sync"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
@ -11,6 +12,8 @@ import (
|
||||
"reichard.io/conduit/types"
|
||||
)
|
||||
|
||||
type ConnBuilder func() (conn io.ReadWriteCloser, err error)
|
||||
|
||||
func NewServerTunnel(name string, wsConn *websocket.Conn) *Tunnel {
|
||||
return &Tunnel{
|
||||
name: name,
|
||||
@ -19,22 +22,40 @@ func NewServerTunnel(name string, wsConn *websocket.Conn) *Tunnel {
|
||||
}
|
||||
}
|
||||
|
||||
func NewClientTunnel(name, target string, wsConn *websocket.Conn) *Tunnel {
|
||||
func NewClientTunnel(name, target string, wsConn *websocket.Conn) (*Tunnel, error) {
|
||||
targetURL, err := url.Parse(target)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var connBuilder ConnBuilder
|
||||
switch targetURL.Scheme {
|
||||
case "http", "https":
|
||||
log.Infof("creating HTTP tunnel: %s -> %s", name, target)
|
||||
connBuilder, err = HTTPConnectionBuilder(targetURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
default:
|
||||
log.Infof("creating TCP tunnel: %s -> %s", name, target)
|
||||
connBuilder = func() (conn io.ReadWriteCloser, err error) {
|
||||
return net.Dial("tcp", target)
|
||||
}
|
||||
}
|
||||
|
||||
return &Tunnel{
|
||||
name: name,
|
||||
wsConn: wsConn,
|
||||
streams: make(map[string]io.ReadWriteCloser),
|
||||
connBuilder: func() (io.ReadWriteCloser, error) {
|
||||
return net.Dial("tcp", target)
|
||||
},
|
||||
}
|
||||
connBuilder: connBuilder,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type Tunnel struct {
|
||||
name string
|
||||
wsConn *websocket.Conn
|
||||
streams map[string]io.ReadWriteCloser
|
||||
connBuilder func() (io.ReadWriteCloser, error)
|
||||
connBuilder ConnBuilder
|
||||
|
||||
wsMu, streamsMu sync.Mutex
|
||||
}
|
||||
|
Loading…
x
Reference in New Issue
Block a user