ACL を使用した認可¶

Apache Kafka® には、そのままで使える、プラグ可能なオーソライザーが付属します。これは Apache ZooKeeper™ を使用してすべての ACL を格納します。ACL を設定することは重要です。オーソライザーが構成されているとき、ACL を設定しない場合、リソースへのアクセスがスーパーユーザーに限定されるためです。デフォルトの動作では、リソースに ACL が関連付けられていない場合、スーパーユーザー以外のユーザーはリソースにアクセスできません。

参考

この構成の動作例については、Confluent Platform デモを参照してください。構成リファレンスについては、デモの docker-compose.yml ファイルを参照してください。

ACL のコンセプト¶

アクセス制御リスト（ACL）により、所属企業の Apache Kafka® クラスターデータに対して、重要な認可の制御を実現します。ACL を作成して使用を始める前に、このセクションで説明する ACL のコンセプトをよく確認してください。ACL のコンセプトを理解することは、ACL を作成および使用して、コンポーネントやクラスターデータへのアクセスを正しく管理するための重要なポイントです。

オーソライザー¶

オーソライザーは操作を認可するために Apache Kafka® で使用されるサーバープラグインです。具体的には、オーソライザーで、プリンシパルとアクセス対象のリソースを基に、操作を認可するかどうかを制御します。デフォルトの Kafka オーソライザーの実装は、AclAuthorizer（kafka.security.authorizer.AclAuthorizer）で、Apache Kafka® 2.4 および Confluent Platform 5.4.0 に導入されました。その前のバージョンでは、オーソライザーは SimpleAclAuthorizer（kafka.security.auth.SimpleAclAuthorizer）という名前でした。

AclAuthorizer を有効にして使用するには、次のとおり server.properties にブローカー構成の完全なクラス名を設定します。

authorizer.class.name=kafka.security.authorizer.AclAuthorizer

注釈

このトピックで説明するのは AclAuthorizer だけですが、Confluent には Confluent LDAP Authorizer によるグループとユーザープリンシパルベースの認可、 Confluent Server Authorizer による、独自の LDAP グループとロールベースアクセス制御（RBAC）、さらに一元的な ACL も用意されています。デフォルトでは、Confluent Server Authorizer は ZooKeeper ベースの ACL をサポートします。

AclAuthorizer は Kafka の ACL 情報を ZooKeeper に格納します。ただし、ZooKeeper ノードへのアクセスを制御しません。ZooKeeper には独自の ACL セキュリティ機能があり、ZooKeeper ノードのアクセス制御を実行します。ZooKeeper の ACL では、どのプリンシパル（たとえば、ブローカープリンシパル）が Kafka クラスターメタデータ（同期レプリカ、トピック構成、Kafka の ACL など）を含む ZooKeeper ノード、およびブローカー間の調整（コントローラーの選出、ブローカーの参加、トピックの削除など）で使用されるノードをアップデートできるのかを制御します。

Kafka ACL では、どのプリンシパルが Kafka リソースで操作を実行できるのかを制御します。Kafka ブローカーでは、ブローカーの構成で ZooKeeper のセキュリティを有効（zookeeper.set.acl=true）にすることで、ZooKeeper の ACL を使用できます。

プリンシパル¶

プリンシパルは、オーソライザーにより認証できるエンティティです。Kafka ブローカーのクライアントは、多様なセキュリティプロトコルを使用して、自身を個別のプリンシパルとして識別します。プリンシパルを識別する方法は、Kafka ブローカーに接続するために使用するセキュリティプロトコル（mTLS、 SASL/GSSAPI または SASL/PLAIN など）で決まります。認証は、Kafka ブローカー内のプリンシパルを識別するために実行されているセキュリティプロトコル（SASL、TLS/SSL など）で決まります。

次の例では、使用されているセキュリティプロトコルに基づいたプリンシパル名のフォーマットを示します。

クライアントが Kafka ブローカーに SSL セキュリティプロトコルで接続するとき、プリンシパル名は SSL 証明書のサブジェクト名の形式、つまり CN=quickstart.confluent.io,OU=TEST,O=Sales,L=PaloAlto,ST=Ca,C=US のようになります。サブジェクトの各部の間にあるコンマの後にスペースがないことに注意してください。
クライアントが Kafka ブローカーに SASL セキュリティプロトコルと GSSAPI（Kerberos）メカニズムで接続するとき、プリンシパルは kafka-client@hostname.com という Kerberos のプリンシパルフォーマットです。詳細については、「Kerberos Principal Names」を参照してください。
クライアントが Kafka ブローカーに SASL セキュリティプロトコルと PLAIN または SCRAM メカニズムで接続するとき、プリンシパルは単なるテキスト文字列の alice、admin、billing_etl_job_03 などになります。

