IT评测·应用市场-qidao123.com

标题: Kafka（十）安全 [打印本页]
作者: 玛卡巴卡的卡巴卡玛    时间: 2024-7-29 00:49
标题: Kafka（十）安全
Kafka安全

安全性需要从系统整体层面考虑，Kafka提供了如下安全特性，可以与现有安全底子办法集成，构建出基于整个系统的安全。
包罗：
1 安全协议

Kafka使用两种标准技能（TLS，SASL）支持4种安全协议。
TLS（Transport Layer Security）：传输层安全，SSL（Secure Socket Layer）安全套接字层作为其前身，支持传输加密和身份验证。
SASL（Simple Authentication Security Layer）：简朴身份验证和安全层，是一个在面向连接的协议中使用不同的机制实现按身份验证的框架。
Kafka共提供有4种可选安全协议，每一个安全协议都结合了TLS或SASL。
1.1 PALINTEXT

无传输层加密，无身份验证，只适合在私有网络传输非敏感数据。
1.2 SSL

支持传输层加密，SSL客户端及服务端身份验证，适用于不安全网络传输。
SSL协议是基于IKP（Public Key Infrastructure公钥底子办法）的协议，所以起首要生成证书。这里基于双向认证，先容生成自署名证书的过程。
1.2.1 生成服务端证书

起首生成服务端CA密钥对，用来签发服务端证书。证书颁发机构（CA, Certificate Authority）即颁发数字证书的机构。是负责发放和管理数字证书的权势巨子机构，并作为电子商务买卖业务中受信托的第三方，承担公钥体系中公钥的正当性查验的责任。在这里，我们充当CA，给Kafka服务器和客户端签发证书。
  1. keytool -genkeypair -keyalg RSA -keysize 2048 -keystore server.ca.p12 -storetype PKCS12 -storepass 123456 -keypass 123456 -alias ca -dname "CN=BrokerCA" -ext bc=ca:true -validity 36500
复制代码
导出CA根证书，用来生成证书链，并安装进客户端的信托证书库：
  1. keytool -export -file server.ca.crt -keystore server.ca.p12 -storetype PKCS12 -storepass 123456 -alias ca -rfc
复制代码
接下来生成服务端密钥对：
  1. keytool -genkey -keyalg RSA -keysize 2048 -keystore server.ks.p12 -storepass 123456 -keypass 123456 -alias server -storetype PKCS12 -dname "CN=Kafka,O=QuPeng,C=China" -validity 36500
复制代码
导出证书请求文件，用于向CA申请证书：
  1. keytool -certreq -file server.csr -keystore server.ks.p12 -storetype PKCS12 -storepass 123456 -keypass 123456 -alias server
复制代码
CA用本身的密钥对签发一个服务端的证书：
  1. keytool -gencert -infile server.csr -outfile server.crt -keystore server.ca.p12 -storetype PKCS12 -storepass 123456 -alias ca -ext SAN=dns:localhost,ip:127.0.0.1,ip:172.26.143.96 -validity 36500
复制代码
将根证书和服务端证书都导入服务端密钥库：
  1. cat server.crt server.ca.crt > serverchain.crt
  2. keytool -importcert -file serverchain.crt -keystore server.ks.p12 -storepass 123456 -keypass 123456 -a
  3. lias server -storetype PKCS12 -noprompt
复制代码
如果集群的broker之间需要认证，将根证书导入信托密钥库：
  1. keytool -import -file server.ca.crt -keystore server.ts.p12 -storetype PKCS12 -storepass 123456 -alias server -noprompt
复制代码
将根证书导入客户端信托密钥库：
  1. keytool -import -file server.ca.crt -keystore client.ts.p12 -storetype PKCS12 -storepass 123456 -alias ca -noprompt
复制代码
1.2.2 生成客户端证书

过程与服务端证书类似。
生成客户端CA证书：
  1. keytool -genkeypair -keyalg RSA -keysize 2048 -keystore client.ca.p12 -storetype PKCS12 -storepass 123456 -keypass 123456 -alias ca -dname CN=ClientCA -ext bc=ca:true -validity 36500
  2. keytool -export -file client.ca.crt -keystore client.ca.p12 -storetype PKCS12 -storepass 123456 -alias ca -rfc
复制代码
导出客户端证书请求：
  1. keytool -genkey -keyalg RSA -keysize 2048 -keystore client.ks.p12 -storepass 123456 -keypass 123456 -alias client -storetype PKCS12 -dname "CN=Kafka Client App,O=QuPeng,C=China" -validity 36500
  2. keytool -certreq -file client.csr -keystore client.ks.p12 -storetype PKCS12 -storepass 123456 -keypass 123456 -alias client
复制代码
签发客户端证书：
  1. keytool -gencert -infile client.csr -outfile client.crt -keystore client.ca.p12 -storetype PKCS12 -sto
  2. repass 123456 -alias ca -validity 36500
复制代码
将客户端根证书和客户端证书都导入密钥库：
  1. cat client.crt client.ca.crt > clientchain.crt
  2. keytool -importcert -file clientchain.crt -keystore client.ks.p12 -storepass 123456 -keypass 123456 -alias client -storetype PKCS12 -noprompt
复制代码
将服务端证书导入信托证书库：
  1. keytool -import -file client.ca.crt -keystore server.ts.p12 -alias client -storetype PKCS12 -storepass 123456 -noprompt
复制代码
1.2.3 修改配置

服务端server.properties：
  1. listeners=PLAINTEXT://172.26.143.96:9092,SSL://172.26.143.96:9093
  2. advertised.listeners=PLAINTEXT://172.26.143.96:9092,SSL://172.26.143.96:9093
  3. listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL
  5. #SSL
  6. ssl.keystore.location=/home/peter/keystore/server.ks.p12
  7. ssl.keystore.password=123456
  8. ssl.key.password=123456
  9. ssl.keystore.type=PKCS12
  10. ssl.truststore.location=/home/peter/keystore/server.ts.p12
  11. ssl.truststore.password=123456
  12. ssl.truststore.type=PKCS12
  13. ssl.client.auth=required
  15. inter.broker.listener.name=PLAINTEXT
  16. #security.inter.broker.protocol=PLAINTEXT
复制代码
下面具体先容一下这些属性：
listeners

listener属性是broker用来监听网络请求的监听器的列表。它的格式是：
  1. listener_name_1://host_name_1:port_1,listener_name_2://host_name_2:port_2
复制代码
listener_name_1：监听器名称，任意定名。并非为示例中的安全协议名称，那只是为了简便。listener.security.protocol.map属性界说了监听器和安全协议的映射。
host_name_1：IP地址。可以为空，为默认网卡的IP；可以为0.0.0.0，监听所有网卡。
port：监听的端口号。
advertised.listeners

按照字面意思明确，发布的监听器。通过Zookeeper发布，也就是告诉给别的broker，客户端程序我的监听器配置是什么，请通过它来与我通讯。
它的格式是：
  1. listener_name_1://host_name_1:port_1,listener_name_2://host_name_2:port_2
复制代码
例如在Zookeeper中查询到：
  1. {
  2.         "features": {},
  3.         "listener_security_protocol_map": {
  4.                 "PLAINTEXT": "PLAINTEXT",
  5.                 "SSL": "SSL"
  6.         },
  7.         "endpoints": ["PLAINTEXT://172.26.143.96:9092,SSL://172.26.143.96:9093"],
  8.         "jmx_port": -1,
  9.         "port": 9092,
  10.         "host": "localhost",
  11.         "version": 5,
  12.         "timestamp": "1647337490945"
  13. }
复制代码
listener.security.protocol.map

监听器名称和安全协议之间的映射关系聚集。
它的格式是：
  1. listener_name_1:security_protocol_name_1,listener_name_2:security_protocol_name_2
复制代码
security_protocol_name_1：4中协议中的一种：PLANTEXT，SSL，SASL_PLANTEXT，SASL_SSL。
inter.broker.listener.name

用于Broker之间通讯的listener的名称。如果未设置，则listener名称由 security.inter.broker.protocol界说(security.inter.broker.protocol默认值是PLAINTEXT)。
同时设置 这个和 security.inter.broker.protocol 属性是错误的。
security.inter.broker.protocol

用于在署理之间举行通讯的安全协议。
有用值为：PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL。
control.plane.listener.name

用于Controller和Broker之间通讯的监听器名称, Broker将会使用control.plane.listener.name 来定位监听器列表中的EndPoint
如果未设置,则默认使用inter.broker.listener.name来通讯,没有专门的链接。
Spring Boot客户端配置：
  1. security.protocol: SSL
  2. ssl.keystore.location: classpath:keystore/client.ks.p12
  3. ssl.keystore.password: 123456
  4. ssl.truststore.location: classpath:keystore/client.ts.p12
  5. ssl.truststore.password: 123456
  6. ssl.key.password: 123456
  7. ssl.keystore.type: PKCS12
复制代码
1.3 SASL_PLAINTEXT

无传输层加密，支持SASL客户端身份验证（一些SAAL机制也支持服务端身份验证），适用于私有网络。
1.4 SASL_SSL

