Authz for gateway and otlp_http inputs

internal/cli/enterprise.go

@@ -20,7 +20,9 @@ import (
 	"github.com/urfave/cli/v2"
 
 	"github.com/redpanda-data/benthos/v4/public/service"
+	"github.com/redpanda-data/common-go/authz"
 
+	"github.com/redpanda-data/connect/v4/internal/gateway"
 	"github.com/redpanda-data/connect/v4/internal/impl/kafka/enterprise"
 	"github.com/redpanda-data/connect/v4/internal/license"
 	"github.com/redpanda-data/connect/v4/internal/rpcplugin"
@@ -56,6 +58,8 @@ func InitEnterpriseCLI(binaryName, version, dateBuilt string, schema *service.Co
 		chrootPath        string
 		chrootPassthrough []string
 		disableTelemetry  bool
+		authzResourceName string
+		authzPolicyFile   string
 	)
 
 	flags := []cli.Flag{
@@ -134,7 +138,15 @@ func InitEnterpriseCLI(binaryName, version, dateBuilt string, schema *service.Co
 				}
 			}
 
-			// Kick off telemetry exporter.
+			// Store authorization configuration if present
+			if authzResourceName != "" && authzPolicyFile != "" {
+				gateway.SetManagerAuthzConfig(pConf.Resources(), gateway.AuthzConfig{
+					ResourceName: authz.ResourceName(authzResourceName),
+					PolicyFile:   authzPolicyFile,
+				})
+			}
+
+			// Kick off telemetry exporter
 			if !disableTelemetry {
 				telemetry.ActivateExporter(instanceID, version, fbLogger, schema, pConf)
 			}
@@ -185,7 +197,7 @@ func InitEnterpriseCLI(binaryName, version, dateBuilt string, schema *service.Co
 				}
 
 				// Parse and resolve cloud auth flags
-				if _, _, err := parseCloudAuthFlags(c.Context, c, secretLookupFn); err != nil {
+				if authzResourceName, authzPolicyFile, err = parseCloudAuthFlags(c.Context, c, secretLookupFn); err != nil {
 					return err
 				}
 

internal/gateway/authz.go

@@ -0,0 +1,126 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package gateway
+
+import (
+	"fmt"
+	"net/http"
+	"sync/atomic"
+
+	"github.com/redpanda-data/benthos/v4/public/service"
+	"github.com/redpanda-data/common-go/authz"
+	"github.com/redpanda-data/common-go/authz/loader"
+)
+
+// AuthzConfig holds the configuration for authorization policy.
+type AuthzConfig struct {
+	ResourceName authz.ResourceName
+	PolicyFile   string
+}
+
+const authzConfigKey = "rp_gateway_authz_config"
+
+// SetManagerAuthzConfig stores the authorization configuration in the resource
+// manager.
+func SetManagerAuthzConfig(mgr *service.Resources, conf AuthzConfig) {
+	mgr.SetGeneric(authzConfigKey, conf)
+}
+
+// ManagerAuthzConfig retrieves the authorization configuration from the
+// resource manager.
+func ManagerAuthzConfig(mgr *service.Resources) (AuthzConfig, bool) {
+	if c, ok := mgr.GetGeneric(authzConfigKey); ok {
+		return c.(AuthzConfig), true
+	}
+	return AuthzConfig{}, false
+}
+
+// FileWatchingAuthzResourcePolicy wraps an authorization policy that
+// automatically reloads when the underlying policy file changes.
+// Thread-safe for concurrent use.
+type FileWatchingAuthzResourcePolicy struct {
+	unwatch loader.PolicyUnwatch
+	value   atomic.Pointer[authz.ResourcePolicy]
+}
+
+// NewFileWatchingAuthzResourcePolicy loads an authorization policy from file and
+// watches it for changes. The notifyError callback is called on reload errors.
+func NewFileWatchingAuthzResourcePolicy(
+	name authz.ResourceName,
+	file string,
+	permissions []authz.PermissionName,
+	notifyError func(error),
+) (*FileWatchingAuthzResourcePolicy, error) {
+	a := new(FileWatchingAuthzResourcePolicy)
+
+	policy, unwatch, err := loader.WatchPolicyFile(file, func(policy authz.Policy, err error) {
+		if err != nil {
+			notifyError(fmt.Errorf("watching authorization policy file: %w", err))
+			return
+		}
+		rp, err := authz.NewResourcePolicy(policy, name, permissions)
+		if err != nil {
+			notifyError(fmt.Errorf("loading authorization policy: %w", err))
+			return
+		}
+		a.value.Store(rp)
+	})
+	if err != nil {
+		return nil, fmt.Errorf("load authorization policy file: %w", err)
+	}
+	a.unwatch = unwatch
+
+	rp, err := authz.NewResourcePolicy(policy, name, permissions)
+	if err != nil {
+		return nil, fmt.Errorf("compile authorization policy: %w", err)
+	}
+	a.value.Store(rp)
+
+	return a, nil
+}
+
+// Close closes the resource policy and stops watching the policy file.
+func (r *FileWatchingAuthzResourcePolicy) Close() error {
+	if r == nil {
+		return nil
+	}
+	return r.unwatch()
+}
+
+// Authorizer returns an [Authorizer] for this resource and the given permission.
+// The permission must have been provided to [NewFileWatchingAuthzResourcePolicy].
+func (r *FileWatchingAuthzResourcePolicy) Authorizer(perm authz.PermissionName) authz.Authorizer {
+	return r.value.Load().Authorizer(perm)
+}
+
+// SubResourceAuthorizer returns an [Authorizer] for a child resource and
+// the given permission. The permission must have been provided to
+// [NewFileWatchingAuthzResourcePolicy].
+func (r *FileWatchingAuthzResourcePolicy) SubResourceAuthorizer(t authz.ResourceType, id authz.ResourceID, perm authz.PermissionName) authz.Authorizer {
+	return r.value.Load().SubResourceAuthorizer(t, id, perm)
+}
+
+// AuthzMiddleware returns an HTTP middleware handler that enforces
+// authorization checks for the given permission before invoking the next
+// handler. If the principal is missing or unauthorized, it responds with
+// 403 Forbidden.
+func AuthzMiddleware(
+	policy *FileWatchingAuthzResourcePolicy,
+	perm authz.PermissionName,
+	next http.Handler,
+) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		principal, ok := ValidatedPrincipalIDFromContext(req.Context())
+		if !ok || !policy.Authorizer(perm).Check(principal) {
+			http.Error(w, "Forbidden", http.StatusForbidden)
+			return
+		}
+		next.ServeHTTP(w, req)
+	})
+}

