[v15] exclude alpn upgrade connections from PROXY line enforcement (#…

…45993)

* exclude alpn upgrade connections from PROXY line enforcement

When clients use multiplex mode behind a TLS-terminating load balancer (LB), Teleport circumvents this limitation by establishing a connection that is upgraded to WebSockets.

On the upgraded WebSocket connection, clients initiate a TLS handshake and send their actual requests.

However, when `proxy_protocol: on` is enabled, the proxy line validation is re-applied to the upgraded connection, causing the request to fail. This occurs because the LB only included the proxy line in the initial WebSocket request, which was consumed. After the upgrade, the connection is routed to the ALPN router, and if it reaches the Kubernetes Proxy, it tries to enforce the presence of the PROXY line.

Since the PROXY line was not present, the request failed.

This PR excludes websocket upgraded connections from PROXY line validation.

* rephrase error message and rename function

* simplify code by unwrapping till net.Conn

integration/proxy/proxy_test.go

@@ -391,43 +391,13 @@ func TestALPNSNIProxyKube(t *testing.T) {
 		teleportCluster := suite.root.Config.Auth.ClusterName.GetClusterName()
 		kubeCluster := "gke_project_europecentral2a_cluster1"
 
-		// Make a mock ALB which points to the Teleport Proxy Service. Then
-		// ALPN local proxies will point to this ALB instead.
-		albProxy := helpers.MustStartMockALBProxy(t, suite.root.Config.Proxy.WebAddr.Addr)
-
-		// Generate a self-signed CA for kube local proxy.
-		localCAKey, localCACert, err := tlsca.GenerateSelfSignedCA(pkix.Name{
-			CommonName: "localhost",
-		}, []string{alpncommon.KubeLocalProxyWildcardDomain(teleportCluster)}, defaults.CATTL)
-		require.NoError(t, err)
-
-		// Create the kube local proxy.
-		lp := mustStartALPNLocalProxyWithConfig(t, alpnproxy.LocalProxyConfig{
-			Listener:                mustCreateKubeLocalProxyListener(t, teleportCluster, localCACert, localCAKey),
-			RemoteProxyAddr:         albProxy.Addr().String(),
-			ALPNConnUpgradeRequired: true,
-			InsecureSkipVerify:      true,
-			SNI:                     localK8SNI,
-			HTTPMiddleware:          mustCreateKubeLocalProxyMiddleware(t, teleportCluster, kubeCluster, k8ClientConfig.CertData, k8ClientConfig.KeyData),
-			Protocols:               []alpncommon.Protocol{alpncommon.ProtocolHTTP},
-		})
-		require.NoError(t, err)
+		k8sClient := createALPNLocalKubeClient(t,
+			suite.root.Config.Proxy.WebAddr,
+			teleportCluster,
+			kubeCluster,
+			k8ClientConfig)
 
-		// Create a proxy-url for kube clients.
-		fp := mustStartKubeForwardProxy(t, lp.GetAddr())
-
-		k8Client, err := kubernetes.NewForConfig(&rest.Config{
-			Host:  "https://" + teleportCluster,
-			Proxy: http.ProxyURL(mustParseURL(t, "http://"+fp.GetAddr())),
-			TLSClientConfig: rest.TLSClientConfig{
-				CAData:     localCACert,
-				CertData:   localCACert, // Client uses same cert as local proxy server.
-				KeyData:    localCAKey,
-				ServerName: alpncommon.KubeLocalProxySNI(teleportCluster, kubeCluster),
-			},
-		})
-		require.NoError(t, err)
-		mustGetKubePod(t, k8Client)
+		mustGetKubePod(t, k8sClient)
 	})
 }
 