支持传输层加密，带有SASL身份验证，适用于不安全网络，支持SASL客户端和SSL服务器端身份验证。
服务端server.properties：
  1. listeners=PLAINTEXT://172.26.143.96:9092,SSL://172.26.143.96:9093,SASL_SSL://172.26.143.96:9094
  2. advertised.listeners=PLAINTEXT://172.26.143.96:9092,SSL://172.26.143.96:9093,SASL_SSL://172.26.143.96:9094
  3. listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
  5. #SSL
  6. ssl.keystore.location=/home/peter/keystore/server.ks.p12
  7. ssl.keystore.password=123456
  8. ssl.key.password=123456
  9. ssl.keystore.type=PKCS12
  10. ssl.truststore.location=/home/peter/keystore/server.ts.p12
  11. ssl.truststore.password=123456
  12. ssl.truststore.type=PKCS12
  13. ssl.client.auth=required
  15. inter.broker.listener.name=PLAINTEXT
  16. #security.inter.broker.protocol=PLAINTEXT
  18. #SASL
  19. sasl.enabled.mechanisms=PLAIN
  20. sasl.mechanism.inter.broker.protocol=PLAIN
  21. listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  22.   username="kafka" password="654321" user_kafka="654321" user_Peter="123456";
复制代码
客户端springboot配置：
  1. security.protocol: SSL
  2. ssl.keystore.location: classpath:keystore/client.ks.p12
  3. ssl.keystore.password: 123456
  4. ssl.truststore.location: classpath:keystore/client.ts.p12
  5. ssl.truststore.password: 123456
  6. ssl.key.password: 123456
  7. ssl.keystore.type: PKCS12
  8. sasl:  mechanism: PLAIN  jaas:    config: org.apache.kafka.common.security.plain.PlainLoginModule required username="Peter" password="123456";
复制代码
2 身份认证

身份验证是通过创建客户端和服务器端身份来验证客户端和服务器端真实性的过程。当客户端连接到首领broker时，可以通过服务器端身份验证来确定本身连接的就是真实的broker。在举行客户端身份验证时，服务器通过验证客户端的凭证(比如暗码或数字证书)来确定Alice的身份，确保连接是来自Alice而不是冒充者。一旦通过身份验证，Alice的身份就与连接相干联，并在整个连接生命周期中起作用。
Kafka用KafkaPrincipal实例表现客户端身份，并用它授予资源访问权限，以及为具有这个客户端身份的连接分配配额。每个连接的KafkaPrincipal实例都是在身份验证过程中基于某种身份验证协议创建的。如果使用了基于暗码的身份验证，那么user_Peter的主体就是User:user_Peter。
   principal.builder.class
可以通过配置broker的principal.builder.class来自界说KafkaPrincipal，此参数的默认值是：org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder。
自界说主题构建类：
  1. package com.qupeng.demo.kafka.kafkaapache.security;
  3. import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder;
  4. import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
  5. import org.apache.kafka.common.security.ssl.SslPrincipalMapper;
  7. import java.util.regex.Matcher;
  8. import java.util.regex.Pattern;
  10. public class UserCertSubjectPrincipalBuilder extends DefaultKafkaPrincipalBuilder {
  11.    public static final String DEFAULT_MAPPER_RULE = "RULE:([cC][nN]=.*?),O=io.strimzi.*$/$1,O=io.strimzi/,RULE:([cC][nN]=.*?),([a-zA-Z]{1,2})=.*$/$1/,DEFAULT";
  13.    public UserCertSubjectPrincipalBuilder() {
  14.       super((KerberosShortNamer)null, new SslPrincipalMapper("RULE:([cC][nN]=.*?),O=io.strimzi.*$/$1,O=io.strimzi/,RULE:([cC][nN]=.*?),([a-zA-Z]{1,2})=.*$/$1/,DEFAULT"));
  15.    }
  17.    public UserCertSubjectPrincipalBuilder(KerberosShortNamer kerberosShortNamer, SslPrincipalMapper sslPrincipalMapper) {
  18.       super(kerberosShortNamer, new SslPrincipalMapper("RULE:([cC][nN]=.*?),O=io.strimzi.*$/$1,O=io.strimzi/,RULE:([cC][nN]=.*?),([a-zA-Z]{1,2})=.*$/$1/,DEFAULT"));
  19.    }
  20. }
复制代码
  匿名链接
User:ANONYMOUS 这个主体被用于未经身份验证的连接，包罗PLAINTEXT监
听器接受的客户端连接和SSL监听器接受的未经身份验证的客户端连接。
  2.1 SSL身份认证

如果监听器的安全协议配置的是SSL或SASL_SSL，那么监听器连接的安全传输层就会使用TLS。创建TLS连接需要一个握手过程，在这个过程中会实行身份验证、协商加密参数，并生成用于加密的共享密钥。客户端通过验证服务器的数字证书来确定服务器的身份。如果启用了SSL客户端身份验证，则服务器也会通过验证客户端数字证书来确定客户端的身份。所有的SSL流量都是加密的，适合用在不安全的网络中。
2.1.1 SSL的性能

因为SSL通道是加密的，所以会增长CPU方面的开销。SSL目前不支持零
复制传输。根据流量模式的不同，增长的开销可能会高达20%~30%。
2.1.2 配置客户端认证服务端

服务端需要配置密钥库，此中包罗broker的私有密匙和证书。同时，客户端也需要配置一个信托存储，此中包罗broker证书或签署了broker 证书的证书颁发机构(CA)的证书。
broker证书需要包罗作为主体别名(subject alternative name，SAN)扩展或公用名称(common name，CN)的broker主机名，客户端可以用它验证服务器的主机名。也可以使用通配符证书，为同一域名下的所有broker使用雷同的密钥存储，以此来简化配置。
   服务器主机名验证
在默认情况下，Kafka客户端会验证生存在服务器证书中的服务器主机名与
客户端连接的主机名是否匹配。连接用的主机名既可以是在客户端配置的引
导服务器地址，也可以是broker通过元数据相应返回给客户端的监听器主
机名。主机名验证是服务器端身份验证的一个关键部分，可以防止中间人攻
击，因此在生产环境中不应该被禁用。
  2.1.3 配置服务端认证客户端

ssl.client.auth=required

可以在broker 端配置 ssl.client.auth=required，让broker对SSL监听器接受的客户端
连接举行身份验证。客户端需要配置一个密钥存储;broker端需要配置一个信托存储，此中包罗客户端证书或客户端证书的CA。如果broker间通讯也启用了SSL，那么broker端的信托存储需要包罗broker证书的CA和客户端证书的CA。在默认情况下，客户端证书中的可识别名称(distinguished name，DN)会被作为KafkaPrincipal，用于授权和配额。可以通过配置 ssl.principal.mapping.rules来自界说主体。配置了SASL_SSL的监听器将禁用TLS客户端身份验证，其安全性主要依赖SASL身份验证和由SASL创建的KafkaPrincipal.
ssl.client.auth=requested

如果配置了 ssl.client.auth=requested，那么SSL客户端身份验证就会酿成
可选的。在这种情况下，没有配置密钥存储的客户端可以完成TLS握手，但
其主体是User:ANONYMOUS。
   忽略信托存储
如果使用了由知名的受信托权势巨子机构签署的证书，就可以省略broker和客户
端的信托存储配置。在这种情况下，将使用Java提供的默认信托存储来创建
信托。第2章先容过相干的安装步调。
  2.1.4 更新秘钥

必须在证书逾期前更新密钥和信托存储，避免TLS握手失败。可以通过直接修改存储文件或将配置参数指向新的存储文件来更新broker的SSL存储。在这两种情况下，都可以使用Admin API或Kafka配置工具来更新存储。下面的例子使用了Kafka配置工具来更新broker 0外部监听器的密钥存储。
  1. bin/kafka-configs.sh -- bootstrap-server localhost:9092
  2. -- command-config admin.props
  3. -- entity-type brokers -- entity-name 0 -- alter -- add-config \
  4. 'listener.name.external.ssl.keystore.location=/path/to/server.ks.p12'
复制代码
2.1.7 安全方面的考虑

使用TLS高版本

TLS广泛用于为多种协议(包罗HTTPS)提供传输层安全性。Kafka默认只启用了较新的TLSv1.2和TLSv1.3，因为旧协议(如TLSv1.1)存在已知的毛病。由于不安全重协商存在已知毛病，因此Kafka不支持TLS连接重协商。
使用256位密钥

在不安全网络中传输数据时，可以使用至少256位的加密密钥来防止加密攻击，并确保数据的完整性。
密钥的安全生存

在默认情况下，包罗私钥的密钥存储直接生存在文件系统中，所以非常有须要对文件系统的访问权限举行限制。如果私钥被泄漏，那么可以使用标准的Java TLS特性来吊销证书。
缩短密钥寿命

还可以使用寿命较短的密钥来低落泄漏的概率。
限制客户端配额

TLS握手过程开销巨大，并占用了broker大量的网络线程时间。对于不安全网络中的TLS监听器，要用连接配额来保护它们不受拒绝服务攻击，从而保证broker的可用性。
耽误发送客户端相应

broker的配置参数 connection.failed.authentication.delay.ms 可以用来耽误发送身份验证失败相应，以低落客户端对身份验证失败的重试速率。
2.2 SASL身份验证

