設定 mTLS 驗證

您可以設定 Managed Service for Apache Kafka 叢集，使用相互傳輸層安全標準 (mTLS) 驗證 Kafka 用戶端。這個方法會使用憑證授權單位服務 (CA 服務) 的用戶端憑證做為驗證依據。這個選項可做為預設 SASL 機制的替代方案，後者會使用 Identity and Access Management (IAM) 主體。

使用 mTLS 時，必須使用 Kafka ACL 設定授權。如要瞭解基礎概念，請參閱下列文件：

事前準備

設定 mTLS 驗證前，請先完成下列步驟：

確認叢集是否符合資格。確認您有在 2025 年 6 月 24 日後建立的現有 Managed Service for Apache Kafka 叢集。只有這些叢集支援 mTLS 驗證。如要驗證叢集的建立日期，請使用 gcloud managed-kafka clusters describe 指令，或在控制台中查看叢集的詳細資料頁面。
設定 CA 服務。設定您打算用來核發用戶端憑證的 CA 服務和 CA 集區。您必須在 CA 集區內建立根憑證和從屬憑證。
1. 建立 CA 集區。記下 CA 集區 ID。
  
  如要進一步瞭解如何建立 CA 集區，請參閱「建立 CA 集區」。
2. 為集區建立並啟用根 CA。
  
  如要進一步瞭解如何為集區啟用根 CA，請參閱「建立根 CA」。
3. 建立並啟用一或多個從屬 CA。建議使用下層 CA 核發用戶端憑證，不要直接使用根 CA。
  
  如要進一步瞭解如何建立從屬 CA，請參閱「建立從屬憑證授權單位」。

必要角色和權限

如要設定 mTLS，請確保您 (使用者) 和 Managed Service for Apache Kafka 服務代理都具備必要的 IAM 權限。無論是建立新叢集或更新現有叢集來設定 mTLS，都適用這項規定。

使用者權限

如要建立或設定 Managed Service for Apache Kafka 叢集以使用 mTLS，您需要建立或更新叢集資源的權限。如要這麼做，請要求管理員授予您包含叢集的專案 Managed Kafka 叢集編輯者 (roles/managedkafka.clusterEditor) 角色。

這個預先定義的角色具備 managedkafka.clusters.create 和 managedkafka.clusters.update 權限。您可透過這些權限建立新叢集，或修改現有叢集，新增 mTLS 必要的 CA 服務 (CA) 集區設定。

只要您擁有 CA 集區的完整資源路徑，就不需要在 CA 服務資源上取得個別權限，即可在 Kafka 叢集上設定 mTLS。不過，如要在Google Cloud 控制台中查看、建立或管理 CA 集區，您需要 CA 服務專用的其他角色，例如「CA 服務管理員」 (roles/privateca.admin) 或「CA 服務運算子」 (roles/privateca.operator)。

服務代理權限

如要讓 mTLS 整合功能正常運作，Managed Service for Apache Kafka 服務代理程式必須有權存取指定的 CA 集區。服務代理程式是專案的 Google 管理服務帳戶。

如果 Managed Service for Apache Kafka 叢集和 CA 集區位於同一專案，服務代理預設會具備必要權限。系統會自動將 managedkafka.serviceAgent 角色授予專案中的服務代理，其中包含必要的 privateca.caPools.get 權限。
如果 CA 集區與 Managed Service for Apache Kafka 叢集位於不同專案，您必須手動授予服務代理存取權。在包含 CA 集區的專案中，授予服務代理人「私人 CA 集區讀取者」 (roles/privateca.poolReader) 角色。

所需權限摘要

如要查看確切的必要權限，請展開下列部分。

所需權限

使用者權限：

managedkafka.clusters.create：建立設定為 mTLS 的新 Kafka 叢集時，必須提供這項資訊。
managedkafka.clusters.update：更新現有 Kafka 叢集以設定 mTLS 時，必須提供這項資訊。

Managed Service for Apache Kafka 服務代理的權限：

privateca.caPools.get：服務代理人存取 CA 集區時必須提供這項資訊。這項權限包含在 roles/privateca.poolReader 角色中。

您或許還可透過自訂角色或其他預先定義的角色取得這些權限。

授予服務代理人 CA 集區存取權

如果 CA Service CA 集區和 Managed Service for Apache Kafka 叢集位於不同 Google Cloud 專案，您必須授予叢集服務代理存取 CA 集區的權限。Managed Service for Apache Kafka 服務代理的名稱為 service-CLUSTER_PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com。