AclAuthorizer でサポートするのは、個別のユーザーだけで、常にプリンシパルがユーザー名と解釈されます。ただし、他のオーソライザー（LDAP Authorizer など）はグループをサポートします。したがって、プリンシパルを指定するとき、プレフィックス User: または Group: （大文字と小文字を区別）を使用したタイプを含める必要があります。たとえば、User:admin、Group:developers または User:CN=quickstart.confluent.io,OU=TEST,O=Sales,L=PaloAlto,ST=Ca,C=US です。

次の ACL では、プレーンテキストのプリンシパル（User:alice、User:fred ）は、指定ホスト（host-1、host-2）の一方から特定リソース（topic）に対して特定の操作（読み取り、書き込み）を許可される Kafka ユーザーとして識別されます。

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
 --add --allow-principal User:alice --allow-principal User:fred --allow-host host-1 \
 --allow-host host-2 --operation read --operation write --topic finance-topic

ベストプラクティスは、アプリケーションごとに 1 つのプリンシパルを作成し、各プリンシパルに必要最小限の ACL を割り当てることです。たとえば、Alice が請求ワークフローを自動化するために、異なるトピックにアクセスする 3 種類のプログラムをコーディングする場合、3 つのプリンシパル billing_etl_job_01、billing_etl_job_02、および billing_etl_job_03 を作成できます。Alice は必要最小限のトピックで各プリンシパルのアクセス許可を付与し、個別のプリンシパルで各プログラムを実行します。

他のアプローチとして、請求プログラムで必要なすべてのトピックにアクセスできる、単一の billing_etl_jobs プリンシパルを作成し、そのプリンシパルで 3 つすべてのプログラムを単に実行するという方式があります。

Alice 自身のプリンシパルでこれらのプログラムを実行することは推奨しません。このジョブで実際に必要とされるよりも広範なアクセス許可を保持していると推定されるためです。また、アプリケーションごとに 1 つのプリンシパルで実行すると、デバッグや監査で大きく役に立ちます。操作をどのアプリケーションが実行しているのかがはっきりしているためです。

ワイルドカードプリンシパル¶

プリンシパルでワイルドカードを使用して（User:*）、すべてのプリンシパルに対する ACL を作成できます。ユーザープリンシパルにワイルドカードを使用する ACL はすべてのユーザーに適用されます。たとえば、次のコマンドではトピック testTopic に対してすべてのユーザーにアクセス許可を付与します。

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add \
--allow-principal User:* --operation All --topic testTopic

グループプリンシパルをサポートするオーソライザー（Confluent Server Authorizer など）を使用する場合、プリンシパル Group:* を使用して、すべてのグループプリンシパルに対して ACL を作成することもできます。プリンシパルにワイルドカードを使用する ACL は、少なくとも 1 つのグループに所属するすべてのユーザーに適用されます。

スーパーユーザーに対して、ワイルドカードを使用する ACL を作成できません。たとえば、次に示すのはユーザープリンシパル User:* をスーパーユーザーにする構成ですが、すべてのユーザーがスーパーユーザーになるわけではありません。ワイルドカードの一致が実行されないためです。

super.users=User:*

注釈

Confluent Server Authorizer を使用している場合、ロールのバインディングではワイルドカードの一致をサポートしないことに注意してください。したがって、ロールを User:* に割り当てても、このロールがすべてのユーザーに付与されるわけではありません。RBAC プリンシパルの詳細については、「ロールベースアクセス制御を使用した認可」を参照してください。

SASL/Kerberos のプリンシパル¶

Kerberos を使用する場合、Kafka のプリンシパルは Kerberos のプリンシパル（例 kafka/kafka1.hostname.com@EXAMPLE.COM）を基に決まります。デフォルトでは、Kafka は Kerberos のプリンシパルのプライマリ名（名前のスラッシュ（ /）より前の部分）のみを使用します。したがって、ブローカーの Kerberos のプリンシパルが kafka/broker1.example.com@EXAMPLE である場合、Kafka のオーソライザーで使用されるプリンシパルは kafka です。ホスト名はすべてのブローカーで異なります。この解析処理は、sasl.kerberos.principal.to.local.rules のデフォルト値を使用して、自動的に実行されています。

Kerberos のプリンシパル名と構成の詳細については、「Kerberos Principals」を参照してください。

注釈

所属組織で Kerberos または Active Directory サーバーを使用している場合、クラスターにある各 Kafka ブローカー用のプリンシパル、および Kerberos 認証で Kafka に（クライアントやツールで）アクセスするすべてのオペレーティングシステムユーザーを、Kerberos の管理者に問い合わせます。サーバープリンシパルは NT_HOSTBASED_SERVICE タイプです。

各ブローカーは、レプリケーションのため、およびコントローラーとして機能するとき、他のすべてのブローカーと通信できる必要があります。ブローカープリンシパルをスーパーユーザーとして追加する必要があります。追加しない場合、Kafka は機能しません。

SASL/Kerberos のユーザー名をカスタマイズする構成オプション¶

