finagle/twitter-server: Add admin endpoint for info about clients con…

…nected to servers

Problem

There is no way to get information about clients currently connected to a
finagle server.

Solution/Result

Add an admin endpoint to list basic information about each client connected, as
well as information about the connection's encryption status, for each listening
server.

JIRA Issues: CSL-8104

Differential Revision: https://phabricator.twitter.biz/D329940

CHANGELOG.rst

@@ -20,6 +20,12 @@ Changes
 
 * Remove the TwitterServer dependency on Netty 3. ``PHAB_ID=D328148``
 
+New Features
+~~~~~~~~~~~~
+
+* Added an admin page, /admin/servers/connections.json with details about incoming connections,
+  including encryption status and remote principal ``PHAB_ID=D329940``
+
 19.5.1
 ------
 

doc/src/sphinx/Admin.rst

@@ -469,3 +469,9 @@ that your service is running.
 
 Surface server information exposed by Finagle. Per-server configuration parameters and
 values for each module are available at `/admin/servers/<server name>`.
+
+/admin/servers/connections
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Expose information about currently connected clients including encryption status,
+if available.

server/src/main/scala/com/twitter/server/Admin.scala

@@ -263,6 +263,13 @@ trait Admin { self: App with AdminHttpServer with Stats =>
         alias = "Favicon",
         group = None,
         includeInIndex = false
+      ),
+      Route(
+        path = Path.Servers + "connections/",
+        handler = new AttachedClientsHandler(),
+        alias = "Incoming Connections",
+        group = Some(Grouping.Utilities),
+        includeInIndex = true
       )
     ).map(Route.isolate)
 

server/src/main/scala/com/twitter/server/handler/AttachedClientsHandler.scala

@@ -0,0 +1,92 @@
+package com.twitter.server.handler
+
+import com.fasterxml.jackson.annotation.JsonInclude
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.PropertyNamingStrategy
+import com.fasterxml.jackson.databind.annotation.JsonNaming
+import com.twitter.finagle.Service
+import com.twitter.finagle.http.{Request, Response}
+import com.twitter.finagle.server.ServerRegistry
+import com.twitter.server.util.JsonConverter
+import com.twitter.util.Future
+import java.net.SocketAddress
+
+private[handler] object AttachedClientsHandler {
+
+  case class ClientConnectionEntry(address: SocketAddress, ssl: Option[ClientSslInfo])
+
+  @JsonNaming(classOf[PropertyNamingStrategy.SnakeCaseStrategy])
+  case class PeerCertInfo(commonName: String)
+
+  @JsonNaming(classOf[PropertyNamingStrategy.SnakeCaseStrategy])
+  @JsonInclude(value = Include.NON_ABSENT, content = Include.NON_ABSENT)
+  case class ClientSslInfo(
+    sessionId: String,
+    cipherSuite: String,
+    peerCertificate: Option[PeerCertInfo])
+
+  case class ServerConnectionInfo(address: SocketAddress, clients: Seq[ClientConnectionEntry])
+
+  case class AttachedClientsConnectionInfo(servers: Seq[ServerConnectionInfo])
+
+  private def render(serverRegistry: ServerRegistry): AttachedClientsConnectionInfo = {
+    AttachedClientsConnectionInfo(serverRegistry.serverAddresses.flatMap { serverAddress =>
+      val connectionRegistry = serverRegistry.connectionRegistry(serverAddress)
+      Some(
+        ServerConnectionInfo(
+          address = serverAddress,
+          clients = connectionRegistry.iterator.map {
+            clientConnection =>
+              ClientConnectionEntry(
+                address = clientConnection.remoteAddress,
+                ssl = if (!clientConnection.sslSessionInfo.usingSsl) {
+                  None
+                } else {
+                  Some(ClientSslInfo(
+                    sessionId = clientConnection.sslSessionInfo.sessionId,
+                    cipherSuite = clientConnection.sslSessionInfo.cipherSuite,
+                    peerCertificate = clientConnection.sslSessionInfo.peerCertificates.headOption
+                      .map { peerCertificate =>
+                        PeerCertInfo(peerCertificate.getSubjectDN.getName)
+                      }
+                  ))
+                }
+              )
+          }.toList
+        ))
+    })
+  }
+}
+
+/**
+ * A handler that reports information about connected clients, by server. For example:
+ *
+ * {{{
+ * {
+ *   "servers": [
+ *     {
+ *       "address": "127.0.0.1:8080",
+ *       "clients": [
+ *         {
+ *           "address": "127.0.0.1:9090",
+ *           "ssl": {
+ *             "session_id": "sessionid",
+ *             "cipher_suite": "cipher?sweeeeet!",
+ *             "peer_certificate": {
+ *               "common_name": "remoteprincipal"
+ *             }
+ *           }
+ *         }
+ *       ]
+ *     }
+ *   ]
+ * }
+ * }}}
+ */
+class AttachedClientsHandler(serverRegistry: ServerRegistry = ServerRegistry)
+    extends Service[Request, Response] {
+  def apply(request: Request): Future[Response] = {
+    val doc = AttachedClientsHandler.render(serverRegistry)
+    Future.value(JsonConverter(doc))
+  }
+}

