diff --git a/clients/src/main/java/org/apache/kafka/common/network/PlaintextTransportLayer.java b/clients/src/main/java/org/apache/kafka/common/network/PlaintextTransportLayer.java index a3567af96e4f9..35d41685dd178 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/PlaintextTransportLayer.java +++ b/clients/src/main/java/org/apache/kafka/common/network/PlaintextTransportLayer.java @@ -37,7 +37,7 @@ public class PlaintextTransportLayer implements TransportLayer { private static final Logger log = LoggerFactory.getLogger(PlaintextTransportLayer.class); private final SelectionKey key; private final SocketChannel socketChannel; - private final Principal principal = new KafkaPrincipal("ANONYMOUS"); + private final Principal principal = KafkaPrincipal.ANONYMOUS; public PlaintextTransportLayer(SelectionKey key) throws IOException { this.key = key; diff --git a/clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java b/clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java index 8ba1b01201df8..f1cd6074de576 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java +++ b/clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java @@ -33,6 +33,7 @@ import javax.net.ssl.SSLSession; import javax.net.ssl.SSLPeerUnverifiedException; +import org.apache.kafka.common.security.auth.KafkaPrincipal; import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -600,7 +601,8 @@ public Principal peerPrincipal() throws IOException { try { return sslEngine.getSession().getPeerPrincipal(); } catch (SSLPeerUnverifiedException se) { - throw new IOException(String.format("Unable to retrieve getPeerPrincipal due to %s", se)); + log.warn("SSL peer is not authenticated, returning ANONYMOUS instead"); + return KafkaPrincipal.ANONYMOUS; } } diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selector.java b/clients/src/main/java/org/apache/kafka/common/network/Selector.java index 12c911cb66c47..f49d54cbc1915 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/Selector.java +++ b/clients/src/main/java/org/apache/kafka/common/network/Selector.java @@ -465,6 +465,7 @@ private void close(KafkaChannel channel) { this.sensors.connectionClosed.record(); } + /** * check if channel is ready */ @@ -475,9 +476,11 @@ public boolean isChannelReady(String id) { } /** - * Get the channel associated with this numeric id + * Get the channel associated with this connection + * Exposing this to allow SocketServer get the Principal from the channel when creating a request + * without making Selector know about Principals */ - private KafkaChannel channelForId(String id) { + public KafkaChannel channelForId(String id) { KafkaChannel channel = this.channels.get(id); if (channel == null) throw new IllegalStateException("Attempt to write to socket for which there is no open connection. Connection id " + id + " existing connections " + channels.keySet().toString()); diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java b/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java index 277b6efb8cf4a..b640ea0f4bdb6 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java +++ b/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java @@ -20,6 +20,7 @@ import java.security.Principal; public class KafkaPrincipal implements Principal { + public final static KafkaPrincipal ANONYMOUS = new KafkaPrincipal("ANONYMOUS"); private final String name; public KafkaPrincipal(String name) { diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala index 20741281dcaa7..37d9e1fa5c0ca 100644 --- a/core/src/main/scala/kafka/network/RequestChannel.scala +++ b/core/src/main/scala/kafka/network/RequestChannel.scala @@ -18,6 +18,7 @@ package kafka.network import java.nio.ByteBuffer +import java.security.Principal import java.util.concurrent._ import com.yammer.metrics.core.Gauge @@ -29,11 +30,12 @@ import kafka.utils.{Logging, SystemTime} import org.apache.kafka.common.network.Send import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol} import org.apache.kafka.common.requests.{AbstractRequest, RequestHeader} +import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.log4j.Logger object RequestChannel extends Logging { - val AllDone = new Request(processor = 1, connectionId = "2", buffer = getShutdownReceive(), startTimeMs = 0, securityProtocol = SecurityProtocol.PLAINTEXT) + val AllDone = new Request(processor = 1, connectionId = "2", new Session(KafkaPrincipal.ANONYMOUS, ""), buffer = getShutdownReceive(), startTimeMs = 0, securityProtocol = SecurityProtocol.PLAINTEXT) def getShutdownReceive() = { val emptyProducerRequest = new ProducerRequest(0, 0, "", 0, 0, collection.mutable.Map[TopicAndPartition, ByteBufferMessageSet]()) @@ -44,7 +46,9 @@ object RequestChannel extends Logging { byteBuffer } - case class Request(processor: Int, connectionId: String, private var buffer: ByteBuffer, startTimeMs: Long, securityProtocol: SecurityProtocol) { + case class Session(principal: Principal, host: String) + + case class Request(processor: Int, connectionId: String, session: Session, private var buffer: ByteBuffer, startTimeMs: Long, securityProtocol: SecurityProtocol) { @volatile var requestDequeueTimeMs = -1L @volatile var apiLocalCompleteTimeMs = -1L @volatile var responseCompleteTimeMs = -1L @@ -113,11 +117,11 @@ object RequestChannel extends Logging { } if(requestLogger.isTraceEnabled) - requestLogger.trace("Completed request:%s from connection %s;totalTime:%d,requestQueueTime:%d,localTime:%d,remoteTime:%d,responseQueueTime:%d,sendTime:%d" - .format(requestDesc, connectionId, totalTime, requestQueueTime, apiLocalTime, apiRemoteTime, responseQueueTime, responseSendTime)) + requestLogger.trace("Completed request:%s from connection %s;totalTime:%d,requestQueueTime:%d,localTime:%d,remoteTime:%d,responseQueueTime:%d,sendTime:%d,securityProtocol:%s,principal:%s" + .format(requestDesc, connectionId, totalTime, requestQueueTime, apiLocalTime, apiRemoteTime, responseQueueTime, responseSendTime, securityProtocol, session.principal)) else if(requestLogger.isDebugEnabled) - requestLogger.debug("Completed request:%s from connection %s;totalTime:%d,requestQueueTime:%d,localTime:%d,remoteTime:%d,responseQueueTime:%d,sendTime:%d" - .format(requestDesc, connectionId, totalTime, requestQueueTime, apiLocalTime, apiRemoteTime, responseQueueTime, responseSendTime)) + requestLogger.debug("Completed request:%s from connection %s;totalTime:%d,requestQueueTime:%d,localTime:%d,remoteTime:%d,responseQueueTime:%d,sendTime:%d,securityProtocol:%s,principal:%s" + .format(requestDesc, connectionId, totalTime, requestQueueTime, apiLocalTime, apiRemoteTime, responseQueueTime, responseSendTime, securityProtocol, session.principal)) } } diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala index 649812d9f8014..24ac1eaa29dd7 100644 --- a/core/src/main/scala/kafka/network/SocketServer.scala +++ b/core/src/main/scala/kafka/network/SocketServer.scala @@ -404,7 +404,10 @@ private[kafka] class Processor(val id: Int, } collection.JavaConversions.collectionAsScalaIterable(selector.completedReceives).foreach(receive => { try { - val req = RequestChannel.Request(processor = id, connectionId = receive.source, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol) + + val channel = selector.channelForId(receive.source); + val session = RequestChannel.Session(channel.principal, channel.socketDescription) + val req = RequestChannel.Request(processor = id, connectionId = receive.source, session = session, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol) requestChannel.sendRequest(req) } catch { case e @ (_: InvalidRequestException | _: SchemaException) => { diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 67f0cad802f90..825dc9e8f2287 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -55,7 +55,8 @@ class KafkaApis(val requestChannel: RequestChannel, */ def handle(request: RequestChannel.Request) { try{ - trace("Handling request: " + request.requestObj + " from connection: " + request.connectionId) + trace("Handling request:%s from connection %s;securityProtocol:%s,principal:%s". + format(request.requestObj, request.connectionId, request.securityProtocol, request.session.principal)) request.requestId match { case RequestKeys.ProduceKey => handleProducerRequest(request) case RequestKeys.FetchKey => handleFetchRequest(request) diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala index 1937943586087..1585e7160af5d 100644 --- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala +++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala @@ -17,21 +17,15 @@ package kafka.network; -import java.io._ + import java.net._ import javax.net.ssl._ import java.io._ -import java.nio.ByteBuffer -import java.util.Random -import kafka.api.ProducerRequest -import kafka.cluster.EndPoint -import kafka.common.TopicAndPartition -import kafka.message.ByteBufferMessageSet -import kafka.producer.SyncProducerConfig import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.common.network.NetworkSend import org.apache.kafka.common.protocol.SecurityProtocol +import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.utils.SystemTime import org.junit.Assert._ import org.junit._ @@ -43,7 +37,6 @@ import java.nio.ByteBuffer import kafka.common.TopicAndPartition import kafka.message.ByteBufferMessageSet import kafka.server.KafkaConfig -import java.nio.channels.SelectionKey import kafka.utils.TestUtils import scala.collection.Map @@ -230,4 +223,12 @@ class SocketServerTest extends JUnitSuite { assertEquals(serializedBytes.toSeq, receiveResponse(sslSocket).toSeq) overrideServer.shutdown() } + + @Test + def testSessionPrincipal(): Unit = { + val socket = connect() + val bytes = new Array[Byte](40) + sendRequest(socket, 0, bytes) + assertEquals(KafkaPrincipal.ANONYMOUS, server.requestChannel.receiveRequest().session.principal) + } }