[IDENTITY-7450] Support Global Key - Store & Use

internal/api-key/command.go

@@ -32,7 +32,7 @@ const (
 	apiKeyNotValidForClusterSuggestions = "Specify the cluster this API key belongs to using the `--resource` flag. Alternatively, first execute the `confluent kafka cluster use` command to set the context to the proper cluster for this key and retry the `confluent api-key store` command."
 	apiKeyUseFailedErrorMsg             = "unable to set active API key"
 	apiKeyUseFailedSuggestions          = "If you did not create this API key with the CLI or created it on another computer, you must first store the API key and secret locally with `confluent api-key store %s <secret>`."
-	nonKafkaNotImplementedErrorMsg      = "functionality not yet available for non-Kafka cluster resources"
+	nonKafkaNotImplementedErrorMsg      = "functionality not yet available for resources other than Kafka clusters and Global API keys"
 	refuseToOverrideSecretSuggestions   = "If you would like to override the existing secret stored for API key \"%s\", use the `--force` flag."
 	unableToStoreApiKeyErrorMsg         = "unable to store API key locally: %w"
 )

internal/api-key/command_create.go

@@ -80,7 +80,7 @@
 	return cmd
 }
 
 func (c *command) create(cmd *cobra.Command, _ []string) error {
 	c.setKeyStoreIfNil()
 
 	description, err := cmd.Flags().GetString("description")
@@ -171,24 +171,41 @@
 		return err
 	}
 
