Status

Current state: "Under Discussion"

Discussion thread: here

JIRA: KAFKA-18005

Motivation

Currently, when describing config for a resource, we'll get `null` if the config is a sensitive config, ex: "ssl.keystore.certificate.chain", "ssl.keystore.password". And when describing configs with them it'll always return something like this:

> bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 2 --describe
Dynamic configs for broker 2 are:
  listener.name.myssl.ssl.keystore.key=null sensitive=true synonyms={DYNAMIC_BROKER_CONFIG:listener.name.myssl.
=null}

Note: The value of `listener.name.myssl.ssl.keystore.key` should be a non-null value , but returning null for security consideration.

It makes sense to not expose sensitive data to the users, but returning null will let the users/operators hard to identify if the config is up-to-date.

Take the example in KIP-412:

Consider the dynamic configurations ssl.keystore.location, ssl.keystore.password and ssl.keystore.type are stored in "vault" and their values are resolved using a VaultConfigProvider. 

...
ssl.keystore.location=${vault:/path/to/variables.properties:ssl.keystore.location}

ssl.keystore.password=${vault:/path/to/variables.properties:ssl.keystore.password}

ssl.keystore.type=${vault:/path/to/variables.properties:ssl.keystore.type}
 
config.providers=vault
config.providers.file.class=org.apache.kafka.connect.configs.VaultConfigProvider

    1. The user will update the ssl.keystore.location, ssl.keystore.password, ssl.keystore.type in the vault.
    2. After the updates are complete he will send a adminClient request to the broker to notify that configs are updated.
    3. Once the Broker receives a alteredConfig request it will invoke the get function in VaultConfigProvider.
    4. The VaultConfigProvider will fetch the actual values for ssl.keystore.location, ssl.keystore.password, ssl.keystore.type from the vault.
    5. The broker will validate these configs and apply the changes.


In the step (2), the operator needs a way to know if the current state of these configs. With current design, the operator will never know it, and blindly run the alter config multiple times, or worse, the operator thought it is already updated and skipped this update, and cause the broker connection failure.


In the Kubernetes world, one of the purpose for the operator is:

Kubernetes Operators manage application logic and are part of the Kubernetes control plane. As such, they are controllers that execute loops to check the actual state of the cluster and the desired state, acting to reconcile them when the two states are drifting apart. 

from the CNCF blog post.


Without the knowledge of the actual state of the resource (kafka pod), the operators cannot reconcile them as expected. It would be great if the broker/controller can return some metadata of these sensitive configs, like "last modified timestamp", to allow the operators have a way to get the current state of these configs.

Public Interfaces

There will be an additional "internal config" added for each dynamic confidential configurations. The definition of dynamic confidential configurations is:

  1. This config is PASSWORD type.
  2. This config is a dynamically changable config (per-broker).


The table lists current configs that fit the above 2 criteria:

existing confidential changable confignewly added internal configsdefault valueupdate mode
sasl.jaas.configsasl.jaas.config.timestamp-1per-broker
ssl.keystore.keyssl.keystore.key.timestamp-1per-broker
ssl.keystore.passwordssl.keystore.password.timestamp-1per-broker
ssl.keystore.certificate.chainssl.keystore.certificate.chain.timestamp-1per-broker
ssl.truststore.certificatesssl.truststore.certificates.timestamp-1per-broker
ssl.truststore.passwordssl.truststore.password.timestamp-1per-broker
ssl.key.passwordssl.key.password.timestamp-1per-broker

For these configs, there will be an "timestamp" config added. Ex: "ssl.keystore.key" will add "ssl.keystore.key.timestamp" as an internal config with default value -1. It has the same update mode (per-broker). The internal config means it is not publicly visible. And users cannot alter values via admin API or broker properties. Besides, there will be a new ConfigDef.Type added: METADATA, to allow brokers restrict users update these configs.

Proposed Changes

ConfigDef type:

