Configura l'autenticazione mTLS

Puoi configurare il cluster Managed Service per Apache Kafka per autenticare i client Kafka utilizzando mutual TLS (mTLS). Questo metodo utilizza i certificati client di Certificate Authority Service (CA Service) come base per l'autenticazione. Questa opzione fornisce un'alternativa al meccanismo SASL predefinito che utilizza i principal Identity and Access Management (IAM).

Quando utilizzi mTLS, l'autorizzazione deve essere configurata utilizzando gli ACL Kafka. Per i concetti di base, consulta i seguenti documenti:

Prima di iniziare

Prima di configurare l'autenticazione mTLS, completa le seguenti operazioni:

  • Conferma l'idoneità del cluster. Verifica di disporre di un cluster Managed Service per Apache Kafka esistente creato dopo il 24 giugno 2025. Solo questi cluster supportano l'autenticazione mTLS. Per verificare la data di creazione del cluster, utilizza il comando gcloud managed-kafka clusters describe o visualizza la pagina dei dettagli del cluster nella console.

  • Configura CA Service. Configura il servizio CA e i pool di CA che intendi utilizzare per l'emissione dei certificati client. Devi creare certificati radice e subordinati all'interno dei pool di CA.

    1. Crea un pool di CA. Prendi nota dell'ID pool di CA.

      Per saperne di più su come creare un pool di CA, vedi Creare un pool di CA.

    2. Crea e attiva una CA radice per il pool.

      Per saperne di più su come attivare una CA radice per il pool, consulta Creare una CA radice.

    3. Crea e attiva una o più CA subordinate. Ti consigliamo di utilizzare una CA subordinata per l'emissione di certificati client anziché utilizzare direttamente una CA radice.

      Per saperne di più su come creare una CA subordinata, consulta Creare un'autorità di certificazione subordinata.

Ruoli e autorizzazioni richiesti

