kafka安全设置ubuntu上如何操作

Kafka在Ubuntu上的安全设置指南
在Ubuntu上为Kafka配置安全设置，需通过认证、加密、授权、防火墙及操作系统权限等多层防护，以下是详细操作步骤：

一、前提准备

1. 安装Java：Kafka依赖Java运行环境，推荐使用OpenJDK 11及以上版本。

   sudo apt update && sudo apt install -y openjdk-11-jdk
   

2. 下载并解压Kafka：从Apache官网下载最新稳定版Kafka（如3.2.0），解压至指定目录（如/opt/kafka）。

   wget https://downloads.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz
   tar -xzf kafka_2.13-3.2.0.tgz -C /opt/
   cd /opt/kafka_2.13-3.2.0
   

二、配置SASL认证（身份验证）

SASL用于验证客户端与Broker的身份，常用PLAIN（明文，需配合SSL）或SCRAM（加密密码）机制。

1. 修改Broker配置（server.properties）

编辑Kafka配置文件，启用SASL并指定机制：

# 启用SASL_PLAINTEXT监听器（生产环境建议用SASL_SSL）
listeners=SASL_PLAINTEXT://:9092
security.inter.broker.protocol=SASL_PLAINTEXT
# 指定Broker间认证机制
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
# 启用客户端支持的机制
sasl.enabled.mechanisms=SCRAM-SHA-512

2. 创建JAAS配置文件

在config/目录下创建kafka_server_jaas.conf，配置Broker用户凭证（以admin用户为例）：

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin-secret";
};

3. 启动Broker时加载JAAS配置

修改Kafka启动脚本kafka-server-start.sh，添加JVM参数指定JAAS文件路径：

export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"

4. 创建客户端用户

使用kafka-configs.sh脚本添加客户端用户（如client1）：

bin/kafka-configs.sh --zookeeper localhost:2181 \
    --alter \
    --add-config "SCRAM-SHA-512=[password=client1-secret]" \
    --entity-type users \
    --entity-name client1

三、配置SSL/TLS加密（数据传输加密）

SSL/TLS用于加密Broker与客户端之间的通信，防止数据泄露。

1. 生成证书（自签名）

使用OpenSSL生成CA证书、Broker证书和客户端证书：

# 创建证书目录
mkdir -p config/certs && cd config/certs

# 生成CA私钥和证书（有效期3650天）
openssl req -new -x509 -days 3650 -keyout ca-key.pem -out ca-cert.pem -subj "/CN=KafkaCA"

# 生成Broker私钥和证书签名请求（CSR）
openssl req -newkey rsa:2048 -days 365 -nodes -keyout broker-key.pem -out broker-req.pem -subj "/CN=kafka-broker"

# 用CA签名Broker CSR，生成Broker证书
openssl x509 -req -in broker-req.pem -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial -out broker-cert.pem -days 365

# 生成客户端私钥和CSR
openssl req -newkey rsa:2048 -days 365 -nodes -keyout client-key.pem -out client-req.pem -subj "/CN=kafka-client"

# 用CA签名客户端CSR，生成客户端证书
openssl x509 -req -in client-req.pem -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial -out client-cert.pem -days 365

2. 导入证书到Java Keystore/Truststore

将证书导入Broker的Keystore（存储自身证书和私钥）和Truststore（存储CA证书）：

# 创建Broker Keystore并导入证书
keytool -importkeystore -srckeystore broker-cert.pem -destkeystore kafka.server.keystore.jks -deststoretype JKS -deststorepass kafka-pass -srcstorepass kafka-pass

# 创建Broker Truststore并导入CA证书
keytool -import -alias ca-cert -file ca-cert.pem -keystore kafka.server.truststore.jks -keystorepass kafka-pass -noprompt

3. 配置Broker启用SSL

修改server.properties，添加SSL配置：

# 启用SSL监听器（生产环境建议用SASL_SSL）
listeners=SSL://:9093
security.inter.broker.protocol=SSL
# 指定Keystore和Truststore路径及密码
ssl.keystore.location=config/certs/kafka.server.keystore.jks
ssl.keystore.password=kafka-pass
ssl.key.password=kafka-pass
ssl.truststore.location=config/certs/kafka.server.truststore.jks
ssl.truststore.password=kafka-pass
# 限制SSL协议版本（禁用TLSv1.0/1.1）
ssl.enabled.protocols=TLSv1.2,TLSv1.3
# 指定加密套件（优先使用AES-GCM）
ssl.cipher.suites=TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384

4. 配置客户端SSL

创建客户端Keystore和Truststore（步骤同Broker），并修改客户端配置（client.properties）：

security.protocol=SSL
ssl.truststore.location=config/certs/kafka.client.truststore.jks
ssl.truststore.password=kafka-pass
ssl.keystore.location=config/certs/kafka.client.keystore.jks
ssl.keystore.password=kafka-pass
ssl.key.password=kafka-pass

四、配置授权（ACL，访问控制）

授权用于控制用户对Kafka资源（如Topic、Group）的操作权限（读、写、创建等）。

1. 启用ACL授权

修改server.properties，启用SimpleAclAuthorizer并设置超级用户：

# 指定授权类（默认已启用SimpleAclAuthorizer）
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
# 禁止未配置ACL的资源被任意访问
allow.everyone.if.no.acl.found=false
# 定义超级用户（可跳过ACL检查）
super.users=User:admin

2. 创建ACL规则

使用kafka-acls.sh脚本添加ACL规则（示例：允许admin用户读写test-topic）：

# 允许admin用户对test-topic的所有操作
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
    --add --allow-principal User:admin --operation All --topic test-topic

# 允许client1用户对test-topic的读操作
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
    --add --allow-principal User:client1 --operation Read --topic test-topic

五、配置防火墙

限制Kafka端口（默认9092/9093）的访问，仅允许可信IP连接：

# 允许本地回环接口
sudo ufw allow from 127.0.0.1 to any port 9092
sudo ufw allow from 127.0.0.1 to any port 9093

# 允许特定IP段（如公司内网）
sudo ufw allow from 192.168.1.0/24 to any port 9092
sudo ufw allow from 192.168.1.0/24 to any port 9093

# 启用防火墙
sudo ufw enable

六、操作系统权限限制

1. 以专用用户运行Kafka：创建kafka用户并修改Kafka目录权限：

   sudo useradd -r -m -d /opt/kafka -s /sbin/nologin kafka
   sudo chown -R kafka:kafka /opt/kafka
   sudo chmod -R 750 /opt/kafka
   

2. 修改启动脚本：使用kafka用户启动Kafka服务：

   sudo -u kafka bin/kafka-server-start.sh config/server.properties
   

七、验证配置

1. 测试SASL认证：使用客户端工具发送消息，指定SASL和SSL配置：

   bin/kafka-console-producer.sh --broker-list localhost:9093 \
       --topic test-topic \
       --producer.config config/client.properties
   

   client.properties需包含SASL和SSL配置（参考上文）。

2. 测试ACL权限：尝试用未授权用户（如client2）访问test-topic，应提示权限不足。

通过以上步骤，可在Ubuntu上为Kafka构建认证+加密+授权+防火墙的多层安全体系，有效防范未授权访问和数据泄露。