TUN-5481: Create abstraction for Origin UDP Connection
Creates an abstraction over UDP Conn for origin "connection" which can be useful for future support of complex protocols that may require changing ports during protocol negotiation (eg. SIP, TFTP) In addition, it removes a dependency from ingress on connection package.
This commit is contained in:
parent
eea3d11e40
commit
7e47667b08
|
@ -17,6 +17,7 @@ import (
|
||||||
"golang.org/x/sync/errgroup"
|
"golang.org/x/sync/errgroup"
|
||||||
|
|
||||||
"github.com/cloudflare/cloudflared/datagramsession"
|
"github.com/cloudflare/cloudflared/datagramsession"
|
||||||
|
"github.com/cloudflare/cloudflared/ingress"
|
||||||
quicpogs "github.com/cloudflare/cloudflared/quic"
|
quicpogs "github.com/cloudflare/cloudflared/quic"
|
||||||
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
|
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
|
||||||
)
|
)
|
||||||
|
@ -176,7 +177,7 @@ func (q *QUICConnection) handleRPCStream(rpcStream *quicpogs.RPCServerStream) er
|
||||||
func (q *QUICConnection) RegisterUdpSession(ctx context.Context, sessionID uuid.UUID, dstIP net.IP, dstPort uint16) error {
|
func (q *QUICConnection) RegisterUdpSession(ctx context.Context, sessionID uuid.UUID, dstIP net.IP, dstPort uint16) error {
|
||||||
// Each session is a series of datagram from an eyeball to a dstIP:dstPort.
|
// Each session is a series of datagram from an eyeball to a dstIP:dstPort.
|
||||||
// (src port, dst IP, dst port) uniquely identifies a session, so it needs a dedicated connected socket.
|
// (src port, dst IP, dst port) uniquely identifies a session, so it needs a dedicated connected socket.
|
||||||
originProxy, err := q.newUDPProxy(dstIP, dstPort)
|
originProxy, err := ingress.DialUDP(dstIP, dstPort)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
q.logger.Err(err).Msgf("Failed to create udp proxy to %s:%d", dstIP, dstPort)
|
q.logger.Err(err).Msgf("Failed to create udp proxy to %s:%d", dstIP, dstPort)
|
||||||
return err
|
return err
|
||||||
|
@ -292,15 +293,6 @@ func isTransferEncodingChunked(req *http.Request) bool {
|
||||||
return strings.Contains(strings.ToLower(transferEncodingVal), "chunked")
|
return strings.Contains(strings.ToLower(transferEncodingVal), "chunked")
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: TUN-5303: Define an UDPProxy in ingress package
|
|
||||||
func (q *QUICConnection) newUDPProxy(dstIP net.IP, dstPort uint16) (*net.UDPConn, error) {
|
|
||||||
dstAddr := &net.UDPAddr{
|
|
||||||
IP: dstIP,
|
|
||||||
Port: int(dstPort),
|
|
||||||
}
|
|
||||||
return net.DialUDP("udp", nil, dstAddr)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO: TUN-5303: Find the local IP once in ingress package
|
// TODO: TUN-5303: Find the local IP once in ingress package
|
||||||
// TODO: TUN-5421 allow user to specify which IP to bind to
|
// TODO: TUN-5421 allow user to specify which IP to bind to
|
||||||
func getLocalIP() (net.IP, error) {
|
func getLocalIP() (net.IP, error) {
|
||||||
|
|
|
@ -4,6 +4,7 @@ import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
@ -19,7 +20,6 @@ import (
|
||||||
"golang.org/x/net/proxy"
|
"golang.org/x/net/proxy"
|
||||||
"golang.org/x/sync/errgroup"
|
"golang.org/x/sync/errgroup"
|
||||||
|
|
||||||
"github.com/cloudflare/cloudflared/connection"
|
|
||||||
"github.com/cloudflare/cloudflared/logger"
|
"github.com/cloudflare/cloudflared/logger"
|
||||||
"github.com/cloudflare/cloudflared/socks"
|
"github.com/cloudflare/cloudflared/socks"
|
||||||
"github.com/cloudflare/cloudflared/websocket"
|
"github.com/cloudflare/cloudflared/websocket"
|
||||||
|
@ -192,8 +192,10 @@ func TestSocksStreamWSOverTCPConnection(t *testing.T) {
|
||||||
|
|
||||||
func TestWsConnReturnsBeforeStreamReturns(t *testing.T) {
|
func TestWsConnReturnsBeforeStreamReturns(t *testing.T) {
|
||||||
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
eyeballConn, err := connection.NewHTTP2RespWriter(r, w, connection.TypeWebsocket)
|
eyeballConn := &readWriter{
|
||||||
assert.NoError(t, err)
|
w: w,
|
||||||
|
r: r.Body,
|
||||||
|
}
|
||||||
|
|
||||||
cfdConn, originConn := net.Pipe()
|
cfdConn, originConn := net.Pipe()
|
||||||
tcpOverWSConn := tcpOverWSConnection{
|
tcpOverWSConn := tcpOverWSConnection{
|
||||||
|
@ -319,3 +321,16 @@ func echoTCPOrigin(t *testing.T, conn net.Conn) {
|
||||||
_, err = conn.Write(testResponse)
|
_, err = conn.Write(testResponse)
|
||||||
assert.NoError(t, err)
|
assert.NoError(t, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type readWriter struct {
|
||||||
|
w io.Writer
|
||||||
|
r io.Reader
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *readWriter) Read(p []byte) (n int, err error) {
|
||||||
|
return r.r.Read(p)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *readWriter) Write(p []byte) (n int, err error) {
|
||||||
|
return r.w.Write(p)
|
||||||
|
}
|
||||||
|
|
|
@ -0,0 +1,27 @@
|
||||||
|
package ingress
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"net"
|
||||||
|
)
|
||||||
|
|
||||||
|
type UDPProxy struct {
|
||||||
|
io.ReadWriteCloser
|
||||||
|
}
|
||||||
|
|
||||||
|
func DialUDP(dstIP net.IP, dstPort uint16) (*UDPProxy, error) {
|
||||||
|
dstAddr := &net.UDPAddr{
|
||||||
|
IP: dstIP,
|
||||||
|
Port: int(dstPort),
|
||||||
|
}
|
||||||
|
|
||||||
|
// We use nil as local addr to force runtime to find the best suitable local address IP given the destination
|
||||||
|
// address as context.
|
||||||
|
udpConn, err := net.DialUDP("udp", nil, dstAddr)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("unable to create UDP proxy to origin (%v:%v): %w", dstIP, dstPort, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &UDPProxy{udpConn}, nil
|
||||||
|
}
|
Loading…
Reference in New Issue