先决条件
—
在我们深入为 Pulsar 创建自定义身份验证机制之前,请确保您具有以下设置:
- Java 17: 确保您的环境中已安装并设置 Java 17。
- Spring Boot Version 3.3.2: 我们将使用 Spring Boot 创建自定义 Pulsar 客户端。
- Docker & Docker Compose: 在容器化环境中运行 Pulsar 代理所必需的。
- Maven: 用于构建和管理 Java 项目中的依赖关系。
Overview
—
在本指南中,我们将为 Apache Pulsar 创建一个自定义身份验证提供程序。我们的自定义提供程序将通过验证请求标头中发送到 REST API 的用户名和密码组合来处理身份验证。如果身份验证成功,将被授予访问 Pulsar 的权限;否则将被拒绝。
我们将按照官方 Pulsar 文档来扩展 Pulsar 的安全部分。
github repo: https://github.com/urdogan0000/PulsarCustomAuthSpringboot
Linked-in: https://www.linkedin.com/in/haydarurdogan/
第 1 步:设置环境
首先,让我们为自定义 Pulsar 代理身份验证设置一个新的 Maven 项目:
项目结构:
pulsar-custom-auth
│
├── src
│ ├── main
│ │ ├── java
│ │ │ └── com
│ │ │ └── liderahenkpulsar
│ │ │ └── auth
│ │ │ └── BasicAuthProvider.java
| | | |_ AuthenticationBasicAuth
| | | |__ CustomDataBasic
│ │ └── resources
│ │ └── application.properties
│ └── test
└── pom.xml
pom.xml:添加以下依赖项:
<dependencies>
<dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-broker-common</artifactId>
<version>3.2.3</version>
</dependency>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>okhttp</artifactId>
<version>4.12.0</version>
</dependency>
</dependencies>
第 2 步:创建自定义代理身份验证
现在,让我们通过实现 Pulsar 的 AuthenticationProvider
接口来创建 BasicAuthProvider
类:
BasicAuthProvider.java:
package com.liderahenkpulsar.auth;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.authentication.AuthenticationProvider;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.naming.AuthenticationException;
import java.io.IOException;
public class BasicAuthProvider implements AuthenticationProvider {
static final String HTTP_HEADER_NAME = "Authorization";
static final String AUTH_METHOD_NAME = "customAuth";
static final String HTTP_HEADER_VALUE_PREFIX = "Basic";
static final String REST_API_URL_NAME = "authRestApiEndpoint";
private static final Logger log = LoggerFactory.getLogger(BasicAuthProvider.class);
private String apiEndpoint;
private OkHttpClient httpClient;
@Override
public void initialize(ServiceConfiguration config) throws PulsarServerException {
httpClient = new OkHttpClient();
this.apiEndpoint = (String) config.getProperties().getOrDefault(REST_API_URL_NAME, "http://localhost:8081/pulsar/send");
log.info("BasicAuthProvider initialized with endpoint: {}", apiEndpoint);
}
@Override
public String getAuthMethodName() {
return AUTH_METHOD_NAME;
}
@Override
public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
String credentials = getUserCredentials(authData);
log.info("Authentication request to endpoint: {}", apiEndpoint);
log.info("Authorization header: {}", credentials);
Request request = new Request.Builder()
.url(apiEndpoint)
.addHeader(HTTP_HEADER_NAME, credentials)
.build();
try (Response response = httpClient.newCall(request).execute()) {
if (response.isSuccessful()) {
assert response.body() != null;
String responseBody = response.body().string();
log.info("Authentication successful: {}", responseBody);
return responseBody;
} else {
log.warn("Authentication failed. HTTP status code: {}, Response: {}",
response.code(),
response.body().string());
throw new AuthenticationException("Authentication failed. Invalid username or password.");
}
} catch (IOException e) {
log.error("Error during authentication: ", e);
throw new AuthenticationException("Authentication process encountered an error.");
}
}
private static String getUserCredentials(AuthenticationDataSource authData) throws AuthenticationException {
if (authData.hasDataFromCommand()) {
String commandData = authData.getCommandData();
log.info("Extracted command data: {}", commandData);
return validateUserCredentials(commandData);
} else if (authData.hasDataFromHttp()) {
String httpHeaderValue = authData.getHttpHeader(HTTP_HEADER_NAME);
if (httpHeaderValue == null) {
throw new AuthenticationException("Invalid HTTP Authorization header");
}
return validateUserCredentials(httpHeaderValue);
} else {
throw new AuthenticationException("No user credentials passed");
}
}
private static String validateUserCredentials(final String userCredentials) throws AuthenticationException {
if (StringUtils.isNotBlank(userCredentials)) {
return userCredentials;
} else {
throw new AuthenticationException("Invalid or blank user credentials found");
}
}
@Override
public void close() {
if (httpClient != null) {
httpClient.connectionPool().evictAll();
}
}
}
第 3 步:实施自定义客户端身份验证
为了实现自定义客户端,我们将创建两个类:
AuthenticationBasicAuth
:此类实现主要身份验证逻辑并向 Pulsar 代理提供凭证。
CustomDataBasic
:此类提供向 Pulsar 发出请求时进行身份验证所需的数据(标头)。
1. AuthenticationBasicAuth
Class
此类负责定义身份验证方法并管理用户凭据。它实现了 Authentication
和 EncodedAuthenticationParameterSupport
接口。
AuthenticationBasicAuth.java:
package com.liderahenkpulsar.auth;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.client.api.EncodedAuthenticationParameterSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Map;
public class AuthenticationBasicAuth implements Authentication, EncodedAuthenticationParameterSupport {
private static final Logger log = LoggerFactory.getLogger(AuthenticationBasicAuth.class);
private static final String AUTH_NAME = "customAuth"; // Ensure this matches your Pulsar broker's expectation
private String userId;
private String password;
// Default constructor for reflection or configuration usage
public AuthenticationBasicAuth() {
log.info("AuthenticationBasicAuth instantiated without parameters. Awaiting configuration.");
}
// Constructor to directly accept userId and password
public AuthenticationBasicAuth(String userId, String password) {
if (userId == null || userId.isEmpty() || password == null || password.isEmpty()) {
throw new IllegalArgumentException("User ID and password must not be null or empty");
}
this.userId = userId;
this.password = password;
log.info("AuthenticationBasicAuth instantiated with userId: {} and password: [PROTECTED]", userId);
}
@Override
public void close() throws IOException {
// No operation needed on close
}
@Override
public String getAuthMethodName() {
return AUTH_NAME;
}
@Override
public AuthenticationDataProvider getAuthData() {
return new CustomDataBasic(userId, password);
}
@Override
public void configure(Map<String, String> authParams) {
// No-op for map configuration
log.info("Configured with authParams: {}", authParams);
}
@Override
public void configure(String encodedAuthParamString) {
// No-op for encoded string configuration
log.info("Configured with encodedAuthParamString: {}", encodedAuthParamString);
}
@Override
public void start() {
log.info("Starting AuthenticationBasicAuth for userId: {}", userId);
}
}
方法说明:
- 构造函数:有两个构造函数 - 一个默认(无参数),当没有直接传递配置时,另一个接受“userId”和“password”。后者确保在继续之前正确设置这些参数。
**getAuthMethodName**
:返回身份验证方法名称,该名称应与自定义代理身份验证 (customAuth
) 中定义的方法名称匹配。
**getAuthData**
:提供 CustomDataBasic
的实例,其中包含身份验证标头。
**configure(Map<String, String> authParams)**
和 **configure(StringencodedAuthParamString)**
:这些方法允许您使用参数配置身份验证实例。它们在这里是无操作的,因为我们通过构造函数处理配置,但为了可见性而包含日志记录。
**start**
:初始化或启动身份验证过程,主要用于日志记录目的。
2. CustomDataBasic
Class
此类提供身份验证所需的实际数据,特别是处理 HTTP 请求的标头和命令数据。
CustomDataBasic.java:
package com.liderahenkpulsar.auth;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
public class CustomDataBasic implements AuthenticationDataProvider {
private static final String HTTP_HEADER_NAME = "Authorization";
private final String commandAuthToken;
private Map<String, String> headers = new HashMap<>();
public CustomDataBasic(String userId, String password) {
// Create the basic auth token
this.commandAuthToken = "Basic " + userId + ":" + password; // Ideally, base64 encode this string
// Initialize headers
headers.put(HTTP_HEADER_NAME, this.commandAuthToken);
this.headers = Collections.unmodifiableMap(this.headers);
}
@Override
public boolean hasDataForHttp() {
return true; // Indicate that HTTP headers are available
}
@Override
public Set<Map.Entry<String, String>> getHttpHeaders() {
return this.headers.entrySet(); // Return the HTTP headers for authentication
}
@Override
public boolean hasDataFromCommand() {
return true; // Indicate that command data is available
}
@Override
public String getCommandData() {
return this.commandAuthToken; // Return the command data for authentication
}
}
方法说明:
- 构造函数:通过连接前缀为
Basic
的 userId
和 password
来初始化 commandAuthToken
。在现实场景中,应该使用 Base64 进行编码。
**hasDataForHttp**
: 返回 true
,表示有数据(标头)可用于 HTTP 身份验证。
**getHttpHeaders**
:提供 Pulsar 身份验证所需的 HTTP 标头,包括 Authorization
标头。
**hasDataFromCommand**
:返回true
,表示命令数据(凭证)可用。
**getCommandData**
:返回身份验证令牌 (commandAuthToken
) 作为命令数据。
第 1 步:构建自定义身份验证 JAR 文件
创建 JAR 文件:确保您的 Java 项目已正确设置所有依赖项。使用 Maven 或 Gradle 编译项目并将其打包成 JAR 文件。
对于Maven:
mvn clean package
构建后,您应该有一个 JAR 文件,通常位于 target
目录中,例如 target/liderahenkpulsar.auth-1.0.jar
。
修改代理配置文件
下载原始 Broker 配置:从 Pulsar GitHub 存储库下载原始 broker.conf
:
wget https://raw.githubusercontent.com/apache/pulsar/branch-3.3/conf/broker.conf -O broker.conf
更新代理配置:编辑 broker.conf
以包含您的自定义身份验证设置。以下是需要修改的关键行:
### --- Authentication --- ###
authenticationEnabled=true
# Specify the authentication providers to use
authenticationProviders=com.liderahenkpulsar.auth.BasicAuthProvider
# Define the authentication method name that matches your AuthenticationBasicAuth class
authenticationMethods=customAuth
# (Optional) Specify the endpoint for your authentication service
authRestApiEndpoint=http://localhost:8083/pulsar/send
# Authentication settings of the broker itself. Used when the broker connects to other brokers,
# either in the same or other clusters
brokerClientAuthenticationPlugin=com.liderahenkpulsar.auth.AuthenticationBasicAuth
brokerClientAuthenticationParameters=
# (Optional) Specify the endpoint for your authentication service
authRestApiEndpoint=http://localhost:8083/pulsar/send //this part is for your login rest api servic
将更新的配置文件保存为 custom-auth-broker.conf
。
为 Pulsar 设置 Docker Compose
- 创建 Docker Compose 文件:
- 下面是完整的 Docker Compose 文件。将其保存为
docker-compose.yml
,位于 JAR 文件和更新的代理配置所在的同一目录中。
version: '3.8'
networks:
pulsar:
driver: bridge
services:
# Start Zookeeper
zookeeper:
image: apachepulsar/pulsar:3.3.1
container_name: zookeeper
restart: on-failure
networks:
- pulsar
volumes:
- ./data/zookeeper:/pulsar/data/zookeeper
environment:
- PULSAR_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m
command: >
bash -c "bin/apply-config-from-env.py conf/zookeeper.conf && \
bin/generate-zookeeper-config.sh conf/zookeeper.conf && \
exec bin/pulsar zookeeper"
healthcheck:
test: ["CMD", "bin/pulsar-zookeeper-ruok.sh"]
interval: 10s
timeout: 5s
retries: 30
# Init cluster metadata
pulsar-init:
image: apachepulsar/pulsar:3.3.1
container_name: pulsar-init
hostname: pulsar-init
networks:
- pulsar
command: >
bin/pulsar initialize-cluster-metadata \
--cluster cluster-a \
--zookeeper zookeeper:2181 \
--configuration-store zookeeper:2181 \
--web-service-url http://broker:8080 \
--broker-service-url pulsar://broker:6650
depends_on:
- zookeeper
# Start Bookie
bookie:
image: apachepulsar/pulsar:3.3.1
container_name: bookie
restart: on-failure
networks:
- pulsar
environment:
- clusterName=cluster-a
- zkServers=zookeeper:2181
- metadataServiceUri=metadata-store:zk:zookeeper:2181
- advertisedAddress=bookie
- BOOKIE_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m
depends_on:
- zookeeper
- pulsar-init
volumes:
- ./data/bookkeeper:/pulsar/data/bookkeeper
command: bash -c "bin/apply-config-from-env.py conf/bookkeeper.conf && exec bin/pulsar bookie"
# Start Broker
broker:
image: apachepulsar/pulsar:3.3.1
container_name: broker
hostname: broker
restart: on-failure
networks:
- pulsar
environment:
- metadataStoreUrl=zk:zookeeper:2181
- zookeeperServers=zookeeper:2181
- clusterName=cluster-a
- managedLedgerDefaultEnsembleSize=1
- managedLedgerDefaultWriteQuorum=1
- managedLedgerDefaultAckQuorum=1
- advertisedAddress=broker
- advertisedListeners=external:pulsar://172.16.102.12:6650
- PULSAR_MEM=-Xms1g -Xmx1g -XX:MaxDirectMemorySize=512m
depends_on:
- zookeeper
- bookie
ports:
- "6650:6650"
- "8080:8080"
volumes:
- ./custom-auth-broker.conf:/pulsar/conf/broker.conf
- ./liderahenkpulsar.auth-1.0.jar:/pulsar/lib/liderahenkpulsar.auth-1.0.jar
command: bash -c "bin/pulsar broker"
运行 Docker Compose 环境
运行 Docker Compose:使用 Docker Compose 启动具有自定义身份验证设置的 Pulsar 集群。
docker-compose up -d
验证设置:
- 检查日志以确保所有服务(Zookeeper、Bookie、Broker)正确启动。
- 确保代理加载自定义身份验证提供程序时不会出现错误。
设置 Spring Boot 项目
您可以使用 Spring Initializr 或具有以下依赖项的首选 IDE 创建新的 Spring Boot 项目:
- Spring Web:适用于 RESTful API(可选,根据您的用例)
- Spring Boot Starter:核心 Spring Boot 依赖项
- Pulsar 客户端:Java 版 Pulsar 客户端
添加依赖项
将 Pulsar 客户端和您的自定义身份验证 JAR 添加到 pom.xml
:
<dependencies>
<!-- Spring Boot Starter Dependencies -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<!-- Pulsar Client Dependency -->
<dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-client</artifactId>
<version>3.3.1</version>
</dependency>
<!-- Custom Authentication Dependency -->
<dependency>
<groupId>com.liderahenkpulsar</groupId>
<artifactId>liderahenkpulsar.auth</artifactId>
<version>1.0</version>
<scope>system</scope>
<systemPath>${project.basedir}/lib/liderahenkpulsar.auth-1.0.jar</systemPath>
</dependency>
</dependencies>
确保自定义 JAR 位于项目目录中的 lib
文件夹中。
使用自定义身份验证配置 Pulsar 客户端
为 Pulsar 客户端设置创建一个配置类:
package com.example.pulsarclient.config;
import com.liderahenkpulsar.auth.AuthenticationBasicAuth;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class PulsarConfig {
@Bean
public PulsarClient pulsarClient() throws Exception {
// Replace with the correct broker service URL
String serviceUrl = "pulsar://localhost:6650";
String userId = "your-username"; // Replace with your actual username
String password = "your-password"; // Replace with your actual password
// Create a Pulsar client with custom authentication
return PulsarClient.builder()
.serviceUrl(serviceUrl)
.authentication(new AuthenticationBasicAuth(userId, password))
.build();
}
}
当您运行应用程序时,您会看到 broker 的日志
2024–09–11T06:38:20,515+0000 [pulsar-io-3–7] INFO com.liderahenkpulsar.auth.BasicAuthProvider — Authentication request to endpoint: [http://172.16.102.215:8083/pulsar/send](http://172.16.102.215:8083/pulsar/send)
2024–09–11T06:38:20,515+0000 [pulsar-io-3–7] INFO com.liderahenkpulsar.auth.BasicAuthProvider — Authorization header: Basic testUser:testPass
2024–09–11T06:38:20,519+0000 [pulsar-io-3–7] INFO com.liderahenkpulsar.auth.BasicAuthProvider — Authentication successful: test
恭喜您可以通过自定义身份验证访问 pulsar。
原文地址