在包含 CA 的個別集區層級 (建議) 或專案中的所有集區，將 CA 集區讀取者 (roles/privateca.poolReader) 角色授予 Managed Service for Apache Kafka 服務代理。這個角色具備必要的 privateca.caPools.get 權限。

個別 CA 集區

建議您只授予單一 CA 集區的權限，因為這樣符合最低權限原則。

執行 gcloud privateca pools add-iam-policy-binding 指令：

gcloud privateca pools add-iam-policy-binding CA_POOL_ID \
    --location=CA_POOL_LOCATION \
    --member='serviceAccount:service-CLUSTER_PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com' \
    --role='roles/privateca.poolReader'

更改下列內容：

CA_POOL_ID：您要授予存取權的 CA 集區 ID。例如：test-mtls-pool1。
CA_POOL_LOCATION：CA 集區所在的 Google Cloud 區域。例如：us-central1。
CLUSTER_PROJECT_NUMBER：包含 Managed Service for Apache Kafka 叢集的專案編號。例如：12341234123。

所有 CA 集區

或者，您也可以在專案層級設定政策，授予服務代理存取專案中所有 CA 集區的權限。

執行 gcloud projects add-iam-policy-binding 指令：

gcloud projects add-iam-policy-binding CA_PROJECT_ID \
    --member='serviceAccount:service-CLUSTER_PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com' \
    --role='roles/privateca.poolReader'

更改下列內容：

CA_PROJECT_ID：包含您要授予存取權的 CA 集區的專案 ID。例如：test-cas-project。
CLUSTER_PROJECT_NUMBER：包含 Managed Service for Apache Kafka 叢集的專案編號。例如：12341234123。

在叢集上啟用 mTLS

如要啟用 mTLS，請為叢集提供一或多個 CA 服務 CA 集區的資源名稱，用於用戶端驗證。建立新叢集時，或更新 2025 年 6 月 24 日後建立的現有叢集時，您都可以執行這項操作。

提供 CA 集區 ID 後，服務會自動從指定集區下載 CA 憑證，並安裝至叢集中每個代理程式的信任儲存區。

控制台

您可以在建立新叢集時啟用 mTLS，也可以編輯現有叢集來啟用 mTLS。

於新叢集

前往 Google Cloud 控制台的「Clusters」(叢集) 頁面。

前往「Clusters」(叢集) 頁面
選取 [Create] (建立)。
「建立 Kafka 叢集」頁面隨即開啟。
請按照「建立叢集」一文中的步驟操作。
在最後一個步驟之前，請找出「Optional mTLS configuration」(選用 mTLS 設定) 一節。
請按照下列格式輸入 CA 集區的完整資源名稱：projects/PROJECT_ID/LOCATION/LOCATION/caPools/POOL_ID。
如要新增更多，請按一下「新增憑證授權單位集區」。最多可以新增 10 個 CA 集區。
(選用) 輸入任何主體對應規則。
按一下「建立」，即可建立啟用 mTLS 的叢集。

於現有叢集

前往 Google Cloud 控制台的「Clusters」(叢集) 頁面。
前往「Clusters」(叢集) 頁面
按一下要更新的叢集名稱。
按一下 [編輯]。
在「mTLS 設定」部分，新增或修改 CA 集區清單。最多可以新增 10 個 CA 集區。
(選用) 輸入或編輯任何主體對應規則。
按一下 [儲存]。

gcloud

於新叢集

In the Google Cloud console, activate Cloud Shell.

Activate Cloud Shell

At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

執行加上 --mtls-ca-pools 旗標的 gcloud managed-kafka clusters create 指令。在本範例中，設定了兩個 CA 集區。

gcloud managed-kafka clusters create CLUSTER_ID \
    --location=LOCATION \
    --cpu=3 \
    --memory=3GiB \
    --subnets=projects/PROJECT_ID/locations/LOCATION/subnetworks/SUBNET_ID \
    --mtls-ca-pools=projects/PROJECT_ID/locations/LOCATION/caPools/POOL_ID_1,projects/PROJECT_ID/locations/LOCATION/caPools/POOL_ID_2

更改下列內容：

CLUSTER_ID：叢集的 ID 或名稱。
如要進一步瞭解如何命名叢集，請參閱「 Managed Service for Apache Kafka 資源命名指南」。例如：test-mtls-cluster。
LOCATION：叢集位置。
如要進一步瞭解支援的位置，請參閱「支援的 Managed Service for Apache Kafka 位置」。例如：us-central1。
SUBNETS：要連線的子網路清單。如有多個子網路值，請使用半形逗號分隔。
子網路的格式為 projects/PROJECT_ID/locations/LOCATION/subnetworks/SUBNET_ID。

