Merge 43bc60e4d1 into de1a4c463c

pkg/tcpproxy/tcp.go

@@ -59,10 +59,30 @@ func (p *TCPProxy) Get(host string) *TCPServer {
 // and open a connection to the passthrough server.
 func (p *TCPProxy) Handle(conn net.Conn) {
 	defer conn.Close()
-	// See: https://www.ibm.com/docs/en/ztpf/1.1.0.15?topic=sessions-ssl-record-format
+	// It appears that the ClientHello must fit into *one* TLSPlaintext message:
-	data := make([]byte, 16384)
+	// When a client first connects to a server, it is REQUIRED to send the ClientHello as its first TLS message.
+	// Source: https://datatracker.ietf.org/doc/html/rfc8446#section-4.1.2
+	//
+	// length:  The length (in bytes) of the following TLSPlaintext.fragment. The length MUST NOT exceed 2^14 bytes.
+	// An endpoint that receives a record that exceeds this length MUST terminate the connection with a "record_overflow" alert.
+	// Source: https://datatracker.ietf.org/doc/html/rfc8446#section-5.1
+	// bytes 0  : content type
+	// bytes 1-2: legacy version
+	// bytes 3-4: length
+	// bytes 5+ : message
+	// https://en.wikipedia.org/wiki/Transport_Layer_Security#TLS_record
+	// Thus, we need to allocate 5 + 16384 bytes
+	data := make([]byte, parser.TLSHeaderLength+16384)
 
-	length, err := conn.Read(data)
+	// read the tls header first
+	_, err := io.ReadFull(conn, data[:parser.TLSHeaderLength])
+	if err != nil {
+		klog.V(4).ErrorS(err, "Error reading TLS header from the connection")
+		return
+	}
+	// get the total data length then read the rest
+	length := min(int(data[3])<<8+int(data[4])+parser.TLSHeaderLength, len(data))
+	_, err = io.ReadFull(conn, data[parser.TLSHeaderLength:length])
 	if err != nil {
 		klog.V(4).ErrorS(err, "Error reading data from the connection")
 		return
@@ -115,7 +135,7 @@ func (p *TCPProxy) Handle(conn net.Conn) {
 	} else {
 		_, err = clientConn.Write(data[:length])
 		if err != nil {
-			klog.Errorf("Error writing the first 4k of proxy data: %v", err)
+			klog.Errorf("Error writing the first %d bytes of proxy data: %v", length, err)
 			clientConn.Close()
 		}
 	}

test/e2e/framework/deployment.go

@@ -527,7 +527,7 @@ func newDeployment(name, namespace, image string, port int32, replicas int32, co
 						{
 							Name:  name,
 							Image: image,
-							Env:   []corev1.EnvVar{},
+							Env:   env,
 							Ports: []corev1.ContainerPort{
 								{
 									Name:          "http",

test/e2e/framework/httpexpect/request.go

@@ -80,22 +80,41 @@ func (h *HTTPRequest) ForceResolve(ip string, port uint16) *HTTPRequest {
 		h.chain.fail(fmt.Sprintf("invalid ip address: %s", ip))
 		return h
 	}
-	dialer := &net.Dialer{
-		Timeout:   h.client.Timeout,
-		KeepAlive: h.client.Timeout,
-		DualStack: true,
-	}
 	resolveAddr := fmt.Sprintf("%s:%d", ip, int(port))
 
+	return h.WithDialContextMiddleware(func(next DialContextFunc) DialContextFunc {
+		return func(ctx context.Context, network, _ string) (net.Conn, error) {
+			return next(ctx, network, resolveAddr)
+		}
+	})
+}
+
+// DialContextFunc is the function signature for `DialContext`
+type DialContextFunc func(ctx context.Context, network, addr string) (net.Conn, error)
+
+// WithDialContextMiddleware sets the `DialContext` function of the client
+// transport with a new function returns from `fn`. An existing `DialContext`
+// is passed into `fn` so the new function can act as a middleware by calling
+// the old one.
+func (h *HTTPRequest) WithDialContextMiddleware(fn func(next DialContextFunc) DialContextFunc) *HTTPRequest {
 	oldTransport, ok := h.client.Transport.(*http.Transport)
 	if !ok {
 		h.chain.fail("invalid old transport address")
 		return h
 	}
-	newTransport := oldTransport.Clone()
+	var nextDialContext DialContextFunc
-	newTransport.DialContext = func(ctx context.Context, network, _ string) (net.Conn, error) {
+	if oldTransport.DialContext != nil {
-		return dialer.DialContext(ctx, network, resolveAddr)
+		nextDialContext = oldTransport.DialContext
+	} else {
+		dialer := &net.Dialer{
+			Timeout:   h.client.Timeout,
+			KeepAlive: h.client.Timeout,
+			DualStack: true,
+		}
+		nextDialContext = dialer.DialContext
 	}
+	newTransport := oldTransport.Clone()
+	newTransport.DialContext = fn(nextDialContext)
 	h.client.Transport = newTransport
 	return h
 }

test/e2e/settings/ssl_passthrough.go

@@ -27,10 +27,10 @@ import (
 	"github.com/stretchr/testify/assert"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"k8s.io/ingress-nginx/test/e2e/framework"
+	"k8s.io/ingress-nginx/test/e2e/framework/httpexpect"
 )
 
 var _ = framework.IngressNginxDescribe("[Flag] enable-ssl-passthrough", func() {
@@ -75,114 +75,210 @@ var _ = framework.IngressNginxDescribe("[Flag] enable-ssl-passthrough", func() {
 				Status(http.StatusNotFound)
 		})
 
-		ginkgo.It("should pass unknown traffic to default backend and handle known traffic", func() {
+		ginkgo.Context("when handling traffic", func() {
+			var tlsConfig *tls.Config
 			host := "testpassthrough.com"
+			url := "https://" + net.JoinHostPort(host, "443")
 			echoName := "echopass"
+			secretName := host
 
-			/* Even with enable-ssl-passthrough enabled, only annotated ingresses may receive the traffic */
+			ginkgo.BeforeEach(func() {
-			annotations := map[string]string{
+				/* Even with enable-ssl-passthrough enabled, only annotated ingresses may receive the traffic */
-				"nginx.ingress.kubernetes.io/ssl-passthrough": "true",
+				annotations := map[string]string{
-			}
+					"nginx.ingress.kubernetes.io/ssl-passthrough": "true",
+				}
 
-			ingressDef := framework.NewSingleIngressWithTLS(host,
+				ingressDef := framework.NewSingleIngress(host,
-				"/",
+					"/",
-				host,
+					host,
-				[]string{host},
+					f.Namespace,
-				f.Namespace,
+					echoName,
-				echoName,
+					80,
-				80,
+					annotations)
-				annotations)
+				var err error
-			tlsConfig, err := framework.CreateIngressTLSSecret(f.KubeClientSet,
+				tlsConfig, err = framework.CreateIngressTLSSecret(f.KubeClientSet,
-				ingressDef.Spec.TLS[0].Hosts,
+					[]string{host},
-				ingressDef.Spec.TLS[0].SecretName,
+					secretName,
-				ingressDef.Namespace)
+					ingressDef.Namespace)
 
-			volumeMount := []corev1.VolumeMount{
+				volumeMount := []corev1.VolumeMount{
-				{
+					{
-					Name:      "certs",
+						Name:      "certs",
-					ReadOnly:  true,
+						ReadOnly:  true,
-					MountPath: "/certs",
+						MountPath: "/certs",
-				},
+					},
-			}
+				}
-			volume := []corev1.Volume{
+				volume := []corev1.Volume{
-				{
+					{
-					Name: "certs",
+						Name: "certs",
-					VolumeSource: corev1.VolumeSource{
+						VolumeSource: corev1.VolumeSource{
-						Secret: &corev1.SecretVolumeSource{
+							Secret: &corev1.SecretVolumeSource{
-							SecretName: ingressDef.Spec.TLS[0].SecretName,
+								SecretName: secretName,
+							},
 						},
 					},
-				},
+				}
-			}
+				envs := []corev1.EnvVar{
-			envs := []corev1.EnvVar{
+					{
-				{
+						Name:  "HTTPBUN_SSL_CERT",
-					Name:  "HTTPBUN_SSL_CERT",
+						Value: "/certs/tls.crt",
-					Value: "/certs/tls.crt",
+					},
-				},
+					{
-				{
+						Name:  "HTTPBUN_SSL_KEY",
-					Name:  "HTTPBUN_SSL_KEY",
+						Value: "/certs/tls.key",
-					Value: "/certs/tls.key",
+					},
-				},
+				}
-			}
 
-			f.NewDeploymentWithOpts("echopass",
+				f.NewDeploymentWithOpts("echopass",
-				framework.HTTPBunImage,
+					framework.HTTPBunImage,
-				80,
+					80,
-				1,
+					1,
-				nil,
+					nil,
-				nil,
+					nil,
-				envs,
+					envs,
-				volumeMount,
+					volumeMount,
-				volume,
+					volume,
-				false)
+					false)
 
-			f.EnsureIngress(ingressDef)
+				f.EnsureIngress(ingressDef)
 
-			assert.Nil(ginkgo.GinkgoT(), err)
+				assert.Nil(ginkgo.GinkgoT(), err)
-			framework.WaitForTLS(f.GetURL(framework.HTTPS), tlsConfig)
+				framework.WaitForTLS(f.GetURL(framework.HTTPS), tlsConfig)
 
-			f.WaitForNginxServer(host,
+				f.WaitForNginxServer(host,
-				func(server string) bool {
+					func(server string) bool {
-					return strings.Contains(server, "listen 442")
+						return strings.Contains(server, "listen 442")
+					})
+			})
+
+			ginkgo.It("should pass unknown traffic to default backend and handle known traffic", func() {
+				/* This one should not receive traffic as it does not contain passthrough annotation */
+				hostBad := "noannotationnopassthrough.com"
+				urlBad := "https://" + net.JoinHostPort(hostBad, "443")
+				ingBad := f.EnsureIngress(framework.NewSingleIngressWithTLS(hostBad,
+					"/",
+					hostBad,
+					[]string{hostBad},
+					f.Namespace,
+					echoName,
+					80,
+					nil))
+				tlsConfigBad, err := framework.CreateIngressTLSSecret(f.KubeClientSet,
+					ingBad.Spec.TLS[0].Hosts,
+					ingBad.Spec.TLS[0].SecretName,
+					ingBad.Namespace)
+				assert.Nil(ginkgo.GinkgoT(), err)
+				framework.WaitForTLS(f.GetURL(framework.HTTPS), tlsConfigBad)
+
+				f.WaitForNginxServer(hostBad,
+					func(server string) bool {
+						return strings.Contains(server, "listen 442")
+					})
+
+				//nolint:gosec // Ignore the gosec error in testing
+				f.HTTPTestClientWithTLSConfig(&tls.Config{ServerName: host, InsecureSkipVerify: true}).
+					GET("/").
+					WithURL(url).
+					ForceResolve(f.GetNginxIP(), 443).
+					Expect().
+					Status(http.StatusOK)
+
+				//nolint:gosec // Ignore the gosec error in testing
+				f.HTTPTestClientWithTLSConfig(&tls.Config{ServerName: hostBad, InsecureSkipVerify: true}).
+					GET("/").
+					WithURL(urlBad).
+					ForceResolve(f.GetNginxIP(), 443).
+					Expect().
+					Status(http.StatusNotFound)
+
+				f.HTTPTestClientWithTLSConfig(tlsConfig).
+					GET("/").
+					WithURL(url).
+					ForceResolve(f.GetNginxIP(), 443).
+					Expect().
+					Status(http.StatusOK)
+
+				f.HTTPTestClientWithTLSConfig(tlsConfigBad).
+					GET("/").
+					WithURL(urlBad).
+					ForceResolve(f.GetNginxIP(), 443).
+					Expect().
+					Status(http.StatusNotFound)
+			})
+
+			ginkgo.Context("on throttled connections", func() {
+				throttleMiddleware := func(next httpexpect.DialContextFunc) httpexpect.DialContextFunc {
+					return func(ctx context.Context, network, addr string) (net.Conn, error) {
+						// Wrap the connection with a throttled writer to simulate real
+						// world traffic where streaming data may arrive in chunks
+						conn, err := next(ctx, network, addr)
+						return &writeThrottledConn{
+							Conn:      conn,
+							chunkSize: len(host) / 3,
+						}, err
+					}
+				}
+
+				ginkgo.It("should handle known traffic without Host header", func() {
+					f.HTTPTestClientWithTLSConfig(tlsConfig).
+						GET("/").
+						WithURL(url).
+						ForceResolve(f.GetNginxIP(), 443).
+						WithDialContextMiddleware(throttleMiddleware).
+						Expect().
+						Status(http.StatusOK)
 				})
 
-			/* This one should not receive traffic as it does not contain passthrough annotation */
+				ginkgo.It("should handle insecure traffic without Host header", func() {
-			hostBad := "noannotationnopassthrough.com"
+					//nolint:gosec // Ignore the gosec error in testing
-			ingBad := f.EnsureIngress(framework.NewSingleIngressWithTLS(hostBad,
+					f.HTTPTestClientWithTLSConfig(&tls.Config{ServerName: host, InsecureSkipVerify: true}).
-				"/",
+						GET("/").
-				hostBad,
+						WithURL(url).
-				[]string{hostBad},
+						ForceResolve(f.GetNginxIP(), 443).
-				f.Namespace,
+						WithDialContextMiddleware(throttleMiddleware).
-				echoName,
+						Expect().
-				80,
+						Status(http.StatusOK)
-				nil))
-			tlsConfigBad, err := framework.CreateIngressTLSSecret(f.KubeClientSet,
-				ingBad.Spec.TLS[0].Hosts,
-				ingBad.Spec.TLS[0].SecretName,
-				ingBad.Namespace)
-			assert.Nil(ginkgo.GinkgoT(), err)
-			framework.WaitForTLS(f.GetURL(framework.HTTPS), tlsConfigBad)
-
-			f.WaitForNginxServer(hostBad,
-				func(server string) bool {
-					return strings.Contains(server, "listen 442")
 				})
 
-			//nolint:gosec // Ignore the gosec error in testing
+				ginkgo.It("should handle known traffic with Host header", func() {
-			f.HTTPTestClientWithTLSConfig(&tls.Config{ServerName: host, InsecureSkipVerify: true}).
+					f.HTTPTestClientWithTLSConfig(tlsConfig).
-				GET("/").
+						GET("/").
-				WithURL("https://"+net.JoinHostPort(host, "443")).
+						WithURL(url).
-				ForceResolve(f.GetNginxIP(), 443).
+						WithHeader("Host", host).
-				Expect().
+						ForceResolve(f.GetNginxIP(), 443).
-				Status(http.StatusOK)
+						WithDialContextMiddleware(throttleMiddleware).
+						Expect().
+						Status(http.StatusOK)
+				})
 
-			//nolint:gosec // Ignore the gosec error in testing
+				ginkgo.It("should handle insecure traffic with Host header", func() {
-			f.HTTPTestClientWithTLSConfig(&tls.Config{ServerName: hostBad, InsecureSkipVerify: true}).
+					//nolint:gosec // Ignore the gosec error in testing
-				GET("/").
+					f.HTTPTestClientWithTLSConfig(&tls.Config{ServerName: host, InsecureSkipVerify: true}).
-				WithURL("https://"+net.JoinHostPort(hostBad, "443")).
+						GET("/").
-				ForceResolve(f.GetNginxIP(), 443).
+						WithURL(url).
-				Expect().
+						WithHeader("Host", host).
-				Status(http.StatusNotFound)
+						ForceResolve(f.GetNginxIP(), 443).
+						WithDialContextMiddleware(throttleMiddleware).
+						Expect().
+						Status(http.StatusOK)
+				})
+			})
 		})
 	})
 })
+
+type writeThrottledConn struct {
+	net.Conn
+	chunkSize int
+}
+
+// Write writes data to the connection `chunkSize` bytes (or less) at a time.
+func (c *writeThrottledConn) Write(b []byte) (n int, err error) {
+	for i := 0; i < len(b); i += c.chunkSize {
+		n, err := c.Conn.Write(b[i:min(i+c.chunkSize, len(b))])
+		if err != nil {
+			return i + n, err
+		}
+	}
+	return len(b), nil
+}