Kafka协议支持使用SASL举行身份验证，并内置支持几种常用的SASL机制。SASL可以和合TLS作为传输层，提供一个有身份验证和加密的安全通道。SASL身份验证通过一系列服务器质询(challenge)和客户端相应来实现，此中SASL机制正义了质询和相应的序列和连接格式。broker支持以下几种SASL机制，并通过回调机制集成现有的安全底子办法。
可以通过 sasl.enabled.mechanisms参数为启用了SASL的broker监听器配置一个或多个SASL机制。客户端可以通过配置sasl.mechanism参数来选择任何一种已被启用的机制。
2.2.1 JAAS

Kafka 使用Java认证和授权服务(JAAS)来配置SASL。配置参数 sasl.jaas.config包罗了一个JAAS配置条目，此中指定了登录模块及相干参数。我们在配置 sasl.jaas.config时会将listener和mechanism作为前缀。
例如，listener.name.external.gssapi.sasl.jaas.config 表现为监听器EXTERNAL的SASL/GSSAPI机制配置JAAS条目。broker和客户端的登录过程将使用JAAS配置来确定用于身份验证的公共和私有凭证。
JAAS 配置文件
JAAS也可以配置在文件中，并通过Java的系统属性 java.security.auth.login.config来指定配置文件路径。不过，还是推荐使用Kafka的 sasl.jaas.config配置参数，因为它支持暗码保护，而且如果监听器启用了多种机制，则可以对每种SASL机制举行单独的配置。
Kafka支持的SASL机制可以通过回调机制与第三方认证服务器集成。我们可以为broker或客户端提供一个登录回调，以此来自界说登录过程，例如，获取用于身份验证的凭证。
也可以提供一个服务器端回调，用它来验证客户端凭证，例如，通过一台外部暗码服务器来验证暗码。还可以提供一个客户端回调，将客户端凭证注入而不是包罗在JAAS配置中。
下面将更具体地探究Kafka支持的SASL机制。
2.2.2 SASL/GSSAPI

Kerberos 是一种被广泛使用的网络身份验证协议，它使用了强加密技能，支持在不安全网络中举行安全的双向身份验证。通用安全服务应用程序编程接口(GSS-API)是一个为使用了不同身份验证机制的应用程序提供安全服务的框架。RFC-4752将GSS-API的Kerberos V5身份验证机制引入到了SASL中。Kerberos服务器的开源和企业级实现让Vorberos 成为许多对女全性有严酷要求的行业的选择。Kafka 通过SASL/GSSAPI支持Kerberos 身份验证。
2.2.2.1 设置SASL/GSSAPI

Kafka 使用包罗在Java运行时环境中的GSSAPI提供程序来支持Kerberos 身份验证。
  1. sasl.enabled.mechanisms=GSSAPI
  2. listener.name.external.gssapi.sasl.jaas.config=|
  3. com.sun.security.auth.module.Krb5LoginModule required \
  4. useKeyTab=true storeKey=true
  5. keyTab="/path/to/broker1.keytab" \
  6. principal="kafka/broker1.example.com@EXAMPLE.COM";
复制代码
如果broker间通讯启用了SASL/GSSAPI，则也需要配置broker间的SASL 机制和Kerberos服务名。
  1. sasl.mechanism.inter.broker.protocol=GSSAPI
  2. sasl.kerberos.service.name=kafka
复制代码
客户端需要在JAAS配置中指定它们本身的密钥表和主体，并用sasl.kerberos.service.name 指定它们想要连接的服务名。
  1. sasl.mechanism=GSSAPI
  2. sasl.kerberos.service.name=kafka
  3. sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
  4. useKeyTab=true storeKey=true
  5. keyTab="/path/to/alice.keytab" \
  6. principal="Alice@EXAMPLE.COM";
复制代码
2.2.2.2 安全方面的考虑

如果在生产环境中使用了Kerberos，那么建议便用SASL_SSL来保护身份验证流程和经过身份验证之后的数据流量。如果不使用TLS米提供安全传输层，则取窃听者有可能可以获取到充足的信息来发动字典攻击或暴力破解工具，并窃取客户端任凭证。使用随机生成的密钥比使用基于暗码生成的密钥更为安全。应该避免使用较弱的加空异法(如DES-MD5)，并通过文件系统权限来限制对密钥表文件的访问，因为能够访问这个文件的用户都可能成为冒充者。
SASL/GSSAPI需要一个安全的DNS服务来举行服务器身份验证。由于针对KDC服务或DNS 服务的拒绝服务攻击有可能会导致客户端身份验证失败，因此有须要对这些服务的可用性举行监控。Kerberos还依赖可配置的松散同步时钟来检测重放攻击，所以要确保时钟同步是安全的。
2.2.3 SASL/PLAIN

RFC-4616界说了一个简朴的用户名和暗码身份验证机制，可以与TLS一起，为我们提供安全的身份验证。在举行身份验证期间，客户端会向服务器发送用户名和暗码，服务器则会根据它的暗码存储来验证暗码。Kafka内置了SASL/PLAIN支持，我们可以通过一个自界说回调与外部暗码数据库集成。
2.2.3.1 配置SASL/PLAIN

SASL/PLAIN的默认实现会将broker的JAAS配置作为暗码存储。所有的客户端用户名和暗码都包罗在登录选项中，broker会验证客户端提供的暗码是否与此中的一个条目相匹配。如果broker间通讯启用了SASL/PLAIN，那么还需要提供broker的用户名和暗码。
  1. sasl.enabled.mechanisms=PLAIN
  2. sasl.mechanism.inter.broker.protocol=PLAIN
  3. listener.name.external.plain.sasl.jaas.config=\
  4. org.apache.kafka.common.security.plain.PlainLoginModule required \
  5. username="kafka" password="kafka-password" \
  6. user_kafka="kafka-password" \
  7. user_Alice="Alice-password";
复制代码
客户端也需要配置用户名和暗码。
  1. sasl.mechanism=PLAIN
  2. sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule \
  3. required username="Alice" password="Alice-password";
复制代码
在JAAS配置文件中生存暗码既不安全也不灵活，因为在添加或删除用户后需要重启所有的broker。
如果是在生产环境中使用SASL/PLAIN，那么可以使用自界说服务器回调将broker与安全的第三方暗码服务器集成在一起。也可以使用自界说回调实现暗码轮换。务器端的回调需要在轮换暗码的重叠时间段内同时支持旧暗码和新暗码，直到所有客户新都切换到新暗码。
下面是一个使用自界说回调的例子，我们对htpasswd生成的暗码举行验证。
  1. public class PasswordVerifier extends PlainServerCallbackHandler {
  2.     private final List<String> passwdFiles = new ArrayList<>();
  4.     public void configure(Map<String, ?> configs, String mechanism,
  5.                           List<AppConfigurationEntry> jaasEntries) {
  6.         Map<String, ?> loginOptions = jaasEntries.get(0).getOptions();
  7.         String files = (String) loginOptions.get("password.files");
  8.         Collections.addAll(passwdFiles, files.split(","));
  9.     }
  11.     @Override
  12.     protected boolean authenticate(String user, char[] password) {
  13.         return passwdFiles.stream()
  14.                 .anyMatch(file -> authenticate(file, user, password));
  15.     }
  17.     private boolean authenticate (String file, String user,char[] password){
  18.         try {
  19.             String cmd = String.format("htpasswd -vb %s %s %s",
  20.                     file, user, new String(password));
  21.             return Runtime.getRuntime().exec(cmd).waitFor() == 0;
  22.         } catch (Exception e) {
  23.             return false;
  24.         }
  25.     }
  26. }
复制代码
broker配置了暗码验证回调处理器和其他选项。
  1. listener.name.external.plain.sasl.jaas.config=\
  2. org.apache.kafka.common.security.plain.PlainLoginModule required \
  3. password.files="/path/to/htpassword.props,/path/to/oldhtpassword.props";
  4. listener.name.external.plain.sasl.server.callback.handler.class=\
  5. com.example.PasswordVerifier
复制代码
在客户端，一个实现了 org.apache.kafka.common.security.auth.AuthenticateCallbackHandler的客户端回调处理器会在创建连接时动态加载暗码，而不是在启动时从JAAS配置中加载静态暗码。为了提高安全性，可以将暗码生存在加密的文件或外部的安全服务器中。下面的例子使用配置类从文件中动态加载暗码。
  1. public class PasswordProvider implements AuthenticateCallbackHandler {
  3.     @Override
  4.     public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
  5.         Properties props = Utils.loadProps("src/main/resources/password.properties"); // 在回调中加载配置文件,确保使用最新的密码来进行密码轮换。
  6.         PasswordConfig config = new PasswordConfig(props);
  7.         String user = config.getString("username");
  8.         String password = config.getPassword("password").value(); // 即使密码被存储在外部,底层的配置库也会返回实际的密码。
  9.         for (Callback callback: callbacks) {
  10.             if (callback instanceof NameCallback)
  11.                 ((NameCallback) callback).setName(user);
  12.             else if (callback instanceof PasswordCallback) {
  13.                 ((PasswordCallback) callback).setPassword(password.toCharArray());
  14.             }
  15.         }
  16.     }
  19.     private static class PasswordConfig extends AbstractConfig {
  20.         static ConfigDef CONFIG = new ConfigDef()
  21.                 .define("username", STRING, HIGH, "User name")
  22.                 .define("password", PASSWORD, HIGH, "User password"); // 将密码定义成PASSWORD类型,确保密码不会被包含在日志中。
  24.         PasswordConfig(Properties props) {
  25.             super(CONFIG, props, false);
  27.         }
  28.     }
  29. }
