Configurar a autenticação mTLS

É possível configurar o cluster do Serviço gerenciado para Apache Kafka para autenticar clientes do Kafka usando TLS mútuo (mTLS). Esse método usa certificados de cliente do Certificate Authority Service (CA Service) como base para autenticação. Essa opção oferece uma alternativa ao mecanismo SASL padrão que usa principais do Identity and Access Management (IAM).

Ao usar o mTLS, a autorização precisa ser configurada com as ACLs do Kafka. Para conceitos básicos, consulte os seguintes documentos:

Antes de começar

Antes de configurar a autenticação mTLS, faça o seguinte:

  • Confirme a qualificação do cluster. Verifique se você tem um cluster do Serviço gerenciado para Apache Kafka criado após 24 de junho de 2025. Somente esses clusters são compatíveis com a autenticação mTLS. Para verificar a data de criação do cluster, use o comando gcloud managed-kafka clusters describe ou consulte a página de detalhes do cluster no console.

  • Configure o serviço de CA. Configure o serviço de CA e os pools de CA que você pretende usar para emitir certificados de cliente. É necessário criar certificados raiz e subordinados nos pools de CA.

    1. Crie um pool de CAs. Anote o ID do pool de ACs.

      Para mais informações sobre como criar um pool de CAs, consulte Criar um pool de CAs.

    2. Crie e ative uma CA raiz para o pool.

      Para mais informações sobre como ativar uma AC raiz para o pool, consulte Criar uma AC raiz.

    3. Crie e ative uma ou mais CAs subordinadas. Recomendamos usar uma CA subordinada para emitir certificados de cliente em vez de usar uma CA raiz diretamente.

      Para mais informações sobre como criar uma AC subordinada, consulte Criar uma autoridade certificadora subordinada.

Papéis e permissões necessárias

Para configurar o mTLS, verifique se você (o usuário) e o agente do Serviço gerenciado para Apache Kafka têm as permissões necessárias do IAM. Isso se aplica à criação de um cluster ou à atualização de um cluster existente para configurar o mTLS.

Permissões do usuário

Para criar ou configurar um cluster do serviço gerenciado para Apache Kafka para mTLS, você precisa de permissões para criar ou atualizar o recurso de cluster. Para isso, peça ao administrador para conceder a você o papel de Editor de cluster do Kafka gerenciado (roles/managedkafka.clusterEditor) no projeto que contém o cluster.

Esse papel predefinido contém as permissões managedkafka.clusters.create e managedkafka.clusters.update. Com essas permissões, é possível criar um cluster ou modificar um já existente para adicionar a configuração do pool do serviço de AC (CA) necessária para o mTLS.

Não é necessário ter permissões separadas nos recursos do CA Service para configurar o mTLS no cluster do Kafka, desde que você tenha o caminho completo do recurso do pool de CAs. No entanto, para ver, criar ou gerenciar pools de ACs no console Google Cloud , você precisaria de outros papéis específicos do serviço de AC, como Administrador do serviço de AC (roles/privateca.admin) ou Operador do serviço de AC (roles/privateca.operator).

Permissões do agente de serviço

Para que a integração do mTLS funcione, o agente de serviço do Managed Service para Apache Kafka precisa de permissão para acessar o pool de ACs especificado. O agente de serviço é uma conta serviço gerenciado pelo Google para seu projeto.

  • Se o cluster do Serviço gerenciado para Apache Kafka e o pool de ACs estiverem no mesmo projeto, o agente de serviço terá as permissões necessárias por padrão. O papel managedkafka.serviceAgent, concedido automaticamente ao agente de serviço no projeto, inclui a permissão privateca.caPools.get necessária.

  • Se o pool de CA estiver em um projeto diferente do cluster do Serviço gerenciado para Apache Kafka, conceda manualmente a permissão ao agente de serviço para acessá-lo. Conceda ao agente de serviço o papel Leitor do pool de Private CA (roles/privateca.poolReader) no projeto que contém o pool de ACs.

Resumo das permissões necessárias

Para conferir as permissões necessárias, expanda a seção a seguir.

Essas permissões também podem ser concedidas com papéis personalizados ou outros papéis predefinidos.

Conceder ao agente de serviço acesso aos pools de ACs

Se o pool de ACs do serviço de AC e o cluster do Serviço gerenciado para Apache Kafka estiverem em projetos Google Cloud diferentes, conceda ao agente de serviço do cluster permissão para acessar o pool de ACs. O agente de serviço do Serviço gerenciado para Apache Kafka é chamado de service-CLUSTER_PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com.

