Repository: linkerd/linkerd2
Pull request #13738 (2025-03-05 21:42:03)

build(policy): fix kubert TLS features
The policy controller runtime crate incorrectly set the 'rustls-tls' feature on the kubert dependency, even though the crate takes top-level feature flags to control its crypto implementation.

This change updates the runtime crate's feature flagging to properly alternate between rustls and openssl.

<!--  Thanks for sending a pull request!

If you already have a well-structured git commit message, chances are GitHub
set the title and description of this PR to the git commit message subject and
body, respectively. If so, you may delete these instructions and submit your PR.

If this is your first time, please read our contributor guide:
https://github.com/linkerd/linkerd2/blob/main/CONTRIBUTING.md

The title and description of your Pull Request should match the git commit
subject and body, respectively. Git commit messages are structured as follows:

```
Subject

Problem

Solution

Validation

Fixes #[GitHub issue ID]

DCO Sign off
```

Example git commit message:

```
Introduce Pull Request Template

GitHub's community guidelines recommend a pull request template, the repo was
lacking one.

Introduce a `PULL_REQUEST_TEMPLATE.md` file.

Once merged, the
[Community profile checklist](https://github.com/linkerd/linkerd2/community)
should indicate the repo now provides a pull request template.

Fixes #3321

Signed-off-by: Jane Smith <jane.smith@example.com>
```

Note the git commit message subject becomes the pull request title.

For more details around git commits, see the section on Committing in our
contributor guide:
https://github.com/linkerd/linkerd2/blob/main/CONTRIBUTING.md#committing
-->