POOL_ID_1：第一個 CA 集區的 ID。例如：test-mtls-pool1。

POOL_ID_2：第二個 CA 集區的 ID。例如：test-mtls-pool2。

於現有叢集

In the Google Cloud console, activate Cloud Shell.

Activate Cloud Shell

At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

使用 gcloud managed-kafka clusters update 指令。這項指令會覆寫目前設定的整組集區。請提供必要 CA 集區的完整清單。在本範例中，設定了兩個 CA 集區。

gcloud managed-kafka clusters update CLUSTER_ID \
    --location=LOCATION \
    --mtls-ca-pools=projects/PROJECT_ID/locations/LOCATION/caPools/POOL_ID_1,projects/PROJECT_ID/locations/LOCATION/caPools/POOL_ID_2

更改下列內容：

CLUSTER_ID：叢集的 ID 或名稱。
如要進一步瞭解如何命名叢集，請參閱「 Managed Service for Apache Kafka 資源命名指南」。例如：test-mtls-cluster。
LOCATION：叢集位置。
如要進一步瞭解支援的位置，請參閱「支援的 Managed Service for Apache Kafka 位置」。例如：us-central1。

POOL_ID_1：第一個 CA 集區的 ID。例如：test-mtls-pool1。

POOL_ID_2：第二個 CA 集區的 ID。例如：test-mtls-pool2。

設定主體名稱對應

當用戶端透過 mTLS 進行驗證時，Kafka 主體預設會衍生自憑證的主體識別名稱 (DN)，且格式為 User:CN=...,OU=...,O=...,L=...,ST=...,C=...。建立對應規則，將憑證的主體 DN 轉換成方便使用的別名，以便在 Kafka 存取控制清單中使用。如要進一步瞭解主體 DN 的格式，請參閱 Apache Kafka 說明文件中的「Customizing SSL User Name」。

這些轉換是由 ssl.principal.mapping.rules Kafka 代理程式屬性定義，該屬性會使用規則運算式擷取並重新格式化憑證主體的部分內容。

舉例來說，您可以套用規則，將完整主體 DN 轉換為別名，如下所示：

憑證主體 DN： CN=order-processing-app,OU=marketing,O=company,C=US
對應規則： RULE:^.*[Cc][Nn]=([a-zA-Z0-9.-]*).*$/$1/L,DEFAULT
產生的 Kafka 主體： order-processing-app

這個範例規則會從憑證主體擷取共用名稱 (CN)，並將其做為 Kafka 中的主體名稱。

如要使用 Google Cloud CLI 在叢集上設定對應規則，請按照下列步驟操作。使用主控台時，您可以在建立或更新叢集時設定對應規則。

如要更新叢集，請使用 gcloud managed-kafka clusters update 指令並搭配 --ssl-principal-mapping-rules 旗標。
```
gcloud managed-kafka clusters update CLUSTER_ID \
    --location=REGION \
    --ssl-principal-mapping-rules='MAPPING_RULE'
```
更改下列內容：
- CLUSTER_ID：待建立的 Managed Service for Apache Kafka 叢集 ID。例如：test-kafka-cluster。
- REGION：用於建立叢集的 Google Cloud 區域。例如：us-central1。
- MAPPING_RULE*：要套用的對應規則。例如：RULE:^.*[Cc][Nn]=([a-zA-Z0-9.-]*).*$/$1/L,DEFAULT。

如要進一步瞭解如何編寫對應規則，請參閱 Apache Kafka 說明文件中的「授權和 ACL」。

為 mTLS 主體設定 Kafka ACL

根據預設，任何使用有效 mTLS 憑證成功完成驗證的用戶端，都會獲得叢集的完整存取權。如要強制執行最低權限原則，您必須建立 Kafka ACL，為 mTLS 主體定義特定權限。mTLS 用戶端的主體是其憑證的主體 DN (或對應的別名)，並以 User: 為前置字元。

如要建立 Kafka ACL，您需要 Managed Kafka ACL 編輯者 (roles/managedkafka.aclEditor) IAM 角色。

假設您有一個應用程式 (由憑證識別)，會將訊息產生至 orders-topic，並從 analytics-topic 接收訊息。應用程式主體經過對應規則簡化後，會變成 order-processing-app。建立 Kafka ACL 時，主體必須加上 User: 前置字元。

將 WRITE ACL 套用至叢集。執行 gcloud managed-kafka acls add-entry 指令，在 orders-topic 上授予 WRITE 權限。
```
gcloud managed-kafka acls add-entry topic/orders-topic \
    --cluster=CLUSTER_ID \
    --location=REGION \
    --principal=User:order-processing-app \
    --operation=WRITE \
    --permission-type=ALLOW \
    --host="*"
```
更改下列內容：
- CLUSTER_ID：待建立的 Managed Service for Apache Kafka 叢集 ID。例如：test-kafka-cluster。
- REGION：用於建立叢集的 Google Cloud 區域。例如：us-central1。

