[D] Custome authentication Plugin Not Working [pulsar]

["urdogan0000 (via GitHub)" <gi...@apache.org>]

GitHub user urdogan0000 created a discussion: Custome authentication Plugin Not Working

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;
    }
}

GitHub link: https://github.com/apache/pulsar/discussions/23270

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org