FIX Pulsar pubsub component initialization when using oauth2ClientSecretPath

common/authentication/oauth2/clientcredentials.go

@@ -17,10 +17,13 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
+	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -33,13 +36,93 @@ import (
 // ClientCredentialsMetadata is the metadata fields which can be used by a
 // component to configure an OIDC client_credentials token source.
 type ClientCredentialsMetadata struct {
-	TokenCAPEM       string   `mapstructure:"oauth2TokenCAPEM"`
-	TokenURL         string   `mapstructure:"oauth2TokenURL"`
-	ClientID         string   `mapstructure:"oauth2ClientID"`
-	ClientSecret     string   `mapstructure:"oauth2ClientSecret"`
-	ClientSecretPath string   `mapstructure:"oauth2ClientSecretPath"`
-	Audiences        []string `mapstructure:"oauth2Audiences"`
-	Scopes           []string `mapstructure:"oauth2Scopes"`
+	TokenCAPEM          string   `mapstructure:"oauth2TokenCAPEM"`
+	TokenURL            string   `mapstructure:"oauth2TokenURL"`
+	ClientID            string   `mapstructure:"oauth2ClientID"`
+	ClientSecret        string   `mapstructure:"oauth2ClientSecret"`
+	ClientSecretPath    string   `mapstructure:"oauth2ClientSecretPath"`
+	CredentialsFilePath string   `mapstructure:"oauth2CredentialsFile"`
+	Audiences           []string `mapstructure:"oauth2Audiences"`
+	Scopes              []string `mapstructure:"oauth2Scopes"`
+}
+
+// ResolveCredentials loads client_id and client_secret from files if configured.
+func (m *ClientCredentialsMetadata) ResolveCredentials() error {
+	if m.CredentialsFilePath != "" && m.ClientSecretPath != "" {
+		return errors.New("'oauth2CredentialsFile' and 'oauth2ClientSecretPath' fields are mutually exclusive")
+	}
+
+	if m.CredentialsFilePath != "" {
+		fileClientID, fileClientSecret, fileIssuerURL, err := LoadCredentialsFromJSONFile(m.CredentialsFilePath)
+		if err != nil {
+			return fmt.Errorf("failed to load credentials from JSON file: %w", err)
+		}
+
+		// Metadata overrides file values
+		if m.ClientID == "" {
+			m.ClientID = fileClientID
+		}
+		if m.ClientSecret == "" {
+			m.ClientSecret = fileClientSecret
+		}
+		if m.TokenURL == "" {
+			m.TokenURL = fileIssuerURL
+		}
+		return nil
+	}
+
+	if m.ClientSecretPath != "" {
+		// Metadata overrides file value
+		if m.ClientSecret == "" {
+			secretBytes, err := os.ReadFile(m.ClientSecretPath)
+			if err != nil {
+				return fmt.Errorf("could not read oauth2 client secret from file %q: %w", m.ClientSecretPath, err)
+			}
+			m.ClientSecret = strings.TrimSpace(string(secretBytes))
+		}
+		return nil
+	}
+
+	return nil
+}
+
+// ToOptions converts ClientCredentialsMetadata to ClientCredentialsOptions.
+func (m *ClientCredentialsMetadata) ToOptions(logger logger.Logger) ClientCredentialsOptions {
+	return ClientCredentialsOptions{
+		Logger:       logger,
+		TokenURL:     m.TokenURL,
+		CAPEM:        []byte(m.TokenCAPEM),
+		ClientID:     m.ClientID,
+		ClientSecret: m.ClientSecret,
+		Scopes:       m.Scopes,
+		Audiences:    m.Audiences,
+	}
+}
+
+// CredentialsFile represents a JSON credentials file.
+type CredentialsFile struct {
+	ClientID     string `json:"client_id"`
+	ClientSecret string `json:"client_secret"`
+	IssuerURL    string `json:"issuer_url"`
+}
+
+// LoadCredentialsFromJSONFile reads client_id, client_secret, and issuer_url from a JSON file.
+func LoadCredentialsFromJSONFile(filePath string) (clientID, clientSecret, issuerURL string, err error) {
+	secretBytes, err := os.ReadFile(filePath)
+	if err != nil {
+		return "", "", "", fmt.Errorf("could not read oauth2 credentials from file %q: %w", filePath, err)
+	}
+
+	var creds CredentialsFile
+	if err := json.Unmarshal(secretBytes, &creds); err != nil {
+		return "", "", "", fmt.Errorf("failed to parse JSON credentials file: %w", err)
+	}
+
+	if creds.ClientID == "" || creds.ClientSecret == "" || creds.IssuerURL == "" {
+		return "", "", "", errors.New("credentials file must contain client_id, client_secret, and issuer_url")
+	}
+
+	return creds.ClientID, creds.ClientSecret, creds.IssuerURL, nil
 }
 
 type ClientCredentialsOptions struct {

common/authentication/oauth2/clientcredentials_test.go

@@ -16,6 +16,7 @@ package oauth2
 import (
 	"context"
 	"net/url"
+	"os"
 	"testing"
 	"time"
 
@@ -116,3 +117,181 @@ func Test_TokenRenewal(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, "new-token", tok)
 }
+
+func TestLoadCredentialsFromJSONFile(t *testing.T) {
+	t.Run("valid JSON", func(t *testing.T) {
+		tmpFile, err := os.CreateTemp(t.TempDir(), "credentials-*.json")
+		require.NoError(t, err)
+		defer os.Remove(tmpFile.Name())
+
+		content := `{"client_id": "test-id", "client_secret": "test-secret", "issuer_url": "https://oauth.example.com/token"}`
+		_, err = tmpFile.WriteString(content)
+		require.NoError(t, err)
+		require.NoError(t, tmpFile.Close())
+
+		clientID, clientSecret, issuerURL, err := LoadCredentialsFromJSONFile(tmpFile.Name())
+		require.NoError(t, err)
+		assert.Equal(t, "test-id", clientID)
+		assert.Equal(t, "test-secret", clientSecret)
+		assert.Equal(t, "https://oauth.example.com/token", issuerURL)
+	})
+
+	t.Run("missing required fields", func(t *testing.T) {
+		tmpFile, err := os.CreateTemp(t.TempDir(), "credentials-*.json")
+		require.NoError(t, err)
+		defer os.Remove(tmpFile.Name())
+
+		_, err = tmpFile.WriteString(`{"client_id": "test-id"}`)
+		require.NoError(t, err)
+		require.NoError(t, tmpFile.Close())
+
+		_, _, _, err = LoadCredentialsFromJSONFile(tmpFile.Name())
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "must contain client_id, client_secret, and issuer_url")
+	})
+
+	t.Run("invalid JSON", func(t *testing.T) {
+		tmpFile, err := os.CreateTemp(t.TempDir(), "credentials-*.json")
+		require.NoError(t, err)
+		defer os.Remove(tmpFile.Name())
+
+		_, err = tmpFile.WriteString("{ invalid json }")
+		require.NoError(t, err)
+		require.NoError(t, tmpFile.Close())
+
+		_, _, _, err = LoadCredentialsFromJSONFile(tmpFile.Name())
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "failed to parse JSON")
+	})
+
+	t.Run("file not found", func(t *testing.T) {
+		_, _, _, err := LoadCredentialsFromJSONFile("/nonexistent/file/path")
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "could not read oauth2 credentials from file")
+	})
+}
+
+func TestClientCredentialsMetadata_ResolveCredentials(t *testing.T) {
+	t.Run("oauth2CredentialsFile with metadata override", func(t *testing.T) {
+		tmpFile, err := os.CreateTemp(t.TempDir(), "credentials-*.json")
+		require.NoError(t, err)
+		defer os.Remove(tmpFile.Name())
+
+		_, err = tmpFile.WriteString(`{"client_id": "file-id", "client_secret": "file-secret", "issuer_url": "https://file.com/token"}`)
+		require.NoError(t, err)
+		require.NoError(t, tmpFile.Close())
+
+		m := ClientCredentialsMetadata{
+			ClientID:            "meta-id",
+			ClientSecret:        "meta-secret",
+			TokenURL:            "https://meta.com/token",
+			CredentialsFilePath: tmpFile.Name(),
+		}
+
+		err = m.ResolveCredentials()
+		require.NoError(t, err)
+		assert.Equal(t, "meta-id", m.ClientID)                // metadata overrides
+		assert.Equal(t, "meta-secret", m.ClientSecret)        // metadata overrides
+		assert.Equal(t, "https://meta.com/token", m.TokenURL) // metadata overrides
+	})
+
+	t.Run("oauth2CredentialsFile without metadata", func(t *testing.T) {
+		tmpFile, err := os.CreateTemp(t.TempDir(), "credentials-*.json")
+		require.NoError(t, err)
+		defer os.Remove(tmpFile.Name())
+
+		_, err = tmpFile.WriteString(`{"client_id": "file-id", "client_secret": "file-secret", "issuer_url": "https://file.com/token"}`)
+		require.NoError(t, err)
+		require.NoError(t, tmpFile.Close())
+
+		m := ClientCredentialsMetadata{CredentialsFilePath: tmpFile.Name()}
+		err = m.ResolveCredentials()
+		require.NoError(t, err)
+		assert.Equal(t, "file-id", m.ClientID)
+		assert.Equal(t, "file-secret", m.ClientSecret)
+		assert.Equal(t, "https://file.com/token", m.TokenURL)
+	})
+
+	t.Run("oauth2ClientSecretPath with metadata override", func(t *testing.T) {
+		tmpFile, err := os.CreateTemp(t.TempDir(), "secret-*.txt")
+		require.NoError(t, err)
+		defer os.Remove(tmpFile.Name())
+
+		_, err = tmpFile.WriteString("file-secret")
+		require.NoError(t, err)
+		require.NoError(t, tmpFile.Close())
+
+		m := ClientCredentialsMetadata{
+			ClientID:         "meta-id",
+			ClientSecret:     "meta-secret",
+			ClientSecretPath: tmpFile.Name(),
+		}
+
+		err = m.ResolveCredentials()
+		require.NoError(t, err)
+		assert.Equal(t, "meta-id", m.ClientID)
+		assert.Equal(t, "meta-secret", m.ClientSecret) // metadata overrides
+	})
+
+	t.Run("oauth2ClientSecretPath without metadata", func(t *testing.T) {
+		tmpFile, err := os.CreateTemp(t.TempDir(), "secret-*.txt")
+		require.NoError(t, err)
+		defer os.Remove(tmpFile.Name())
+
+		_, err = tmpFile.WriteString("file-secret")
+		require.NoError(t, err)
+		require.NoError(t, tmpFile.Close())
+
+		m := ClientCredentialsMetadata{ClientSecretPath: tmpFile.Name()}
+		err = m.ResolveCredentials()
+		require.NoError(t, err)
+		assert.Equal(t, "file-secret", m.ClientSecret)
+	})
+
+	t.Run("error both fields set", func(t *testing.T) {
+		jsonFile, err := os.CreateTemp(t.TempDir(), "credentials-*.json")
+		require.NoError(t, err)
+		defer os.Remove(jsonFile.Name())
+		_, err = jsonFile.WriteString(`{"client_id": "id", "client_secret": "secret", "issuer_url": "https://example.com"}`)
+		require.NoError(t, err)
+		require.NoError(t, jsonFile.Close())
+
+		txtFile, err := os.CreateTemp(t.TempDir(), "secret-*.txt")
+		require.NoError(t, err)
+		defer os.Remove(txtFile.Name())
+		_, err = txtFile.WriteString("secret")
+		require.NoError(t, err)
+		require.NoError(t, txtFile.Close())
+
+		m := ClientCredentialsMetadata{
+			CredentialsFilePath: jsonFile.Name(),
+			ClientSecretPath:    txtFile.Name(),
+		}
+
+		err = m.ResolveCredentials()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "mutually exclusive")
+	})
+}
+
+func TestClientCredentialsMetadata_ToOptions(t *testing.T) {
+	logger := logger.NewLogger("test")
+	metadata := ClientCredentialsMetadata{
+		TokenURL:     "https://token.example.com",
+		TokenCAPEM:   "cert-pem-content",
+		ClientID:     "test-client-id",
+		ClientSecret: "test-client-secret",
+		Scopes:       []string{"scope1", "scope2"},
+		Audiences:    []string{"audience1"},
+	}
+
+	opts := metadata.ToOptions(logger)
+
+	assert.Equal(t, logger, opts.Logger)
+	assert.Equal(t, "https://token.example.com", opts.TokenURL)
+	assert.Equal(t, []byte("cert-pem-content"), opts.CAPEM)
+	assert.Equal(t, "test-client-id", opts.ClientID)
+	assert.Equal(t, "test-client-secret", opts.ClientSecret)
+	assert.Equal(t, []string{"scope1", "scope2"}, opts.Scopes)
+	assert.Equal(t, []string{"audience1"}, opts.Audiences)
+}

