Files
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
c855109044 build(deps): bump github.com/twmb/franz-go/pkg/kadm (#5800)
Bumps [github.com/twmb/franz-go/pkg/kadm](https://github.com/twmb/franz-go) from 1.16.0 to 1.17.1.
- [Changelog](https://github.com/twmb/franz-go/blob/master/CHANGELOG.md)
- [Commits](https://github.com/twmb/franz-go/compare/v1.16.0...v1.17.1)

---
updated-dependencies:
- dependency-name: github.com/twmb/franz-go/pkg/kadm
  dependency-version: 1.17.1
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2025-11-05 08:38:06 +00:00

1171 lines
38 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package kadm
import (
"context"
"fmt"
"math"
"strings"
"sync"
"github.com/twmb/franz-go/pkg/kerr"
"github.com/twmb/franz-go/pkg/kmsg"
)
// ACLBuilder is a builder that is used for batch creating / listing / deleting
// ACLS.
//
// An ACL consists of five components:
//
// - the user (principal)
// - the host the user runs on
// - what resource to access (topic name, group id, etc.)
// - the operation (read, write)
// - whether to allow or deny the above
//
// This builder allows for adding the above five components in batches and then
// creating, listing, or deleting a batch of ACLs in one go. This builder
// merges the fifth component (allowing or denying) into allowing principals
// and hosts and denying principals and hosts. The builder must always have an
// Allow or Deny. For creating, the host is optional and defaults to the
// wildcard * that allows or denies all hosts. For listing / deleting, the host
// is also required (specifying no hosts matches all hosts, but you must
// specify this).
//
// Building works on a multiplying factor: every user, every host, every
// resource, and every operation is combined (principals * hosts * resources *
// operations).
//
// With the Kafka simple authorizer (and most reimplementations), all
// principals are required to have the "User:" prefix. The PrefixUserExcept
// function can be used to easily add the "User:" prefix if missing.
//
// The full set of operations and which requests require what operations is
// described in a large doc comment on the ACLOperation type.
//
// Lastly, resources to access / deny access to can be created / matched based
// on literal (exact) names, or on prefix names, or more. See the ACLPattern
// docs for more information.
type ACLBuilder struct {
any []string
anyResource bool
topics []string
anyTopic bool
groups []string
anyGroup bool
anyCluster bool
txnIDs []string
anyTxn bool
tokens []string
anyToken bool
allow []string
anyAllow bool
allowHosts []string
anyAllowHosts bool
deny []string
anyDeny bool
denyHosts []string
anyDenyHosts bool
ops []ACLOperation
pattern ACLPattern
}
// PrefixUser prefixes all allowed and denied principals with "User:".
func (b *ACLBuilder) PrefixUser() {
b.PrefixUserExcept()
}
// PrefixUserExcept prefixes all allowed and denied principals with "User:",
// unless they have any of the given except prefixes.
func (b *ACLBuilder) PrefixUserExcept(except ...string) {
replace := func(u string) string {
if !strings.HasPrefix(u, "User:") {
for _, e := range except {
if strings.HasPrefix(u, e) {
return u
}
}
return "User:" + u
}
return u
}
for i, u := range b.allow {
b.allow[i] = replace(u)
}
for i, u := range b.deny {
b.deny[i] = replace(u)
}
}
// NewACLs returns a new ACL builder.
func NewACLs() *ACLBuilder {
return new(ACLBuilder)
}
// AnyResource lists & deletes ACLs of any type matching the given names
// (pending other filters). If no names are given, this matches all names.
//
// This returns the input pointer.
//
// This function does nothing for creating.
func (b *ACLBuilder) AnyResource(name ...string) *ACLBuilder {
b.any = name
if len(name) == 0 {
b.anyResource = true
}
return b
}
// Topics lists/deletes/creates ACLs of resource type "topic" for the given
// topics.
//
// This returns the input pointer.
//
// For listing or deleting, if this is provided no topics, all "topic" resource
// type ACLs are matched. For creating, if no topics are provided, this
// function does nothing.
func (b *ACLBuilder) Topics(t ...string) *ACLBuilder {
b.topics = t
if len(t) == 0 {
b.anyTopic = true
}
return b
}
// MaybeTopics is the same as Topics, but does not match all topics if none are
// provided.
func (b *ACLBuilder) MaybeTopics(t ...string) *ACLBuilder { b.topics = t; return b }
// Groups lists/deletes/creates ACLs of resource type "group" for the given
// groups.
//
// This returns the input pointer.
//
// For listing or deleting, if this is provided no groups, all "group" resource
// type ACLs are matched. For creating, if no groups are provided, this
// function does nothing.
func (b *ACLBuilder) Groups(g ...string) *ACLBuilder {
b.groups = g
if len(g) == 0 {
b.anyGroup = true
}
return b
}
// MaybeGroups is the same as Groups, but does not match all groups if none are
// provided.
func (b *ACLBuilder) MaybeGroups(g ...string) *ACLBuilder { b.groups = g; return b }
// Clusters lists/deletes/creates ACLs of resource type "cluster".
//
// This returns the input pointer.
//
// There is only one type of cluster in Kafka, "kafka-cluster". Opting in to
// listing or deleting by cluster inherently matches all ACLS of resource type
// cluster. For creating, this function allows for creating cluster ACLs.
func (b *ACLBuilder) Clusters() *ACLBuilder {
b.anyCluster = true
return b
}
// MaybeClusters is the same as Clusters, but only matches clusters if c is
// true.
func (b *ACLBuilder) MaybeClusters(c bool) *ACLBuilder { b.anyCluster = c; return b }
// TransactionalIDs lists/deletes/creates ACLs of resource type
// "transactional_id" for the given transactional IDs.
//
// This returns the input pointer.
//
// For listing or deleting, if this is provided no IDs, all "transactional_id"
// resource type ACLs matched. For creating, if no IDs are provided, this
// function does nothing.
func (b *ACLBuilder) TransactionalIDs(x ...string) *ACLBuilder {
b.txnIDs = x
if len(x) == 0 {
b.anyTxn = true
}
return b
}
// MaybeTransactionalIDs is the same as TransactionalIDs, but does not match
// all transactional ID's if none are provided.
func (b *ACLBuilder) MaybeTransactionalIDs(x ...string) *ACLBuilder { b.txnIDs = x; return b }
// DelegationTokens lists/deletes/creates ACLs of resource type
// "delegation_token" for the given delegation tokens.
//
// This returns the input pointer.
//
// For listing or deleting, if this is provided no tokens, all
// "delegation_token" resource type ACLs are matched. For creating, if no
// tokens are provided, this function does nothing.
func (b *ACLBuilder) DelegationTokens(t ...string) *ACLBuilder {
b.tokens = t
if len(t) == 0 {
b.anyToken = true
}
return b
}
// MaybeDelegationTokens is the same as DelegationTokens, but does not match
// all tokens if none are provided.
func (b *ACLBuilder) MaybeDelegationTokens(t ...string) *ACLBuilder { b.tokens = t; return b }
// Allow sets the principals to add allow permissions for. For listing and
// deleting, you must also use AllowHosts.
//
// This returns the input pointer.
//
// For creating, if this is not paired with AllowHosts, the user will have
// access to all hosts (the wildcard *).
//
// For listing & deleting, if the principals are empty, this matches any user.
func (b *ACLBuilder) Allow(principals ...string) *ACLBuilder {
b.allow = principals
if len(principals) == 0 {
b.anyAllow = true
}
return b
}
// MaybeAllow is the same as Allow, but does not match all allowed principals
// if none are provided.
func (b *ACLBuilder) MaybeAllow(principals ...string) *ACLBuilder { b.allow = principals; return b }
// AllowHosts sets the hosts to add allow permissions for. If using this, you
// must also use Allow.
//
// This returns the input pointer.
//
// For creating, if this is empty, the user will have access to all hosts (the
// wildcard *) and this function is actually not necessary.
//
// For listing & deleting, if the hosts are empty, this matches any host.
func (b *ACLBuilder) AllowHosts(hosts ...string) *ACLBuilder {
b.allowHosts = hosts
if len(hosts) == 0 {
b.anyAllowHosts = true
}
return b
}
// MaybeAllowHosts is the same as AllowHosts, but does not match all allowed
// hosts if none are provided.
func (b *ACLBuilder) MaybeAllowHosts(hosts ...string) *ACLBuilder { b.allowHosts = hosts; return b }
// Deny sets the principals to add deny permissions for. For listing and
// deleting, you must also use DenyHosts.
//
// This returns the input pointer.
//
// For creating, if this is not paired with DenyHosts, the user will be denied
// access to all hosts (the wildcard *).
//
// For listing & deleting, if the principals are empty, this matches any user.
func (b *ACLBuilder) Deny(principals ...string) *ACLBuilder {
b.deny = principals
if len(principals) == 0 {
b.anyDeny = true
}
return b
}
// MaybeDeny is the same as Deny, but does not match all denied principals if
// none are provided.
func (b *ACLBuilder) MaybeDeny(principals ...string) *ACLBuilder { b.deny = principals; return b }
// DenyHosts sets the hosts to add deny permissions for. If using this, you
// must also use Deny.
//
// This returns the input pointer.
//
// For creating, if this is empty, the user will be denied access to all hosts
// (the wildcard *) and this function is actually not necessary.
//
// For listing & deleting, if the hosts are empty, this matches any host.
func (b *ACLBuilder) DenyHosts(hosts ...string) *ACLBuilder {
b.denyHosts = hosts
if len(hosts) == 0 {
b.anyDenyHosts = true
}
return b
}
// MaybeDenyHosts is the same as DenyHosts, but does not match all denied
// hosts if none are provided.
func (b *ACLBuilder) MaybeDenyHosts(hosts ...string) *ACLBuilder { b.denyHosts = hosts; return b }
// ACLOperation is a type alias for kmsg.ACLOperation, which is an enum
// containing all Kafka ACL operations and has helper functions.
//
// Kafka requests require the following operations (broker <=> broker ACLs
// elided):
//
// PRODUCING/CONSUMING
// ===================
// Produce WRITE on TOPIC for topics
// WRITE on TRANSACTIONAL_ID for txn id (if transactionally producing)
//
// Fetch READ on TOPIC for topics
//
// ListOffsets DESCRIBE on TOPIC for topics
//
// Metadata DESCRIBE on TOPIC for topics
// CREATE on CLUSTER for kafka-cluster (if automatically creating new topics)
// CREATE on TOPIC for topics (if automatically creating new topics)
//
// OffsetForLeaderEpoch DESCRIBE on TOPIC for topics
//
// GROUPS
// ======
// FindCoordinator DESCRIBE on GROUP for group (if finding group coordinator)
// DESCRIBE on TRANSACTIONAL_ID for id (if finding transactiona coordinator)
//
// OffsetCommit READ on GROUP for group
// READ on TOPIC for topics
//
// OffsetFetch DESCRIBE on GROUP for group
// DESCRIBE on TOPIC for topics
//
// OffsetDelete DELETE on GROUP For group
// READ on TOPIC for topics
//
// JoinGroup READ on GROUP for group
// Heartbeat READ on GROUP for group
// LeaveGroup READ on GROUP for group
// SyncGroup READ on GROUP for group
//
// DescribeGroup DESCRIBE on GROUP for groups
//
// ListGroups DESCRIBE on GROUP for groups
// or, DESCRIBE on CLUSTER for kafka-cluster
//
// DeleteGroups DELETE on GROUP for groups
//
// TRANSACTIONS (including FindCoordinator above)
// ============
// InitProducerID WRITE on TRANSACTIONAL_ID for id, if using transactions
// or, IDEMPOTENT_WRITE on CLUSTER for kafka-cluster, if pre Kafka 3.0
// or, WRITE on TOPIC for any topic, if Kafka 3.0+
//
// AddPartitionsToTxn WRITE on TRANSACTIONAL_ID for id
// WRITE on TOPIC for topics
//
// AddOffsetsToTxn WRITE on TRANSACTIONAL_ID for id
// READ on GROUP for group
//
// EndTxn WRITE on TRANSACTIONAL_ID for id
//
// TxnOffsetCommit WRITE on TRANSACTIONAL_ID for id
// READ on GROUP for group
// READ on TOPIC for topics
//
// TOPIC ADMIN
// ===========
// CreateTopics CREATE on CLUSTER for kafka-cluster
// CREATE on TOPIC for topics
// DESCRIBE_CONFIGS on TOPIC for topics, for returning topic configs on create
//
// CreatePartitions ALTER on TOPIC for topics
//
// DeleteTopics DELETE on TOPIC for topics
// DESCRIBE on TOPIC for topics, if deleting by topic id (in addition to prior ACL)
//
// DeleteRecords DELETE on TOPIC for topics
//
// CONFIG ADMIN
// ============
// DescribeConfigs DESCRIBE_CONFIGS on CLUSTER for kafka-cluster, for broker or broker-logger describing
// DESCRIBE_CONFIGS on TOPIC for topics, for topic describing
//
// AlterConfigs ALTER_CONFIGS on CLUSTER for kafka-cluster, for broker altering
// ALTER_CONFIGS on TOPIC for topics, for topic altering
//
// IncrementalAlterConfigs ALTER_CONFIGS on CLUSTER for kafka-cluster, for broker or broker-logger altering
// ALTER_CONFIGS on TOPIC for topics, for topic altering
//
//
// MISC ADMIN
// ==========
// AlterReplicaLogDirs ALTER on CLUSTER for kafka-cluster
// DescribeLogDirs DESCRIBE on CLUSTER for kafka-cluster
//
// AlterPartitionAssignments ALTER on CLUSTER for kafka-cluster
// ListPartitionReassignments DESCRIBE on CLUSTER for kafka-cluster
//
// DescribeDelegationTokens DESCRIBE on DELEGATION_TOKEN for id
//
// ElectLeaders ALTER on CLUSTER for kafka-cluster
//
// DescribeClientQuotas DESCRIBE_CONFIGS on CLUSTER for kafka-cluster
// AlterClientQuotas ALTER_CONFIGS on CLUSTER for kafka-cluster
//
// DescribeUserScramCredentials DESCRIBE on CLUSTER for kafka-cluster
// AlterUserScramCredentials ALTER on CLUSTER for kafka-cluster
//
// UpdateFeatures ALTER on CLUSTER for kafka-cluster
//
// DescribeCluster DESCRIBE on CLUSTER for kafka-cluster
//
// DescribeProducerIDs READ on TOPIC for topics
// DescribeTransactions DESCRIBE on TRANSACTIONAL_ID for ids
// DESCRIBE on TOPIC for topics
// ListTransactions DESCRIBE on TRANSACTIONAL_ID for ids
type ACLOperation = kmsg.ACLOperation
const (
// OpUnknown is returned for unknown operations.
OpUnknown ACLOperation = kmsg.ACLOperationUnknown
// OpAny, used for listing and deleting, matches any operation.
OpAny ACLOperation = kmsg.ACLOperationAny
// OpAll is a shortcut for allowing / denying all operations.
OpAll ACLOperation = kmsg.ACLOperationAll
// OpRead is the READ operation.
OpRead ACLOperation = kmsg.ACLOperationRead
// OpWrite is the WRITE operation.
OpWrite ACLOperation = kmsg.ACLOperationWrite
// OpCreate is the CREATE operation.
OpCreate ACLOperation = kmsg.ACLOperationCreate
// OpDelete is the DELETE operation.
OpDelete ACLOperation = kmsg.ACLOperationDelete
// OpAlter is the ALTER operation.
OpAlter ACLOperation = kmsg.ACLOperationAlter
// OpDescribe is the DESCRIBE operation.
OpDescribe ACLOperation = kmsg.ACLOperationDescribe
// OpClusterAction is the CLUSTER_ACTION operation. This operation is
// used for any broker<=>broker communication and is not needed by
// clients.
OpClusterAction ACLOperation = kmsg.ACLOperationClusterAction
// OpDescribeConfigs is the DESCRIBE_CONFIGS operation.
OpDescribeConfigs ACLOperation = kmsg.ACLOperationDescribeConfigs
// OpAlterConfigs is the ALTER_CONFIGS operation.
OpAlterConfigs ACLOperation = kmsg.ACLOperationAlterConfigs
// OpIdempotentWrite is the IDEMPOTENT_WRITE operation. As of Kafka
// 3.0+, this has been deprecated and replaced by the ability to WRITE
// on any topic.
OpIdempotentWrite ACLOperation = kmsg.ACLOperationIdempotentWrite
)
// Operations sets operations to allow or deny. Passing no operations defaults
// to OpAny.
//
// This returns the input pointer.
//
// For creating, OpAny returns an error, for it is strictly used for filters
// (listing & deleting).
func (b *ACLBuilder) Operations(operations ...ACLOperation) *ACLBuilder {
b.ops = operations
if len(operations) == 0 {
b.ops = []ACLOperation{OpAny}
}
return b
}
// MaybeOperations is the same as Operations, but does not match all operations
// if none are provided.
func (b *ACLBuilder) MaybeOperations(operations ...ACLOperation) *ACLBuilder {
if len(operations) > 0 {
b.Operations(operations...)
}
return b
}
// ACLPattern is a type alias for kmsg.ACLResourcePatternType, which is an enum
// containing all Kafka ACL resource pattern options.
//
// Creating/listing/deleting ACLs works on a resource name basis: every ACL
// created has a name, and every ACL filtered for listing / deleting matches by
// name. The name by default is "literal", meaning created ACLs will have the
// exact name, and matched ACLs must match completely.
//
// Prefixed names allow for creating an ACL that matches any prefix: principals
// foo-bar and foo-baz both have the prefix "foo-", meaning a READ on TOPIC for
// User:foo- with prefix pattern will allow both of those principals to read
// the topic.
//
// Any and match are used for listing and deleting. Any will match any name, be
// it literal or prefix or a wildcard name. There is no need for specifying
// topics, groups, etc. when using any resource pattern.
//
// Alternatively, match requires a name, but it matches any literal name (exact
// match), any prefix, and any wildcard.
type ACLPattern = kmsg.ACLResourcePatternType
const (
// ACLPatternUnknown is returned for unknown patterns.
ACLPatternUnknown ACLPattern = kmsg.ACLResourcePatternTypeUnknown
// ACLPatternAny is the ANY resource pattern.
ACLPatternAny ACLPattern = kmsg.ACLResourcePatternTypeAny
// ACLPatternMatch is the MATCH resource pattern.
ACLPatternMatch ACLPattern = kmsg.ACLResourcePatternTypeMatch
// ACLPatternLiteral is the LITERAL resource pattern, the default.
ACLPatternLiteral ACLPattern = kmsg.ACLResourcePatternTypeLiteral
// ACLPatternPrefixed is the PREFIXED resource pattern.
ACLPatternPrefixed ACLPattern = kmsg.ACLResourcePatternTypePrefixed
)
// ResourcePatternType sets the pattern type to use when creating or filtering
// ACL resource names, overriding the default of LITERAL.
//
// This returns the input pointer.
//
// For creating, only LITERAL and PREFIXED are supported.
func (b *ACLBuilder) ResourcePatternType(pattern ACLPattern) *ACLBuilder {
b.pattern = pattern
return b
}
// ValidateCreate returns an error if the builder is invalid for creating ACLs.
func (b *ACLBuilder) ValidateCreate() error {
for _, op := range b.ops {
switch op {
case OpAny, OpUnknown:
return fmt.Errorf("invalid operation %s for creating ACLs", op)
}
}
switch b.pattern {
case ACLPatternLiteral, ACLPatternPrefixed:
default:
return fmt.Errorf("invalid acl resource pattern %s for creating ACLs", b.pattern)
}
if len(b.allowHosts) != 0 && len(b.allow) == 0 {
return fmt.Errorf("invalid allow hosts with no allow principals")
}
if len(b.denyHosts) != 0 && len(b.deny) == 0 {
return fmt.Errorf("invalid deny hosts with no deny principals")
}
return nil
}
// ValidateDelete is an alias for ValidateFilter.
func (b *ACLBuilder) ValidateDelete() error { return b.ValidateFilter() }
// ValidateDescribe is an alias for ValidateFilter.
func (b *ACLBuilder) ValidateDescribe() error { return b.ValidateFilter() }
// ValidateFilter returns an error if the builder is invalid for deleting or
// describing ACLs (which both operate on a filter basis).
func (b *ACLBuilder) ValidateFilter() error {
if len(b.allowHosts) != 0 && len(b.allow) == 0 && !b.anyAllow {
return fmt.Errorf("invalid allow hosts with no allow principals")
}
if len(b.allow) != 0 && len(b.allowHosts) == 0 && !b.anyAllowHosts {
return fmt.Errorf("invalid allow principals with no allow hosts")
}
if len(b.denyHosts) != 0 && len(b.deny) == 0 && !b.anyDeny {
return fmt.Errorf("invalid deny hosts with no deny principals")
}
if len(b.deny) != 0 && len(b.denyHosts) == 0 && !b.anyDenyHosts {
return fmt.Errorf("invalid deny principals with no deny hosts")
}
return nil
}
// HasAnyFilter returns whether any field in this builder is opted into "any",
// meaning a wide glob. This would be if you used Topics with no topics, and so
// on. This function can be used to detect if you accidentally opted into a
// non-specific ACL.
//
// The evaluated fields are: resources, principals/hosts, a single OpAny
// operation, and an Any pattern.
func (b *ACLBuilder) HasAnyFilter() bool {
return b.anyResource ||
b.anyTopic ||
b.anyGroup ||
b.anyTxn ||
b.anyToken ||
b.anyAllow ||
b.anyAllowHosts ||
b.anyDeny ||
b.anyDenyHosts ||
b.hasOpAny() ||
b.pattern == ACLPatternAny
}
func (b *ACLBuilder) hasOpAny() bool {
for _, op := range b.ops {
if op == OpAny {
return true
}
}
return false
}
// HasResource returns true if the builder has a non-empty resource (topic,
// group, ...), or if any resource has "any" set to true.
func (b *ACLBuilder) HasResource() bool {
l := len(b.any) +
len(b.topics) +
len(b.groups) +
len(b.txnIDs) +
len(b.tokens)
return l > 0 ||
b.anyResource ||
b.anyTopic ||
b.anyGroup ||
b.anyCluster ||
b.anyTxn ||
b.anyToken
}
// HasPrincipals returns if any allow or deny principals have been set, or if
// their "any" field is true.
func (b *ACLBuilder) HasPrincipals() bool {
return len(b.allow) > 0 ||
b.anyAllow ||
len(b.deny) > 0 ||
b.anyDeny
}
// HasHosts returns if any allow or deny hosts have been set, or if their "any"
// field is true.
func (b *ACLBuilder) HasHosts() bool {
return len(b.allowHosts) > 0 ||
b.anyAllowHosts ||
len(b.denyHosts) > 0 ||
b.anyDenyHosts
}
func (b *ACLBuilder) dup() *ACLBuilder { // shallow copy
d := *b
return &d
}
// CreateACLsResult is a result for an individual ACL creation.
type CreateACLsResult struct {
Principal string
Host string
Type kmsg.ACLResourceType // Type is the type of resource this is.
Name string // Name is the name of the resource allowed / denied.
Pattern ACLPattern // Pattern is the name pattern.
Operation ACLOperation // Operation is the operation allowed / denied.
Permission kmsg.ACLPermissionType // Permission is whether this is allowed / denied.
Err error // Err is the error for this ACL creation.
ErrMessage string // ErrMessage a potential extra message describing any error.
}
// CreateACLsResults contains all results to created ACLs.
type CreateACLsResults []CreateACLsResult
// CreateACLs creates a batch of ACLs using the ACL builder, validating the
// input before issuing the CreateACLs request.
//
// If the input is invalid, or if the response fails, or if the response does
// not contain as many ACLs as we issued in our create request, this returns an
// error.
func (cl *Client) CreateACLs(ctx context.Context, b *ACLBuilder) (CreateACLsResults, error) {
if err := b.ValidateCreate(); err != nil {
return nil, err
}
if len(b.allow) != 0 && len(b.allowHosts) == 0 {
b.allowHosts = []string{"*"}
}
if len(b.deny) != 0 && len(b.denyHosts) == 0 {
b.denyHosts = []string{"*"}
}
var clusters []string
if b.anyCluster {
clusters = []string{"kafka-cluster"}
}
req := kmsg.NewPtrCreateACLsRequest()
for _, typeNames := range []struct {
t kmsg.ACLResourceType
names []string
}{
{kmsg.ACLResourceTypeTopic, b.topics},
{kmsg.ACLResourceTypeGroup, b.groups},
{kmsg.ACLResourceTypeCluster, clusters},
{kmsg.ACLResourceTypeTransactionalId, b.txnIDs},
{kmsg.ACLResourceTypeDelegationToken, b.tokens},
} {
for _, name := range typeNames.names {
for _, op := range b.ops {
for _, perm := range []struct {
principals []string
hosts []string
permType kmsg.ACLPermissionType
}{
{b.allow, b.allowHosts, kmsg.ACLPermissionTypeAllow},
{b.deny, b.denyHosts, kmsg.ACLPermissionTypeDeny},
} {
for _, principal := range perm.principals {
for _, host := range perm.hosts {
c := kmsg.NewCreateACLsRequestCreation()
c.ResourceType = typeNames.t
c.ResourceName = name
c.ResourcePatternType = b.pattern
c.Operation = op
c.Principal = principal
c.Host = host
c.PermissionType = perm.permType
req.Creations = append(req.Creations, c)
}
}
}
}
}
}
resp, err := req.RequestWith(ctx, cl.cl)
if err != nil {
return nil, err
}
if len(resp.Results) != len(req.Creations) {
return nil, fmt.Errorf("received %d results to %d creations", len(resp.Results), len(req.Creations))
}
var rs CreateACLsResults
for i, r := range resp.Results {
c := &req.Creations[i]
rs = append(rs, CreateACLsResult{
Principal: c.Principal,
Host: c.Host,
Type: c.ResourceType,
Name: c.ResourceName,
Pattern: c.ResourcePatternType,
Operation: c.Operation,
Permission: c.PermissionType,
Err: kerr.ErrorForCode(r.ErrorCode),
ErrMessage: unptrStr(r.ErrorMessage),
})
}
return rs, nil
}
// DeletedACL an ACL that was deleted.
type DeletedACL struct {
Principal string // Principal is this deleted ACL's principal.
Host string // Host is this deleted ACL's host.
Type kmsg.ACLResourceType // Type is this deleted ACL's resource type.
Name string // Name is this deleted ACL's resource name.
Pattern ACLPattern // Pattern is this deleted ACL's resource name pattern.
Operation ACLOperation // Operation is this deleted ACL's operation.
Permission kmsg.ACLPermissionType // Permission this deleted ACLs permission.
Err error // Err is non-nil if this match has an error.
ErrMessage string // ErrMessage a potential extra message describing any error.
}
// DeletedACLs contains ACLs that were deleted from a single delete filter.
type DeletedACLs []DeletedACL
// DeleteACLsResult contains the input used for a delete ACL filter, and the
// deletes that the filter matched or the error for this filter.
//
// All fields but Deleted and Err are set from the request input. The response
// sets either Deleted (potentially to nothing if the filter matched nothing)
// or Err.
type DeleteACLsResult struct {
Principal *string // Principal is the optional user that was used in this filter.
Host *string // Host is the optional host that was used in this filter.
Type kmsg.ACLResourceType // Type is the type of resource used for this filter.
Name *string // Name is the name of the resource used for this filter.
Pattern ACLPattern // Pattern is the name pattern used for this filter.
Operation ACLOperation // Operation is the operation used for this filter.
Permission kmsg.ACLPermissionType // Permission is permission used for this filter.
Deleted DeletedACLs // Deleted contains all ACLs this delete filter matched.
Err error // Err is non-nil if this filter has an error.
ErrMessage string // ErrMessage a potential extra message describing any error.
}
// DeleteACLsResults contains all results to deleted ACLs.
type DeleteACLsResults []DeleteACLsResult
// DeleteACLs deletes a batch of ACLs using the ACL builder, validating the
// input before issuing the DeleteACLs request.
//
// If the input is invalid, or if the response fails, or if the response does
// not contain as many ACL results as we issued in our delete request, this
// returns an error.
//
// Deleting ACLs works on a filter basis: a single filter can match many ACLs.
// For example, deleting with operation ANY matches any operation. For safety /
// verification purposes, you an DescribeACLs with the same builder first to
// see what would be deleted.
func (cl *Client) DeleteACLs(ctx context.Context, b *ACLBuilder) (DeleteACLsResults, error) {
dels, _, err := createDelDescACL(b)
if err != nil {
return nil, err
}
req := kmsg.NewPtrDeleteACLsRequest()
req.Filters = dels
resp, err := req.RequestWith(ctx, cl.cl)
if err != nil {
return nil, err
}
if len(resp.Results) != len(req.Filters) {
return nil, fmt.Errorf("received %d results to %d filters", len(resp.Results), len(req.Filters))
}
var rs DeleteACLsResults
for i, r := range resp.Results {
f := &req.Filters[i]
var ms DeletedACLs
for _, m := range r.MatchingACLs {
ms = append(ms, DeletedACL{
Principal: m.Principal,
Host: m.Host,
Type: m.ResourceType,
Name: m.ResourceName,
Pattern: m.ResourcePatternType,
Operation: m.Operation,
Permission: m.PermissionType,
Err: kerr.ErrorForCode(m.ErrorCode),
ErrMessage: unptrStr(m.ErrorMessage),
})
}
rs = append(rs, DeleteACLsResult{
Principal: f.Principal,
Host: f.Host,
Type: f.ResourceType,
Name: f.ResourceName,
Pattern: f.ResourcePatternType,
Operation: f.Operation,
Permission: f.PermissionType,
Deleted: ms,
Err: kerr.ErrorForCode(r.ErrorCode),
ErrMessage: unptrStr(r.ErrorMessage),
})
}
return rs, nil
}
// DescribedACL is an ACL that was described.
type DescribedACL struct {
Principal string // Principal is this described ACL's principal.
Host string // Host is this described ACL's host.
Type kmsg.ACLResourceType // Type is this described ACL's resource type.
Name string // Name is this described ACL's resource name.
Pattern ACLPattern // Pattern is this described ACL's resource name pattern.
Operation ACLOperation // Operation is this described ACL's operation.
Permission kmsg.ACLPermissionType // Permission this described ACLs permission.
}
// DescribedACLs contains ACLs that were described from a single describe
// filter.
type DescribedACLs []DescribedACL
// DescribeACLsResult contains the input used for a describe ACL filter, and
// the describes that the filter matched or the error for this filter.
//
// All fields but Described and Err are set from the request input. The
// response sets either Described (potentially to nothing if the filter matched
// nothing) or Err.
type DescribeACLsResult struct {
Principal *string // Principal is the optional user that was used in this filter.
Host *string // Host is the optional host that was used in this filter.
Type kmsg.ACLResourceType // Type is the type of resource used for this filter.
Name *string // Name is the name of the resource used for this filter.
Pattern ACLPattern // Pattern is the name pattern used for this filter.
Operation ACLOperation // Operation is the operation used for this filter.
Permission kmsg.ACLPermissionType // Permission is permission used for this filter.
Described DescribedACLs // Described contains all ACLs this describe filter matched.
Err error // Err is non-nil if this filter has an error.
ErrMessage string // ErrMessage a potential extra message describing any error.
}
// DescribeACLsResults contains all results to described ACLs.
type DescribeACLsResults []DescribeACLsResult
// DescribeACLs describes a batch of ACLs using the ACL builder, validating the
// input before issuing DescribeACLs requests.
//
// If the input is invalid, or if any response fails, this returns an error.
//
// Listing ACLs works on a filter basis: a single filter can match many ACLs.
// For example, describing with operation ANY matches any operation. Under the
// hood, this method issues one describe request per filter, because describing
// ACLs does not work on a batch basis (unlike creating & deleting). The return
// of this function can be used to see what would be deleted given the same
// builder input.
func (cl *Client) DescribeACLs(ctx context.Context, b *ACLBuilder) (DescribeACLsResults, error) {
_, descs, err := createDelDescACL(b)
if err != nil {
return nil, err
}
var (
ictx, cancel = context.WithCancel(ctx)
mu sync.Mutex
wg sync.WaitGroup
firstErr error
resps = make([]*kmsg.DescribeACLsResponse, len(descs))
)
defer cancel()
for i := range descs {
req := descs[i] // each req is unique per loop, we are not reusing req, this is safe
myIdx := i
wg.Add(1)
go func() {
defer wg.Done()
resp, err := req.RequestWith(ictx, cl.cl)
resps[myIdx] = resp
if err == nil {
return
}
cancel()
mu.Lock()
defer mu.Unlock()
if firstErr == nil { // keep the first err
firstErr = err
}
}()
}
wg.Wait()
if firstErr != nil {
return nil, firstErr
}
var rs DescribeACLsResults
for i, r := range resps {
f := descs[i]
var ds DescribedACLs
for _, resource := range r.Resources {
for _, acl := range resource.ACLs {
ds = append(ds, DescribedACL{
Principal: acl.Principal,
Host: acl.Host,
Type: resource.ResourceType,
Name: resource.ResourceName,
Pattern: resource.ResourcePatternType,
Operation: acl.Operation,
Permission: acl.PermissionType,
})
}
}
rs = append(rs, DescribeACLsResult{
Principal: f.Principal,
Host: f.Host,
Type: f.ResourceType,
Name: f.ResourceName,
Pattern: f.ResourcePatternType,
Operation: f.Operation,
Permission: f.PermissionType,
Described: ds,
Err: kerr.ErrorForCode(r.ErrorCode),
ErrMessage: unptrStr(r.ErrorMessage),
})
}
return rs, nil
}
var sliceAny = []string{"any"}
func createDelDescACL(b *ACLBuilder) ([]kmsg.DeleteACLsRequestFilter, []*kmsg.DescribeACLsRequest, error) {
if err := b.ValidateFilter(); err != nil {
return nil, nil, err
}
// As a special shortcut, if we have any allow and deny principals and
// hosts, we collapse these into one "any" group. The anyAny and
// anyAnyHosts vars are used in our looping below, and if we do this,
// we dup and set all the relevant fields to false to not expand them
// in our loops.
var anyAny, anyAnyHosts bool
if b.anyAllow && b.anyDeny && b.anyAllowHosts && b.anyDenyHosts {
anyAny = true
anyAnyHosts = true
b = b.dup()
b.allow = nil
b.allowHosts = nil
b.deny = nil
b.denyHosts = nil
b.anyAllow = false
b.anyAllowHosts = false
b.anyDeny = false
b.anyDenyHosts = false
}
var clusters []string
if b.anyCluster {
clusters = []string{"kafka-cluster"}
}
var deletions []kmsg.DeleteACLsRequestFilter
var describes []*kmsg.DescribeACLsRequest
for _, typeNames := range []struct {
t kmsg.ACLResourceType
names []string
any bool
}{
{kmsg.ACLResourceTypeAny, b.any, b.anyResource},
{kmsg.ACLResourceTypeTopic, b.topics, b.anyTopic},
{kmsg.ACLResourceTypeGroup, b.groups, b.anyGroup},
{kmsg.ACLResourceTypeCluster, clusters, b.anyCluster},
{kmsg.ACLResourceTypeTransactionalId, b.txnIDs, b.anyTxn},
{kmsg.ACLResourceTypeDelegationToken, b.tokens, b.anyToken},
} {
if typeNames.any {
typeNames.names = sliceAny
}
for _, name := range typeNames.names {
for _, op := range b.ops {
for _, perm := range []struct {
principals []string
anyPrincipal bool
hosts []string
anyHost bool
permType kmsg.ACLPermissionType
}{
{
b.allow,
b.anyAllow,
b.allowHosts,
b.anyAllowHosts,
kmsg.ACLPermissionTypeAllow,
},
{
b.deny,
b.anyDeny,
b.denyHosts,
b.anyDenyHosts,
kmsg.ACLPermissionTypeDeny,
},
{
nil,
anyAny,
nil,
anyAnyHosts,
kmsg.ACLPermissionTypeAny,
},
} {
if perm.anyPrincipal {
perm.principals = sliceAny
}
if perm.anyHost {
perm.hosts = sliceAny
}
for _, principal := range perm.principals {
for _, host := range perm.hosts {
deletion := kmsg.NewDeleteACLsRequestFilter()
describe := kmsg.NewPtrDescribeACLsRequest()
deletion.ResourceType = typeNames.t
describe.ResourceType = typeNames.t
if !typeNames.any {
deletion.ResourceName = kmsg.StringPtr(name)
describe.ResourceName = kmsg.StringPtr(name)
}
deletion.ResourcePatternType = b.pattern
describe.ResourcePatternType = b.pattern
deletion.Operation = op
describe.Operation = op
if !perm.anyPrincipal {
deletion.Principal = kmsg.StringPtr(principal)
describe.Principal = kmsg.StringPtr(principal)
}
if !perm.anyHost {
deletion.Host = kmsg.StringPtr(host)
describe.Host = kmsg.StringPtr(host)
}
deletion.PermissionType = perm.permType
describe.PermissionType = perm.permType
deletions = append(deletions, deletion)
describes = append(describes, describe)
}
}
}
}
}
}
return deletions, describes, nil
}
// DecodeACLOperations decodes an int32 bitfield into a slice of
// kmsg.ACLOperation values.
//
// This function is used to interpret the `AuthorizedOperations` field returned
// by the Kafka APIs, which specifies the operations a client is allowed to
// perform on a cluster, topic, or consumer group. It is utilized in multiple
// Kafka API responses, including Metadata, DescribeCluster, and
// DescribeGroupsResponseGroup.
//
// Caveats with Metadata API
// 1. To include authorized operations in the Metadata response, the client must explicitly
// opt in by setting `IncludeClusterAuthorizedOperations` and/or `IncludeTopicAuthorizedOperations`.
// These options were introduced in Kafka 2.3.0 as part of KIP-430.
// 2. In Kafka 2.8.0 (Metadata v11), the `AuthorizedOperations` for the cluster was removed from the
// Metadata response. Instead, clients should use the DescribeCluster API to retrieve cluster-level
// permissions.
//
// Function Behavior
// - If the bitfield equals `math.MinInt32` (-2147483648), it indicates that "AUTHORIZED_OPERATIONS_OMITTED"
// is set, and the function returns an empty slice.
// - For non-omitted values, the function iterates through all 32 bits of the bitfield. Each bit that
// is set (`1`) corresponds to an ACL operation, which is then mapped to its respective `kmsg.ACLOperation` value.
// - Undefined or unknown bits (e.g., bit 0 for `kmsg.ACLOperationUnknown`) are ignored.
//
// Supported Use Cases
// - Cluster Operations: Retrieved via the DescribeCluster API or older Metadata API versions (v8–v10).
// - Topic Operations: Retrieved via the Metadata API when `IncludeTopicAuthorizedOperations` is set.
// - Group Operations: Retrieved in the DescribeGroups API response.
func DecodeACLOperations(bitfield int32) []ACLOperation {
var operations []ACLOperation
// MinInt32 represents "AUTHORIZED_OPERATIONS_OMITTED"
if bitfield == math.MinInt32 {
return operations
}
// Helper function to determine if an operation is valid.
isValidOperation := func(op kmsg.ACLOperation) bool {
return op >= kmsg.ACLOperationRead && op <= kmsg.ACLOperationDescribeTokens
}
for i := 0; i < 32; i++ {
if bitfield&(1<<i) != 0 {
operation := kmsg.ACLOperation(i)
if isValidOperation(operation) {
operations = append(operations, operation)
}
}
}
return operations
}