將 READ ACL 套用至叢集。執行 gcloud managed-kafka acls add-entry 指令，在 analytics-topic 上授予 READ 權限。

gcloud managed-kafka acls add-entry topic/analytics-topic \
    --cluster=CLUSTER_ID \
    --location=REGION \
    --principal=User:order-processing-app \
    --operation=READ \
    --permission-type=ALLOW \
    --host="*"

套用這些 ACL 後，order-processing-app 用戶端就只會擁有您授予的特定權限。如要進一步瞭解如何建立 ACL，請參閱「建立 Kafka ACL」。

設定 Kafka 用戶端

在叢集上設定 mTLS 後，您必須設定用戶端應用程式，才能使用這個方法進行驗證。這個程序包括建立用戶端憑證，以及設定用戶端屬性以使用該憑證。

在用戶端機器上建立並下載用戶端憑證。執行 gcloud privateca certificates create 指令，從您為叢集設定的其中一個 CA 集區發出新憑證。

這項指令會將憑證 client-cert.pem 和私密金鑰 client-key.pem 下載至本機環境。
```
gcloud privateca certificates create CERTIFICATE_ID \
    --project=PROJECT_ID \
    --issuer-location=REGION \
    --issuer-pool=POOL_ID \
    --ca=CA_NAME \
    --generate-key \
    --dns-san="client.example.com" \
    --subject="CN=test-client-app" \
    --key-output-file=client-key.pem \
    --cert-output-file=client-cert.pem
```
更改下列內容：
- CERTIFICATE_ID：憑證物件的專屬名稱。例如：order-app-cert。
- PROJECT_ID：包含 CA 集區的專案 ID。例如：test-project-12345。
- REGION：CA 集區所在的區域。例如：us-central1。
- POOL_ID：要核發憑證的 CA 集區 ID。例如：test-mtls-pool1。
- CA_NAME：集區中的憑證授權單位名稱。例如：test-sub-ca。
- --dns-san="client.example.com"：DNS 主體替代名稱。您可以為用戶端使用任何相關值。
- --subject="CN=test-client-app"：主體 DN。除非您已設定主體名稱對應規則，否則這個名稱會做為 mTLS 主體。

查看用戶端憑證、憑證主體，以及驗證 ssl.principal.mapping.rules。執行 gcloud privateca certificates describe 指令：

gcloud privateca certificates describe CERTIFICATE_ID \
   --issuer-pool=POOL_ID \
   --issuer-location=REGION

更改下列內容：

CERTIFICATE_ID：憑證物件的專屬名稱。例如：order-app-cert。
POOL_ID：您核發憑證的 CA 集區 ID。例如：test-mtls-pool1。
REGION：CA 集區所在的區域。例如：us-central1。

輸出結果會與下列內容相似：

certificateDescription:
  aiaIssuingCertificateUrls:
  - http://privateca-content-68e092f4-0000-288c-95cf-30fd3814648c.storage.googleapis.com/a6553d092bbedd752e34/ca.crt
  authorityKeyId:
    keyId: 9568aa9d2baa11a097addc2e24adeaebea0d6a2a
  certFingerprint:
    sha256Hash: 230e52b8411fd094048fca194fc6cf80e41b3e8561298aec3519e13cb1fd05eb
  ...
  subjectDescription:
    hexSerialNumber: 2107b74cf5a814043a38a87eeb6cd7c7891a5f
    lifetime: P30D
    notAfterTime: '2025-07-13T15:34:43Z'
    notBeforeTime: '2025-06-13T15:34:44Z'
    subject:
      commonName: test-client-app
    subjectAltName:
      dnsNames:
      - client.example.com
  ...

建立 Java KeyStore。將憑證和私密金鑰合併為 PKCS#12 檔案，然後匯入 Java KeyStore (.jks) 檔案。

# Create a password for the keystore
export KEYSTORE_PASSWORD="KEYSTORE_PASSWORD"

# Combine the key and cert into a PKCS#12 file
openssl pkcs12 -export -inkey client-key.pem -in client-cert.pem \
  -name client -out client-keystore.p12 -password "pass:$KEYSTORE_PASSWORD"

# Import the PKCS#12 file into a Java KeyStore
keytool -importkeystore -srckeystore client-keystore.p12 -srcstoretype pkcs12 \
  -destkeystore client-keystore.jks -srcstorepass "$KEYSTORE_PASSWORD" -deststorepass "$KEYSTORE_PASSWORD"