server/src/test/scala/com/twitter/server/handler/AttachedClientsHandlerTest.scala

@@ -0,0 +1,106 @@
+package com.twitter.server.handler
+
+import com.twitter.conversions.DurationOps._
+import com.twitter.finagle.http.{Request, Status}
+import com.twitter.finagle.server.ServerRegistry
+import com.twitter.finagle.service.StatsFilter
+import com.twitter.finagle.ssl.session.SslSessionInfo
+import com.twitter.finagle.{ClientConnection, Service, ServiceFactory, Stack, param}
+import com.twitter.io.Buf
+import com.twitter.util.{Await, Future, Time}
+import java.net.{InetSocketAddress, SocketAddress}
+import java.security.Principal
+import java.security.cert.X509Certificate
+import javax.net.ssl.SSLSession
+import org.mockito.Mockito
+import org.scalatest.FunSuite
+
+object AttachedClientsHandlerTest {
+  def await[A](f: Future[A]): A = Await.result(f, 2.seconds)
+
+  val remoteSocketAddress = InetSocketAddress.createUnresolved("/127.0.0.1", 9090)
+
+  def registerServer(registry: ServerRegistry, addr: SocketAddress, name: String): Unit = {
+    val ok = ServiceFactory.const(Service.const(Future.value("ok")))
+    val leaf = StatsFilter.module.toStack(Stack.leaf(Stack.Role("ok"), ok))
+
+    registry.register(addr.toString, leaf, Stack.Params.empty + param.Label(name))
+
+    val peerCertificate = Mockito.mock(classOf[X509Certificate])
+    val remotePrincipal = Mockito.mock(classOf[Principal])
+    Mockito.when(peerCertificate.getSubjectDN).thenReturn(remotePrincipal)
+    Mockito.when(remotePrincipal.getName).thenReturn("remoteprincipal")
+
+    registry
+      .connectionRegistry(addr).register(new ClientConnection {
+        override def remoteAddress: SocketAddress = remoteSocketAddress
+        override def localAddress: SocketAddress = addr
+        override def onClose: Future[Unit] = ???
+        override def close(deadline: Time): Future[Unit] = ???
+        override def sslSessionInfo: SslSessionInfo = new SslSessionInfo {
+          override def usingSsl: Boolean = true
+          override def session: SSLSession = ???
+          override def sessionId: String = "sessionid"
+          override def cipherSuite: String = "cipher?sweeeeet!"
+          override def localCertificates: Seq[X509Certificate] = ???
+          override def peerCertificates: Seq[X509Certificate] = Seq(peerCertificate)
+        }
+      })
+  }
+}
+
+class AttachedClientsHandlerTest extends FunSuite {
+  import AttachedClientsHandlerTest._
+
+  private[this] def assertJsonResponse(actualResponse: String, expectedResponse: String) = {
+    val actual = stripWhitespace(actualResponse)
+    val expected = stripWhitespace(expectedResponse)
+    assert(actual == expected)
+  }
+
+  private[this] def stripWhitespace(string: String): String =
+    string.filter { case c => !c.isWhitespace }
+
+  test("initial state") {
+    val registry = new ServerRegistry()
+    val handler = new AttachedClientsHandler(registry)
+    val response = await(handler(Request("/connections/")))
+    val Buf.Utf8(content) = response.content
+    assertJsonResponse(content, """{"servers":[]}""")
+  }
+
+  test("add a client connection") {
+
+    val registry = new ServerRegistry()
+    registerServer(registry, InetSocketAddress.createUnresolved("/127.0.0.1", 8080), "server0")
+
+    val handler = new AttachedClientsHandler(registry)
+    val response = await(handler(Request("/connections/")))
+
+    assert(response.status == Status.Ok)
+    assertJsonResponse(
+      response.contentString,
+      """
+        |{
+        |  "servers": [
+        |    {
+        |      "address": "127.0.0.1:8080",
+        |      "clients": [
+        |        {
+        |          "address": "127.0.0.1:9090",
+        |          "ssl": {
+        |            "session_id": "sessionid",
+        |            "cipher_suite": "cipher?sweeeeet!",
+        |            "peer_certificate": {
+        |              "common_name": "remoteprincipal"
+        |            }
+        |          }
+        |        }
+        |      ]
+        |    }
+        |  ]
+        |}
+      """.stripMargin
+    )
+  }
+}