小B在实现了统一日志分析平台后，想到的第一个问题就是平台的安全性，用于安全分析的平台被攻击成功，那么小B将颜面扫地还可能被扫地出门。所以小B认证分析了平台可能存在的安全风险：

• 平台产品自身的安全漏洞(解决方案：及时安装补丁或升级版本)；

• 数据泄露：在采集存储使用等过程中造成的数据泄露问题(解决方案：加密)；

• 未授权访问：平台无任何访问控制措施导致未授权访问(解决方案：见后文)；

对于平台的安全性，小B觉得首要的是使用没有已知漏洞的产品版本，其次就是添加访问控制。对于小B的日志分析平台，其中有这么几个地方需要额外注意：ES与Kibana的安全性、Kafka的安全性。

通用访问控制实现

# 如果操作系统是Centos系列7以上版本，推荐使用
# 保护Kafka与Zookeeper
# 只允许Kafka的机器访问ZK，这种方法也是最有效防护ZK未授权访问导致信息泄露的办法
firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="10.10.10.9" port port="2181" protocol="tcp" accept'
# 只允许Beats机器访问Kafka
firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="10.10.10.0/24" port port="9092" protocol="tcp" accept'
# 只允许办公网络IP访问Kibana
firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="182.*.*.107" port port="5601" protocol="tcp" accept'
# 如果白名单IP比较分散，可以使用firewall结合ipset，关于ipset的使用大家就自行百度吧！
firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source ipset="ipset_name" port port="9092" protocol="tcp" accept'

Elastic安全性

# 停止Kibana与elasticsearch服务
systemctl stop kibana.service
systemctl stop elasticsearch.service
# 创建es证书颁发机构
/usr/share/elasticsearch/bin/elasticsearch-certutil ca
# 创建es集群通信证书
/usr/share/elasticsearch/bin/elasticsearch-certutil cert --ca /etc/elasticsearch/elastic-stack-ca.p12

# 修改证书权限
chmod 644 /etc/elasticsearch/elastic-certificates.p12
# 修改配置文件
vim /etc/elasticsearch/elasticsearch.yml
# 在文件末尾添加如下内容

xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: /etc/elasticsearch/elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: /etc/elasticsearch/elastic-certificates.p12

# 如果配置了证书密码
# /usr/share/elasticsearch/bin/elasticsearch-keystore add xpack.security.transport.ssl.keystore.secure_password
# /usr/share/elasticsearch/bin/elasticsearch-keystore add xpack.security.transport.ssl.truststore.secure_password
# 启动Elasticsearch服务
systemctl start elasticsearch.service
# 创建账号密码
/usr/share/elasticsearch/bin/elasticsearch-setup-passwords interactive

• elastic：超级用户
• apm_system：使用APM时的用户
• kibana：用来连接ES的用户
• logstash_system：使用Logstash时的用户
• beats_system：使用Beats时的用户
• remote_monitoring_user：收集和存储监控信息的用户

# 修改kibana配置文件
vim /etc/kibana/kibana.yml
# 在文件末尾添加如下内容

elasticsearch.username: "kibana"
elasticsearch.password: "kibana"

# 启动kibana
systemctl start kibana.service
# 使用elastc账户登录kibana

• 方法一：使用Nginx Proxy实现

# 安装认证模块
yum install httpd-tools.x86_64 -y
# 创建认证账号
htpasswd -cm /etc/nginx/kibana-user kibana
# 配置nginx
vim /etc/nginx/nginx.conf

server {
        listen       80;
        server_name  kibana;
        auth_basic      "Restricted Access";
        auth_basic_user_file    /opt/nginx/conf/kibana-user;
        #charset koi8-r;
        #access_log  logs/host.access.log  main;
        location / {
            proxy_pass http://10.10.10.9:5601;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection 'upgrade';
            proxy_set_header Host $host;
            proxy_cache_bypass $http_upgrade;
        }
    }

• 方法二：使用Nginx Lua实现

• SearchGuard：http://docs.search-guard.com/latest/index (来自小B朋友的用后感：不稳定、插件容易出问题)
• Shield：https://www.elastic.co/guide/en/shield/current/index.html (没有使用过不做评论)

Kafka与Zookeeper安全性

• TLS：提供加密，是否能解决未授权访问？未知。
• SASL：只针对客户端连接访问，无法解决Zookeeper未授权访问；
• ACL：只针对节点内的授权，无法解决Zookeeper未授权访问；

• 存放Broker节点的配置信息和controller节点的配置信息；
• 监听父节点的子节点列表来进行controller的选举；
• partition信息的存储；

1. 修改Zookeeper的配置文件，加入官网提供的配置文件

vim /opt/zookeeper-3.4.14/conf/zoo.cfg
# 在文件末尾添加如下内容

# 打开SASL开关
quorum.auth.enableSasl=true
# 打开Client-to-Server authentication
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
quorum.cnxn.threads.size=20
# 认证模式使用SASL
requireClientAuthScheme=sasl

2. 创建Zookeeper的认证文件