デフォルトでは、Kafka のプリンシパルは、Kerberos のプリンシパルのプライマリ部分です。server.properties で sasl.kerberos.principal.to.local.rules にカスタマイズしたルールを指定することにより、この動作を変更できます。sasl.kerberos.principal.to.local.rules のフォーマットは、Kerberos 構成ファイル（krb5.conf）の auth_to_local と同様に、各ルールの内容を記述したリスト形式をとります。これらのルールは小文字（/L）または大文字（/U ）オプションもサポートしており、変換結果を強制的にすべて小文字または大文字にします。これは、ルールの末尾に /L または /U を追加することによって行われます。各ルールは RULE: で始まり、式が含まれます。以下の例は、フォーマットと構文を示しています。

RULE:[n:string](regexp)s/pattern/replacement/
RULE:[n:string](regexp)s/pattern/replacement/g
RULE:[n:string](regexp)s/pattern/replacement//L
RULE:[n:string](regexp)s/pattern/replacement/g/L
RULE:[n:string](regexp)s/pattern/replacement//U
RULE:[n:string](regexp)s/pattern/replacement/g/U

このルールでは、デフォルトルールを維持しながら user@MYDOMAIN.COM が user に変換されます。

sasl.kerberos.principal.to.local.rules=RULE:[1:$1@$0](.*@MYDOMAIN.COM)s/@.*//,DEFAULT

TLS/SSL のプリンシパルユーザー名¶

前に示したように、プリンシパルの識別は、ユーザーがブローカーで認証される方法に基づいており、これは使用するセキュリティプロトコルで決まります。TLS/SSL のプリンシパルを使用するには、ユーザー名を正確に表す方法を理解する必要があります。

デフォルトでは、TLS/SSL 証明書で識別されるプリンシパルの名前は、証明書の DN（X.500 Distinguished Name: 識別名、別名サブジェクト）で、CN=writeuser,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown という形式を使用します。ssl.principal.mapping.rules を使用して、DN を管理しやすいプリンシパル名に変換できます。詳細については、「TLS/SSL のユーザー名をカスタマイズする構成オプション」を参照してください。

SSL は有効であるが、クライアント認証が構成されていない場合、クライアントは SSL ポートを使用して匿名で接続し、サーバーはユーザー名 ANONYMOUS として処理します。このような構成では、暗号化とサーバー認証が実行されますが、クライアントは匿名で接続します。サーバーに ANONYMOUS ユーザーが表示される他の場合は、PLAINTEXT セキュリティプロトコルを使用している場合です。読み取りや書き込みアクセス許可を ANONYMOUS ユーザーに付与することにより、どのユーザーも認証なしでブローカーにアクセスすることが許可されます。つまり、すべてのユーザーにこのアクセス許可を付与する意図がない限り、ANONYMOUS ユーザーにアクセス許可を付与しないでください。

TLS/SSL のユーザー名をカスタマイズする構成オプション¶

デフォルトでは、SSL のユーザー名は CN=writeuser,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown の形式です。この構成では、X.500 識別名（DN）を短い名前にマッピングするためのルールのリストを利用できます。ルールは順番に評価され、DN と一致する最初のルールが、短い名前へのマッピングに使用されます。リスト内のそれ以降のルールはすべて無視されます。

ssl.principal.mapping.rules のフォーマットは、"RULE:" で始まり、次のフォーマットの式が続く、ルールのリストです。デフォルトのルールは、X.500 証明書 DN の文字列表現を返します。DN がパターンと一致する場合、置換コマンドがその名前に対して実行されます。このコマンドは小文字と大文字のオプションもサポートしており、変換結果を強制的にすべて小文字または大文字に置き換えます。この処理は、ルールの末尾に "/L" または "/U" を追加することによって実行されます。

RULE:pattern/replacement/
RULE:pattern/replacement/[LU]

たとえば、ssl.principal.mapping.rules の値は次のとおりです。

RULE:^CN=(.*?),OU=ServiceUsers.*$/$1/,
RULE:^CN=(.*?),OU=(.*?),O=(.*?),L=(.*?),ST=(.*?),C=(.*?)$/$1@$2/L,
RULE:^.*[Cc][Nn]=([a-zA-Z0-9.]*).*$/$1/L,
DEFAULT

これらのルールは DN CN=serviceuser,OU=ServiceUsers,O=Unknown,L=Unknown,ST=Unknown,C=Unknown を serviceuser に、 CN=adminUser,OU=Admin,O=Unknown,L=Unknown,ST=Unknown,C=Unknown を adminuser@admin に変換します。

操作¶

操作はリソースで実行するアクションです。ACL ではユーザーやグループがアクセスできるリソースを識別するとともに、それらのユーザーやグループが認可されている操作も識別します。各リソースに対して、操作はそのリソースで利用できる 1 つまたは複数の Kafka API またはリクエストタイプにマップされます。たとえば、Topic リソースの READ 操作は Fetch、OffsetCommit および TxnOffsetCommit にマップされます。または Topic リソースの WRITE 操作は Produce および AddPartitionsToTxn にマップされます。

次の表では、Confluent Platform の各リソースタイプに利用できる有効な操作を示し、操作、リソース、API の間の関係について説明します。

