本文主要介绍基于SCRAM进行身份验证，使用Kafka ACL进行授权，SSL进行加密以及使用camel-Kafka连接Kafka群集以使用camel路由生产和消费消息的过程。
加密与认证
AMQ Streams支持加密和身份验证，这是侦听器配置的一部分。
侦听器配置
Kafka代理中的加密和身份验证是针对每个侦听器配置的。
Kafka代理中的每个侦听器都配置有自己的安全协议。配置属性"listener.security.protocal"定义哪个侦听器使用哪种安全协议。然后将每个侦听器名称映射到其安全协议中。
支持的安全协议有：
- 纯文本：侦听器，无需任何加密或身份验证。
- SSL协议：使用TLS加密的侦听器，以及使用TLS客户端证书的身份验证（可选）。
- SASL_PLAINTEXT：侦听器不加密，但具有基于SASL的身份验证。
- SASL_SSL：具有基于TLS的加密和基于SASL的身份验证的侦听器。
以下为指定SASL_SSL协议侦听器配置：
listeners=SASL_SSL://localhost:9092 advertised.listeners=SASL_SSL://localhost:9092 security.inter.broker.protocol=SASL_SSL
TLS加密
为了使用TLS加密和服务器身份验证，必须提供包含私钥和公钥的密钥库。通常使用Java密钥存储（JKS）格式的文件来完成此操作。在"ssl.keystore.location"属性中设置此文件的路径。"ssl.keystore.password"为集群中的所有Kafka代理生成TLS证书。证书应在其通用名称或主题备用名称中具有其公告的地址和引导地址。
编辑所有Kafka集群节点上的/opt/kafka/config/server.properties 配置文件，以进行以下操作：
更改"listener.security.protocol.map"字段以为要使用TLS加密的侦听器指定SSL协议。
将"ssl.keystore.location"选项设置为带有代理证书的JKS密钥库的路径。
将"ssl.keystore.password"选项设置为用于保护密钥库的密码。
SASL认证
使用Java身份验证和授权服务（JAAS）配置SASL身份验证。JAAS还用于验证Kafka和ZooKeeper之间的连接。
JAAS使用其自己的配置文件。该文件位于/opt/kafka/config/jaas.conf，通过普通的未加密连接以及通过TLS连接都支持SASL身份验证。可以分别为每个侦听器启用SASL。要启用它，listener.security.protocol.map中的安全协议必须是SASL_PLAINTEXT或SASL_SSL。
Kafka中的SASL身份验证支持几种不同的机制：
- 普通
根据用户名和密码实施身份验证。用户名和密码以Kafka配置存储在本地。
- SCRAM-SHA-256和SCRAM-SHA-512
使用Salted Challenge Response身份验证机制（SCRAM）实现身份验证。SCRAM凭证集中存储在ZooKeeper中。SCRAM可以用于ZooKeeper群集节点在专用网络中隔离运行的情况。
- GSSAPI
针对Kerberos服务器实施身份验证
通过JAAS配置文件配置SASL机制。Kafka使用名为Kafka服务器的JAAS上下文。在JAAS中配置它们之后，必须在Kafka配置中启用SASL机制。这可以使用sasl.enabled.mechanisms属性完成
- SASL SCRAM
Kafka中的SCRAM身份验证包含两种机制：SCRAM-SHA-256和SCRAM-SHA-512。这些机制仅在使用的哈希算法上有所不同-SHA-256与更强的SHA-512。要启用SCRAM身份验证，JAAS配置文件必须包含以下配置：
[administrator@JavaLangOutOfMemory ~ ]% vi ${kafka_home}/config/kafka_server_jass.conf KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin123"; };
并在server.properties文件中启用SASL身份验证:
[administrator@JavaLangOutOfMemory ~ ]% vi ${kafka_home}/config/server.properties # With SASL & SSL encryption scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \ username="admin" \ password="admin@2016.08.19.com";
在$kafka-home/config目录下创建ssl-user-config.properties文件，如下所示：
[administrator@JavaLangOutOfMemory ~ ]% vi ${kafka_home}/config/ssl-user-config.properties security.protocol=SASL_SSL sasl.mechanism=SCRAM-SHA-512 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com"; ssl.truststore.location=$kafka_home/config/truststore/kafka.truststore.jks ssl.truststore.password=luga@2016.08.19.com
SCRAM机制的用户凭据存储在ZooKeeper中。 可借助kafka-configs.sh工具进行管理。
[administrator@JavaLangOutOfMemory ~ ]%/bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password='admin@2016.08.19.com']' --entity-type users --entity-name admin Warning: --zookeeper is deprecated and will be removed in a future version of Kafka. Use --bootstrap-server instead to specify a broker to connect to. Completed updating config for entity: user-principal 'admin'. ./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password='luga@2016.08.19.com']' --entity-type users --entity-name luga Warning: --zookeeper is deprecated and will be removed in a future version of Kafka. Use --bootstrap-server instead to specify a broker to connect to. Completed updating config for entity: user-principal 'luga'.
完整的$ {kafka-home} /config/server.properties文件如下所示:
# Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # see kafka.server.KafkaConfig for additional details and defaults ############################# Server Basics ############################# # The id of the broker. This must be set to a unique integer for each broker. broker.id=0 ############################# Socket Server Settings ############################# # The address the socket server listens on. It will get the value returned from # java.net.InetAddress.getCanonicalHostName() if not configured. # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 listeners=SASL_SSL://localhost:9092 # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). advertised.listeners=SASL_SSL://localhost:9092 security.inter.broker.protocol=SASL_SSL ssl.endpoint.identification.algorithm= ssl.client.auth=required sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512 sasl.enabled.mechanisms=SCRAM-SHA-512 # Broker security settings 46 ssl.truststore.location=/${kafka_home}/config/truststore/kafka.truststore.jks ssl.truststore.password=luga@2016.08.19.com ssl.keystore.location=/${kafka_home}/config/keystore/kafka.keystore.jks ssl.keystore.password=luga@2016.08.19.com ssl.key.password=luga@2016.08.19.com # ACLs authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer super.users=User:admin # #zookeeper SASL zookeeper.set.acl=false # # With SASL & SSL encryption scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \ username="admin" \ password="admin123"; # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL # The number of threads that the server uses for receiving requests from the network and sending responses to the network num.network.threads=3 # The number of threads that the server uses for processing requests, which may include disk I/O num.io.threads=8 # The send buffer (SO_SNDBUF) used by the socket server socket.send.buffer.bytes=102400 # The receive buffer (SO_RCVBUF) used by the socket server socket.receive.buffer.bytes=102400 # The maximum size of a request that the socket server will accept (protection against OOM) socket.request.max.bytes=104857600 ############################# Log Basics ############################# # A comma separated list of directories under which to store log files log.dirs=/tmp/kafka-logs # The default number of log partitions per topic. More partitions allow greater # parallelism for consumption, but this will also result in more files across # the brokers. num.partitions=1 # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. # This value is recommended to be increased for installations with data dirs located in RAID array. num.recovery.threads.per.data.dir=1 ############################# Internal Topic Settings ############################# # The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state" # For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3. offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 ############################# Log Flush Policy ############################# # Messages are immediately written to the filesystem but by default we only fsync() to sync # the OS cache lazily. The following configurations control the flush of data to disk. # There are a few important trade-offs here: # 1. Durability: Unflushed data may be lost if you are not using replication. # 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush. # 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks. # The settings below allow one to configure the flush policy to flush data after a period of time or # every N messages (or both). This can be done globally and overridden on a per-topic basis. # The number of messages to accept before forcing a flush of data to disk #log.flush.interval.messages=10000 # The maximum amount of time a message can sit in a log before we force a flush #log.flush.interval.ms=1000 ############################# Log Retention Policy ############################# # The following configurations control the disposal of log segments. The policy can # be set to delete segments after a period of time, or after a given size has accumulated. # A segment will be deleted whenever *either* of these criteria are met. Deletion always happens # from the end of the log. # The minimum age of a log file to be eligible for deletion due to age log.retention.hours=168 # A size-based retention policy for logs. Segments are pruned from the log unless the remaining # segments drop below log.retention.bytes. Functions independently of log.retention.hours. #log.retention.bytes=1073741824 # The maximum size of a log segment file. When this size is reached a new log segment will be created. log.segment.bytes=1073741824 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies log.retention.check.interval.ms=300000 ############################# Zookeeper ############################# # Zookeeper connection string (see zookeeper docs for details). # This is a comma separated host:port pairs, each corresponding to a zk # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". # You can also append an optional chroot string to the urls to specify the # root directory for all kafka znodes. zookeeper.connect=localhost:2181 # Timeout in ms for connecting to zookeeper zookeeper.connection.timeout.ms=18000 ############################# Group Coordinator Settings ############################# # The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance. # The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms. # The default value for this is 3 seconds. # We override this to 0 here as it makes for a better out-of-the-box experience for development and testing. # However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup. group.initial.rebalance.delay.ms=0
用JAAS启动Kafka，命令如下所示：
[administrator@JavaLangOutOfMemory ~ ]%export KAFKA_OPTS=-Djava.security.auth.login.config=/${kafka_home}/config/kafka_server_jaas.conf [administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-server-start.sh /${kafka_home}/config/server.properties
创建主题Demo-Topic:
[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:luga --operation Create --operation Describe --topic demo-topic Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW) (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW) Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW) (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW)
同理，我们给生产者和消费者授予相关权限：
[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:luga --producer --topic demo-topic Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW) (principal=User:luga, host=*, operation=WRITE, permissionType=ALLOW) (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW) Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW) (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW) (principal=User:luga, host=*, operation=WRITE, permissionType=ALLOW)
[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:luga --consumer --topic demo-topic --group demo-consumer-group Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: (principal=User:luga, host=*, operation=READ, permissionType=ALLOW) (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW) Adding ACLs for resource `ResourcePattern(resourceType=GROUP, name=demo-consumer-group, patternType=LITERAL)`: (principal=User:luga, host=*, operation=READ, permissionType=ALLOW) Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`: (principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW) (principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW) (principal=User:luga, host=*, operation=WRITE, permissionType=ALLOW) (principal=User:luga, host=*, operation=READ, permissionType=ALLOW) Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=demo-consumer-group, patternType=LITERAL)`: (principal=User:luga, host=*, operation=READ, permissionType=ALLOW)
开始生产消息：
[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic demo-topic --producer.config config/ssl-producer.properties >Hi this is docker >
生产者配置ssl-producer.properties：
[administrator@JavaLangOutOfMemory ~ ]% vi /${kafka_home}/config/ssl-producer.properties bootstrap.servers=localhost:9092 compression.type=none ### SECURITY ###### security.protocol=SASL_SSL sasl.mechanism=SCRAM-SHA-512 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com"; ssl.truststore.location=/{kafka_home}/config/truststore/kafka.truststore.jks ssl.truststore.password=luga@2016.08.19.com
消费者配置ssl-consumer.properties：
[administrator@JavaLangOutOfMemory ~ ]% vi /${kafka_home}/config/ssl-consumer.properties bootstrap.servers=localhost:9092 # consumer group id group.id=test-consumer-group # What to do when there is no initial offset in Kafka or if the current # offset does not exist any more on the server: latest, earliest, none #auto.offset.reset= ### SECURITY ###### security.protocol=SASL_SSL sasl.mechanism=SCRAM-SHA-512 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com"; ssl.truststore.location=/${kafka_home}/config/truststore/kafka.truststore.jks ssl.truststore.password=luga@2016.08.19.com
开始消费消息：
[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo-topic --from-beginning --consumer.config config/ssl-consumer.properties
现在基于came路由启动Spring Boot应用程序进行消息的生产与消费：
public class KafkaRouteBuilder extends RouteBuilder { @Override public void configure() throws Exception { //producer from("timer://foo?period={{period}}") .setBody(constant("Hi This is kafka example")) .to("kafka:{{kafka.topic}}?brokers={{kafka.bootstrap.url}}" + "&keySerializerClass=org.apache.kafka.common.serialization.StringSerializer" + "&serializerClass=org.apache.kafka.common.serialization.StringSerializer" + "&securityProtocol={{security.protocol}}&saslJaasConfig={{sasl.jaas.config}}" + "&saslMechanism={{sasl.mechanism}}&sslTruststoreLocation={{ssl.truststore.location}}" + "&sslTruststorePassword={{ssl.truststore.password}}&sslTruststoreType={{ssl.truststore.type}}") .log("${body}"); //consumer from("kafka:{{consumer.topic}}?brokers={{kafka.bootstrap.url}}&maxPollRecords={{consumer.max.poll.records}}" + "&keySerializerClass=org.apache.kafka.common.serialization.StringSerializer" + "&serializerClass=org.apache.kafka.common.serialization.StringSerializer" + "&groupId={{consumer.group}}&securityProtocol={{security.protocol}}&saslJaasConfig={{sasl.jaas.config}}" + "&saslMechanism={{sasl.mechanism}}&sslTruststoreLocation={{ssl.truststore.location}}" + "&sslTruststorePassword={{ssl.truststore.password}}&sslTruststoreType={{ssl.truststore.type}}" + "&autoOffsetReset={{consumer.auto.offset.reset}}&autoCommitEnable={{consumer.auto.commit.enable}}") .log("${body}"); }
application.yml文件配置如下：
kafka.topic=demo-topic kafka.bootstrap.url=localhost:9092 period=10000&repeatCount=5 kafka.key.serializer=org.apache.kafka.common.serialization.StringSerializer kafka.value.serializer=org.apache.kafka.common.serialization.StringSerializer consumer.topic=demo-topic consumer.group=demo-consumer-group consumer.max.poll.records=1 consumer.threads=10 consumer.consumersCount=1 consumer.auto.offset.reset=earliest consumer.auto.commit.enable=true consumer.receive.buffer.bytes=-1 security.protocol = SASL_SSL ssl.truststore.location =/luga/git/{...}/src/main/truststore/kafka.truststore.jks ssl.truststore.password =luga@2016.08.19.com ssl.truststore.type = JKS sasl.mechanism = SCRAM-SHA-512 #sasl.kerberos.service.name= sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com";
测试运行：
[administrator@JavaLangOutOfMemory ~ ]% mvn spring-boot:run [INFO] Scanning for projects... [INFO] [INFO] ---------< org.apache.camel.example:camel-example-kafka-sasl >---------- [INFO] Building Camel :: Example :: Kafka :: sasl 1.0 [INFO] --------------------------------[ jar ]--------------------------------- 。。。
至此，简单的场景运行起来。
