feat: add basic authentication

Dockerfile

@@ -30,6 +30,8 @@ COPY integration-tests/certs /pulsar/certs
 COPY integration-tests/tokens /pulsar/tokens
 COPY integration-tests/standalone.conf /pulsar/conf
 COPY integration-tests/client.conf /pulsar/conf
+COPY integration-tests/.htpasswd /pulsar/conf
+ENV PULSAR_EXTRA_OPTS="-Dpulsar.auth.basic.conf=/pulsar/conf/.htpasswd"
 COPY pulsar-test-service-start.sh /pulsar/bin
 COPY pulsar-test-service-stop.sh /pulsar/bin
 COPY run-ci.sh /pulsar/bin

integration-tests/.htpasswd

@@ -0,0 +1 @@
+admin:$apr1$FG4AO6aX$KGYPuMoLUou3i6vUkPUUf.

integration-tests/license_test.go

@@ -68,6 +68,7 @@ var skip = map[string]bool{
 	"../pulsar/internal/pulsar_proto/PulsarApi.pb.go": true,
 	"../.github/workflows/bot.yaml":                   true,
 	"../integration-tests/pb/hello.pb.go":             true,
+	"../integration-tests/.htpasswd":                  true,
 }
 
 func TestLicense(t *testing.T) {

integration-tests/standalone.conf

@@ -98,7 +98,7 @@ anonymousUserRole=anonymous
 authenticationEnabled=true
 
 # Autentication provider name list, which is comma separated list of class names
-authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls,org.apache.pulsar.broker.authentication.AuthenticationProviderToken
+authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls,org.apache.pulsar.broker.authentication.AuthenticationProviderToken,org.apache.pulsar.broker.authentication.AuthenticationProviderBasic
 
 # Enforce authorization
 authorizationEnabled=true

pulsar/client.go

@@ -78,6 +78,11 @@ func NewAuthenticationOAuth2(authParams map[string]string) Authentication {
 	return oauth
 }
 
+// NewAuthenticationBasic Creates Basic Authentication provider
+func NewAuthenticationBasic(username, password string) (Authentication, error) {
+	return auth.NewAuthenticationBasic(username, password)
+}
+
 // ClientOptions is used to construct a Pulsar Client instance.
 type ClientOptions struct {
 	// Configure the service URL for the Pulsar service.

pulsar/client_impl_test.go

@@ -28,6 +28,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
@@ -1019,3 +1021,48 @@ func TestHTTPOAuth2AuthFailed(t *testing.T) {
 
 	client.Close()
 }
+
+func TestHTTPBasicAuth(t *testing.T) {
+	basicAuth, err := NewAuthenticationBasic("admin", "123456")
+	require.NoError(t, err)
+	require.NotNil(t, basicAuth)
+
+	client, err := NewClient(ClientOptions{
+		URL:            webServiceURL,
+		Authentication: basicAuth,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, client)
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: newAuthTopicName(),
+	})
+
+	require.NoError(t, err)
+	require.NotNil(t, producer)
+
+	client.Close()
+}
+
+func TestHTTPSBasicAuth(t *testing.T) {
+	basicAuth, err := NewAuthenticationBasic("admin", "123456")
+	require.NoError(t, err)
+	require.NotNil(t, basicAuth)
+
+	client, err := NewClient(ClientOptions{
+		URL:                   webServiceURLTLS,
+		TLSTrustCertsFilePath: caCertsPath,
+		TLSValidateHostname:   true,
+		Authentication:        basicAuth,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, client)
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: newAuthTopicName(),
+	})
+	require.NoError(t, err)
+	require.NotNil(t, producer)
+
+	client.Close()
+}

pulsar/internal/auth/basic.go

@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package auth
+
+import (
+	"crypto/tls"
+	"encoding/base64"
+	"errors"
+	"net/http"
+)
+
+type basicAuthProvider struct {
+	rt               http.RoundTripper
+	commandAuthToken []byte
+	httpAuthToken    string
+}
+
+func NewAuthenticationBasic(username, password string) (Provider, error) {
+	if username == "" {
+		return nil, errors.New("username cannot be empty")
+	}
+	if password == "" {
+		return nil, errors.New("password cannot be empty")
+	}
+
+	commandAuthToken := []byte(username + ":" + password)
+	return &basicAuthProvider{
+		commandAuthToken: commandAuthToken,
+		httpAuthToken:    "Basic " + base64.StdEncoding.EncodeToString(commandAuthToken),
+	}, nil
+}
+
+func NewAuthenticationBasicWithParams(params map[string]string) (Provider, error) {
+	return NewAuthenticationBasic(params["username"], params["password"])
+}
+
+func (b *basicAuthProvider) Init() error {
+	return nil
+}
+
+func (b *basicAuthProvider) Name() string {
+	return "basic"
+}
+
+func (b *basicAuthProvider) GetTLSCertificate() (*tls.Certificate, error) {
+	return nil, nil
+}
+
+func (b *basicAuthProvider) GetData() ([]byte, error) {
+	return b.commandAuthToken, nil
+}
+
+func (b *basicAuthProvider) Close() error {
+	return nil
+}
+
+func (b *basicAuthProvider) RoundTrip(req *http.Request) (*http.Response, error) {
+	req.Header.Add("Authorization", b.httpAuthToken)
+	return b.rt.RoundTrip(req)
+}
+
+func (b *basicAuthProvider) Transport() http.RoundTripper {
+	return b.rt
+}
+
+func (b *basicAuthProvider) WithTransport(tr http.RoundTripper) error {
+	b.rt = tr
+	return nil
+}

pulsar/internal/auth/basic_test.go

@@ -0,0 +1,70 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package auth
+
+import (
+	"errors"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewAuthenticationBasicWithParams(t *testing.T) {
+	username := "admin"
+	password := "123456"
+
+	provider, err := NewAuthenticationBasic(username, password)
+	require.NoError(t, err)
+	require.NotNil(t, provider)
+
+	data, err := provider.GetData()
+	require.NoError(t, err)
+	require.Equal(t, []byte(username+":"+password), data)
+
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(r.Header.Get("Authorization")))
+	}))
+
+	client := s.Client()
+	err = provider.WithTransport(client.Transport)
+	require.NoError(t, err)
+	client.Transport = provider
+
+	resp, err := client.Get(s.URL)
+	require.NoError(t, err)
+
+	body, err := ioutil.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	require.NoError(t, err)
+	require.Equal(t, []byte("Basic YWRtaW46MTIzNDU2"), body)
+}
+
+func TestNewAuthenticationBasicWithInvalidParams(t *testing.T) {
+	username := "admin"
+	password := "123456"
+	provider, err := NewAuthenticationBasic("", password)
+	require.Equal(t, errors.New("username cannot be empty"), err)
+	require.Nil(t, provider)
+
+	provider, err = NewAuthenticationBasic(username, "")
+	require.Equal(t, errors.New("password cannot be empty"), err)
+	require.Nil(t, provider)
+}

pulsar/internal/auth/provider.go

@@ -80,6 +80,9 @@ func NewProvider(name string, params string) (Provider, error) {
 	case "oauth2", "org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2":
 		return NewAuthenticationOAuth2WithParams(m)
 
+	case "basic", "org.apache.pulsar.client.impl.auth.AuthenticationBasic":
+		return NewAuthenticationBasicWithParams(m)
+
 	default:
 		return nil, fmt.Errorf("invalid auth provider '%s'", name)
 	}