Demo: Feat/proxy connection settings (#389)

* Update spec

* wip proxy demo

* Fix tests

* Fix and add tests

* Add test case

* Add proxy url to example ui

* Use custom dialer for websocket http/https proxies

* clear previous settings

* Add new import to internal/examples

* Only use ProxyConnectDialer when connect headers are specified

* Review feedback

* Add WS client test

* add invalid url tests

* Revert change to dialer instantiation

* Remove testing instructions

* Update opamp-spec to v0.14.0, fix tests

Author: michel-laterman

client/httpclient.go

@@ -55,6 +55,12 @@ func (c *httpClient) Start(ctx context.Context, settings types.StartSettings) er
 		c.sender.EnableCompression()
 	}
 
+	if settings.ProxyURL != "" {
+		if err := c.sender.SetProxy(settings.ProxyURL, settings.ProxyHeaders); err != nil {
+			return err
+		}
+	}
+
 	// Prepare the first message to send.
 	err := c.common.PrepareFirstMessage(ctx)
 	if err != nil {

client/internal/httpsender.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"net/url"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -84,6 +85,35 @@ func NewHTTPSender(logger types.Logger) *HTTPSender {
 	return h
 }
 
+// SetProxy will force each request to use passed proxy and use the passed headers when making a CONNECT request to the proxy.
+// If the proxy has no schema http is used.
+// This method is not thread safe and must be called before h.client is used.
+func (h *HTTPSender) SetProxy(proxy string, headers http.Header) error {
+	proxyURL, err := url.Parse(proxy)
+	if err != nil || proxyURL.Scheme == "" || proxyURL.Host == "" { // error or bad URL - try to use http as scheme to resolve
+		proxyURL, err = url.Parse("http://" + proxy)
+		if err != nil {
+			return err
+		}
+	}
+	if proxyURL.Hostname() == "" {
+		return url.InvalidHostError(proxy)
+	}
+
+	proxyTransport := &http.Transport{}
+	if h.client.Transport != nil {
+		transport, ok := h.client.Transport.(*http.Transport)
+		if !ok {
+			return fmt.Errorf("unable to coorce client transport as *http.Transport detected type is: %T", h.client.Transport)
+		}
+		proxyTransport = transport.Clone()
+	}
+	proxyTransport.Proxy = http.ProxyURL(proxyURL)
+	proxyTransport.ProxyConnectHeader = headers
+	h.client.Transport = proxyTransport
+	return nil
+}
+
 // Run starts the processing loop that will perform the HTTP request/response.
 // When there are no more messages to send Run will suspend until either there is
 // a new message to send or the polling interval elapses.

client/internal/httpsender_test.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -356,3 +357,147 @@ func TestPackageUpdatesWithError(t *testing.T) {
 
 	cancel()
 }
+
+func TestHTTPSenderSetProxy(t *testing.T) {
+	tests := []struct {
+		name string
+		url  string
+		err  error
+	}{{
+		name: "http proxy",
+		url:  "http://proxy.internal:8080",
+		err:  nil,
+	}, {
+		name: "socks5 proxy",
+		url:  "socks5://proxy.internal:8080",
+		err:  nil,
+	}, {
+		name: "no schema",
+		url:  "proxy.internal:8080",
+		err:  nil,
+	}, {
+		name: "empty url",
+		url:  "",
+		err:  url.InvalidHostError(""),
+	}, {
+		name: "invalid url",
+		url:  "this is not valid",
+		err:  url.InvalidHostError("this is not valid"),
+	}}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			sender := NewHTTPSender(&sharedinternal.NopLogger{})
+			err := sender.SetProxy(tc.url, nil)
+			if tc.err != nil {
+				assert.ErrorAs(t, err, &tc.err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+
+	t.Run("old transport settings are preserved", func(t *testing.T) {
+		sender := &HTTPSender{
+			client: &http.Client{
+				Transport: &http.Transport{
+					MaxResponseHeaderBytes: 1024,
+				},
+			},
+		}
+		err := sender.SetProxy("https://proxy.internal:8080", nil)
+		assert.NoError(t, err)
+		transport, ok := sender.client.Transport.(*http.Transport)
+		if !ok {
+			t.Logf("Transport: %v", sender.client.Transport)
+			t.Fatalf("Unable to coorce as *http.Transport detected type: %T", sender.client.Transport)
+		}
+		assert.NotNil(t, transport.Proxy)
+		assert.Equal(t, int64(1024), transport.MaxResponseHeaderBytes)
+	})
+
+	t.Run("test https proxy", func(t *testing.T) {
+		var connected atomic.Bool
+		// HTTPS Connect proxy, no auth required
+		proxyServer := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+			t.Logf("Request: %+v", req)
+			if req.Method != http.MethodConnect {
+				w.WriteHeader(http.StatusMethodNotAllowed)
+				return
+			}
+			connected.Store(true)
+
+			targetConn, err := net.DialTimeout("tcp", req.Host, 10*time.Second)
+			if err != nil {
+				w.WriteHeader(http.StatusBadGateway)
+				return
+			}
+			defer targetConn.Close()
+
+			hijacker, ok := w.(http.Hijacker)
+			if !ok {
+				w.WriteHeader(http.StatusBadGateway)
+				return
+			}
+			clientConn, _, err := hijacker.Hijack()
+			if err != nil {
+				t.Logf("Hijack error: %v", err)
+				w.WriteHeader(http.StatusBadGateway)
+				return
+			}
+			clientConn.Write([]byte("HTTP/1.1 200 Connection established\r\n\r\n"))
+			defer clientConn.Close()
+
+			var wg sync.WaitGroup
+			wg.Add(2)
+			go func() {
+				defer wg.Done()
+				_, err := io.Copy(targetConn, clientConn)
+				assert.NoError(t, err, "proxy encountered an error copying to destination")
+			}()
+			go func() {
+				defer wg.Done()
+				_, err := io.Copy(clientConn, targetConn)
+				assert.NoError(t, err, "proxy encountered an error copying to client")
+			}()
+			wg.Wait()
+		}))
+		t.Cleanup(proxyServer.Close)
+
+		srv := StartTLSMockServer(t)
+		t.Cleanup(srv.Close)
+		srv.OnRequest = func(w http.ResponseWriter, _ *http.Request) {
+			w.WriteHeader(http.StatusOK)
+		}
+
+		sender := NewHTTPSender(&sharedinternal.NopLogger{})
+		sender.client = proxyServer.Client()
+		err := sender.SetProxy(proxyServer.URL, http.Header{"test-header": []string{"test-value"}})
+		assert.NoError(t, err)
+
+		t.Logf("Proxy URL: %s", proxyServer.URL)
+
+		sender.NextMessage().Update(func(msg *protobufs.AgentToServer) {
+			msg.AgentDescription = &protobufs.AgentDescription{
+				IdentifyingAttributes: []*protobufs.KeyValue{{
+					Key: "service.name",
+					Value: &protobufs.AnyValue{
+						Value: &protobufs.AnyValue_StringValue{StringValue: "test-service"},
+					},
+				}},
+			}
+		})
+		sender.callbacks = types.Callbacks{
+			OnConnect: func(_ context.Context) {
+			},
+			OnConnectFailed: func(_ context.Context, err error) {
+				t.Logf("sender failed to connect: %v", err)
+			},
+		}
+		sender.url = "https://" + srv.Endpoint
+
+		resp, err := sender.sendRequestWithRetries(context.Background())
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+		assert.True(t, connected.Load(), "test request did not use proxy")
+	})
+}