Cluster リソースタイプに利用できる操作

操作	リソース	許可されている API
Alter	クラスター	AlterReplicaLogDirs、CreateAcls、DeleteAcls
AlterConfigs	クラスター	AlterConfigs
ClusterAction	クラスター	Fetch（レプリケーション用のみ）、LeaderAndIsr、OffsetForLeaderEpoch、StopReplica、UpdateMetadata、ControlledShutdown、WriteTxnMarkers
作成	クラスター	CreateTopics、Metadata（`auto.create.topics.enable` の場合）
Describe	クラスター	DescribeAcls、DescribeLogDirs、ListGroups
DescribeConfigs	クラスター	DescribeConfigs
IdempotentWrite	クラスター	InitProducerId、Produce

Topic リソースタイプに利用できる操作

操作	リソース	許可されている API
Alter	トピック	CreatePartitions
AlterConfigs	トピック	AlterConfigs
作成	トピック	Metadata（`auto.create.topics.enable` の場合）、CreateTopics
削除	トピック	DeleteRecords、DeleteTopics
Describe	トピック	ListOffsets、Metadata、OffsetFetch、OffsetForLeaderEpoch
DescribeConfigs	トピック	DescribeConfigs
読み取り	トピック	Fetch、OffsetCommit、TxnOffsetCommit
書き込み	トピック	Produce、AddPartitionsToTxn

Group リソースタイプに利用できる操作

操作	リソース	許可されている API
削除	グループ	DeleteGroups
Describe	グループ	DescribeGroup、FindCoordinator、ListGroups
読み取り	グループ	AddOffsetsToTxn、Heartbeat、JoinGroup、LeaveGroup、OffsetCommit、OffsetFetch、SyncGroup、TxnOffsetCommit

DelegationToken リソースタイプに利用できる操作

操作	リソース	許可されている API
Describe	DelegationToken	DescribeTokens

TransactionalId リソースタイプに利用できる操作

操作	リソース	許可されている API
Describe	TransactionalId	FindCoordinator
書き込み	TransactionalId	Produce、AddPartitionsToTxn、AddOffsetsToTxn、EndTxn、InitProducerId、TxnOffsetCommit

前に示した表の操作は、クライアント（プロデューサー、コンシューマー、管理）およびクラスターのブローカー間の操作に対するものです。セキュアなクラスターでは、クライアントリクエストおよびブローカー間の操作には認可が必要です。ブローカー間の操作は 2 つのクラス、cluster と topic に分かれます。クラスターの操作は、ブローカーおよびパーティションメタデータのアップデート、リーダーおよびパーティションの同期レプリカセットの変更、制御されたシャットダウンのトリガーなど、クラスターの管理に必要な操作を指します。

トピックのパーティションのレプリケーションが内部で動作する方法のため、ブローカーのプリンシパルは、ブローカーがリーダーからフォロワーに正しくトピックをレプリケーションできるように、スーパーユーザーであることが必要です。

プロデューサーとコンシューマーは、トピックに対する操作を実行するために認可される必要がありますが、ブローカーと異なるプリンシパルで構成する必要があります。プロデューサーで認可が必要なメインの操作は WRITE と READ です。管理ユーザーはコマンドラインツールを実行し、認可を要求できます。管理ユーザーで認可が必要になることがある操作は、DELETE、CREATE および ALTER です。プロデューサーとコンシューマーにワイルドカード（*）を使用できます。このため設定が必要なのは 1 回だけです。

暗黙的に認められる操作¶

一部の操作では、操作の暗黙的アクセス許可がユーザーに追加されます。

READ、WRITE または DELETE のアクセス許可が付与されると、ユーザーは DESCRIBE 操作の権限を暗黙的に獲得します。

ALTER_CONFIGS のアクセス許可が付与されると、ユーザーは暗黙的に DESCRIBE_CONFIGS 操作権限を獲得します。

ACL の優先順位¶

許可の ACL と拒否の ACL がある状況では、拒否の ACL が許可の ACL に優先されます。

リソース¶

ユーザーは特定の Kafka および Confluent Platform リソースにアクセスし操作を実行できます。リソースには、クラスター、グループ、Apache Kafka® トピック、トランザクション ID または委任トークンがあります。ACL では、指定したリソースにアクセスできるユーザーとそのリソースに対して実行を許可された操作を指定します。Kafka には次のリソースがあります。

クラスター

Kafka クラスターです。クラスター全体に影響を及ぼす操作（制御されたシャットダウン、新しいトピックの作成など）を実行することが必要なユーザーでは、クラスターリソースに対する権限の割り当てが必要です。

委任トークン

委任トークンは Apache Kafka® のブローカーとクライアント間で共有するシークレットです。委任トークンに基づく認証は、既存の SASL/SSL メソッドを補完するために使用できる軽量な認証メカニズムです。詳細については、「委任トークンを使用した認証」を参照してください。

グループ