pubsub/pulsar/metadata.yaml

@@ -34,8 +34,13 @@ authenticationProfiles:
       - name: oauth2ClientSecretPath
         type: string
         description: |
-          The path to the OAuth Client Secret.
-        example: "/path/to/oauth2/client_secret.json"
+          The path to a plain text file containing the OAuth Client Secret.
+        example: "/path/to/oauth2/client_secret.txt"
+      - name: oauth2CredentialsFile
+        type: string
+        description: |
+          The path to a JSON file containing both client_id and client_secret.
+        example: "/path/to/oauth2/credentials.json"
       - name: oauth2TokenCAPEM
         type: string
         description: |

pubsub/pulsar/pulsar.go

@@ -213,6 +213,11 @@ func parsePulsarMetadata(meta pubsub.Metadata) (*pulsarMetadata, error) {
 		}
 	}
 
+	// Resolve credentials from file if ClientSecretPath is set
+	if err := m.ClientCredentialsMetadata.ResolveCredentials(); err != nil {
+		return nil, err
+	}
+
 	return &m, nil
 }
 
@@ -235,28 +240,13 @@ func (p *Pulsar) Init(ctx context.Context, metadata pubsub.Metadata) error {
 	case len(m.Token) > 0:
 		options.Authentication = pulsar.NewAuthenticationToken(m.Token)
 	case len(m.ClientCredentialsMetadata.TokenURL) > 0:
-		credsOpts := oauth2.ClientCredentialsOptions{
-			Logger:       p.logger,
-			TokenURL:     m.ClientCredentialsMetadata.TokenURL,
-			CAPEM:        []byte(m.ClientCredentialsMetadata.TokenCAPEM),
-			ClientID:     m.ClientCredentialsMetadata.ClientID,
-			ClientSecret: m.ClientCredentialsMetadata.ClientSecret,
-			Scopes:       m.ClientCredentialsMetadata.Scopes,
-			Audiences:    m.ClientCredentialsMetadata.Audiences,
-		}
-		if len(m.ClientCredentialsMetadata.ClientSecretPath) > 0 {
-			if _, err = oauth2.NewClientCredentials(ctx, credsOpts); err != nil {
-				return fmt.Errorf("could not instantiate oauth2 token provider: %w", err)
-			}
-			options.Authentication = pulsar.NewAuthenticationTokenFromFile(m.ClientSecretPath)
-		} else {
-			var cliCreds *oauth2.ClientCredentials
-			cliCreds, err = oauth2.NewClientCredentials(ctx, credsOpts)
-			if err != nil {
-				return fmt.Errorf("could not instantiate oauth2 token provider: %w", err)
-			}
-			options.Authentication = pulsar.NewAuthenticationTokenFromSupplier(cliCreds.Token)
+		credsOpts := m.ClientCredentialsMetadata.ToOptions(p.logger)
+		var cliCreds *oauth2.ClientCredentials
+		cliCreds, err = oauth2.NewClientCredentials(ctx, credsOpts)
+		if err != nil {
+			return fmt.Errorf("could not instantiate oauth2 token provider: %w", err)
 		}
+		options.Authentication = pulsar.NewAuthenticationTokenFromSupplier(cliCreds.Token)
 	}
 
 	client, err := p.newClientFn(options)