Patch:
diff --git a/Cargo.lock b/Cargo.lock
index 3b87e7369..e69bb9173 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1262,6 +1262,7 @@ dependencies = [
  "futures-util",
  "http-body-util",
  "hyper",
+ "hyper-openssl",
  "hyper-util",
  "k8s-openapi",
  "kube-client",
@@ -1269,6 +1270,8 @@ dependencies = [
  "kube-runtime",
  "kubert-prometheus-process",
  "kubert-prometheus-tokio",
+ "once_cell",
+ "openssl",
  "parking_lot",
  "pin-project-lite",
  "prometheus-client",
@@ -1277,6 +1280,7 @@ dependencies = [
  "serde_json",
  "thiserror 2.0.11",
  "tokio",
+ "tokio-openssl",
  "tokio-rustls",
  "tower 0.5.2",
  "tower-http",
@@ -2503,6 +2507,17 @@ dependencies = [
  "tokio-stream",
 ]
 
+[[package]]
+name = "tokio-openssl"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "59df6849caa43bb7567f9a36f863c447d95a11d5903c9cc334ba32576a27eadd"
+dependencies = [
+ "openssl",
+ "openssl-sys",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-rustls"
 version = "0.26.2"
diff --git a/controller/api/destination/endpoint_profile_translator.go b/controller/api/destination/endpoint_profile_translator.go
index c5aeef331..f651f3caf 100644
--- a/controller/api/destination/endpoint_profile_translator.go
+++ b/controller/api/destination/endpoint_profile_translator.go
@@ -12,7 +12,8 @@ import (
 )
 
 type endpointProfileTranslator struct {
-	enableH2Upgrade     bool
+	forceOpaqueTransport,
+	enableH2Upgrade bool
 	controllerNS        string
 	identityTrustDomain string
 	defaultOpaquePorts  map[uint32]struct{}
@@ -44,6 +45,7 @@ var endpointProfileUpdatesQueueOverflowCounter = promauto.NewCounter(
 // newEndpointProfileTranslator translates pod updates and profile updates to
 // DestinationProfiles for endpoints
 func newEndpointProfileTranslator(
+	forceOpaqueTransport bool,
 	enableH2Upgrade bool,
 	controllerNS,
 	identityTrustDomain string,
@@ -54,10 +56,11 @@ func newEndpointProfileTranslator(
 	log *logging.Entry,
 ) *endpointProfileTranslator {
 	return &endpointProfileTranslator{
-		enableH2Upgrade:     enableH2Upgrade,
-		controllerNS:        controllerNS,
-		identityTrustDomain: identityTrustDomain,
-		defaultOpaquePorts:  defaultOpaquePorts,
+		forceOpaqueTransport: forceOpaqueTransport,
+		enableH2Upgrade:      enableH2Upgrade,
+		controllerNS:         controllerNS,
+		identityTrustDomain:  identityTrustDomain,
+		defaultOpaquePorts:   defaultOpaquePorts,
 
 		meshedHttp2ClientParams: meshedHTTP2ClientParams,
 
@@ -158,10 +161,10 @@ func (ept *endpointProfileTranslator) createEndpoint(address watcher.Address, op
 	var weightedAddr *pb.WeightedAddr
 	var err error
 	if address.ExternalWorkload != nil {
-		weightedAddr, err = createWeightedAddrForExternalWorkload(address, opaquePorts, ept.meshedHttp2ClientParams)
+		weightedAddr, err = createWeightedAddrForExternalWorkload(address, ept.forceOpaqueTransport, opaquePorts, ept.meshedHttp2ClientParams)
 	} else {
 		weightedAddr, err = createWeightedAddr(address, opaquePorts,
-			ept.enableH2Upgrade, ept.identityTrustDomain, ept.controllerNS, ept.meshedHttp2ClientParams)
+			ept.forceOpaqueTransport, ept.enableH2Upgrade, ept.identityTrustDomain, ept.controllerNS, ept.meshedHttp2ClientParams)
 	}
 	if err != nil {
 		return nil, err
diff --git a/controller/api/destination/endpoint_translator.go b/controller/api/destination/endpoint_translator.go
index 628480632..b88fc8fe3 100644
--- a/controller/api/destination/endpoint_translator.go
+++ b/controller/api/destination/endpoint_translator.go
@@ -38,6 +38,7 @@ type (
 		nodeName            string
 		defaultOpaquePorts  map[uint32]struct{}
 
+		forceOpaqueTransport,
 		enableH2Upgrade,
 		enableEndpointFiltering,
 		enableIPv6,
@@ -83,6 +84,7 @@ var updatesQueueOverflowCounter = promauto.NewCounterVec(
 func newEndpointTranslator(
 	controllerNS string,
 	identityTrustDomain string,
+	forceOpaqueTransport,
 	enableH2Upgrade,
 	enableEndpointFiltering,
 	enableIPv6,
@@ -115,6 +117,7 @@ func newEndpointTranslator(
 		nodeTopologyZone,
 		srcNodeName,
 		defaultOpaquePorts,
+		forceOpaqueTransport,
 		enableH2Upgrade,
 		enableEndpointFiltering,
 		enableIPv6,
@@ -409,14 +412,14 @@ func (et *endpointTranslator) sendClientAdd(set watcher.AddressSet) {
 		if address.Pod != nil {
 			opaquePorts = watcher.GetAnnotatedOpaquePorts(address.Pod, et.defaultOpaquePorts)
 			wa, err = createWeightedAddr(address, opaquePorts,
-				et.enableH2Upgrade, et.identityTrustDomain, et.controllerNS, et.meshedHTTP2ClientParams)
+				et.forceOpaqueTransport, et.enableH2Upgrade, et.identityTrustDomain, et.controllerNS, et.meshedHTTP2ClientParams)
 			if err != nil {
 				et.log.Errorf("Failed to translate Pod endpoints to weighted addr: %s", err)
 				continue
 			}
 		} else if address.ExternalWorkload != nil {
 			opaquePorts = watcher.GetAnnotatedOpaquePortsForExternalWorkload(address.ExternalWorkload, et.defaultOpaquePorts)
-			wa, err = createWeightedAddrForExternalWorkload(address, opaquePorts, et.meshedHTTP2ClientParams)
+			wa, err = createWeightedAddrForExternalWorkload(address, et.forceOpaqueTransport, opaquePorts, et.meshedHTTP2ClientParams)
 			if err != nil {
 				et.log.Errorf("Failed to translate ExternalWorkload endpoints to weighted addr: %s", err)
 				continue
@@ -531,6 +534,7 @@ func toAddr(address watcher.Address) (*net.TcpAddress, error) {
 
 func createWeightedAddrForExternalWorkload(
 	address watcher.Address,
+	forceOpaqueTransport bool,
 	opaquePorts map[uint32]struct{},
 	http2 *pb.Http2ClientParams,
 ) (*pb.WeightedAddr, error) {
@@ -556,21 +560,23 @@ func createWeightedAddrForExternalWorkload(
 	weightedAddr.Http2 = http2
 
 	_, opaquePort := opaquePorts[address.Port]
+	opaquePort = opaquePort || address.OpaqueProtocol
+
+	if forceOpaqueTransport || opaquePort {
+		port, err := getInboundPortFromExternalWorkload(&address.ExternalWorkload.Spec)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read inbound port from external workload: %w", err)
+		}
+		weightedAddr.ProtocolHint.OpaqueTransport = &pb.ProtocolHint_OpaqueTransport{InboundPort: port}
+	}
+
 	// If address is set as opaque by a Server, or its port is set as
 	// opaque by annotation or default value, then set the hinted protocol to
 	// Opaque.
-	if address.OpaqueProtocol || opaquePort {
+	if opaquePort {
 		weightedAddr.ProtocolHint.Protocol = &pb.ProtocolHint_Opaque_{
 			Opaque: &pb.ProtocolHint_Opaque{},
 		}
-
-		port, err := getInboundPortFromExternalWorkload(&address.ExternalWorkload.Spec)
-		if err != nil {
-			return nil, fmt.Errorf("failed to read inbound port: %w", err)
-		}
-		weightedAddr.ProtocolHint.OpaqueTransport = &pb.ProtocolHint_OpaqueTransport{
-			InboundPort: port,
-		}
 	} else {
 		weightedAddr.ProtocolHint.Protocol = &pb.ProtocolHint_H2_{
 			H2: &pb.ProtocolHint_H2{},
@@ -603,6 +609,7 @@ func createWeightedAddrForExternalWorkload(
 func createWeightedAddr(
 	address watcher.Address,
 	opaquePorts map[uint32]struct{},
+	forceOpaqueTransport bool,
 	enableH2Upgrade bool,
 	identityTrustDomain string,
 	controllerNS string,
@@ -645,20 +652,22 @@ func createWeightedAddr(
 		weightedAddr.ProtocolHint = &pb.ProtocolHint{}
 
 		_, opaquePort := opaquePorts[address.Port]
-		// If address is set as opaque by a Server, or its port is set as
-		// opaque by annotation or default value, then set the hinted protocol to
-		// Opaque.
-		if address.OpaqueProtocol || opaquePort {
-			weightedAddr.ProtocolHint.Protocol = &pb.ProtocolHint_Opaque_{
-				Opaque: &pb.ProtocolHint_Opaque{},
-			}
+		opaquePort = opaquePort || address.OpaqueProtocol
 
+		if forceOpaqueTransport || opaquePort {
 			port, err := getInboundPort(&address.Pod.Spec)
 			if err != nil {
 				return nil, fmt.Errorf("failed to read inbound port: %w", err)
 			}
-			weightedAddr.ProtocolHint.OpaqueTransport = &pb.ProtocolHint_OpaqueTransport{
-				InboundPort: port,
+			weightedAddr.ProtocolHint.OpaqueTransport = &pb.ProtocolHint_OpaqueTransport{InboundPort: port}
+		}
+
+		// If address is set as opaque by a Server, or its port is set as
+		// opaque by annotation or default value, then set the hinted protocol to
+		// Opaque.
+		if opaquePort {
+			weightedAddr.ProtocolHint.Protocol = &pb.ProtocolHint_Opaque_{
+				Opaque: &pb.ProtocolHint_Opaque{},
 			}
 		} else if enableH2Upgrade {
 			// If the pod is controlled by any Linkerd control plane, then it can be
diff --git a/controller/api/destination/federated_service_watcher.go b/controller/api/destination/federated_service_watcher.go
index 4cc356a05..5c385f26c 100644
--- a/controller/api/destination/federated_service_watcher.go
+++ b/controller/api/destination/federated_service_watcher.go
@@ -351,6 +351,7 @@ func (fs *federatedService) remoteDiscoverySubscribe(
 	translator := newEndpointTranslator(
 		fs.config.ControllerNS,
 		remoteConfig.TrustDomain,
+		fs.config.ForceOpaqueTransport,
 		fs.config.EnableH2Upgrade,
 		false, // Disable endpoint filtering for remote discovery.
 		fs.config.EnableIPv6,
@@ -399,6 +400,7 @@ func (fs *federatedService) localDiscoverySubscribe(
 	translator := newEndpointTranslator(
 		fs.config.ControllerNS,
 		fs.config.IdentityTrustDomain,
+		fs.config.ForceOpaqueTransport,
 		fs.config.EnableH2Upgrade,
 		true,
 		fs.config.EnableIPv6,
diff --git a/controller/api/destination/server.go b/controller/api/destination/server.go
index 637aeafcb..496ef003f 100644
--- a/controller/api/destination/server.go
+++ b/controller/api/destination/server.go
@@ -30,6 +30,7 @@ type (
 		IdentityTrustDomain,
 		ClusterDomain string
 
+		ForceOpaqueTransport,
 		EnableH2Upgrade,
 		EnableEndpointSlices,
 		EnableIPv6,
@@ -205,6 +206,7 @@ func (s *server) Get(dest *pb.GetDestination, stream pb.Destination_GetServer) e
 		translator := newEndpointTranslator(
 			s.config.ControllerNS,
 			remoteConfig.TrustDomain,
+			s.config.ForceOpaqueTransport,
 			s.config.EnableH2Upgrade,
 			false, // Disable endpoint filtering for remote discovery.
 			s.config.EnableIPv6,
@@ -239,6 +241,7 @@ func (s *server) Get(dest *pb.GetDestination, stream pb.Destination_GetServer) e
 		translator := newEndpointTranslator(
 			s.config.ControllerNS,
 			s.config.IdentityTrustDomain,
+			s.config.ForceOpaqueTransport,
 			s.config.EnableH2Upgrade,
 			true,
 			s.config.EnableIPv6,
@@ -531,6 +534,7 @@ func (s *server) subscribeToEndpointProfile(
 	canceled := stream.Context().Done()
 	streamEnd := make(chan struct{})
 	translator := newEndpointProfileTranslator(
+		s.config.ForceOpaqueTransport,
 		s.config.EnableH2Upgrade,
 		s.config.ControllerNS,
 		s.config.IdentityTrustDomain,
diff --git a/controller/cmd/destination/main.go b/controller/cmd/destination/main.go
index 5532ee8ae..697cba43f 100644
--- a/controller/cmd/destination/main.go
+++ b/controller/cmd/destination/main.go
@@ -32,6 +32,8 @@ func Main(args []string) {
 	metricsAddr := cmd.String("metrics-addr", ":9996", "address to serve scrapable metrics on")
 	kubeConfigPath := cmd.String("kubeconfig", "", "path to kube config")
 	controllerNamespace := cmd.String("controller-namespace", "linkerd", "namespace in which Linkerd is installed")
+	outboundTransportMode := cmd.String("outbound-transport-mode", "transparent",
+		"Force proxies to use the legacy transport for meshed traffic, i.e. transparently add TLS to the destination instead of routing to the proxy's inbound port")
 	enableH2Upgrade := cmd.Bool("enable-h2-upgrade", true,
 		"Enable transparently upgraded HTTP2 connections among pods in the service mesh")
 	enableEndpointSlices := cmd.Bool("enable-endpoint-slices", true,
@@ -166,11 +168,23 @@ func Main(args []string) {
 		log.Fatalf("Failed to initialize Cluster Store: %s", err)
 	}
 
+	var forceOpaqueTransport bool
+	switch *outboundTransportMode {
+	case "transport-header":
+		forceOpaqueTransport = true
+	case "transparent":
+		forceOpaqueTransport = false
+	default:
+		log.Errorf("Unknown value for 'outboundTransportMode': %s, defaulting to \"transparent\"", *outboundTransportMode)
+		forceOpaqueTransport = false
+	}
+
 	config := destination.Config{
 		ControllerNS:            *controllerNamespace,
 		IdentityTrustDomain:     *trustDomain,
 		ClusterDomain:           *clusterDomain,
 		DefaultOpaquePorts:      opaquePorts,
+		ForceOpaqueTransport:    forceOpaqueTransport,
 		EnableH2Upgrade:         *enableH2Upgrade,
 		EnableEndpointSlices:    *enableEndpointSlices,
 		EnableIPv6:              *enableIPv6,
diff --git a/policy-controller/runtime/Cargo.toml b/policy-controller/runtime/Cargo.toml
index 117b26d4f..1f165c075 100644
--- a/policy-controller/runtime/Cargo.toml
+++ b/policy-controller/runtime/Cargo.toml
@@ -7,11 +7,11 @@ license = "Apache-2.0"
 publish = false
 
 [features]
-default = ["openssl-tls", "openssl-vendored"]
-openssl-tls = ["kube/openssl-tls"]
+default = ["rustls-tls"]
+openssl-tls = ["kube/openssl-tls", "kubert/openssl-tls"]
 # Vendor openssl to statically link lib
-openssl-vendored = ["openssl/vendored"]
-rustls-tls = ["kube/rustls-tls"]
+openssl-vendored = ["openssl-tls", "openssl/vendored"]
+rustls-tls = ["kube/rustls-tls", "kubert/rustls-tls"]
 
 [dependencies]
 anyhow = "1"
@@ -60,7 +60,6 @@ features = [
     "prometheus-client",
     "runtime",
     "server",
-    "rustls-tls",
 ]
 
 [dependencies.tokio]


Test patch:
diff --git a/controller/api/destination/endpoint_profile_translator_test.go b/controller/api/destination/endpoint_profile_translator_test.go
index 5007e4a61..3d4e0b296 100644
--- a/controller/api/destination/endpoint_profile_translator_test.go
+++ b/controller/api/destination/endpoint_profile_translator_test.go
@@ -40,7 +40,7 @@ func TestEndpointProfileTranslator(t *testing.T) {
 		}
 		log := logging.WithField("test", t.Name())
 		translator := newEndpointProfileTranslator(
-			true, "cluster", "identity", make(map[uint32]struct{}), nil,
+			false, true, "cluster", "identity", make(map[uint32]struct{}), nil,
 			mockGetProfileServer,
 			nil,
 			log,
@@ -84,7 +84,7 @@ func TestEndpointProfileTranslator(t *testing.T) {
 		log := logging.WithField("test", t.Name())
 		endStream := make(chan struct{})
 		translator := newEndpointProfileTranslator(
-			true, "cluster", "identity", make(map[uint32]struct{}), nil,
+			false, true, "cluster", "identity", make(map[uint32]struct{}), nil,
 			mockGetProfileServer,
 			endStream,
 			log,
diff --git a/controller/api/destination/endpoint_translator_test.go b/controller/api/destination/endpoint_translator_test.go
index e40458a0c..57a637f7d 100644
--- a/controller/api/destination/endpoint_translator_test.go
+++ b/controller/api/destination/endpoint_translator_test.go
@@ -36,6 +36,17 @@ var (
 			},
 			Spec: corev1.PodSpec{
 				ServiceAccountName: "serviceaccount-name",
+				Containers: []corev1.Container{
+					{
+						Name: k8s.ProxyContainerName,
+						Env: []corev1.EnvVar{
+							{
+								Name:  envInboundListenAddr,
+								Value: "0.0.0.0:4143",
+							},
+						},
+					},
+				},
 			},
 		},
 		OwnerKind: "replicationcontroller",
@@ -56,6 +67,17 @@ var (
 			},
 			Spec: corev1.PodSpec{
 				ServiceAccountName: "serviceaccount-name",
+				Containers: []corev1.Container{
+					{
+						Name: k8s.ProxyContainerName,
+						Env: []corev1.EnvVar{
+							{
+								Name:  envInboundListenAddr,
+								Value: "[::1]:4143",
+							},
+						},
+					},
+				},
 			},
 		},
 		OwnerKind: "replicationcontroller",
@@ -74,6 +96,19 @@ var (
 					k8s.ProxyDeploymentLabel: "deployment-name",
 				},
 			},
+			Spec: corev1.PodSpec{
+				Containers: []corev1.Container{
+					{
+						Name: k8s.ProxyContainerName,
+						Env: []corev1.EnvVar{
+							{
+								Name:  envInboundListenAddr,
+								Value: "0.0.0.0:4143",
+							},
+						},
+					},
+				},
+			},
 		},
 	}
 
@@ -89,6 +124,19 @@ var (
 					k8s.ProxyDeploymentLabel: "deployment-name",
 				},
 			},
+			Spec: corev1.PodSpec{
+				Containers: []corev1.Container{
+					{
+						Name: k8s.ProxyContainerName,
+						Env: []corev1.EnvVar{
+							{
+								Name:  envInboundListenAddr,
+								Value: "[::1]:4143",
+							},
+						},
+					},
+				},
+			},
 		},
 	}
 
@@ -463,6 +511,37 @@ func TestEndpointTranslatorForPods(t *testing.T) {
 		checkAddress(t, addressesRemoved[0], pod3)
 	})
 
+	t.Run("Sends addresses with opaque transport", func(t *testing.T) {
+		expectedProtocolHint := &pb.ProtocolHint{
+			Protocol: &pb.ProtocolHint_H2_{
+				H2: &pb.ProtocolHint_H2{},
+			},
+			OpaqueTransport: &pb.ProtocolHint_OpaqueTransport{
+				InboundPort: 4143,
+			},
+		}
+
+		mockGetServer, translator := makeEndpointTranslatorWithOpaqueTransport(t, true)
+		translator.Start()
+		defer translator.Stop()
+
+		translator.Add(mkAddressSetForPods(t, pod1, pod2, pod3))
+
+		addressesAdded := (<-mockGetServer.updatesReceived).GetAdd().Addrs
+		actualNumberOfAdded := len(addressesAdded)
+		expectedNumberOfAdded := 3
+		if actualNumberOfAdded != expectedNumberOfAdded {
+			t.Fatalf("Expecting [%d] addresses to be added, got [%d]: %v", expectedNumberOfAdded, actualNumberOfAdded, addressesAdded)
+		}
+
+		for i := 0; i < 3; i++ {
+			actualProtocolHint := addressesAdded[i].GetProtocolHint()
+			if diff := deep.Equal(actualProtocolHint, expectedProtocolHint); diff != nil {
+				t.Fatalf("ProtocolHint: %v", diff)
+			}
+		}
+	})
+
 	t.Run("Sends metric labels with added addresses", func(t *testing.T) {
 		mockGetServer, translator := makeEndpointTranslator(t)
 		translator.Start()
diff --git a/controller/api/destination/test_util.go b/controller/api/destination/test_util.go
index e8e7573e0..1a6328a13 100644
--- a/controller/api/destination/test_util.go
+++ b/controller/api/destination/test_util.go
@@ -1045,6 +1045,10 @@ func (m *mockDestinationGetProfileServer) Send(profile *pb.DestinationProfile) e
 }
 
 func makeEndpointTranslator(t *testing.T) (*mockDestinationGetServer, *endpointTranslator) {
+	return makeEndpointTranslatorWithOpaqueTransport(t, false)
+}
+
+func makeEndpointTranslatorWithOpaqueTransport(t *testing.T, forceOpaqueTransport bool) (*mockDestinationGetServer, *endpointTranslator) {
 	t.Helper()
 	node := `apiVersion: v1
 kind: Node
@@ -1072,9 +1076,10 @@ metadata:
 	translator := newEndpointTranslator(
 		"linkerd",
 		"trust.domain",
-		true,
-		true,
+		forceOpaqueTransport,
+		true,  // enableH2Upgrade
 		true,  // enableEndpointFiltering
+		true,  // enableIPv6
 		false, // extEndpointZoneWeights
 		nil,   // meshedHttp2ClientParams
 		"service-name.service-ns",