ブローカー内のグループです。グループで機能する（グループの参加など）すべてのプロトコルコールは、サブジェクトで、グループの対応する権限が必要です。グループ（group.id）では、コンシューマーグループ、ストリームグループ（application.id）、Connect ワーカーグループ、またはコンシューマーグループプロトコルを使用する他のグループ（Schema Registry クラスターなど）を表すことができます。

トピック

すべての Kafka メッセージはトピック（およびパーティション）に整理されます。トピックにアクセスするには、ACL で定義された対応する操作（READ や WRITE など）が必要です。

トランザクション ID

トランザクション ID（transactional.id ）は、アプリケーションが再起動されても、単一のプロデューサーインスタンスを識別し、単一のライターを保証する方法になります。これは "厳密に 1 回" のセマンティクス（EOS）で必要です。各 transactional.id でアクティブにできるのは、常に 1 つのプロデューサーだけです。プロデューサーが起動されると、まず自身の transactional.id であるプロデューサーで中断しているトランザクションがあるかどうかがチェックされます。トランザクションが存在する場合、終了（中断またはコミット）するまで待機する必要があります。これにより、プロデューサーが常に一貫したステートから起動することが保証されます。

トランザクションを使用するとき、プロデューサーでトランザクション ID を操作できること、すべてのアクセス許可セットを保持していることが必要です。たとえば、次の ACL では、システムのすべてのユーザーに EOS プロデューサーへのアクセスを許可します。

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
   --add --transactional-id * --allow-principal User:* --operation write

Kafka クラスターの ACL を作成して、ストリームを厳密に 1 回（EOS）処理できるようにする必要がある場合:

# Allow Streams EOS:
 bin/kafka-acls ... --add --allow-principal User:team1 --operation WRITE \
   --operation DESCRIBE --transactional-id team1-streams-app1 \
   --resource-pattern-type prefixed

トランザクション ID のロールの追加情報については、「Transactions in Apache Kafka」を参照してください。

ユーザーが使用できる操作は、ユーザーがアクセス権限を付与されたリソースによって異なります。すべてのリソースには、一意に識別できるリソース ID があります。たとえば、リソースタイプがトピックの場合は、リソース ID はトピック名で、リソースタイプがグループの場合は、リソース ID はグループ名です。

特定リソースの ACL を表示するには、--list オプションを使用できます。たとえば、トピック test-topic の ACL をすべて表示するには、次のコマンドを実行します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
   --list --topic test-topic

プレフィックス付き ACL¶

LITERAL 値（デフォルト）、PREFIXED パターンタイプ、またはすべてを許可するワイルドカード（*）で、ACL リソースを指定できます。

LITERAL を使用してリソースを識別する場合、Kafka では、完全なリソース名（トピックやコンシューマーグループなど）を ACL に指定されたリソースと一致させようとします。場合により、アスタリスク（*）を使用してすべてのリソースを指定することもできます。

PREFIXED を使用してリソースを指定すると、Kafka では、リソース名のプレフィックスと ACL で指定されたリソースを一致させようとします。

たとえば、ユーザー User:kafka/kafka1.host-1.com@bigdata.com に対して、プレフィックス Test- を使用した名前のトピックを生成するための ACL を追加できます。次のコマンドを実行して、これを実行できます。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
   --add --allow-principal User:kafka/kafka1.host-1.com@bigdata.com \
   --producer --topic Test- --resource-pattern-type prefixed

次に示すのは、Kafka Java SDK でビルドされた "BillingPublisher" というプログラムで、プレフィックス billing- のトピックだけに書き込みできる ACL が必要になる場合の例です。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add \
--allow-principal User:BillingPublisher \
--allow-host 198.51.100.0 --producer --topic billing- --resource-pattern-type prefixed

単一のコマンドで、すべてのグループ * （ワイルドカード）へのアクセスを許可する一方で、トピックの PREFIXED リソースパターンタイプを使用することはできないことに注意してください。代わりに、別のコマンドに分けて、アクセス許可を指定します。たとえば、プレフィックス付きのトピックでユーザーに READ と DESCRIBE アクセスを付与します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add \
--allow-principal User:username --operation Read --operation Describe --topic topicprefix --resource-pattern-type prefixed

次にすべてのグループへの READ アクセスをユーザーに付与します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add \
--allow-principal User:username --operation Read --group '*'

スーパーユーザー¶

デフォルトでは、リソースに ACL が関連付けられていない場合、スーパーユーザー以外のユーザーはリソースにアクセスできません。動作を変更する場合は、server.properties に allow.everyone.if.no.acl.found=true を含めます。

注釈

本稼働環境では allow.everyone.if.no.acl.found 構成オプションを使用しないことを強くお勧めします。

ACL があるという前提でこのオプションを指定し、最後の ACL が削除される場合、実質的にすべてのユーザーに対して Kafka クラスターを開きます。
このオプションを使用して ACL を無効にする場合は、次の点に注意してください。ACL が追加されると、以前にアクセス権を持っていたすべてのユーザーがそのアクセス権を失います。