internal/gateway/authz_test.go

@@ -0,0 +1,215 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package gateway_test
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/redpanda-data/common-go/authz"
+	"github.com/redpanda-data/connect/v4/internal/gateway"
+)
+
+const (
+	authzTestResourceName authz.ResourceName   = "organization/test-org/resourcegroup/default/dataplane/test-service"
+	authzTestPermRead     authz.PermissionName = "test_service_read"
+	authzTestPermWrite    authz.PermissionName = "test_service_write"
+	authzTestPrincipal    authz.PrincipalID    = "test@example.com"
+	authzOtherPrincipal   authz.PrincipalID    = "other@example.com"
+)
+
+// testHandler is a simple HTTP handler that writes "OK" on success
+var testHandler = http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+	w.WriteHeader(http.StatusOK)
+	_, _ = w.Write([]byte("OK"))
+})
+
+func TestAuthzMiddlewareAllowAll(t *testing.T) {
+	t.Log("Given: Policy file granting all permissions")
+	policy := setupPolicy(t, "testdata/policies/allow_all.yaml")
+	defer policy.Close()
+
+	t.Log("And: Middleware protecting a handler with read permission")
+	middleware := gateway.AuthzMiddleware(policy, authzTestPermRead, testHandler)
+
+	t.Log("When: Request with valid principal in context")
+	req := httptest.NewRequest(http.MethodGet, "/test", http.NoBody)
+	req = req.WithContext(gateway.ContextWithValidatedPrincipalID(req.Context(), authzTestPrincipal))
+	rec := httptest.NewRecorder()
+	middleware.ServeHTTP(rec, req)
+
+	t.Log("Then: Request succeeds")
+	assert.Equal(t, http.StatusOK, rec.Code)
+}
+
+func TestAuthzMiddlewareDenyAll(t *testing.T) {
+	t.Log("Given: Policy file denying all permissions")
+	policy := setupPolicy(t, "testdata/policies/deny_all.yaml")
+	defer policy.Close()
+
+	t.Log("And: Middleware protecting a handler with read permission")
+	middleware := gateway.AuthzMiddleware(policy, authzTestPermRead, testHandler)
+
+	t.Log("When: Request with valid principal but no permissions")
+	req := httptest.NewRequest(http.MethodGet, "/test", http.NoBody)
+	req = req.WithContext(gateway.ContextWithValidatedPrincipalID(req.Context(), authzTestPrincipal))
+	rec := httptest.NewRecorder()
+	middleware.ServeHTTP(rec, req)
+
+	t.Log("Then: Request is forbidden")
+	assert.Equal(t, http.StatusForbidden, rec.Code)
+	assert.Contains(t, rec.Body.String(), "Forbidden")
+}
+
+func TestAuthzMiddlewareNoPrincipal(t *testing.T) {
+	t.Log("Given: Policy file granting all permissions")
+	policy := setupPolicy(t, "testdata/policies/allow_all.yaml")
+	defer policy.Close()
+
+	t.Log("And: Middleware protecting a handler with read permission")
+	middleware := gateway.AuthzMiddleware(policy, authzTestPermRead, testHandler)
+
+	t.Log("When: Request without principal in context")
+	req := httptest.NewRequest(http.MethodGet, "/test", nil)
+	rec := httptest.NewRecorder()
+	middleware.ServeHTTP(rec, req)
+
+	t.Log("Then: Request is forbidden")
+	assert.Equal(t, http.StatusForbidden, rec.Code)
+	assert.Contains(t, rec.Body.String(), "Forbidden")
+}
+
+func TestAuthzMiddlewareSelective(t *testing.T) {
+	t.Log("Given: Policy file granting only read permission")
+	policy := setupPolicy(t, "testdata/policies/selective.yaml")
+	defer policy.Close()
+
+	tests := []struct {
+		name       string
+		permission authz.PermissionName
+		wantCode   int
+		wantBody   string
+	}{
+		{
+			name:       "allowed_read",
+			permission: authzTestPermRead,
+			wantCode:   http.StatusOK,
+			wantBody:   "OK",
+		},
+		{
+			name:       "denied_write",
+			permission: authzTestPermWrite,
+			wantCode:   http.StatusForbidden,
+			wantBody:   "Forbidden",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Logf("When: Request requires %s permission", tc.permission)
+			middleware := gateway.AuthzMiddleware(policy, tc.permission, testHandler)
+			req := httptest.NewRequest(http.MethodGet, "/test", nil)
+			req = req.WithContext(gateway.ContextWithValidatedPrincipalID(req.Context(), authzTestPrincipal))
+			rec := httptest.NewRecorder()
+			middleware.ServeHTTP(rec, req)
+
+			t.Logf("Then: Request %s", tc.wantBody)
+			assert.Equal(t, tc.wantCode, rec.Code)
+			assert.Contains(t, rec.Body.String(), tc.wantBody)
+		})
+	}
+}
+
+func TestAuthzMiddlewareWrongPrincipal(t *testing.T) {
+	t.Log("Given: Policy file granting permissions to specific principal")
+	policy := setupPolicy(t, "testdata/policies/allow_all.yaml")
+	defer policy.Close()
+
+	t.Log("And: Middleware protecting a handler with read permission")
+	middleware := gateway.AuthzMiddleware(policy, authzTestPermRead, testHandler)
+
+	t.Log("When: Request with different principal not in policy")
+	req := httptest.NewRequest(http.MethodGet, "/test", nil)
+	req = req.WithContext(gateway.ContextWithValidatedPrincipalID(req.Context(), authzOtherPrincipal))
+	rec := httptest.NewRecorder()
+	middleware.ServeHTTP(rec, req)
+
+	t.Log("Then: Request is forbidden")
+	assert.Equal(t, http.StatusForbidden, rec.Code)
+	assert.Contains(t, rec.Body.String(), "Forbidden")
+}
+
+func TestAuthzMiddlewarePolicyReload(t *testing.T) {
+	t.Log("Given: Policy file with allow_all")
+	dir := t.TempDir()
+	policyFile := filepath.Join(dir, "policy.yaml")
+
+	copyPolicyFile := func(src string) {
+		data, err := os.ReadFile(src)
+		require.NoError(t, err)
+		require.NoError(t, os.WriteFile(policyFile, data, 0o644))
+	}
+
+	copyPolicyFile("testdata/policies/allow_all.yaml")
+	policy := setupPolicy(t, policyFile)
+	defer policy.Close()
+
+	t.Log("And: Middleware protecting a handler with read permission")
+	middleware := gateway.AuthzMiddleware(policy, authzTestPermRead, testHandler)
+
+	t.Run("allow_all", func(t *testing.T) {
+		t.Log("When: Request with valid principal")
+		req := httptest.NewRequest(http.MethodGet, "/test", nil)
+		req = req.WithContext(gateway.ContextWithValidatedPrincipalID(req.Context(), authzTestPrincipal))
+		rec := httptest.NewRecorder()
+		middleware.ServeHTTP(rec, req)
+
+		t.Log("Then: Request succeeds")
+		assert.Equal(t, http.StatusOK, rec.Code)
+	})
+
+	t.Log("Given: Policy file updated to deny_all")
+	copyPolicyFile("testdata/policies/deny_all.yaml")
+	time.Sleep(100 * time.Millisecond)
+
+	t.Run("deny_all", func(t *testing.T) {
+		t.Log("When: Request with valid principal")
+		req := httptest.NewRequest(http.MethodGet, "/test", nil)
+		req = req.WithContext(gateway.ContextWithValidatedPrincipalID(req.Context(), authzTestPrincipal))
+		rec := httptest.NewRecorder()
+		middleware.ServeHTTP(rec, req)
+
+		t.Log("Then: Request fails")
+		assert.Equal(t, http.StatusForbidden, rec.Code)
+		assert.Contains(t, rec.Body.String(), "Forbidden")
+	})
+}
+
+// setupPolicy creates a FileWatchingAuthzResourcePolicy for testing
+func setupPolicy(t *testing.T, policyFile string) *gateway.FileWatchingAuthzResourcePolicy {
+	t.Helper()
+	policy, err := gateway.NewFileWatchingAuthzResourcePolicy(
+		authzTestResourceName,
+		policyFile,
+		[]authz.PermissionName{authzTestPermRead, authzTestPermWrite},
+		func(err error) {
+			t.Fatalf("Policy watch error: %v", err)
+		},
+	)
+	require.NoError(t, err)
+
+	return policy
+}

internal/gateway/testdata/policies/allow_all.yaml

@@ -0,0 +1,10 @@
+roles:
+  - id: test.admin
+    permissions:
+      - test_service_read
+      - test_service_write
+
+bindings:
+  - role: test.admin
+    principal: test@example.com
+    scope: organization/test-org/resourcegroup/default/dataplane/test-service