-	if resourceType == resource.KafkaCluster {
+	switch resourceType {
+	case resource.KafkaCluster:
 		if err := c.keystore.StoreAPIKey(c.V2Client, userKey, resourceId); err != nil {
 			return fmt.Errorf(unableToStoreApiKeyErrorMsg, err)
 		}
+	case resource.Global:
+		// Global keys' secrets are irretrievable after creation, so always store them locally — same
+		// rationale as Kafka cluster keys. The user can still copy the printed secret if they want.
+		if err := c.keystore.StoreGlobalAPIKey(userKey); err != nil {
+			return fmt.Errorf(unableToStoreApiKeyErrorMsg, err)
+		}
 	}
 
 	use, err := cmd.Flags().GetBool("use")
 	if err != nil {
 		return err
 	}
 	if use {
-		if resourceType != resource.KafkaCluster {
+		switch resourceType {
+		case resource.KafkaCluster:
+			if err := c.useAPIKey(userKey.Key, resourceId); err != nil {
+				return errors.NewWrapErrorWithSuggestions(err, apiKeyUseFailedErrorMsg, fmt.Sprintf(apiKeyUseFailedSuggestions, userKey.Key))
+			}
+			output.Printf(c.Config.EnableColor, useAPIKeyMsg, userKey.Key)
+		case resource.Global:
+			if err := c.Context.SetActiveGlobalAPIKey(userKey.Key); err != nil {
+				return errors.NewWrapErrorWithSuggestions(err, apiKeyUseFailedErrorMsg, fmt.Sprintf(apiKeyUseFailedSuggestions, userKey.Key))
+			}
+			if err := c.Config.Save(); err != nil {
+				return err
+			}
+			output.Printf(c.Config.EnableColor, useGlobalAPIKeyMsg, userKey.Key)
+		default:
 			return fmt.Errorf("`--use` set but ineffective: %s", nonKafkaNotImplementedErrorMsg)
 		}
-		if err := c.useAPIKey(userKey.Key, resourceId); err != nil {
-			return errors.NewWrapErrorWithSuggestions(err, apiKeyUseFailedErrorMsg, fmt.Sprintf(apiKeyUseFailedSuggestions, userKey.Key))
-		}
-		output.Printf(c.Config.EnableColor, useAPIKeyMsg, userKey.Key)
 	}
 
 	return nil

internal/api-key/command_store.go

@@ -47,15 +47,51 @@
 	return cmd
 }
 
 func (c *command) store(cmd *cobra.Command, args []string) error {
 	c.setKeyStoreIfNil()
 
+	key := args[0]
+	secret := args[1]
+
+	force, err := cmd.Flags().GetBool("force")
+	if err != nil {
+		return err
+	}
+
+	// Check if API key exists server-side
+	apiKey, httpResp, err := c.V2Client.GetApiKey(key)
+	if err != nil {
+		return errors.CatchApiKeyForbiddenAccessError(err, getOperation, httpResp)
+	}
+
+	resourceType, clusterId, _, resolveErr := c.resolveResourceId(cmd, c.V2Client)
+	isGlobalKey := apiKey.GetSpec().Resource.GetKind() == "Global"
+
+	// Detect Global API keys by either the explicit --resource global flag or the server-side resource Kind.
+	// Global keys are org-scoped and stored separately from cluster-scoped keys.
+	if resourceType == resource.Global || isGlobalKey {
+		if resourceType == resource.Global && !isGlobalKey {
+			return errors.NewErrorWithSuggestions(
+				fmt.Sprintf("API key %q is not a Global API key", key),
+				"Omit `--resource global`, or pass the correct resource ID.",
+			)
+		}
+		if isGlobalKey && resourceType != "" && resourceType != resource.Global {
+			return errors.NewErrorWithSuggestions(
+				fmt.Sprintf("API key %q is a Global API key, but --resource was set to %q", key, resourceType),
+				"Re-run with `--resource global`, or omit `--resource` to auto-detect.",
+			)
+		}
+		return c.storeGlobal(key, secret, force)
+	}
+
 	var cluster *config.KafkaClusterConfig
 
 	// Attempt to get cluster from --resource flag if set; if that doesn't work,
-	// attempt to fall back to the currently active Kafka cluster
-	resourceType, clusterId, _, err := c.resolveResourceId(cmd, c.V2Client)
-	if err == nil && clusterId != "" {
+	// attempt to fall back to the currently active Kafka cluster.
+	// Preserve historical behavior: if --resource resolution failed, silently fall through to the
+	// active cluster (the cluster/key-mismatch check below will surface real problems).
+	if resolveErr == nil && clusterId != "" {
 		if resourceType != resource.KafkaCluster {
 			return errors.New(nonKafkaNotImplementedErrorMsg)
 		}
@@ -74,20 +110,6 @@
 		}
 	}
 
-	key := args[0]
-	secret := args[1]
-
-	force, err := cmd.Flags().GetBool("force")
-	if err != nil {
-		return err
-	}
-
-	// Check if API key exists server-side
-	apiKey, httpResp, err := c.V2Client.GetApiKey(key)
-	if err != nil {
-		return errors.CatchApiKeyForbiddenAccessError(err, getOperation, httpResp)
-	}
-
 	apiKeyIsValidForTargetCluster := cluster.GetId() != "" && cluster.GetId() == apiKey.GetSpec().Resource.GetId()
 
 	if !apiKeyIsValidForTargetCluster {
@@ -114,3 +136,17 @@
 	output.ErrPrintf(c.Config.EnableColor, "Stored secret for API key \"%s\".\n", key)
 	return nil
 }
+
+func (c *command) storeGlobal(key, secret string, force bool) error {
+	if c.keystore.HasGlobalAPIKey(key) && !force {
+		return errors.NewErrorWithSuggestions(
+			fmt.Sprintf(`refusing to overwrite existing secret for API Key "%s"`, key),
+			fmt.Sprintf(refuseToOverrideSecretSuggestions, key),
+		)
+	}
+	if err := c.keystore.StoreGlobalAPIKey(&config.APIKeyPair{Key: key, Secret: secret}); err != nil {
+		return fmt.Errorf(unableToStoreApiKeyErrorMsg, err)
+	}
+	output.ErrPrintf(c.Config.EnableColor, "Stored secret for Global API key \"%s\".\n", key)
+	return nil
+}

internal/api-key/command_use.go

@@ -12,7 +12,10 @@ import (
 	"github.com/confluentinc/cli/v4/pkg/resource"
 )
 
-const useAPIKeyMsg = "Using API Key \"%s\".\n"
+const (
+	useAPIKeyMsg       = "Using API Key \"%s\".\n"
+	useGlobalAPIKeyMsg = "Using Global API Key \"%s\".\n"
+)
 
 func (c *command) newUseCommand() *cobra.Command {
 	cmd := &cobra.Command{
@@ -34,6 +37,21 @@ func (c *command) newUseCommand() *cobra.Command {
 func (c *command) use(cmd *cobra.Command, args []string) error {
 	c.setKeyStoreIfNil()
 
+	apiKey := args[0]
+
+	// Global keys are stored on the Context, not on a specific Kafka cluster. Check there first so
+	// `confluent api-key use <global-key>` without --resource works seamlessly.
+	if !cmd.Flags().Changed("resource") && c.Context.HasGlobalAPIKey(apiKey) {
+		if err := c.Context.SetActiveGlobalAPIKey(apiKey); err != nil {
+			return errors.NewWrapErrorWithSuggestions(err, apiKeyUseFailedErrorMsg, fmt.Sprintf(apiKeyUseFailedSuggestions, apiKey))
+		}
+		if err := c.Config.Save(); err != nil {
+			return err
+		}
+		output.Printf(c.Config.EnableColor, useGlobalAPIKeyMsg, apiKey)
+		return nil
+	}
+
 	var clusterId string
 
 	if cmd.Flags().Changed("resource") {
@@ -46,20 +64,20 @@ func (c *command) use(cmd *cobra.Command, args []string) error {
 		}
 		clusterId = resourceId
 	} else {
-		clusterId = c.Context.KafkaClusterContext.FindApiKeyClusterId(args[0])
+		clusterId = c.Context.KafkaClusterContext.FindApiKeyClusterId(apiKey)
 		if clusterId == "" {
 			return errors.NewErrorWithSuggestions(
-				fmt.Sprintf(`API key "%s" and associated Kafka cluster are not stored in local CLI state`, args[0]),
-				fmt.Sprintf(apiKeyUseFailedSuggestions, args[0]),
+				fmt.Sprintf(`API key "%s" and associated Kafka cluster are not stored in local CLI state`, apiKey),
+				fmt.Sprintf(apiKeyUseFailedSuggestions, apiKey),
 			)
 		}
 	}
 
-	if err := c.useAPIKey(args[0], clusterId); err != nil {
-		return errors.NewWrapErrorWithSuggestions(err, apiKeyUseFailedErrorMsg, fmt.Sprintf(apiKeyUseFailedSuggestions, args[0]))
+	if err := c.useAPIKey(apiKey, clusterId); err != nil {
+		return errors.NewWrapErrorWithSuggestions(err, apiKeyUseFailedErrorMsg, fmt.Sprintf(apiKeyUseFailedSuggestions, apiKey))
 	}
 
-	output.Printf(c.Config.EnableColor, useAPIKeyMsg, args[0])
+	output.Printf(c.Config.EnableColor, useAPIKeyMsg, apiKey)
 	return nil
 }
 

internal/asyncapi/command_export.go

@@ -462,7 +462,7 @@
 	return bindings, nil
 }
 
 func (c *command) getClusterDetails(details *accountDetails, flags *flags, cmd *cobra.Command) error {
 	cluster, err := pkafka.GetClusterForCommand(c.V2Client, c.Context)
 	if err != nil {
 		return fmt.Errorf(`failed to find Kafka cluster: %w`, err)
@@ -491,12 +491,20 @@
 		}
 		clusterCreds = cluster.APIKeys[flags.kafkaApiKey]
 	} else {
-		clusterCreds = cluster.APIKeys[cluster.APIKey]
+		// No explicit --kafka-api-key: resolve the cluster-scoped active key, falling back to the
+		// active Global API key (see Context.ResolveKafkaAPIKey).
+		apiKey, apiSecret, err := c.Context.ResolveKafkaAPIKey(cluster)
+		if err != nil {
+			return err
+		}
+		if apiKey != "" {
+			clusterCreds = &config.APIKeyPair{Key: apiKey, Secret: apiSecret}
+		}
 	}
 	if clusterCreds == nil {
 		return errors.NewErrorWithSuggestions(
 			"API key not set for the Kafka cluster",
-			"Set an API key pair for the Kafka cluster using `confluent api-key create --resource <cluster-id>` and then use it with `--kafka-api-key`.",
+			"Set an API key pair for the Kafka cluster using `confluent api-key create --resource <cluster-id>` and then use it with `--kafka-api-key`, or set an active Global API key with `confluent api-key use`.",
 		)
 	}
 

internal/kafka/command_clientconfig_create.go

@@ -179,7 +179,7 @@ func (c *clientConfigCommand) setKafkaCluster(cmd *cobra.Command, configFile str
 		return "", err
 	}
 
-	if err := addApiKeyToCluster(cmd, kafkaCluster); err != nil {
+	if err := addApiKeyToCluster(cmd, c.Context, kafkaCluster); err != nil {
 		return "", err
 	}
 
@@ -200,11 +200,16 @@ func (c *clientConfigCommand) setKafkaCluster(cmd *cobra.Command, configFile str
 		}
 	}
 
+	apiKey, apiSecret, err := c.Context.ResolveKafkaAPIKey(kafkaCluster)
+	if err != nil {
+		return "", err
+	}
+
 	// replace BROKER_ENDPOINT, CLUSTER_API_KEY, and CLUSTER_API_SECRET templates
 	configFile = replaceTemplates(configFile, map[string]string{
 		brokerEndpointTemplate:   kafkaCluster.Bootstrap,
-		clusterApiKeyTemplate:    kafkaCluster.APIKey,
-		clusterApiSecretTemplate: kafkaCluster.GetApiSecret(),
+		clusterApiKeyTemplate:    apiKey,
+		clusterApiSecretTemplate: apiSecret,
 	})
 	return configFile, nil
 }
@@ -283,7 +288,7 @@ func (c *clientConfigCommand) getSchemaRegistryCluster() (*srcmv3.SrcmV3Cluster,
 }
 
 func (c *clientConfigCommand) validateKafkaCredentials(kafkaCluster *config.KafkaClusterConfig) error {
-	configMap, err := getCommonConfig(kafkaCluster, c.clientId)
+	configMap, err := getCommonConfig(c.Context, kafkaCluster, c.clientId)
 	if err != nil {
 		return err
 	}

internal/kafka/command_topic.go

@@ -173,7 +173,7 @@ func (c *command) prepareAnonymousContext(cmd *cobra.Command) error {
 	return nil
 }
 
-func addApiKeyToCluster(cmd *cobra.Command, cluster *config.KafkaClusterConfig) error {
+func addApiKeyToCluster(cmd *cobra.Command, ctx *config.Context, cluster *config.KafkaClusterConfig) error {
 	apiKey, err := cmd.Flags().GetString("api-key")
 	if err != nil {
 		return err
@@ -192,6 +192,12 @@ func addApiKeyToCluster(cmd *cobra.Command, cluster *config.KafkaClusterConfig)
 		}
 	}
 
+	// If the cluster has no scoped API key, accept an active Global API key as a fallback. The Kafka
+	// REST/admin path will resolve credentials via Context.ResolveKafkaAPIKey() downstream.
+	if cluster.APIKey == "" && ctx.GetActiveGlobalAPIKey() != "" {
+		return nil
+	}
+
 	if cluster.APIKey == "" {
 		return &errors.UnspecifiedAPIKeyError{ClusterID: cluster.ID}
 	}

internal/kafka/command_topic_consume.go

@@ -111,7 +111,7 @@
 	return c.consumeOnPrem(cmd, args)
 }
 
 func (c *command) consumeCloud(cmd *cobra.Command, args []string) error {
 	topic := args[0]
 
 	cluster, err := kafka.GetClusterForCommand(c.V2Client, c.Context)
@@ -127,7 +127,7 @@
 		cluster.Bootstrap = bootstrap
 	}
 
-	if err := addApiKeyToCluster(cmd, cluster); err != nil {
+	if err := addApiKeyToCluster(cmd, c.Context, cluster); err != nil {
 		return err
 	}
 
@@ -193,7 +193,7 @@
 		}
 	}
 
-	consumer, err := newConsumer(group, cluster, c.clientID, configFile, config)
+	consumer, err := newConsumer(group, c.Context, cluster, c.clientID, configFile, config)
 	if err != nil {
 		return fmt.Errorf(errors.FailedToCreateConsumerErrorMsg, err)
 	}

internal/kafka/command_topic_produce.go

@@ -127,7 +127,7 @@
 	return c.produceOnPrem(cmd, args)
 }
 
 func (c *command) produceCloud(cmd *cobra.Command, args []string) error {
 	topic := args[0]
 
 	cluster, err := kafka.GetClusterForCommand(c.V2Client, c.Context)
@@ -143,7 +143,7 @@
 		cluster.Bootstrap = bootstrap
 	}
 
-	if err := addApiKeyToCluster(cmd, cluster); err != nil {
+	if err := addApiKeyToCluster(cmd, c.Context, cluster); err != nil {
 		return err
 	}
 
@@ -184,7 +184,7 @@
 		return err
 	}
 
-	producer, err := newProducer(cluster, c.clientID, configFile, config)
+	producer, err := newProducer(c.Context, cluster, c.clientID, configFile, config)
 	if err != nil {
 		return fmt.Errorf(errors.FailedToCreateProducerErrorMsg, err)
 	}

internal/kafka/confluent_kafka.go

@@ -161,17 +161,17 @@ func (c *command) retrieveUnsecuredToken(e ckgo.OAuthBearerTokenRefresh) (ckgo.O
 	return oauthBearerToken, nil
 }
 
-func newProducer(kafka *config.KafkaClusterConfig, clientID, configPath string, configStrings []string) (*ckgo.Producer, error) {
-	configMap, err := getProducerConfigMap(kafka, clientID)
+func newProducer(ctx *config.Context, kafka *config.KafkaClusterConfig, clientID, configPath string, configStrings []string) (*ckgo.Producer, error) {
+	configMap, err := getProducerConfigMap(ctx, kafka, clientID)
 	if err != nil {
 		return nil, fmt.Errorf(errors.FailedToGetConfigurationErrorMsg, err)
 	}
 
 	return newProducerWithOverwrittenConfigs(configMap, configPath, configStrings)
 }
 
-func newConsumer(group string, kafka *config.KafkaClusterConfig, clientID, configPath string, configStrings []string) (*ckgo.Consumer, error) {
-	configMap, err := getConsumerConfigMap(group, kafka, clientID)
+func newConsumer(group string, ctx *config.Context, kafka *config.KafkaClusterConfig, clientID, configPath string, configStrings []string) (*ckgo.Consumer, error) {
+	configMap, err := getConsumerConfigMap(group, ctx, kafka, clientID)
 	if err != nil {
 		return nil, fmt.Errorf(errors.FailedToGetConfigurationErrorMsg, err)
 	}

internal/kafka/confluent_kafka_configs.go

@@ -27,26 +27,31 @@ type PartitionFilter struct {
 	Index   int32
 }
 
-func getCommonConfig(kafka *config.KafkaClusterConfig, clientId string) (*ckgo.ConfigMap, error) {
+func getCommonConfig(ctx *config.Context, kafka *config.KafkaClusterConfig, clientId string) (*ckgo.ConfigMap, error) {
 	if err := kafka.DecryptAPIKeys(); err != nil {
 		return nil, err
 	}
 
+	apiKey, apiSecret, err := ctx.ResolveKafkaAPIKey(kafka)
+	if err != nil {
+		return nil, err
+	}
+
 	configMap := &ckgo.ConfigMap{
 		"security.protocol":                     "SASL_SSL",
 		"sasl.mechanism":                        "PLAIN",
 		"ssl.endpoint.identification.algorithm": "https",
 		"client.id":                             clientId,
 		"bootstrap.servers":                     kafka.Bootstrap,
-		"sasl.username":                         kafka.APIKey,
-		"sasl.password":                         kafka.GetApiSecret(),
+		"sasl.username":                         apiKey,
+		"sasl.password":                         apiSecret,
 	}
 
 	return configMap, nil
 }
 
-func getProducerConfigMap(kafka *config.KafkaClusterConfig, clientID string) (*ckgo.ConfigMap, error) {
-	configMap, err := getCommonConfig(kafka, clientID)
+func getProducerConfigMap(ctx *config.Context, kafka *config.KafkaClusterConfig, clientID string) (*ckgo.ConfigMap, error) {
+	configMap, err := getCommonConfig(ctx, kafka, clientID)
 	if err != nil {
 		return nil, err
 	}
@@ -62,8 +67,8 @@ func getProducerConfigMap(kafka *config.KafkaClusterConfig, clientID string) (*c
 	return configMap, nil
 }
 
-func getConsumerConfigMap(group string, kafka *config.KafkaClusterConfig, clientID string) (*ckgo.ConfigMap, error) {
-	configMap, err := getCommonConfig(kafka, clientID)
+func getConsumerConfigMap(group string, ctx *config.Context, kafka *config.KafkaClusterConfig, clientID string) (*ckgo.ConfigMap, error) {
+	configMap, err := getCommonConfig(ctx, kafka, clientID)
 	if err != nil {
 		return nil, err
 	}