复制代码
客户端和为broker间通讯启用了SASL/PLAIN的broker都可以配置客户端回调。
  1. sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule \
  2. required file="/path/to/credentials.props";
  3. sasl.client.callback.handler.class=com.qupeng.demo.kafka.kafkaapache.security.PasswordProvider
复制代码
2.2.3.2 安全方面的考虑

由于SASL/PLAIN会通过网络传输明文暗码，因此PLAIN机制应该与SASL_SSL一起使用，以此来提供安全传输层。
2.2.4 SASL/SCRAM

RFC-5802中引入了一种安全的用户名和暗码身份验证机制，办理了普通暗码身份验证机制(如SASL/PLAIN)的安全题目。加盐质询相应认证机制(salted challenge response authentication mechanism， SCRAM)不传输明文暗码，而是以一种无法被客户端冒充的格式生存暗码。密云与一些随机数组合在一起(也就是所谓的加盐)，并对其应用单向加密哈希函数。Kafka促供了一个内置的SCRAM实现，可以与安全的ZooKeeper部署在一起，不需要额外的暗码服务器。Kafka的SCRAM机制支持SCRAM-SHA-256和SCRAM-SHA-512。
2.2.4.1 设置SASL/SCRAM

可以在启动ZooKeeper之后创建初始用户，然后再启动broker。broker会在启动期间将SCRAM用户元数据加载到内存中，确保所有用户(包罗用于broker间通讯的broker用户)都能乐成地举行身份验证。我们可以随时添加或删除用户。
broker会使用基于ZooKeeper监听器的通知机制来更新缓存。下面的这个例子创建了一个用户，主体是User:Alice，暗码是Alice-password，使用的机制是SCRAM-SHA-512。
  1. $ bin/kafka-configs.sh -- zookeeper localhost:2181 -- alter -- add-config \
  2. 'SCRAM-SHA-512=[iterations=8192,password=Peter-password]'
  3. -- entity-type users -- entity-name Peter
复制代码
一个监听器可以配置一个或多个SCRAM机制。只有当监听器用于broker间通讯时，才需要用到broker的用户名和暗码。
  1. sasl.enabled.mechanisms=SCRAM-SHA-512
  2. sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
  3. listener.name.external.scram-sha-512.sasl.jaas.config=\
  4. org.apache.kafka.common.security.scram.ScramLoginModule required \
  5. username="kafka" password="kafka-password";
复制代码
客户端必须配置一个在broker端启用的SASL机制，而且客户端JAAS配置中必须包罗用户名和暗码。
  1. sasl.mechanism=SCRAM-SHA-512
  2. sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule \
  3. required username="Peter" password="Peter-password";
复制代码
可以用配置管理工具的-- add-config选项添加新SCRAM用户，用-- delete-config选项删除用户。当现有用户被删除时，我们无法为这个用户创建新连接，但这个用户的现有连接可以继续使用。可以为broker配置一个重新认证时间隔断，用于限制用户被删除后可以继续使用现有连接的时间。下面的例子中删除了Alice的SCRAM-SHA-512机制的凭证。
  1. $ bin/kafka-configs.sh -- zookeeper localhost:2181 -- alter -- delete-config |
  2. 'SCRAM-SHA-512' -- entity-type users -- entity-name Peter
复制代码
2.2.4.2 安全方面的考虑

SCRAM在暗码中加入了随机数，并对其应用了单向加密哈希函数，以避
免通过网络传输或在数据库中生存真实的暗码。然而，任何基于暗码的系统的安全性仅取决于暗码本身。必须接纳强暗码计谋，保护系统免受暴力或字典攻击。Kafka只支持强哈希算法 SHA-256和SHA-512，不接纳较弱的SHA-1，从而提供了安全保障。这与默认的4096迭代次数和密钥随机盐相结合，限制了当ZooKeeper安全遭破坏时给Kafka造成的影响。
我们还是要为握手过程中传输的密钥和生存在ZooKeeper中的密钥提供额外的保护，以历止暴力破解攻击。SCRAM必须与SASL_SSL一起使用，避免窃听者在身份验证过程中获取哈希密钥。ZooKeeper也必须启用SSL，同时ZooKeeper数诺必须举行磁盘加密，确生存储的密钥即使在遭受攻击时也不会被获取到。如果没有与安全的ZooKeeper部署在一起，则可以使用SCRAM回调，并与安全的外部凭证存储系统集成在一起。
2.2.5 SASL/OAUTHBEARER

OAuth是一个用于限制应用程序访问HTTP服务的授权框架。RFC-7628界说了
OAUTHBEARER SASL 机制，可以用通过OAuth 2.0获取的凭证访问受保护的非HTTP协议资源。OAUTHBEARER使用的是寿命较短且资源访问受限的OAuth 2.0不记名令牌，避免了在恒久暗码机制中存在的安全毛病。Kafka支持SASL/OAUTHBEARER 客户端身份验证，并可以与第三方OAuth服务器集成。OAUTHBEARER的内置实现使用了不安全的JSON Web Token(JWT)，不适合被用在生产环境中。可以通过自界说回调将其与标准的OAuth服务器集成，以便在生产环境中使用OAUTHBEARER机制提供安全的身份验证。
2.2.5.1 配置SASL/OAUTHBEARER

Kafka内置的SASL/OAUTHBEARER实现不验证令牌，因此只需要在JAAS配置中指定登录模块。如果监听器被用于broker间通讯，则还需要提供broker在发起客户端连接时需要用到的令牌的相干信息。unsecuredLoginStringClaim_sub指定了默认的主体声明。
  1. sasl.enabled.mechanisms=OAUTHBEARER
  2. sasl.mechanism.inter.broker.protocol=OAUTHBEARER
  3. listener.name.external.oauthbearer.sasl.jaas.config=\
  4. org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule \
  5. required unsecuredLoginStringClaim_sub="kafka";  # broker间连接使用的令牌的主体声明。
复制代码
客户端必须配置主体声明选项unsecuredLoginStringClaim_sub。也可以配置其他声明和令牌生命周期。
  1. sasl.mechanism=OAUTHBEARER
  2. sasl.jaas.config=\
  3. org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule \
  4. required unsecuredLoginStringClaim_sub="Peter"; // User:Peter是默认的KafkaPrincipal
复制代码
为了将Kafka与第三方OAuth服务器集成，以便在生产环境中使用不记名令牌，Kafka客户端必须配置 sasl.login.callback.handler.class，通过恒久暗码或刷新令牌从OAuth服务器获取令牌。如果broker间通讯启用了OAUTHBEARER，则broker也需要配置一个登录回调处理器，用于获取broker在发起客户端连接时需要用到的令牌。
  1. public class OAuthTokenProvider implements AuthenticateCallbackHandler {
  3.     @Override
  4.     public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
  5.         OAuthBearerToken token = null;
  6.         for (Callback callback : callbacks) {
  7.             if (callback instanceof OAuthBearerTokenCallback) {
  8.                 token = acquireToken(); // 客户端必须从OAuth服务器获取一个令牌,并将其传给回调。
  9.                 ((OAuthBearerTokenCallback) callback).token(token);
  10.             } else if (callback instanceof SaslExtensionsCallback) { // 客户端可能还用了其他回调。
  11.                 ((SaslExtensionsCallback) callback).extensions(processExtensions(token));
  12.             } else {
  13.                 throw new UnsupportedCallbackException(callback);
  14.             }
  15.         }
  16.     }
  18.     private SaslExtensions processExtensions(OAuthBearerToken token) {
  19.         return null;
  20.     }
  22.     private OAuthBearerToken acquireToken() {
  23.         return null;
  24.     }
  25. }
复制代码
broker还必须配置listener.name.external.oauthbearer.sasl.server.callback.handler.class，用于验证客户端提供的令牌。
  1. public class OAuthTokenVerifier implements AuthenticateCallbackHandler {
  3.     @Override
  4.         public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
  5.         for (Callback callback : callbacks) {
  6.             if (callback instanceof OAuthBearerValidatorCallback) {
  7.                 OAuthBearerValidatorCallback cb = (OAuthBearerValidatorCallback) callback;
  8.                 try {
  9.                     cb.token(validatedToken(cb.tokenValue())); // OAuthBearerValidatorCallback包含来自客户端的令牌,broker会验证这个令牌。
  10.                 } catch (OAuthBearerIllegalTokenException e) {
  11.                     OAuthBearerValidationResult r = e.reason();
  12.                         cb.error(errorStatus(г), г.failureScope(), r.failure0penIdConfig());
  13.                 }
  14.             } else if (callback instanceof OAuthBearerExtensionsValidatorCallback) {
  15.                 OAuthBearerExtensionsValidatorCallback ecb =
  16.                         (OAuthBearerExtensionsValidatorCallback) callback;
  17.                         ecb.inputExtensions().map().forEach((k, v) ->
  18.                                 ecb.valid(validateExtension(k, v))); // broker会验证来自客户端的可选扩展。
  19.             } else {
  20.                 throw new UnsupportedCallbackException(callback);
  21.             }
  22.         }
  23.     }
  25.     private OAuthBearerToken validatedToken(String s) {
  26.         return null;
  27.     }
  28. }