次のとおりスーパーユーザーを server.properties に追加できます（区切り文字がセミコロンであることに注意。SSL ユーザー名にコンマが含まれるため）。

super.users=User:Bob;User:Alice

ACL とモニタリングインターセプター¶

Confluent Monitoring Interceptor はデフォルトで、_confluent-monitoring トピックにメッセージを生成します。_confluent-monitoring トピックを confluent.monitoring.interceptor.topic 属性を使用して構成できます。アプリケーションつまりプログラムからの _confluent-monitoring トピックへのアクセスには、WRITE および DESCRIBE アクセスが必要です。_confluent-monitoring トピックが存在しない場合、作成するために、クラスターレベルの CREATE および DESCRIBE アクセスが必要です。_confluent-monitoring に対して、トピックレベルの CREATE、DESCRIBE、READ および WRITE アクセスも必要です。

インターセプターを使用する各クライアントプリンシパルに個別にアクセス許可を付与することができます。または、ワイルドカードを使用すると、すべてのクライアントを許可できます。次の例では、_confluent-monitoring トピックへのプリンシパルのアクセスを許可する ACL を示します。

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
--add --topic _confluent-monitoring --allow-principal User:username --operation write --operation Describe

Confluent Control Center プリンシパルには、_confluent-monitoring トピックへの READ、DESCRIBE および CREATE アクセスが必要になります。control-center-set-acls スクリプトを使用して、Confluent Control Center プリンシパルがこのトピックにアクセスするための適切なアクセス許可を設定します。詳細については、「Control Center UI 認証」を参照してください。

ACL の使用¶

次のセクションの例では、bin/kafka-acls （Kafka Authorization management CLI）を使用して、ACL の追加、削除、リスト表示を実行します。サポートされるオプションの詳細については、bin/kafka-acls --help を実行します。ACL は ZooKeeper に格納され、各ブローカーに非同期で伝播されます。つまりコマンドの応答が返った後でも、変更が有効になるまでに遅延が発生することがあります。

Kafka AdminClient API を使用して ACL を管理することもできます。

ちなみに

トランザクション（--transactional-id）を使用している場合、IdempotentWrite ACL が暗黙的に想定されます。
トランザクションを使用していない場合、--idempotent オプションを使用して、IdempotentWrite ACL を有効にできます。

よくある ACL のユースケースの一部を次に示します。

トピックを作成、クライアントのプリンシパルでは、Topic または Cluster リソースに対して CREATE および DESCRIBE 操作が必要になります。
トピックに生成、プロデューサーのプリンシパルでは Topic リソースに対する WRITE 操作が必要になります。
トピックからの消費、コンシューマーのプリンシパルでは、Topic および Group リソースに対する READ 操作が必要になります。

作成、生成、消費ができるためには、サーバーを適切な ACL で構成する必要があることに注意してください。サーバーでは、メタデータをアップデートし（CLUSTER_ACTION）、レプリケーション目的でトピックから読み取る（READ）ための認可が必要になります。

ACL フォーマット¶

Kafka の ACL は "プリンシパル P のホスト H からの操作 O をリソースパターン RP に一致するリソースで許可または拒否する" という一般フォーマットで定義されます。

ワイルドカードをどのリソースにも適用できます。
トピックとグループのワイルドカードによるアクセスを、すべてのトピックとグループにアクセスするアクセス許可を保持するユーザー（管理ユーザーなど）に付与できます。この方式を使用する場合、ユーザーに対して各トピックとグループに個別のルールを作成する必要はありません。たとえば、次のコマンドで Alice にワイルドカードアクセスを付与できます。
```
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal \
User:Alice --operation All --topic '*' --group '*'
```

構成プロパティファイルの使用¶

プロデューサー API、コンシューマー API または Streams API を Kafka で以前に使用していた場合、クラスターへの接続の詳細情報は構成プロパティを使用して指定されていることを把握しているかもしれません。Kafka とやり取りするように開発されたアプリケーションでこれを認識しているユーザーもいれば、Kafka に付属する管理ツールが同様に動作することを知らないユーザーもいます。つまり構成プロパティ（多くの場合、config.properties ファイルのフォーム）を定義した後、アプリケーションまたはツールでクラスターに接続できます。

ユーザーのホームディレクトリに構成プロパティファイルを作成すると、発行する後続のコマンド（必ず構成ファイルのパスを含める）はそのファイルを読み込み、それを使用して、Kafka クラスターへの接続を確立します。つまりネイティブ Kafka ツールを使用して、Kafka クラスターとやり取りするために、まず必要なのは、構成プロパティファイルを生成することです。

--command-config 引数で、Confluent CLI ツールに Kafka クラスターに接続する必要があるツールの構成プロパティを、.properties ファイルフォーマットで指定します。これに含まれるのは一般的に、クラスターが接続に使用する security.protocol と、クラスターで認証するために必要なすべての情報です。以下に例を示します。

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="alice" \
  password="s3cr3t";

