JWT Authentication with service intentions: xds package update

agent/consul/fsm/fsm.go

@@ -426,6 +426,12 @@ func (c *FSM) registerStreamSnapshotHandlers() {
 		return c.State().SamenessGroupSnapshot(req, buf)
 	}, true)
 	panicIfErr(err)
+
+	err = c.deps.Publisher.RegisterHandler(state.EventTopicJWTProvider, func(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
+		return c.State().JWTProviderSnapshot(req, buf)
+	}, true)
+
+	panicIfErr(err)
 }
 
 func panicIfErr(err error) {

agent/consul/state/config_entry_events.go

@@ -27,6 +27,7 @@ var configEntryKindToTopic = map[string]stream.Topic{
 	structs.BoundAPIGateway:   EventTopicBoundAPIGateway,
 	structs.RateLimitIPConfig: EventTopicIPRateLimit,
 	structs.SamenessGroup:     EventTopicSamenessGroup,
+	structs.JWTProvider:       EventTopicJWTProvider,
 }
 
 // EventSubjectConfigEntry is a stream.Subject used to route and receive events
@@ -169,6 +170,12 @@ func (s *Store) SamenessGroupSnapshot(req stream.SubscribeRequest, buf stream.Sn
 	return s.configEntrySnapshot(structs.SamenessGroup, req, buf)
 }
 
+// JWTProviderSnapshot is a stream.SnapshotFunc that returns a snapshot of
+// jwt-provider config entries.
+func (s *Store) JWTProviderSnapshot(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
+	return s.configEntrySnapshot(structs.JWTProvider, req, buf)
+}
+
 func (s *Store) configEntrySnapshot(kind string, req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
 	var (
 		idx     uint64

agent/consul/state/events.go

@@ -45,7 +45,7 @@ func PBToStreamSubscribeRequest(req *pbsubscribe.SubscribeRequest, entMeta acl.E
 			}
 		case EventTopicMeshConfig, EventTopicServiceResolver, EventTopicIngressGateway,
 			EventTopicServiceIntentions, EventTopicServiceDefaults, EventTopicAPIGateway,
-			EventTopicTCPRoute, EventTopicHTTPRoute, EventTopicInlineCertificate,
+			EventTopicTCPRoute, EventTopicHTTPRoute, EventTopicJWTProvider, EventTopicInlineCertificate,
 			EventTopicBoundAPIGateway, EventTopicSamenessGroup:
 			subject = EventSubjectConfigEntry{
 				Name:           named.Key,

agent/consul/state/memdb.go

@@ -211,6 +211,7 @@ var (
 	EventTopicBoundAPIGateway      = pbsubscribe.Topic_BoundAPIGateway
 	EventTopicIPRateLimit          = pbsubscribe.Topic_IPRateLimit
 	EventTopicSamenessGroup        = pbsubscribe.Topic_SamenessGroup
+	EventTopicJWTProvider          = pbsubscribe.Topic_JWTProvider
 )
 
 func processDBChanges(tx ReadTxn, changes Changes) ([]stream.Event, error) {

agent/proxycfg-glue/config_entry.go

@@ -75,6 +75,8 @@ func newConfigEntryRequest(req *structs.ConfigEntryQuery, deps ServerDataSourceD
 		topic = pbsubscribe.Topic_BoundAPIGateway
 	case structs.RateLimitIPConfig:
 		topic = pbsubscribe.Topic_IPRateLimit
+	case structs.JWTProvider:
+		topic = pbsubscribe.Topic_JWTProvider
 	default:
 		return nil, fmt.Errorf("cannot map config entry kind: %q to a topic", req.Kind)
 	}

agent/proxycfg/connect_proxy.go

@@ -90,6 +90,20 @@ func (s *handlerConnectProxy) initialize(ctx context.Context) (ConfigSnapshot, e
 		return snap, err
 	}
 
+	// Watch for JWT provider updates.
+	// While we could optimize by only watching providers referenced by intentions,
+	// this should be okay because we expect few JWT providers and infrequent JWT
+	// provider updates.
+	err = s.dataSources.ConfigEntryList.Notify(ctx, &structs.ConfigEntryQuery{
+		Kind:           structs.JWTProvider,
+		Datacenter:     s.source.Datacenter,
+		QueryOptions:   structs.QueryOptions{Token: s.token},
+		EnterpriseMeta: *structs.DefaultEnterpriseMetaInPartition(s.proxyID.PartitionOrDefault()),
+	}, jwtProviderID, s.ch)
+	if err != nil {
+		return snap, err
+	}
+
 	// Get information about the entire service mesh.
 	err = s.dataSources.ConfigEntry.Notify(ctx, &structs.ConfigEntryQuery{
 		Kind:           structs.MeshConfig,
@@ -322,6 +336,23 @@ func (s *handlerConnectProxy) handleUpdate(ctx context.Context, u UpdateEvent, s
 		snap.ConnectProxy.Intentions = resp
 		snap.ConnectProxy.IntentionsSet = true
 
+	case u.CorrelationID == jwtProviderID:
+		resp, ok := u.Result.(*structs.IndexedConfigEntries)
+
+		if !ok {
+			return fmt.Errorf("invalid type for response: %T", u.Result)
+		}
+
+		providers := make(map[string]*structs.JWTProviderConfigEntry, len(resp.Entries))
+		for _, entry := range resp.Entries {
+			jwtEntry, ok := entry.(*structs.JWTProviderConfigEntry)
+			if !ok {
+				return fmt.Errorf("invalid type for response: %T", entry)
+			}
+			providers[jwtEntry.Name] = jwtEntry
+		}
+
+		snap.JWTProviders = providers
 	case u.CorrelationID == peeredUpstreamsID:
 		resp, ok := u.Result.(*structs.IndexedPeeredServiceList)
 		if !ok {

agent/proxycfg/proxycfg.deepcopy.go

@@ -29,6 +29,77 @@ func (o *ConfigSnapshot) DeepCopy() *ConfigSnapshot {
 		retV := o.Proxy.DeepCopy()
 		cp.Proxy = *retV
 	}
+	if o.JWTProviders != nil {
+		cp.JWTProviders = make(map[string]*structs.JWTProviderConfigEntry, len(o.JWTProviders))
+		for k2, v2 := range o.JWTProviders {
+			var cp_JWTProviders_v2 *structs.JWTProviderConfigEntry
+			if v2 != nil {
+				cp_JWTProviders_v2 = new(structs.JWTProviderConfigEntry)
+				*cp_JWTProviders_v2 = *v2
+				if v2.JSONWebKeySet != nil {
+					cp_JWTProviders_v2.JSONWebKeySet = new(structs.JSONWebKeySet)
+					*cp_JWTProviders_v2.JSONWebKeySet = *v2.JSONWebKeySet
+					if v2.JSONWebKeySet.Local != nil {
+						cp_JWTProviders_v2.JSONWebKeySet.Local = new(structs.LocalJWKS)
+						*cp_JWTProviders_v2.JSONWebKeySet.Local = *v2.JSONWebKeySet.Local
+					}
+					if v2.JSONWebKeySet.Remote != nil {
+						cp_JWTProviders_v2.JSONWebKeySet.Remote = new(structs.RemoteJWKS)
+						*cp_JWTProviders_v2.JSONWebKeySet.Remote = *v2.JSONWebKeySet.Remote
+						if v2.JSONWebKeySet.Remote.RetryPolicy != nil {
+							cp_JWTProviders_v2.JSONWebKeySet.Remote.RetryPolicy = new(structs.JWKSRetryPolicy)
+							*cp_JWTProviders_v2.JSONWebKeySet.Remote.RetryPolicy = *v2.JSONWebKeySet.Remote.RetryPolicy
+							if v2.JSONWebKeySet.Remote.RetryPolicy.RetryPolicyBackOff != nil {
+								cp_JWTProviders_v2.JSONWebKeySet.Remote.RetryPolicy.RetryPolicyBackOff = new(structs.RetryPolicyBackOff)
+								*cp_JWTProviders_v2.JSONWebKeySet.Remote.RetryPolicy.RetryPolicyBackOff = *v2.JSONWebKeySet.Remote.RetryPolicy.RetryPolicyBackOff
+							}
+						}
+					}
+				}
+				if v2.Audiences != nil {
+					cp_JWTProviders_v2.Audiences = make([]string, len(v2.Audiences))
+					copy(cp_JWTProviders_v2.Audiences, v2.Audiences)
+				}
+				if v2.Locations != nil {
+					cp_JWTProviders_v2.Locations = make([]*structs.JWTLocation, len(v2.Locations))
+					copy(cp_JWTProviders_v2.Locations, v2.Locations)
+					for i5 := range v2.Locations {
+						if v2.Locations[i5] != nil {
+							cp_JWTProviders_v2.Locations[i5] = new(structs.JWTLocation)
+							*cp_JWTProviders_v2.Locations[i5] = *v2.Locations[i5]
+							if v2.Locations[i5].Header != nil {
+								cp_JWTProviders_v2.Locations[i5].Header = new(structs.JWTLocationHeader)
+								*cp_JWTProviders_v2.Locations[i5].Header = *v2.Locations[i5].Header
+							}
+							if v2.Locations[i5].QueryParam != nil {
+								cp_JWTProviders_v2.Locations[i5].QueryParam = new(structs.JWTLocationQueryParam)
+								*cp_JWTProviders_v2.Locations[i5].QueryParam = *v2.Locations[i5].QueryParam
+							}
+							if v2.Locations[i5].Cookie != nil {
+								cp_JWTProviders_v2.Locations[i5].Cookie = new(structs.JWTLocationCookie)
+								*cp_JWTProviders_v2.Locations[i5].Cookie = *v2.Locations[i5].Cookie
+							}
+						}
+					}
+				}
+				if v2.Forwarding != nil {
+					cp_JWTProviders_v2.Forwarding = new(structs.JWTForwardingConfig)
+					*cp_JWTProviders_v2.Forwarding = *v2.Forwarding
+				}
+				if v2.CacheConfig != nil {
+					cp_JWTProviders_v2.CacheConfig = new(structs.JWTCacheConfig)
+					*cp_JWTProviders_v2.CacheConfig = *v2.CacheConfig
+				}
+				if v2.Meta != nil {
+					cp_JWTProviders_v2.Meta = make(map[string]string, len(v2.Meta))
+					for k5, v5 := range v2.Meta {
+						cp_JWTProviders_v2.Meta[k5] = v5
+					}
+				}
+			}
+			cp.JWTProviders[k2] = cp_JWTProviders_v2
+		}
+	}
 	if o.Roots != nil {
 		cp.Roots = o.Roots.DeepCopy()
 	}

agent/proxycfg/snapshot.go

@@ -1020,6 +1020,7 @@ type ConfigSnapshot struct {
 	Datacenter            string
 	IntentionDefaultAllow bool
 	Locality              GatewayKey
+	JWTProviders          map[string]*structs.JWTProviderConfigEntry
 
 	ServerSNIFn ServerSNIFunc
 	Roots       *structs.IndexedCARoots

agent/proxycfg/state.go

@@ -44,6 +44,7 @@ const (
 	serviceResolverIDPrefix            = "service-resolver:"
 	serviceIntentionsIDPrefix          = "service-intentions:"
 	intentionUpstreamsID               = "intention-upstreams"
+	jwtProviderID                      = "jwt-provider"
 	peerServersWatchID                 = "peer-servers"
 	peeredUpstreamsID                  = "peered-upstreams"
 	intentionUpstreamsDestinationID    = "intention-upstreams-destination"