kafka mTLS authentication issue with External Certificate

[anandgupta0404]

Hello @scholzj

we are getting the authentication issue with when we are using the user with type:tls-external.
as our external CA which sign the certificate for user, also include C=<some_value> OU=<some_value> in subject name. (and this is mandate in our organization )

as i understood from the post here that in subject name only the CN=<user_name> should be there.
6515

so as you suggested that we should use the kafkaAdmin API, but i am not sure how to use it to manage user separately (not from operator).

can you please suggest some other solutions?

Please let me know if you wanted to see our code.

[scholzj]

Yes, the subject of the certificates should be CN=<NameOfTheKafkaUser>. The only other options are to disable the USer OPerator and manage users yourself using the Kafka Admin API or Kafka CLI tools (the docs are on the Kafka website). Or create and use a custom principal builder that will remove the other parts of your certificate subject.

[anandgupta0404]

Thanks for the quick response @scholzj

There are 2 things which wanted to ask.

1. • Admin-API There is documentation for the admin API but here I dont not see API or method for creating user they say in documentation that its just managing the ACLs using API.
2. • create and use a custom principal builder that will remove the other parts of your certificate subject.
     I am not sure how to do that - can you share any documentation for the same ?
     I went through using this document but did not understand 6460
     I do not have experience in java :-)

[scholzj]

1. • Admin-API There is documentation for the admin API but here I dont not see API or method for creating user they say in documentation that its just managing the ACLs using API.

With mTLS, you never create a user. The user is basically the certificate. So you just need to set the ACLs or the quotas for the right principal. That is important to understand. Because with mTLS using some external CA as you try to do, any certificate issues by given CA is a valid user and can authenticate. That might be fine or might not be fine - that depends on what CA you use etc.

2. • create and use a custom principal builder that will remove the other parts of your certificate subject.
     I am not sure how to do that - can you share any documentation for the same ?
     I went through using this document but did not understand 6460
     I do not have experience in java :-)

I do not have any docs. All I can give you is this simple demo: https://github.com/scholzj/custom-strimzi-principal-builder (I would expect it to still mostly work. But keep in mind I tested it when I wrote it 10 months ago.) In general, this is of course advanced topic, so it is not completely simple.

[anandgupta0404]

Hello @scholzj

Thanks for you quick response and Sorry for the delay response from my side (there was vacations).

We tried the method of using principal builder class and build the custom image of kafka. Also, we have given this new image and principal builder class in our kafka cluster configuration (snip of code attached below)

but we are getting below error. (Attached the screenshot).

1. Also below is code for public builder class which you have suggested to use.

package cz.scholz.customprincipalbuilder;

import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.security.auth.AuthenticationContext;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder;
import org.apache.kafka.common.security.auth.KafkaPrincipalSerde;
import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder;
import org.apache.kafka.common.security.ssl.SslPrincipalMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;

/**
 * Custom Principal Builder class that can be used with Strimzi-based Kafka clusters. It keeps the default mapping rules
 * for mTLS users for the internal listeners used by Strimzi components. That is important to keep Strimzi working. But
 * it allows to customize the mapping rules for the other listeners using the {@code CUSTOM_SSL_PRINCIPAL_MAPPING_RULES}
 * constant.
 */
public class CustomPrincipalBuilder implements KafkaPrincipalBuilder, KafkaPrincipalSerde {
    private static final Logger LOGGER = LoggerFactory.getLogger(CustomPrincipalBuilder.class);
    private static final Set<String> INTERNAL_LISTENERS = Set.of("CONTROLPLANE-9090", "REPLICATION-9091");

    // Customize the mapping rule to match you you want to do. This example matches the regular Strimzi users and givens
    // them the user principal <Username>@my-cluster. But you can customize it to for example remove some parts of the
    // distinguished name etc.
    private static final String CUSTOM_SSL_PRINCIPAL_MAPPING_RULES = "RULE:^[C][N]=([a-zA-Z0-9.-]*),C=IN,O=MYORG IN$/U";

    private final DefaultKafkaPrincipalBuilder internalPrincipalBuilder;
    private final DefaultKafkaPrincipalBuilder userDefinedPrincipalBuilder;

    /**
     * Constructs the Custom Principal Builder
     */
    public CustomPrincipalBuilder() {
        LOGGER.info("Creating new CustomPrincipalBuilder instance with default mapping");
        internalPrincipalBuilder = new DefaultKafkaPrincipalBuilder(null, SslPrincipalMapper.fromRules(BrokerSecurityConfigs.DEFAULT_SSL_PRINCIPAL_MAPPING_RULES));

        LOGGER.info("Configuring CustomPrincipalBuilder with custom mapping rules {}", CUSTOM_SSL_PRINCIPAL_MAPPING_RULES);
        userDefinedPrincipalBuilder = new DefaultKafkaPrincipalBuilder(null, SslPrincipalMapper.fromRules(CUSTOM_SSL_PRINCIPAL_MAPPING_RULES));
    }