ACL の追加¶

次の ACL を追加することを考えます。プリンシパル User:CN=Jane Smith,OU=Sales,O=Unknown,L=Unknown,ST=Unknown,C=Unknown と User:CN=Bob Thomas,OU=Sales,O=Unknown,L=Unknown,ST=NY,C=Unknown が、読み取りと書き込みの操作をトピック test-topic に対して、IP 198.51.100.0 および IP 198.51.100.1 から許可される ACL です。これには次のとおり実行します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add \
 --allow-principal "User:CN=Bob Thomas,OU=Sales,O=Unknown,L=Unknown,ST=NY,C=Unknown" \
 --allow-principal "User:CN=Jane Smith,OU=Sales,O=Unknown,L=Unknown,ST=Unknown,C=Unknown" \
 --allow-host 198.51.100.0 --allow-host 198.51.100.1 \
 --operation Read --operation Write --topic test-topic

デフォルトでは、リソースにアクセスするための操作を許可する明示的な ACL を保持していないすべてのプリンシパルは拒否されます。まれなケースですが、一部を除くすべてのプリンシパルへのアクセスを許可する ACL が必要な場合は、--deny-principal と --deny-host オプションを使用できます。たとえば、次のコマンドで、test-topic からの読み取りをすべてのユーザーに許可しますが、IP 198.51.100.3 からの User:kafka/kafka6.host-1.com@bigdata.com のみを拒否します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add \
 --allow-principal User:'*' --allow-host '*' --deny-principal User:kafka/kafka6.host-1.com@bigdata.com --deny-host 198.51.100.3 \
 --operation Read --topic test-topic

Kafka では証明書失効リスト（CRL）をサポートしていません。つまりクライアントの証明書を無効にできません。したがって、代わりに実行できるのは、ACL でユーザーのアクセスを無効にすることだけです。

kafka-acls --bootstrap-server localhost:9092 --add --deny-principal "User:CN=Bob,O=Sales" --cluster --topic '*'

--allow-host と deny-host は IP アドレスのみをサポートすることに注意してください（ホスト名は未サポートです）。一方 IPv6 アドレスがサポートされ、ACL で使用できます。

前に示した例では、--topic [topic-name] をリソースパターンオプションに指定することで、ACL をトピックに追加します。同様に、--cluster を指定してクラスターに、さらに --group [group-name] を指定してグループに ACL を追加できます。すべてのグループにアクセス許可を付与する場合、次のコマンドのとおり --group='*' を指定して実行できます。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add \
--allow-principal User:'*' --operation read --topic test --group '*'

プレフィックス付きリソースパターンに対して ACL を追加できます。たとえば、部門名（OU） ServiceUsers （この部門は TLS/SSL 認証で使用）に所属するユーザーが、名前が Test- で始まるトピックに生成できるようにする ACL を追加できます。これには次のオプションの CLI を実行します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal \
  User:CN=serviceuser,OU=ServiceUsers,O=Unknown,L=Unknown,ST=Unknown,C=Unknown --producer --topic Test- --resource-pattern-type prefixed

--resource-pattern-type のデフォルトは literal で、まったく同じ名前のリソースにのみ影響があります。ただし、ワイルドカードのリソース名 '*' の場合は、あらゆる名前のリソースに影響があります。

ACL の削除¶

ACL の削除は追加と同様ですが、--remove オプションを --add の代わりに指定します。前に示した最初の例で追加した ACL を削除するには、次のとおり実行します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --remove \
 --allow-principal "User:CN=Bob Thomas,OU=Sales,O=Unknown,L=Unknown,ST=NY,C=Unknown" \
 --allow-principal "User:CN=Jane Smith,OU=Sales,O=Unknown,L=Unknown,ST=Unknown,C=Unknown" \
 --allow-host 198.51.100.0 --allow-host 198.51.100.1 \
 --operation Read --operation Write --topic test-topic

この例にあるプレフィックス付きリソースパターンに追加された ACL を削除する場合は、次のオプションで CLI を実行します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --remove \
  --allow-principal User:CN=Jane Smith,OU=Sales,O=Unknown,L=Unknown,ST=Unknown,C=Unknown \
  --producer --topic Test- --resource-pattern-type Prefixed

ACL のリスト表示¶

--list オプションとリソースを指定することにより、対象リソースの ACL リストを表示できます。たとえば、test-topic に対するすべての ACL をリスト表示するには、次のコマンドを実行します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
 --list --topic test-topic

ただし、これにより返るのは、まったく同じリソースパターンに追加された ACL だけです。このトピックへのアクセスに影響する他の ACL が存在することがあります（たとえば、ワイルドカード '*' のトピックに対する ACL またはプレフィックス付きリソースパターンの ACL）。ワイルドカードのリソースパターンに対する ACL を明示的にクエリできます。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --list --topic '*'