Conceda o papel Leitor do pool de CA (roles/privateca.poolReader) ao agente de serviço do Serviço Gerenciado para Apache Kafka no nível do pool individual (recomendado) que contém suas CAs ou em todos os pools do projeto. Esse papel fornece a permissão privateca.caPools.get necessária.

Pool de CA individual

Conceder permissões a um único pool de CA é a abordagem recomendada, já que segue o princípio de privilégio mínimo.

Execute o 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'

Substitua:

  • CA_POOL_ID: o ID do pool de ACs a que você está concedendo acesso. Por exemplo, test-mtls-pool1.

  • CA_POOL_LOCATION: a Google Cloud região em que o pool de CAs está localizado. Por exemplo, us-central1.

  • CLUSTER_PROJECT_NUMBER: o número do projeto que contém seu cluster do Serviço Gerenciado para Apache Kafka. Por exemplo, 12341234123.

Todos os pools de CA

Outra opção é conceder ao agente de serviço permissão para acessar todos os pools de CA em um projeto definindo a política no nível do projeto.

Execute o 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'

Substitua:

  • CA_PROJECT_ID: o ID do projeto que contém os pools de ACs a que você está concedendo acesso. Por exemplo, test-cas-project.

  • CLUSTER_PROJECT_NUMBER: o número do projeto que contém seu cluster do Serviço Gerenciado para Apache Kafka. Por exemplo, 12341234123.

Ativar o mTLS em um cluster

Para ativar o mTLS, forneça ao cluster os nomes de recursos de um ou mais pools de AC do serviço de CA para usar na autenticação do cliente. É possível fazer isso ao criar um cluster ou atualizar um que foi criado após 24 de junho de 2025.

Depois que você fornece os identificadores do pool de AC, o serviço baixa automaticamente os certificados de AC dos pools especificados e os instala no truststore de cada broker no cluster.

Console

É possível ativar o mTLS em um novo cluster durante a criação ou em um cluster atual editando-o.

Em um novo cluster

  1. No Google Cloud console, acesse a página Clusters.

    Acessar Clusters

  2. Selecione Criar.

    A página Criar cluster do Kafka é aberta.

  3. Siga as etapas em Criar um cluster.
  4. Antes da etapa final, localize a seção Configuração opcional de mTLS.
  5. Insira o nome completo do recurso de um pool de ACs no formato projects/PROJECT_ID/LOCATION/LOCATION/caPools/POOL_ID.
  6. Para adicionar mais, clique em Adicionar pool de ACs. É possível adicionar até 10 pools de CA.
  7. (Opcional) Insira as regras de mapeamento de principais.
  8. Clique em Criar para criar o cluster com o mTLS ativado.

Em um cluster atual

  1. No Google Cloud console, acesse a página Clusters.

    Acessar Clusters

  2. Clique no nome do cluster que você quer atualizar.
  3. Clique em Editar.
  4. Na seção Configuração de mTLS, adicione ou modifique a lista de pools de CA. É possível adicionar até 10 pools de CA.
  5. (Opcional) Insira ou edite as regras de mapeamento de principais.
  6. Clique em Salvar.

gcloud

Em um novo 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. Execute o comando gcloud managed-kafka clusters create com a flag --mtls-ca-pools. Neste exemplo, dois pools de CA estão configurados.

    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

Substitua:

  • CLUSTER_ID: o ID ou nome do cluster.

    Para mais informações sobre como nomear um cluster, consulte as diretrizes de nomeação de recursos do Serviço gerenciado para Apache Kafka. Por exemplo: test-mtls-cluster.

  • LOCATION: o local do cluster.

    Para mais informações sobre os locais com suporte, consulte Locais com suporte do Serviço gerenciado para Apache Kafka. Por exemplo: us-central1.

  • SUBNETS: a lista de sub-redes a serem conectadas. Use vírgulas para separar vários valores de sub-rede.

    O formato da sub-rede é projects/PROJECT_ID/locations/LOCATION/subnetworks/SUBNET_ID.

    • POOL_ID_1: o ID do primeiro pool de ACs. Por exemplo: test-mtls-pool1.

  • POOL_ID_2: o ID do segundo pool de ACs. Por exemplo: test-mtls-pool2.

Em um cluster atual

  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. Use o comando gcloud managed-kafka clusters update. Esse comando substitui todo o conjunto de pools configurados no momento. Forneça a lista completa dos pools de ACs necessários. Neste exemplo, dois pools de CA estão configurados.

    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

Substitua:

Configurar o mapeamento de nomes principais

