diff --git a/pkg/consuldp/config.go b/pkg/consuldp/config.go index a3a9fc77..68067c31 100644 --- a/pkg/consuldp/config.go +++ b/pkg/consuldp/config.go @@ -320,6 +320,13 @@ type EnvoyConfig struct { ExtraArgs []string } +const ( + // VirtualDNSInlinePort is the port for Envoy's inline DNS table listener. + VirtualDNSInlinePort = 8653 + // VirtualDNSEgressPort is the port for Envoy's external DNS forwarder listener. + VirtualDNSEgressPort = 8654 +) + // XDSServer contains the configuration of the xDS server. type XDSServer struct { // BindAddress is the address on which the Envoy xDS server will be available. diff --git a/pkg/consuldp/consul_dataplane.go b/pkg/consuldp/consul_dataplane.go index 48d5c758..695bdd76 100644 --- a/pkg/consuldp/consul_dataplane.go +++ b/pkg/consuldp/consul_dataplane.go @@ -47,6 +47,11 @@ type ConsulDataplane struct { aclToken string metricsConfig *metricsConfig lifecycleConfig *lifecycleConfig + // upstreamIndex maps upstream service identities decoded from CDS SNIs on + // the proxied xDS stream. It is consulted during virtual-FQDN expansion so + // the dataplane fills missing tokens from the real upstream identity rather + // than the source proxy defaults. + upstreamIndex *dns.UpstreamIndex } // NewConsulDP creates a new instance of ConsulDataplane @@ -66,8 +71,9 @@ func NewConsulDP(cfg *Config) (*ConsulDataplane, error) { }) return &ConsulDataplane{ - logger: logger, - cfg: cfg, + logger: logger, + cfg: cfg, + upstreamIndex: dns.NewUpstreamIndex(), }, nil } @@ -186,7 +192,7 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error { if cdp.cfg.Mode == ModeTypeDNSProxy { // start up DNS server with the configuration from the consul-dataplane flags / environment variables since // envoy bootstrapping is bypassed. - if err = cdp.startDNSProxy(ctx, cdp.cfg.DNSServer, cdp.cfg.Proxy.Namespace, cdp.cfg.Proxy.Partition); err != nil { + if err = cdp.startDNSProxy(ctx, cdp.cfg.DNSServer, cdp.cfg.Proxy.Namespace, cdp.cfg.Proxy.Partition, ""); err != nil { cdp.logger.Error("failed to start the dns proxy", "error", err) return err } @@ -214,7 +220,7 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error { cdp.logger.Debug("generated envoy bootstrap params", "params", bootstrapParams) // start up DNS server with envoy bootstrap params. - if err = cdp.startDNSProxy(ctx, cdp.cfg.DNSServer, bootstrapParams.Namespace, bootstrapParams.Partition); err != nil { + if err = cdp.startDNSProxy(ctx, cdp.cfg.DNSServer, bootstrapParams.Namespace, bootstrapParams.Partition, bootstrapParams.Datacenter); err != nil { cdp.logger.Error("failed to start the dns proxy", "error", err) return err } @@ -287,17 +293,28 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error { } func (cdp *ConsulDataplane) startDNSProxy(ctx context.Context, - dnsConfig *DNSServerConfig, namespace, partition string) error { + dnsConfig *DNSServerConfig, namespace, partition, datacenter string) error { dnsClientInterface := pbdns.NewDNSServiceClient(cdp.serverConn) + var ( + virtualDNSInlineAddr string + virtualDNSEgressAddr string + ) + virtualDNSInlineAddr = fmt.Sprintf("127.0.0.1:%d", VirtualDNSInlinePort) + virtualDNSEgressAddr = fmt.Sprintf("127.0.0.1:%d", VirtualDNSEgressPort) + dnsServer, err := dns.NewDNSServer(dns.DNSServerParams{ - BindAddr: dnsConfig.BindAddr, - Port: dnsConfig.Port, - Client: dnsClientInterface, - Logger: cdp.logger, - Partition: partition, - Namespace: namespace, - Token: cdp.aclToken, + BindAddr: dnsConfig.BindAddr, + Port: dnsConfig.Port, + Client: dnsClientInterface, + Logger: cdp.logger, + Partition: partition, + Namespace: namespace, + Token: cdp.aclToken, + Datacenter: datacenter, + VirtualDNSInlineAddr: virtualDNSInlineAddr, + VirtualDNSEgressAddr: virtualDNSEgressAddr, + UpstreamIndex: cdp.upstreamIndex, }) if err == dns.ErrServerDisabled { cdp.logger.Info("dns server disabled: configure the Consul DNS port to enable") diff --git a/pkg/consuldp/xds.go b/pkg/consuldp/xds.go index f352ae8d..0740a908 100644 --- a/pkg/consuldp/xds.go +++ b/pkg/consuldp/xds.go @@ -15,6 +15,8 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" + + "github.com/hashicorp/consul-dataplane/pkg/dns" ) const ( @@ -118,15 +120,23 @@ func (cdp *ConsulDataplane) xdsServerExited() chan struct{} { return cdp.xdsServ func (cdp *ConsulDataplane) streamInterceptor() grpc.StreamServerInterceptor { return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { - return handler(srv, &metricServerStream{ss}) + return handler(srv, &metricServerStream{ServerStream: ss, upstreamIndex: cdp.upstreamIndex}) } } type metricServerStream struct { grpc.ServerStream + // upstreamIndex, when set, receives CDS cluster SNIs decoded from the + // server->Envoy frames that pass through this stream. + upstreamIndex *dns.UpstreamIndex } func (s *metricServerStream) SendMsg(m interface{}) error { + // Tap CDS frames on their way to Envoy to keep the upstream identity index + // current. This reads a copy of the frame's bytes and never mutates m, so + // opaque passthrough is unaffected. + tapClusterFrame(s.upstreamIndex, m) + err := s.ServerStream.SendMsg(m) if err == nil { metrics.SetGauge([]string{"envoy_connected"}, 1) diff --git a/pkg/consuldp/xds_cluster_tap.go b/pkg/consuldp/xds_cluster_tap.go new file mode 100644 index 00000000..84846724 --- /dev/null +++ b/pkg/consuldp/xds_cluster_tap.go @@ -0,0 +1,132 @@ +// Copyright IBM Corp. 2022, 2026 +// SPDX-License-Identifier: MPL-2.0 + +package consuldp + +import ( + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/types/known/emptypb" + + "github.com/hashicorp/consul-dataplane/pkg/dns" +) + +// clusterTypeURL is the xDS type URL carried by CDS responses. Only responses +// with this type URL contribute upstream identities to the index. +const clusterTypeURL = "type.googleapis.com/envoy.config.cluster.v3.Cluster" + +// Field numbers within envoy.service.discovery.v3.DeltaDiscoveryResponse. +// These mirror the generated proto (see go-control-plane discovery.pb.go) and +// are the raw wire identifiers we scan for so the dataplane does not need to +// pull in the full envoy proto dependency just to read cluster names. +const ( + deltaFieldResources = 2 // repeated Resource + deltaFieldTypeURL = 4 // string + deltaFieldRemovedResources = 6 // repeated string + resourceFieldName = 3 // string (the cluster SNI) +) + +// tapClusterFrame inspects a single server->Envoy xDS frame flowing through the +// proxy. The proxy carries each frame as an opaque *emptypb.Empty whose unknown +// fields hold the wire-format DiscoveryResponse, so we read those raw bytes and +// hand-parse just the cluster names. When the frame is a CDS response, the +// added and removed cluster SNIs are applied to the upstream index. Any other +// frame (or an unparseable one) is ignored; the frame itself is never modified, +// so passthrough is unaffected. +func tapClusterFrame(index *dns.UpstreamIndex, m interface{}) { + if index == nil { + return + } + empty, ok := m.(*emptypb.Empty) + if !ok { + return + } + raw := empty.ProtoReflect().GetUnknown() + if len(raw) == 0 { + return + } + added, removed, isCluster := extractClusterSNIs(raw) + if !isCluster { + return + } + index.Update(added, removed) +} + +// extractClusterSNIs scans a wire-format DeltaDiscoveryResponse for the cluster +// SNIs it added/updated and removed. It returns isCluster=false (and no names) +// when the frame is not a CDS response or cannot be parsed. +func extractClusterSNIs(raw []byte) (added, removed []string, isCluster bool) { + var ( + typeURL string + names []string + gone []string + ) + b := raw + for len(b) > 0 { + num, typ, n := protowire.ConsumeTag(b) + if n < 0 { + return nil, nil, false + } + b = b[n:] + switch { + case num == deltaFieldTypeURL && typ == protowire.BytesType: + v, vn := protowire.ConsumeBytes(b) + if vn < 0 { + return nil, nil, false + } + typeURL = string(v) + b = b[vn:] + case num == deltaFieldResources && typ == protowire.BytesType: + v, vn := protowire.ConsumeBytes(b) + if vn < 0 { + return nil, nil, false + } + if name, ok := extractResourceName(v); ok { + names = append(names, name) + } + b = b[vn:] + case num == deltaFieldRemovedResources && typ == protowire.BytesType: + v, vn := protowire.ConsumeBytes(b) + if vn < 0 { + return nil, nil, false + } + gone = append(gone, string(v)) + b = b[vn:] + default: + vn := protowire.ConsumeFieldValue(num, typ, b) + if vn < 0 { + return nil, nil, false + } + b = b[vn:] + } + } + if typeURL != clusterTypeURL { + return nil, nil, false + } + return names, gone, true +} + +// extractResourceName scans a wire-format DeltaDiscoveryResponse.Resource for +// its name field, which for a CDS cluster is the upstream SNI. +func extractResourceName(raw []byte) (string, bool) { + b := raw + for len(b) > 0 { + num, typ, n := protowire.ConsumeTag(b) + if n < 0 { + return "", false + } + b = b[n:] + if num == resourceFieldName && typ == protowire.BytesType { + v, vn := protowire.ConsumeBytes(b) + if vn < 0 { + return "", false + } + return string(v), true + } + vn := protowire.ConsumeFieldValue(num, typ, b) + if vn < 0 { + return "", false + } + b = b[vn:] + } + return "", false +} diff --git a/pkg/consuldp/xds_cluster_tap_test.go b/pkg/consuldp/xds_cluster_tap_test.go new file mode 100644 index 00000000..fd3ecec1 --- /dev/null +++ b/pkg/consuldp/xds_cluster_tap_test.go @@ -0,0 +1,102 @@ +// Copyright IBM Corp. 2022, 2026 +// SPDX-License-Identifier: MPL-2.0 + +package consuldp + +import ( + "testing" + + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/types/known/emptypb" + + "github.com/hashicorp/consul-dataplane/pkg/dns" +) + +func encodeResource(name string) []byte { + var b []byte + b = protowire.AppendTag(b, resourceFieldName, protowire.BytesType) + b = protowire.AppendBytes(b, []byte(name)) + return b +} + +func encodeDeltaResponse(typeURL string, names, removed []string) []byte { + var b []byte + b = protowire.AppendTag(b, deltaFieldTypeURL, protowire.BytesType) + b = protowire.AppendBytes(b, []byte(typeURL)) + for _, n := range names { + b = protowire.AppendTag(b, deltaFieldResources, protowire.BytesType) + b = protowire.AppendBytes(b, encodeResource(n)) + } + for _, r := range removed { + b = protowire.AppendTag(b, deltaFieldRemovedResources, protowire.BytesType) + b = protowire.AppendBytes(b, []byte(r)) + } + return b +} + +func emptyWithBytes(b []byte) *emptypb.Empty { + e := &emptypb.Empty{} + e.ProtoReflect().SetUnknown(protoreflect.RawFields(b)) + return e +} + +func TestExtractClusterSNIs(t *testing.T) { + const td = "e5b1a4d3.consul" + + t.Run("cluster response", func(t *testing.T) { + raw := encodeDeltaResponse(clusterTypeURL, + []string{"api.default.dc1.internal." + td}, + []string{"web.default.partition-vms.dc1.internal-v1." + td}, + ) + added, removed, ok := extractClusterSNIs(raw) + if !ok { + t.Fatal("expected cluster response") + } + if len(added) != 1 || added[0] != "api.default.dc1.internal."+td { + t.Fatalf("unexpected added: %v", added) + } + if len(removed) != 1 || removed[0] != "web.default.partition-vms.dc1.internal-v1."+td { + t.Fatalf("unexpected removed: %v", removed) + } + }) + + t.Run("non-cluster response ignored", func(t *testing.T) { + raw := encodeDeltaResponse( + "type.googleapis.com/envoy.config.listener.v3.Listener", + []string{"ignored"}, nil) + if _, _, ok := extractClusterSNIs(raw); ok { + t.Fatal("expected non-cluster response to be ignored") + } + }) + + t.Run("empty bytes", func(t *testing.T) { + if _, _, ok := extractClusterSNIs(nil); ok { + t.Fatal("expected empty input to be ignored") + } + }) +} + +func TestTapClusterFrame(t *testing.T) { + const td = "e5b1a4d3.consul" + idx := dns.NewUpstreamIndex() + + raw := encodeDeltaResponse(clusterTypeURL, + []string{"api.default.partition-vms.dc1.internal-v1." + td}, nil) + tapClusterFrame(idx, emptyWithBytes(raw)) + + if _, ok := idx.Lookup("api", "", "partition-vms", "dc1"); !ok { + t.Fatal("expected cluster SNI to be indexed") + } + + // Non-Empty message must be ignored without panicking. + tapClusterFrame(idx, "not-an-empty") + + // A removal frame drops the entry. + rm := encodeDeltaResponse(clusterTypeURL, nil, + []string{"api.default.partition-vms.dc1.internal-v1." + td}) + tapClusterFrame(idx, emptyWithBytes(rm)) + if _, ok := idx.Lookup("api", "", "partition-vms", "dc1"); ok { + t.Fatal("expected entry to be removed") + } +} diff --git a/pkg/dns/dns.go b/pkg/dns/dns.go index 6b70d9d4..b18eb3a0 100644 --- a/pkg/dns/dns.go +++ b/pkg/dns/dns.go @@ -11,11 +11,13 @@ import ( "io" "math" "net" + "strings" "sync" "time" "github.com/hashicorp/consul/proto-public/pbdns" "github.com/hashicorp/go-hclog" + "golang.org/x/net/dns/dnsmessage" "google.golang.org/grpc/metadata" ) @@ -25,6 +27,11 @@ var ErrServerDisabled error = errors.New("server is disabled") // ErrServerRunning is returned when the server is already running var ErrServerRunning error = errors.New("server is already running") +const ( + envoyDNSForwardTimeout = 300 * time.Millisecond + listenerUnhealthyTTL = 5 * time.Second +) + // DNSServerParams is the configuration for creating a new DNS server type DNSServerParams struct { BindAddr string @@ -32,9 +39,18 @@ type DNSServerParams struct { Logger hclog.Logger Client pbdns.DNSServiceClient - Partition string - Namespace string - Token string + Partition string + Namespace string + Token string + Datacenter string + // VirtualDNSInlineAddr is the address of Envoy's inline DNS listener (127.0.0.1:8653). + VirtualDNSInlineAddr string + // VirtualDNSEgressAddr is the address of Envoy's egress DNS listener (127.0.0.1:8654). + VirtualDNSEgressAddr string + // UpstreamIndex, when set, is consulted during virtual-FQDN expansion to + // fill missing namespace/partition/datacenter tokens from the real upstream + // identity (decoded from CDS SNIs) instead of the source proxy defaults. + UpstreamIndex *UpstreamIndex } // DNSServerInterface is the interface for athe DNSServer @@ -59,9 +75,17 @@ type DNSServer struct { connUDP net.PacketConn listenerTCP net.Listener - partition string - namespace string - token string + partition string + namespace string + token string + datacenter string + virtualDNSInlineAddr string + virtualDNSEgressAddr string + upstreamIndex *UpstreamIndex + + listenerHealthLock sync.Mutex + inlineListenerUnavailableTill time.Time + egressListenerUnavailableTill time.Time } // NewDNSServer creates a new DNS proxy server @@ -78,8 +102,12 @@ func NewDNSServer(p DNSServerParams) (DNSServerInterface, error) { s.client = p.Client s.logger = p.Logger.Named("dns-proxy") s.partition = p.Partition + s.datacenter = p.Datacenter + s.virtualDNSInlineAddr = p.VirtualDNSInlineAddr + s.virtualDNSEgressAddr = p.VirtualDNSEgressAddr s.namespace = p.Namespace s.token = p.Token + s.upstreamIndex = p.UpstreamIndex return s, nil } @@ -203,32 +231,15 @@ func (d *DNSServer) proxyUDP(ctx context.Context) { func (d *DNSServer) queryConsulAndRespondUDP(buf []byte, addr net.Addr) { logger := d.logger.Named("udp") - req := &pbdns.QueryRequest{ - Msg: buf, - Protocol: pbdns.Protocol_PROTOCOL_UDP, - } - ctx, done := context.WithTimeout(context.Background(), time.Minute*1) - defer done() - - ctx = metadata.AppendToOutgoingContext(ctx, - "x-consul-partition", d.partition, - "x-consul-namespace", d.namespace, - "x-consul-token", d.token, - ) - - logger.Debug("querying through udp", "partition", d.partition, "namespace", d.namespace) - - resp, err := d.client.Query(ctx, req) + respMsg, err := d.triageAndResolve(buf, pbdns.Protocol_PROTOCOL_UDP) if err != nil { - logger.Error("error resolving consul request", "error", err) + logger.Error("error resolving dns request", "error", err) return } - logger.Debug("dns messaged received from consul", "length", len(resp.Msg)) - _, err = d.connUDP.WriteTo(resp.Msg, addr) + _, err = d.connUDP.WriteTo(respMsg, addr) if err != nil { logger.Error("error sending response", "error", err) - return } } @@ -244,11 +255,11 @@ func (d *DNSServer) proxyTCP(ctx context.Context) { if err != nil { d.logger.Warn("failure to accept tcp connection", "error", err) } - go d.proxyTCPAcceptedConn(ctx, c, d.client) + go d.proxyTCPAcceptedConn(ctx, c) } } -func (d *DNSServer) proxyTCPAcceptedConn(ctx context.Context, conn net.Conn, client pbdns.DNSServiceClient) { +func (d *DNSServer) proxyTCPAcceptedConn(ctx context.Context, conn net.Conn) { defer conn.Close() logger := d.logger.Named("tcp") for { @@ -290,45 +301,29 @@ func (d *DNSServer) proxyTCPAcceptedConn(ctx context.Context, conn net.Conn, cli continue } - // Now that we have the request we can forward the dnsrequest to consul - req := &pbdns.QueryRequest{ - Msg: data, - Protocol: pbdns.Protocol_PROTOCOL_TCP, - } - - ctx, done := context.WithTimeout(context.Background(), time.Minute*1) - defer done() - - ctx = metadata.AppendToOutgoingContext(ctx, - "x-consul-partition", d.partition, - "x-consul-namespace", d.namespace, - "x-consul-token", d.token, - ) - - logger.Debug("querying through tcp", "partition", d.partition, "namespace", d.namespace) - - resp, err := client.Query(ctx, req) + logger.Debug("triaging dns request", "partition", d.partition, "namespace", d.namespace) + responseMsg, err := d.triageAndResolve(data, pbdns.Protocol_PROTOCOL_TCP) if err != nil { - logger.Error("error resolving consul request", "error", err) + logger.Error("error resolving dns request", "error", err) return } - logger.Debug("total data length of dns response from consul", "size", len(resp.Msg)) + logger.Debug("total data length of dns response", "size", len(responseMsg)) // This is a guard and shouldn't happen but if the response is > 65535 // then we will just close the connection. - if len(resp.Msg) > math.MaxUint16 { - logger.Error("consul response too large for DNS spec", "error", err) + if len(responseMsg) > math.MaxUint16 { + logger.Error("dns response too large for DNS spec") return } // TCP DNS requests add a two byte length field prefixed to the message. // Source: RFC1035 4.2.2. - err = binary.Write(conn, binary.BigEndian, uint16(len(resp.Msg))) + err = binary.Write(conn, binary.BigEndian, uint16(len(responseMsg))) if err != nil { logger.Warn("error writing length", "error", err) return } - _, err = conn.Write(resp.Msg) + _, err = conn.Write(responseMsg) if err != nil { logger.Error("error writing response", "error", err) return @@ -345,3 +340,374 @@ func (d *DNSServer) Stop() { } d.cancel() } + +// ----------------------------------------------------------------------------- +// DNS Triage Logic +// ----------------------------------------------------------------------------- + +// domainClass categorises an incoming DNS query domain name. +type domainClass int + +const ( + domainClassVirtual domainClass = iota // *.virtual.*consul + domainClassConsul // any other *.consul domain + domainClassExternal // everything else +) + +// classifyDomain returns the class of the fully-qualified domain name. +// name must be in canonical (lower-case, trailing-dot-stripped) form. +func classifyDomain(name string) domainClass { + name = strings.ToLower(strings.TrimSuffix(name, ".")) + if strings.HasSuffix(name, ".consul") || name == "consul" { + // Check for virtual: must contain ".virtual." segment + if strings.Contains(name, ".virtual.") { + return domainClassVirtual + } + return domainClassConsul + } + return domainClassExternal +} + +// parseVirtualTokens splits a virtual domain name into its service name and any +// explicitly supplied namespace/partition/datacenter tokens. Tokens that are +// not present in the query are returned as empty strings (callers decide how to +// fill them). ok is false when name is not a virtual domain. +func parseVirtualTokens(name string) (svc, ns, partition, dc string, ok bool) { + name = strings.ToLower(strings.TrimSuffix(name, ".")) + + // Find the service name: everything up to the first ".virtual." segment. + virtualIdx := strings.Index(name, ".virtual.") + if virtualIdx < 0 { + return "", "", "", "", false + } + svc = name[:virtualIdx] + svc = strings.TrimSuffix(svc, ".service") + // Remainder after ".virtual." — may be empty (just "consul") or have + // ns/ap/dc qualifiers. + remainder := name[virtualIdx+len(".virtual."):] + // Strip trailing ".consul" if present. + remainder = strings.TrimSuffix(remainder, ".consul") + + // Parse remainder tokens separated by ".": + // Possible patterns of (label, qualifier) pairs: ns, ap, dc in any order. + parts := strings.Split(remainder, ".") + for i := 0; i+1 < len(parts); i++ { + switch parts[i+1] { + case "ns": + ns = parts[i] + i++ // skip qualifier + case "ap": + partition = parts[i] + i++ + case "dc": + dc = parts[i] + i++ + } + } + + return svc, ns, partition, dc, true +} + +// expandVirtualFQDN expands any of the 8 short-form virtual domain names into +// the canonical form: +// +// .virtual..ns..ap..dc.consul +// +// Missing components are filled from the provided default namespace, partition +// and datacenter. +// func expandVirtualFQDN(name, defaultNS, defaultPartition, defaultDC string) string { +// svc, ns, partition, dc, ok := parseVirtualTokens(name) +// if !ok { +// return strings.ToLower(strings.TrimSuffix(name, ".")) +// } +// if ns == "" { +// ns = defaultNS +// } +// if partition == "" { +// partition = defaultPartition +// } +// if dc == "" { +// dc = defaultDC +// } +// return fmt.Sprintf("%s.virtual.%s.ns.%s.ap.%s.dc.consul", svc, ns, partition, dc) +// } + +// expandVirtualName expands a short-form virtual domain name into its canonical +// FQDN. Missing namespace/partition/datacenter tokens are first resolved from +// the upstream identity advertised through CDS (the same raw identity the +// control plane used to build Envoy's inline DNS table); any still-missing +// tokens fall back to the source proxy's own namespace/partition/datacenter. +// This keeps the dataplane's expansion consistent with Envoy's inline answer in +// cross-partition, cross-namespace, and cross-datacenter scenarios. +func (d *DNSServer) expandVirtualName(name string) string { + svc, ns, partition, dc, ok := parseVirtualTokens(name) + if !ok { + return strings.ToLower(strings.TrimSuffix(name, ".")) + } + + // Consult the upstream index using any explicit tokens as constraints. + // A unique match fills the tokens the query left unspecified. + if comp, found := d.upstreamIndex.Lookup(svc, ns, partition, dc); found { + if ns == "" { + ns = comp.Namespace + } + if partition == "" { + partition = comp.Partition + } + if dc == "" { + dc = comp.Datacenter + } + } + + // Fill anything still unspecified from the source proxy defaults. + if ns == "" { + ns = d.namespace + } + if partition == "" { + partition = d.partition + } + if dc == "" { + dc = d.datacenter + } + return fmt.Sprintf("%s.virtual.%s.ns.%s.ap.%s.dc.consul", svc, ns, partition, dc) +} + +// rewriteQueryName rewrites the question section of a DNS message with +// newName and returns the modified raw bytes. The original name in the +// question is replaced in-place at the wire level by re-encoding it. +func rewriteQueryName(raw []byte, newName string) ([]byte, string, error) { + var msg dnsmessage.Message + if err := msg.Unpack(raw); err != nil { + return nil, "", fmt.Errorf("unpack dns message: %w", err) + } + if len(msg.Questions) == 0 { + return raw, "", nil + } + originalName := msg.Questions[0].Name.String() + fqdn := newName + if !strings.HasSuffix(fqdn, ".") { + fqdn += "." + } + name, err := dnsmessage.NewName(fqdn) + if err != nil { + return nil, "", fmt.Errorf("build dns name %q: %w", fqdn, err) + } + msg.Questions[0].Name = name + // Re-pack the message to re-encode the updated QNAME. + rewritten, err := msg.Pack() + if err != nil { + return nil, "", fmt.Errorf("pack dns message: %w", err) + } + return rewritten, originalName, nil +} + +// rewriteResponseName replaces all occurrences of expandedName in the +// response's answer/authority/additional sections with originalName so the +// caller receives a response that matches the name it queried. +func rewriteResponseName(raw []byte, expandedName, originalName string) ([]byte, error) { + var msg dnsmessage.Message + if err := msg.Unpack(raw); err != nil { + return raw, nil // best-effort; return original on parse failure + } + expanded, err := dnsmessage.NewName(canonicalName(expandedName)) + if err != nil { + return raw, nil + } + original, err := dnsmessage.NewName(canonicalName(originalName)) + if err != nil { + return raw, nil + } + + rewrite := func(name *dnsmessage.Name) { + if name.String() == expanded.String() { + *name = original + } + } + for i := range msg.Questions { + rewrite(&msg.Questions[i].Name) + } + for i := range msg.Answers { + rewrite(&msg.Answers[i].Header.Name) + } + for i := range msg.Authorities { + rewrite(&msg.Authorities[i].Header.Name) + } + for i := range msg.Additionals { + rewrite(&msg.Additionals[i].Header.Name) + } + out, err := msg.Pack() + if err != nil { + return raw, nil + } + return out, nil +} + +func canonicalName(n string) string { + n = strings.ToLower(n) + if !strings.HasSuffix(n, ".") { + n += "." + } + return n +} + +// isNXDOMAIN returns true when the DNS response carries an NXDOMAIN rcode. +func isNXDOMAIN(raw []byte) bool { + var msg dnsmessage.Message + if err := msg.Unpack(raw); err != nil { + return false + } + return msg.RCode == dnsmessage.RCodeNameError +} + +// forwardUDP sends a raw DNS query to addr and returns the raw response. +func forwardUDP(addr string, query []byte, timeout time.Duration) ([]byte, error) { + conn, err := net.DialTimeout("udp", addr, timeout) + if err != nil { + return nil, fmt.Errorf("dial udp %s: %w", addr, err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(timeout)); err != nil { + return nil, err + } + if _, err := conn.Write(query); err != nil { + return nil, fmt.Errorf("write udp query: %w", err) + } + buf := make([]byte, 65535) + n, err := conn.Read(buf) + if err != nil { + return nil, fmt.Errorf("read udp response: %w", err) + } + return buf[:n], nil +} + +// queryConsul forwards a raw DNS message to the Consul server via gRPC and +// returns the raw response bytes. +func (d *DNSServer) queryConsul(raw []byte, proto pbdns.Protocol) ([]byte, error) { + req := &pbdns.QueryRequest{Msg: raw, Protocol: proto} + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + ctx = metadata.AppendToOutgoingContext(ctx, + "x-consul-partition", d.partition, + "x-consul-namespace", d.namespace, + "x-consul-token", d.token, + ) + resp, err := d.client.Query(ctx, req) + if err != nil { + return nil, err + } + return resp.Msg, nil +} + +func (d *DNSServer) canTryInlineListener() bool { + d.listenerHealthLock.Lock() + defer d.listenerHealthLock.Unlock() + return time.Now().After(d.inlineListenerUnavailableTill) +} + +func (d *DNSServer) canTryEgressListener() bool { + d.listenerHealthLock.Lock() + defer d.listenerHealthLock.Unlock() + return time.Now().After(d.egressListenerUnavailableTill) +} + +func (d *DNSServer) markInlineListenerUnavailable() { + d.listenerHealthLock.Lock() + defer d.listenerHealthLock.Unlock() + d.inlineListenerUnavailableTill = time.Now().Add(listenerUnhealthyTTL) +} + +func (d *DNSServer) markEgressListenerUnavailable() { + d.listenerHealthLock.Lock() + defer d.listenerHealthLock.Unlock() + d.egressListenerUnavailableTill = time.Now().Add(listenerUnhealthyTTL) +} + +// triageAndResolve is the main entry point for the virtual DNS triage logic. +// It classifies the domain, expands FQDNs, forwards to the right backend, and +// handles NXDOMAIN fallback. proto determines which protocol label is used for +// Consul gRPC queries. +func (d *DNSServer) triageAndResolve(raw []byte, proto pbdns.Protocol) ([]byte, error) { + // Parse domain from the first question. + var msg dnsmessage.Message + if err := msg.Unpack(raw); err != nil || len(msg.Questions) == 0 { + // Unparseable — fall back to Consul server. + return d.queryConsul(raw, proto) + } + originalName := strings.TrimSuffix(msg.Questions[0].Name.String(), ".") + + class := classifyDomain(originalName) + + switch class { + case domainClassConsul: + // Standard Consul domain — unchanged path. + return d.queryConsul(raw, proto) + + case domainClassExternal: + // Non-consul domain. + if d.virtualDNSEgressAddr == "" || !d.canTryEgressListener() { + return d.queryConsul(raw, proto) + } + + // Forward to Envoy egress DNS listener (UDP only — c-ares plugin is UDP). + resp, err := forwardUDP(d.virtualDNSEgressAddr, raw, envoyDNSForwardTimeout) + if err != nil { + d.markEgressListenerUnavailable() + d.logger.Debug("egress listener unavailable, falling back to consul", "error", err) + return d.queryConsul(raw, proto) + } + return resp, nil + + case domainClassVirtual: + if d.virtualDNSInlineAddr == "" { + return d.queryConsul(raw, proto) + } + + // Expand the short form to the full FQDN. + expandedName := d.expandVirtualName(originalName) + + // Rewrite the query with the expanded name before forwarding to Envoy. + rewrittenQuery, _, err := rewriteQueryName(raw, expandedName) + d.logger.Debug("virtual dns query", "original_name", originalName, "expanded_name", expandedName, "rewritten_query_len", len(rewrittenQuery), "error", err) + if err != nil { + d.logger.Warn("failed to rewrite query name, falling back to consul", "error", err) + return d.queryConsul(raw, proto) + } + + // Forward to Envoy's inline DNS listener (UDP — dns_filter is UDP-only). + if !d.canTryInlineListener() { + envoyErr := errors.New("inline listener in backoff window") + d.logger.Debug("virtual dns inline listener in backoff, falling back to consul", "domain", originalName, "error", envoyErr) + return d.queryConsul(raw, proto) + } + + envoyResp, envoyErr := forwardUDP(d.virtualDNSInlineAddr, rewrittenQuery, envoyDNSForwardTimeout) + if envoyErr == nil && !isNXDOMAIN(envoyResp) { + // Hit — rewrite the response name back to the original and return. + out, err := rewriteResponseName(envoyResp, expandedName, originalName) + if err != nil { + return envoyResp, nil + } + return out, nil + } + if envoyErr != nil { + d.markInlineListenerUnavailable() + } + + // NXDOMAIN from Envoy (or forwarding error) — fall back to Consul server. + // Use the original (unexpanded) query so the server applies its own + // expansion logic. + d.logger.Debug("virtual dns miss, falling back to consul server", + "domain", originalName, "envoy_error", envoyErr) + consulResp, consulErr := d.queryConsul(raw, proto) + if consulErr != nil { + if envoyErr != nil { + return nil, fmt.Errorf("virtual dns: envoy error: %v; consul fallback error: %w", envoyErr, consulErr) + } + return nil, consulErr + } + return consulResp, nil + } + + // Should never reach here. + return d.queryConsul(raw, proto) +} diff --git a/pkg/dns/dns_test.go b/pkg/dns/dns_test.go index 5b4ac7be..f7913923 100644 --- a/pkg/dns/dns_test.go +++ b/pkg/dns/dns_test.go @@ -19,6 +19,7 @@ import ( "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" + "golang.org/x/net/dns/dnsmessage" "google.golang.org/grpc/metadata" "github.com/hashicorp/consul-dataplane/pkg/dns/mocks" @@ -330,3 +331,397 @@ func (s *DNSTestSuite) Test_ProxydnsTCP() { }) } } + +func (s *DNSTestSuite) Test_ClassifyDomain() { + testCases := map[string]domainClass{ + "service.virtual.default.ns.default.ap.dc1.dc.consul": domainClassVirtual, + "service.default.consul": domainClassConsul, + "consul": domainClassConsul, + "google.com": domainClassExternal, + } + + for domain, expected := range testCases { + s.Run(domain, func() { + s.Require().Equal(expected, classifyDomain(domain)) + }) + } +} + +// func (s *DNSTestSuite) Test_ExpandVirtualFQDN() { +// testCases := []struct { +// name string +// input string +// expected string +// defaultNS string +// defaultAP string +// defaultDC string +// }{ +// { +// name: "short form", +// input: "service.virtual.consul", +// expected: "service.virtual.default.ns.default.ap.dc1.dc.consul", +// defaultNS: "default", +// defaultAP: "default", +// defaultDC: "dc1", +// }, +// { +// name: "override namespace and datacenter", +// input: "service.virtual.other.ns.dc2.dc.consul", +// expected: "service.virtual.other.ns.default.ap.dc2.dc.consul", +// defaultNS: "default", +// defaultAP: "default", +// defaultDC: "dc1", +// }, +// { +// name: "override all qualifiers", +// input: "service.virtual.team.ns.part.ap.dc2.dc.consul", +// expected: "service.virtual.team.ns.part.ap.dc2.dc.consul", +// defaultNS: "default", +// defaultAP: "default", +// defaultDC: "dc1", +// }, +// { +// name: "service alias short form", +// input: "service-db.service.virtual.consul", +// expected: "service-db.virtual.default.ns.default.ap.dc1.dc.consul", +// defaultNS: "default", +// defaultAP: "default", +// defaultDC: "dc1", +// }, +// } + +// for _, tc := range testCases { +// s.Run(tc.name, func() { +// out := expandVirtualFQDN(tc.input, tc.defaultNS, tc.defaultAP, tc.defaultDC) +// s.Require().Equal(tc.expected, out) +// }) +// } +// } + +func (s *DNSTestSuite) Test_TriageAndResolve_ConsulDomain() { + mockedDNSConsulClient := mocks.NewDNSServiceClient(s.T()) + server := DNSServer{ + client: mockedDNSConsulClient, + logger: hclog.Default(), + partition: "test-partition", + namespace: "test-namespace", + token: "test-token", + } + + query := buildDNSQuery(s.T(), "service.default.consul") + consulResp := buildDNSAnswerResponse(s.T(), "service.default.consul", "service.default.consul", dnsmessage.RCodeSuccess) + + mockedDNSConsulClient.On("Query", mock.Anything, mock.Anything). + Return(&pbdns.QueryResponse{Msg: consulResp}, nil). + Once() + + resp, err := server.triageAndResolve(query, pbdns.Protocol_PROTOCOL_UDP) + s.Require().NoError(err) + s.Require().Equal(consulResp, resp) +} + +func (s *DNSTestSuite) Test_TriageAndResolve_ExternalDomain_EgressForwardingAndFallback() { + s.Run("uses egress listener for external domains", func() { + mockedDNSConsulClient := mocks.NewDNSServiceClient(s.T()) + query := buildDNSQuery(s.T(), "www.example.com") + expectedResp := buildDNSAnswerResponse(s.T(), "www.example.com", "www.example.com", dnsmessage.RCodeSuccess) + + udpConn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + s.Require().NoError(err) + defer udpConn.Close() + + done := make(chan struct{}) + go func() { + defer close(done) + buf := make([]byte, 4096) + n, addr, readErr := udpConn.ReadFromUDP(buf) + if readErr != nil { + return + } + _, _ = udpConn.WriteToUDP(expectedResp, addr) + _ = n + }() + + server := DNSServer{ + client: mockedDNSConsulClient, + logger: hclog.Default(), + virtualDNSEgressAddr: udpConn.LocalAddr().String(), + } + + resp, err := server.triageAndResolve(query, pbdns.Protocol_PROTOCOL_UDP) + s.Require().NoError(err) + s.Require().Equal(expectedResp, resp) + <-done + }) + + s.Run("falls back to consul on egress listener error", func() { + mockedDNSConsulClient := mocks.NewDNSServiceClient(s.T()) + query := buildDNSQuery(s.T(), "www.example.com") + consulResp := buildDNSAnswerResponse(s.T(), "www.example.com", "www.example.com", dnsmessage.RCodeSuccess) + + server := DNSServer{ + client: mockedDNSConsulClient, + logger: hclog.Default(), + partition: "test-partition", + namespace: "test-namespace", + token: "test-token", + virtualDNSEgressAddr: "127.0.0.1:1", + } + + mockedDNSConsulClient.On("Query", mock.Anything, mock.Anything). + Return(&pbdns.QueryResponse{Msg: consulResp}, nil). + Once() + + resp, err := server.triageAndResolve(query, pbdns.Protocol_PROTOCOL_UDP) + s.Require().NoError(err) + s.Require().Equal(consulResp, resp) + s.Require().False(server.canTryEgressListener()) + }) +} + +func (s *DNSTestSuite) Test_TriageAndResolve_VirtualDomain() { + s.Run("inline hit rewrites response back to original name", func() { + mockedDNSConsulClient := mocks.NewDNSServiceClient(s.T()) + + originalName := "service.virtual.consul" + expandedName := "service.virtual.default.ns.partition-vms.ap.dc1.dc.consul" + query := buildDNSQuery(s.T(), originalName) + + udpConn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + s.Require().NoError(err) + defer udpConn.Close() + + done := make(chan struct{}) + go func() { + defer close(done) + buf := make([]byte, 4096) + n, addr, readErr := udpConn.ReadFromUDP(buf) + if readErr != nil { + return + } + receivedName := firstQuestionNameFromRaw(s.T(), buf[:n]) + s.Equal(canonicalName(expandedName), receivedName) + response := buildDNSAnswerResponse(s.T(), expandedName, expandedName, dnsmessage.RCodeSuccess) + _, _ = udpConn.WriteToUDP(response, addr) + }() + + server := DNSServer{ + client: mockedDNSConsulClient, + logger: hclog.Default(), + namespace: "default", + partition: "partition-vms", + datacenter: "dc1", + virtualDNSInlineAddr: udpConn.LocalAddr().String(), + } + + resp, err := server.triageAndResolve(query, pbdns.Protocol_PROTOCOL_UDP) + s.Require().NoError(err) + s.Require().Equal(canonicalName(originalName), firstQuestionNameFromRaw(s.T(), resp)) + s.Require().Equal(canonicalName(originalName), firstAnswerNameFromRaw(s.T(), resp)) + <-done + }) + + s.Run("service alias inline hit rewrites response back to original name", func() { + mockedDNSConsulClient := mocks.NewDNSServiceClient(s.T()) + + originalName := "service-db.service.virtual.consul" + expandedName := "service-db.virtual.default.ns.partition-vms.ap.dc1.dc.consul" + query := buildDNSQuery(s.T(), originalName) + + udpConn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + s.Require().NoError(err) + defer udpConn.Close() + + done := make(chan struct{}) + go func() { + defer close(done) + buf := make([]byte, 4096) + n, addr, readErr := udpConn.ReadFromUDP(buf) + if readErr != nil { + return + } + receivedName := firstQuestionNameFromRaw(s.T(), buf[:n]) + s.Equal(canonicalName(expandedName), receivedName) + response := buildDNSAnswerResponse(s.T(), expandedName, expandedName, dnsmessage.RCodeSuccess) + _, _ = udpConn.WriteToUDP(response, addr) + }() + + server := DNSServer{ + client: mockedDNSConsulClient, + logger: hclog.Default(), + namespace: "default", + partition: "partition-vms", + datacenter: "dc1", + virtualDNSInlineAddr: udpConn.LocalAddr().String(), + } + + resp, err := server.triageAndResolve(query, pbdns.Protocol_PROTOCOL_UDP) + s.Require().NoError(err) + s.Require().Equal(canonicalName(originalName), firstQuestionNameFromRaw(s.T(), resp)) + s.Require().Equal(canonicalName(originalName), firstAnswerNameFromRaw(s.T(), resp)) + <-done + }) + + s.Run("nxdomain from inline listener falls back to consul", func() { + mockedDNSConsulClient := mocks.NewDNSServiceClient(s.T()) + + originalName := "service.virtual.consul" + expandedName := "service.virtual.default.ns.partition-vms.ap.dc1.dc.consul" + query := buildDNSQuery(s.T(), originalName) + consulResp := buildDNSAnswerResponse(s.T(), originalName, originalName, dnsmessage.RCodeSuccess) + + udpConn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + s.Require().NoError(err) + defer udpConn.Close() + + done := make(chan struct{}) + go func() { + defer close(done) + buf := make([]byte, 4096) + n, addr, readErr := udpConn.ReadFromUDP(buf) + if readErr != nil { + return + } + receivedName := firstQuestionNameFromRaw(s.T(), buf[:n]) + s.Equal(canonicalName(expandedName), receivedName) + nx := buildDNSRCodeResponse(s.T(), expandedName, dnsmessage.RCodeNameError) + _, _ = udpConn.WriteToUDP(nx, addr) + }() + + server := DNSServer{ + client: mockedDNSConsulClient, + logger: hclog.Default(), + partition: "partition-vms", + namespace: "default", + token: "test-token", + datacenter: "dc1", + virtualDNSInlineAddr: udpConn.LocalAddr().String(), + } + + mockedDNSConsulClient.On("Query", mock.Anything, mock.Anything). + Return(&pbdns.QueryResponse{Msg: consulResp}, nil). + Once() + + resp, err := server.triageAndResolve(query, pbdns.Protocol_PROTOCOL_UDP) + s.Require().NoError(err) + s.Require().Equal(consulResp, resp) + <-done + }) + + s.Run("inline listener error falls back to consul and marks listener unavailable", func() { + mockedDNSConsulClient := mocks.NewDNSServiceClient(s.T()) + query := buildDNSQuery(s.T(), "service.virtual.consul") + consulResp := buildDNSAnswerResponse(s.T(), "service.virtual.consul", "service.virtual.consul", dnsmessage.RCodeSuccess) + + server := DNSServer{ + client: mockedDNSConsulClient, + logger: hclog.Default(), + partition: "partition-vms", + namespace: "default", + token: "test-token", + datacenter: "dc1", + virtualDNSInlineAddr: "127.0.0.1:1", + } + + mockedDNSConsulClient.On("Query", mock.Anything, mock.Anything). + Return(&pbdns.QueryResponse{Msg: consulResp}, nil). + Once() + + resp, err := server.triageAndResolve(query, pbdns.Protocol_PROTOCOL_UDP) + s.Require().NoError(err) + s.Require().Equal(consulResp, resp) + s.Require().False(server.canTryInlineListener()) + }) +} + +func buildDNSQuery(t *testing.T, name string) []byte { + t.Helper() + qName, err := dnsmessage.NewName(canonicalName(name)) + require.NoError(t, err) + + msg := dnsmessage.Message{ + Header: dnsmessage.Header{ID: 1, RecursionDesired: true}, + Questions: []dnsmessage.Question{{ + Name: qName, + Type: dnsmessage.TypeA, + Class: dnsmessage.ClassINET, + }}, + } + + raw, err := msg.Pack() + require.NoError(t, err) + return raw +} + +func buildDNSRCodeResponse(t *testing.T, questionName string, rcode dnsmessage.RCode) []byte { + t.Helper() + qName, err := dnsmessage.NewName(canonicalName(questionName)) + require.NoError(t, err) + + msg := dnsmessage.Message{ + Header: dnsmessage.Header{ + ID: 1, + Response: true, + RCode: rcode, + }, + Questions: []dnsmessage.Question{{ + Name: qName, + Type: dnsmessage.TypeA, + Class: dnsmessage.ClassINET, + }}, + } + + raw, err := msg.Pack() + require.NoError(t, err) + return raw +} + +func buildDNSAnswerResponse(t *testing.T, questionName, answerName string, rcode dnsmessage.RCode) []byte { + t.Helper() + qName, err := dnsmessage.NewName(canonicalName(questionName)) + require.NoError(t, err) + aName, err := dnsmessage.NewName(canonicalName(answerName)) + require.NoError(t, err) + + msg := dnsmessage.Message{ + Header: dnsmessage.Header{ + ID: 1, + Response: true, + RCode: rcode, + }, + Questions: []dnsmessage.Question{{ + Name: qName, + Type: dnsmessage.TypeA, + Class: dnsmessage.ClassINET, + }}, + Answers: []dnsmessage.Resource{{ + Header: dnsmessage.ResourceHeader{ + Name: aName, + Type: dnsmessage.TypeA, + Class: dnsmessage.ClassINET, + TTL: 1, + }, + Body: &dnsmessage.AResource{A: [4]byte{127, 0, 0, 1}}, + }}, + } + + raw, err := msg.Pack() + require.NoError(t, err) + return raw +} + +func firstQuestionNameFromRaw(t *testing.T, raw []byte) string { + t.Helper() + var msg dnsmessage.Message + require.NoError(t, msg.Unpack(raw)) + require.NotEmpty(t, msg.Questions) + return msg.Questions[0].Name.String() +} + +func firstAnswerNameFromRaw(t *testing.T, raw []byte) string { + t.Helper() + var msg dnsmessage.Message + require.NoError(t, msg.Unpack(raw)) + require.NotEmpty(t, msg.Answers) + return msg.Answers[0].Header.Name.String() +} diff --git a/pkg/dns/upstream_index.go b/pkg/dns/upstream_index.go new file mode 100644 index 00000000..de780ad9 --- /dev/null +++ b/pkg/dns/upstream_index.go @@ -0,0 +1,174 @@ +// Copyright IBM Corp. 2022, 2026 +// SPDX-License-Identifier: MPL-2.0 + +package dns + +import ( + "strings" + "sync" +) + +// UpstreamComponents captures the real identity of an upstream service as +// advertised by the control plane through the upstream cluster SNI. These are +// the raw tokens the control plane itself used to build Envoy's inline DNS +// table, so filling missing virtual-FQDN tokens from them keeps the dataplane's +// expansion consistent with what Envoy will actually answer. +type UpstreamComponents struct { + Service string + Namespace string + Partition string + Datacenter string +} + +// UpstreamIndex is a thread-safe index of upstream identities keyed by their +// SNI. It is populated by decoding the CDS (Cluster) resources that already +// flow through the dataplane's proxied xDS stream, so it introduces no new +// watch and no control-plane change. The last decoded state is retained across +// xDS reconnects so virtual expansion keeps working during a control-plane +// outage (BCDR resilience). +type UpstreamIndex struct { + mu sync.RWMutex + // entries is keyed by the full SNI (the CDS resource name) so that delta + // removals can drop exactly the upstream that went away. + entries map[string]UpstreamComponents +} + +// NewUpstreamIndex returns an empty, ready-to-use UpstreamIndex. +func NewUpstreamIndex() *UpstreamIndex { + return &UpstreamIndex{ + entries: make(map[string]UpstreamComponents), + } +} + +// Update applies a delta of CDS cluster SNIs to the index. added holds SNIs of +// clusters that were added or modified; removed holds SNIs of clusters that +// were removed. Non-internal SNIs (external, peered, gateway, prepared-query) +// are ignored because they do not participate in virtual-FQDN expansion. +func (idx *UpstreamIndex) Update(added, removed []string) { + if idx == nil { + return + } + idx.mu.Lock() + defer idx.mu.Unlock() + for _, sni := range removed { + delete(idx.entries, sni) + } + for _, sni := range added { + comp, ok := ParseServiceSNI(sni) + if !ok { + continue + } + idx.entries[sni] = comp + } +} + +// Lookup resolves the identity of the named service, using any explicitly +// provided namespace/partition/datacenter tokens as constraints. Empty +// constraint tokens match any value. It returns the matched components only +// when the constraints resolve to a single, unambiguous identity; otherwise it +// returns false so the caller can fall back to the source proxy's defaults. +func (idx *UpstreamIndex) Lookup(service, namespace, partition, datacenter string) (UpstreamComponents, bool) { + if idx == nil || service == "" { + return UpstreamComponents{}, false + } + idx.mu.RLock() + defer idx.mu.RUnlock() + + var ( + match UpstreamComponents + found bool + ) + for _, comp := range idx.entries { + if comp.Service != service { + continue + } + if namespace != "" && comp.Namespace != namespace { + continue + } + if partition != "" && comp.Partition != partition { + continue + } + if datacenter != "" && comp.Datacenter != datacenter { + continue + } + if found && comp != match { + // More than one distinct identity satisfies the constraints; + // the query is ambiguous, so refuse to guess. + return UpstreamComponents{}, false + } + match = comp + found = true + } + return match, found +} + +const ( + sniMarkerInternal = "internal" + sniMarkerInternalV1 = "internal-v1" +) + +// ParseServiceSNI parses a Consul upstream service SNI into its identity +// components. It understands the internal (mesh) SNI schemes emitted by the +// Consul control plane: +// +// default partition, no subset: ...internal. +// default partition, with subset: ....internal. +// other partition, no subset: ....internal-v1. +// other partition, with subset: .....internal-v1. +// +// The subset label, when present, is ignored: virtual-FQDN expansion keys off +// the base service name, and every subset of a service shares its identity. +// +// This parsing is intentionally coupled to Consul's SNI label scheme (see +// agent/connect/sni.go in the Consul server). SNIs that are external, peered, +// gateway, or prepared-query are not internal upstreams and return ok=false. +func ParseServiceSNI(sni string) (UpstreamComponents, bool) { + sni = strings.ToLower(strings.TrimSuffix(sni, ".")) + if sni == "" { + return UpstreamComponents{}, false + } + labels := strings.Split(sni, ".") + + // Locate the scheme marker. internal-v1 is checked before internal because + // it is the more specific token. + for i, label := range labels { + switch label { + case sniMarkerInternalV1: + // prefix = labels[:i] + switch i { + case 4: // svc, ns, ap, dc + return UpstreamComponents{ + Service: labels[0], + Namespace: labels[1], + Partition: labels[2], + Datacenter: labels[3], + }, true + case 5: // subset, svc, ns, ap, dc + return UpstreamComponents{ + Service: labels[1], + Namespace: labels[2], + Partition: labels[3], + Datacenter: labels[4], + }, true + } + case sniMarkerInternal: + switch i { + case 3: // svc, ns, dc (default partition) + return UpstreamComponents{ + Service: labels[0], + Namespace: labels[1], + Partition: "default", + Datacenter: labels[2], + }, true + case 4: // subset, svc, ns, dc (default partition) + return UpstreamComponents{ + Service: labels[1], + Namespace: labels[2], + Partition: "default", + Datacenter: labels[3], + }, true + } + } + } + return UpstreamComponents{}, false +} diff --git a/pkg/dns/upstream_index_test.go b/pkg/dns/upstream_index_test.go new file mode 100644 index 00000000..33fc4199 --- /dev/null +++ b/pkg/dns/upstream_index_test.go @@ -0,0 +1,131 @@ +// Copyright IBM Corp. 2022, 2026 +// SPDX-License-Identifier: MPL-2.0 + +package dns + +import "testing" + +func TestParseServiceSNI(t *testing.T) { + const td = "e5b1a4d3.consul" + cases := []struct { + name string + sni string + want UpstreamComponents + ok bool + }{ + { + name: "default partition no subset", + sni: "service-response.default.dc1.internal." + td, + want: UpstreamComponents{Service: "service-response", Namespace: "default", Partition: "default", Datacenter: "dc1"}, + ok: true, + }, + { + name: "default partition with subset", + sni: "v2.service-response.default.dc1.internal." + td, + want: UpstreamComponents{Service: "service-response", Namespace: "default", Partition: "default", Datacenter: "dc1"}, + ok: true, + }, + { + name: "non-default partition no subset", + sni: "service-response.default.partition-vms.dc1.internal-v1." + td, + want: UpstreamComponents{Service: "service-response", Namespace: "default", Partition: "partition-vms", Datacenter: "dc1"}, + ok: true, + }, + { + name: "non-default partition with subset", + sni: "v2.service-response.ns1.partition-vms.dc2.internal-v1." + td, + want: UpstreamComponents{Service: "service-response", Namespace: "ns1", Partition: "partition-vms", Datacenter: "dc2"}, + ok: true, + }, + { + name: "trailing dot tolerated", + sni: "api.default.dc1.internal." + td + ".", + want: UpstreamComponents{Service: "api", Namespace: "default", Partition: "default", Datacenter: "dc1"}, + ok: true, + }, + { + name: "external not indexed", + sni: "web.default.default.peer1.external." + td, + ok: false, + }, + { + name: "empty", + sni: "", + ok: false, + }, + { + name: "unrelated fqdn", + sni: "example.com", + ok: false, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, ok := ParseServiceSNI(tc.sni) + if ok != tc.ok { + t.Fatalf("ParseServiceSNI(%q) ok = %v, want %v", tc.sni, ok, tc.ok) + } + if ok && got != tc.want { + t.Fatalf("ParseServiceSNI(%q) = %+v, want %+v", tc.sni, got, tc.want) + } + }) + } +} + +func TestUpstreamIndexLookup(t *testing.T) { + const td = "e5b1a4d3.consul" + idx := NewUpstreamIndex() + idx.Update([]string{ + "service-response.default.partition-vms.dc1.internal-v1." + td, + "service-response.default.partition-vms.dc2.internal-v1." + td, + "api.default.dc1.internal." + td, + }, nil) + + t.Run("unique by service", func(t *testing.T) { + got, ok := idx.Lookup("api", "", "", "") + if !ok { + t.Fatal("expected match") + } + if got.Partition != "default" || got.Datacenter != "dc1" || got.Namespace != "default" { + t.Fatalf("unexpected components: %+v", got) + } + }) + + t.Run("ambiguous without constraint", func(t *testing.T) { + if _, ok := idx.Lookup("service-response", "", "", ""); ok { + t.Fatal("expected ambiguous lookup to fail") + } + }) + + t.Run("disambiguated by datacenter", func(t *testing.T) { + got, ok := idx.Lookup("service-response", "", "", "dc2") + if !ok { + t.Fatal("expected match") + } + if got.Partition != "partition-vms" || got.Datacenter != "dc2" { + t.Fatalf("unexpected components: %+v", got) + } + }) + + t.Run("unknown service", func(t *testing.T) { + if _, ok := idx.Lookup("missing", "", "", ""); ok { + t.Fatal("expected no match") + } + }) + + t.Run("removal drops entry", func(t *testing.T) { + idx.Update(nil, []string{"api.default.dc1.internal." + td}) + if _, ok := idx.Lookup("api", "", "", ""); ok { + t.Fatal("expected entry to be removed") + } + }) + + t.Run("nil index is safe", func(t *testing.T) { + var nilIdx *UpstreamIndex + nilIdx.Update([]string{"api.default.dc1.internal." + td}, nil) + if _, ok := nilIdx.Lookup("api", "", "", ""); ok { + t.Fatal("expected nil index to return no match") + } + }) +}