复制代码
2.2.5.2 安全方面的考虑

由于SASL/OAUTHBEARER客户端通过网络发送OAuth 2.0不记名令牌，这些令牌可能会被用于冒充客户端，因此必须启用TLS来加密身份验证流量。如果令牌遭到破坏，那么可以使用短期令牌来限制泄漏范围。可以开启broker的重新认证，防止连接的存活时间超过令牌有用期。为broker配置重新认证时间隔断，并与可打消令牌相结合，有用限制了已有连接被打消后可以继续使用令牌的时间。
2.2.6 委托令牌

委托令牌是一种在broker和客户端之间共享的密钥，它提供了一种轻量级的配置方式，无须向客户端应用程序分发SSL密钥存储或Kerberos密钥表。可以用委托令牌来减少身份验址服务器(如Kerberos密钥分发中心，KDC)的负载。一些框架(如Kafka Connect)可以用委托令牌来简化worker的安全配置。一个已经通过broker身份验证的客户端可以为同一个用户主体创建委托令牌，并将它们分发给worker，worker可以直接用这些令牌通过broker的身份验证。每个委托令牌由令牌标识符和作为共享秘密使用的哈希消息验证码(HMAC)组成。基于委托令牌的客户端身份验证使用的是SASL/SCRAM机制，令牌标识符就是用户名，HMAC就是暗码。
可以使用Kafka Admin API或delegation-tokens下令创建或更新委托令牌。要为主体
Usereter创建委托令牌，必须先用Alice的凭据对客户端举行身份验证(除委托令牌之外的任何一种身份验证协议)。使用委托令牌举行身份验证的客户端不能创建其他委托令牌。
  1. $ bin/kafka-delegation-tokens.sh -- bootstrap-server localhost:9092 |
  2. -- command-config admin.props -- create -- max-life-time-period -1 \
  3. # 如果Peter运行了这个命令，那么生成的令牌就可以用来冒充Peter。这个令牌的所有者是User:Peter。我们将User:Bob配置为令牌的续订者。
  4. $ bin/kafka-delegation-tokens.sh -- bootstrap-server localhost:9092 |
  5. -- command-config admin.props -- renew -- renew-time-period -1 -- hmac c2Vjcmve #更新命令可以由令牌所有者(Peter)或令牌续订者(Bob)来执行。
复制代码
2.2.6.1 配置委托令牌

要创建和验证委托令牌，所有broker都必须使用 delegation.token.master.key
参数配置雷同的主密钥。这个主密钥只能通过重启所有broker举行轮换。所有已有的令牌都必须在更新主密钥之前删除，因为它们不会再被使用，应该在所有broker都更新了密钥之后再创建新令牌。
要用委托令牌举行身份验证，broker至少要启用一种SASL/SCRAM机制。客户端也需要配置SCRAM机制，将令牌标识符作为用户名，HMAC作为暗码。KafkaPrincipal就是与令牌相干联的原始主体，比如User:Alice。
  1. sasl.mechanism=SCRAM-SHA-512
  2. sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule \
  3. required tokenauth="true" username="MTIz" password="c2VjcmV0"; #用带有 tokenauth的SCRAM机制配置委托令牌。
复制代码
2.2.6.2 安全方面的考虑

与内置的SCRAM实现一样，委托令牌只适合用于有安全ZooKeeper的
环境中。SCRAM的所有安全注意事项也都适用于委托令牌。
用于生成令牌的主密钥必须通过加密或外部化到安全暗码存储的方式举行保护。如果令牌遭到破坏，则可以用短期委托令牌来限制泄漏范围。可以开启broker的重新认证，防止连接使用逾期的令牌，并限制已有连接在令牌被删除后继续存在的时间。
2.7 重新认证

当客户端向broker发起连接时，broker会对客户端举行身份验证。broker会验证客户端凭证，如果凭证有用，验证就通过。一些安全机制(比如Kerberos和OAuth)使用的凭证的生存期是有限的。Kafka的一个背景登录线程负责在旧凭证逾期之前获取新凭证，但在默认情况下，新凭证只用于验证新连接。使用旧凭证验证的旧连接将继续处理请求，直到因为某些原因（比如请求超时、空闲超时或网络错误）断开连接。在用于验证连接的凭证过长寿命连接可能会继续处理请求。可以通过配置connections.max.reauth.ms参数让broker举行重新认证。
connections.max.reauth.ms=0

可以添加监听器和认证机制作为前缀，例如：listener.name.<_ listener-name_>.oauthbearer.connections.max.reauth.ms=3600000
如果这个参数的值被设置为正整数，那么broker就会在举行SASL 握手期间将SASL连接的会话生存期告诉客户端。会话生存期是凭证剩余生存期和ronnections.max.reauth.ms二者当中的较小者。任安在这个隔断内没有举行重新认证的连接都将被broker终止。客户端可以使用背景登录线程获取的最新凭证或自界说回调注入的最新凭据来实行重新认证。
重新认证可在以下几种场景中加强安全性。
   被泄漏的用户身份
如果用户身份被泄漏，则必须尽快将其从系统中移除。一旦用户被从认证服务器上移除，该用户所有的新连接都将无法通过broker的身份验证。已有连接将继续处理请求，直到实行下一次重新认证。如果没有配置connections.max.reauth.ms，就不会举行重新认证，已有连接可能会在很长一段时间内继续使用被泄漏的用户身份。Kafka不支持SSL重新协商，因为旧SSL协议的重新协商过程存在已知的毛病。较新的协议(如TLSv1.3)不支持重新协商。
因此，已有SSL连接可以继续使用已打消或逾期的证书。可以用用户主体的
Deny ACL 来阻止这些连接继续实行操作。因为ACL变动可以很快被应用于
所有broker，所以这是禁用被泄漏用户访问权限最快的方法。
  2.8 安全更新不停机

Kafka需要定期轮换密钥、应用安全补丁以及更新到最新的安全协议。大部分维护任务是通过滚动更新的方式举行的，也就是用新的配置逐个重启broker，而像更新SSL密钥存储和信托存储这类任务则可以通过动态配置更新来实行，无须重启broker。
在向已有集群添加新的安全协议时，可以在保存旧监听器的同时加入新监听器，确保在更新安全协议期间客户端可以继续使用旧监听器。例如，可以按照下面的顺序将已有集群的PLAINTEXT 切换到SASL_SSL。
在向已有SASL监听器添加SASL机制或从已有SASL监听备中删除SASL机制时，可以在同一个监听器端口上举行滚动更新，以此来避免停机。可以按照下面的顺序将PLAIN切换到SCRAM-SHA-256。
4. 使用Kafka配置工具将所有已有用户添加到SCRAM存储中。
5. 设置 sasl.enabled.mechanisms=PLAIN，SCRAM-SHA-256，并为监听器配置 listener.name.<_ listener-name_>.scram-sha-256.sasl.jaas.config，然后举行broker滚动更新。
6. 修改所有的客户端应用程序，设置sasl.mechanism=SCRAM-SHA-256，并使用SCRAM更新 sasl.jaas.config。
7. 如果监听器被用于broker间通讯，则需要设置sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256，并举行broker滚动更新。
8. 移除PLAIN机制，再举行一次broker滚动更新。设置sasl.enabled.mechanisms=SCRAM-SHA-256，并移除 listener.name .< listener-name>.plain.sasl.jaas.config以及与PLAIN相干的配置参数。
3 加密

加密被用于保护数据的隐私和完整性。
3.1 传输层加密

启用了SSL和SASL_SSL安全协议的Kafka监听器将TLS作为传输层，提供了安全的加密通道，以用来保护在不安全网络上传输的数据。
3.2 数据存储加密

还需要接纳额外的步伐来保护静态数据，确保有Kafka日志存储磁盘物理访问权限的用户也无法检索敏感数据。物理磁盘可以接纳全磁盘加密或数据卷加密，即使磁盘被盗，也可以避免数据泄漏。
3.3 端到端加密

虽然传输层和数据存储加密已经提供了充足的保护，但仍旧需要提供额外的保护步伐，避免将自动数据访问权限授予平台管理员。broker内存中未加密的数据可能会出现在堆转储中，有磁盘访问权限的管理员可以直接访问这些数据和包罗埋伏敏感数据的Kafka日志。
如果集群中有高度敏感的数据或PII，则需要接纳额外步伐来保护数据隐私。为了符合监管合规性，特别是在云端，需要确保平台管理员或云供应商不能以任何方式访问秘密数据。可以将自界说加密提供程序内嵌到Kafka客户端中，以此来实现端到端加密，保证整个数据流都是加密的。
   PII（Personally Identifiable Information，个人可识别信息）是指可以用来唯一识别、接洽或定位个人的数据。这些数据包罗但不限于姓名、地址、电话号码、电子邮件地址、社会安全号码等。
  3.3.1 端到端加密流程