Quando um cliente se autentica com mTLS, por padrão, o principal do Kafka é derivado do nome distinto (DN) do assunto do certificado e tem o formato User:CN=...,OU=...,O=...,L=...,ST=...,C=.... Crie regras de mapeamento para transformar o DN do assunto do certificado em um alias conveniente que seja mais fácil de usar nas ACLs do Kafka. Para mais informações sobre o formulário do DN do assunto, consulte Personalizar o nome de usuário SSL na documentação do Apache Kafka.

Essas transformações são definidas pela propriedade do broker do Kafka ssl.principal.mapping.rules, que usa expressões regulares para extrair e reformatar partes do assunto do certificado.

Por exemplo, é possível aplicar uma regra para transformar um DN de assunto completo em um alias da seguinte maneira:

  • DN do assunto do certificado: CN=order-processing-app,OU=marketing,O=company,C=US

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

  • Principal do Kafka resultante:order-processing-app

Esta regra de exemplo extrai o nome comum (CN) do assunto do certificado e o usa como o nome principal no Kafka.

Para definir uma regra de mapeamento no seu cluster usando a Google Cloud CLI, siga estas etapas. Ao usar o console, é possível definir as regras de mapeamento ao criar ou atualizar um cluster.

  • Para atualizar o cluster, use o comando gcloud managed-kafka clusters update com a flag --ssl-principal-mapping-rules.

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

    Substitua:

    • CLUSTER_ID: o ID do cluster do Serviço Gerenciado para Apache Kafka que você está criando. Por exemplo, test-kafka-cluster.

    • REGION: a Google Cloud região em que o cluster será criado. Por exemplo: us-central1.

    • MAPPING_RULE*: a regra de mapeamento que você quer aplicar. Por exemplo: RULE:^.*[Cc][Nn]=([a-zA-Z0-9.-]*).*$/$1/L,DEFAULT.

Para mais informações sobre como escrever regras de mapeamento, consulte Autorização e ACLs na documentação do Apache Kafka.

Configurar ACLs do Kafka para principais do mTLS

Por padrão, qualquer cliente que se autenticar com um certificado mTLS válido recebe acesso total ao cluster. Para aplicar o princípio de privilégio mínimo, crie ACLs do Kafka para definir permissões específicas para seus principais do mTLS. O principal de um cliente mTLS é o DN do assunto do certificado (ou um alias mapeado), com o prefixo User:.

Para criar ACLs do Kafka, você precisa do papel do IAM Editor de ACL do Kafka gerenciado (roles/managedkafka.aclEditor).

Suponha que você tenha um aplicativo identificado pelo certificado que produz mensagens para orders-topic e consome mensagens de analytics-topic. O principal do aplicativo, depois de simplificado com uma regra de mapeamento, é order-processing-app. Ao criar ACLs do Kafka, é necessário prefixar o principal com User:.

  1. Aplique a ACL WRITE ao cluster. Execute o comando gcloud managed-kafka acls add-entry para conceder a permissão WRITE no 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="*"

    Substitua:

    • CLUSTER_ID: o ID do cluster do Serviço Gerenciado para Apache Kafka que você está criando. Por exemplo, test-kafka-cluster.

    • REGION: a Google Cloud região em que o cluster será criado. Por exemplo: us-central1.

  2. Aplique a ACL READ ao cluster. Execute o comando gcloud managed-kafka acls add-entry para conceder a permissão READ no 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="*"

Depois de aplicar essas ACLs, o cliente order-processing-app terá apenas as permissões específicas que você concedeu. Para mais informações sobre como criar ACLs, consulte Criar ACLs do Kafka.

Configurar clientes do Kafka

