Implement ACL APIs

This change implements createAcls, describeAcls, and deleteAcls on the
admin client.

Author: lp247

docker-compose.yml

@@ -95,6 +95,9 @@ services:
       # Consumer group options
       KAFKA_CFG_GROUP_COORDINATOR_REBALANCE_PROTOCOLS: "classic,consumer"
       KAFKA_CFG_GROUP_INITIAL_REBALANCE_DELAY_MS: "0"
+      # ACL options
+      KAFKA_AUTHORIZER_CLASS_NAME: "org.apache.kafka.metadata.authorizer.StandardAuthorizer"
+      KAFKA_SUPER_USERS: "User:ANONYMOUS"
 
   broker-cluster-2:
     image: *image

docs/admin.md

@@ -98,10 +98,10 @@ The return value is an object specifying quotas for the requested user/client co
 
 Options:
 
-| Property   | Type                                          | Description                                                                                              |
-| ---------- | --------------------------------------------- | -------------------------------------------------------------------------------------------------------- |
-| components | `DescribeClientQuotasRequestComponent[]`      | Array of components specifying the entity types and match criteria for which to describe client quotas.  |
-| strict     | `boolean`                                     | Whether to use strict matching for components. Defaults to `false`.                                      |
+| Property   | Type                                     | Description                                                                                             |
+| ---------- | ---------------------------------------- | ------------------------------------------------------------------------------------------------------- |
+| components | `DescribeClientQuotasRequestComponent[]` | Array of components specifying the entity types and match criteria for which to describe client quotas. |
+| strict     | `boolean`                                | Whether to use strict matching for components. Defaults to `false`.                                     |
 
 ### `alterClientQuotas(options[, callback])`
 
@@ -111,10 +111,48 @@ The return value is a list of entities for which quotas have been changed.
 
 Options:
 
-| Property     | Type                              | Description                                                                               |
-| ------------ | --------------------------------- | ----------------------------------------------------------------------------------------- |
-| entries      | `AlterClientQuotasRequestEntry[]` | Array of entries specifying the entities and quotas to change.                            |
-| validateOnly | `boolean`                         | Whether to only validate the request without applying changes. Defaults to `false`.       |
+| Property     | Type                              | Description                                                                         |
+| ------------ | --------------------------------- | ----------------------------------------------------------------------------------- |
+| entries      | `AlterClientQuotasRequestEntry[]` | Array of entries specifying the entities and quotas to change.                      |
+| validateOnly | `boolean`                         | Whether to only validate the request without applying changes. Defaults to `false`. |
+
+### `createAcls(options[, callback])`
+
+Creates Access Control List (ACL) entries to define permissions for Kafka resources.
+
+The return value is `void`.
+
+Options:
+
+| Property   | Type    | Description                     |
+| ---------- | ------- | ------------------------------- |
+| creations  | `Acl[]` | Array of ACL entries to create. |
+
+### `describeAcls(options[, callback])`
+
+Describes existing Access Control List (ACL) entries that match the specified filter criteria.
+
+The return value is an array of resources with their associated ACL entries.
+
+Options:
+
+| Property | Type        | Description                               |
+| -------- | ----------- | ----------------------------------------- |
+| filter   | `AclFilter` | Filter criteria for matching ACL entries. |
+
+The filter contains the same properties as ACL entries, but `resourceName`, `principal`, and `host` can be `null` to match any value.
+
+### `deleteAcls(options[, callback])`
+
+Deletes Access Control List (ACL) entries that match the specified filter criteria.
+
+The return value is an array of deleted ACL entries.
+
+Options:
+
+| Property | Type          | Description                                         |
+| -------- | ------------- | --------------------------------------------------- |
+| filters  | `AclFilter[]` | Array of filter criteria for ACL entries to delete. |
 
 ### `describeLogDirs(options[, callback])`
 

docs/diagnostic.md

@@ -73,6 +73,7 @@ Each tracing channel publishes events with the following common properties:
 | `plt:kafka:admin:groups`            | `Admin`          | Traces a `Admin.listGroups`, `Admin.describeGroups` or `Admin.deleteGroups` request.              |
 | `plt:kafka:admin:clientQuotas`      | `Admin`          | Traces a `Admin.describeClientQuotas` or `Admin.alterClientQuotas` request.                       |
 | `plt:kafka:admin:logDirs`           | `Admin`          | Traces a `Admin.describeLogDirs` request.                                                         |