可以将序列化器和反序列化器与加密库集成，在序列化期间举行消息加密，在反序列化期间举行消息解密。消息加密通常使用对称加密算法，比如AES。生产者可以用生存在密钥管理系统(KMS)中的共享加密密钥加密消息，消费者可以用它解密消息。broker不需要访问加密密钥，并且永远看不到消息里未加密的内容，因此在云端使用这种方法是安全的。解密消息所需的加密参数可以生存在消息标头或消息体(针对不支持消息标头的旧客户端)中。还可以在消息标头中包罗数字署名，用于验证消息的完整性。
例如一个端到端加密数据流：
3.3.2 密钥管理

生产者和消费者必须配置凭证才能从KMS获取共享密钥。建议定期轮换密钥，以此来加强安全性，因为频仍轮换密钥可以限制被泄漏的消息的数目，还可以防止暴力破解攻击。如果使用旧密钥加密的消息仍旧符合保存计谋，那么新旧密钥都必须可用。有许多KMS系统支持优雅的对称加密密钥轮换，无须在Kafka客户端做任何特殊处理。对于压实型主题，使用旧密钥加密的消息可能会被保存很长时间，而且可能需要对旧消息举行重新加密。为了避免新消息对其造成干扰，生产者和消费者在这个过程中需要离线。
3.3.3 压缩

压缩加密的消息
与加密前压缩消息相比，加密后压缩消息并不会在存储空间方面提供任何优
势。可以通过配置让序列化器在加密消息之前压缩消息，或者让应用程序在
生产消息之前压缩消息。不管是哪一种情况，最好都禁用Kafka的压缩，因
为它们增长了开销却不会带来任何额外的好处。对于在不安全传输层上传输
的消息，还需要考虑已压缩的加密消息的安全传输题目。
3.3.4 消息键加密

在许多环境中，特别是在使用TLS作为传输层时，消息键不需要加密，因为它们通常不
会像消息体那样包罗敏感数据。但在某些情况下，明文的消息键可能不符合监管要求。由于在分区和压实时会用到消息的键，因此在对键举行转换时必须保存等价哈希，确保即使加密参数发生变革，键的哈希值也不会变。一种方法是将原始消息键的安全哈希作为消息键，并将加密的消息键生存在消息体或消息标头中。因为消息键和消息体的序列化是独立举行的，所以可以用生产者拦截器来实行这个转换。
4 授权

授权是决定你可以对哪些资源实行哪些操作的过程。broker会使用一个可定制的授权器举行访问控制管理。前面讲过，每当客户端向broker发起连接时，broker会对客户端举行身份验证，并将代表客户端身份的KafkaPrincipal与连接关联起来。在处理请求时，broker会验证与当前连接关联的主体是否被授权实行这个请求。
Kafka 提供了一个内置的授权器AclAuthorizer，我们可以通过下面的配置来启用它。
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
   SimpleAclAuthorizer
AclAuthorizer 是在Kafka 2.3中引入的。从Kafka 0.9.0.0开始的旧版本中有
一个内置的授权器 kafka.security.auth.SimpleAclAuthorizer，虽然该授权
器已被弃用，但仍受支持。
  AclAuthorizer 支持使用访问控制列表(ACL)对Kafka资源举行细粒度的访问控制。ACL生存在ZooKeeper中，broker也会将它们缓存在内存中，以便在授权请求时提供高性能的查找。broker在启动时会将ACL加载到缓存中，并通过ZooKeeper 的watcher通知机制来更新缓存。在授权请求时，broker会验证与当前连接关联的KafkaPrincipal是否有权限操作被请求的资源。
每个ACL由以下几部分组成:

如果一个动作没有匹配的Deny ACL，并且至少有一个匹配的Allow ACL，那么
AclAuthorizer将对这个动作举行授权。如果授予了Read权限、Write权限、Alter权限或Delete权限，那么也会隐含授予Describe权限;如果授予了AlterConfigs权限，那么也会隐含授予DescribeConfigs权限。
Kafka 提供了一个使用broker中的授权器管理ACL的工具。
  1. $KAFKA_HOME/bin/kafka-acls.sh --bootstrap-server 172.26.143.96:9092 --add --resource-pattern-type Literal --topic product Write --allow-principal User:Peter
  3. $KAFKA_HOME/bin/kafka-acls.sh --bootstrap-server 172.26.143.96:9092 --add --topic product Describe --allow-principal User:ANONYMOUS
复制代码
也可以直接在ZooKeeper中创建ACL，如果需要在启动broker之前创建ACL，那么这就非常有用。
  1. $KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
  2. --add --topic product \
  3. --resource-pattern-type LITERAL \
  4. --operation Write \
  5. --allow-principal User:Peter
  7. $KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
  8. --add --topic test.* \
  9. --resource-pattern-type PREFIXED \
  10. --producer #producer 是一种生产端的方便式写法，相当于 --operation WRITE --operation CREATE，包括的权限有：WRITE、CREATE
  11. --allow-principal User:Peter
  13. $KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
  14. --add --topic product Describe \
  15. --allow-principal User:ANONYMOUS
  17. $KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
  18. --add \
  19. --consumer \ #consumer 是一种消费端的方便式写法，相当于 --operation READ  ，包括的READ
  20. --topic test.1 \
  21. --group c1 \
  22. --allow-principal User:consumer
  24. $KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
  25. --add \
  26. --operation Read \
  27. --operation Describe \
  28. --group consumers \
  29. --resource-pattern-type LITERAL \
  30. --allow-principal User:consumer
复制代码
AclAuthorizer 提供了两个配置选项，可用于对资源或主体举行大范围的授权，以此来简化ACL管理，特别是在第一次为已有集群添加授权时。服务端server.properties：
  1. #Authorization
  2. authorizer.class.name=kafka.security.authorizer.AclAuthorizer
  3. #超级用户可以访问所有资源,不受任何限制,而且不能用Deny ACL禁止超级用户的权限。
  4. #如果admin的凭证被泄露,则必须将admin从super.users中移除,然后重启broker,让变更重生效。在生产环境中,通过ACL为用户授予特定权限会更为安全,因为在必要时可以
  5. #与Kafka中其他使用逗号分隔的配置参数不同,super.users使用分号来分隔多个用户,因为用户主体(比如SSL证书的可识别名称)通常会包含逗号。
  6. super.users=User:admin;User:MaYun
  7. #如果allow.everyone.if.no.acl.found被设置为true,那么所有用户都可以访问没有配置ACL的资源。如果是第一次在集群中启用授权或在开发过程中启用授权,那么这个可能很有用,但不适合用在生产环境中,因为这样有可能在无意中授予了访问新资源的权限。而且,如果添加了与前缀或通配符匹配的ACL,则不再满足no.acl.found条件,访问权限有可能被意外移除。
  8. allow.everyone.if.no.acl.found=true
复制代码
4.1 自界说授权

在Kafka中，可以自界说授权这样就可以实，现额外的控制或增长新的访问控制类型，比如基于角色的访问控制。
下面的自界说授权器限制了只有内部监听器可以处理某些请求。
  1. package com.qupeng.demo.kafka.kafkaapache.security;
  3. import kafka.security.authorizer.AclAuthorizer;
  4. import org.apache.kafka.common.utils.Utils;
  5. import org.apache.kafka.server.authorizer.Action;
  6. import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
  7. import org.apache.kafka.server.authorizer.AuthorizationResult;
  9. import java.util.Collections;
  10. import java.util.List;
  11. import java.util.Set;
  13. import static org.apache.kafka.common.protocol.ApiKeys.CREATE_ACLS;
  14. import static org.apache.kafka.common.protocol.ApiKeys.DELETE_ACLS;
  15. import static org.apache.kafka.server.authorizer.AuthorizationResult.DENIED;
  17. public class CustomAuthorizer extends AclAuthorizer {
  18.     private static final Set<Short> internalOps = Utils.mkSet(CREATE_ACLS.id, DELETE_ACLS.id);
  19.     private static final String internalListener = "INTERNAL";
  21.     @Override
  22.     public List<AuthorizationResult> authorize(AuthorizableRequestContext context, List<Action> actions) {
  23.         // 如果不是内部请求，并且ACL是创建和删除的操作，禁止执行
  24.         if (!context.listenerName().equals(internalListener) &&
  25.                 internalOps.contains((short) context.requestType())) {
  26.             return Collections.nCopies(actions.size(), DENIED);
  27.         }  else {
  28.             // 如果是内部请求，执行默认的授权逻辑
  29.             return super.authorize(context, actions);
  30.         }
  31.     }
  32. }
复制代码
4.2 安全方面的考虑

5 审计

broker 可以生成用于审计和调试的log4j日志。可以在log4j.properties文件中指定日志级别、日志记载器及其配置选项。用于授权日志的kafka.authorizer.logger和用于请求日志的kafka.request.logger 可以单独配置日志级别和保存计谋。在生产环境中,可以用一些框架(如Elastic Stack)来分析和可视化这些日志。
授权器会记载INFO级别的拒绝访问日志和DEBUG级别的授权访问日志。
  1. DEBUG Principal = User:Alice is Allowed Operation = Write from host = 127.0.0.1
  2. on resource = Topic:LITERAL:customerOrders for request = Produce with resourceR-
  3. efCount =1 (kafka.authorizer.logger)
  4. INFO Principal = User:Mallory is Denied Operation = Describe from host =
  5. 10.0.0.13 on resource = Topic:LITERAL:customerOrders for request = Metadata
  6. with resourceRefCount = 1 (kafka.authorizer.logger)
