Securing an Apache Kafka broker - part II

In the previous post, we looked at how to configure an Apache Kafka broker to require SSL client authentication. In this post we will add authorization to the example, making sure that only authorized producers can send messages to the broker. In addition, we will show how to enforce authorization rules per-topic for consumers.

1) Configure authorization in the broker

Configure Apache Kafka as per the previous tutorial. To enforce some custom authorization rules in Kafka, we will need to implement the Kafka Authorizer interface. This interface contains an "authorize" method, which supplies a Session Object, where you can obtain the current principal, as well as the Operation and Resource upon which to enforce an authorization decision.

In terms of the example detailed in the previous post, we created broker, service (producer) and client (consumer) principals. We want to enforce authorization decisions as follows:

• Let the broker principal do anything
• Let the producer principal read/write on all topics
• Let the consumer principal read/describe only on topics starting with "test".

There is a sample Authorizer implementation available in some Kafka unit test I wrote in github that can be used in this example - CustomAuthorizer:

	@Override
	public boolean authorize(Session arg0, Operation arg1, Resource arg2) {
	if (arg0.principal() == null) {
	return false;
	}
	String principal = arg0.principal().getName();
	if (principal.startsWith("CN=Client")
	&& ("Read".equals(arg1.name()) || "Describe".equals(arg1.name()))
	&& arg2.name().startsWith("test")) {
	return true;
	} else if (principal.startsWith("CN=Service") && ("Read".equals(arg1.name()) || "Describe".equals(arg1.name())
	|| "Write".equals(arg1.name()) || "Create".equals(arg1.name()))) {
	return true;
	} else if (principal.startsWith("CN=Broker")) {
	return true;
	}
	return false;
	}

view raw gistfile1.txt hosted with ❤ by GitHub

Next we need to package up the CustomAuthorizer in a jar so that it can be used in the broker. You can do this by checking out the testcases github repo, and invoking "mvn clean package jar:test-jar -DskipTests" in the "apache/bigdata/kafka" directory. Now copy the resulting test jar in "target" to the "libs" directory in your Kafka installation. Finally, edit the "config/server.properties" file and add the following configuration item:

• authorizer.class.name=org.apache.coheigea.bigdata.kafka.CustomAuthorizer

2) Test authorization

Now lets test the authorization logic. Restart the broker and the producer:

• bin/kafka-server-start.sh config/server.properties
• bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test --producer.config config/producer.properties

Send a few messages to check that the producer is authorized correctly. Now start the consumer:

• bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --consumer.config config/consumer.properties --new-consumer

If everything is configured correctly then it should work as in the first tutorial. Now we will create a new topic called "messages":

• bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic messages

Restart the producer to send messages to "messages" instead of "test". This should work correctly. Now try to consume from "messages" instead of "test". This should result in an authorization failure, as the "client" principal can only consume from the "test" topic according to the authorization rules.