+| `plt:kafka:admin:acls`              | `Admin`          | Traces a `Admin.createAcls`, `Admin.describeAcls` or `Admin.deleteAcls` request.                  |
 | `plt:kafka:producer:initIdempotent` | `Producer`       | Traces a `Producer.initIdempotentProducer` request.                                               |
 | `plt:kafka:producer:sends`          | `Producer`       | Traces a `Producer.send` request.                                                                 |
 | `plt:kafka:consumer:group`          | `Consumer`       | Traces a `Consumer.findGroupCoordinator`, `Consumer.joinGroup` or `Consumer.leaveGroup` requests. |

playground/apis/admin/acl.ts

@@ -1,12 +1,7 @@
 import { api as createAclsV3 } from '../../../src/apis/admin/create-acls-v3.ts'
 import { api as deleteAclsV3 } from '../../../src/apis/admin/delete-acls-v3.ts'
 import { api as describeAclsV3 } from '../../../src/apis/admin/describe-acls-v3.ts'
-import {
-  AclOperations,
-  AclPermissionTypes,
-  ResourcePatternTypes,
-  ResourceTypes
-} from '../../../src/apis/enumerations.ts'
+import { AclOperations, AclPermissionTypes, PatternTypes, ResourceTypes } from '../../../src/apis/enumerations.ts'
 import { Connection } from '../../../src/network/connection.ts'
 import { performAPICallWithRetry } from '../../utils.ts'
 
