Add a customized principal builder

[xuzha]

Add a CommonNamePrincipalBuilder, only merge to 0.10.2.1 branch.

Why we need this:

By default, the SSL user name will be of the form CN=writeuser,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown. We can change that by setting a customized PrincipalBuilder in the server.conf.

This class: CommonNamePrincipalBuilder gets the Common Name from the cert, which is writeuser, as the username.

By using this, we can do the following command to grant the produce permission:

./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=locahost:2181 --add --allow-principal User:{CnName} --producer --topic test.topic

[xuzha]

This working in our testing env

Granted UserA with producer permission before the testing.

UserA, with producer permission, server side log:

2017-06-26 - 22:58:10,416 DEBUG kafka-network-thread-1151118090-ListenerName(SSL)-SSL-48 SslTransportLayer.handshakeFinished - SSL handshake completed successfully with peerHost '205.205.205.123' peerPort 41328 peerPrincipal 'CN=UserA, OU=B, O=C, L=D, ST=E, C=F' cipherSuite 'TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256'
2017-06-26 - 22:58:10,498 DEBUG kafka-request-handler-23 logger.$anonfun$aclMatch$2 - operation = Describe on resource = Topic:topicA from host = 205.205.205.123 is Allow based on acl = User:UserA has Allow permission for operations: Describe from hosts: *
2017-06-26 - 22:58:10,498 DEBUG kafka-request-handler-23 logger.logAuditMessage - Principal = User:UserA is Allowed Operation = Describe from host = 205.205.205.123 on resource = Topic:topicA
2017-06-26 - 22:58:10,499 DEBUG kafka-request-handler-23 logger.$anonfun$aclMatch$2 - operation = Write on resource = Topic:topicA from host = 205.205.205.123 is Allow based on acl = User:UserA has Allow permission for operations: Write from hosts: *
2017-06-26 - 22:58:10,499 DEBUG kafka-request-handler-23 logger.logAuditMessage - Principal = User:UserA is Allowed Operation = Write from host = 205.205.205.123 on resource = Topic:topicA
2017-06-26 - 22:58:10,500 DEBUG kafka-request-handler-23 ReplicaManager.debug - [Replica Manager on Broker 1151118090]: Request key topicA-0 unblocked 0 fetch requests.

UserB, without producer permission, server side log:

2017-06-26 - 22:57:04,719 DEBUG kafka-network-thread-1151118090-ListenerName(SSL)-SSL-39 SslTransportLayer.handshakeFinished - SSL handshake completed successfully with peerHost '205.205.205.123` peerPort 6525 peerPrincipal 'CN=UserB, OU=B, O=C, L=D, ST=E, C=F' cipherSuite 'TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256'
2017-06-26 - 22:57:04,829 DEBUG kafka-request-handler-22 logger.logAuditMessage - Principal = User:UserB is Denied Operation = Describe from host = 205.205.205.123 on resource = Topic:topicA

ANONYMOUS, connecting Kafka on plaintext port 9092

2017-06-26T22:56:11+00:00 i-091f31a0a7c696438 local3.debug 2017-06-26 - 22:56:11,755 DEBUG kafka-request-handler-3 logger.logAuditMessage - Principal = User:ANONYMOUS is Denied Operation = Describe from host =205.205.205.123, on resource = Topic:topicA

[xuzha]

If you guys have a moment: @SonicWang @alexism @parasitew

[alexism]

a small detail: why limit = 6 if the only [0] is used?

[xuzha]

Thanks for reviewing this man.

Basically this function extracts the common name from the distinguished name. For example, the distinguished name is in this format: CN=xu_zhang,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown, the function will extract the xu_zhang from it. Then when we using ACL, we can just put user:xu_zhang instead of user:CN=xu_zhang,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown.

[alexism]

I'm lacking context and it's hard to understand the purpose of this class based on the existing documentation.
but otherwise code LGTM.

[xuzha]

@alexism thanks for reviewing this man ;-). I will add some internal documentation about this.