diff --git a/proxy/network_proxy_test.go b/proxy/network_proxy_test.go deleted file mode 100644 index 9a73548..0000000 --- a/proxy/network_proxy_test.go +++ /dev/null @@ -1,216 +0,0 @@ -package proxy - -import ( - "bytes" - "fmt" - "io" - "net" - "strings" - "testing" - "time" -) - -var testBuf = []byte("Buffalo buffalo Buffalo buffalo buffalo buffalo Buffalo buffalo") -var testBufSize = len(testBuf) - -type EchoServer interface { - Run() - Close() - LocalAddr() net.Addr -} - -type TCPEchoServer struct { - listener net.Listener - testCtx *testing.T -} - -type UDPEchoServer struct { - conn net.PacketConn - testCtx *testing.T -} - -func NewEchoServer(t *testing.T, proto, address string) EchoServer { - var server EchoServer - if strings.HasPrefix(proto, "tcp") { - listener, err := net.Listen(proto, address) - if err != nil { - t.Fatal(err) - } - server = &TCPEchoServer{listener: listener, testCtx: t} - } else { - socket, err := net.ListenPacket(proto, address) - if err != nil { - t.Fatal(err) - } - server = &UDPEchoServer{conn: socket, testCtx: t} - } - return server -} - -func (server *TCPEchoServer) Run() { - go func() { - for { - client, err := server.listener.Accept() - if err != nil { - return - } - go func(client net.Conn) { - if _, err := io.Copy(client, client); err != nil { - server.testCtx.Logf("can't echo to the client: %v\n", err.Error()) - } - client.Close() - }(client) - } - }() -} - -func (server *TCPEchoServer) LocalAddr() net.Addr { return server.listener.Addr() } -func (server *TCPEchoServer) Close() { server.listener.Close() } - -func (server *UDPEchoServer) Run() { - go func() { - readBuf := make([]byte, 1024) - for { - read, from, err := server.conn.ReadFrom(readBuf) - if err != nil { - return - } - for i := 0; i != read; { - written, err := server.conn.WriteTo(readBuf[i:read], from) - if err != nil { - break - } - i += written - } - } - }() -} - -func (server *UDPEchoServer) LocalAddr() net.Addr { return server.conn.LocalAddr() } -func (server *UDPEchoServer) Close() { server.conn.Close() } - -func testProxyAt(t *testing.T, proto string, proxy Proxy, addr string) { - defer proxy.Close() - go proxy.Run() - client, err := net.Dial(proto, addr) - if err != nil { - t.Fatalf("Can't connect to the proxy: %v", err) - } - defer client.Close() - client.SetDeadline(time.Now().Add(10 * time.Second)) - if _, err = client.Write(testBuf); err != nil { - t.Fatal(err) - } - recvBuf := make([]byte, testBufSize) - if _, err = client.Read(recvBuf); err != nil { - t.Fatal(err) - } - if !bytes.Equal(testBuf, recvBuf) { - t.Fatal(fmt.Errorf("Expected [%v] but got [%v]", testBuf, recvBuf)) - } -} - -func testProxy(t *testing.T, proto string, proxy Proxy) { - testProxyAt(t, proto, proxy, proxy.FrontendAddr().String()) -} - -func TestTCP4Proxy(t *testing.T) { - backend := NewEchoServer(t, "tcp", "127.0.0.1:0") - defer backend.Close() - backend.Run() - frontendAddr := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0} - proxy, err := NewProxy(frontendAddr, backend.LocalAddr()) - if err != nil { - t.Fatal(err) - } - testProxy(t, "tcp", proxy) -} - -func TestTCP6Proxy(t *testing.T) { - backend := NewEchoServer(t, "tcp", "[::1]:0") - defer backend.Close() - backend.Run() - frontendAddr := &net.TCPAddr{IP: net.IPv6loopback, Port: 0} - proxy, err := NewProxy(frontendAddr, backend.LocalAddr()) - if err != nil { - t.Fatal(err) - } - testProxy(t, "tcp", proxy) -} - -func TestTCPDualStackProxy(t *testing.T) { - // If I understand `godoc -src net favoriteAddrFamily` (used by the - // net.Listen* functions) correctly this should work, but it doesn't. - t.Skip("No support for dual stack yet") - backend := NewEchoServer(t, "tcp", "[::1]:0") - defer backend.Close() - backend.Run() - frontendAddr := &net.TCPAddr{IP: net.IPv6loopback, Port: 0} - proxy, err := NewProxy(frontendAddr, backend.LocalAddr()) - if err != nil { - t.Fatal(err) - } - ipv4ProxyAddr := &net.TCPAddr{ - IP: net.IPv4(127, 0, 0, 1), - Port: proxy.FrontendAddr().(*net.TCPAddr).Port, - } - testProxyAt(t, "tcp", proxy, ipv4ProxyAddr.String()) -} - -func TestUDP4Proxy(t *testing.T) { - backend := NewEchoServer(t, "udp", "127.0.0.1:0") - defer backend.Close() - backend.Run() - frontendAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0} - proxy, err := NewProxy(frontendAddr, backend.LocalAddr()) - if err != nil { - t.Fatal(err) - } - testProxy(t, "udp", proxy) -} - -func TestUDP6Proxy(t *testing.T) { - backend := NewEchoServer(t, "udp", "[::1]:0") - defer backend.Close() - backend.Run() - frontendAddr := &net.UDPAddr{IP: net.IPv6loopback, Port: 0} - proxy, err := NewProxy(frontendAddr, backend.LocalAddr()) - if err != nil { - t.Fatal(err) - } - testProxy(t, "udp", proxy) -} - -func TestUDPWriteError(t *testing.T) { - frontendAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0} - // Hopefully, this port will be free: */ - backendAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 25587} - proxy, err := NewProxy(frontendAddr, backendAddr) - if err != nil { - t.Fatal(err) - } - defer proxy.Close() - go proxy.Run() - client, err := net.Dial("udp", "127.0.0.1:25587") - if err != nil { - t.Fatalf("Can't connect to the proxy: %v", err) - } - defer client.Close() - // Make sure the proxy doesn't stop when there is no actual backend: - client.Write(testBuf) - client.Write(testBuf) - backend := NewEchoServer(t, "udp", "127.0.0.1:25587") - defer backend.Close() - backend.Run() - client.SetDeadline(time.Now().Add(10 * time.Second)) - if _, err = client.Write(testBuf); err != nil { - t.Fatal(err) - } - recvBuf := make([]byte, testBufSize) - if _, err = client.Read(recvBuf); err != nil { - t.Fatal(err) - } - if !bytes.Equal(testBuf, recvBuf) { - t.Fatal(fmt.Errorf("Expected [%v] but got [%v]", testBuf, recvBuf)) - } -} diff --git a/proxy/proxy.go b/proxy/proxy.go deleted file mode 100644 index 4e24e5f..0000000 --- a/proxy/proxy.go +++ /dev/null @@ -1,37 +0,0 @@ -// Package proxy provides a network Proxy interface and implementations for TCP -// and UDP. -package proxy - -import ( - "fmt" - "net" -) - -// Proxy defines the behavior of a proxy. It forwards traffic back and forth -// between two endpoints : the frontend and the backend. -// It can be used to do software port-mapping between two addresses. -// e.g. forward all traffic between the frontend (host) 127.0.0.1:3000 -// to the backend (container) at 172.17.42.108:4000. -type Proxy interface { - // Run starts forwarding traffic back and forth between the front - // and back-end addresses. - Run() - // Close stops forwarding traffic and close both ends of the Proxy. - Close() - // FrontendAddr returns the address on which the proxy is listening. - FrontendAddr() net.Addr - // BackendAddr returns the proxied address. - BackendAddr() net.Addr -} - -// NewProxy creates a Proxy according to the specified frontendAddr and backendAddr. -func NewProxy(frontendAddr, backendAddr net.Addr) (Proxy, error) { - switch frontendAddr.(type) { - case *net.UDPAddr: - return NewUDPProxy(frontendAddr.(*net.UDPAddr), backendAddr.(*net.UDPAddr)) - case *net.TCPAddr: - return NewTCPProxy(frontendAddr.(*net.TCPAddr), backendAddr.(*net.TCPAddr)) - default: - panic(fmt.Errorf("Unsupported protocol")) - } -} diff --git a/proxy/stub_proxy.go b/proxy/stub_proxy.go deleted file mode 100644 index 571749e..0000000 --- a/proxy/stub_proxy.go +++ /dev/null @@ -1,31 +0,0 @@ -package proxy - -import ( - "net" -) - -// StubProxy is a proxy that is a stub (does nothing). -type StubProxy struct { - frontendAddr net.Addr - backendAddr net.Addr -} - -// Run does nothing. -func (p *StubProxy) Run() {} - -// Close does nothing. -func (p *StubProxy) Close() {} - -// FrontendAddr returns the frontend address. -func (p *StubProxy) FrontendAddr() net.Addr { return p.frontendAddr } - -// BackendAddr returns the backend address. -func (p *StubProxy) BackendAddr() net.Addr { return p.backendAddr } - -// NewStubProxy creates a new StubProxy -func NewStubProxy(frontendAddr, backendAddr net.Addr) (Proxy, error) { - return &StubProxy{ - frontendAddr: frontendAddr, - backendAddr: backendAddr, - }, nil -} diff --git a/proxy/tcp_proxy.go b/proxy/tcp_proxy.go deleted file mode 100644 index 8f42580..0000000 --- a/proxy/tcp_proxy.go +++ /dev/null @@ -1,96 +0,0 @@ -package proxy - -import ( - "io" - "net" - "sync" - "syscall" - - "github.com/Sirupsen/logrus" -) - -// TCPProxy is a proxy for TCP connections. It implements the Proxy interface to -// handle TCP traffic forwarding between the frontend and backend addresses. -type TCPProxy struct { - listener *net.TCPListener - frontendAddr *net.TCPAddr - backendAddr *net.TCPAddr -} - -// NewTCPProxy creates a new TCPProxy. -func NewTCPProxy(frontendAddr, backendAddr *net.TCPAddr) (*TCPProxy, error) { - listener, err := net.ListenTCP("tcp", frontendAddr) - if err != nil { - return nil, err - } - // If the port in frontendAddr was 0 then ListenTCP will have a picked - // a port to listen on, hence the call to Addr to get that actual port: - return &TCPProxy{ - listener: listener, - frontendAddr: listener.Addr().(*net.TCPAddr), - backendAddr: backendAddr, - }, nil -} - -func (proxy *TCPProxy) clientLoop(client *net.TCPConn, quit chan bool) { - backend, err := net.DialTCP("tcp", nil, proxy.backendAddr) - if err != nil { - logrus.Printf("Can't forward traffic to backend tcp/%v: %s\n", proxy.backendAddr, err) - client.Close() - return - } - - var wg sync.WaitGroup - var broker = func(to, from *net.TCPConn) { - if _, err := io.Copy(to, from); err != nil { - // If the socket we are writing to is shutdown with - // SHUT_WR, forward it to the other end of the pipe: - if err, ok := err.(*net.OpError); ok && err.Err == syscall.EPIPE { - from.CloseWrite() - } - } - to.CloseRead() - wg.Done() - } - - wg.Add(2) - go broker(client, backend) - go broker(backend, client) - - finish := make(chan struct{}) - go func() { - wg.Wait() - close(finish) - }() - - select { - case <-quit: - case <-finish: - } - client.Close() - backend.Close() - <-finish -} - -// Run starts forwarding the traffic using TCP. -func (proxy *TCPProxy) Run() { - quit := make(chan bool) - defer close(quit) - for { - client, err := proxy.listener.Accept() - if err != nil { - logrus.Printf("Stopping proxy on tcp/%v for tcp/%v (%s)", proxy.frontendAddr, proxy.backendAddr, err) - return - } - go proxy.clientLoop(client.(*net.TCPConn), quit) - } -} - -// Close stops forwarding the traffic. -func (proxy *TCPProxy) Close() { proxy.listener.Close() } - -// FrontendAddr returns the TCP address on which the proxy is listening. -func (proxy *TCPProxy) FrontendAddr() net.Addr { return proxy.frontendAddr } - -// BackendAddr returns the TCP proxied address. -func (proxy *TCPProxy) BackendAddr() net.Addr { return proxy.backendAddr } diff --git a/proxy/udp_proxy.go b/proxy/udp_proxy.go deleted file mode 100644 index b8375c3..0000000 --- a/proxy/udp_proxy.go +++ /dev/null @@ -1,169 +0,0 @@ -package proxy - -import ( - "encoding/binary" - "net" - "strings" - "sync" - "syscall" - "time" - - "github.com/Sirupsen/logrus" -) - -const ( - // UDPConnTrackTimeout is the timeout used for UDP connection tracking - UDPConnTrackTimeout = 90 * time.Second - // UDPBufSize is the buffer size for the UDP proxy - UDPBufSize = 65507 -) - -// A net.Addr where the IP is split into two fields so you can use it as a key -// in a map: -type connTrackKey struct { - IPHigh uint64 - IPLow uint64 - Port int -} - -func newConnTrackKey(addr *net.UDPAddr) *connTrackKey { - if len(addr.IP) == net.IPv4len { - return &connTrackKey{ - IPHigh: 0, - IPLow: uint64(binary.BigEndian.Uint32(addr.IP)), - Port: addr.Port, - } - } - return &connTrackKey{ - IPHigh: binary.BigEndian.Uint64(addr.IP[:8]), - IPLow: binary.BigEndian.Uint64(addr.IP[8:]), - Port: addr.Port, - } -} - -type connTrackMap map[connTrackKey]*net.UDPConn - -// UDPProxy is proxy for which handles UDP datagrams. It implements the Proxy -// interface to handle UDP traffic forwarding between the frontend and backend -// addresses. -type UDPProxy struct { - listener *net.UDPConn - frontendAddr *net.UDPAddr - backendAddr *net.UDPAddr - connTrackTable connTrackMap - connTrackLock sync.Mutex -} - -// NewUDPProxy creates a new UDPProxy. -func NewUDPProxy(frontendAddr, backendAddr *net.UDPAddr) (*UDPProxy, error) { - listener, err := net.ListenUDP("udp", frontendAddr) - if err != nil { - return nil, err - } - return &UDPProxy{ - listener: listener, - frontendAddr: listener.LocalAddr().(*net.UDPAddr), - backendAddr: backendAddr, - connTrackTable: make(connTrackMap), - }, nil -} - -func (proxy *UDPProxy) replyLoop(proxyConn *net.UDPConn, clientAddr *net.UDPAddr, clientKey *connTrackKey) { - defer func() { - proxy.connTrackLock.Lock() - delete(proxy.connTrackTable, *clientKey) - proxy.connTrackLock.Unlock() - proxyConn.Close() - }() - - readBuf := make([]byte, UDPBufSize) - for { - proxyConn.SetReadDeadline(time.Now().Add(UDPConnTrackTimeout)) - again: - read, err := proxyConn.Read(readBuf) - if err != nil { - if err, ok := err.(*net.OpError); ok && err.Err == syscall.ECONNREFUSED { - // This will happen if the last write failed - // (e.g: nothing is actually listening on the - // proxied port on the container), ignore it - // and continue until UDPConnTrackTimeout - // expires: - goto again - } - return - } - for i := 0; i != read; { - written, err := proxy.listener.WriteToUDP(readBuf[i:read], clientAddr) - if err != nil { - return - } - i += written - } - } -} - -// Run starts forwarding the traffic using UDP. -func (proxy *UDPProxy) Run() { - readBuf := make([]byte, UDPBufSize) - for { - read, from, err := proxy.listener.ReadFromUDP(readBuf) - if err != nil { - // NOTE: Apparently ReadFrom doesn't return - // ECONNREFUSED like Read do (see comment in - // UDPProxy.replyLoop) - if !isClosedError(err) { - logrus.Printf("Stopping proxy on udp/%v for udp/%v (%s)", proxy.frontendAddr, proxy.backendAddr, err) - } - break - } - - fromKey := newConnTrackKey(from) - proxy.connTrackLock.Lock() - proxyConn, hit := proxy.connTrackTable[*fromKey] - if !hit { - proxyConn, err = net.DialUDP("udp", nil, proxy.backendAddr) - if err != nil { - logrus.Printf("Can't proxy a datagram to udp/%s: %s\n", proxy.backendAddr, err) - proxy.connTrackLock.Unlock() - continue - } - proxy.connTrackTable[*fromKey] = proxyConn - go proxy.replyLoop(proxyConn, from, fromKey) - } - proxy.connTrackLock.Unlock() - for i := 0; i != read; { - written, err := proxyConn.Write(readBuf[i:read]) - if err != nil { - logrus.Printf("Can't proxy a datagram to udp/%s: %s\n", proxy.backendAddr, err) - break - } - i += written - } - } -} - -// Close stops forwarding the traffic. -func (proxy *UDPProxy) Close() { - proxy.listener.Close() - proxy.connTrackLock.Lock() - defer proxy.connTrackLock.Unlock() - for _, conn := range proxy.connTrackTable { - conn.Close() - } -} - -// FrontendAddr returns the UDP address on which the proxy is listening. -func (proxy *UDPProxy) FrontendAddr() net.Addr { return proxy.frontendAddr } - -// BackendAddr returns the proxied UDP address. -func (proxy *UDPProxy) BackendAddr() net.Addr { return proxy.backendAddr } - -func isClosedError(err error) bool { - /* This comparison is ugly, but unfortunately, net.go doesn't export errClosing. - * See: - * http://golang.org/src/pkg/net/net.go - * https://code.google.com/p/go/issues/detail?id=4337 - * https://groups.google.com/forum/#!msg/golang-nuts/0_aaCvBmOcM/SptmDyX1XJMJ - */ - return strings.HasSuffix(err.Error(), "use of closed network connection") -}