@@ -18,7 +13,7 @@ await performAPICallWithRetry('CreateAcls', () =>
     {
       resourceType: ResourceTypes.TOPIC,
       resourceName: 'temp',
-      resourcePatternType: ResourcePatternTypes.LITERAL,
+      patternType: PatternTypes.LITERAL,
       principal: 'abc:cde',
       host: '*',
       operation: AclOperations.READ,
@@ -27,37 +22,35 @@ await performAPICallWithRetry('CreateAcls', () =>
   ]))
 
 await performAPICallWithRetry('DescribeAcls', () =>
-  describeAclsV3.async(
-    connection,
-    ResourceTypes.TOPIC,
-    'temp',
-    ResourcePatternTypes.LITERAL,
-    null,
-    null,
-    AclOperations.READ,
-    AclPermissionTypes.DENY
-  ))
+  describeAclsV3.async(connection, {
+    resourceType: ResourceTypes.TOPIC,
+    resourceName: 'temp',
+    patternType: PatternTypes.LITERAL,
+    principal: null,
+    host: null,
+    operation: AclOperations.READ,
+    permissionType: AclPermissionTypes.DENY
+  }))
 
 await performAPICallWithRetry('DescribeAcls', () =>
-  describeAclsV3.async(
-    connection,
-    ResourceTypes.TOPIC,
-    'temp',
-    ResourcePatternTypes.LITERAL,
-    null,
-    null,
-    AclOperations.READ,
-    AclPermissionTypes.ALLOW
-  ))
+  describeAclsV3.async(connection, {
+    resourceType: ResourceTypes.TOPIC,
+    resourceName: 'temp',
+    patternType: PatternTypes.LITERAL,
+    principal: null,
+    host: null,
+    operation: AclOperations.READ,
+    permissionType: AclPermissionTypes.ALLOW
+  }))
 
 await performAPICallWithRetry('DeleteAcls', () =>
   deleteAclsV3.async(connection, [
     {
-      resourceTypeFilter: ResourceTypes.TOPIC,
-      resourceNameFilter: 'temp',
-      patternTypeFilter: ResourcePatternTypes.LITERAL,
-      principalFilter: null,
-      hostFilter: null,
+      resourceType: ResourceTypes.TOPIC,
+      resourceName: 'temp',
+      patternType: PatternTypes.LITERAL,
+      principal: null,
+      host: null,
       operation: AclOperations.READ,
       permissionType: AclPermissionTypes.DENY
     }

src/apis/admin/create-acls-v3.ts

@@ -3,16 +3,7 @@ import { type NullableString } from '../../protocol/definitions.ts'
 import { type Reader } from '../../protocol/reader.ts'
 import { Writer } from '../../protocol/writer.ts'
 import { createAPI, type ResponseErrorWithLocation } from '../definitions.ts'
-
-export interface CreateAclsRequestCreation {
-  resourceType: number
-  resourceName: string
-  resourcePatternType: number
-  principal: string
-  host: string
-  operation: number
-  permissionType: number
-}
+import { type Acl } from '../types.ts'
 
 export type CreateAclsRequest = Parameters<typeof createRequest>
 
@@ -37,12 +28,12 @@ CreateAcls Request (Version: 3) => [creations] TAG_BUFFER
     operation => INT8
     permission_type => INT8
 */
-export function createRequest (creations: CreateAclsRequestCreation[]): Writer {
+export function createRequest (creations: Acl[]): Writer {
   return Writer.create()
     .appendArray(creations, (w, c) => {
       w.appendInt8(c.resourceType)
         .appendString(c.resourceName)
-        .appendInt8(c.resourcePatternType)
+        .appendInt8(c.patternType)
         .appendString(c.principal)
         .appendString(c.host)
         .appendInt8(c.operation)

src/apis/admin/delete-acls-v3.ts

@@ -3,28 +3,14 @@ import { type NullableString } from '../../protocol/definitions.ts'
 import { type Reader } from '../../protocol/reader.ts'
 import { Writer } from '../../protocol/writer.ts'
 import { createAPI, type ResponseErrorWithLocation } from '../definitions.ts'
+import { type AclOperation, type AclPermissionType, type PatternType, type ResourceType } from '../enumerations.ts'
+import { type Acl, type AclFilter } from '../types.ts'
 
-export interface DeleteAclsRequestFilter {
-  resourceTypeFilter: number
-  resourceNameFilter?: NullableString
-  patternTypeFilter: number
-  principalFilter?: NullableString
-  hostFilter?: NullableString
-  operation: number
-  permissionType: number
-}
 export type DeleteAclsRequest = Parameters<typeof createRequest>
 
-export interface DeleteAclsResponseMatchingAcl {
+export interface DeleteAclsResponseMatchingAcl extends Acl {
   errorCode: number
   errorMessage: NullableString
-  resourceType: number
-  resourceName: string
-  patternType: number
-  principal: string
-  host: string
-  operation: number
-  permissionType: number
 }
 
 export interface DeleteAclsResponseFilterResults {
@@ -48,14 +34,14 @@ export interface DeleteAclsResponse {
       operation => INT8
       permission_type => INT8
 */
-export function createRequest (filters: DeleteAclsRequestFilter[]): Writer {
+export function createRequest (filters: AclFilter[]): Writer {
   return Writer.create()
     .appendArray(filters, (w, f) => {
-      w.appendInt8(f.resourceTypeFilter)
-        .appendString(f.resourceNameFilter)
-        .appendInt8(f.patternTypeFilter)
-        .appendString(f.principalFilter)
-        .appendString(f.hostFilter)
+      w.appendInt8(f.resourceType)
+        .appendString(f.resourceName)
+        .appendInt8(f.patternType)
+        .appendString(f.principal)
+        .appendString(f.host)
         .appendInt8(f.operation)
         .appendInt8(f.permissionType)
     })
@@ -109,13 +95,13 @@ export function parseResponse (
           return {
             errorCode,
             errorMessage: r.readNullableString(),
-            resourceType: r.readInt8(),
+            resourceType: r.readInt8() as ResourceType,
             resourceName: r.readString(),
-            patternType: r.readInt8(),
+            patternType: r.readInt8() as PatternType,
             principal: r.readString(),
             host: r.readString(),
-            operation: r.readInt8(),
-            permissionType: r.readInt8()
+            operation: r.readInt8() as AclOperation,
+            permissionType: r.readInt8() as AclPermissionType
           }
         })
       }

src/apis/admin/describe-acls-v3.ts

@@ -3,21 +3,13 @@ import { type NullableString } from '../../protocol/definitions.ts'
 import { type Reader } from '../../protocol/reader.ts'
 import { Writer } from '../../protocol/writer.ts'
 import { createAPI } from '../definitions.ts'
+import { type AclOperation, type AclPermissionType, type PatternType, type ResourceType } from '../enumerations.ts'
+import { type AclPermission, type AclTarget, type AclFilter } from '../types.ts'
 
 export type DescribeAclsRequest = Parameters<typeof createRequest>
 
-export interface DescribeAclsResponseAcl {
-  principal: string
-  host: string
-  operation: number
-  permissionType: number
-}
-
-export interface DescribeAclsResponseResource {
-  resourceType: number
-  resourceName: string
-  patternType: number
-  acls: DescribeAclsResponseAcl[]
+export interface DescribeAclsResponseResource extends AclTarget {
+  acls: AclPermission[]
 }
 export interface DescribeAclsResponse {
   throttleTimeMs: number
@@ -36,23 +28,15 @@ export interface DescribeAclsResponse {
     operation => INT8
     permission_type => INT8
 */
-export function createRequest (
-  resourceTypeFilter: number,
-  resourceNameFilter: NullableString,
-  patternTypeFilter: number,
-  principalFilter: NullableString,
-  hostFilter: NullableString,
-  operation: number,
-  permissionType: number
-): Writer {
+export function createRequest (filter: AclFilter): Writer {
   return Writer.create()
-    .appendInt8(resourceTypeFilter)
-    .appendString(resourceNameFilter)
-    .appendInt8(patternTypeFilter)
-    .appendString(principalFilter)
-    .appendString(hostFilter)
-    .appendInt8(operation)
-    .appendInt8(permissionType)
+    .appendInt8(filter.resourceType)
+    .appendString(filter.resourceName)
+    .appendInt8(filter.patternType)
+    .appendString(filter.principal)
+    .appendString(filter.host)
+    .appendInt8(filter.operation)
+    .appendInt8(filter.permissionType)
     .appendTaggedFields()
 }
 
@@ -83,15 +67,15 @@ export function parseResponse (
     errorMessage: reader.readNullableString(),
     resources: reader.readArray(r => {
       return {
-        resourceType: r.readInt8(),
+        resourceType: r.readInt8() as ResourceType,
         resourceName: r.readString(),
-        patternType: r.readInt8(),
+        patternType: r.readInt8() as PatternType,
         acls: r.readArray(r => {
           return {
             principal: r.readString(),
             host: r.readString(),
-            operation: r.readInt8(),
-            permissionType: r.readInt8()
+            operation: r.readInt8() as AclOperation,
+            permissionType: r.readInt8() as AclPermissionType
           }
         })
       }

src/apis/enumerations.ts

@@ -48,12 +48,13 @@ export const ResourceTypes = {
   GROUP: 3,
   CLUSTER: 4,
   TRANSACTIONAL_ID: 5,
-  DELEGATION_TOKEN: 6
+  DELEGATION_TOKEN: 6,
+  USER: 7
 } as const
-export type ResourceType = keyof typeof ResourceTypes
+export type ResourceType = (typeof ResourceTypes)[keyof typeof ResourceTypes]
 
-export const ResourcePatternTypes = { UNKNOWN: 0, ANY: 1, MATCH: 2, LITERAL: 3, PREFIXED: 4 } as const
-export type ResourcePatternType = keyof typeof ResourcePatternTypes
+export const PatternTypes = { UNKNOWN: 0, ANY: 1, MATCH: 2, LITERAL: 3, PREFIXED: 4 } as const
+export type PatternType = (typeof PatternTypes)[keyof typeof PatternTypes]
 
 export const AclOperations = {
   UNKNOWN: 0,
@@ -68,12 +69,15 @@ export const AclOperations = {
   CLUSTER_ACTION: 9,
   DESCRIBE_CONFIGS: 10,
   ALTER_CONFIGS: 11,
-  IDEMPOTENT_WRITE: 12
+  IDEMPOTENT_WRITE: 12,
+  CREATE_TOKENS: 13,
+  DESCRIBE_TOKENS: 14,
+  TWO_PHASE_COMMIT: 15
 } as const
-export type AclOperation = keyof typeof AclOperations
+export type AclOperation = (typeof AclOperations)[keyof typeof AclOperations]
 
 export const AclPermissionTypes = { UNKNOWN: 0, ANY: 1, DENY: 2, ALLOW: 3 } as const
-export type AclPermissionType = keyof typeof AclPermissionTypes
+export type AclPermissionType = (typeof AclPermissionTypes)[keyof typeof AclPermissionTypes]
 
 // ./admin/*-configs.ts
 export const ConfigSources = {

src/apis/index.ts

@@ -2,6 +2,7 @@
 export * from './callbacks.ts'
 export * from './definitions.ts'
 export * from './enumerations.ts'
+export * from './types.ts'
 
 // Low-level APIs
 export * from './admin/index.ts'