Depois de configurar o mTLS no cluster, configure os aplicativos clientes para autenticar usando esse método. O processo envolve criar um certificado do cliente e configurar as propriedades dele para usá-lo.

  1. Crie e faça o download de um certificado do cliente na máquina dele. Execute o comando gcloud privateca certificates create para emitir um novo certificado de um dos pools de CA que você configurou para o cluster.

    Esse comando baixa o certificado client-cert.pem e a chave privada client-key.pem para seu ambiente local.

    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

    Substitua:

    • CERTIFICATE_ID: um nome exclusivo para o objeto de certificado. Por exemplo: order-app-cert.

    • PROJECT_ID: o ID do projeto que contém o pool de ACs. Por exemplo: test-project-12345.

    • REGION: a região em que o pool de CAs está localizado. Por exemplo: us-central1.

    • POOL_ID: o ID do pool de ACs de que o certificado será emitido. Por exemplo: test-mtls-pool1.

    • CA_NAME: o nome da autoridade certificadora no pool. Por exemplo: test-sub-ca.

    • --dns-san="client.example.com": o nome alternativo do assunto do DNS. Você pode usar qualquer valor relevante para seu cliente.

    • --subject="CN=test-client-app": o DN do titular. Esse nome é usado como o principal do mTLS, a menos que você tenha configurado uma regra de mapeamento de nome principal.

  2. Confira o certificado do cliente, o assunto do certificado e verifique ssl.principal.mapping.rules. Execute o comando gcloud privateca certificates describe:

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

    Substitua:

    • CERTIFICATE_ID: o nome exclusivo do objeto de certificado. Por exemplo: order-app-cert.

    • POOL_ID: o ID do pool de ACs de que você emitiu o certificado. Por exemplo: test-mtls-pool1.

    • REGION: a região em que o pool de CAs está localizado. Por exemplo: us-central1.

    O resultado será o seguinte:

    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. Crie um keystore Java. Combine o certificado e a chave privada em um arquivo PKCS#12 e importe para um arquivo 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"

    Para verificar se a chave foi armazenada, execute o seguinte comando:

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

    O resultado será o seguinte:

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

    A linha Owner mostra o DN do assunto do certificado. Por padrão, o Kafka define o principal do Kafka nesse formato exato: CN=...,OU=...,O=...,L=...,ST=...,C=.... Para mais informações, consulte Autorização e ACLs na documentação do Apache Kafka.

  4. Configure as propriedades do cliente Kafka e o endereço de inicialização. No aplicativo cliente do Kafka, defina as seguintes propriedades para usar o keystore em uma conexão SSL. Além disso, use o endereço de bootstrap correto com a porta 9192. Para mais informações sobre como configurar um cliente, consulte Início rápido: criar um cluster do Serviço gerenciado para Apache Kafka e conectar um cliente.

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

    Substitua:

    • KEYSTORE_LOCATION: o caminho para o arquivo .jks.

    • KEYSTORE_PASSWORD: a senha do keystore.

    • CLUSTER_BOOTSTRAP_ADDRESS: o endereço de bootstrap do cluster. Para encontrar o endereço de bootstrap, consulte Ver detalhes do cluster. Adicione o número da porta como 9192.

Proteger a configuração do cliente

O exemplo anterior envolve o armazenamento local de chaves privadas e senhas, por isso não o recomendamos para ambientes de produção. Para produção, processe os secrets do cliente com segurança. As opções incluem:

  • Armazene o keystore e a senha dele como secrets no Secret Manager do Google Cloud e recupere-os durante a execução no código do aplicativo.

  • Se você implantar o aplicativo no GKE, use o complemento do Secret Manager para ativar os secrets no sistema de arquivos do aplicativo durante a execução.

Monitorar mTLS

É possível monitorar a integridade das atualizações de certificado mTLS usando métricas e registros no Cloud Monitoring e no Cloud Logging.

Para monitorar proativamente a integridade das atualizações de certificado mTLS, use a métrica managedkafka.googleapis.com/mtls_truststore_update_count no Monitoring. Essa métrica conta as tentativas de atualização do truststore e inclui um rótulo STATUS, que pode ser SUCCESS ou um motivo de falha como CA_POOL_FETCH_ERROR.

O serviço gerenciado para Apache Kafka tenta atualizar o truststore uma vez por hora para cada cluster. Recomendamos que você crie um alerta que seja acionado quando essa métrica informar uma contagem persistente de erros por mais de três horas, já que isso pode indicar uma configuração incorreta que exige intervenção manual.

As atualizações do Truststore consomem a cota da API Certificate Authority Service. É importante entender o seguinte:

  • O processo de atualização chama o método FetchCaCerts, que está sujeito à cota API requests per minute per region.

  • Esse uso de cota é atribuído ao projeto que contém o pool de CA referenciado, e não ao projeto do Serviço gerenciado para Apache Kafka.

  • O limite padrão é de 400 consultas por segundo (QPS) por região. Devido à baixa frequência de uma solicitação por cluster por hora, é improvável que essas atualizações de truststore causem o estouro da cota.

Também é possível acompanhar as atualizações do truststore visualizando os registros no Logging. Procure as seguintes entradas de registro para confirmar se as atualizações foram bem-sucedidas:

  • Managed Service for Apache Kafka updated the mTLS trust store

  • Added root CA certificate to trust store

A seguir

Apache Kafka® é uma marca registrada da The Apache Software Foundation ou afiliadas nos Estados Unidos e/ou em outros países.