The METADATA configDef type will be added:

    /**
     * The type for a configuration value
     */
    public enum Type {
        /**
         * Used for boolean values. Values can be provided as a Boolean object or as a String with values
         * <code>true</code> or <code>false</code> (this is not case-sensitive), otherwise a {@link ConfigException} is
         * thrown.
         */
        BOOLEAN,
        ...

        /**
         * Used for string values containing sensitive data such as a password or key. The values of configurations with
         * of this type are not included in logs and instead replaced with "[hidden]". Values must be provided as a
         * String object, otherwise a {@link ConfigException} is thrown.
         */
        PASSWORD,
    
        // newly added below
        /**
         * Used for string values containing the metadata of the according sensitive field such as a password or key. 
         * Values must be provided as a String object, otherwise a {@link ConfigException} is thrown.
         */
        METADATA;


Restriction for METADATA type configs:

(1) alter

These METADATA type configs cannot be updated via admin API, otherwise, exception will be thrown.

> bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 2 --alter --add-config listener.name.ssl.ssl.keystore.key.timestamp=123

Error while executing config command with args '--bootstrap-server localhost:9092 --entity-type brokers --entity-name 2 --alter --add-config listener.name.ssl.ssl.keystore.key.timestamp=123'
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.InvalidRequestException: Cannot update metadata type config:HashMap(listener.name.ssl.ssl.keystore.key.timestamp -> 123)


Metadata type config update:

The metadata type config can only be updated when the mapping config getting updated. For example:

> bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 2 --alter --add-config listener.name.ssl.ssl.key.password=000000

Completed updating config for broker 2.


Since the config: listener.name.ssl.ssl.key.password  is a confidential chanagable config, in the controller we'll also add one more entry to the metadata log to update listener.name.ssl.ssl.key.password.timestamp metadata config, and set the value as current timestamp in milliseconds. When checking the controller log, you'll see these lines:


[2024-11-15 14:26:10,963] INFO [QuorumController id=1] Replayed ConfigRecord for ConfigResource(type=BROKER, name='2') which set configuration listener.name.ssl.ssl.key.password to null (org.apache.kafka.controller.ConfigurationControlManager)

// added by controller 
[2024-11-15 14:26:10,964] INFO [QuorumController id=1] Replayed ConfigRecord for ConfigResource(type=BROKER, name='2') which set configuration listener.name.ssl.ssl.key.password.timestamp to 1731651970963 (org.apache.kafka.controller.ConfigurationControlManager)


(2) describe

Because the metadata configs are internal configs, it cannot be shown as an config entry when describing configs. Instead, the metadata value will be merged into the mapping confidential config as "lastUpdatedTimestamp" field. When describing the config, we'll see it:


> bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 2  --describe --all

// non-sensitive field will not show "lastUpdatedTimestamp" field as before
advertised.listeners=PLAINTEXT://localhost:9092 sensitive=false synonyms={STATIC_BROKER_CONFIG:advertised.listeners=PLAINTEXT://localhost:9092}
...
// sensitive field will show "lastUpdatedTimestamp" field. If it is not changed before, it'll be -1.
listener.name.sslnochange.ssl.key.password=null sensitive=true synonyms={STATIC_BROKER_CONFIG:listener.name.plaintext.ssl.key.password=null, STATIC_BROKER_CONFIG:ssl.key.password=null} lastUpdatedTimestamp=-1
// If it is changed, it'll return the last updated timestamp
listener.name.ssl.ssl.key.password=null sensitive=true synonyms={STATIC_BROKER_CONFIG:listener.name.ssl.ssl.key.password=null, STATIC_BROKER_CONFIG:ssl.key.password=null} lastUpdatedTimestamp=1731651970963

Note:

(1) When the lastUpdatedTimestamp is -1, it means it has never dynamically updated before.

(2) The value of metadata configs will only be honored for DYNAMIC_BROKER_CONFIG. That is, if the value is from DYNAMIC_DEFAULT_BROKER_CONFIG or STATIC_BROKER_CONFIG (from properties file), it'll be ignored because the sensitive configs are all per-broker configs, only DYNAMIC_BROKER_CONFIG update is possible.



Thus, in DescribeConfigsResponse, we need to add one "LastUpdatedTimestamp" field to it.

DescribeConfigs

Request:

Bump the version to 5 because of DescribeConfigsResponse change.

Response:

Bump version to 5 to add "LastUpdatedTimestamp" field

{
  "apiKey": 32,
  "type": "response",
  "name": "DescribeConfigsResponse",
  // Version 1 adds ConfigSource and the synonyms.
  // Starting in version 2, on quota violation, brokers send out responses before throttling.
  // Version 4 enables flexible versions.
  // Version 5 adds "LastUpdateTimestampMs" field
  "validVersions": "0-5",
  "flexibleVersions": "5+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "Results", "type": "[]DescribeConfigsResult", "versions": "0+",
      "about": "The results for each resource.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The error code, or 0 if we were able to successfully describe the configurations." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
        "about": "The error message, or null if we were able to successfully describe the configurations." },
      { "name": "ResourceType", "type": "int8", "versions": "0+",
        "about": "The resource type." },
      { "name": "ResourceName", "type": "string", "versions": "0+",
        "about": "The resource name." },
      { "name": "Configs", "type": "[]DescribeConfigsResourceResult", "versions": "0+",
        "about": "Each listed configuration.", "fields": [
        { "name": "Name", "type": "string", "versions": "0+",
          "about": "The configuration name." },
        { "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+",
          "about": "The configuration value." },
        { "name": "ReadOnly", "type": "bool", "versions": "0+",
          "about": "True if the configuration is read-only." },
        { "name": "IsDefault", "type": "bool", "versions": "0",
          "about": "True if the configuration is not set." },
        // Note: the v0 default for this field that should be exposed to callers is
        // context-dependent. For example, if the resource is a broker, this should default to 4.
        // -1 is just a placeholder value.
        { "name": "ConfigSource", "type": "int8", "versions": "1+", "default": "-1", "ignorable": true,
          "about": "The configuration source." },
        { "name": "IsSensitive", "type": "bool", "versions": "0+",
          "about": "True if this configuration is sensitive." },
        { "name": "Synonyms", "type": "[]DescribeConfigsSynonym", "versions": "1+", "ignorable": true,
          "about": "The synonyms for this configuration key.", "fields": [
          { "name": "Name", "type": "string", "versions": "1+",
            "about": "The synonym name." },
          { "name": "Value", "type": "string", "versions": "1+", "nullableVersions": "0+",
            "about": "The synonym value." },
          { "name": "Source", "type": "int8", "versions": "1+",
            "about": "The synonym source." }
        ]},
        { "name": "ConfigType", "type": "int8", "versions": "3+", "default": "0", "ignorable": true,
          "about": "The configuration data type. Type can be one of the following values - BOOLEAN, STRING, INT, SHORT, LONG, DOUBLE, LIST, CLASS, PASSWORD" },
        { "name": "Documentation", "type": "string", "versions": "3+", "nullableVersions": "0+", "ignorable": true,
          "about": "The configuration documentation." },

        // newly added field
        { "name": "LastUpdateTimestampMs", "type": "int64", "versions": "5+", "ignorable": true,
          "about": "The last updated timestamp for this config." }
      ]}
    ]}
  ]
}


Compatibility, Deprecation, and Migration Plan

This change will not have compatibility issue.

When old client connects to the new broker, it'll use DescribeConfigs V.4.

When new client connects to the old broker, it'll also use DescribeConfigs V.4.

Test Plan

This KIP will be tested using unittest, integration tests.

Rejected Alternatives

Returning Hash(confidential value) when describing configs

We can return hash of the confidential value to the clients. It will make the implementation very simple. But it opens a door for bad users to have a way to guess the confidential value with brute force way.

Adding a new ACL/principle to be allowed to read confidential configs

We can also add a new ACL (ex: "DescribeConfidentialConfigs") or add a new principle (ex: "super.users") to allow to read the confidential configs. It is not good because if some environment grant "all" too some users, and now it'll be able to read confidential configs. The same applies to super users, it'll get some extra power after this KIP.

  • No labels