11.9 Securing the platform

*"Security design for a production system should use a THREAT MODEL that addresses security threats not just for individual components but also for the system as a whole. Threat models build an abstraction of the system and identify potential threats and the associated risks. Once the threats are evaluated, DOCUMENTED, and PRIORITIZED BASED ON RISKS, mitigation strategies must be implemented for EACH potential threat.

When assessing potential threats, it is important to consider EXTERNAL threats AS WELL AS INSIDER THREATS."*

The platform-level defenses:

• Network firewall solutions to protect the network.
• Encryption to protect physical storage.
• “Key stores, trust stores, and Kerberos keytab files that contain credentials used for authentication must be protected using filesystem permissions.”
• “Access to configuration files containing security-critical information like credentials must be restricted.”
• “Since passwords stored in clear-text in configuration files are insecure even if access is restricted, Kafka supports externalizing passwords in a secure store.”

9.1 Password protection — two mechanisms

A. Custom ConfigProvider (works for brokers and clients)

public class GpgProvider implements ConfigProvider {
  @Override public void configure(Map<String, ?> configs) {}

  @Override
  public ConfigData get(String path) {
    try {
      String passphrase = System.getenv("PASSPHRASE");        // ① from the ENV
      String data = Shell.execCommand(                        // ② gpg decrypt
        "gpg", "--decrypt", "--passphrase", passphrase, path);
      Properties props = new Properties();
      props.load(new StringReader(data));                     // ③ parse
      Map<String, String> map = new HashMap<>();
      for (String name : props.stringPropertyNames())
        map.put(name, props.getProperty(name));
      return new ConfigData(map);
    } catch (IOException e) {
      throw new RuntimeException(e);                          // ④ FAIL FAST
    }
  }

  @Override
  public ConfigData get(String path, Set<String> keys) {       // ⑤ subset
    ConfigData configData = get(path);
    Map<String, String> data = configData.data().entrySet()
      .stream().filter(e -> keys.contains(e.getKey()))
      .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    return new ConfigData(data, configData.ttl());
  }

  @Override public void close() {}
}

Encrypt the credentials file:

gpg --symmetric --output credentials.props.gpg \
  --passphrase "$PASSPHRASE" credentials.props

Reference it indirectly — the ${provider:path:key} syntax:

username=${gpg:/path/to/credentials.props.gpg:username}
password=${gpg:/path/to/credentials.props.gpg:password}
config.providers=gpg
config.providers.gpg.class=com.example.GpgProvider

(Same mechanism Ch. 9 §2.6 referenced for Connect secret providers — Vault/AWS/Azure providers exist in the community.)

B. Broker-only: encrypted configs in ZooKeeper, no custom code

bin/kafka-configs.sh --zookeeper localhost:2181 --alter \
  --entity-type brokers --entity-name 0 --add-config      \
  'listener.name.external.ssl.keystore.password=server-ks-password,\
password.encoder.secret=encoder-secret'

"The following command can be executed BEFORE STARTING BROKERS to store encrypted SSL key store passwords for brokers in ZooKeeper. THE PASSWORD ENCODER SECRET MUST BE CONFIGURED IN EACH BROKER'S CONFIGURATION FILE TO DECRYPT THE VALUE."

The chain of trust bottoms out somewhere:

• Config file holds: password.encoder.secret (or $PASSPHRASE in the env)
• ZooKeeper holds: the Encrypted real password

► You’ve reduced the secret surface from “many passwords in many files” to “one secret per broker.” That’s the win — not perfect secrecy.