執行下列指令，即可確認金鑰是否已成功儲存：

keytool -v -list -keystore client-keystore.jks -storepass "$KEYSTORE_PASSWORD"

輸出結果會與下列內容相似：

Keystore type: JKS
Keystore provider: SUN

Your keystore contains 1 entry

Alias name: client
Creation date: Jun 13, 2024
Entry type: Private key entry
Certificate chain length: 1
Certificate[1]:
Owner: CN=test-client-app
Issuer: CN=test-sub-ca
...

請注意，Owner 行會顯示憑證主體 DN。根據預設，Kafka 會將 Kafka 主體設為以下確切格式：CN=...,OU=...,O=...,L=...,ST=...,C=...。詳情請參閱 Apache Kafka 說明文件中的「授權和 ACL」。

設定 Kafka 用戶端屬性和啟動位址。在 Kafka 用戶端應用程式中，設定下列屬性，以便使用金鑰儲存區進行 SSL 連線。此外，請務必使用正確的啟動程序位址和連接埠 9192。如要進一步瞭解如何設定用戶端，請參閱「快速入門：建立 Managed Service for Apache Kafka 叢集並連線至用戶端」。
```
security.protocol=SSL
ssl.keystore.location=KEYSTORE_LOCATION
ssl.keystore.password=KEYSTORE_PASSWORD
bootstrap.servers=CLUSTER_BOOTSTRAP_ADDRESS
```
更改下列內容：
- KEYSTORE_LOCATION：.jks 檔案的路徑。
- KEYSTORE_PASSWORD：金鑰存放區的密碼。
- CLUSTER_BOOTSTRAP_ADDRESS：叢集的啟動位址。如要尋找啟動位址，請參閱「查看叢集詳細資料」。請務必新增通訊埠編號 9192。

保護用戶端設定

前述範例涉及在本機儲存私密金鑰和密碼，因此不建議用於正式環境。在正式環境中，請妥善處理用戶端密鑰。選項包括：

將金鑰庫及其密碼儲存在 Google Cloud Secret Manager 中做為密鑰，並在應用程式程式碼的執行階段擷取這些密鑰。
如果您在 GKE 上部署應用程式，請使用 Secret Manager 外掛程式，在執行階段將密鑰掛接到應用程式的檔案系統。

監控 mTLS

您可以使用 Cloud Monitoring 和 Cloud Logging 中的指標和記錄，監控 mTLS 憑證更新的健康狀態。

如要主動監控 mTLS 憑證更新的健康狀態，請使用 Monitoring 中的 managedkafka.googleapis.com/mtls_truststore_update_count 指標。這項指標會計算信任儲存區的更新嘗試次數，並包含 STATUS 標籤，該標籤可以是 SUCCESS 或失敗原因，例如 CA_POOL_FETCH_ERROR。

Managed Service for Apache Kafka 服務會嘗試每小時為每個叢集重新整理一次信任儲存區。建議您建立警報，在指標回報錯誤的持續計數超過三小時時觸發警報，因為這可能表示設定錯誤，需要手動介入。

更新信任儲存區會耗用 Certificate Authority Service API 的配額。請務必瞭解下列事項：

更新程序會呼叫 FetchCaCerts 方法，該方法受限於 API requests per minute per region 配額。
這項配額用量會歸因於包含所參照 CA 集區的專案，而非 Managed Service for Apache Kafka 專案。
預設上限為每個區域每秒 400 次查詢 (QPS)。由於每小時每個叢集只會發出一次要求，因此您不太可能因為更新信任儲存區而超出配額。

您也可以在 Logging 中查看記錄，追蹤信任儲區更新。查看下列記錄項目，確認更新是否成功：

Managed Service for Apache Kafka updated the mTLS trust store
Added root CA certificate to trust store

後續步驟

瞭解如何建立叢集。
瞭解如何查看叢集詳細資料。

Apache Kafka® 是 The Apache Software Foundation 或其關聯企業在美國與/或其他國家/地區的註冊商標。

設定 mTLS 驗證 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

事前準備

必要角色和權限

所需權限

授予服務代理人 CA 集區存取權

個別 CA 集區

所有 CA 集區

在叢集上啟用 mTLS

控制台

於新叢集

於現有叢集

gcloud

於新叢集

於現有叢集

設定主體名稱對應

為 mTLS 主體設定 Kafka ACL

設定 Kafka 用戶端

保護用戶端設定

監控 mTLS

後續步驟

設定 mTLS 驗證