@@ -517,6 +487,7 @@ func TestKubePROXYProtocol(t *testing.T) {
 		desc              string
 		proxyListenerMode types.ProxyListenerMode
 		proxyProtocolMode multiplexer.PROXYProtocolMode
+		useALPNUpgrade    bool
 	}{
 		{
 			desc:              "PROXY protocol on, separate Proxy listeners",
@@ -548,6 +519,18 @@ func TestKubePROXYProtocol(t *testing.T) {
 			proxyProtocolMode: multiplexer.PROXYProtocolUnspecified,
 			proxyListenerMode: types.ProxyListenerMode_Multiplex,
 		},
+		{
+			desc:              "PROXY protocol on, multiplexed Proxy listeners with ALPN upgrade",
+			proxyProtocolMode: multiplexer.PROXYProtocolOn,
+			proxyListenerMode: types.ProxyListenerMode_Multiplex,
+			useALPNUpgrade:    true,
+		},
+		{
+			desc:              "PROXY protocol off, multiplexed Proxy listeners with ALPN upgrade",
+			proxyProtocolMode: multiplexer.PROXYProtocolOff,
+			proxyListenerMode: types.ProxyListenerMode_Multiplex,
+			useALPNUpgrade:    true,
+		},
 	}
 
 	username := helpers.MustGetCurrentUser(t).Username
@@ -639,7 +622,7 @@ func TestKubePROXYProtocol(t *testing.T) {
 				}
 
 				// Create kube client that we'll use to test that connection is working correctly.
-				k8Client, _, err := kube.ProxyClient(kube.ProxyConfig{
+				k8Client, kubeConfig, err := kube.ProxyClient(kube.ProxyConfig{
 					T:                   testCluster,
 					Username:            kubeRoleSpec.Allow.Logins[0],
 					KubeUsers:           kubeRoleSpec.Allow.KubeGroups,
@@ -651,17 +634,69 @@ func TestKubePROXYProtocol(t *testing.T) {
 				})
 				require.NoError(t, err)
 
+				if tt.useALPNUpgrade {
+					k8Client = createALPNLocalKubeClient(t,
+						targetAddr,
+						testCluster.Secrets.SiteName,
+						kubeCluster,
+						kubeConfig)
+				}
+
 				resp, err := k8Client.CoreV1().Pods("default").List(context.Background(), metav1.ListOptions{})
 				require.NoError(t, err)
 				require.Len(t, resp.Items, 3, "pods item length mismatch")
 			}
 
-			checkForTargetAddr(testCluster.Config.Proxy.Kube.ListenAddr)
+			// kube listener does not support ALPN upgrade
+			if !tt.useALPNUpgrade {
+				checkForTargetAddr(testCluster.Config.Proxy.Kube.ListenAddr)
+			}
 			if tt.proxyListenerMode == types.ProxyListenerMode_Multiplex {
 				checkForTargetAddr(testCluster.Config.Proxy.WebAddr)
 			}
 		})
 	}