    @Override
    public KafkaPrincipal build(AuthenticationContext context) {
        if (INTERNAL_LISTENERS.contains(context.listenerName()))    {
            return internalPrincipalBuilder.build(context);
        } else {
            if (userDefinedPrincipalBuilder != null)    {
                return userDefinedPrincipalBuilder.build(context);
            } else {
                throw new KafkaException("Custom Principal Builder was not initialized yet!");
            }
        }
    }

    @Override
    public byte[] serialize(KafkaPrincipal principal) throws SerializationException {
        // Serialization does not depend on the mapping rules, so we can just use the serialization from the internal one
        return internalPrincipalBuilder.serialize(principal);
    }

    @Override
    public KafkaPrincipal deserialize(byte[] bytes) throws SerializationException {
        // Deserialization does not depend on the mapping rules, so we can just use the deserialization from the internal one
        return internalPrincipalBuilder.deserialize(bytes);
    }
}```

2. Below is our kafka configuration with new image and principal builder class.

```kafka:
  authorization:
    type: simple
  replicas: 1
  resources:
    limits:
      cpu: 500m
      memory: 1Gi
    requests:
      cpu: 250m
      memory: 512Mi
  version: "3.6.0"
  config:
    default.replication.factor: 1
    inter.broker.protocol.version: '3.6'
    log.retention.hours: 168
    min.insync.replicas: 1
    offsets.topic.replication.factor: 1
    transaction.state.log.min.isr: 1
    transaction.state.log.replication.factor: 1
    inter.broker.listener.name: internal
    principal.builder.class: cz.scholz.customprincipalbuilder.CustomPrincipalBuilder
  
  listeners:
    - name: client
      port: 9094
      tls: true
      type: internal
      authentication:
        type: tls```


Can you please help me with this error, I am assuming this error is caused by the regulaer expression which we have given.

but our reguler expression works fine we already tried tested it here -> [https://regex101.com/](url)

[scholzj]

I do not know anything about these rules and how the format / validate. Also, most of what you shared is unformatted and therefore unreadable.

[anandgupta0404]

Hello @scholzj
I do not know anything about these rules and how the format / validate. --> Okay
is there any reference document which i can refer to fix that issue of "invalid rule"

I have changed the formatting of principal builder class code and kafka cluster deployment code.

[scholzj]

I guess Kafka documentation?

[anandgupta0404]

Hello @scholzj

we tried to use the same regex as you have used in principal builder class.
but its giving below INFO continuously with SSL error. mentioned in log.

2024-10-22 10:27:45,200 INFO Creating new CustomPrincipalBuilder instance with default mapping (cz.scholz.customprincipalbuilder.CustomPrincipalBuilder) [data-plane-kafka-network-thread-0-ListenerName(CLIENT-9094)-SSL-4]

2024-10-22 10:27:45,200 INFO Configuring CustomPrincipalBuilder with custom mapping rules RULE:^CN=(.*?)$/$1@my-cluster/L (cz.scholz.customprincipalbuilder.CustomPrincipalBuilder) [data-plane-kafka-network-thread-0-ListenerName(CLIENT-9094)-SSL-4]

2024-10-22 10:27:45,204 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /100.75.0.180 (channelId=100.73.3.134:9094-100.75.0.180:52916-622) (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(CLIENT-9094)-SSL-4]
2

[scholzj]

Once again, I'm not an expert on this. But my take would be that this has nothing to do with the principal builder. You are just failing authentication.

[ravneet28]

Hi @scholzj ,
I work with @anandgupta0404 and found that the authentication and regex is properly configured as the user we created is able to authorize when we add it as a super user in the cluster config as below, but when we try to use kafkauser ACLs, we get the Authorization error status 409.
We would like to use kafkauser ACLs for different applications for more granular control.

This way it works as the super user.

kafka:
  authorization:
    type: simple
    superUsers:
      - someuservalue.com

Kafkauser CRD, this way it fails to authorize(Status 409):

    - name: someuservalue.com
      authentication:
        type: tls-external
      authorization:
        type: simple
        acls:
          - resource:
              type: topic
              name: '*'
              patternType: literal
            operations:
              - All
            host: '*'

[scholzj]

I have no idea what gives you 409. I do not think the KAfka authorizer gives any HTTP code based errors (but maybe I just neevr noticed it). In any case, the KafkaUSer CRD you shared doe sno tlook like a KafkaUser CRD. You have to share the exact KafkaUser CRD (ideally including its status) as it exists in your Kubernetes cluster.

[ravneet28]

Thanks for the quick response. Below is the CRD, along with the broker and client applications logs.

1. For the status 409 and authorization failed, I forgot to mention that this status messages show up in the client applications i.e. AKHQ and a sample producer application which we built on java.

Error Logs

Java application

AKHQ log

2024-11-06 11:05:39,939 WARN  -thread-78 org.akhq.log.access        [Date: 2024-11-06T11:05:39.8922Z] [Duration: 46 ms] [Url: POST /api/kafka-cluster/topic] [Status: 409] [Ip: /100.72.6.2] [User: -]

Broker logs

2024-11-06 11:15:12,053 INFO Principal = User:someuservalue.com is Denied Operation = Create from host = 100.75.8.118 on resource = Topic:LITERAL:akhqtesttopic for request = CreateTopics with resourceRefCount = 1 (kafka.authorizer.logger) [data-plane-kafka-request-handler-0]

2. kafkauser CRD,

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  annotations:
    kubectl.kubernetes.io/last-applied-configuration: |
      {"apiVersion":"kafka.strimzi.io/v1beta2","kind":"KafkaUser","metadata":{"annotations":{},"labels":{"argocd.argoproj.io/instance":"kafka-cluster","strimzi.io/cluster":"kafka-cluster"},"name":"someuservalue.com","namespace":"testing"},"spec":{"authentication":{"type":"tls-external"},"authorization":{"acls":[{"host":"*","operations":["All"],"resource":{"type":"cluster"}},{"host":"*","operations":["Read","Write","Create","Delete","Alter","Describe","IdempotentWrite","AlterConfigs","ClusterAction"],"resource":{"name":"*","patternType":"literal","type":"topic"}},{"host":"*","operations":["All"],"resource":{"name":"*","patternType":"literal","type":"group"}}],"type":"simple"}}}
  resourceVersion: '3349805284'
  name: someuservalue.com
  uid: 927e7696-861b-4d1b-b881-370c9ea76fac
  creationTimestamp: '2024-11-04T13:45:17Z'
  generation: 3
  managedFields:
    - apiVersion: kafka.strimzi.io/v1beta2
      fieldsType: FieldsV1
      fieldsV1:
        'f:metadata':
          'f:annotations':
            .: {}
            'f:kubectl.kubernetes.io/last-applied-configuration': {}
          'f:labels':
            .: {}
            'f:argocd.argoproj.io/instance': {}
            'f:strimzi.io/cluster': {}
        'f:spec':
          .: {}
          'f:authentication':
            .: {}
            'f:type': {}
          'f:authorization':
            .: {}
            'f:acls': {}
            'f:type': {}
      manager: argocd-controller
      operation: Update
      time: '2024-11-06T07:06:35Z'
    - apiVersion: kafka.strimzi.io/v1beta2
      fieldsType: FieldsV1
      fieldsV1:
        'f:status':
          .: {}
          'f:conditions': {}
          'f:observedGeneration': {}
          'f:username': {}
      manager: strimzi-user-operator
      operation: Update
      subresource: status
      time: '2024-11-06T07:06:35Z'
  namespace: testing
  labels:
    argocd.argoproj.io/instance: kafka-cluster
    strimzi.io/cluster: kafka-cluster
spec:
  authentication:
    type: tls-external
  authorization:
    acls:
      - host: '*'
        operations:
          - All
        resource:
          type: cluster
      - host: '*'
        operations:
          - Read
          - Write
          - Create
          - Delete
          - Alter
          - Describe
          - IdempotentWrite
          - AlterConfigs
          - ClusterAction
        resource:
          name: '*'
          patternType: literal
          type: topic
      - host: '*'
        operations:
          - All
        resource:
          name: '*'
          patternType: literal
          type: group
    type: simple
status:
  conditions:
    - lastTransitionTime: '2024-11-06T07:06:35.588464058Z'
      status: 'True'
      type: Ready
  observedGeneration: 3
  username: CN=someuservalue.com

[scholzj]

Well, all you need to do is to properly read the thinsg you are sharing ...

* The broker log talks about user someuservalue.com

• The KafkaUser talks about user CN=someuservalue.com

So that is a mismatch. If you want to use the User OPerator to manage qoutas or ACLs for your users, you need to use the same principals ... i.e. for tls-external authentication it has to be CN=<KafkaUserResourceName>.

[anandgupta0404]

Hello @scholzj

Thank you so much for the help, principal builder class is working fine now with ACLs format suggested by you.
Closing this discussion.