Custome authentication Plugin Not Working

[urdogan0000]

I create my customAuthenticationPlugin for broker and client.My pulsar container run and up with this custom AuthenticationPlugin when ı try to connect with client it gives an error like,
//
Caused by: org.apache.pulsar.client.api.PulsarClientException$UnsupportedAuthenticationException: org.apache.pulsar.client.api.PulsarClientException$UnsupportedAuthenticationException: java.lang.NullPointerException: Cannot invoke "com.google.gson.JsonObject.has(String)" because "params" is null
at org.apache.pulsar.client.api.AuthenticationFactory.create(AuthenticationFactory.java:92) ~[org.apache.pulsar-pulsar-client-api-3.2.3.jar:3.2.3]
((

My custom plugin

public class BasicAuthProvider implements AuthenticationProvider {
static final String HTTP_HEADER_NAME = "Authorization";
static final String HTTP_HEADER_VALUE_PREFIX = "Basic";
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();

    // Initialize the REST client and read configuration
    this.apiEndpoint = (String) config.getProperties().getOrDefault("authRestApiEndpoint", "http://localhost:8081/pulsar/send");

    log.info("BasicAuthProvider initialized with endpoint: {}", apiEndpoint);
}

@Override
public String getAuthMethodName() {
    return "customAuth";
}

@Override
public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
    String credentials = getUserCredentials(authData);

    // Log the incoming request and Authorization header
    log.info("Authentication request to endpoint: {}", apiEndpoint);
    log.info("Authorization header: {}", credentials);

    // Create a GET request with Basic Authentication header using OkHttp
    Request request = new Request.Builder()
            .url(apiEndpoint)
            .addHeader(HTTP_HEADER_NAME, credentials)  // The credentials already contain "Basic <encoded>"
            .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);
            // You can further parse the response if needed to determine success
            return responseBody;  // Return the result as per your application's requirements
        } else {
            if (log.isWarnEnabled()) {
                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.");
    }
}

public 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);
        log.info("Extracted HTTP header value: {}", httpHeaderValue);
        if (httpHeaderValue == null) {
            log.error("Extracted HTTP header value: {}", httpHeaderValue);
            throw new AuthenticationException("Invalid HTTP Authorization header");
        }
        return validateUserCredentials(httpHeaderValue);
    } else {
        log.error("Extracted HTTP header value: {}");
        throw new AuthenticationException("No user credentials passed");
    }
}

private static String validateUserCredentials(final String userCredentials) throws AuthenticationException {
    if (StringUtils.isNotBlank(userCredentials) ) {
        return userCredentials;
    } else {
        log.error("Extracted HTTP header value: {}",HTTP_HEADER_VALUE_PREFIX);
        log.error(userCredentials);
        log.error("test");
        throw new AuthenticationException("Invalid or blank user credentials found");
    }
}

@Override
public void close() {
    // Cleanup resources if needed
    if (httpClient != null) {
        httpClient.connectionPool().evictAll();  // Close all connections in the pool
    }
}

}

public class AuthenticationBasicAuth implements Authentication, EncodedAuthenticationParameterSupport {

private static final Logger log = LoggerFactory.getLogger(AuthenticationBasicAuth.class);
private String userId;
private String password;

// Default constructor
public AuthenticationBasicAuth() {
    // No initialization here, config will be called
}

public AuthenticationBasicAuth(String userId, String password) {
    this.userId = userId;
    this.password = password;
}

@Override
public void close() throws IOException {
    // No-op
}

@Override
public String getAuthMethodName() {
    return "customAuth";
}

@Override
public CustomDataBasic getAuthData() throws PulsarClientException {
    if (userId == null || password == null) {
        throw new PulsarClientException.InvalidConfigurationException("userId or password is not set");
    }
    return new CustomDataBasic(userId, password);
}

@Override
public void configure(Map<String, String> authParams) {
    // Configure using a Map
    configure(new Gson().toJson(authParams)); // Convert map to JSON string and delegate to the other configure method
}

@Override
public void configure(String encodedAuthParamString) {
    // Configure using an encoded JSON string
    try {
        JsonObject params = new Gson().fromJson(encodedAuthParamString, JsonObject.class);
        this.userId = params.has("userId") ? params.get("userId").getAsString() : null;
        this.password = params.has("password") ? params.get("password").getAsString() : null;

        // Log the received parameters for debugging
        log.debug("Configured with userId: {}, password: {}", userId, password != null ? "[PROTECTED]" : "null");

        // Validate the parameters
        if (userId == null || password == null) {
            throw new IllegalArgumentException("userId and password must be provided in the authentication parameters.");
        }
    } catch (com.google.gson.JsonSyntaxException e) {
        log.error("Invalid JSON format for authentication parameters: {}", encodedAuthParamString, e);
        throw new IllegalArgumentException("Invalid JSON format for authentication parameters.", e);
    }
}

@Override
public void start() throws PulsarClientException {
    // No-op
}

}

public class CustomDataBasic implements AuthenticationDataProvider {
private static final String HTTP_HEADER_NAME = "Authorization";
private final String commandAuthToken;
private final Map<String, String> headers;

public CustomDataBasic(String userId, String password) {
    // Base64 encode the credentials
    String credentials = userId + ":" + password;
   // String base64Credentials = Base64.getEncoder().encodeToString(credentials.getBytes());
    this.commandAuthToken = "Basic " + userId+":"+password;

    // Initialize headers
    this.headers = Collections.unmodifiableMap(new HashMap<String, String>(){
        {
            this.put(HTTP_HEADER_NAME, commandAuthToken);
            this.put(PULSAR_AUTH_METHOD_NAME, "customAuth");
        }
    });
}

@Override
public boolean hasDataForHttp() {
    return true;
}

@Override
public Set<Map.Entry<String, String>> getHttpHeaders() {
    return this.headers.entrySet();
}


@Override
public boolean hasDataFromCommand() {
    return true;
}

@Override
public String getCommandData() {
    return this.commandAuthToken;
}

}