Transport layer change to GRPC over TCP

Owned by Sebastian Bach
Last updated: Apr 27, 2018

Motivation

Communication layer uses UDP as transport layer for inter-node communication.

The handshake between nodes is divided into two phases:

  1. Encryption handshake (public key exchange)
  2. Protocol handshake (nonce)

The encryption part is provided by an Rchain custom solution. On the other hand the communication pattern between nodes corresponds to request-response cycles. So UPD is not a good choice since this protocol corresponds to a fire-and-forget semantic.

Background

GRPC is a communication protocol based on Google Protocol Buffers (protobuf) over TCP. The implementation for Java provides an HTTP/2 TLS communication out of the box. It's a natural choice for efficient communication based on request-response cycles. On the other hand TLS is a proven and reliable secure communication protocol. Changing 

Caveat

Nodes are identified in the Rchain P2P-network by their public key. Although GRPC is convenient to use, there is quite a hassle to extract the peer public key from the underlying TLS communication. To extract the keys it is required to inject call interceptors on client and server side. A POC implementation can be found here Github: grpc-example. Interceptor excerpts from this repository:

SslSessionClientInterceptor.scala
class SslSessionClientCallInterceptor[ReqT, RespT](next: ClientCall[ReqT, RespT]) extends ClientCall[ReqT, RespT] {
  self =>

  private val logger = Logger(this.getClass)

  def cancel(message: String, cause: Throwable): Unit = next.cancel(message, cause)
  def request(numMessages: Int): Unit = next.request(numMessages)
  def sendMessage(message: ReqT): Unit = next.sendMessage(message)
  def halfClose(): Unit = next.halfClose()

  override def isReady: Boolean = next.isReady
  override def setMessageCompression(enabled: Boolean): Unit = next.setMessageCompression(enabled)
  override def getAttributes: Attributes = next.getAttributes

  def start(responseListener: ClientCall.Listener[RespT], headers: Metadata): Unit =
    next.start(new InterceptionListener(responseListener), headers)

  private class InterceptionListener(next: ClientCall.Listener[RespT]) extends ClientCall.Listener[RespT] {
    override def onClose(status: Status, trailers: Metadata): Unit = next.onClose(status, trailers)
    override def onReady(): Unit = next.onReady()
    override def onHeaders(headers: Metadata): Unit = next.onHeaders(headers)

    override def onMessage(message: RespT): Unit = {
      message match {
        case handshake: HandshakeReply =>
          val sslSession: Option[SSLSession] = Option(self.getAttributes.get(Grpc.TRANSPORT_ATTR_SSL_SESSION))
          if (sslSession.isEmpty) {
            logger.error("No SSL Session found in client call")
            close()
          } else {
            sslSession.foreach { session =>
              val pubKey = Base64.getEncoder.encodeToString(session.getPeerCertificates.head.getPublicKey.getEncoded)
              if (pubKey == handshake.key) {
                next.onMessage(message)
              } else {
                logger.error("Wrong public key")
                close()
              }
            }
          }

        case _ => next.onMessage(message)
      }
    }

    private def close(): Unit =
      throw Status.UNAUTHENTICATED.withDescription("Wrong public key").asRuntimeException()

  }
}
SslSessionServerInterceptor.scala
class SslSessionServerInterceptor() extends ServerInterceptor {

  private val logger = Logger(this.getClass)

  def interceptCall[ReqT, RespT](
    call: ServerCall[ReqT, RespT],
    headers: Metadata,
    next: ServerCallHandler[ReqT, RespT]
  ): ServerCall.Listener[ReqT] = new InterceptionListener(next.startCall(call, headers), call)

  private class InterceptionListener[ReqT, RespT](next: ServerCall.Listener[ReqT], call: ServerCall[ReqT, RespT]) extends ServerCall.Listener[ReqT] {
    override def onHalfClose(): Unit = next.onHalfClose()
    override def onCancel(): Unit = next.onCancel()
    override def onComplete(): Unit = next.onComplete()
    override def onReady(): Unit = next.onReady()

    override def onMessage(message: ReqT): Unit = {
      message match {
        case handshake: HandshakeRequest =>
          val sslSession: Option[SSLSession] = Option(call.getAttributes.get(Grpc.TRANSPORT_ATTR_SSL_SESSION))
          if (sslSession.isEmpty) {
            logger.error("No SSL Session found in server call")
            close()
          } else {
            sslSession.foreach { session =>
              val pubKey = Base64.getEncoder.encodeToString(session.getPeerCertificates.head.getPublicKey.getEncoded)
              if (pubKey == handshake.key) {
                next.onMessage(message)
              } else {
                logger.error("Wrong public key")
                close()
              }
            }
          }
        case _ => next.onMessage(message)
      }
    }

    private def close(): Unit =
      throw Status.UNAUTHENTICATED.withDescription("Wrong public key").asRuntimeException()
  }
}

GRPC-Java and secp256k1 certificates

Motivation

Rchain tokens (RHOC) are ERC20 Ethereum based. On the other hand Ethereum uses the Elliptic-curve secp256k1 cryptographic algorithm (ECC) for private/public key creation. During the swap from RHOC to REV (the actual RChain currency) it's desirable that users can keep their public address (private/public keys) and just change the Ethereum address prefix from 0x the the RChain address prefix.

Node dependencies

In our first attempt we used for TLS encryption a netty library that is statically linked with Google's BoringSSL (netty-tcnative-boringssl-static). However, it turned out that BoringSSL doesn't support the secp256k1 group of ECC. To run the node with secp256k1 with GRPC-Java it must be used together with Openssl on the host machine. This means, that Openssl must be configured as a dependency in each of the deployment profiles (DEB, RPM, Docker and JAR).