vim /opt/zookeeper-3.4.14/conf/zookeeper-server.conf
# 创建一个用户名为kafka，密码为123456的用户信息

Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    user_kafka="123456";
};

3. 修改Zookeeper运行的环境变量

vim /opt/zookeeper-3.4.14/bin/zkServer.sh
# 在140行，nohup后面加入运行的环境变量

# 原文 nohup "$JAVA" "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" \
# 添加 -DJava.security.auth.login.config信息
nohup "$JAVA" "-Djava.security.auth.login.config=/opt/zookeeper-3.4.14/conf/zookeeper-server.conf" "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" \

4. 启动Zookeeper验证SASL是否配置成功

# 启动服务端
/opt/zookeeper-3.4.14/bin/zkServer.sh start
# 使用客户端连接
/opt/zookeeper-3.4.14/bin/zkCli.sh

5. 创建客户端认证文件

vim /opt/zookeeper-3.4.14/conf/zookeeper-client.conf

Client {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="kafka"
    password="12345";
};

1. 在上面我们已经配置好了Zookeeper的SASL了，这里我们直接进行Kafka的SASL配置

# 首先修改Kafka服务端配置文件，启动SASL/PLAIN认证
vim /opt/kafka_2.11-2.3.1/config/server.properties

# PLAIN是SASL的一种认证方式，PLAINTEXT是通信过程是明文，不加密。
listeners=SASL_PLAINTEXT://10.10.10.9:9092
# 启用的认证模式PLAIN，也可使用GSSAPI
sasl.enabled.mechanisms=PLAIN
# kafka broker之间也需要使用PLAIN方式认证，也可使用GSSAPI
sasl.mechanism.inter.broker.protocol=PLAIN
# 通信为明文。如果需要使用SSL加密通信，则使用SASL_SSL，不过需要配置证书
security.inter.broker.protocol=SASL_PLAINTEXT

2. 创建一个Kafka的认证文件

vim /opt/kafka_2.11-2.3.1/config/kafka.conf
# kafkaServer为KafkaServer的SASL
# Client是Kafka用来连接Zookeeper的配置，如果不配置Client，会提示WARN不会报错，原理在上面已经解释过了

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka_user"
    password="kafka_pass"
    user_producer="producer"
    user_consumer="consumer";
};
Client {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="123456";
};

3. 修改Kafka运行的环境变量

vim /opt/kafka_2.11-2.3.1/bin/kafka-run-class.sh
# 找到KAKFA_OPTS

if [ -z "$KAFKA_OPTS" ]; then
  KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka_2.11-2.3.1/config/kafka.conf"
fi

4. 启动Kafka并测试Kafka认证

# 启动Kafka服务
/opt/kafka_2.11-2.3.1/bin/kafka-server-start.sh -daemon /opt/kafka_2.11-2.3.1/config/server.properties
# 启动Kafka Producer测试是否能使用Kafka，记住先创建一个system-messages的topic
/opt/kafka_2.11-2.3.1/bin/kafka-console-producer.sh --broker-list 10.10.10.9:9092 --topic system-messages

5. 配置Kafka Client SASL认证

# 首先是修改Producer的配置文件
vim /opt/kafka_2.11-2.3.1/config/producer.properties
# 在文件末尾添加如下内容
# 如果不添加还是不能使用

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

# 其次是创建Kafka认证的文件，我这里Kafka CLient与Server是同一台，所以配置文件我写在同一个文件中
vim /opt/kafka_2.11-2.3.1/config/kafka.conf
# 在文件末尾添加如下内容

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="producer"
    password="producer";
};

6. Kafka客户端配置完成后，我们再来测试

# 切记，必须制定producer的配置文件
/opt/kafka_2.11-2.3.1/bin/kafka-console-producer.sh --broker-list 10.10.10.9:9092 --topic system-messages

7. 与Producer一样的配置Consumer。如果你与小B一样使用的是一个配置文件就不用重复配置了，如果不是记得配置，并且需要指定运行环境变量。

1. 首先是Filebeat成为Kafka的Producer

vim /etc/filebeat/filebeat.yml

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/test.log
output.kafka:
  hosts: ["10.10.10.9:9092"]
  topic: "test"
  username: "producer"
  password: "producer"

2. 其次就是Logstash成为Kafka的Consumer

vim /etc/logstash/conf.d/kafka.conf

input {
	kafka {
		bootstrap_servers => "10.10.10.9:9092"
		security_protocol => "SASL_PLAINTEXT"
		sasl_mechanism => "PLAIN"
		jaas_path => "/etc/logstash/kafka-client-jaas.conf"
		topics => ["test"]
	}
}
output {
    stdout {}
}

# 配置Kafka的Client信息
vim /etc/logstash/kafka-client-jaas.conf

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="consumer"
    password="consumer";
};

 

- HISTORY -

《日志分析系列(一)：介绍篇》

《日志分析系列(二)：平台实现篇》

《日志分析系列(外传一)：Nginx透过代理获取真实客户端IP》

《日志分析系列(外传二)：Nginx日志统一化》