client/types/startsettings.go

@@ -26,6 +26,12 @@ type StartSettings struct {
 	// Optional TLS config for HTTP connection.
 	TLSConfig *tls.Config
 
+	// Optional Proxy configuration
+	// The ProxyURL may be http(s) or socks5; if no schema is specified http is assumed.
+	ProxyURL string
+	// ProxyHeaders gives the headers an HTTP client will present on a proxy CONNECT request.
+	ProxyHeaders http.Header
+
 	// Agent information.
 	InstanceUid InstanceUid
 

client/wsclient.go

@@ -2,15 +2,19 @@ package client
 
 import (
 	"context"
+	"crypto/tls"
 	"errors"
+	"net"
 	"net/http"
 	"net/url"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/gorilla/websocket"
+	dialer "github.com/michel-laterman/proxy-connect-dialer-go"
 
 	"github.com/open-telemetry/opamp-go/client/internal"
 	"github.com/open-telemetry/opamp-go/client/types"
@@ -81,6 +85,12 @@ func (c *wsClient) Start(ctx context.Context, settings types.StartSettings) erro
 	// Prepare connection settings.
 	c.dialer = *websocket.DefaultDialer
 
+	if settings.ProxyURL != "" {
+		if err := c.useProxy(settings.ProxyURL, settings.ProxyHeaders, settings.TLSConfig); err != nil {
+			return err
+		}
+	}
+
 	var err error
 	c.url, err = url.Parse(settings.OpAMPServerURL)
 	if err != nil {
@@ -426,3 +436,52 @@ func (c *wsClient) runUntilStopped(ctx context.Context) {
 		c.runOneCycle(ctx)
 	}
 }
+
+// useProxy sets the websocket dialer to use the passed proxy URL.
+// If the proxy has no schema http is used.
+// This method is not thread safe and must be called before c.dialer is used.
+func (c *wsClient) useProxy(proxy string, headers http.Header, cfg *tls.Config) error {
+	proxyURL, err := url.Parse(proxy)
+	if err != nil || proxyURL.Scheme == "" || proxyURL.Host == "" { // error or bad URL - try to use http as scheme to resolve
+		proxyURL, err = url.Parse("http://" + proxy)
+		if err != nil {
+			return err
+		}
+	}
+	if proxyURL.Hostname() == "" {
+		return url.InvalidHostError(proxy)
+	}
+
+	// Clear previous settings
+	c.dialer.Proxy = nil
+	c.dialer.NetDialContext = nil
+	c.dialer.NetDialTLSContext = nil
+
+	switch strings.ToLower(proxyURL.Scheme) {
+	case "http":
+		// FIXME: dialer.NetDialContext is currently used as a work around instead of setting dialer.Proxy as gorilla/websockets does not have 1st class support for setting proxy connect headers
+		// Once http://github.com/gorilla/websocket/issues/479 is complete, we should use dialer.Proxy, and dialer.ProxyConnectHeader
+		if len(headers) > 0 {
+			dialer, err := dialer.NewProxyConnectDialer(proxyURL, &net.Dialer{}, dialer.WithProxyConnectHeaders(headers))
+			if err != nil {
+				return err
+			}
+			c.dialer.NetDialContext = dialer.DialContext
+			return nil
+		}
+		c.dialer.Proxy = http.ProxyURL(proxyURL) // No connect headers, use a regular proxy
+	case "https":
+		if len(headers) > 0 {
+			dialer, err := dialer.NewProxyConnectDialer(proxyURL, &net.Dialer{}, dialer.WithTLS(cfg), dialer.WithProxyConnectHeaders(headers))
+			if err != nil {
+				return err
+			}
+			c.dialer.NetDialTLSContext = dialer.DialContext
+			return nil
+		}
+		c.dialer.Proxy = http.ProxyURL(proxyURL) // No connect headers, use a regular proxy
+	default: // catches socks5
+		c.dialer.Proxy = http.ProxyURL(proxyURL)
+	}
+	return nil
+}