Test-topic に一致するプレフィックス付きリソースパターンに対する ACL を必ずしも明示的にクエリできません。そのようなパターンの名前がわからないことがあるためです。--resource-pattern-type match を使用することで、Test-topic に影響を及ぼすすべての ACL をリスト表示できます。以下に例を示します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --list --topic Test-topic --resource-pattern-type match

このコマンドは、リテラル、ワイルドカード、プレフィックス付きリソースパターンにすべて一致する ACL のリストを表示します。

内部トピックの ACL を表示するには、次のとおり実行します。

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --list --topic __consumer_offsets

プロデューサーまたはコンシューマーであるプリンシパルの追加や削除¶

ACL 管理の最も一般的なユースケースは、プロデューサーまたはコンシューマーとしてプリンシパルを追加または削除することです。ユーザー "Jane Doe"（Kerberos プラットフォーム User:janedoe@bigdata.com）を test-topic のプロデューサーとして追加するには、次のとおり実行します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
 --add --allow-principal User:janedoe@bigdata.com \
 --producer --topic test-topic

User:janedoe@bigdata.com を test-topic でグループ Group-1 のコンシューマーとして追加するには、--consumer と --group オプションを指定します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
 --add --allow-principal User:janedoe@bigdata.com \
 --consumer --topic test-topic --group Group-1

プリンシパルをプロデューサーまたはコンシューマーロールから削除するには、--remove オプションを指定します。

べき等 API とトランザクション API に対する認可の有効化¶

プロデューサーでは enable.idempotence=true を構成して、各メッセージの厳密に 1 つのコピーがストリームに書き込まれることを保証できます。べき等性が有効なプロデューサーによって使用されるプリンシパルは、クラスターで IdempotentWrite を実行することを認可される必要があります。

べき等性が有効なプロデューサーを使用して Bob がメッセージを生成できるようにするには、次のコマンドを実行します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
 --add --allow-principal User:Bob \
 --producer --topic test-topic --idempotent

プロデューサーで空ではない transactional.id を構成し、複数のプロデューサーセッションにまたがる信頼性セマンティクスでトランザクションデリバリーを有効にすることもできます。トランザクションプロデューサーで使用されるプリンシパルには、Describe および Write 操作を、構成された transactional.id で認可する必要があります。

transactional.id=test-txn のトランザクションプロデューサーを使用して Alice がメッセージを生成できるようにするには、次のコマンドを実行します。

bin/kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
 --add --allow-principal User:Alice \
 --producer --topic test-topic --transactional-id test-txn

構成されたトランザクション ID に対して ACL が構成されているトランザクションプロデューサーに対して、べき等性書き込みアクセスが自動的に許可されることに注意してください。

スーパーユーザー以外の ACL 管理者の作成¶

ACL を作成または削除できるが、スーパーユーザーのロールが与えられていないユーザーが必要な場合、現在のスーパーユーザーが別のユーザー（ここでは ACL 管理者と呼びます）に ALTER --cluster アクセス制御エントリ（ACE）を許可できます。これは操作（この場合は "alter"）をリソース "cluster" にバインドします。ACL 管理者に ALTER --cluster ACE が付与されると、そのユーザーはクラスターの対象リソースで ACL を作成および削除できます。

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
--add --allow-principal User:notSuper \
--operation  ALTER --cluster

注釈

ALTER --cluster をグループに割り当てる場合は、Group:groupName も有効ですが、使用しているオーソライザーで、グループの処理と許可をできるようにする必要があります。
ALTER --cluster をユーザーやグループに割り当てるときに注意が必要です。そのようなユーザーはリソースへの自身のアクセスを制御する ACL を作成および削除できるためです。

REST Proxy と Schema Registry での認可¶

Kafka の ACL を使用して REST Proxy と Schema Registry で認可を実行できます。これらには、REST Proxy セキュリティプラグインと Schema Registry セキュリティプラグインが必要です。

デバッグ¶

log4j.properties ファイルに変更を加えることで DEBUG モードでオーソライザーログ付きで実行することができます。デフォルトの log4j.properties ファイルを使用している場合、次の行を WARN から DEBUG モードに変更します。

log4j.logger.kafka.authorizer.logger=WARN, authorizerAppender

log4j.properties ファイルは、Kafka の構成ディレクトリにあります（/etc/kafka/log4j.properties）。Confluent Platform の前のバージョンを使用している場合、または独自の log4j.properties ファイルを使用している場合、次の行を構成に追加する必要があります。

log4j.appender.authorizerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.authorizerAppender.File=${kafka.logs.dir}/kafka-authorizer.log
log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.logger.kafka.authorizer.logger=DEBUG, authorizerAppender
log4j.additivity.kafka.authorizer.logger=false

有効にするには、ブローカーを再起動する必要があります。これにより認可されたすべてのリクエストと関連ユーザー名がログに記録されます。ログは $kafka_logs_dir/kafka-authorizer.log に記録されます。ログの場所はパッケージ形式によって異なり、kafka_logs_dir は、rpm 形式および debian 形式では /var/log/kafka になり、アーカイブ形式では $base_dir/logs になります。