Per configurare mTLS, devi assicurarti che sia tu (l'utente) sia l'agente di servizio Managed Service per Apache Kafka dispongano delle autorizzazioni IAM necessarie. Questo vale sia per la creazione di un nuovo cluster sia per l'aggiornamento di uno esistente per la configurazione di mTLS.

Autorizzazioni utente

Per creare o configurare un cluster Managed Service per Apache Kafka per mTLS, devi disporre delle autorizzazioni per creare o aggiornare la risorsa cluster. Per farlo, chiedi all'amministratore di concederti il ruolo Editor cluster Kafka gestito (roles/managedkafka.clusterEditor) nel progetto contenente il cluster.

Questo ruolo predefinito contiene le autorizzazioni managedkafka.clusters.create e managedkafka.clusters.update. Queste autorizzazioni ti consentono di creare un nuovo cluster o modificarne uno esistente per aggiungere la configurazione del pool del servizio CA (CA) richiesta per mTLS.

Non sono necessarie autorizzazioni separate per le risorse del servizio CA per configurare mTLS sul cluster Kafka, purché tu disponga del percorso completo della risorsa del pool di CA. Tuttavia, per visualizzare, creare o gestire pool di CA nella consoleGoogle Cloud , avrai bisogno di ruoli aggiuntivi specifici per CA Service, come Amministratore servizio CA (roles/privateca.admin) o Operatore servizio CA (roles/privateca.operator).

Autorizzazioni service agent

Affinché l'integrazione mTLS funzioni, l'agente di servizio Managed Service per Apache Kafka richiede l'autorizzazione per accedere al pool di CA specificato. Il service agent è un account di serviziot gestito da Google per il tuo progetto.

  • Se il cluster Managed Service per Apache Kafka e il pool CA si trovano nello stesso progetto, il service agent dispone delle autorizzazioni necessarie per impostazione predefinita. Il ruolo managedkafka.serviceAgent, concesso automaticamente all'agente di servizio del tuo progetto, include l'autorizzazione privateca.caPools.get richiesta.

  • Se il pool di CA si trova in un progetto diverso dal cluster Managed Service per Apache Kafka, devi concedere manualmente all'agente di servizio l'autorizzazione di accesso. Concedi il ruolo Lettore pool di CA privata (roles/privateca.poolReader) all'agente di servizio nel progetto che contiene il pool di CA.

Riepilogo delle autorizzazioni richieste

Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione seguente.

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

Concedi all'agente di servizio l'accesso ai pool di CA

Se il pool di CA del servizio CA e il cluster Managed Service for Apache Kafka si trovano in Google Cloud progetti diversi, devi concedere all'agente di servizio del cluster l'autorizzazione di accesso al pool di CA. L'agente di servizio Managed Service per Apache Kafka si chiama service-CLUSTER_PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com.

Concedi il ruolo Lettore pool di CA (roles/privateca.poolReader) all'agente di servizio Managed Service per Apache Kafka a livello di singolo pool (consigliato) che contiene le tue CA o in tutti i pool del progetto. Questo ruolo fornisce l'autorizzazione privateca.caPools.get necessaria.

Pool di CA individuale

La concessione delle autorizzazioni a un singolo pool di CA è l'approccio consigliato, in quanto segue il principio del privilegio minimo.

Esegui il comando 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'

Sostituisci quanto segue:

  • CA_POOL_ID: L'ID del pool di CA a cui stai concedendo l'accesso. Ad esempio, test-mtls-pool1.

  • CA_POOL_LOCATION: la regione Google Cloud in cui si trova il pool di CA. Ad esempio: us-central1.

  • CLUSTER_PROJECT_NUMBER: il numero del progetto contenente il cluster Managed Service per Apache Kafka. Ad esempio, 12341234123.

Tutti i pool di CA

In alternativa, puoi concedere all'agente di servizio l'autorizzazione ad accedere a tutti i pool di CA all'interno di un progetto impostando il criterio a livello di progetto.

Esegui il comando 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'

Sostituisci quanto segue:

  • CA_PROJECT_ID: l'ID del progetto contenente i pool di CA a cui stai concedendo l'accesso. Ad esempio, test-cas-project.

  • CLUSTER_PROJECT_NUMBER: il numero del progetto contenente il cluster Managed Service per Apache Kafka. Ad esempio, 12341234123.

Abilitare mTLS su un cluster

Per attivare mTLS, fornisci al cluster i nomi delle risorse di uno o più pool di CA del servizio CA da utilizzare per l'autenticazione client. Puoi farlo quando crei un nuovo cluster o aggiornandone uno esistente creato dopo il 24 giugno 2025.

Dopo aver fornito gli identificatori del pool di CA, il servizio scarica automaticamente i certificati CA dai pool specificati e li installa nell'archivio attendibile di ogni broker nel cluster.

Console

Puoi attivare mTLS su un nuovo cluster durante la creazione o su un cluster esistente modificandolo.

Su un nuovo cluster

  1. Nella console Google Cloud , vai alla pagina Cluster.

    Vai a Cluster

  2. Seleziona Crea.

    Viene visualizzata la pagina Crea cluster Kafka.

  3. Segui i passaggi descritti in Creare un cluster.
  4. Prima del passaggio finale, individua la sezione Configurazione mTLS facoltativa.
  5. Inserisci il nome completo della risorsa di un pool di CA nel formato projects/PROJECT_ID/LOCATION/LOCATION/caPools/POOL_ID.
  6. Per aggiungerne altri, fai clic su Aggiungi pool di CA. Puoi aggiungere fino a 10 pool di CA.
  7. (Facoltativo) Inserisci eventuali regole di mappatura delle entità.
  8. Fai clic su Crea per creare il cluster con mTLS abilitato.

Su un cluster esistente

  1. Nella console Google Cloud , vai alla pagina Cluster.

    Vai a Cluster

  2. Fai clic sul nome del cluster da aggiornare.
  3. Fai clic su Modifica.
  4. Nella sezione Configurazione mTLS, aggiungi o modifica l'elenco dei pool di CA. Puoi aggiungere fino a 10 pool di CA.
  5. (Facoltativo) Inserisci o modifica le regole di mappatura delle entità.
  6. Fai clic su Salva.

gcloud

Su un nuovo cluster

  1. 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.

  2. Esegui il comando gcloud managed-kafka clusters create con il flag --mtls-ca-pools. In questo esempio, sono configurati due pool di 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

Sostituisci quanto segue:

  • CLUSTER_ID: l'ID o il nome del cluster.

    Per maggiori informazioni su come assegnare un nome a un cluster, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka. Ad esempio, test-mtls-cluster.

  • LOCATION: la posizione del cluster.

    Per saperne di più sulle località supportate, consulta Località di Managed Service per Apache Kafka supportate. Ad esempio, us-central1.

  • SUBNETS: l'elenco delle subnet a cui connettersi. Utilizza le virgole per separare più valori di subnet.

    Il formato della subnet è projects/PROJECT_ID/locations/LOCATION/subnetworks/SUBNET_ID.

    • POOL_ID_1: l'ID del primo pool di CA. Ad esempio, test-mtls-pool1.

  • POOL_ID_2: l'ID del secondo pool di CA. Ad esempio, test-mtls-pool2.

Su un cluster esistente

  1. 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.

  2. Utilizza il comando gcloud managed-kafka clusters update. Questo comando sovrascrive l'intero insieme di pool attualmente configurati. Fornisci l'elenco completo dei pool di CA richiesti. In questo esempio, sono configurati due pool di 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

Sostituisci quanto segue:

Configura la mappatura dei nomi principali

Quando un client esegue l'autenticazione con mTLS, per impostazione predefinita l'entità Kafka viene derivata dal nome distinto (DN) del soggetto del certificato e ha il formato User:CN=...,OU=...,O=...,L=...,ST=...,C=.... Crea regole di mapping per trasformare il nome distinto del soggetto del certificato in un alias pratico più facile da utilizzare negli ACL di Kafka. Per saperne di più sul modulo per il nome distinto del soggetto, consulta Personalizzazione del nome utente SSL nella documentazione di Apache Kafka.

Queste trasformazioni sono definite dalla proprietà del broker Kafka ssl.principal.mapping.rules, che utilizza espressioni regolari per estrarre e riformattare parti dell'oggetto del certificato.

Ad esempio, puoi applicare una regola per trasformare un nome distinto del soggetto completo in un alias nel seguente modo:

  • DN soggetto del certificato: CN=order-processing-app,OU=marketing,O=company,C=US

  • Regola di mappatura: RULE:^.*[Cc][Nn]=([a-zA-Z0-9.-]*).*$/$1/L,DEFAULT

  • Entità Kafka risultante: order-processing-app

Questa regola di esempio estrae il nome comune (CN) dal soggetto del certificato e lo utilizza come nome principale in Kafka.

Per impostare una regola di mapping sul cluster utilizzando Google Cloud CLI, segui questi passaggi. Quando utilizzi la console, puoi impostare le regole di mappatura durante la creazione o l'aggiornamento di un cluster.

  • Per aggiornare il cluster, utilizza il comando gcloud managed-kafka clusters update con il flag --ssl-principal-mapping-rules.

    gcloud managed-kafka clusters update CLUSTER_ID \
    --location=REGION \
    --ssl-principal-mapping-rules='MAPPING_RULE'

    Sostituisci quanto segue:

    • CLUSTER_ID: l'ID del cluster Managed Service per Apache Kafka che stai creando. Ad esempio, test-kafka-cluster.

    • REGION: la Google Cloud regione in cui creare il cluster. Ad esempio, us-central1.

    • MAPPING_RULE*: la regola di mappatura che vuoi applicare. Ad esempio, RULE:^.*[Cc][Nn]=([a-zA-Z0-9.-]*).*$/$1/L,DEFAULT.

Per saperne di più su come scrivere regole di mappatura, consulta Autorizzazione e ACL nella documentazione di Apache Kafka.

Configura gli ACL Kafka per i principal mTLS

Per impostazione predefinita, a qualsiasi client che esegue l'autenticazione con un certificato mTLS valido viene concesso l'accesso completo al cluster. Per applicare il principio del privilegio minimo, devi creare ACL Kafka per definire autorizzazioni specifiche per i tuoi principal mTLS. Il principal per un client mTLS è il nome distinto (DN) del soggetto del certificato (o un alias mappato), con il prefisso User:.

Per creare ACL Kafka, devi disporre del ruolo IAM Editor ACL Kafka gestite (roles/managedkafka.aclEditor).

Supponiamo di avere un'applicazione identificata dal relativo certificato che produce messaggi per orders-topic e utilizza messaggi da analytics-topic. L'entità dell'applicazione, dopo essere stata semplificata con una regola di mappatura, è order-processing-app. Quando crei gli ACL Kafka, devi aggiungere il prefisso User: al principal.

  1. Applica l'ACL WRITE al cluster. Esegui il comando gcloud managed-kafka acls add-entry per concedere l'autorizzazione WRITE su orders-topic.

    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="*"

    Sostituisci quanto segue:

    • CLUSTER_ID: l'ID del cluster Managed Service per Apache Kafka che stai creando. Ad esempio, test-kafka-cluster.

    • REGION: la Google Cloud regione in cui creare il cluster. Ad esempio, us-central1.

  2. Applica l'ACL READ al cluster. Esegui il comando gcloud managed-kafka acls add-entry per concedere l'autorizzazione READ su analytics-topic.

    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="*"

Dopo aver applicato queste ACL, il client order-processing-app dispone solo delle autorizzazioni specifiche che hai concesso. Per saperne di più su come creare ACL, consulta Crea ACL Kafka.

Configura i client Kafka

Dopo aver configurato mTLS sul cluster, devi configurare le applicazioni client per l'autenticazione utilizzando questo metodo. La procedura prevede la creazione di un certificato client e la configurazione delle proprietà del client per utilizzarlo.

  1. Crea e scarica un certificato client sul computer client. Esegui il comando gcloud privateca certificates create per emettere un nuovo certificato da uno dei pool di CA che hai configurato per il tuo cluster.

    Questo comando scarica il certificato client-cert.pem e la relativa chiave privata client-key.pem nell'ambiente locale.

    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

    Sostituisci quanto segue:

    • CERTIFICATE_ID: un nome univoco per l'oggetto certificato. Ad esempio, order-app-cert.

    • PROJECT_ID: l'ID del progetto contenente il pool di CA. Ad esempio, test-project-12345.

    • REGION: la regione in cui si trova il pool di CA. Ad esempio, us-central1.

    • POOL_ID: l'ID del pool di CA da cui emettere il certificato. Ad esempio, test-mtls-pool1.

    • CA_NAME: il nome dell'autorità di certificazione all'interno del pool. Ad esempio, test-sub-ca.

    • --dns-san="client.example.com": il nome alternativo del soggetto DNS. Puoi utilizzare qualsiasi valore pertinente per il tuo cliente.

    • --subject="CN=test-client-app": il DN soggetto. Questo nome viene utilizzato come entità mTLS, a meno che tu non abbia configurato una regola di mappatura dei nomi delle entità.

  2. Visualizza il certificato client, il soggetto del certificato e verifica ssl.principal.mapping.rules. Esegui il comando gcloud privateca certificates describe:

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

    Sostituisci quanto segue:

    • CERTIFICATE_ID: il nome univoco dell'oggetto certificato. Ad esempio, order-app-cert.

    • POOL_ID: l'ID del pool di CA da cui hai emesso il certificato. Ad esempio, test-mtls-pool1.

    • REGION: la regione in cui si trova il pool di CA. Ad esempio, us-central1.

    L'output è simile al seguente:

    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
  ...

  3. Crea un archivio chiavi Java. Combina il certificato e la chiave privata in un file PKCS#12, quindi importalo in un file 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"

    Puoi verificare che la chiave sia stata archiviata correttamente eseguendo questo comando:

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

    L'output è simile al seguente:

    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
...

    Tieni presente che la riga Owner mostra il nome distinto del soggetto del certificato. Per impostazione predefinita, Kafka imposta il principal Kafka su questo formato esatto: CN=...,OU=...,O=...,L=...,ST=...,C=.... Per saperne di più, consulta la sezione Autorizzazione e ACL nella documentazione di Apache Kafka.

  4. Configura le proprietà del client Kafka e l'indirizzo di bootstrap. Nell'applicazione client Kafka, imposta le seguenti proprietà per utilizzare il keystore per una connessione SSL. Inoltre, assicurati di utilizzare l'indirizzo di bootstrap corretto con la porta 9192. Per maggiori informazioni su come configurare un client, vedi Guida rapida: crea un cluster Managed Service per Apache Kafka e connetti un client.

    security.protocol=SSL
ssl.keystore.location=KEYSTORE_LOCATION
ssl.keystore.password=KEYSTORE_PASSWORD
bootstrap.servers=CLUSTER_BOOTSTRAP_ADDRESS

    Sostituisci quanto segue:

    • KEYSTORE_LOCATION: il percorso del file .jks.

    • KEYSTORE_PASSWORD: la password per il keystore.

    • CLUSTER_BOOTSTRAP_ADDRESS: l'indirizzo di bootstrap del tuo cluster. Per trovare l'indirizzo di bootstrap, consulta Visualizzare i dettagli del cluster. Assicurati di aggiungere il numero di porta come 9192.

Proteggere la configurazione del client

L'esempio precedente prevede l'archiviazione locale di chiavi private e password, pertanto non lo consigliamo per gli ambienti di produzione. Per la produzione, gestisci i segreti client in modo sicuro. Sono incluse le seguenti opzioni:

  • Archivia il keystore e la relativa password come secret in Google Cloud Secret Manager e recuperali in fase di runtime nel codice dell'applicazione.

  • Se esegui il deployment dell'applicazione su GKE, utilizza il componente aggiuntivo Secret Manager per montare i secret nel file system dell'applicazione in fase di runtime.

Monitorare mTLS

Puoi monitorare lo stato degli aggiornamenti dei certificati mTLS utilizzando metriche e log in Cloud Monitoring e Cloud Logging.

Per monitorare in modo proattivo lo stato degli aggiornamenti dei certificati mTLS, utilizza la metrica managedkafka.googleapis.com/mtls_truststore_update_count in Monitoring. Questa metrica conteggia i tentativi di aggiornamento dell'archivio attendibile e include un'etichetta STATUS, che può essere SUCCESS o un motivo di errore come CA_POOL_FETCH_ERROR.

Il servizio Managed Service per Apache Kafka tenta di aggiornare l'archivio attendibile una volta all'ora per ogni cluster. Ti consigliamo di creare un avviso che si attiva quando questa metrica segnala un conteggio persistente di errori per più di tre ore, in quanto ciò potrebbe indicare una configurazione errata che richiede un intervento manuale.

Gli aggiornamenti del truststore utilizzano la quota dell'API Certificate Authority Service. È importante comprendere quanto segue:

  • Il processo di aggiornamento chiama il metodo FetchCaCerts, che è soggetto alla quota API requests per minute per region.

  • Questo utilizzo della quota è attribuito al progetto che contiene il pool di CA a cui viene fatto riferimento e non al progetto Managed Service per Apache Kafka.

  • Il limite predefinito è di 400 query al secondo (QPS) per regione. Data la bassa frequenza di una richiesta per cluster all'ora, è improbabile che questi aggiornamenti del truststore ti facciano superare questa quota.

Puoi anche monitorare gli aggiornamenti del truststore visualizzando i log in Logging. Cerca le seguenti voci di log per verificare che gli aggiornamenti siano stati eseguiti correttamente:

  • Managed Service for Apache Kafka updated the mTLS trust store

  • Added root CA certificate to trust store

Passaggi successivi

Apache Kafka® è un marchio registrato di Apache Software Foundation o delle sue affiliate negli Stati Uniti e/o in altri paesi.