复制代码
DEBUG级别的请求日志中包罗了用户主体和客户端主机的相干信息。如果请求日志设置了TRACE级别，那么日志中也会包罗请求细节。
  1. DEBUG Completed request:RequestHeader(apikey=PRODUCE, apiVersion=8,
  2. clientId=producer-1, correlationId=6) --
  3. {acks =- 1,timeout=30000,partitionSizes=[customerOrders-0=15514]},response:
  4. {responses=[{topic=customerOrders,partition_responses=[{partition=0,error_code=0
  5. ,base_offset=13,log_append_time =- 1,log_start_offset=0,record_errors=[],error_mes
  6. sage=null}]}],throttle_time_ms=0} from connection
  7. 127.0.0.1:9094-127.0.0.1:61040-0;totalTime:2.42,requestQueueTime:0.112,local-
  8. Time:2.15,remoteTime:0.0,throttleTime:0,responseQueueTime:0.04,sendTime:
  9. 0.118, securityProtocol: SASL_SSL,principal:User: Alice, listener : SASL_SSL, clientInf
  10. ormation:ClientInformation(softwareName=apache-kafka-java,
  11. softwareVersion=2.7.0-SNAPSHOT) (kafka.request.logger)
复制代码
可以通过分析授权器和请求日志来检测可疑活动。身份验证失败率指标和授权失败日志对审计来说非常有用，而且在遭到攻击或发生未授权访问时可以为我们提供有价值的信息。为了能够举行端到端的消息审计和追踪，可以在生成消息时将审计元数据放在消息标头中。端到端加密可用于保护元数据的完整性。
6 配额

请参考：
https://blog.csdn.net/yunyun1886358/article/details/135434017
7 Zookeeper安全

用于维护Kafka集群可用性的元数据就生存在ZooKeeper中，所以，除了保护Kafka，也
很有须要保护ZooKeeper。ZooKeeper支持基于SASL/GSSAPI的Kerberos身份验证和基于SASL/DIGEST-MD5的用户名和暗码身份验证。ZooKeeper 3.5.0中加入了对TLS的支持，可以举行双向身份验证和传输数据加密。需要注意是，SASL/DIGEST-MD5应该与TLS加密一起使用，并且不适合用在生产环境中，因为它存在已知的毛病。
7.1 SASL

ZooKeeper的SASL是通过Java系统属性 java.security.auth.login.config来配置的。
这个属性必须指向JAAS配置文件，此中包罗了一个登录模块以及与ZooKeeper服务器相干的配置参数。broker必须配置ZooKeeper客户端登录模块，以便与启用了SASL的ZooKeeper服务器通讯。下面的ZooKeeper服务器端JAAS配置启用了Kerberos身份验证。
  1.         Server {
  2.                 com.sun.security.auth.module.Krb5LoginModule required
  3.                 useKeyTab=true storeKey=true
  4.                 keyTab="/path/to/zk.keytab"
  5.                 principal="zookeeper/zk1.example.com@EXAMPLE.COM";
  6.         }:
复制代码
要启用ZooKeeper服务器端的SASL 身份验证，需要在ZooKeeper配置文件中配置身份验证提供程序。
  1. authProvider.sasl=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
  2. kerberos.removeHostFromPrincipal=true
  3. kerberos.removeRealmFromPrincipal=true
复制代码
broker 主体
在默认情况下,ZooKeeper用完整的Kerberos主体(如kafka/broker1.example.com@EXAMPLE.COM)来识别客户端。如果ZooKeeper的身份验证启用了ACL,那么ZooKeeper服务器端需要配置 kerberos.removeHostFromPrincipal=true和 kerberos.removeRealmFromPrincipal=true,确保所有的broker都有雷同的主体。
Kafka broker 需要启用到ZooKeeper的SASL身份验证，为此需要使用包罗客户端凭证的JAAS配置文件。
  1.         Client {
  2.                 com.sun.security.auth.module.Krb5LoginModule required
  3.                 useKeyTab=true storeKey=true
  4.                 keyTab="/path/to/broker1.keytab"
  5.                 principal="kafka/broker1.example.com@EXAMPLE.COM";
  6.         };
复制代码
7.2 SSL

ZooKeeper的每一个端点都应该启用SSL包罗使用SASL身份验证的那些。与Kafka一样，启用客户端身份验证需要配置SSL，但与Kafka不同，ZooKeeper会用SASL和SSL两种协议举行客户端身份验证，并将多个主体与连接关联起来。如果任意一个与某个连接关联的主体有访问某个资源的权限，那么ZooKeeper就会授予对这个资源的访问权限。
要为ZooKeeper服务器配置SSL，需要提供一个包罗服务器主机名或者通配符主机名的密钥存储。如果启用了客户端身份验证，则还需要提供一个用于验证客户端证书的信托存储。
  1. secureClientPort=2181
  2. serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
  3. authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
  4. ssl.keyStore.location=/path/to/zk.ks.p12
  5. ssl.keyStore. password=zk-ks-password
  6. ssl.keyStore.type=PKCS12
  7. ssl.trustStore.location=/path/to/zk.ts.p12
  8. ssl.trustStore.password=zk-ts-password
  9. ssl.trustStore.type=PKCS12
复制代码
要为Kafka配置SSL以便连接到ZooKeeper，需要提供一个用于验证ZooKeeper证书的信托存储。如果启用了客户端身份验证，则还需要提供一个密钥存储。
  1. zookeeper.ssl.client.enable=true
  2. zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
  3. zookeeper.ssl.keystore.location=/path/to/zkclient.ks.p12
  4. zookeeper.ssl.keystore.password=zkclient-ks-password
  5. zookeeper.ssl.keystore.type=PKCS12
  6. zookeeper.ssl.truststore.location=/path/to/zkclient.ts.p12
  7. zookeeper.ssl.truststore.password=zkclient-ts-password
  8. zookeeper.ssl.truststore.type=PKCS12
复制代码
7.3 授权

可以通过设置路径ACL来启用ZooKeeper节点授权。如果broker配置了 zookeeper.set.acl=true，就会在创建节点时设置ZooKeeper节点ACL。在默认情况下，元数据节点是可公开访问的，但只有broker能修改。如果内部管理员用户需要直接通过ZooKeeper修改元数据，那么需要添加额外的ACL。敏感路径(比如包罗SCRAM凭证的节点)在默认情况下是不公开的。
8 保护平台

前文讨论了如何保护Kafka和ZooKeeper。生产环境的系统所接纳的安全威胁模子不仅要涵盖单个组件的安全威胁，还要涵盖整个系统的安全威胁。威胁模子提供了一个系统抽象，用于识别埋伏的威胁和相干风险。在评估、记载并基于风险等级确定了威胁优先级之后，必须实施针对每个埋伏威胁的缓解计谋，以确保整个系统的安全。在评估埋伏威胁时，需要考虑外部威胁和内部威胁。对于存储PII或其他敏感数据的系统，还必须实施符合监管政策的步伐。不过，与威胁模子建模技能相干的内容超出了本章的讨论范围。
除了用安全的身份验证、授权和加密来保护Kafka中的数据和ZooKeeper中的元数据，还必须接纳额外的步伐来确保平台的安全性。可以用网络防火墙来保护网络，用加密来保护物理存储，用文件系统权限来保护包罗身份验证凭证的密钥存储、信托存储和Kerberos密钥表文件。必须对包罗安全关键信息(如凭证)的配置文件举行访问限制。由于在配置文件中生存明文暗码是不安全的(即使对访问权限举行了限制)，因此Kafka支持将暗码外部化到安全存储中。
8.1 保护暗码

可以为broker和客户端配置提供程序，用于从安全的第三方暗码存储库获取暗码。也可以将暗码加密后生存在配置文件中，并提供用于解密的提供程序。
下面的提供程序使用gpg来解密生存在文件中的broker或客户端属性。
  1. package com.qupeng.demo.kafka.kafkaapache.security;
  3. import org.apache.kafka.common.config.ConfigData;
  4. import org.apache.kafka.common.config.provider.ConfigProvider;
  5. import org.apache.kafka.common.utils.Shell;
  7. import java.io.IOException;
  8. import java.io.StringReader;
  9. import java.util.HashMap;
  10. import java.util.Map;
  11. import java.util.Properties;
  12. import java.util.Set;
  13. import java.util.stream.Collectors;
  15. public class GpgProvider implements ConfigProvider {
  17.     @Override
  18.     public void configure(Map<String, ?> configs) {
  19.     }
  21.     @Override
  22.     public void close() {
  23.     }
  25.     @Override
  26.     public ConfigData get(String path) {
  27.         try {
  28.             String passphrase = System.getenv("PASSPHRASE"); //通过环境变量PASSPHRASE来提供用于解密密码的密码短语。
  29.             String data = Shell.execCommand( //用gpg解密配置文件。返回值中包含了全部的解密配置参数。
  30.                     "gpg", " -- decrypt", " -- passphrase", passphrase, path);
  31.             Properties props = new Properties();
  32.             props.load(new StringReader(data)); //将配置数据解析成Java属性。
  33.             Map<String, String> map = new HashMap<>();
  34.             for (String name : props.stringPropertyNames())
  35.                 map.put(name, props.getProperty(name));
  36.             return new ConfigData(map);
  37.         } catch (IOException e) {
  38.             throw new RuntimeException(e); //如果遇到错误,就抛出RuntimeException异常。
  39.         }
  40.     }
  42.     @Override
  43.     public ConfigData get(String path, Set<String> keys) { //调用者可能想要获取指定文件中的部分属性,我们在这里加载所有属性,并返回调用者请求的部分。
  44.         ConfigData configData = get(path);
  45.         Map<String, String> data = configData.data().entrySet()
  46.                 .stream().filter(e -> keys.contains(e.getKey()))
  47.                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
  48.         return new ConfigData(data, configData.ttl());
  49.     }
  50. }
