Feature/amqp

.devcontainer/Dockerfile

@@ -0,0 +1,33 @@
+FROM mcr.microsoft.com/vscode/devcontainers/go:1.21
+
+ARG NODE_VERSION="16"
+ARG GOLANGCI_LINT_VERSION="1.57.1"
+
+# https://github.com/microsoft/vscode-dev-containers/blob/main/containers/go/.devcontainer/base.Dockerfile
+ENV USERNAME=vscode
+ENV LIBRARY_SCRIPTS_SRC="https://raw.githubusercontent.com/microsoft/vscode-dev-containers/main/containers/go/.devcontainer/library-scripts/node-debian.sh"
+ENV NVM_DIR=/usr/local/share/nvm
+ENV NVM_SYMLINK_CURRENT=true \
+  PATH=${NVM_DIR}/current/bin:${PATH}
+RUN mkdir /tmp/library-scripts \
+  && curl -fsSL -o /tmp/library-scripts/node-debian.sh "${LIBRARY_SCRIPTS_SRC}"
+RUN bash /tmp/library-scripts/node-debian.sh "${NVM_DIR}" "${NODE_VERSION}" "${USERNAME}" \
+  && apt-get clean -y && rm -rf /var/lib/apt/lists/* \
+  && rm -rf /tmp/library-scripts
+
+RUN echo 'deb [trusted=yes] https://repo.goreleaser.com/apt/ /' \
+  | tee /etc/apt/sources.list.d/goreleaser.list
+RUN apt-get update \
+  && export DEBIAN_FRONTEND=noninteractive \
+  && apt-get -y install --no-install-recommends goreleaser
+
+RUN curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh \
+  | sh -s -- -b $(go env GOPATH)/bin v$GOLANGCI_LINT_VERSION
+
+USER vscode
+WORKDIR /home/vscode
+
+RUN mkdir -p .config/git \
+  && echo ".vscode/*" >> .config/git/ignore \
+  && echo "*.code-workspace" >> .config/git/ignore \
+  && echo ".history/" >> .config/git/ignore

.devcontainer/devcontainer.json

@@ -0,0 +1,33 @@
+{
+  "name": "Benthos Dev Container",
+  "build": {
+    "dockerfile": "Dockerfile",
+    "args": {
+      "NODE_VERSION": "16",
+      "GOLANGCI_LINT_VERSION": "1.46.2"
+    }
+  },
+  "runArgs": ["--network=host", "--privileged"],
+  "customizations": {
+    // Configure properties specific to VS Code.
+    "vscode": {
+      "extensions": [
+        "golang.Go",
+        "dbaeumer.vscode-eslint",
+        "EditorConfig.EditorConfig",
+        "esbenp.prettier-vscode",
+        "github.vscode-github-actions",
+        "jebbs.plantuml",
+        "GitHub.copilot",
+        "github.vscode-github-actions",
+        "ms-vscode.makefile-tools",
+        "GeorgesHaidar.vsc-benthos"
+      ]
+    }
+  },
+  "features": {
+    "ghcr.io/devcontainers/features/docker-in-docker:2": {}
+  },
+  "postCreateCommand": "go mod download",
+  "remoteUser": "vscode"
+}

.github/workflows/cheetah_release.yaml

@@ -0,0 +1,41 @@
+name: Release cheetah-benthos
+
+on:
+  workflow_dispatch:
+
+  push:
+    branches: ["cheetah-main"]
+    tags:
+      - "v*"
+  pull_request:
+    branches: ["cheetah-main"]
+
+env:
+  IMAGE_NAME: ${{ github.repository }}
+  DOCKERFILE_PATH: ./resources/docker/Dockerfile
+  CONTEXT: ./
+
+permissions:
+  contents: read
+  packages: write
+
+jobs:
+  push_image:
+    name: "Build and push image"
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4
+
+      - name: Build and push
+        uses: trifork/cheetah-infrastructure-utils-workflows/.github/actions/build-image/default@main
+        with:
+          read_package_pat: ${{ secrets.PACKAGE_PAT }} # we need this, as GITHUB_TOKEN only have permission to its own repo
+          context: ${{ env.CONTEXT }}
+          image_name: ${{ env.IMAGE_NAME }}
+          github_run_id: ${{ github.run_id }}
+          dockerfile_path: ${{ env.DOCKERFILE_PATH }}
+          push_image: ${{ (github.event_name == 'pull_request' || github.ref_type == 'tag' || github.event_name == 'workflow_dispatch') && 'true' || 'false' }}
+          upload_image: "false"

internal/impl/kafka/sasl.go

@@ -2,8 +2,14 @@ package kafka
 
 import (
 	"context"
+	"encoding/base64"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
 
 	"github.com/IBM/sarama"
 
@@ -43,6 +49,18 @@ func saslField() *service.ConfigField {
 		service.NewStringField("token").
 			Description("The token to use for a single session's OAUTHBEARER authentication.").
 			Default(""),
+		service.NewStringField("tokenEndpoint").
+			Description("The endpoint to use for OAUTHBEARER token acquisition.").
+			Default(""),
+		service.NewStringField("clientId").
+			Description("The client ID to use for OAUTHBEARER token acquisition.").
+			Default(""),
+		service.NewStringField("clientSecret").
+			Description("The client secret to use for OAUTHBEARER token acquisition.").
+			Default("").Secret(),
+		service.NewStringField("scope").
+			Description("The scope to use for OAUTHBEARER token acquisition.").
+			Default(""),
 		service.NewStringMapField("extensions").
 			Description("Key/value pairs to add to OAUTHBEARER authentication requests.").
 			Optional(),
@@ -130,21 +148,98 @@ func plainSaslFromConfig(c *service.ParsedConfig) (sasl.Mechanism, error) {
 
 func oauthSaslFromConfig(c *service.ParsedConfig) (sasl.Mechanism, error) {
 	token, err := c.FieldString("token")
+
+	if err != nil && token != "" {
+		var extensions map[string]string
+		if c.Contains("extensions") {
+			if extensions, err = c.FieldStringMap("extensions"); err != nil {
+				return nil, err
+			}
+		}
+		return oauth.Oauth(func(c context.Context) (oauth.Auth, error) {
+			return oauth.Auth{
+				Token:      token,
+				Extensions: extensions,
+			}, nil
+		}), nil
+	} else if c.Contains("tokenEndpoint") {
+		return oauth.Oauth(func(ctx context.Context) (oauth.Auth, error) {
+			shortToken, err := acquireToken(ctx, c)
+			return oauth.Auth{Token: shortToken}, err
+		}), nil
+	}
+	return nil, errors.New("field 'token' or 'tokenEndpoint' was not found in the config")
+}
+
+func acquireToken(ctx context.Context, c *service.ParsedConfig) (string, error) {
+
+	tokenEndpoint, err := c.FieldString("tokenEndpoint")
 	if err != nil {
-		return nil, err
+		return "", err
 	}
-	var extensions map[string]string
-	if c.Contains("extensions") {
-		if extensions, err = c.FieldStringMap("extensions"); err != nil {
-			return nil, err
-		}
+
+	clientID, err := c.FieldString("clientId")
+	if err != nil {
+		return "", err
 	}
-	return oauth.Oauth(func(c context.Context) (oauth.Auth, error) {
-		return oauth.Auth{
-			Token:      token,
-			Extensions: extensions,
-		}, nil
-	}), nil
+
+	clientSecret, err := c.FieldString("clientSecret")
+	if err != nil {
+		return "", err
+	}
+
+	scope, err := c.FieldString("scope")
+	if err != nil {
+		return "", err
+	}
+
+	authHeaderValue := base64.StdEncoding.EncodeToString([]byte(clientID + ":" + clientSecret))
+
+	queryParams := url.Values{}
+	queryParams.Set("grant_type", "client_credentials")
+	queryParams.Set("scope", scope)
+
+	req, err := http.NewRequestWithContext(ctx, "POST", tokenEndpoint, strings.NewReader(queryParams.Encode()))
+	if err != nil {
+		return "", err
+	}
+
+	req.URL.RawQuery = queryParams.Encode()
+
+	req.Header.Set("Authorization", "Basic "+authHeaderValue)
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", err
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+
+	if err := resp.Body.Close(); err != nil {
+		return "", err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("token request failed with status code %d", resp.StatusCode)
+	}
+
+	var tokenResponse map[string]interface{}
+	err = json.Unmarshal(body, &tokenResponse)
+	if err != nil {
+		return "", fmt.Errorf("failed to parse token response: %s", err)
+	}
+
+	accessToken, ok := tokenResponse["access_token"].(string)
+	if !ok {
+		return "", errors.New("access_token not found in token response")
+	}
+
+	return accessToken, nil
 }
 
 func scram256SaslFromConfig(c *service.ParsedConfig) (sasl.Mechanism, error) {
@@ -223,6 +318,7 @@ func SaramaSASLField() *service.ConfigField {
 			Secret(),
 		service.NewStringField(saramaFieldSASLAccessToken).
 			Description("A static OAUTHBEARER access token").
+			Secret().
 			Default(""),
 		service.NewStringField(saramaFieldSASLTokenCache).
 			Description("Instead of using a static `access_token` allows you to query a [`cache`](/docs/components/caches/about) resource to fetch OAUTHBEARER tokens from").

internal/impl/kafka/sasl/sasl.go

@@ -33,7 +33,7 @@ func FieldSpec() docs.FieldSpec {
 		),
 		docs.FieldString("user", "A PLAIN username. It is recommended that you use environment variables to populate this field.", "${USER}"),
 		docs.FieldString("password", "A PLAIN password. It is recommended that you use environment variables to populate this field.", "${PASSWORD}").Secret(),
-		docs.FieldString("access_token", "A static OAUTHBEARER access token"),
+		docs.FieldString("access_token", "A static OAUTHBEARER access token").Secret(),
 		docs.FieldString("token_cache", "Instead of using a static `access_token` allows you to query a [`cache`](/docs/components/caches/about) resource to fetch OAUTHBEARER tokens from"),
 		docs.FieldString("token_key", "Required when using a `token_cache`, the key to query the cache with for tokens."),
 	).Advanced()