+
+}
+
+func createALPNLocalKubeClient(t *testing.T, targetAddr utils.NetAddr, teleportCluster, kubeCluster string, k8ClientConfig *rest.Config) *kubernetes.Clientset {
+	// Generate a self-signed CA for kube local proxy.
+	localCAKey, localCACert, err := tlsca.GenerateSelfSignedCA(pkix.Name{
+		CommonName: "localhost",
+	}, []string{alpncommon.KubeLocalProxyWildcardDomain(teleportCluster)}, defaults.CATTL)
+	require.NoError(t, err)
+
+	// Make a mock ALB which points to the Teleport Proxy Service. Then
+	// ALPN local proxies will point to this ALB instead.
+	albProxy := helpers.MustStartMockALBProxy(t, targetAddr.String())
+
+	// Create the kube local proxy.
+	lp := mustStartALPNLocalProxyWithConfig(t, alpnproxy.LocalProxyConfig{
+		Listener:                mustCreateKubeLocalProxyListener(t, teleportCluster, localCACert, localCAKey),
+		RemoteProxyAddr:         albProxy.Addr().String(),
+		ALPNConnUpgradeRequired: true,
+		InsecureSkipVerify:      true,
+		SNI:                     constants.KubeTeleportProxyALPNPrefix + "teleport.cluster.local",
+		HTTPMiddleware:          mustCreateKubeLocalProxyMiddleware(t, teleportCluster, kubeCluster, k8ClientConfig.CertData, k8ClientConfig.KeyData),
+		Protocols:               []alpncommon.Protocol{alpncommon.ProtocolHTTP},
+	})
+	require.NoError(t, err)
+
+	// Create a proxy-url for kube clients.
+	fp := mustStartKubeForwardProxy(t, lp.GetAddr())
+
+	k8Client, err := kubernetes.NewForConfig(&rest.Config{
+		Host:  "https://" + teleportCluster,
+		Proxy: http.ProxyURL(mustParseURL(t, "http://"+fp.GetAddr())),
+		TLSClientConfig: rest.TLSClientConfig{
+			CAData:     localCACert,
+			CertData:   localCACert, // Client uses same cert as local proxy server.
+			KeyData:    localCAKey,
+			ServerName: alpncommon.KubeLocalProxySNI(teleportCluster, kubeCluster),
+		},
+	})
+	require.NoError(t, err)
+	return k8Client
 }
 
 func TestKubeIPPinning(t *testing.T) {

lib/multiplexer/multiplexer.go

@@ -483,10 +483,8 @@ const (
 	invalidProxyLineError                 = "invalid PROXY line"
 	invalidProxyV2LineError               = "invalid PROXY v2 line"
 	invalidProxySignatureError            = "could not verify PROXY signature for connection"
-	missingProxyLineError                 = `connection (%s -> %s) rejected because PROXY protocol is enabled but required
-PROXY protocol line wasn't received. 
-Make sure you have correct configuration, only enable "proxy_protocol: on" in config if Teleport is running behind L4 
-load balancer with enabled PROXY protocol.`
+	missingProxyLineError                 = `connection (%s -> %s) rejected: PROXY protocol required, but PROXY protocol line not received. Please verify your configuration. 
+Enable "proxy_protocol: on" only if Teleport is behind an L4 load balancer with PROXY protocol enabled.`
 	unknownProtocolError     = "unknown protocol"
 	unexpectedPROXYLineError = `received unexpected PROXY protocol line. Connection will be allowed, but this is usually a result of misconfiguration - 
 if Teleport is running behind L4 load balancer with enabled PROXY protocol you should explicitly set config field "proxy_protocol" to "on".
@@ -655,30 +653,29 @@ func (m *Mux) checkPROXYProtocolRequirement(conn net.Conn, unsignedPROXYLineRece
 		return trace.Wrap(err)
 	}
 
-	// We try to get inner multiplexer connection, if we succeed and there is on, it means conn was passed
-	// to us from another multiplexer listener and unsigned PROXY protocol requirement was handled there.
-	innerConn := unwrapMuxConn(conn)
-
-	if !selfConnection && innerConn == nil && !unsignedPROXYLineReceived {
+	if !selfConnection && !isInternalConn(conn) && !unsignedPROXYLineReceived {
 		return trace.BadParameter(missingProxyLineError, conn.RemoteAddr().String(), conn.LocalAddr().String())
 	}
 
 	return nil
 }
 
-func unwrapMuxConn(conn net.Conn) *Conn {
+// isInternalConn determines if the connection is a multiplexer Conn.
+// If the check is successful, it indicates that the connection was provided by another multiplexer listener,
+// and that the unsigned PROXY protocol requirement has already been handled.
+func isInternalConn(conn net.Conn) bool {
 	type netConn interface {
 		NetConn() net.Conn
 	}
 
 	for {
-		if muxConn, ok := conn.(*Conn); ok {
-			return muxConn
+		if _, ok := conn.(*Conn); ok {
+			return true
 		}
 
 		connGetter, ok := conn.(netConn)
 		if !ok {
-			return nil
+			return false
 		}
 		conn = connGetter.NetConn()
 	}

lib/web/conn_upgrade.go

@@ -243,6 +243,10 @@ func (conn *waitConn) Close() error {
 	return trace.Wrap(err)
 }
 
+func (conn *waitConn) NetConn() net.Conn {
+	return conn.Conn
+}
+
 type websocketALPNServerConn struct {
 	net.Conn
 	readBuffer []byte
@@ -262,6 +266,10 @@ func newWebSocketALPNServerConn(ctx context.Context, conn net.Conn, logger *slog
 	}
 }
 
+func (c *websocketALPNServerConn) NetConn() net.Conn {
+	return c.Conn
+}
+
 func (c *websocketALPNServerConn) Read(b []byte) (int, error) {
 	c.readMutex.Lock()
 	defer c.readMutex.Unlock()