From 6b7ccd06cfd6e2bfa5b277614a5d7821899d6beb Mon Sep 17 00:00:00 2001 From: Nitya Dhanushkodi Date: Tue, 15 Aug 2023 11:57:07 -0700 Subject: [PATCH] [NET-4799] [OSS] xdsv2: listeners L4 support for connect proxies (#18436) * refactor to avoid future import cycles --- agent/xds/clusters.go | 22 +- agent/xds/{ => config}/config.go | 2 +- agent/xds/{ => config}/config_test.go | 2 +- agent/xds/configfetcher/config_fetcher.go | 7 + agent/xds/listeners.go | 33 +- agent/xds/listeners_apigateway.go | 3 +- agent/xds/listeners_ingress.go | 3 +- agent/xds/listeners_test.go | 115 +- agent/xds/{ => naming}/naming.go | 12 +- agent/xds/{ => platform}/net_fallback.go | 4 +- agent/xds/{ => platform}/net_linux.go | 4 +- agent/xds/proxystateconverter/clusters.go | 74 + agent/xds/proxystateconverter/converter.go | 118 ++ agent/xds/proxystateconverter/listeners.go | 1694 +++++++++++++++++ agent/xds/resources.go | 5 +- agent/xds/routes.go | 5 +- agent/xds/server.go | 18 +- agent/xdsv2/cluster_resources.go | 6 + agent/xdsv2/listener_resources.go | 967 ++++++++++ agent/xdsv2/resources.go | 69 + agent/xdsv2/route_resources.go | 20 + .../pbmesh/v1alpha1/proxy_state.pb.go | 3 +- .../pbmesh/v1alpha1/proxy_state.proto | 3 +- test/integration/connect/envoy/helpers.bash | 2 +- .../connect/envoy/helpers.windows.bash | 2 +- 25 files changed, 3104 insertions(+), 89 deletions(-) rename agent/xds/{ => config}/config.go (99%) rename agent/xds/{ => config}/config_test.go (99%) create mode 100644 agent/xds/configfetcher/config_fetcher.go rename agent/xds/{ => naming}/naming.go (52%) rename agent/xds/{ => platform}/net_fallback.go (70%) rename agent/xds/{ => platform}/net_linux.go (92%) create mode 100644 agent/xds/proxystateconverter/clusters.go create mode 100644 agent/xds/proxystateconverter/converter.go create mode 100644 agent/xds/proxystateconverter/listeners.go create mode 100644 agent/xdsv2/cluster_resources.go create mode 100644 agent/xdsv2/listener_resources.go create mode 100644 agent/xdsv2/resources.go create mode 100644 agent/xdsv2/route_resources.go diff --git a/agent/xds/clusters.go b/agent/xds/clusters.go index 6907b9bc46..f56700c02a 100644 --- a/agent/xds/clusters.go +++ b/agent/xds/clusters.go @@ -19,6 +19,8 @@ import ( envoy_upstreams_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/upstreams/http/v3" envoy_matcher_v3 "github.com/envoyproxy/go-control-plane/envoy/type/matcher/v3" envoy_type_v3 "github.com/envoyproxy/go-control-plane/envoy/type/v3" + "github.com/hashicorp/consul/agent/xds/config" + "github.com/hashicorp/consul/agent/xds/naming" "github.com/hashicorp/go-hclog" "google.golang.org/protobuf/encoding/protojson" @@ -366,7 +368,7 @@ func makePassthroughClusters(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, !meshConf.TransparentProxy.MeshDestinationsOnly { clusters = append(clusters, &envoy_cluster_v3.Cluster{ - Name: OriginalDestinationClusterName, + Name: naming.OriginalDestinationClusterName, ClusterDiscoveryType: &envoy_cluster_v3.Cluster_Type{ Type: envoy_cluster_v3.Cluster_ORIGINAL_DST, }, @@ -1041,7 +1043,7 @@ func (s *ResourceGenerator) configIngressUpstreamCluster(c *envoy_cluster_v3.Clu if svc != nil { override = svc.PassiveHealthCheck } - outlierDetection := ToOutlierDetection(cfgSnap.IngressGateway.Defaults.PassiveHealthCheck, override, false) + outlierDetection := config.ToOutlierDetection(cfgSnap.IngressGateway.Defaults.PassiveHealthCheck, override, false) c.OutlierDetection = outlierDetection } @@ -1050,7 +1052,7 @@ func (s *ResourceGenerator) makeAppCluster(cfgSnap *proxycfg.ConfigSnapshot, nam var c *envoy_cluster_v3.Cluster var err error - cfg, err := ParseProxyConfig(cfgSnap.Proxy.Config) + cfg, err := config.ParseProxyConfig(cfgSnap.Proxy.Config) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -1144,7 +1146,7 @@ func (s *ResourceGenerator) makeUpstreamClusterForPeerService( clusterName := generatePeeredClusterName(uid, tbs) - outlierDetection := ToOutlierDetection(upstreamConfig.PassiveHealthCheck, nil, true) + outlierDetection := config.ToOutlierDetection(upstreamConfig.PassiveHealthCheck, nil, true) // We can't rely on health checks for services on cluster peers because they // don't take into account service resolvers, splitters and routers. Setting // MaxEjectionPercent too 100% gives outlier detection the power to eject the @@ -1279,7 +1281,7 @@ func (s *ResourceGenerator) makeUpstreamClusterForPreparedQuery(upstream structs CircuitBreakers: &envoy_cluster_v3.CircuitBreakers{ Thresholds: makeThresholdsIfNeeded(cfg.Limits), }, - OutlierDetection: ToOutlierDetection(cfg.PassiveHealthCheck, nil, true), + OutlierDetection: config.ToOutlierDetection(cfg.PassiveHealthCheck, nil, true), } if cfg.Protocol == "http2" || cfg.Protocol == "grpc" { if err := s.setHttp2ProtocolOptions(c); err != nil { @@ -1499,7 +1501,7 @@ func (s *ResourceGenerator) makeUpstreamClustersForDiscoveryChain( CircuitBreakers: &envoy_cluster_v3.CircuitBreakers{ Thresholds: makeThresholdsIfNeeded(upstreamConfig.Limits), }, - OutlierDetection: ToOutlierDetection(upstreamConfig.PassiveHealthCheck, nil, true), + OutlierDetection: config.ToOutlierDetection(upstreamConfig.PassiveHealthCheck, nil, true), } var lb *structs.LoadBalancer @@ -1676,7 +1678,7 @@ type clusterOpts struct { // makeGatewayCluster creates an Envoy cluster for a mesh or terminating gateway func (s *ResourceGenerator) makeGatewayCluster(snap *proxycfg.ConfigSnapshot, opts clusterOpts) *envoy_cluster_v3.Cluster { - cfg, err := ParseGatewayConfig(snap.Proxy.Config) + cfg, err := config.ParseGatewayConfig(snap.Proxy.Config) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -1819,7 +1821,7 @@ func configureClusterWithHostnames( // makeExternalIPCluster creates an Envoy cluster for routing to IP addresses outside of Consul // This is used by terminating gateways for Destinations func (s *ResourceGenerator) makeExternalIPCluster(snap *proxycfg.ConfigSnapshot, opts clusterOpts) *envoy_cluster_v3.Cluster { - cfg, err := ParseGatewayConfig(snap.Proxy.Config) + cfg, err := config.ParseGatewayConfig(snap.Proxy.Config) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -1858,7 +1860,7 @@ func (s *ResourceGenerator) makeExternalIPCluster(snap *proxycfg.ConfigSnapshot, // makeExternalHostnameCluster creates an Envoy cluster for hostname endpoints that will be resolved with DNS // This is used by both terminating gateways for Destinations, and Mesh Gateways for peering control plane traffice func (s *ResourceGenerator) makeExternalHostnameCluster(snap *proxycfg.ConfigSnapshot, opts clusterOpts) *envoy_cluster_v3.Cluster { - cfg, err := ParseGatewayConfig(snap.Proxy.Config) + cfg, err := config.ParseGatewayConfig(snap.Proxy.Config) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -2044,7 +2046,7 @@ func (s *ResourceGenerator) getTargetClusterName(upstreamsSnapshot *proxycfg.Con clusterName = generatePeeredClusterName(targetUID, tbs) } - clusterName = CustomizeClusterName(clusterName, chain) + clusterName = naming.CustomizeClusterName(clusterName, chain) if forMeshGateway { clusterName = meshGatewayExportedClusterNamePrefix + clusterName } diff --git a/agent/xds/config.go b/agent/xds/config/config.go similarity index 99% rename from agent/xds/config.go rename to agent/xds/config/config.go index 958326afaa..32c46d07d3 100644 --- a/agent/xds/config.go +++ b/agent/xds/config/config.go @@ -1,7 +1,7 @@ // Copyright (c) HashiCorp, Inc. // SPDX-License-Identifier: BUSL-1.1 -package xds +package config import ( "strings" diff --git a/agent/xds/config_test.go b/agent/xds/config/config_test.go similarity index 99% rename from agent/xds/config_test.go rename to agent/xds/config/config_test.go index 72ef0bb1e5..72e9a0e614 100644 --- a/agent/xds/config_test.go +++ b/agent/xds/config/config_test.go @@ -1,7 +1,7 @@ // Copyright (c) HashiCorp, Inc. // SPDX-License-Identifier: BUSL-1.1 -package xds +package config import ( "testing" diff --git a/agent/xds/configfetcher/config_fetcher.go b/agent/xds/configfetcher/config_fetcher.go new file mode 100644 index 0000000000..8d4d311caf --- /dev/null +++ b/agent/xds/configfetcher/config_fetcher.go @@ -0,0 +1,7 @@ +package configfetcher + +// ConfigFetcher is the interface the agent needs to expose +// for the xDS server to fetch agent config, currently only one field is fetched +type ConfigFetcher interface { + AdvertiseAddrLAN() string +} diff --git a/agent/xds/listeners.go b/agent/xds/listeners.go index b3b58bc2ad..68c7ebdfad 100644 --- a/agent/xds/listeners.go +++ b/agent/xds/listeners.go @@ -29,6 +29,9 @@ import ( envoy_tcp_proxy_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/tcp_proxy/v3" envoy_tls_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3" envoy_type_v3 "github.com/envoyproxy/go-control-plane/envoy/type/v3" + "github.com/hashicorp/consul/agent/xds/config" + "github.com/hashicorp/consul/agent/xds/naming" + "github.com/hashicorp/consul/agent/xds/platform" "github.com/hashicorp/go-hclog" "google.golang.org/protobuf/encoding/protojson" @@ -50,8 +53,6 @@ import ( "github.com/hashicorp/consul/types" ) -const virtualIPTag = "virtual" - // listenersFromSnapshot returns the xDS API representation of the "listeners" in the snapshot. func (s *ResourceGenerator) listenersFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) { if cfgSnap == nil { @@ -118,7 +119,7 @@ func (s *ResourceGenerator) listenersFromSnapshotConnectProxy(cfgSnap *proxycfg. } } - proxyCfg, err := ParseProxyConfig(cfgSnap.Proxy.Config) + proxyCfg, err := config.ParseProxyConfig(cfgSnap.Proxy.Config) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -258,7 +259,7 @@ func (s *ResourceGenerator) listenersFromSnapshotConnectProxy(cfgSnap *proxycfg. // We only match on this virtual IP if the upstream is in the proxy's partition. // This is because the IP is not guaranteed to be unique across k8s clusters. if acl.EqualPartitions(e.Node.PartitionOrDefault(), cfgSnap.ProxyID.PartitionOrDefault()) { - if vip := e.Service.TaggedAddresses[virtualIPTag]; vip.Address != "" { + if vip := e.Service.TaggedAddresses[naming.VirtualIPTag]; vip.Address != "" { uniqueAddrs[vip.Address] = struct{}{} } } @@ -462,7 +463,7 @@ func (s *ResourceGenerator) listenersFromSnapshotConnectProxy(cfgSnap *proxycfg. // The virtualIPTag is used by consul-k8s to store the ClusterIP for a service. // For services imported from a peer,the partition will be equal in all cases. if acl.EqualPartitions(e.Node.PartitionOrDefault(), cfgSnap.ProxyID.PartitionOrDefault()) { - if vip := e.Service.TaggedAddresses[virtualIPTag]; vip.Address != "" { + if vip := e.Service.TaggedAddresses[naming.VirtualIPTag]; vip.Address != "" { uniqueAddrs[vip.Address] = struct{}{} } } @@ -552,8 +553,8 @@ func (s *ResourceGenerator) listenersFromSnapshotConnectProxy(cfgSnap *proxycfg. filterChain, err := s.makeUpstreamFilterChain(filterChainOpts{ accessLogs: &cfgSnap.Proxy.AccessLogs, - clusterName: OriginalDestinationClusterName, - filterName: OriginalDestinationClusterName, + clusterName: naming.OriginalDestinationClusterName, + filterName: naming.OriginalDestinationClusterName, protocol: "tcp", }) if err != nil { @@ -787,7 +788,7 @@ func parseCheckPath(check structs.CheckType) (structs.ExposePath, error) { // listenersFromSnapshotGateway returns the "listener" for a terminating-gateway or mesh-gateway service func (s *ResourceGenerator) listenersFromSnapshotGateway(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) { - cfg, err := ParseGatewayConfig(cfgSnap.Proxy.Config) + cfg, err := config.ParseGatewayConfig(cfgSnap.Proxy.Config) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -1171,7 +1172,7 @@ func createDownstreamTransportSocketForConnectTLS(cfgSnap *proxycfg.ConfigSnapsh // Determine listener protocol type from configured service protocol. Don't hard fail on a config typo, //The parse func returns default config if there is an error, so it's safe to continue. - cfg, _ := ParseProxyConfig(cfgSnap.Proxy.Config) + cfg, _ := config.ParseProxyConfig(cfgSnap.Proxy.Config) // Create TLS validation context for mTLS with leaf certificate and root certs. tlsContext := makeCommonTLSContext( @@ -1263,7 +1264,7 @@ func (s *ResourceGenerator) makeInboundListener(cfgSnap *proxycfg.ConfigSnapshot var l *envoy_listener_v3.Listener var err error - cfg, err := ParseProxyConfig(cfgSnap.Proxy.Config) + cfg, err := config.ParseProxyConfig(cfgSnap.Proxy.Config) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -1513,7 +1514,7 @@ func (s *ResourceGenerator) finalizePublicListenerFromConfig(l *envoy_listener_v } func (s *ResourceGenerator) makeExposedCheckListener(cfgSnap *proxycfg.ConfigSnapshot, cluster string, path structs.ExposePath) (proto.Message, error) { - cfg, err := ParseProxyConfig(cfgSnap.Proxy.Config) + cfg, err := config.ParseProxyConfig(cfgSnap.Proxy.Config) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -1588,7 +1589,7 @@ func (s *ResourceGenerator) makeExposedCheckListener(cfgSnap *proxycfg.ConfigSna &envoy_core_v3.CidrRange{AddressPrefix: advertise, PrefixLen: &wrapperspb.UInt32Value{Value: uint32(advertiseLen)}}, ) - if ok, err := kernelSupportsIPv6(); err != nil { + if ok, err := platform.SupportsIPv6(); err != nil { return nil, err } else if ok { ranges = append(ranges, @@ -1639,7 +1640,7 @@ func (s *ResourceGenerator) makeTerminatingGatewayListener( intentions := cfgSnap.TerminatingGateway.Intentions[svc] svcConfig := cfgSnap.TerminatingGateway.ServiceConfigs[svc] - cfg, err := ParseProxyConfig(svcConfig.ProxyConfig) + cfg, err := config.ParseProxyConfig(svcConfig.ProxyConfig) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -1683,7 +1684,7 @@ func (s *ResourceGenerator) makeTerminatingGatewayListener( intentions := cfgSnap.TerminatingGateway.Intentions[svc] svcConfig := cfgSnap.TerminatingGateway.ServiceConfigs[svc] - cfg, err := ParseProxyConfig(svcConfig.ProxyConfig) + cfg, err := config.ParseProxyConfig(svcConfig.ProxyConfig) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -1807,7 +1808,7 @@ func (s *ResourceGenerator) makeFilterChainTerminatingGateway(cfgSnap *proxycfg. filterChain.Filters = append(filterChain.Filters, authFilter) } - proxyCfg, err := ParseProxyConfig(cfgSnap.Proxy.Config) + proxyCfg, err := config.ParseProxyConfig(cfgSnap.Proxy.Config) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -2128,7 +2129,7 @@ func (s *ResourceGenerator) makeMeshGatewayPeerFilterChain( if err != nil { return nil, err } - clusterName = meshGatewayExportedClusterNamePrefix + CustomizeClusterName(target.Name, chain) + clusterName = meshGatewayExportedClusterNamePrefix + naming.CustomizeClusterName(target.Name, chain) } uid := proxycfg.NewUpstreamIDFromServiceName(svc) diff --git a/agent/xds/listeners_apigateway.go b/agent/xds/listeners_apigateway.go index 3078c740a0..07566017ce 100644 --- a/agent/xds/listeners_apigateway.go +++ b/agent/xds/listeners_apigateway.go @@ -9,6 +9,7 @@ import ( envoy_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" envoy_listener_v3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" envoy_tls_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3" + "github.com/hashicorp/consul/agent/xds/naming" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/wrapperspb" @@ -70,7 +71,7 @@ func (s *ResourceGenerator) makeAPIGatewayListeners(address string, cfgSnap *pro if err != nil { return nil, err } - clusterName = CustomizeClusterName(target.Name, chain) + clusterName = naming.CustomizeClusterName(target.Name, chain) } filterName := fmt.Sprintf("%s.%s.%s.%s", chain.ServiceName, chain.Namespace, chain.Partition, chain.Datacenter) diff --git a/agent/xds/listeners_ingress.go b/agent/xds/listeners_ingress.go index 3dfd9705d9..61ef21d76a 100644 --- a/agent/xds/listeners_ingress.go +++ b/agent/xds/listeners_ingress.go @@ -9,6 +9,7 @@ import ( envoy_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" envoy_listener_v3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" envoy_tls_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3" + "github.com/hashicorp/consul/agent/xds/naming" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" @@ -62,7 +63,7 @@ func (s *ResourceGenerator) makeIngressGatewayListeners(address string, cfgSnap if err != nil { return nil, err } - clusterName = CustomizeClusterName(target.Name, chain) + clusterName = naming.CustomizeClusterName(target.Name, chain) } filterName := fmt.Sprintf("%s.%s.%s.%s", chain.ServiceName, chain.Namespace, chain.Partition, chain.Datacenter) diff --git a/agent/xds/listeners_test.go b/agent/xds/listeners_test.go index ab19ebc056..b5d6c66f00 100644 --- a/agent/xds/listeners_test.go +++ b/agent/xds/listeners_test.go @@ -11,8 +11,7 @@ import ( "text/template" "github.com/stretchr/testify/assert" - - "github.com/hashicorp/consul/agent/xds/testcommon" + "google.golang.org/protobuf/proto" envoy_listener_v3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" testinf "github.com/mitchellh/go-testing-interface" @@ -20,6 +19,10 @@ import ( "github.com/hashicorp/consul/agent/proxycfg" "github.com/hashicorp/consul/agent/structs" + "github.com/hashicorp/consul/agent/xds/configfetcher" + "github.com/hashicorp/consul/agent/xds/proxystateconverter" + "github.com/hashicorp/consul/agent/xds/testcommon" + "github.com/hashicorp/consul/agent/xdsv2" "github.com/hashicorp/consul/envoyextensions/xdscommon" "github.com/hashicorp/consul/sdk/testutil" "github.com/hashicorp/consul/types" @@ -33,6 +36,7 @@ type listenerTestCase struct { // test input. overrideGoldenName string generatorSetup func(*ResourceGenerator) + alsoRunTestForV2 bool } func makeListenerDiscoChainTests(enterprise bool) []listenerTestCase { @@ -70,6 +74,7 @@ func makeListenerDiscoChainTests(enterprise bool) []listenerTestCase { create: func(t testinf.T) *proxycfg.ConfigSnapshot { return proxycfg.TestConfigSnapshotDiscoveryChain(t, "simple", enterprise, nil, nil) }, + alsoRunTestForV2: true, }, { name: "connect-proxy-with-http-chain", @@ -118,6 +123,7 @@ func makeListenerDiscoChainTests(enterprise bool) []listenerTestCase { create: func(t testinf.T) *proxycfg.ConfigSnapshot { return proxycfg.TestConfigSnapshotDiscoveryChain(t, "external-sni", enterprise, nil, nil) }, + alsoRunTestForV2: true, }, { name: "connect-proxy-with-chain-and-overrides", @@ -130,12 +136,14 @@ func makeListenerDiscoChainTests(enterprise bool) []listenerTestCase { create: func(t testinf.T) *proxycfg.ConfigSnapshot { return proxycfg.TestConfigSnapshotDiscoveryChain(t, "failover-through-remote-gateway", enterprise, nil, nil) }, + alsoRunTestForV2: true, }, { name: "connect-proxy-with-tcp-chain-failover-through-local-gateway", create: func(t testinf.T) *proxycfg.ConfigSnapshot { return proxycfg.TestConfigSnapshotDiscoveryChain(t, "failover-through-local-gateway", enterprise, nil, nil) }, + alsoRunTestForV2: true, }, { name: "connect-proxy-with-jwt-config-entry-with-local", @@ -226,6 +234,7 @@ func TestListenersFromSnapshot(t *testing.T) { }, }) }, + alsoRunTestForV2: true, }, { name: "connect-proxy-with-tls-incoming-min-version", @@ -245,6 +254,7 @@ func TestListenersFromSnapshot(t *testing.T) { }, }) }, + alsoRunTestForV2: true, }, { name: "connect-proxy-with-tls-incoming-max-version", @@ -264,6 +274,7 @@ func TestListenersFromSnapshot(t *testing.T) { }, }) }, + alsoRunTestForV2: true, }, { name: "connect-proxy-with-tls-incoming-cipher-suites", @@ -286,6 +297,7 @@ func TestListenersFromSnapshot(t *testing.T) { }, }) }, + alsoRunTestForV2: true, }, { name: "grpc-public-listener", @@ -302,6 +314,7 @@ func TestListenersFromSnapshot(t *testing.T) { ns.Proxy.Config["bind_address"] = "127.0.0.2" }, nil) }, + alsoRunTestForV2: true, }, { name: "listener-bind-port", @@ -310,6 +323,7 @@ func TestListenersFromSnapshot(t *testing.T) { ns.Proxy.Config["bind_port"] = 8888 }, nil) }, + alsoRunTestForV2: true, }, { name: "listener-bind-address-port", @@ -319,6 +333,7 @@ func TestListenersFromSnapshot(t *testing.T) { ns.Proxy.Config["bind_port"] = 8888 }, nil) }, + alsoRunTestForV2: true, }, { name: "listener-unix-domain-socket", @@ -330,6 +345,7 @@ func TestListenersFromSnapshot(t *testing.T) { ns.Proxy.Upstreams[0].LocalBindSocketMode = "0640" }, nil) }, + alsoRunTestForV2: true, }, { name: "listener-max-inbound-connections", @@ -338,6 +354,7 @@ func TestListenersFromSnapshot(t *testing.T) { ns.Proxy.Config["max_inbound_connections"] = 222 }, nil) }, + alsoRunTestForV2: true, }, { name: "http2-public-listener", @@ -354,6 +371,7 @@ func TestListenersFromSnapshot(t *testing.T) { ns.Proxy.Config["balance_inbound_connections"] = "exact_balance" }, nil) }, + alsoRunTestForV2: true, }, { name: "listener-balance-outbound-connections-bind-port", @@ -362,6 +380,7 @@ func TestListenersFromSnapshot(t *testing.T) { ns.Proxy.Upstreams[0].Config["balance_outbound_connections"] = "exact_balance" }, nil) }, + alsoRunTestForV2: true, }, { name: "http-public-listener", @@ -559,7 +578,7 @@ func TestListenersFromSnapshot(t *testing.T) { }, { // NOTE: if IPv6 is not supported in the kernel per - // kernelSupportsIPv6() then this test will fail because the golden + // platform.SupportsIPv6() then this test will fail because the golden // files were generated assuming ipv6 support was present name: "expose-checks-http", create: proxycfg.TestConfigSnapshotExposeChecks, @@ -571,7 +590,7 @@ func TestListenersFromSnapshot(t *testing.T) { }, { // NOTE: if IPv6 is not supported in the kernel per - // kernelSupportsIPv6() then this test will fail because the golden + // platform.SupportsIPv6() then this test will fail because the golden // files were generated assuming ipv6 support was present name: "expose-checks-http-with-bind-override", create: proxycfg.TestConfigSnapshotExposeChecksWithBindOverride, @@ -583,7 +602,7 @@ func TestListenersFromSnapshot(t *testing.T) { }, { // NOTE: if IPv6 is not supported in the kernel per - // kernelSupportsIPv6() then this test will fail because the golden + // platform.SupportsIPv6() then this test will fail because the golden // files were generated assuming ipv6 support was present name: "expose-checks-grpc", create: proxycfg.TestConfigSnapshotExposeChecksGRPC, @@ -1170,16 +1189,19 @@ func TestListenersFromSnapshot(t *testing.T) { create: proxycfg.TestConfigSnapshotTransparentProxyResolverRedirectUpstream, }, { - name: "transparent-proxy-catalog-destinations-only", - create: proxycfg.TestConfigSnapshotTransparentProxyCatalogDestinationsOnly, + name: "transparent-proxy-catalog-destinations-only", + create: proxycfg.TestConfigSnapshotTransparentProxyCatalogDestinationsOnly, + alsoRunTestForV2: true, }, { - name: "transparent-proxy-dial-instances-directly", - create: proxycfg.TestConfigSnapshotTransparentProxyDialDirectly, + name: "transparent-proxy-dial-instances-directly", + create: proxycfg.TestConfigSnapshotTransparentProxyDialDirectly, + alsoRunTestForV2: true, }, { - name: "transparent-proxy-terminating-gateway", - create: proxycfg.TestConfigSnapshotTransparentProxyTerminatingGatewayCatalogDestinationsOnly, + name: "transparent-proxy-terminating-gateway", + create: proxycfg.TestConfigSnapshotTransparentProxyTerminatingGatewayCatalogDestinationsOnly, + alsoRunTestForV2: true, }, { name: "custom-trace-listener", @@ -1242,6 +1264,7 @@ func TestListenersFromSnapshot(t *testing.T) { }, nil) }, + alsoRunTestForV2: true, }, { name: "connect-proxy-without-tproxy-and-permissive-mtls", @@ -1251,6 +1274,7 @@ func TestListenersFromSnapshot(t *testing.T) { }, nil) }, + alsoRunTestForV2: true, }, } @@ -1275,26 +1299,26 @@ func TestListenersFromSnapshot(t *testing.T) { // golder files for every test case and so not be any use! testcommon.SetupTLSRootsAndLeaf(t, snap) - // Need server just for logger dependency - g := NewResourceGenerator(testutil.Logger(t), nil, false) - g.ProxyFeatures = sf - if tt.generatorSetup != nil { - tt.generatorSetup(g) - } + var listeners []proto.Message - listeners, err := g.listenersFromSnapshot(snap) - require.NoError(t, err) + t.Run("current-xdsv1", func(t *testing.T) { + // Need server just for logger dependency + g := NewResourceGenerator(testutil.Logger(t), nil, false) + g.ProxyFeatures = sf + if tt.generatorSetup != nil { + tt.generatorSetup(g) + } + listeners, err = g.listenersFromSnapshot(snap) + require.NoError(t, err) + // The order of listeners returned via LDS isn't relevant, so it's safe + // to sort these for the purposes of test comparisons. + sort.Slice(listeners, func(i, j int) bool { + return listeners[i].(*envoy_listener_v3.Listener).Name < listeners[j].(*envoy_listener_v3.Listener).Name + }) - // The order of listeners returned via LDS isn't relevant, so it's safe - // to sort these for the purposes of test comparisons. - sort.Slice(listeners, func(i, j int) bool { - return listeners[i].(*envoy_listener_v3.Listener).Name < listeners[j].(*envoy_listener_v3.Listener).Name - }) + r, err := createResponse(xdscommon.ListenerType, "00000001", "00000001", listeners) + require.NoError(t, err) - r, err := createResponse(xdscommon.ListenerType, "00000001", "00000001", listeners) - require.NoError(t, err) - - t.Run("current", func(t *testing.T) { gotJSON := protoToJSON(t, r) gName := tt.name @@ -1305,6 +1329,39 @@ func TestListenersFromSnapshot(t *testing.T) { expectedJSON := goldenEnvoy(t, filepath.Join("listeners", gName), envoyVersion, latestEnvoyVersion, gotJSON) require.JSONEq(t, expectedJSON, gotJSON) }) + + if tt.alsoRunTestForV2 { + t.Run("current-xdsv2", func(t *testing.T) { + generator := xdsv2.NewResourceGenerator(testutil.Logger(t)) + converter := proxystateconverter.NewConverter(testutil.Logger(t), nil) + proxyState, err := converter.ProxyStateFromSnapshot(snap) + require.NoError(t, err) + + res, err := generator.AllResourcesFromIR(proxyState) + require.NoError(t, err) + + listeners = res[xdscommon.ListenerType] + // The order of listeners returned via LDS isn't relevant, so it's safe + // to sort these for the purposes of test comparisons. + sort.Slice(listeners, func(i, j int) bool { + return listeners[i].(*envoy_listener_v3.Listener).Name < listeners[j].(*envoy_listener_v3.Listener).Name + }) + + r, err := createResponse(xdscommon.ListenerType, "00000001", "00000001", listeners) + require.NoError(t, err) + + gotJSON := protoToJSON(t, r) + + gName := tt.name + if tt.overrideGoldenName != "" { + gName = tt.overrideGoldenName + } + + expectedJSON := goldenEnvoy(t, filepath.Join("listeners", gName), envoyVersion, latestEnvoyVersion, gotJSON) + require.JSONEq(t, expectedJSON, gotJSON) + }) + } + }) } }) @@ -1462,7 +1519,7 @@ func customTraceJSON(t testinf.T) string { type configFetcherFunc func() string -var _ ConfigFetcher = (configFetcherFunc)(nil) +var _ configfetcher.ConfigFetcher = (configFetcherFunc)(nil) func (f configFetcherFunc) AdvertiseAddrLAN() string { return f() diff --git a/agent/xds/naming.go b/agent/xds/naming/naming.go similarity index 52% rename from agent/xds/naming.go rename to agent/xds/naming/naming.go index d9a2de2c32..3e19d93270 100644 --- a/agent/xds/naming.go +++ b/agent/xds/naming/naming.go @@ -1,7 +1,7 @@ // Copyright (c) HashiCorp, Inc. // SPDX-License-Identifier: BUSL-1.1 -package xds +package naming import ( "fmt" @@ -9,6 +9,16 @@ import ( "github.com/hashicorp/consul/agent/structs" ) +const ( + // OriginalDestinationClusterName is the name we give to the passthrough + // cluster which redirects transparently-proxied requests to their original + // destination outside the mesh. This cluster prevents Consul from blocking + // connections to destinations outside of the catalog when in transparent + // proxy mode. + OriginalDestinationClusterName = "original-destination" + VirtualIPTag = "virtual" +) + func CustomizeClusterName(clusterName string, chain *structs.CompiledDiscoveryChain) string { if chain == nil || chain.CustomizationHash == "" { return clusterName diff --git a/agent/xds/net_fallback.go b/agent/xds/platform/net_fallback.go similarity index 70% rename from agent/xds/net_fallback.go rename to agent/xds/platform/net_fallback.go index e35ed3c99a..e37c97633b 100644 --- a/agent/xds/net_fallback.go +++ b/agent/xds/platform/net_fallback.go @@ -4,8 +4,8 @@ //go:build !linux // +build !linux -package xds +package platform -func kernelSupportsIPv6() (bool, error) { +func SupportsIPv6() (bool, error) { return true, nil } diff --git a/agent/xds/net_linux.go b/agent/xds/platform/net_linux.go similarity index 92% rename from agent/xds/net_linux.go rename to agent/xds/platform/net_linux.go index 59e535f6de..acf8c53a85 100644 --- a/agent/xds/net_linux.go +++ b/agent/xds/platform/net_linux.go @@ -4,7 +4,7 @@ //go:build linux // +build linux -package xds +package platform import ( "fmt" @@ -20,7 +20,7 @@ var ( ipv6SupportedErr error ) -func kernelSupportsIPv6() (bool, error) { +func SupportsIPv6() (bool, error) { ipv6SupportOnce.Do(func() { ipv6Supported, ipv6SupportedErr = checkIfKernelSupportsIPv6() }) diff --git a/agent/xds/proxystateconverter/clusters.go b/agent/xds/proxystateconverter/clusters.go new file mode 100644 index 0000000000..b5f10bf32e --- /dev/null +++ b/agent/xds/proxystateconverter/clusters.go @@ -0,0 +1,74 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxystateconverter + +import ( + "fmt" + "strings" + + "github.com/hashicorp/consul/agent/connect" + "github.com/hashicorp/consul/agent/proxycfg" + "github.com/hashicorp/consul/agent/structs" + "github.com/hashicorp/consul/agent/xds/naming" + "github.com/hashicorp/consul/proto/private/pbpeering" +) + +const ( + meshGatewayExportedClusterNamePrefix = "exported~" +) + +func makeExposeClusterName(destinationPort int) string { + return fmt.Sprintf("exposed_cluster_%d", destinationPort) +} + +func clusterNameForDestination(cfgSnap *proxycfg.ConfigSnapshot, name string, address string, namespace string, partition string) string { + name = destinationSpecificServiceName(name, address) + sni := connect.ServiceSNI(name, "", namespace, partition, cfgSnap.Datacenter, cfgSnap.Roots.TrustDomain) + + // Prefixed with destination to distinguish from non-passthrough clusters for the same upstream. + return "destination." + sni +} + +func destinationSpecificServiceName(name string, address string) string { + address = strings.ReplaceAll(address, ":", "-") + address = strings.ReplaceAll(address, ".", "-") + return fmt.Sprintf("%s.%s", address, name) +} + +// generatePeeredClusterName returns an SNI-like cluster name which mimics PeeredServiceSNI +// but excludes partition information which could be ambiguous (local vs remote partition). +func generatePeeredClusterName(uid proxycfg.UpstreamID, tb *pbpeering.PeeringTrustBundle) string { + return strings.Join([]string{ + uid.Name, + uid.NamespaceOrDefault(), + uid.Peer, + "external", + tb.TrustDomain, + }, ".") +} + +func (s *Converter) getTargetClusterName(upstreamsSnapshot *proxycfg.ConfigSnapshotUpstreams, chain *structs.CompiledDiscoveryChain, tid string, forMeshGateway bool) string { + target := chain.Targets[tid] + clusterName := target.Name + targetUID := proxycfg.NewUpstreamIDFromTargetID(tid) + if targetUID.Peer != "" { + tbs, ok := upstreamsSnapshot.UpstreamPeerTrustBundles.Get(targetUID.Peer) + // We can't generate cluster on peers without the trust bundle. The + // trust bundle should be ready soon. + if !ok { + s.Logger.Debug("peer trust bundle not ready for discovery chain target", + "peer", targetUID.Peer, + "target", tid, + ) + return "" + } + + clusterName = generatePeeredClusterName(targetUID, tbs) + } + clusterName = naming.CustomizeClusterName(clusterName, chain) + if forMeshGateway { + clusterName = meshGatewayExportedClusterNamePrefix + clusterName + } + return clusterName +} diff --git a/agent/xds/proxystateconverter/converter.go b/agent/xds/proxystateconverter/converter.go new file mode 100644 index 0000000000..3a8037b44b --- /dev/null +++ b/agent/xds/proxystateconverter/converter.go @@ -0,0 +1,118 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxystateconverter + +import ( + "fmt" + + "github.com/hashicorp/go-hclog" + + "github.com/hashicorp/consul/agent/proxycfg" + "github.com/hashicorp/consul/agent/structs" + "github.com/hashicorp/consul/agent/xds/configfetcher" + pbmesh "github.com/hashicorp/consul/proto-public/pbmesh/v1alpha1" + "github.com/hashicorp/consul/proto-public/pbmesh/v1alpha1/pbproxystate" +) + +// Converter converts a single snapshot into a ProxyState. +type Converter struct { + Logger hclog.Logger + CfgFetcher configfetcher.ConfigFetcher + proxyState *pbmesh.ProxyState +} + +func NewConverter( + logger hclog.Logger, + cfgFetcher configfetcher.ConfigFetcher, +) *Converter { + return &Converter{ + Logger: logger, + CfgFetcher: cfgFetcher, + proxyState: &pbmesh.ProxyState{ + Listeners: make([]*pbproxystate.Listener, 0), + Clusters: make(map[string]*pbproxystate.Cluster), + Routes: make(map[string]*pbproxystate.Route), + Endpoints: make(map[string]*pbproxystate.Endpoints), + }, + } +} + +func (g *Converter) ProxyStateFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot) (*pbmesh.ProxyState, error) { + err := g.resourcesFromSnapshot(cfgSnap) + if err != nil { + return nil, fmt.Errorf("failed to generate FullProxyState: %v", err) + } + + return g.proxyState, nil +} + +func (g *Converter) resourcesFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot) error { + err := g.tlsConfigFromSnapshot(cfgSnap) + if err != nil { + return err + } + + err = g.listenersFromSnapshot(cfgSnap) + if err != nil { + return err + } + return nil +} + +const localPeerKey = "local" + +func (g *Converter) tlsConfigFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot) error { + proxyStateTLS := &pbproxystate.TLS{} + g.proxyState.TrustBundles = make(map[string]*pbproxystate.TrustBundle) + g.proxyState.LeafCertificates = make(map[string]*pbproxystate.LeafCertificate) + + // Set the TLS in the top level proxyState + g.proxyState.Tls = proxyStateTLS + + // Add local trust bundle + g.proxyState.TrustBundles[localPeerKey] = &pbproxystate.TrustBundle{ + TrustDomain: cfgSnap.Roots.TrustDomain, + Roots: []string{cfgSnap.RootPEMs()}, + } + + // Add peered trust bundles for remote peers that will dial this proxy. + for _, peeringTrustBundle := range cfgSnap.PeeringTrustBundles() { + g.proxyState.TrustBundles[peeringTrustBundle.PeerName] = &pbproxystate.TrustBundle{ + TrustDomain: peeringTrustBundle.GetTrustDomain(), + Roots: peeringTrustBundle.RootPEMs, + } + } + + // Add upstream peer trust bundles for dialing upstreams in remote peers. + upstreamsSnapshot, err := cfgSnap.ToConfigSnapshotUpstreams() + if err != nil { + if !(cfgSnap.Kind == structs.ServiceKindMeshGateway || cfgSnap.Kind == structs.ServiceKindTerminatingGateway) { + return err + } + } + if upstreamsSnapshot != nil { + upstreamsSnapshot.UpstreamPeerTrustBundles.ForEachKeyE(func(k proxycfg.PeerName) error { + tbs, ok := upstreamsSnapshot.UpstreamPeerTrustBundles.Get(k) + if ok { + g.proxyState.TrustBundles[k] = &pbproxystate.TrustBundle{ + TrustDomain: tbs.TrustDomain, + Roots: tbs.RootPEMs, + } + } + return nil + }) + } + + if cfgSnap.MeshConfigTLSOutgoing() != nil { + proxyStateTLS.OutboundTlsParameters = makeTLSParametersFromTLSConfig(cfgSnap.MeshConfigTLSOutgoing().TLSMinVersion, + cfgSnap.MeshConfigTLSOutgoing().TLSMaxVersion, cfgSnap.MeshConfigTLSOutgoing().CipherSuites) + } + + if cfgSnap.MeshConfigTLSIncoming() != nil { + proxyStateTLS.InboundTlsParameters = makeTLSParametersFromTLSConfig(cfgSnap.MeshConfigTLSIncoming().TLSMinVersion, + cfgSnap.MeshConfigTLSIncoming().TLSMaxVersion, cfgSnap.MeshConfigTLSIncoming().CipherSuites) + } + + return nil +} diff --git a/agent/xds/proxystateconverter/listeners.go b/agent/xds/proxystateconverter/listeners.go new file mode 100644 index 0000000000..a530fbf4fe --- /dev/null +++ b/agent/xds/proxystateconverter/listeners.go @@ -0,0 +1,1694 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxystateconverter + +import ( + "errors" + "fmt" + "net" + "net/url" + "regexp" + "strconv" + "strings" + "time" + + envoy_http_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3" + "github.com/hashicorp/go-uuid" + + "github.com/hashicorp/consul/agent/xds/config" + "github.com/hashicorp/consul/agent/xds/naming" + "github.com/hashicorp/consul/agent/xds/platform" + "github.com/hashicorp/consul/proto-public/pbmesh/v1alpha1/pbproxystate" + + "github.com/hashicorp/go-hclog" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/wrapperspb" + + "github.com/hashicorp/consul/acl" + "github.com/hashicorp/consul/agent/connect" + "github.com/hashicorp/consul/agent/proxycfg" + "github.com/hashicorp/consul/agent/structs" + "github.com/hashicorp/consul/envoyextensions/xdscommon" + "github.com/hashicorp/consul/sdk/iptables" + "github.com/hashicorp/consul/types" +) + +// listenersFromSnapshot adds listeners to pbmesh.ProxyState using the config snapshot. +func (s *Converter) listenersFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot) error { + if cfgSnap == nil { + return errors.New("nil config given") + } + + switch cfgSnap.Kind { + case structs.ServiceKindConnectProxy: + return s.listenersFromSnapshotConnectProxy(cfgSnap) + case structs.ServiceKindTerminatingGateway, + structs.ServiceKindMeshGateway, + structs.ServiceKindIngressGateway, + structs.ServiceKindAPIGateway: + // TODO(proxystate): gateway support will be added in the future + //return s.listenersFromSnapshotGateway(cfgSnap) + default: + return fmt.Errorf("Invalid service kind: %v", cfgSnap.Kind) + } + return nil +} + +// listenersFromSnapshotConnectProxy returns the "listeners" for a connect proxy service +func (s *Converter) listenersFromSnapshotConnectProxy(cfgSnap *proxycfg.ConfigSnapshot) error { + // This is the list of listeners we add to. It will be empty to start. + listeners := s.proxyState.Listeners + var err error + + // Configure inbound listener. + inboundListener, err := s.makeInboundListener(cfgSnap, xdscommon.PublicListenerName) + if err != nil { + return err + } + listeners = append(listeners, inboundListener) + + // This outboundListener is exclusively used when transparent proxy mode is active. + // In that situation there is a single listener where we are redirecting outbound traffic, + // and each upstream gets a filter chain attached to that listener. + var outboundListener *pbproxystate.Listener + + if cfgSnap.Proxy.Mode == structs.ProxyModeTransparent { + port := iptables.DefaultTProxyOutboundPort + if cfgSnap.Proxy.TransparentProxy.OutboundListenerPort != 0 { + port = cfgSnap.Proxy.TransparentProxy.OutboundListenerPort + } + + opts := makeListenerOpts{ + name: xdscommon.OutboundListenerName, + //accessLogs: cfgSnap.Proxy.AccessLogs, + addr: "127.0.0.1", + port: port, + direction: pbproxystate.Direction_DIRECTION_OUTBOUND, + logger: s.Logger, + } + outboundListener = makeListener(opts) + if outboundListener.Capabilities == nil { + outboundListener.Capabilities = []pbproxystate.Capability{} + } + outboundListener.Capabilities = append(outboundListener.Capabilities, pbproxystate.Capability_CAPABILITY_TRANSPARENT) + } + + // TODO(proxystate): tracing escape hatch will be added in the future. It will be added to the top level in proxystate, and used in xds generation. + //proxyCfg, err := config.ParseProxyConfig(cfgSnap.Proxy.Config) + //if err != nil { + // // Don't hard fail on a config typo, just warn. The parse func returns + // // default config if there is an error so it's safe to continue. + // s.Logger.Warn("failed to parse Connect.Proxy.Config", "error", err) + //} + //var tracing *envoy_http_v3.HttpConnectionManager_Tracing + //if proxyCfg.ListenerTracingJSON != "" { + // if tracing, err = makeTracingFromUserConfig(proxyCfg.ListenerTracingJSON); err != nil { + // s.Logger.Warn("failed to parse ListenerTracingJSON config", "error", err) + // } + //} + + upstreamsSnapshot, err := cfgSnap.ToConfigSnapshotUpstreams() + if err != nil { + return err + } + + for uid, chain := range cfgSnap.ConnectProxy.DiscoveryChain { + upstreamCfg, skip := cfgSnap.ConnectProxy.GetUpstream(uid, &cfgSnap.ProxyID.EnterpriseMeta) + if skip { + // Discovery chain is not associated with a known explicit or implicit upstream so it is skipped. + continue + } + + cfg := s.getAndModifyUpstreamConfigForListener(uid, upstreamCfg, chain) + + // If escape hatch is present, create a listener from it and move on to the next + if cfg.EnvoyListenerJSON != "" { + upstreamListener := &pbproxystate.Listener{ + EscapeHatchListener: cfg.EnvoyListenerJSON, + } + listeners = append(listeners, upstreamListener) + continue + } + + // RDS, Envoy's Route Discovery Service, is only used for HTTP services with a customized discovery chain. + useRDS := chain.Protocol != "tcp" && !chain.Default + + var clusterName string + if !useRDS { + // When not using RDS we must generate a cluster name to attach to the filter chain. + // With RDS, cluster names get attached to the dynamic routes instead. + target, err := simpleChainTarget(chain) + if err != nil { + return err + } + + clusterName = s.getTargetClusterName(upstreamsSnapshot, chain, target.ID, false) + if clusterName == "" { + continue + } + } + + filterName := fmt.Sprintf("%s.%s.%s.%s", chain.ServiceName, chain.Namespace, chain.Partition, chain.Datacenter) + + // Generate the upstream listeners for when they are explicitly set with a local bind port or socket path + if upstreamCfg != nil && upstreamCfg.HasLocalPortOrSocket() { + router, err := s.makeUpstreamRouter(routerOpts{ + // TODO(proxystate): access logs and tracing will be added in the future. + //accessLogs: &cfgSnap.Proxy.AccessLogs, + routeName: uid.EnvoyID(), + clusterName: clusterName, + filterName: filterName, + protocol: cfg.Protocol, + useRDS: useRDS, + //tracing: tracing, + }) + if err != nil { + return err + } + + opts := makeListenerOpts{ + name: uid.EnvoyID(), + //accessLogs: cfgSnap.Proxy.AccessLogs, + direction: pbproxystate.Direction_DIRECTION_OUTBOUND, + logger: s.Logger, + upstream: upstreamCfg, + } + upstreamListener := makeListener(opts) + upstreamListener.BalanceConnections = balanceConnections[cfg.BalanceOutboundConnections] + + upstreamListener.Routers = append(upstreamListener.Routers, router) + listeners = append(listeners, upstreamListener) + + // Avoid creating filter chains below for upstreams that have dedicated listeners + continue + } + + // The rest of this loop is used exclusively for transparent proxies. + // Below we create a filter chain per upstream, rather than a listener per upstream + // as we do for explicit upstreams above. + + upstreamRouter, err := s.makeUpstreamRouter(routerOpts{ + //accessLogs: &cfgSnap.Proxy.AccessLogs, + routeName: uid.EnvoyID(), + clusterName: clusterName, + filterName: filterName, + protocol: cfg.Protocol, + useRDS: useRDS, + //tracing: tracing, + }) + if err != nil { + return err + } + + endpoints := cfgSnap.ConnectProxy.WatchedUpstreamEndpoints[uid][chain.ID()] + uniqueAddrs := make(map[string]struct{}) + + if chain.Partition == cfgSnap.ProxyID.PartitionOrDefault() { + for _, ip := range chain.AutoVirtualIPs { + uniqueAddrs[ip] = struct{}{} + } + for _, ip := range chain.ManualVirtualIPs { + uniqueAddrs[ip] = struct{}{} + } + } + + // Match on the virtual IP for the upstream service (identified by the chain's ID). + // We do not match on all endpoints here since it would lead to load balancing across + // all instances when any instance address is dialed. + for _, e := range endpoints { + if e.Service.Kind == structs.ServiceKind(structs.TerminatingGateway) { + key := structs.ServiceGatewayVirtualIPTag(chain.CompoundServiceName()) + + if vip := e.Service.TaggedAddresses[key]; vip.Address != "" { + uniqueAddrs[vip.Address] = struct{}{} + } + + continue + } + if vip := e.Service.TaggedAddresses[structs.TaggedAddressVirtualIP]; vip.Address != "" { + uniqueAddrs[vip.Address] = struct{}{} + } + + // The virtualIPTag is used by consul-k8s to store the ClusterIP for a service. + // We only match on this virtual IP if the upstream is in the proxy's partition. + // This is because the IP is not guaranteed to be unique across k8s clusters. + if acl.EqualPartitions(e.Node.PartitionOrDefault(), cfgSnap.ProxyID.PartitionOrDefault()) { + if vip := e.Service.TaggedAddresses[naming.VirtualIPTag]; vip.Address != "" { + uniqueAddrs[vip.Address] = struct{}{} + } + } + } + if len(uniqueAddrs) > 2 { + s.Logger.Debug("detected multiple virtual IPs for an upstream, all will be used to match traffic", + "upstream", uid, "ip_count", len(uniqueAddrs)) + } + + // For every potential address we collected, create the appropriate address prefix to match on. + // In this case we are matching on exact addresses, so the prefix is the address itself, + // and the prefix length is based on whether it's IPv4 or IPv6. + upstreamRouter.Match = makeRouterMatchFromAddrs(uniqueAddrs) + + // Only attach the filter chain if there are addresses to match on + if upstreamRouter.Match != nil && len(upstreamRouter.Match.PrefixRanges) > 0 { + outboundListener.Routers = append(outboundListener.Routers, upstreamRouter) + } + } + requiresTLSInspector := false + requiresHTTPInspector := false + + configuredPorts := make(map[int]interface{}) + err = cfgSnap.ConnectProxy.DestinationsUpstream.ForEachKeyE(func(uid proxycfg.UpstreamID) error { + svcConfig, ok := cfgSnap.ConnectProxy.DestinationsUpstream.Get(uid) + if !ok || svcConfig == nil { + return nil + } + + if structs.IsProtocolHTTPLike(svcConfig.Protocol) { + if _, ok := configuredPorts[svcConfig.Destination.Port]; ok { + return nil + } + configuredPorts[svcConfig.Destination.Port] = struct{}{} + const name = "~http" // name used for the shared route name + routeName := clusterNameForDestination(cfgSnap, name, fmt.Sprintf("%d", svcConfig.Destination.Port), svcConfig.NamespaceOrDefault(), svcConfig.PartitionOrDefault()) + upstreamRouter, err := s.makeUpstreamRouter(routerOpts{ + //accessLogs: &cfgSnap.Proxy.AccessLogs, + routeName: routeName, + filterName: routeName, + protocol: svcConfig.Protocol, + useRDS: true, + //tracing: tracing, + }) + if err != nil { + return err + } + upstreamRouter.Match = makeRouterMatchFromAddressWithPort("", svcConfig.Destination.Port) + outboundListener.Routers = append(outboundListener.Routers, upstreamRouter) + requiresHTTPInspector = true + } else { + for _, address := range svcConfig.Destination.Addresses { + clusterName := clusterNameForDestination(cfgSnap, uid.Name, address, uid.NamespaceOrDefault(), uid.PartitionOrDefault()) + + upstreamRouter, err := s.makeUpstreamRouter(routerOpts{ + //accessLogs: &cfgSnap.Proxy.AccessLogs, + routeName: uid.EnvoyID(), + clusterName: clusterName, + filterName: clusterName, + protocol: svcConfig.Protocol, + //tracing: tracing, + }) + if err != nil { + return err + } + + upstreamRouter.Match = makeRouterMatchFromAddressWithPort(address, svcConfig.Destination.Port) + outboundListener.Routers = append(outboundListener.Routers, upstreamRouter) + + requiresTLSInspector = len(upstreamRouter.Match.ServerNames) != 0 || requiresTLSInspector + } + } + return nil + }) + if err != nil { + return err + } + + if requiresTLSInspector { + outboundListener.Capabilities = append(outboundListener.Capabilities, pbproxystate.Capability_CAPABILITY_L4_TLS_INSPECTION) + } + + if requiresHTTPInspector { + outboundListener.Capabilities = append(outboundListener.Capabilities, pbproxystate.Capability_CAPABILITY_L7_PROTOCOL_INSPECTION) + } + + // Looping over explicit and implicit upstreams is only needed for cross-peer + // because they do not have discovery chains. + for _, uid := range cfgSnap.ConnectProxy.PeeredUpstreamIDs() { + upstreamCfg, skip := cfgSnap.ConnectProxy.GetUpstream(uid, &cfgSnap.ProxyID.EnterpriseMeta) + if skip { + // Not associated with a known explicit or implicit upstream so it is skipped. + continue + } + + peerMeta, found := cfgSnap.ConnectProxy.UpstreamPeerMeta(uid) + if !found { + s.Logger.Warn("failed to fetch upstream peering metadata for listener", "uid", uid) + } + cfg := s.getAndModifyUpstreamConfigForPeeredListener(uid, upstreamCfg, peerMeta) + + // If escape hatch is present, create a listener from it and move on to the next + if cfg.EnvoyListenerJSON != "" { + upstreamListener := &pbproxystate.Listener{ + EscapeHatchListener: cfg.EnvoyListenerJSON, + } + listeners = append(listeners, upstreamListener) + continue + } + + tbs, ok := cfgSnap.ConnectProxy.UpstreamPeerTrustBundles.Get(uid.Peer) + if !ok { + // this should never happen since we loop through upstreams with + // set trust bundles + return fmt.Errorf("trust bundle not ready for peer %s", uid.Peer) + } + + clusterName := generatePeeredClusterName(uid, tbs) + + // Generate the upstream listeners for when they are explicitly set with a local bind port or socket path + if upstreamCfg != nil && upstreamCfg.HasLocalPortOrSocket() { + upstreamRouter, err := s.makeUpstreamRouter(routerOpts{ + //accessLogs: &cfgSnap.Proxy.AccessLogs, + clusterName: clusterName, + filterName: fmt.Sprintf("%s.%s.%s", + upstreamCfg.DestinationName, + upstreamCfg.DestinationNamespace, + upstreamCfg.DestinationPeer), + routeName: uid.EnvoyID(), + protocol: cfg.Protocol, + useRDS: false, + statPrefix: "upstream_peered.", + }) + if err != nil { + return err + } + + opts := makeListenerOpts{ + name: uid.EnvoyID(), + //accessLogs: cfgSnap.Proxy.AccessLogs, + direction: pbproxystate.Direction_DIRECTION_OUTBOUND, + logger: s.Logger, + upstream: upstreamCfg, + } + upstreamListener := makeListener(opts) + upstreamListener.BalanceConnections = balanceConnections[cfg.BalanceOutboundConnections] + + upstreamListener.Routers = []*pbproxystate.Router{ + upstreamRouter, + } + listeners = append(listeners, upstreamListener) + + // Avoid creating filter chains below for upstreams that have dedicated listeners + continue + } + + // The rest of this loop is used exclusively for transparent proxies. + // Below we create a filter chain per upstream, rather than a listener per upstream + // as we do for explicit upstreams above. + + upstreamRouter, err := s.makeUpstreamRouter(routerOpts{ + //accessLogs: &cfgSnap.Proxy.AccessLogs, + routeName: uid.EnvoyID(), + clusterName: clusterName, + filterName: fmt.Sprintf("%s.%s.%s", + uid.Name, + uid.NamespaceOrDefault(), + uid.Peer), + protocol: cfg.Protocol, + useRDS: false, + statPrefix: "upstream_peered.", + //tracing: tracing, + }) + if err != nil { + return err + } + + endpoints, _ := cfgSnap.ConnectProxy.PeerUpstreamEndpoints.Get(uid) + uniqueAddrs := make(map[string]struct{}) + + // Match on the virtual IP for the upstream service (identified by the chain's ID). + // We do not match on all endpoints here since it would lead to load balancing across + // all instances when any instance address is dialed. + for _, e := range endpoints { + if vip := e.Service.TaggedAddresses[structs.TaggedAddressVirtualIP]; vip.Address != "" { + uniqueAddrs[vip.Address] = struct{}{} + } + + // The virtualIPTag is used by consul-k8s to store the ClusterIP for a service. + // For services imported from a peer,the partition will be equal in all cases. + if acl.EqualPartitions(e.Node.PartitionOrDefault(), cfgSnap.ProxyID.PartitionOrDefault()) { + if vip := e.Service.TaggedAddresses[naming.VirtualIPTag]; vip.Address != "" { + uniqueAddrs[vip.Address] = struct{}{} + } + } + } + if len(uniqueAddrs) > 2 { + s.Logger.Debug("detected multiple virtual IPs for an upstream, all will be used to match traffic", + "upstream", uid, "ip_count", len(uniqueAddrs)) + } + + // For every potential address we collected, create the appropriate address prefix to match on. + // In this case we are matching on exact addresses, so the prefix is the address itself, + // and the prefix length is based on whether it's IPv4 or IPv6. + upstreamRouter.Match = makeRouterMatchFromAddrs(uniqueAddrs) + + // Only attach the filter chain if there are addresses to match on + if upstreamRouter.Match != nil && len(upstreamRouter.Match.PrefixRanges) > 0 { + outboundListener.Routers = append(outboundListener.Routers, upstreamRouter) + } + + } + + if outboundListener != nil { + // Add a passthrough for every mesh endpoint that can be dialed directly, + // as opposed to via a virtual IP. + var passthroughRouters []*pbproxystate.Router + + for _, targets := range cfgSnap.ConnectProxy.PassthroughUpstreams { + for tid, addrs := range targets { + uid := proxycfg.NewUpstreamIDFromTargetID(tid) + + sni := connect.ServiceSNI( + uid.Name, "", uid.NamespaceOrDefault(), uid.PartitionOrDefault(), cfgSnap.Datacenter, cfgSnap.Roots.TrustDomain) + + routerName := fmt.Sprintf("%s.%s.%s.%s", uid.Name, uid.NamespaceOrDefault(), uid.PartitionOrDefault(), cfgSnap.Datacenter) + + upstreamRouter, err := s.makeUpstreamRouter(routerOpts{ + //accessLogs: &cfgSnap.Proxy.AccessLogs, + clusterName: "passthrough~" + sni, + filterName: routerName, + protocol: "tcp", + }) + if err != nil { + return err + } + upstreamRouter.Match = makeRouterMatchFromAddrs(addrs) + + passthroughRouters = append(passthroughRouters, upstreamRouter) + } + } + + outboundListener.Routers = append(outboundListener.Routers, passthroughRouters...) + + // Add a catch-all filter chain that acts as a TCP proxy to destinations outside the mesh + if meshConf := cfgSnap.MeshConfig(); meshConf == nil || + !meshConf.TransparentProxy.MeshDestinationsOnly { + + upstreamRouter, err := s.makeUpstreamRouter(routerOpts{ + //accessLogs: &cfgSnap.Proxy.AccessLogs, + clusterName: naming.OriginalDestinationClusterName, + filterName: naming.OriginalDestinationClusterName, + protocol: "tcp", + }) + if err != nil { + return err + } + outboundListener.DefaultRouter = upstreamRouter + } + + // Only add the outbound listener if configured. + if len(outboundListener.Routers) > 0 || outboundListener.DefaultRouter != nil { + listeners = append(listeners, outboundListener) + + } + } + + // Looping over explicit upstreams is only needed for prepared queries because they do not have discovery chains + for uid, u := range cfgSnap.ConnectProxy.UpstreamConfig { + if u.DestinationType != structs.UpstreamDestTypePreparedQuery { + continue + } + + cfg, err := structs.ParseUpstreamConfig(u.Config) + if err != nil { + // Don't hard fail on a config typo, just warn. The parse func returns + // default config if there is an error so it's safe to continue. + s.Logger.Warn("failed to parse", "upstream", uid, "error", err) + } + + // If escape hatch is present, create a listener from it and move on to the next + if cfg.EnvoyListenerJSON != "" { + upstreamListener := &pbproxystate.Listener{ + EscapeHatchListener: cfg.EnvoyListenerJSON, + } + listeners = append(listeners, upstreamListener) + continue + } + + opts := makeListenerOpts{ + name: uid.EnvoyID(), + //accessLogs: cfgSnap.Proxy.AccessLogs, + direction: pbproxystate.Direction_DIRECTION_OUTBOUND, + logger: s.Logger, + upstream: u, + } + upstreamListener := makeListener(opts) + upstreamListener.BalanceConnections = balanceConnections[cfg.BalanceOutboundConnections] + + upstreamRouter, err := s.makeUpstreamRouter(routerOpts{ + // TODO (SNI partition) add partition for upstream SNI + //accessLogs: &cfgSnap.Proxy.AccessLogs, + clusterName: connect.UpstreamSNI(u, "", cfgSnap.Datacenter, cfgSnap.Roots.TrustDomain), + filterName: uid.EnvoyID(), + routeName: uid.EnvoyID(), + protocol: cfg.Protocol, + //tracing: tracing, + }) + if err != nil { + return err + } + upstreamListener.Routers = []*pbproxystate.Router{ + upstreamRouter, + } + listeners = append(listeners, upstreamListener) + } + + cfgSnap.Proxy.Expose.Finalize() + paths := cfgSnap.Proxy.Expose.Paths + + // Add service health checks to the list of paths to create listeners for if needed + if cfgSnap.Proxy.Expose.Checks { + psid := structs.NewServiceID(cfgSnap.Proxy.DestinationServiceID, &cfgSnap.ProxyID.EnterpriseMeta) + for _, check := range cfgSnap.ConnectProxy.WatchedServiceChecks[psid] { + p, err := parseCheckPath(check) + if err != nil { + s.Logger.Warn("failed to create listener for", "check", check.CheckID, "error", err) + continue + } + paths = append(paths, p) + } + } + + // Configure additional listener for exposed check paths + for _, path := range paths { + clusterName := xdscommon.LocalAppClusterName + if path.LocalPathPort != cfgSnap.Proxy.LocalServicePort { + clusterName = makeExposeClusterName(path.LocalPathPort) + } + + l, err := s.makeExposedCheckListener(cfgSnap, clusterName, path) + if err != nil { + return err + } + listeners = append(listeners, l) + } + + // Set listeners on the proxy state. + s.proxyState.Listeners = listeners + + return nil +} + +func makeRouterMatchFromAddrs(addrs map[string]struct{}) *pbproxystate.Match { + ranges := make([]*pbproxystate.CidrRange, 0) + + for addr := range addrs { + ip := net.ParseIP(addr) + if ip == nil { + continue + } + + pfxLen := uint32(32) + if ip.To4() == nil { + pfxLen = 128 + } + ranges = append(ranges, &pbproxystate.CidrRange{ + AddressPrefix: addr, + PrefixLen: &wrapperspb.UInt32Value{Value: pfxLen}, + }) + } + + return &pbproxystate.Match{ + PrefixRanges: ranges, + } +} + +func makeRouterMatchFromAddressWithPort(address string, port int) *pbproxystate.Match { + ranges := make([]*pbproxystate.CidrRange, 0) + + ip := net.ParseIP(address) + if ip == nil { + if address != "" { + return &pbproxystate.Match{ + ServerNames: []string{address}, + DestinationPort: &wrapperspb.UInt32Value{Value: uint32(port)}, + } + } + return &pbproxystate.Match{ + DestinationPort: &wrapperspb.UInt32Value{Value: uint32(port)}, + } + } + + pfxLen := uint32(32) + if ip.To4() == nil { + pfxLen = 128 + } + ranges = append(ranges, &pbproxystate.CidrRange{ + AddressPrefix: address, + PrefixLen: &wrapperspb.UInt32Value{Value: pfxLen}, + }) + + return &pbproxystate.Match{ + PrefixRanges: ranges, + DestinationPort: &wrapperspb.UInt32Value{Value: uint32(port)}, + } +} + +func parseCheckPath(check structs.CheckType) (structs.ExposePath, error) { + var path structs.ExposePath + + if check.HTTP != "" { + path.Protocol = "http" + + // Get path and local port from original HTTP target + u, err := url.Parse(check.HTTP) + if err != nil { + return path, fmt.Errorf("failed to parse url '%s': %v", check.HTTP, err) + } + path.Path = u.Path + + _, portStr, err := net.SplitHostPort(u.Host) + if err != nil { + return path, fmt.Errorf("failed to parse port from '%s': %v", check.HTTP, err) + } + path.LocalPathPort, err = strconv.Atoi(portStr) + if err != nil { + return path, fmt.Errorf("failed to parse port from '%s': %v", check.HTTP, err) + } + + // Get listener port from proxied HTTP target + u, err = url.Parse(check.ProxyHTTP) + if err != nil { + return path, fmt.Errorf("failed to parse url '%s': %v", check.ProxyHTTP, err) + } + + _, portStr, err = net.SplitHostPort(u.Host) + if err != nil { + return path, fmt.Errorf("failed to parse port from '%s': %v", check.ProxyHTTP, err) + } + path.ListenerPort, err = strconv.Atoi(portStr) + if err != nil { + return path, fmt.Errorf("failed to parse port from '%s': %v", check.ProxyHTTP, err) + } + } + + if check.GRPC != "" { + path.Path = "/grpc.health.v1.Health/Check" + path.Protocol = "http2" + + // Get local port from original GRPC target of the form: host/service + proxyServerAndService := strings.SplitN(check.GRPC, "/", 2) + _, portStr, err := net.SplitHostPort(proxyServerAndService[0]) + if err != nil { + return path, fmt.Errorf("failed to split host/port from '%s': %v", check.GRPC, err) + } + path.LocalPathPort, err = strconv.Atoi(portStr) + if err != nil { + return path, fmt.Errorf("failed to parse port from '%s': %v", check.GRPC, err) + } + + // Get listener port from proxied GRPC target of the form: host/service + proxyServerAndService = strings.SplitN(check.ProxyGRPC, "/", 2) + _, portStr, err = net.SplitHostPort(proxyServerAndService[0]) + if err != nil { + return path, fmt.Errorf("failed to split host/port from '%s': %v", check.ProxyGRPC, err) + } + path.ListenerPort, err = strconv.Atoi(portStr) + if err != nil { + return path, fmt.Errorf("failed to parse port from '%s': %v", check.ProxyGRPC, err) + } + } + + path.ParsedFromCheck = true + + return path, nil +} + +// TODO(proxystate): Gateway support will be added in the future. +// Functions to add from agent/xds/listeners.go: +// func listenersFromSnapshotGateway + +// makeListener returns a listener with name and bind details set. Routers and destinations +// must be added before it's useful. +// +// Note on names: Envoy listeners attempt graceful transitions of connections +// when their config changes but that means they can't have their bind address +// or port changed in a running instance. Since our users might choose to change +// a bind address or port for the public or upstream listeners, we need to +// encode those into the unique name for the listener such that if the user +// changes them, we actually create a whole new listener on the new address and +// port. Envoy should take care of closing the old one once it sees it's no +// longer in the config. +type makeListenerOpts struct { + addr string + //accessLogs structs.AccessLogsConfig + logger hclog.Logger + mode string + name string + path string + port int + direction pbproxystate.Direction + upstream *structs.Upstream +} + +func makeListener(opts makeListenerOpts) *pbproxystate.Listener { + if opts.upstream != nil && opts.upstream.LocalBindPort == 0 && opts.upstream.LocalBindSocketPath != "" { + opts.path = opts.upstream.LocalBindSocketPath + opts.mode = opts.upstream.LocalBindSocketMode + return makePipeListener(opts) + } + if opts.upstream != nil { + opts.port = opts.upstream.LocalBindPort + opts.addr = opts.upstream.LocalBindAddress + return makeListenerWithDefault(opts) + } + + return makeListenerWithDefault(opts) +} + +func makeListenerWithDefault(opts makeListenerOpts) *pbproxystate.Listener { + if opts.addr == "" { + opts.addr = "127.0.0.1" + } + // TODO(proxystate): Access logs will be added in the future. It will be added to top level IR, and used by xds code generation. + //accessLog, err := accesslogs.MakeAccessLogs(&opts.accessLogs, true) + //if err != nil && opts.logger != nil { + // // Since access logging is non-essential for routing, warn and move on + // opts.logger.Warn("error generating access log xds", err) + //} + return &pbproxystate.Listener{ + Name: fmt.Sprintf("%s:%s:%d", opts.name, opts.addr, opts.port), + //AccessLog: accessLog, + BindAddress: &pbproxystate.Listener_HostPort{ + HostPort: &pbproxystate.HostPortAddress{ + Host: opts.addr, + Port: uint32(opts.port), + }, + }, + Direction: opts.direction, + } +} + +func makePipeListener(opts makeListenerOpts) *pbproxystate.Listener { + // TODO(proxystate): Access logs will be added in the future. It will be added to top level IR, and used by xds code generation. + //accessLog, err := accesslogs.MakeAccessLogs(&opts.accessLogs, true) + //if err != nil && opts.logger != nil { + // // Since access logging is non-essential for routing, warn and move on + // opts.logger.Warn("error generating access log xds", err) + //} + return &pbproxystate.Listener{ + Name: fmt.Sprintf("%s:%s", opts.name, opts.path), + //AccessLog: accessLog, + BindAddress: &pbproxystate.Listener_UnixSocket{ + UnixSocket: &pbproxystate.UnixSocketAddress{Path: opts.path, Mode: opts.mode}, + }, + Direction: opts.direction, + } +} + +// TODO(proxystate): Escape hatches will be added in the future. +// Functions to add from agent/xds/listeners.go: +// func makeListenerFromUserConfig + +// TODO(proxystate): Intentions will be added in the future +// Functions to add from agent/xds/listeners.go: +// func injectConnectFilters + +// TODO(proxystate): httpConnectionManager constants will need to be added when used for listeners L7 in the future. +// Constants to add from agent/xds/listeners.go: +// const httpConnectionManagerOldName +// const httpConnectionManagerNewName + +// TODO(proxystate): Extracting RDS resource names will be used when wiring up xds v2 server in the future. +// Functions to add from agent/xds/listeners.go: +// func extractRdsResourceNames + +// TODO(proxystate): Intentions will be added in the future. +// Functions to add from agent/xds/listeners.go: +// func injectHTTPFilterOnFilterChains + +// NOTE: This method MUST only be used for connect proxy public listeners, +// since TLS validation will be done against root certs for all peers +// that might dial this proxy. +func (s *Converter) injectConnectTLSForPublicListener(cfgSnap *proxycfg.ConfigSnapshot, listener *pbproxystate.Listener) error { + transportSocket, err := s.createInboundMeshMTLS(cfgSnap) + if err != nil { + return err + } + + for idx := range listener.Routers { + listener.Routers[idx].InboundTls = transportSocket + } + return nil +} + +func getAlpnProtocols(protocol string) []string { + var alpnProtocols []string + + switch protocol { + case "grpc", "http2": + alpnProtocols = append(alpnProtocols, "h2", "http/1.1") + case "http": + alpnProtocols = append(alpnProtocols, "http/1.1") + } + + return alpnProtocols +} + +func (s *Converter) createInboundMeshMTLS(cfgSnap *proxycfg.ConfigSnapshot) (*pbproxystate.TransportSocket, error) { + switch cfgSnap.Kind { + case structs.ServiceKindConnectProxy: + case structs.ServiceKindMeshGateway: + default: + return nil, fmt.Errorf("cannot inject peering trust bundles for kind %q", cfgSnap.Kind) + } + + cfg, err := config.ParseProxyConfig(cfgSnap.Proxy.Config) + if err != nil { + // Don't hard fail on a config typo, just warn. The parse func returns + // default config if there is an error so it's safe to continue. + s.Logger.Warn("failed to parse Connect.Proxy.Config", "error", err) + } + + // Add all trust bundle peer names, including local. + trustBundlePeerNames := []string{"local"} + for _, tb := range cfgSnap.PeeringTrustBundles() { + trustBundlePeerNames = append(trustBundlePeerNames, tb.PeerName) + } + // Arbitrary UUID to reference the identity by. + uuid, err := uuid.GenerateUUID() + if err != nil { + return nil, err + } + // Create the transport socket + ts := &pbproxystate.TransportSocket{} + ts.ConnectionTls = &pbproxystate.TransportSocket_InboundMesh{ + InboundMesh: &pbproxystate.InboundMeshMTLS{ + IdentityKey: uuid, + ValidationContext: &pbproxystate.MeshInboundValidationContext{ + TrustBundlePeerNameKeys: trustBundlePeerNames, + }, + }, + } + s.proxyState.LeafCertificates[uuid] = &pbproxystate.LeafCertificate{ + Cert: cfgSnap.Leaf().CertPEM, + Key: cfgSnap.Leaf().PrivateKeyPEM, + } + ts.TlsParameters = makeTLSParametersFromProxyTLSConfig(cfgSnap.MeshConfigTLSIncoming()) + ts.AlpnProtocols = getAlpnProtocols(cfg.Protocol) + + return ts, nil +} + +func (s *Converter) makeInboundListener(cfgSnap *proxycfg.ConfigSnapshot, name string) (*pbproxystate.Listener, error) { + l := &pbproxystate.Listener{} + l.Routers = make([]*pbproxystate.Router, 0) + var err error + + cfg, err := config.ParseProxyConfig(cfgSnap.Proxy.Config) + if err != nil { + // Don't hard fail on a config typo, just warn. The parse func returns + // default config if there is an error so it's safe to continue. + s.Logger.Warn("failed to parse Connect.Proxy.Config", "error", err) + } + + // This controls if we do L4 or L7 intention checks. + useHTTPFilter := structs.IsProtocolHTTPLike(cfg.Protocol) + + // TODO(proxystate): Escape hatches will be added in the future. This one is a top level escape hatch. + // Generate and return custom public listener from config if one was provided. + //if cfg.PublicListenerJSON != "" { + // l, err = makeListenerFromUserConfig(cfg.PublicListenerJSON) + // if err != nil { + // return nil, err + // } + // + // // For HTTP-like services attach an RBAC http filter and do a best-effort insert + // if useHTTPFilter { + // httpAuthzFilter, err := makeRBACHTTPFilter( + // cfgSnap.ConnectProxy.Intentions, + // cfgSnap.IntentionDefaultAllow, + // rbacLocalInfo{ + // trustDomain: cfgSnap.Roots.TrustDomain, + // datacenter: cfgSnap.Datacenter, + // partition: cfgSnap.ProxyID.PartitionOrDefault(), + // }, + // cfgSnap.ConnectProxy.InboundPeerTrustBundles, + // ) + // if err != nil { + // return nil, err + // } + // + // // Try our best to inject the HTTP RBAC filter. + // if err := injectHTTPFilterOnFilterChains(l, httpAuthzFilter); err != nil { + // s.Logger.Warn( + // "could not inject the HTTP RBAC filter to enforce intentions on user-provided "+ + // "'envoy_public_listener_json' config; falling back on the RBAC network filter instead", + // "proxy", cfgSnap.ProxyID, + // "error", err, + // ) + // + // // If we get an error inject the RBAC network filter instead. + // useHTTPFilter = false + // } + // } + // + // err := s.finalizePublicListenerFromConfig(l, cfgSnap, useHTTPFilter) + // if err != nil { + // return nil, fmt.Errorf("failed to attach Consul filters and TLS context to custom public listener: %v", err) + // } + // return l, nil + //} + + // No JSON user config, use default listener address + // Default to listening on all addresses, but override with bind address if one is set. + addr := cfgSnap.Address + if addr == "" { + addr = "0.0.0.0" + } + if cfg.BindAddress != "" { + addr = cfg.BindAddress + } + + // Override with bind port if one is set, otherwise default to + // proxy service's address + port := cfgSnap.Port + if cfg.BindPort != 0 { + port = cfg.BindPort + } + + opts := makeListenerOpts{ + name: name, + //accessLogs: cfgSnap.Proxy.AccessLogs, + addr: addr, + port: port, + direction: pbproxystate.Direction_DIRECTION_INBOUND, + logger: s.Logger, + } + l = makeListener(opts) + l.BalanceConnections = balanceConnections[cfg.BalanceInboundConnections] + + // TODO(proxystate): Escape hatches will be added in the future. This one is a top level escape hatch. + //var tracing *envoy_http_v3.HttpConnectionManager_Tracing + //if cfg.ListenerTracingJSON != "" { + // if tracing, err = makeTracingFromUserConfig(cfg.ListenerTracingJSON); err != nil { + // s.Logger.Warn("failed to parse ListenerTracingJSON config", "error", err) + // } + //} + + // make local app cluster router + localAppRouter := &pbproxystate.Router{} + + destOpts := destinationOpts{ + protocol: cfg.Protocol, + filterName: name, + routeName: name, + cluster: xdscommon.LocalAppClusterName, + requestTimeoutMs: cfg.LocalRequestTimeoutMs, + idleTimeoutMs: cfg.LocalIdleTimeoutMs, + //tracing: tracing, + //accessLogs: &cfgSnap.Proxy.AccessLogs, + logger: s.Logger, + } + + err = s.addRouterDestination(destOpts, localAppRouter) + if err != nil { + return nil, err + } + + if useHTTPFilter { + l7Dest := localAppRouter.GetL7() + if l7Dest == nil { + return nil, fmt.Errorf("l7 destination on inbound listener should not be empty") + } + l7Dest.AddEmptyIntention = true + + // TODO(proxystate): L7 Intentions and JWT Auth will be added in the future. + //jwtFilter, jwtFilterErr := makeJWTAuthFilter(cfgSnap.JWTProviders, cfgSnap.ConnectProxy.Intentions) + //if jwtFilterErr != nil { + // return nil, jwtFilterErr + //} + //rbacFilter, err := makeRBACHTTPFilter( + // cfgSnap.ConnectProxy.Intentions, + // cfgSnap.IntentionDefaultAllow, + // rbacLocalInfo{ + // trustDomain: cfgSnap.Roots.TrustDomain, + // datacenter: cfgSnap.Datacenter, + // partition: cfgSnap.ProxyID.PartitionOrDefault(), + // }, + // cfgSnap.ConnectProxy.InboundPeerTrustBundles, + //) + //if err != nil { + // return nil, err + //} + // + //filterOpts.httpAuthzFilters = []*envoy_http_v3.HttpFilter{rbacFilter} + // + //if jwtFilter != nil { + // filterOpts.httpAuthzFilters = append(filterOpts.httpAuthzFilters, jwtFilter) + //} + + meshConfig := cfgSnap.MeshConfig() + includeXFCC := meshConfig == nil || meshConfig.HTTP == nil || !meshConfig.HTTP.SanitizeXForwardedClientCert + l7Dest.IncludeXfcc = includeXFCC + l7Dest.Protocol = l7Protocols[cfg.Protocol] + if cfg.MaxInboundConnections > 0 { + l7Dest.MaxInboundConnections = uint64(cfg.MaxInboundConnections) + } + } else { + l4Dest := localAppRouter.GetL4() + if l4Dest == nil { + return nil, fmt.Errorf("l4 destination on inbound listener should not be empty") + } + + if cfg.MaxInboundConnections > 0 { + l4Dest.MaxInboundConnections = uint64(cfg.MaxInboundConnections) + } + + // TODO(proxystate): Intentions will be added to l4 destination in the future. This is currently done in finalizePublicListenerFromConfig. + l4Dest.AddEmptyIntention = true + } + l.Routers = append(l.Routers, localAppRouter) + + err = s.finalizePublicListenerFromConfig(l, cfgSnap) + if err != nil { + return nil, fmt.Errorf("failed to attach Consul filters and TLS context to custom public listener: %v", err) + } + + // When permissive mTLS mode is enabled, include an additional router + // that matches on the `destination_port == `. Traffic sent + // directly to the service port is passed through to the application + // unmodified. + if cfgSnap.Proxy.Mode == structs.ProxyModeTransparent && + cfgSnap.Proxy.MutualTLSMode == structs.MutualTLSModePermissive { + router, err := makePermissiveRouter(cfgSnap, destOpts) + if err != nil { + return nil, fmt.Errorf("unable to add permissive mtls router: %w", err) + } + if router == nil { + s.Logger.Debug("no service port defined for service in permissive mTLS mode; not adding filter chain for non-mTLS traffic") + } else { + l.Routers = append(l.Routers, router) + + // With tproxy, the REDIRECT iptables target rewrites the destination ip/port + // to the proxy ip/port (e.g. 127.0.0.1:20000) for incoming packets. + // We need the original_dst filter to recover the original destination address. + l.Capabilities = append(l.Capabilities, pbproxystate.Capability_CAPABILITY_TRANSPARENT) + } + } + return l, err +} + +func makePermissiveRouter(cfgSnap *proxycfg.ConfigSnapshot, opts destinationOpts) (*pbproxystate.Router, error) { + servicePort := cfgSnap.Proxy.LocalServicePort + if servicePort <= 0 { + // No service port means the service does not accept incoming traffic, so + // the connect proxy does not need to listen for incoming non-mTLS traffic. + return nil, nil + } + + opts.statPrefix += "permissive_" + dest, err := makeL4Destination(opts) + if err != nil { + return nil, err + } + + router := &pbproxystate.Router{ + Match: &pbproxystate.Match{ + DestinationPort: &wrapperspb.UInt32Value{Value: uint32(servicePort)}, + }, + Destination: &pbproxystate.Router_L4{L4: dest}, + } + return router, nil +} + +// finalizePublicListenerFromConfig is used for best-effort injection of L4 intentions and TLS onto listeners. +func (s *Converter) finalizePublicListenerFromConfig(l *pbproxystate.Listener, cfgSnap *proxycfg.ConfigSnapshot) error { + // TODO(proxystate): L4 intentions will be added in the future. + //if !useHTTPFilter { + // // Best-effort injection of L4 intentions + // if err := s.injectConnectFilters(cfgSnap, l); err != nil { + // return nil + // } + //} + + // Always apply TLS certificates + if err := s.injectConnectTLSForPublicListener(cfgSnap, l); err != nil { + return nil + } + + return nil +} + +func (s *Converter) makeExposedCheckListener(cfgSnap *proxycfg.ConfigSnapshot, cluster string, path structs.ExposePath) (*pbproxystate.Listener, error) { + cfg, err := config.ParseProxyConfig(cfgSnap.Proxy.Config) + if err != nil { + // Don't hard fail on a config typo, just warn. The parse func returns + // default config if there is an error so it's safe to continue. + s.Logger.Warn("failed to parse Connect.Proxy.Config", "error", err) + } + + // No user config, use default listener + addr := cfgSnap.Address + + // Override with bind address if one is set, otherwise default to 0.0.0.0 + if cfg.BindAddress != "" { + addr = cfg.BindAddress + } else if addr == "" { + addr = "0.0.0.0" + } + + // Strip any special characters from path to make a valid and hopefully unique name + r := regexp.MustCompile(`[^a-zA-Z0-9]+`) + strippedPath := r.ReplaceAllString(path.Path, "") + listenerName := fmt.Sprintf("exposed_path_%s", strippedPath) + + listenerOpts := makeListenerOpts{ + name: listenerName, + //accessLogs: cfgSnap.Proxy.AccessLogs, + addr: addr, + port: path.ListenerPort, + direction: pbproxystate.Direction_DIRECTION_INBOUND, + logger: s.Logger, + } + l := makeListener(listenerOpts) + + filterName := fmt.Sprintf("exposed_path_filter_%s_%d", strippedPath, path.ListenerPort) + + destOpts := destinationOpts{ + useRDS: false, + protocol: path.Protocol, + filterName: filterName, + routeName: filterName, + cluster: cluster, + statPrefix: "", + routePath: path.Path, + httpAuthzFilters: nil, + //accessLogs: &cfgSnap.Proxy.AccessLogs, + logger: s.Logger, + // in the exposed check listener we don't set the tracing configuration + } + + router := &pbproxystate.Router{} + err = s.addRouterDestination(destOpts, router) + if err != nil { + return nil, err + } + + // For registered checks restrict traffic sources to localhost and Consul's advertise addr + if path.ParsedFromCheck { + + // For the advertise addr we use a CidrRange that only matches one address + advertise := s.CfgFetcher.AdvertiseAddrLAN() + + // Get prefix length based on whether address is ipv4 (32 bits) or ipv6 (128 bits) + advertiseLen := 32 + ip := net.ParseIP(advertise) + if ip != nil && strings.Contains(advertise, ":") { + advertiseLen = 128 + } + + ranges := make([]*pbproxystate.CidrRange, 0, 3) + ranges = append(ranges, + &pbproxystate.CidrRange{AddressPrefix: "127.0.0.1", PrefixLen: &wrapperspb.UInt32Value{Value: 8}}, + &pbproxystate.CidrRange{AddressPrefix: advertise, PrefixLen: &wrapperspb.UInt32Value{Value: uint32(advertiseLen)}}, + ) + + if ok, err := platform.SupportsIPv6(); err != nil { + return nil, err + } else if ok { + ranges = append(ranges, + &pbproxystate.CidrRange{AddressPrefix: "::1", PrefixLen: &wrapperspb.UInt32Value{Value: 128}}, + ) + } + + router.Match = &pbproxystate.Match{ + SourcePrefixRanges: ranges, + } + } + + l.Routers = []*pbproxystate.Router{router} + + return l, err +} + +// TODO(proxystate): Gateway support will be added in the future. +// Functions and types to convert from agent/xds/listeners.go: +// func makeTerminatingGatewayListener +// type terminatingGatewayFilterChainOpts +// func makeFilterChainTerminatingGateway +// func makeMeshGatewayListener +// func makeMeshGatewayPeerFilterChain + +type routerOpts struct { + //accessLogs *structs.AccessLogsConfig + routeName string + clusterName string + filterName string + protocol string + useRDS bool + statPrefix string + //forwardClientDetails bool + //forwardClientPolicy envoy_http_v3.HttpConnectionManager_ForwardClientCertDetails + //tracing *envoy_http_v3.HttpConnectionManager_Tracing +} + +func (g *Converter) makeUpstreamRouter(opts routerOpts) (*pbproxystate.Router, error) { + if opts.statPrefix == "" { + opts.statPrefix = "upstream." + } + + router := &pbproxystate.Router{} + + err := g.addRouterDestination(destinationOpts{ + useRDS: opts.useRDS, + protocol: opts.protocol, + filterName: opts.filterName, + routeName: opts.routeName, + cluster: opts.clusterName, + statPrefix: opts.statPrefix, + //forwardClientDetails: opts.forwardClientDetails, + //forwardClientPolicy: opts.forwardClientPolicy, + //tracing: opts.tracing, + //accessLogs: opts.accessLogs, + logger: g.Logger, + }, router) + if err != nil { + return nil, err + } + + return router, nil +} + +// simpleChainTarget returns the discovery target for a chain with a single node. +// A chain can have a single target if it is for a TCP service or an HTTP service without +// multiple splits/routes/failovers. +func simpleChainTarget(chain *structs.CompiledDiscoveryChain) (*structs.DiscoveryTarget, error) { + startNode := chain.Nodes[chain.StartNode] + if startNode == nil { + return nil, fmt.Errorf("missing first node in compiled discovery chain for: %s", chain.ServiceName) + } + if startNode.Type != structs.DiscoveryGraphNodeTypeResolver { + return nil, fmt.Errorf("expected discovery chain with single node, found unexpected start node: %s", startNode.Type) + } + targetID := startNode.Resolver.Target + return chain.Targets[targetID], nil +} + +func (s *Converter) getAndModifyUpstreamConfigForListener( + uid proxycfg.UpstreamID, + u *structs.Upstream, + chain *structs.CompiledDiscoveryChain, +) structs.UpstreamConfig { + var ( + cfg structs.UpstreamConfig + err error + ) + + configMap := make(map[string]interface{}) + if u != nil { + configMap = u.Config + } + if chain == nil || chain.Default { + cfg, err = structs.ParseUpstreamConfigNoDefaults(configMap) + if err != nil { + // Don't hard fail on a config typo, just warn. The parse func returns + // default config if there is an error so it's safe to continue. + s.Logger.Warn("failed to parse", "upstream", uid, "error", err) + } + } else { + // Use NoDefaults here so that we can set the protocol to the chain + // protocol if necessary + cfg, err = structs.ParseUpstreamConfigNoDefaults(configMap) + if err != nil { + // Don't hard fail on a config typo, just warn. The parse func returns + // default config if there is an error so it's safe to continue. + s.Logger.Warn("failed to parse", "upstream", uid, "error", err) + } + + if cfg.EnvoyListenerJSON != "" { + s.Logger.Warn("ignoring escape hatch setting because already configured for", + "discovery chain", chain.ServiceName, "upstream", uid, "config", "envoy_listener_json") + + // Remove from config struct so we don't use it later on + cfg.EnvoyListenerJSON = "" + } + } + protocol := cfg.Protocol + if chain != nil { + if protocol == "" { + protocol = chain.Protocol + } + if protocol == "" { + protocol = "tcp" + } + } else { + protocol = "tcp" + } + + // set back on the config so that we can use it from return value + cfg.Protocol = protocol + + return cfg +} + +func (s *Converter) getAndModifyUpstreamConfigForPeeredListener( + uid proxycfg.UpstreamID, + u *structs.Upstream, + peerMeta structs.PeeringServiceMeta, +) structs.UpstreamConfig { + var ( + cfg structs.UpstreamConfig + err error + ) + + configMap := make(map[string]interface{}) + if u != nil { + configMap = u.Config + } + + cfg, err = structs.ParseUpstreamConfigNoDefaults(configMap) + if err != nil { + // Don't hard fail on a config typo, just warn. The parse func returns + // default config if there is an error so it's safe to continue. + s.Logger.Warn("failed to parse", "upstream", uid, "error", err) + } + + // Ignore the configured protocol for peer upstreams, since it is defined by the remote + // cluster, which we cannot control. + protocol := peerMeta.Protocol + if protocol == "" { + protocol = "tcp" + } + + // set back on the config so that we can use it from return value + cfg.Protocol = protocol + + if cfg.ConnectTimeoutMs == 0 { + cfg.ConnectTimeoutMs = 5000 + } + + if cfg.MeshGateway.Mode == "" && u != nil { + cfg.MeshGateway = u.MeshGateway + } + + return cfg +} + +type destinationOpts struct { + // All listener filters + // TODO(proxystate): access logs support will be added later + //accessLogs *structs.AccessLogsConfig + cluster string + filterName string + logger hclog.Logger + protocol string + statPrefix string + + // HTTP listener filter options + forwardClientDetails bool + forwardClientPolicy envoy_http_v3.HttpConnectionManager_ForwardClientCertDetails + httpAuthzFilters []*envoy_http_v3.HttpFilter + idleTimeoutMs *int + requestTimeoutMs *int + routeName string + routePath string + // TODO(proxystate): tracing support will be added later + //tracing *envoy_http_v3.HttpConnectionManager_Tracing + useRDS bool +} + +func (g *Converter) addRouterDestination(opts destinationOpts, router *pbproxystate.Router) error { + switch opts.protocol { + case "grpc", "http2", "http": + dest, err := g.makeL7Destination(opts) + if err != nil { + return err + } + router.Destination = &pbproxystate.Router_L7{ + L7: dest, + } + return nil + case "tcp": + fallthrough + default: + if opts.useRDS { + return fmt.Errorf("RDS is not compatible with the tcp proxy filter") + } else if opts.cluster == "" { + return fmt.Errorf("cluster name is required for a tcp proxy filter") + } + dest, err := makeL4Destination(opts) + if err != nil { + return err + } + router.Destination = &pbproxystate.Router_L4{ + L4: dest, + } + return nil + } +} + +func makeL4Destination(opts destinationOpts) (*pbproxystate.L4Destination, error) { + // TODO(proxystate): implement access logs at top level + //accessLogs, err := accesslogs.MakeAccessLogs(opts.accessLogs, false) + //if err != nil && opts.logger != nil { + // opts.logger.Warn("could not make access log xds for tcp proxy", err) + //} + + l4Dest := &pbproxystate.L4Destination{ + //AccessLog: accessLogs, + Name: opts.cluster, + StatPrefix: makeStatPrefix(opts.statPrefix, opts.filterName), + } + return l4Dest, nil +} + +func makeStatPrefix(prefix, filterName string) string { + // Replace colons here because Envoy does that in the metrics for the actual + // clusters but doesn't in the stat prefix here while dashboards assume they + // will match. + return fmt.Sprintf("%s%s", prefix, strings.Replace(filterName, ":", "_", -1)) +} + +func (g *Converter) makeL7Destination(opts destinationOpts) (*pbproxystate.L7Destination, error) { + dest := &pbproxystate.L7Destination{} + + // TODO(proxystate) access logs will be added to proxystate top level and in xds generation + //accessLogs, err := accesslogs.MakeAccessLogs(opts.accessLogs, false) + //if err != nil && opts.logger != nil { + // opts.logger.Warn("could not make access log xds for http connection manager", err) + //} + + // An L7 Destination's name will be the route name, so during xds generation the route can be looked up. + dest.Name = opts.routeName + dest.StatPrefix = makeStatPrefix(opts.statPrefix, opts.filterName) + + // TODO(proxystate) tracing will be added at the top level proxystate and xds generation + //if opts.tracing != nil { + // cfg.Tracing = opts.tracing + //} + + if opts.useRDS { + if opts.cluster != "" { + return nil, fmt.Errorf("cannot specify cluster name when using RDS") + } + } else { + dest.StaticRoute = true + + if opts.cluster == "" { + return nil, fmt.Errorf("must specify cluster name when not using RDS") + } + + routeRule := &pbproxystate.RouteRule{ + Match: &pbproxystate.RouteMatch{ + PathMatch: &pbproxystate.PathMatch{ + PathMatch: &pbproxystate.PathMatch_Prefix{ + Prefix: "/", + }, + }, + // TODO(banks) Envoy supports matching only valid GRPC + // requests which might be nice to add here for gRPC services + // but it's not supported in our current envoy SDK version + // although docs say it was supported by 1.8.0. Going to defer + // that until we've updated the deps. + }, + Destination: &pbproxystate.RouteDestination{ + Destination: &pbproxystate.RouteDestination_Cluster{ + Cluster: &pbproxystate.DestinationCluster{ + Name: opts.cluster, + }, + }, + }, + } + + var timeoutCfg *pbproxystate.TimeoutConfig + r := routeRule.GetDestination() + + if opts.requestTimeoutMs != nil { + if timeoutCfg == nil { + timeoutCfg = &pbproxystate.TimeoutConfig{} + } + timeoutCfg.Timeout = durationpb.New(time.Duration(*opts.requestTimeoutMs) * time.Millisecond) + } + + if opts.idleTimeoutMs != nil { + if timeoutCfg == nil { + timeoutCfg = &pbproxystate.TimeoutConfig{} + } + timeoutCfg.IdleTimeout = durationpb.New(time.Duration(*opts.idleTimeoutMs) * time.Millisecond) + } + r.DestinationConfiguration = &pbproxystate.DestinationConfiguration{ + TimeoutConfig: timeoutCfg, + } + + // If a path is provided, do not match on a catch-all prefix + if opts.routePath != "" { + routeRule.Match.PathMatch = &pbproxystate.PathMatch{ + PathMatch: &pbproxystate.PathMatch_Exact{ + Exact: opts.routePath, + }, + } + } + + // Create static route object + route := &pbproxystate.Route{ + VirtualHosts: []*pbproxystate.VirtualHost{ + { + Name: opts.filterName, + Domains: []string{"*"}, + RouteRules: []*pbproxystate.RouteRule{ + routeRule, + }, + }, + }, + } + // Save the route to proxy state. + g.proxyState.Routes[opts.routeName] = route + } + + dest.Protocol = l7Protocols[opts.protocol] + + // TODO(proxystate) need to include xfcc policy in future L7 task + //// Note the default leads to setting HttpConnectionManager_SANITIZE + //if opts.forwardClientDetails { + // cfg.ForwardClientCertDetails = opts.forwardClientPolicy + // cfg.SetCurrentClientCertDetails = &envoy_http_v3.HttpConnectionManager_SetCurrentClientCertDetails{ + // Subject: &wrapperspb.BoolValue{Value: true}, + // Cert: true, + // Chain: true, + // Dns: true, + // Uri: true, + // } + //} + + // Like injectConnectFilters for L4, here we ensure that the first filter + // (other than the "envoy.grpc_http1_bridge" filter) in the http filter + // chain of a public listener is the authz filter to prevent unauthorized + // access and that every filter chain uses our TLS certs. + if len(opts.httpAuthzFilters) > 0 { + // TODO(proxystate) support intentions in the future + dest.Intentions = make([]*pbproxystate.L7Intention, 0) + //cfg.HttpFilters = append(opts.httpAuthzFilters, cfg.HttpFilters...) + } + + // TODO(proxystate) add grpc http filters in xds in future L7 task + //if opts.protocol == "grpc" { + // grpcHttp1Bridge, err := makeEnvoyHTTPFilter( + // "envoy.filters.http.grpc_http1_bridge", + // &envoy_grpc_http1_bridge_v3.Config{}, + // ) + // if err != nil { + // return nil, err + // } + // + // // In envoy 1.14.x the default value "stats_for_all_methods=true" was + // // deprecated, and was changed to "false" in 1.18.x. Avoid using the + // // default. TODO: we may want to expose this to users somehow easily. + // grpcStatsFilter, err := makeEnvoyHTTPFilter( + // "envoy.filters.http.grpc_stats", + // &envoy_grpc_stats_v3.FilterConfig{ + // PerMethodStatSpecifier: &envoy_grpc_stats_v3.FilterConfig_StatsForAllMethods{ + // StatsForAllMethods: makeBoolValue(true), + // }, + // }, + // ) + // if err != nil { + // return nil, err + // } + // + // // Add grpc bridge before router and authz, and the stats in front of that. + // cfg.HttpFilters = append([]*envoy_http_v3.HttpFilter{ + // grpcStatsFilter, + // grpcHttp1Bridge, + // }, cfg.HttpFilters...) + //} + + return dest, nil +} + +var tlsVersionsWithConfigurableCipherSuites = map[types.TLSVersion]struct{}{ + // Remove these two if Envoy ever sets TLS 1.3 as default minimum + types.TLSVersionUnspecified: {}, + types.TLSVersionAuto: {}, + + types.TLSv1_0: {}, + types.TLSv1_1: {}, + types.TLSv1_2: {}, +} + +func makeTLSParametersFromProxyTLSConfig(tlsConf *structs.MeshDirectionalTLSConfig) *pbproxystate.TLSParameters { + if tlsConf == nil { + return &pbproxystate.TLSParameters{} + } + + return makeTLSParametersFromTLSConfig(tlsConf.TLSMinVersion, tlsConf.TLSMaxVersion, tlsConf.CipherSuites) +} + +func makeTLSParametersFromTLSConfig( + tlsMinVersion types.TLSVersion, + tlsMaxVersion types.TLSVersion, + cipherSuites []types.TLSCipherSuite, +) *pbproxystate.TLSParameters { + tlsParams := pbproxystate.TLSParameters{} + + if tlsMinVersion != types.TLSVersionUnspecified { + tlsParams.MinVersion = tlsVersions[tlsMinVersion] + } + if tlsMaxVersion != types.TLSVersionUnspecified { + tlsParams.MaxVersion = tlsVersions[tlsMaxVersion] + } + if len(cipherSuites) != 0 { + var suites []pbproxystate.TLSCipherSuite + for _, cs := range cipherSuites { + suites = append(suites, tlsCipherSuites[cs]) + } + tlsParams.CipherSuites = suites + } + + return &tlsParams +} + +var tlsVersions = map[types.TLSVersion]pbproxystate.TLSVersion{ + types.TLSVersionAuto: pbproxystate.TLSVersion_TLS_VERSION_AUTO, + types.TLSv1_0: pbproxystate.TLSVersion_TLS_VERSION_1_0, + types.TLSv1_1: pbproxystate.TLSVersion_TLS_VERSION_1_1, + types.TLSv1_2: pbproxystate.TLSVersion_TLS_VERSION_1_2, + types.TLSv1_3: pbproxystate.TLSVersion_TLS_VERSION_1_3, +} + +var tlsCipherSuites = map[types.TLSCipherSuite]pbproxystate.TLSCipherSuite{ + types.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_AES128_GCM_SHA256, + types.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_CHACHA20_POLY1305, + types.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_AES128_GCM_SHA256, + types.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_CHACHA20_POLY1305, + types.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_AES128_SHA, + types.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_AES128_SHA, + types.TLS_RSA_WITH_AES_128_GCM_SHA256: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_AES128_GCM_SHA256, + types.TLS_RSA_WITH_AES_128_CBC_SHA: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_AES128_SHA, + types.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_AES256_GCM_SHA384, + types.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_AES256_GCM_SHA384, + types.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_AES256_SHA, + types.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_AES256_SHA, + types.TLS_RSA_WITH_AES_256_GCM_SHA384: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_AES256_GCM_SHA384, + types.TLS_RSA_WITH_AES_256_CBC_SHA: pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_AES256_SHA, +} + +var l7Protocols = map[string]pbproxystate.L7Protocol{ + "http": pbproxystate.L7Protocol_L7_PROTOCOL_HTTP, + "http2": pbproxystate.L7Protocol_L7_PROTOCOL_HTTP2, + "grpc": pbproxystate.L7Protocol_L7_PROTOCOL_GRPC, +} + +var balanceConnections = map[string]pbproxystate.BalanceConnections{ + "": pbproxystate.BalanceConnections_BALANCE_CONNECTIONS_DEFAULT, + structs.ConnectionExactBalance: pbproxystate.BalanceConnections_BALANCE_CONNECTIONS_EXACT, +} diff --git a/agent/xds/resources.go b/agent/xds/resources.go index 15378861f3..fc76104002 100644 --- a/agent/xds/resources.go +++ b/agent/xds/resources.go @@ -6,6 +6,7 @@ package xds import ( "fmt" + "github.com/hashicorp/consul/agent/xds/configfetcher" "github.com/hashicorp/go-hclog" "google.golang.org/protobuf/proto" @@ -18,7 +19,7 @@ import ( // resources for a single client. type ResourceGenerator struct { Logger hclog.Logger - CfgFetcher ConfigFetcher + CfgFetcher configfetcher.ConfigFetcher IncrementalXDS bool ProxyFeatures xdscommon.SupportedProxyFeatures @@ -26,7 +27,7 @@ type ResourceGenerator struct { func NewResourceGenerator( logger hclog.Logger, - cfgFetcher ConfigFetcher, + cfgFetcher configfetcher.ConfigFetcher, incrementalXDS bool, ) *ResourceGenerator { return &ResourceGenerator{ diff --git a/agent/xds/routes.go b/agent/xds/routes.go index 92b7d6db0c..31b1c4a932 100644 --- a/agent/xds/routes.go +++ b/agent/xds/routes.go @@ -22,6 +22,7 @@ import ( "github.com/hashicorp/consul/agent/consul/discoverychain" "github.com/hashicorp/consul/agent/proxycfg" "github.com/hashicorp/consul/agent/structs" + "github.com/hashicorp/consul/agent/xds/config" ) // routesFromSnapshot returns the xDS API representation of the "routes" in the @@ -140,7 +141,7 @@ func (s *ResourceGenerator) routesForTerminatingGateway(cfgSnap *proxycfg.Config var resources []proto.Message for _, svc := range cfgSnap.TerminatingGateway.ValidServices() { clusterName := connect.ServiceSNI(svc.Name, "", svc.NamespaceOrDefault(), svc.PartitionOrDefault(), cfgSnap.Datacenter, cfgSnap.Roots.TrustDomain) - cfg, err := ParseProxyConfig(cfgSnap.TerminatingGateway.ServiceConfigs[svc].ProxyConfig) + cfg, err := config.ParseProxyConfig(cfgSnap.TerminatingGateway.ServiceConfigs[svc].ProxyConfig) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. @@ -168,7 +169,7 @@ func (s *ResourceGenerator) routesForTerminatingGateway(cfgSnap *proxycfg.Config for _, address := range svcConfig.Destination.Addresses { clusterName := clusterNameForDestination(cfgSnap, svc.Name, address, svc.NamespaceOrDefault(), svc.PartitionOrDefault()) - cfg, err := ParseProxyConfig(cfgSnap.TerminatingGateway.ServiceConfigs[svc].ProxyConfig) + cfg, err := config.ParseProxyConfig(cfgSnap.TerminatingGateway.ServiceConfigs[svc].ProxyConfig) if err != nil { // Don't hard fail on a config typo, just warn. The parse func returns // default config if there is an error so it's safe to continue. diff --git a/agent/xds/server.go b/agent/xds/server.go index c8a5fa1086..c8ffeef842 100644 --- a/agent/xds/server.go +++ b/agent/xds/server.go @@ -10,6 +10,7 @@ import ( "time" envoy_discovery_v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" + "github.com/hashicorp/consul/agent/xds/configfetcher" "github.com/hashicorp/consul/envoyextensions/xdscommon" @@ -70,13 +71,6 @@ const ( // services named "local_agent" in the future. LocalAgentClusterName = "local_agent" - // OriginalDestinationClusterName is the name we give to the passthrough - // cluster which redirects transparently-proxied requests to their original - // destination outside the mesh. This cluster prevents Consul from blocking - // connections to destinations outside of the catalog when in transparent - // proxy mode. - OriginalDestinationClusterName = "original-destination" - // DefaultAuthCheckFrequency is the default value for // Server.AuthCheckFrequency to use when the zero value is provided. DefaultAuthCheckFrequency = 5 * time.Minute @@ -88,12 +82,6 @@ const ( // coupling this to the agent. type ACLResolverFunc func(id string) (acl.Authorizer, error) -// ConfigFetcher is the interface the agent needs to expose -// for the xDS server to fetch agent config, currently only one field is fetched -type ConfigFetcher interface { - AdvertiseAddrLAN() string -} - // ProxyConfigSource is the interface xds.Server requires to consume proxy // config updates. type ProxyConfigSource interface { @@ -110,7 +98,7 @@ type Server struct { Logger hclog.Logger CfgSrc ProxyConfigSource ResolveToken ACLResolverFunc - CfgFetcher ConfigFetcher + CfgFetcher configfetcher.ConfigFetcher // AuthCheckFrequency is how often we should re-check the credentials used // during a long-lived gRPC Stream after it has been initially established. @@ -161,7 +149,7 @@ func NewServer( logger hclog.Logger, cfgMgr ProxyConfigSource, resolveTokenSecret ACLResolverFunc, - cfgFetcher ConfigFetcher, + cfgFetcher configfetcher.ConfigFetcher, ) *Server { return &Server{ NodeName: nodeName, diff --git a/agent/xdsv2/cluster_resources.go b/agent/xdsv2/cluster_resources.go new file mode 100644 index 0000000000..c162d05c8d --- /dev/null +++ b/agent/xdsv2/cluster_resources.go @@ -0,0 +1,6 @@ +package xdsv2 + +// TODO(proxystate): In a future PR this will create clusters and add it to ProxyResources.proxyState +func (pr *ProxyResources) makeCluster(name string) error { + return nil +} diff --git a/agent/xdsv2/listener_resources.go b/agent/xdsv2/listener_resources.go new file mode 100644 index 0000000000..6d6ce7c800 --- /dev/null +++ b/agent/xdsv2/listener_resources.go @@ -0,0 +1,967 @@ +package xdsv2 + +import ( + "fmt" + "sort" + "strconv" + + envoy_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + envoy_listener_v3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" + envoy_rbac_v3 "github.com/envoyproxy/go-control-plane/envoy/config/rbac/v3" + envoy_grpc_http1_bridge_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/grpc_http1_bridge/v3" + envoy_grpc_stats_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/grpc_stats/v3" + envoy_http_router_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/router/v3" + envoy_extensions_filters_listener_http_inspector_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/listener/http_inspector/v3" + envoy_original_dst_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/listener/original_dst/v3" + envoy_tls_inspector_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/listener/tls_inspector/v3" + envoy_connection_limit_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/connection_limit/v3" + envoy_http_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3" + envoy_network_rbac_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/rbac/v3" + envoy_sni_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/sni_cluster/v3" + envoy_tcp_proxy_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/tcp_proxy/v3" + envoy_tls_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3" + envoy_matcher_v3 "github.com/envoyproxy/go-control-plane/envoy/type/matcher/v3" + envoy_type_v3 "github.com/envoyproxy/go-control-plane/envoy/type/v3" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/wrapperspb" + + "github.com/hashicorp/consul/envoyextensions/xdscommon" + "github.com/hashicorp/consul/lib" + "github.com/hashicorp/consul/proto-public/pbmesh/v1alpha1/pbproxystate" +) + +const ( + envoyNetworkFilterName = "envoy.filters.network.tcp_proxy" + envoyOriginalDestinationListenerFilterName = "envoy.filters.listener.original_dst" + envoyTLSInspectorListenerFilterName = "envoy.filters.listener.tls_inspector" + envoyHttpInspectorListenerFilterName = "envoy.filters.listener.http_inspector" + envoyHttpConnectionManagerFilterName = "envoy.filters.network.http_connection_manager" +) + +func (pr *ProxyResources) makeListener(listener *pbproxystate.Listener) (*envoy_listener_v3.Listener, error) { + envoyListener := &envoy_listener_v3.Listener{} + + // Listener Address + var address *envoy_core_v3.Address + switch listener.BindAddress.(type) { + case *pbproxystate.Listener_HostPort: + address = makeIpPortEnvoyAddress(listener.BindAddress.(*pbproxystate.Listener_HostPort)) + case *pbproxystate.Listener_UnixSocket: + address = makeUnixSocketEnvoyAddress(listener.BindAddress.(*pbproxystate.Listener_UnixSocket)) + default: + // This should be impossible to reach because we're using protobufs. + return nil, fmt.Errorf("invalid listener bind address type: %t", listener.BindAddress) + } + envoyListener.Address = address + + // Listener Direction + var direction envoy_core_v3.TrafficDirection + switch listener.Direction { + case pbproxystate.Direction_DIRECTION_OUTBOUND: + direction = envoy_core_v3.TrafficDirection_OUTBOUND + case pbproxystate.Direction_DIRECTION_INBOUND: + direction = envoy_core_v3.TrafficDirection_INBOUND + case pbproxystate.Direction_DIRECTION_UNSPECIFIED: + direction = envoy_core_v3.TrafficDirection_UNSPECIFIED + default: + return nil, fmt.Errorf("no direction for listener %+v", listener.Name) + } + envoyListener.TrafficDirection = direction + + // Before creating the filter chains, sort routers by match to avoid draining if the list is provided out of order. + sortRouters(listener.Routers) + + // Listener filter chains + for _, r := range listener.Routers { + filterChain, err := pr.makeEnvoyListenerFilterChain(r) + if err != nil { + return nil, fmt.Errorf("could not make filter chain: %w", err) + } + envoyListener.FilterChains = append(envoyListener.FilterChains, filterChain) + } + + if listener.DefaultRouter != nil { + defaultFilterChain, err := pr.makeEnvoyListenerFilterChain(listener.DefaultRouter) + if err != nil { + return nil, fmt.Errorf("could not make filter chain: %w", err) + } + envoyListener.DefaultFilterChain = defaultFilterChain + } + + // Envoy builtin listener filters + for _, c := range listener.Capabilities { + listenerFilter, err := makeEnvoyListenerFilter(c) + if err != nil { + return nil, fmt.Errorf("could not make listener filter: %w", err) + } + envoyListener.ListenerFilters = append(envoyListener.ListenerFilters, listenerFilter) + } + + err := addEnvoyListenerConnectionBalanceConfig(listener.BalanceConnections, envoyListener) + if err != nil { + return nil, err + } + + envoyListener.Name = listener.Name + envoyListener.Address = address + envoyListener.TrafficDirection = direction + + return envoyListener, nil +} + +func makeEnvoyConnectionLimitFilter(maxInboundConns uint64) (*envoy_listener_v3.Filter, error) { + cfg := &envoy_connection_limit_v3.ConnectionLimit{ + StatPrefix: "inbound_connection_limit", + MaxConnections: wrapperspb.UInt64(maxInboundConns), + } + + return makeEnvoyFilter("envoy.filters.network.connection_limit", cfg) +} + +func addEnvoyListenerConnectionBalanceConfig(balanceType pbproxystate.BalanceConnections, listener *envoy_listener_v3.Listener) error { + switch balanceType { + case pbproxystate.BalanceConnections_BALANCE_CONNECTIONS_DEFAULT: + // Default with no balancing. + return nil + case pbproxystate.BalanceConnections_BALANCE_CONNECTIONS_EXACT: + listener.ConnectionBalanceConfig = &envoy_listener_v3.Listener_ConnectionBalanceConfig{ + BalanceType: &envoy_listener_v3.Listener_ConnectionBalanceConfig_ExactBalance_{}, + } + return nil + default: + // This should be impossible using protobufs. + return fmt.Errorf("unsupported connection balance option: %+v", balanceType) + } +} + +func makeIpPortEnvoyAddress(address *pbproxystate.Listener_HostPort) *envoy_core_v3.Address { + return &envoy_core_v3.Address{ + Address: &envoy_core_v3.Address_SocketAddress{ + SocketAddress: &envoy_core_v3.SocketAddress{ + Address: address.HostPort.Host, + PortSpecifier: &envoy_core_v3.SocketAddress_PortValue{ + PortValue: address.HostPort.Port, + }, + }, + }, + } +} + +func makeUnixSocketEnvoyAddress(address *pbproxystate.Listener_UnixSocket) *envoy_core_v3.Address { + modeInt, err := strconv.ParseUint(address.UnixSocket.Mode, 0, 32) + if err != nil { + modeInt = 0 + } + return &envoy_core_v3.Address{ + Address: &envoy_core_v3.Address_Pipe{ + Pipe: &envoy_core_v3.Pipe{ + Path: address.UnixSocket.Path, + Mode: uint32(modeInt), + }, + }, + } +} + +func (pr *ProxyResources) makeEnvoyListenerFilterChain(router *pbproxystate.Router) (*envoy_listener_v3.FilterChain, error) { + envoyFilterChain := &envoy_listener_v3.FilterChain{} + + if router == nil { + return nil, fmt.Errorf("no router to create filter chain") + } + + // Router Match + match := makeEnvoyFilterChainMatch(router.Match) + if match != nil { + envoyFilterChain.FilterChainMatch = match + } + + // Router Destination + var envoyFilters []*envoy_listener_v3.Filter + switch router.Destination.(type) { + case *pbproxystate.Router_L4: + l4Filters, err := pr.makeEnvoyResourcesForL4Destination(router.Destination.(*pbproxystate.Router_L4)) + if err != nil { + return nil, err + } + envoyFilters = append(envoyFilters, l4Filters...) + case *pbproxystate.Router_L7: + l7 := router.Destination.(*pbproxystate.Router_L7) + l7Filters, err := pr.makeEnvoyResourcesForL7Destination(l7) + if err != nil { + return nil, err + } + + // Inject ALPN protocols to router's TLS if destination is L7 + if router.InboundTls != nil { + router.InboundTls.AlpnProtocols = getAlpnProtocols(l7.L7.Protocol) + } + envoyFilters = append(envoyFilters, l7Filters...) + case *pbproxystate.Router_Sni: + sniFilters, err := pr.makeEnvoyResourcesForSNIDestination(router.Destination.(*pbproxystate.Router_Sni)) + if err != nil { + return nil, err + } + envoyFilters = append(envoyFilters, sniFilters...) + default: + // This should be impossible using protobufs. + return nil, fmt.Errorf("unsupported destination type: %t", router.Destination) + + } + + // Router TLS + ts, err := pr.makeEnvoyTransportSocket(router.InboundTls) + if err != nil { + return nil, err + } + envoyFilterChain.TransportSocket = ts + + envoyFilterChain.Filters = envoyFilters + return envoyFilterChain, err +} + +func makeEnvoyFilterChainMatch(routerMatch *pbproxystate.Match) *envoy_listener_v3.FilterChainMatch { + var envoyFilterChainMatch *envoy_listener_v3.FilterChainMatch + if routerMatch != nil { + envoyFilterChainMatch = &envoy_listener_v3.FilterChainMatch{} + + envoyFilterChainMatch.DestinationPort = routerMatch.DestinationPort + + if len(routerMatch.ServerNames) > 0 { + var serverNames []string + for _, n := range routerMatch.ServerNames { + serverNames = append(serverNames, n) + } + envoyFilterChainMatch.ServerNames = serverNames + } + if len(routerMatch.PrefixRanges) > 0 { + sortPrefixRanges(routerMatch.PrefixRanges) + var ranges []*envoy_core_v3.CidrRange + for _, r := range routerMatch.PrefixRanges { + cidrRange := &envoy_core_v3.CidrRange{ + PrefixLen: r.PrefixLen, + AddressPrefix: r.AddressPrefix, + } + ranges = append(ranges, cidrRange) + } + envoyFilterChainMatch.PrefixRanges = ranges + } + if len(routerMatch.SourcePrefixRanges) > 0 { + var ranges []*envoy_core_v3.CidrRange + for _, r := range routerMatch.SourcePrefixRanges { + cidrRange := &envoy_core_v3.CidrRange{ + PrefixLen: r.PrefixLen, + AddressPrefix: r.AddressPrefix, + } + ranges = append(ranges, cidrRange) + } + envoyFilterChainMatch.SourcePrefixRanges = ranges + } + } + return envoyFilterChainMatch +} + +func (pr *ProxyResources) makeEnvoyResourcesForSNIDestination(sni *pbproxystate.Router_Sni) ([]*envoy_listener_v3.Filter, error) { + var envoyFilters []*envoy_listener_v3.Filter + sniFilter, err := makeEnvoyFilter("envoy.filters.network.sni_cluster", &envoy_sni_cluster_v3.SniCluster{}) + if err != nil { + return nil, err + } + tcp := &envoy_tcp_proxy_v3.TcpProxy{ + StatPrefix: sni.Sni.StatPrefix, + ClusterSpecifier: &envoy_tcp_proxy_v3.TcpProxy_Cluster{Cluster: ""}, + } + tcpFilter, err := makeEnvoyFilter(envoyNetworkFilterName, tcp) + if err != nil { + return nil, err + } + envoyFilters = append(envoyFilters, sniFilter, tcpFilter) + return envoyFilters, err +} + +func (pr *ProxyResources) makeEnvoyResourcesForL4Destination(l4 *pbproxystate.Router_L4) ([]*envoy_listener_v3.Filter, error) { + err := pr.makeCluster(l4.L4.Name) + if err != nil { + return nil, err + } + envoyFilters, err := makeL4Filters(l4.L4) + return envoyFilters, err +} + +func (pr *ProxyResources) makeEnvoyResourcesForL7Destination(l7 *pbproxystate.Router_L7) ([]*envoy_listener_v3.Filter, error) { + envoyFilters, err := pr.makeL7Filters(l7.L7) + if err != nil { + return nil, err + } + return envoyFilters, err +} + +func getAlpnProtocols(protocol pbproxystate.L7Protocol) []string { + var alpnProtocols []string + + switch protocol { + case pbproxystate.L7Protocol_L7_PROTOCOL_GRPC, pbproxystate.L7Protocol_L7_PROTOCOL_HTTP2: + alpnProtocols = append(alpnProtocols, "h2", "http/1.1") + case pbproxystate.L7Protocol_L7_PROTOCOL_HTTP: + alpnProtocols = append(alpnProtocols, "http/1.1") + } + + return alpnProtocols +} + +func makeL4Filters(l4 *pbproxystate.L4Destination) ([]*envoy_listener_v3.Filter, error) { + var envoyFilters []*envoy_listener_v3.Filter + if l4 != nil { + // Add rbac filter. RBAC filter needs to be added first so any + // unauthorized connections will get rejected. + // TODO(proxystate): Intentions will be added in the future. + if l4.AddEmptyIntention { + rbacFilter, err := makeEmptyRBACNetworkFilter() + if err != nil { + return nil, err + } + envoyFilters = append(envoyFilters, rbacFilter) + } + + if l4.MaxInboundConnections > 0 { + connectionLimitFilter, err := makeEnvoyConnectionLimitFilter(l4.MaxInboundConnections) + if err != nil { + return nil, err + } + envoyFilters = append(envoyFilters, connectionLimitFilter) + } + + // Add tcp proxy filter + tcp := &envoy_tcp_proxy_v3.TcpProxy{ + ClusterSpecifier: &envoy_tcp_proxy_v3.TcpProxy_Cluster{Cluster: l4.Name}, + StatPrefix: l4.StatPrefix, + } + tcpFilter, err := makeEnvoyFilter(envoyNetworkFilterName, tcp) + if err != nil { + return nil, err + } + envoyFilters = append(envoyFilters, tcpFilter) + } + return envoyFilters, nil + +} + +func makeEmptyRBACNetworkFilter() (*envoy_listener_v3.Filter, error) { + cfg := &envoy_network_rbac_v3.RBAC{ + StatPrefix: "connect_authz", + Rules: &envoy_rbac_v3.RBAC{}, + } + filter, err := makeEnvoyFilter("envoy.filters.network.rbac", cfg) + if err != nil { + return nil, err + } + return filter, nil +} + +// TODO: Forward client cert details will be added as part of L7 listeners task. +func (pr *ProxyResources) makeL7Filters(l7 *pbproxystate.L7Destination) ([]*envoy_listener_v3.Filter, error) { + var envoyFilters []*envoy_listener_v3.Filter + var httpConnMgr *envoy_http_v3.HttpConnectionManager + + if l7 != nil { + // TODO: Intentions will be added in the future. + if l7.MaxInboundConnections > 0 { + connLimitFilter, err := makeEnvoyConnectionLimitFilter(l7.MaxInboundConnections) + if err != nil { + return nil, err + } + envoyFilters = append(envoyFilters, connLimitFilter) + } + envoyHttpRouter, err := makeEnvoyHTTPFilter("envoy.filters.http.router", &envoy_http_router_v3.Router{}) + if err != nil { + return nil, err + } + + httpConnMgr = &envoy_http_v3.HttpConnectionManager{ + StatPrefix: l7.StatPrefix, + CodecType: envoy_http_v3.HttpConnectionManager_AUTO, + HttpFilters: []*envoy_http_v3.HttpFilter{ + envoyHttpRouter, + }, + Tracing: &envoy_http_v3.HttpConnectionManager_Tracing{ + // Don't trace any requests by default unless the client application + // explicitly propagates trace headers that indicate this should be + // sampled. + RandomSampling: &envoy_type_v3.Percent{Value: 0.0}, + }, + // Explicitly enable WebSocket upgrades for all HTTP listeners + UpgradeConfigs: []*envoy_http_v3.HttpConnectionManager_UpgradeConfig{ + {UpgradeType: "websocket"}, + }, + } + + routeConfig, err := pr.makeRoute(l7.Name) + if err != nil { + return nil, err + } + + if l7.StaticRoute { + httpConnMgr.RouteSpecifier = &envoy_http_v3.HttpConnectionManager_RouteConfig{ + RouteConfig: routeConfig, + } + } else { + // Add Envoy route under the route resource since it's not inlined. + pr.envoyResources[xdscommon.RouteType] = append(pr.envoyResources[xdscommon.RouteType], routeConfig) + + httpConnMgr.RouteSpecifier = &envoy_http_v3.HttpConnectionManager_Rds{ + Rds: &envoy_http_v3.Rds{ + RouteConfigName: l7.Name, + ConfigSource: &envoy_core_v3.ConfigSource{ + ResourceApiVersion: envoy_core_v3.ApiVersion_V3, + ConfigSourceSpecifier: &envoy_core_v3.ConfigSource_Ads{ + Ads: &envoy_core_v3.AggregatedConfigSource{}, + }, + }, + }, + } + } + + // Add http2 protocol options + if l7.Protocol == pbproxystate.L7Protocol_L7_PROTOCOL_HTTP2 || l7.Protocol == pbproxystate.L7Protocol_L7_PROTOCOL_GRPC { + httpConnMgr.Http2ProtocolOptions = &envoy_core_v3.Http2ProtocolOptions{} + } + + // Add grpc envoy http filters. + if l7.Protocol == pbproxystate.L7Protocol_L7_PROTOCOL_GRPC { + grpcHttp1Bridge, err := makeEnvoyHTTPFilter( + "envoy.filters.http.grpc_http1_bridge", + &envoy_grpc_http1_bridge_v3.Config{}, + ) + if err != nil { + return nil, err + } + + // In envoy 1.14.x the default value "stats_for_all_methods=true" was + // deprecated, and was changed to "false" in 1.18.x. Avoid using the + // default. TODO: we may want to expose this to users somehow easily. + grpcStatsFilter, err := makeEnvoyHTTPFilter( + "envoy.filters.http.grpc_stats", + &envoy_grpc_stats_v3.FilterConfig{ + PerMethodStatSpecifier: &envoy_grpc_stats_v3.FilterConfig_StatsForAllMethods{ + StatsForAllMethods: &wrapperspb.BoolValue{Value: true}, + }, + }, + ) + if err != nil { + return nil, err + } + + // Add grpc bridge before envoyRouter and authz, and the stats in front of that. + httpConnMgr.HttpFilters = append([]*envoy_http_v3.HttpFilter{ + grpcStatsFilter, + grpcHttp1Bridge, + }, httpConnMgr.HttpFilters...) + } + + httpFilter, err := makeEnvoyFilter(envoyHttpConnectionManagerFilterName, httpConnMgr) + if err != nil { + return nil, err + } + envoyFilters = append(envoyFilters, httpFilter) + } + return envoyFilters, nil +} + +func (pr *ProxyResources) makeEnvoyTLSParameters(defaultParams *pbproxystate.TLSParameters, overrideParams *pbproxystate.TLSParameters) *envoy_tls_v3.TlsParameters { + tlsParams := &envoy_tls_v3.TlsParameters{} + + if overrideParams != nil { + if overrideParams.MinVersion != pbproxystate.TLSVersion_TLS_VERSION_UNSPECIFIED { + if minVersion, ok := envoyTLSVersions[overrideParams.MinVersion]; ok { + tlsParams.TlsMinimumProtocolVersion = minVersion + } + } + if overrideParams.MaxVersion != pbproxystate.TLSVersion_TLS_VERSION_UNSPECIFIED { + if maxVersion, ok := envoyTLSVersions[overrideParams.MaxVersion]; ok { + tlsParams.TlsMaximumProtocolVersion = maxVersion + } + } + if len(overrideParams.CipherSuites) != 0 { + tlsParams.CipherSuites = marshalEnvoyTLSCipherSuiteStrings(overrideParams.CipherSuites) + } + return tlsParams + } + + if defaultParams != nil { + if defaultParams.MinVersion != pbproxystate.TLSVersion_TLS_VERSION_UNSPECIFIED { + if minVersion, ok := envoyTLSVersions[defaultParams.MinVersion]; ok { + tlsParams.TlsMinimumProtocolVersion = minVersion + } + } + if defaultParams.MaxVersion != pbproxystate.TLSVersion_TLS_VERSION_UNSPECIFIED { + if maxVersion, ok := envoyTLSVersions[defaultParams.MaxVersion]; ok { + tlsParams.TlsMaximumProtocolVersion = maxVersion + } + } + if len(defaultParams.CipherSuites) != 0 { + tlsParams.CipherSuites = marshalEnvoyTLSCipherSuiteStrings(defaultParams.CipherSuites) + } + return tlsParams + } + + return tlsParams + +} + +func (pr *ProxyResources) makeEnvoyTransportSocket(ts *pbproxystate.TransportSocket) (*envoy_core_v3.TransportSocket, error) { + if ts == nil { + return nil, nil + } + commonTLSContext := &envoy_tls_v3.CommonTlsContext{} + + // Create connection TLS. Listeners should only look at inbound TLS. + switch ts.ConnectionTls.(type) { + case *pbproxystate.TransportSocket_InboundMesh: + downstreamContext := &envoy_tls_v3.DownstreamTlsContext{} + downstreamContext.CommonTlsContext = commonTLSContext + // Set TLS Parameters. + tlsParams := pr.makeEnvoyTLSParameters(pr.proxyState.Tls.InboundTlsParameters, ts.TlsParameters) + commonTLSContext.TlsParams = tlsParams + + // Set the certificate config on the tls context. + // For inbound mesh, we need to add the identity certificate + // and the validation context for the mesh depending on the provided trust bundle names. + if pr.proxyState.Tls == nil { + // if tls is nil but connection tls is provided, then the proxy state is misconfigured + return nil, fmt.Errorf("proxyState.Tls is required to generate router's transport socket") + } + im := ts.ConnectionTls.(*pbproxystate.TransportSocket_InboundMesh).InboundMesh + leaf, ok := pr.proxyState.LeafCertificates[im.IdentityKey] + if !ok { + return nil, fmt.Errorf("failed to create transport socket: leaf certificate %q not found", im.IdentityKey) + } + err := pr.makeEnvoyCertConfig(commonTLSContext, leaf) + if err != nil { + return nil, fmt.Errorf("failed to create transport socket: %w", err) + } + + // Create validation context. + // When there's only one trust bundle name, we create a simple validation context + if len(im.ValidationContext.TrustBundlePeerNameKeys) == 1 { + peerName := im.ValidationContext.TrustBundlePeerNameKeys[0] + tb, ok := pr.proxyState.TrustBundles[peerName] + if !ok { + return nil, fmt.Errorf("failed to create transport socket: provided trust bundle name does not exist in proxystate trust bundle map: %s", peerName) + } + commonTLSContext.ValidationContextType = &envoy_tls_v3.CommonTlsContext_ValidationContext{ + ValidationContext: &envoy_tls_v3.CertificateValidationContext{ + // TODO(banks): later for L7 support we may need to configure ALPN here. + TrustedCa: &envoy_core_v3.DataSource{ + Specifier: &envoy_core_v3.DataSource_InlineString{ + InlineString: RootPEMsAsString(tb.Roots), + }, + }, + }, + } + } else if len(im.ValidationContext.TrustBundlePeerNameKeys) > 1 { + cfg := &envoy_tls_v3.SPIFFECertValidatorConfig{ + TrustDomains: make([]*envoy_tls_v3.SPIFFECertValidatorConfig_TrustDomain, 0, len(im.ValidationContext.TrustBundlePeerNameKeys)), + } + + for _, peerName := range im.ValidationContext.TrustBundlePeerNameKeys { + // Look up the trust bundle ca in the map. + tb, ok := pr.proxyState.TrustBundles[peerName] + if !ok { + return nil, fmt.Errorf("failed to create transport socket: provided bundle name does not exist in trust bundle map: %s", peerName) + } + cfg.TrustDomains = append(cfg.TrustDomains, &envoy_tls_v3.SPIFFECertValidatorConfig_TrustDomain{ + Name: tb.TrustDomain, + TrustBundle: &envoy_core_v3.DataSource{ + Specifier: &envoy_core_v3.DataSource_InlineString{ + InlineString: RootPEMsAsString(tb.Roots), + }, + }, + }) + } + // Sort the trust domains so the output is stable. + sortTrustDomains(cfg.TrustDomains) + + spiffeConfig, err := anypb.New(cfg) + if err != nil { + return nil, err + } + commonTLSContext.ValidationContextType = &envoy_tls_v3.CommonTlsContext_ValidationContext{ + ValidationContext: &envoy_tls_v3.CertificateValidationContext{ + CustomValidatorConfig: &envoy_core_v3.TypedExtensionConfig{ + // The typed config name is hard-coded because it is not available as a wellknown var in the control plane lib. + Name: "envoy.tls.cert_validator.spiffe", + TypedConfig: spiffeConfig, + }, + }, + } + } + // Always require client certificate + downstreamContext.RequireClientCertificate = &wrapperspb.BoolValue{Value: true} + transportSocket, err := makeTransportSocket("tls", downstreamContext) + if err != nil { + return nil, err + } + return transportSocket, nil + case *pbproxystate.TransportSocket_InboundNonMesh: + downstreamContext := &envoy_tls_v3.DownstreamTlsContext{} + downstreamContext.CommonTlsContext = commonTLSContext + // Set TLS Parameters + tlsParams := pr.makeEnvoyTLSParameters(pr.proxyState.Tls.InboundTlsParameters, ts.TlsParameters) + commonTLSContext.TlsParams = tlsParams + // For non-mesh, we don't care about validation context as currently we don't support mTLS for non-mesh connections. + nonMeshTLS := ts.ConnectionTls.(*pbproxystate.TransportSocket_InboundNonMesh).InboundNonMesh + err := pr.addNonMeshCertConfig(commonTLSContext, nonMeshTLS) + if err != nil { + return nil, fmt.Errorf("failed to create transport socket: %w", err) + } + transportSocket, err := makeTransportSocket("tls", downstreamContext) + if err != nil { + return nil, err + } + return transportSocket, nil + case *pbproxystate.TransportSocket_OutboundMesh: + upstreamContext := &envoy_tls_v3.UpstreamTlsContext{} + upstreamContext.CommonTlsContext = commonTLSContext + // Set TLS Parameters + tlsParams := pr.makeEnvoyTLSParameters(pr.proxyState.Tls.OutboundTlsParameters, ts.TlsParameters) + commonTLSContext.TlsParams = tlsParams + // For outbound mesh, we need to insert the mesh identity certificate + // and the validation context for the mesh depending on the provided trust bundle names. + if pr.proxyState.Tls == nil { + // if tls is nil but connection tls is provided, then the proxy state is misconfigured + return nil, fmt.Errorf("proxyState.Tls is required to generate router's transport socket") + } + om := ts.ConnectionTls.(*pbproxystate.TransportSocket_OutboundMesh).OutboundMesh + leaf, ok := pr.proxyState.LeafCertificates[om.IdentityKey] + if !ok { + return nil, fmt.Errorf("leaf %s not found in proxyState", om.IdentityKey) + } + err := pr.makeEnvoyCertConfig(commonTLSContext, leaf) + if err != nil { + return nil, fmt.Errorf("failed to create transport socket: %w", err) + } + + // Create validation context + peerName := om.ValidationContext.TrustBundlePeerNameKey + tb, ok := pr.proxyState.TrustBundles[peerName] + if !ok { + return nil, fmt.Errorf("failed to create transport socket: provided peer name does not exist in trust bundle map: %s", peerName) + } + + var matchers []*envoy_matcher_v3.StringMatcher + if len(om.ValidationContext.SpiffeIds) > 0 { + matchers = make([]*envoy_matcher_v3.StringMatcher, 0) + for _, m := range om.ValidationContext.SpiffeIds { + matchers = append(matchers, &envoy_matcher_v3.StringMatcher{ + MatchPattern: &envoy_matcher_v3.StringMatcher_Exact{ + Exact: m, + }, + }) + } + } + commonTLSContext.ValidationContextType = &envoy_tls_v3.CommonTlsContext_ValidationContext{ + ValidationContext: &envoy_tls_v3.CertificateValidationContext{ + // TODO(banks): later for L7 support we may need to configure ALPN here. + TrustedCa: &envoy_core_v3.DataSource{ + Specifier: &envoy_core_v3.DataSource_InlineString{ + InlineString: RootPEMsAsString(tb.Roots), + }, + }, + MatchSubjectAltNames: matchers, + }, + } + + upstreamContext.Sni = om.Sni + transportSocket, err := makeTransportSocket("tls", upstreamContext) + if err != nil { + return nil, err + } + return transportSocket, nil + default: + return nil, nil + } + +} + +func (pr *ProxyResources) makeEnvoyCertConfig(common *envoy_tls_v3.CommonTlsContext, certificate *pbproxystate.LeafCertificate) error { + if certificate == nil { + return fmt.Errorf("no leaf certificate provided") + } + common.TlsCertificates = []*envoy_tls_v3.TlsCertificate{ + { + CertificateChain: &envoy_core_v3.DataSource{ + Specifier: &envoy_core_v3.DataSource_InlineString{ + InlineString: lib.EnsureTrailingNewline(certificate.Cert), + }, + }, + PrivateKey: &envoy_core_v3.DataSource{ + Specifier: &envoy_core_v3.DataSource_InlineString{ + InlineString: lib.EnsureTrailingNewline(certificate.Key), + }, + }, + }, + } + return nil +} + +func (pr *ProxyResources) makeEnvoySDSCertConfig(common *envoy_tls_v3.CommonTlsContext, certificate *pbproxystate.SDSCertificate) error { + if certificate == nil { + return fmt.Errorf("no SDS certificate provided") + } + common.TlsCertificateSdsSecretConfigs = []*envoy_tls_v3.SdsSecretConfig{ + { + Name: certificate.CertResource, + SdsConfig: &envoy_core_v3.ConfigSource{ + ConfigSourceSpecifier: &envoy_core_v3.ConfigSource_ApiConfigSource{ + ApiConfigSource: &envoy_core_v3.ApiConfigSource{ + ApiType: envoy_core_v3.ApiConfigSource_GRPC, + TransportApiVersion: envoy_core_v3.ApiVersion_V3, + // Note ClusterNames can't be set here - that's only for REST type + // we need a full GRPC config instead. + GrpcServices: []*envoy_core_v3.GrpcService{ + { + TargetSpecifier: &envoy_core_v3.GrpcService_EnvoyGrpc_{ + EnvoyGrpc: &envoy_core_v3.GrpcService_EnvoyGrpc{ + ClusterName: certificate.ClusterName, + }, + }, + Timeout: &durationpb.Duration{Seconds: 5}, + }, + }, + }, + }, + ResourceApiVersion: envoy_core_v3.ApiVersion_V3, + }, + }, + } + return nil +} + +func (pr *ProxyResources) addNonMeshCertConfig(common *envoy_tls_v3.CommonTlsContext, tls *pbproxystate.InboundNonMeshTLS) error { + if tls == nil { + return fmt.Errorf("no inbound non-mesh TLS provided") + } + + switch tls.Identity.(type) { + case *pbproxystate.InboundNonMeshTLS_LeafKey: + leafKey := tls.Identity.(*pbproxystate.InboundNonMeshTLS_LeafKey).LeafKey + leaf, ok := pr.proxyState.LeafCertificates[leafKey] + if !ok { + return fmt.Errorf("leaf key %s not found in leaf certificate map", leafKey) + } + common.TlsCertificates = []*envoy_tls_v3.TlsCertificate{ + { + CertificateChain: &envoy_core_v3.DataSource{ + Specifier: &envoy_core_v3.DataSource_InlineString{ + InlineString: lib.EnsureTrailingNewline(leaf.Cert), + }, + }, + PrivateKey: &envoy_core_v3.DataSource{ + Specifier: &envoy_core_v3.DataSource_InlineString{ + InlineString: lib.EnsureTrailingNewline(leaf.Key), + }, + }, + }, + } + case *pbproxystate.InboundNonMeshTLS_Sds: + c := tls.Identity.(*pbproxystate.InboundNonMeshTLS_Sds).Sds + common.TlsCertificateSdsSecretConfigs = []*envoy_tls_v3.SdsSecretConfig{ + { + Name: c.CertResource, + SdsConfig: &envoy_core_v3.ConfigSource{ + ConfigSourceSpecifier: &envoy_core_v3.ConfigSource_ApiConfigSource{ + ApiConfigSource: &envoy_core_v3.ApiConfigSource{ + ApiType: envoy_core_v3.ApiConfigSource_GRPC, + TransportApiVersion: envoy_core_v3.ApiVersion_V3, + // Note ClusterNames can't be set here - that's only for REST type + // we need a full GRPC config instead. + GrpcServices: []*envoy_core_v3.GrpcService{ + { + TargetSpecifier: &envoy_core_v3.GrpcService_EnvoyGrpc_{ + EnvoyGrpc: &envoy_core_v3.GrpcService_EnvoyGrpc{ + ClusterName: c.ClusterName, + }, + }, + Timeout: &durationpb.Duration{Seconds: 5}, + }, + }, + }, + }, + ResourceApiVersion: envoy_core_v3.ApiVersion_V3, + }, + }, + } + } + + return nil +} + +func makeTransportSocket(name string, config proto.Message) (*envoy_core_v3.TransportSocket, error) { + any, err := anypb.New(config) + if err != nil { + return nil, err + } + return &envoy_core_v3.TransportSocket{ + Name: name, + ConfigType: &envoy_core_v3.TransportSocket_TypedConfig{ + TypedConfig: any, + }, + }, nil +} + +func makeEnvoyListenerFilter(c pbproxystate.Capability) (*envoy_listener_v3.ListenerFilter, error) { + var lf proto.Message + var name string + + switch c { + case pbproxystate.Capability_CAPABILITY_TRANSPARENT: + lf = &envoy_original_dst_v3.OriginalDst{} + name = envoyOriginalDestinationListenerFilterName + case pbproxystate.Capability_CAPABILITY_L4_TLS_INSPECTION: + name = envoyTLSInspectorListenerFilterName + lf = &envoy_tls_inspector_v3.TlsInspector{} + case pbproxystate.Capability_CAPABILITY_L7_PROTOCOL_INSPECTION: + name = envoyHttpInspectorListenerFilterName + lf = &envoy_extensions_filters_listener_http_inspector_v3.HttpInspector{} + default: + return nil, fmt.Errorf("unsupported listener captability: %s", c) + } + lfAsAny, err := anypb.New(lf) + if err != nil { + return nil, err + } + + return &envoy_listener_v3.ListenerFilter{ + Name: name, + ConfigType: &envoy_listener_v3.ListenerFilter_TypedConfig{TypedConfig: lfAsAny}, + }, nil +} + +func makeEnvoyFilter(name string, cfg proto.Message) (*envoy_listener_v3.Filter, error) { + any, err := anypb.New(cfg) + if err != nil { + return nil, err + } + + return &envoy_listener_v3.Filter{ + Name: name, + ConfigType: &envoy_listener_v3.Filter_TypedConfig{TypedConfig: any}, + }, nil +} + +func makeEnvoyHTTPFilter(name string, cfg proto.Message) (*envoy_http_v3.HttpFilter, error) { + any, err := anypb.New(cfg) + if err != nil { + return nil, err + } + + return &envoy_http_v3.HttpFilter{ + Name: name, + ConfigType: &envoy_http_v3.HttpFilter_TypedConfig{TypedConfig: any}, + }, nil +} + +func RootPEMsAsString(rootPEMs []string) string { + var rootPEMsString string + for _, root := range rootPEMs { + rootPEMsString += lib.EnsureTrailingNewline(root) + } + return rootPEMsString +} + +func marshalEnvoyTLSCipherSuiteStrings(cipherSuites []pbproxystate.TLSCipherSuite) []string { + envoyTLSCipherSuiteStrings := map[pbproxystate.TLSCipherSuite]string{ + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_AES128_GCM_SHA256: "ECDHE-ECDSA-AES128-GCM-SHA256", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_CHACHA20_POLY1305: "ECDHE-ECDSA-CHACHA20-POLY1305", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_AES128_GCM_SHA256: "ECDHE-RSA-AES128-GCM-SHA256", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_CHACHA20_POLY1305: "ECDHE-RSA-CHACHA20-POLY1305", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_AES128_SHA: "ECDHE-ECDSA-AES128-SHA", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_AES128_SHA: "ECDHE-RSA-AES128-SHA", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_AES128_GCM_SHA256: "AES128-GCM-SHA256", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_AES128_SHA: "AES128-SHA", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_AES256_GCM_SHA384: "ECDHE-ECDSA-AES256-GCM-SHA384", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_AES256_GCM_SHA384: "ECDHE-RSA-AES256-GCM-SHA384", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_ECDSA_AES256_SHA: "ECDHE-ECDSA-AES256-SHA", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_ECDHE_RSA_AES256_SHA: "ECDHE-RSA-AES256-SHA", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_AES256_GCM_SHA384: "AES256-GCM-SHA384", + pbproxystate.TLSCipherSuite_TLS_CIPHER_SUITE_AES256_SHA: "AES256-SHA", + } + + var cipherSuiteStrings []string + + for _, c := range cipherSuites { + if s, ok := envoyTLSCipherSuiteStrings[c]; ok { + cipherSuiteStrings = append(cipherSuiteStrings, s) + } + } + + return cipherSuiteStrings +} + +var envoyTLSVersions = map[pbproxystate.TLSVersion]envoy_tls_v3.TlsParameters_TlsProtocol{ + pbproxystate.TLSVersion_TLS_VERSION_AUTO: envoy_tls_v3.TlsParameters_TLS_AUTO, + pbproxystate.TLSVersion_TLS_VERSION_1_0: envoy_tls_v3.TlsParameters_TLSv1_0, + pbproxystate.TLSVersion_TLS_VERSION_1_1: envoy_tls_v3.TlsParameters_TLSv1_1, + pbproxystate.TLSVersion_TLS_VERSION_1_2: envoy_tls_v3.TlsParameters_TLSv1_2, + pbproxystate.TLSVersion_TLS_VERSION_1_3: envoy_tls_v3.TlsParameters_TLSv1_3, +} + +// Sort the trust domains so that the output is stable. +// This benefits tests but also prevents Envoy from mistakenly thinking the listener +// changed and needs to be drained only because this ordering is different. +func sortTrustDomains(trustDomains []*envoy_tls_v3.SPIFFECertValidatorConfig_TrustDomain) { + sort.Slice(trustDomains, func(i int, j int) bool { + return trustDomains[i].Name < trustDomains[j].Name + }) +} + +// sortRouters stable sorts routers with a Match to avoid draining if the list is provided out of order. +// xdsv1 used to sort the filter chains on outbound listeners, so this adds that functionality by sorting routers with matches. +func sortRouters(routers []*pbproxystate.Router) { + if routers == nil { + return + } + sort.SliceStable(routers, func(i, j int) bool { + si := "" + sj := "" + if routers[i].Match != nil { + if len(routers[i].Match.PrefixRanges) > 0 { + si += routers[i].Match.PrefixRanges[0].AddressPrefix + + "/" + routers[i].Match.PrefixRanges[0].PrefixLen.String() + + ":" + routers[i].Match.DestinationPort.String() + } + if len(routers[i].Match.ServerNames) > 0 { + si += routers[i].Match.ServerNames[0] + + ":" + routers[i].Match.DestinationPort.String() + } else { + si += routers[i].Match.DestinationPort.String() + } + } + + if routers[j].Match != nil { + if len(routers[j].Match.PrefixRanges) > 0 { + sj += routers[j].Match.PrefixRanges[0].AddressPrefix + + "/" + routers[j].Match.PrefixRanges[0].PrefixLen.String() + + ":" + routers[j].Match.DestinationPort.String() + } + if len(routers[j].Match.ServerNames) > 0 { + sj += routers[j].Match.ServerNames[0] + + ":" + routers[j].Match.DestinationPort.String() + } else { + sj += routers[j].Match.DestinationPort.String() + } + } + + return si < sj + }) +} + +func sortPrefixRanges(prefixRanges []*pbproxystate.CidrRange) { + if prefixRanges == nil { + return + } + sort.SliceStable(prefixRanges, func(i, j int) bool { + return prefixRanges[i].AddressPrefix < prefixRanges[j].AddressPrefix + }) +} diff --git a/agent/xdsv2/resources.go b/agent/xdsv2/resources.go new file mode 100644 index 0000000000..6bc1df58c6 --- /dev/null +++ b/agent/xdsv2/resources.go @@ -0,0 +1,69 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package xdsv2 + +import ( + "fmt" + + pbmesh "github.com/hashicorp/consul/proto-public/pbmesh/v1alpha1" + "github.com/hashicorp/go-hclog" + "google.golang.org/protobuf/proto" + + "github.com/hashicorp/consul/envoyextensions/xdscommon" +) + +// ResourceGenerator is associated with a single gRPC stream and creates xDS +// resources for a single client. +type ResourceGenerator struct { + Logger hclog.Logger + ProxyFeatures xdscommon.SupportedProxyFeatures +} + +func NewResourceGenerator( + logger hclog.Logger, +) *ResourceGenerator { + return &ResourceGenerator{ + Logger: logger, + } +} + +type ProxyResources struct { + proxyState *pbmesh.ProxyState + envoyResources map[string][]proto.Message +} + +func (g *ResourceGenerator) AllResourcesFromIR(proxyState *pbmesh.ProxyState) (map[string][]proto.Message, error) { + pr := &ProxyResources{ + proxyState: proxyState, + envoyResources: make(map[string][]proto.Message), + } + err := pr.generateXDSResources() + if err != nil { + return nil, fmt.Errorf("failed to generate xDS resources for ProxyState: %v", err) + } + return pr.envoyResources, nil +} + +func (pr *ProxyResources) generateXDSResources() error { + listeners := make([]proto.Message, 0) + clusters := make([]proto.Message, 0) + routes := make([]proto.Message, 0) + endpoints := make([]proto.Message, 0) + + for _, l := range pr.proxyState.Listeners { + protoListener, err := pr.makeListener(l) + // TODO: aggregate errors for listeners and still return any properly formed listeners. + if err != nil { + return err + } + listeners = append(listeners, protoListener) + } + + pr.envoyResources[xdscommon.ListenerType] = listeners + pr.envoyResources[xdscommon.ClusterType] = clusters + pr.envoyResources[xdscommon.RouteType] = routes + pr.envoyResources[xdscommon.EndpointType] = endpoints + + return nil +} diff --git a/agent/xdsv2/route_resources.go b/agent/xdsv2/route_resources.go new file mode 100644 index 0000000000..1433534e8f --- /dev/null +++ b/agent/xdsv2/route_resources.go @@ -0,0 +1,20 @@ +package xdsv2 + +import ( + "fmt" + + envoy_route_v3 "github.com/envoyproxy/go-control-plane/envoy/config/route/v3" +) + +func (pr *ProxyResources) makeRoute(name string) (*envoy_route_v3.RouteConfiguration, error) { + var route *envoy_route_v3.RouteConfiguration + // TODO(proxystate): This will make routes in the future. This function should distinguish between static routes + // inlined into listeners and non-static routes that should be added as top level Envoy resources. + _, ok := pr.proxyState.Routes[name] + if !ok { + // This should not happen with a valid proxy state. + return nil, fmt.Errorf("could not find route in ProxyState: %s", name) + + } + return route, nil +} diff --git a/proto-public/pbmesh/v1alpha1/proxy_state.pb.go b/proto-public/pbmesh/v1alpha1/proxy_state.pb.go index 4f8fc7914a..193050474c 100644 --- a/proto-public/pbmesh/v1alpha1/proxy_state.pb.go +++ b/proto-public/pbmesh/v1alpha1/proxy_state.pb.go @@ -105,8 +105,7 @@ type ProxyState struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - // id is this proxy's identity. This should correspond to the workload identity that this proxy of - // the workload this proxy represents. + // identity is a reference to the WorkloadIdentity associated with this proxy. Identity *pbresource.Reference `protobuf:"bytes,1,opt,name=identity,proto3" json:"identity,omitempty"` // listeners is a list of listeners for this proxy. Listeners []*pbproxystate.Listener `protobuf:"bytes,2,rep,name=listeners,proto3" json:"listeners,omitempty"` diff --git a/proto-public/pbmesh/v1alpha1/proxy_state.proto b/proto-public/pbmesh/v1alpha1/proxy_state.proto index 94280a261f..756d7dd307 100644 --- a/proto-public/pbmesh/v1alpha1/proxy_state.proto +++ b/proto-public/pbmesh/v1alpha1/proxy_state.proto @@ -30,8 +30,7 @@ message ProxyStateTemplate { } message ProxyState { - // id is this proxy's identity. This should correspond to the workload identity that this proxy of - // the workload this proxy represents. + // identity is a reference to the WorkloadIdentity associated with this proxy. hashicorp.consul.resource.Reference identity = 1; // listeners is a list of listeners for this proxy. repeated pbproxystate.Listener listeners = 2; diff --git a/test/integration/connect/envoy/helpers.bash b/test/integration/connect/envoy/helpers.bash index cb10a345a3..01e59b6c28 100755 --- a/test/integration/connect/envoy/helpers.bash +++ b/test/integration/connect/envoy/helpers.bash @@ -210,7 +210,7 @@ function assert_envoy_expose_checks_listener_count { RANGES=$(echo "$BODY" | jq '.active_state.listener.filter_chains[0].filter_chain_match.source_prefix_ranges | length') echo "RANGES = $RANGES (expect 3)" # note: if IPv6 is not supported in the kernel per - # agent/xds:kernelSupportsIPv6() then this will only be 2 + # agent/xds/platform:SupportsIPv6() then this will only be 2 [ "${RANGES:-0}" -eq 3 ] HCM=$(echo "$BODY" | jq '.active_state.listener.filter_chains[0].filters[0]') diff --git a/test/integration/connect/envoy/helpers.windows.bash b/test/integration/connect/envoy/helpers.windows.bash index f455244205..5b6969ca85 100644 --- a/test/integration/connect/envoy/helpers.windows.bash +++ b/test/integration/connect/envoy/helpers.windows.bash @@ -255,7 +255,7 @@ function assert_envoy_expose_checks_listener_count { RANGES=$(echo "$BODY" | jq '.active_state.listener.filter_chains[0].filter_chain_match.source_prefix_ranges | length') echo "RANGES = $RANGES (expect 3)" # note: if IPv6 is not supported in the kernel per - # agent/xds:kernelSupportsIPv6() then this will only be 2 + # agent/xds/platform:SupportsIPv6() then this will only be 2 [ "${RANGES:-0}" -eq 3 ] HCM=$(echo "$BODY" | jq '.active_state.listener.filter_chains[0].filters[0]')