I would like to run a Kafka cluster with 2 listeners. One SSL and one PLAINTEXT listener. The SSL listener should be secured by authentication/authorization via SSL and ACL. Only known users should be allowed to access it.

On the PLAINTEXT port any user or anonymous user should have full access.

If I set the property

allow.everyone.if.no.acl.found=false

however, anonymous users will no longer be able to access the PLAINTEXT port. Is there any way to configure the system as described above?

CodePudding user response：

I can think of 2 ways to achieve that. Depending on your exact configuration you may prefer one or the other.

1. ACLs only

   Each connection to Kafka is associated with a Principal. Principal can be customized using a custom PrincipalBuilder but the default implementation should build different principals for users connecting over SSL and PLAINTEXT. Via PLAINTEXT, all users should be ANONYMOUS and via SSL they should get names based on these SSL mapping rules.

   So you should be able to add an ACL granted everything to Principal User:ANONYMOUS and set specific ACLs for the SSL users.

2. Custom Authorizer

   This is slightly more involved but it should grant you more freedom. You can implement a simple custom Authorizer that makes decisions based on the listener.

   If you already have a custom Authorizer, you can retrieve the listener from AuthorizableRequestContext.listenerName() and return allowed if it's the PLAINTEXT listener.

   If you are using the default AclAuthorizer that comes with Kafka, you can extend it to check the listener name:

   public class AllowPlaintextAuthSslAuthorizer extends AclAuthorizer {
   
       @Override
       public List<AuthorizationResult> authorize(AuthorizableRequestContext requestContext, List<Action> actions) {
           if ("PLAINTEXT".equals(requestContext.listenerName())) {
               return actions.stream()
                   .map((Action a) -> AuthorizationResult.ALLOWED)
                   .collect(Collectors.toList());
           } else {
               return super.authorize(requestContext, actions);
           }
       }
   }
   

   Then in your server.properties file, set authorizer.class.name to use your custom class.

   Note that AclAuthorizer is not part of the public API so it may change in the future.