Pulsar支持可插拔的身份认证和授权机制，Pulsar Proxy或者Pulsar Broker都支持该机制。认证和授权机制一起保证了客户端对Pulsar Topic、命名空间、租户的访问权限。

默认情况下，Pulsar并不会启用加密、身份认证和授权机制。之前我们使用pulsar-admin,pulsar-client等命令行工具，使用Java或Go语言开发的Consumer和Produmer，访问使用docker容器启动的单机Pulsar时，都未用到任何身份认证信息。 也就是说，Pulsar的默认配置是完全开放状态的，任何人都可以访问它。因此，必须启用Pulsar的认证和授权机制，为Pulsar各个组件开启安全防护。

本节先学习Pulsar的认证机制。

身份认证相关的概念

角色(Role): 在Pulsar中角色标识就是一个字符串，例如admin, app1，每个角色都标识了一个或多个Pulsar客户端。角色标识是身份认证和授权的主体(subject)。

身份认证提供者(Authentication Provider): Pulsar使用身份认证提供者来建立客户端的身份，然后将角色Token分配给该客户端。 当前Pulsar支持以下身份验证提供者:

• TLS认证
• Athenz
• Kerberos
• JWT认证

本节将只学习如何为Pulsar开启基于JWT的身份认证。

开启JWT Token认证

JWT支持通过两种不同的秘钥生成和验证Token：

• 对称秘钥：
• 使用单个Secret key来生成和验证Token
• 非对称秘钥：包含由私钥和公钥组成的一对密钥
• 使用Private key生成Token
• 使用Public key验证Token

这里使用非对称秘钥即公钥和私钥的形式。首先要由Pulsar的系统管理员生成密钥对。 在Pulsar的安装目录里使用下面的命令生成密钥对：

bin/pulsar tokens create-key-pair --output-private-key ./conf/jwt-private.key --output-public-key ./conf/jwt-public.key

这里生成密钥对jwt-private.key和jwt-public.key到了Pulsar安装目录的conf目录下，注意需要把jwt-private.key保存到一个安全的位置，确保只有管理员才能能使用它来创建新的Token。 jwt-public.key则会分发给所有的Broker，Broker将用它来验证Token。

先使用私钥来创建一个管理员用户的Token:

bin/pulsar tokens create --private-key ./conf/jwt-private.key --subject admin
eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiJ9.ZTR5pCrUXZ083QUvtW9wq5Otrh8PvLpi5L8xWJKVlgWS_U9PDjNStgQFLyx1cv7xlfEPwi3ep41MyMdDXWLzVvY_UolfPPpNpUiGHbdwchqgQ3bkdUubAoF4FxXNj22UweRDB_Jp_gz1VDc_4fiN85kjLgD7cS6XUM-D1Banq6Ll2GL532YJQkjAXk7DlWiWAYZIJP8tzNSGmlx8NzxwjFya1SxC-G3zsztfmkLFBYY6woHuBWAY62cJemHu9hRIj1qTagW5O3UIjtab92Ir44mT_YudfYcQA4g4k6eMeHSSny_k9v76x3WqTmUV4TImEUQN8ZfIYFrDTx92HVsVvQ

再使用私钥来创建一个普通用户的Token：

bin/pulsar tokens create --private-key ./conf/jwt-private.key --subject test-user

eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJ0ZXN0LXVzZXIifQ.ZXbMZ4OTXgJFN5tmw8jLd43spaiDYjbd4lmGTLd2_ge0C3V_lVsyv1VSGC5NNn5cik1rkQTS6dRHQ_GzczCCNNdvKr2z5zEFnCdqO8dRtFXliXALMwpno7epctKHa9TvRzozkmYw3zIKCXdjK9WEKkROyiJmqpCrKUOqXiEYH_oFMgBGAacy8VAkjSQlU8BhXGPhnKvLNolH41XRIkHP5EMcxaOC7NPQjDD-oucH2iCnz4AftBhGmredNBzReztZYN0-RvHCS-dwWwwrqO1Zxlnh1BfTSQblEkTMtaG_qiqRNg_gD6CGCp2oJdcv_p2pndgPmgPIuYpIGpgVP_HXGQ

注意这里的subject就是前面说的角色。另外创建token的时候还可以通过--expiry-time参数为token设置一个有效期，例如bin/pulsar tokens create --private-key ./conf/jwt-private.key --subject test-user --expiry-time 1y将创建一年有效期的Token。

上面的命令虽然已经创建了一个Token，但是Token本身并不会关联任何权限。Pulsar中的权限是和角色相关联的，可以使用pulsar-admin对角色(即Token中的主体)进行授权，例如下面的命令对普通用户test-user进行授权，允许test-user对my-tenant/my-namespace下的topic执行produce和consume操作：

bin/pulsar-admin namespaces grant-permission my-tenant/my-namespace --role test-user --actions produce,consume

在创建了Token，并对Token中的主体进行授权之后，就可以在Broker上开启认证了。要让Broker开启认证只需要在broker.conf配置文件中添加如下的配置:

# 启用认证和鉴权
authenticationEnabled=true
authorizationEnabled=true
authenticationRefreshCheckSeconds=60
authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken

# 设置Broker自身的认证, Broker连接其他Broker时用到
brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationToken
brokerClientAuthenticationParameters={"token":"eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiJ9.ZTR5pCrUXZ083QUvtW9wq5Otrh8PvLpi5L8xWJKVlgWS_U9PDjNStgQFLyx1cv7xlfEPwi3ep41MyMdDXWLzVvY_UolfPPpNpUiGHbdwchqgQ3bkdUubAoF4FxXNj22UweRDB_Jp_gz1VDc_4fiN85kjLgD7cS6XUM-D1Banq6Ll2GL532YJQkjAXk7DlWiWAYZIJP8tzNSGmlx8NzxwjFya1SxC-G3zsztfmkLFBYY6woHuBWAY62cJemHu9hRIj1qTagW5O3UIjtab92Ir44mT_YudfYcQA4g4k6eMeHSSny_k9v76x3WqTmUV4TImEUQN8ZfIYFrDTx92HVsVvQ"}

authenticateOriginalAuthData=true

superUserRoles=admin
tokenPublicKey=/pulsar/conf/jwt-public.key

上面的第9行配置的token为前面创建的admin的token，superUserRoles指定admin用户为超级用户。

注意集群部署的Pulsar，在Broker上开启认证修改的是各个Broker节点的broker.conf配置文件，对于使用docker容器启动的单机Pulsar，需要修改的配置文件是standalone.conf。

配置命令行工具

上面为broker开启认证以后，再使用pulsar-admin就会报认证授权失败的错误，例如下面查看命名空间权限配置的命令:

bin/pulsar-admin namespaces permissions my-tenant/my-namespace

HTTP 401 Unauthorized
Reason: HTTP 401 Unauthorized

需要在client.conf中做如下配置:

webServiceUrl=http://localhost:8080/
brokerServiceUrl=pulsar://localhost:6650/
authPlugin=org.apache.pulsar.client.impl.auth.AuthenticationToken
authParams=token:eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiJ9.ZTR5pCrUXZ083QUvtW9wq5Otrh8PvLpi5L8xWJKVlgWS_U9PDjNStgQFLyx1cv7xlfEPwi3ep41MyMdDXWLzVvY_UolfPPpNpUiGHbdwchqgQ3bkdUubAoF4FxXNj22UweRDB_Jp_gz1VDc_4fiN85kjLgD7cS6XUM-D1Banq6Ll2GL532YJQkjAXk7DlWiWAYZIJP8tzNSGmlx8NzxwjFya1SxC-G3zsztfmkLFBYY6woHuBWAY62cJemHu9hRIj1qTagW5O3UIjtab92Ir44mT_YudfYcQA4g4k6eMeHSSny_k9v76x3WqTmUV4TImEUQN8ZfIYFrDTx92HVsVvQ

因为这里是为命令行工具pulsar-admin访问身份，所以authParams中的token配置的是admin的token。

再次使用查看命名空间权限配置的命令测试，可以正常使用了:

bin/pulsar-admin namespaces permissions my-tenant/my-namespace
"test-user    [produce, consume]"

Pulsar的Java和Go客户端

Broker开启JWT认证后，在使用各个编程语言创建Pulsar Client时，就需要设置具体的Token了。

Java的示例代码如下:

PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://broker.example.com:6650/")
    .authentication(
        AuthenticationFactory.token("eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJ0ZXN0LXVzZXIifQ.ZXbMZ4OTXgJFN5tmw8jLd43spaiDYjbd4lmGTLd2_ge0C3V_lVsyv1VSGC5NNn5cik1rkQTS6dRHQ_GzczCCNNdvKr2z5zEFnCdqO8dRtFXliXALMwpno7epctKHa9TvRzozkmYw3zIKCXdjK9WEKkROyiJmqpCrKUOqXiEYH_oFMgBGAacy8VAkjSQlU8BhXGPhnKvLNolH41XRIkHP5EMcxaOC7NPQjDD-oucH2iCnz4AftBhGmredNBzReztZYN0-RvHCS-dwWwwrqO1Zxlnh1BfTSQblEkTMtaG_qiqRNg_gD6CGCp2oJdcv_p2pndgPmgPIuYpIGpgVP_HXGQ")）
    .build();

Go的示例代码如下:

client, err := NewClient(ClientOptions{
    URL:            "pulsar://localhost:6650",
    Authentication: NewAuthenticationToken("eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJ0ZXN0LXVzZXIifQ.ZXbMZ4OTXgJFN5tmw8jLd43spaiDYjbd4lmGTLd2_ge0C3V_lVsyv1VSGC5NNn5cik1rkQTS6dRHQ_GzczCCNNdvKr2z5zEFnCdqO8dRtFXliXALMwpno7epctKHa9TvRzozkmYw3zIKCXdjK9WEKkROyiJmqpCrKUOqXiEYH_oFMgBGAacy8VAkjSQlU8BhXGPhnKvLNolH41XRIkHP5EMcxaOC7NPQjDD-oucH2iCnz4AftBhGmredNBzReztZYN0-RvHCS-dwWwwrqO1Zxlnh1BfTSQblEkTMtaG_qiqRNg_gD6CGCp2oJdcv_p2pndgPmgPIuYpIGpgVP_HXGQ"),
})