复制代码
可以用gpg来加密凭证文件：
  1. gpg -- symmetric -- output credentials.props.gpg \
  2. -- passphrase "$PASSPHRASE" credentials.props
复制代码
现在在原始属性文件中添加间接配置和提供程序,这样Kafka客户端就可以从被加密的文件中加载它们的凭证了。
  1. username=${gpg:/path/to/credentials.props.gpg:username}
  2. password=${gpg:/path/to/credentials.props.gpg:password}
  3. config.providers=gpg
  4. config.providers.gpg.class=com.example.GpgProvider
复制代码
也可以用Kafka配置工具将加密过的敏感broker配置信息生存在ZooKeeper中。可以在启动broker之前实行如下下令,以将加密过的SSL密钥存储暗码生存到ZooKeeper中。必须在每个broker的配置文件中配置用于解密的加密密钥。
  1. $ bin/kafka-configs.sh -- zookeeper localhost:2181 -- alter |
  2. -- entity-type brokers -- entity-name 0 -- add-config
  3. 'listener.name.external.ssl.keystore.password=server-ks-
  4. password,password.encoder.secret=encoder-secret'
复制代码
附录

1 JAAS

JAAS(Java Authentication Authorization Service)，即 Java 认证与授权，使用可插拔方式将认证与授权服务和应用程序分脱离，提供了灵活和可伸缩的机制来保证客户端或服务器端的 Java 程序；本文主要先容 JAAS 的根本概念及使用方法。
1.1 简介

Java 早期的安全框架夸大的是通过验证代码的来源和作者，保护用户避免受到下载下来的代码的攻击。JAAS 夸大的是通过验证谁在运行代码以及他的权限来保护系统免受攻击。它让你能够将一些标准的安全机制，例如 Solaris NIS（网络信息服务）、Windows NT、LDAP（轻量目次存取协议），Kerberos 等通过一种通用的，可配置的方式集成到系统当中去。
Java 安全框架最初会合在保护用户运行埋伏的不可信托代码，是基于代码的来源（URL）和谁创建的代码（certificate）来给移动代码举行授权。Java 2 SDK 1.3 引入了JAAS（ Java Authentication and Authorization Service）,增长了基于用户的访问控制本领，即根据谁在运行代码来举行授权。JAAS 已经整合进了Java 2 SDK 1.4，作为标准的用户认证与授权模子。
JAAS 认证被实现为可插拔的方式，允许应用程序同底层的具体认证技能保持独立，新增或者更新认证方法并不需要更改应用程序本身。应用程序通过实例化 LoginContext 对象开始认证过程，引用配置文件中的具体认证方法，即 LoginModule 对象，来实行认证。
1.2 核心类及接口

1.2.1 认证

Subject

认证的主体，可以代表一个人，一个角色一个历程等。认证乐成之后可以从 LoginContext 获取 Subject，代表一种身份，用户可以通过这个身份实行一些需要权限才能运行的逻辑。
Principals

可以为是一种权限。一个 Subject 可以包罗多个 Principal。用户认证乐成之后，把授予的 Principal 加入到和该用户关联的 Subject 中，该用户便具有了这个 Principal 的权限。
Credentials

凭证，包罗公共凭证(Public credentials，如：名称或公共密钥)和私有凭证(Private credentials，如：口令或私有密钥)。凭证并不是一个特定的类或捏词，它可以是任何对象。凭证中可以包罗任何特定安全系统需要的验证信息，例如标签（ticket），密钥或口令。
LoginContext

LoginContext 提供了对 Subject 举行身份验证的根本方法，并提供了一种不依赖于底层身份验证技能的应用程序开发方法。LoginContext 查询配置以确定为应用程序配置的身份验证服务或LoginModule。因此，可以在应用程序下插入不同的 Loginmodule，而不需要对应用程序本身举行任何修改。
LoginModule

登录模块，不同的 LoginModule 对应了不同的认证方式。例如：Krb5LoginModule 使用 Kerberos 作为认证方式，FileLoginModule 使用外部生存的暗码文件来认证，UsernameLoginModule 通过用户名暗码来认证。
CallbackHandler

在某些情况下，LoginModule 需要与用户通讯以获取身份验证信息；LoginModule 使用 CallbackHandler 来实现此目标。应用程序实现 CallbackHandler 接口并将其通报给 LoginContext, LoginContext 将其直接转发给底层 LoginModule。LoginModule 使用 CallbackHandler 来网络来自用户的输入(例如暗码或智能卡 pin 号)或向用户提供信息(例如状态信息)。应用重新通过指定 CallbackHandler 来实现与用户交互的各种不同方式。例如，GUI 应用程序CallbackHandler 的实现可能会显示一个窗口来征求用户的输入。另一方面，非 GUI 工具的 CallbackHandler 的实现可能只是简朴地提示用户直接从下令行输入。
Callback

javax.security.auth.callback 包中包罗了 Callback 接口及几个实现类。LoginModules 将一个 Callback 数组通报给 CallbackHandler 的 handle 方法。
1.2.2 授权

Policy

policy 类是一个抽象类，用于表现系统范围的访问控制计谋。policy 有对应的 policy 文件来配置访问权限。
AuthPermission

AuthPermission 类封装了 JAAS 所需的根本权限；AuthPermission包罗一个名称，但不包罗操作列表。
PrivateCredentialPermission

PrivateCredentialPermission 类用于保护对 Subject 私有凭证的访问。
1.3 配置文件

1.3.1 JAAS Login Configuration File

JAAS 登录配置文件，主要用于用户的认证；一个配置文件中可以包罗多个 entry，一个 entry 的格式如下：
  1. <name used by application to refer to this entry> {
  2.   <LoginModule> <flag> <LoginModule options>;
  3.   <optional additional LoginModules, flags and options>;
  4. };
复制代码
一个 entry 的配置主要包罗 LoginModule、flag、LoginModule options，一个配置例子如下：
  1. MyLogin {
  2.     com.abc.demo.general.jaas.MyLoginModule required;
  3. };
复制代码

   整个认证流程是否乐成的判断标准：
    
1.3.2 Policy File

policy 文件主要用来配置权限，其格式为：
  1. 复制代码
  2. grant signedBy "signer_names", codeBase "URL",
  3.       principal principal_class_name "principal_name",
  4.       principal principal_class_name "principal_name",
  5.       ... {
  7.     permission permission_class_name "target_name", "action",
  8.         signedBy "signer_names";
  9.     permission permission_class_name "target_name", "action",
  10.         signedBy "signer_names";
  11.     ...
  12. };
复制代码

  1. //读取d盘各级目录所有文件的权限
  2. permission java.io.FilePermission "d:/-", "read";
  4. //读取/设置环境变量的权限
  5. permission java.util.PropertyPermission "os.name", "read";
  6. permission java.util.PropertyPermission "java.security.auth.login.config", "write";
  7. 复制代码
  8. JAAS 支持很多种类型的权限，它们都继承至 java.security.Permission 类，可查看它的子类来确定具体的权限；常用的权限如下：
  10. AuthPermission：认证操作权限
  11. FilePermission：文件访问权限
  12. PropertyPermission：属性访问权限
  13. AllPermission：所有权限
  14. SocketPermission：网络通信权限
复制代码
1.3 示例代码

https://gitee.com/qp1886358/com.qupeng.demo.jaas
2 PGP

2.1 什么是 GPG

GPG 可以分成两部分来看：
第一部分：GPG 是一个加密、解密、署名、验证工具。前面先容了加解密的根本概念、相干的算法。但这些概念和算法并不能直接使用，而 GPG 就是一个使用这些算法，对信息举行加密、解密、署名、验证的工具。具体先容见如何使用部分。
第二部分：GPG 还是一个密钥管理工具，可以用于管理本身的私钥，其他人的公钥，以及提供了一套公钥信托体系。前面加解密概念中提到，公钥要广而告之，GPG 可以资助我们更加安全高效的互换公钥。具体先容见管理密钥部分。
别的在使用 GPG 时，可能还会看到 PGP、OpenPGP 等名词，他们是什么关系？PGP (Pretty Good Privacy) 是最早的于 1991 年发布的此类工具。PGP 很好用，但他是商业软件。于是 GNU 筹划在 1999 年发布了开源版本的 GNU Privacy Guard (GnuPG 或 GPG)。而 OpenPGP 是于 1997 年制定的一套标准，GPG、PGP 等工具都实现了这套标准。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/) Powered by Discuz! X3.4