如何使用akka在scala中使用TLS打开TCP连接

发布时间：2020-12-16 18:39:53 所属栏目：安全来源：网络整理

导读：我想编写一个 Scala客户端,通过与TLS的tcp连接来讨论专有协议. 基本上,我想从Scala中的Node.js重写以下代码： var conn_options = { host: endpoint,port: port};tlsSocket = tls.connect(conn_options,function() { if (tlsSocket.authorized) { logger.inf

我想编写一个 Scala客户端,通过与TLS的tcp连接来讨论专有协议.

基本上,我想从Scala中的Node.js重写以下代码：

var conn_options = {
        host: endpoint,port: port
};
tlsSocket = tls.connect(conn_options,function() {
      if (tlsSocket.authorized) {
        logger.info('Successfully established a connection');

        // Now that the connection has been established,let's perform the handshake
        // Identification frame:
        // 1 | I | id_size | id
        var idFrameTypeAndVersion = "1I";
        var clientIdString = "foorbar";
        var idDataBuffer = new Buffer(idFrameTypeAndVersion.length + 1 + clientIdString.length);

        idDataBuffer.write(idFrameTypeAndVersion,idFrameTypeAndVersion.length);

        idDataBuffer.writeUIntBE(clientIdString.length,idFrameTypeAndVersion.length,1);
        idDataBuffer.write(clientIdString,idFrameTypeAndVersion.length + 1,clientIdString.length);

        // Send the identification frame to Logmet
        tlsSocket.write(idDataBuffer);

      }
      ...
}

从akka documentation我发现一个很好的例子与Akka相比普通的tcp,但我不知道如何使用TLS套接字连接增强示例.有一些旧版本的文档显示了一个示例with ssl/tls,但在较新版本中遗漏了这个版本.

我在Akka找到了关于TLS对象的文档,但我没有找到任何关于它的好例子.

提前谢谢了！

解决方法

得到它使用以下代码,并希望分享.

基本上,我开始关注从akka社区获得的TcpTlsEcho.java.

我按照akka-streams的文档.另一个非常好的例子,显示和说明akka-streams的用法可以在以下blog post中找到

连接设置和流程如下所示：

/**
    +---------------------------+               +---------------------------+
    | Flow                      |               | tlsConnectionFlow         |
    |                           |               |                           |
    | +------+        +------+  |               |  +------+        +------+ |
    | | SRC  | ~Out~> |      | ~~> O2   --  I1 ~~> |      |  ~O1~> |      | |
    | |      |        | LOGG |  |               |  | TLS  |        | CONN | |
    | | SINK | <~In~  |      | <~~ I2   --  O2 <~~ |      | <~I2~  |      | |
    | +------+        +------+  |               |  +------+        +------+ |
    +---------------------------+               +---------------------------+
**/
// the tcp connection to the server
val connection = Tcp().outgoingConnection(address,port)

// ignore the received data for now. There are different actions to implement the Sink.
val sink = Sink.ignore

// create a source as an actor reference
val source = Source.actorRef(1000,OverflowStrategy.fail)

// join the TLS BidiFlow (see below) with the connection
val tlsConnectionFlow = tlsStage(TLSRole.client).join(connection)

// run the source with the TLS conection flow that is joined with a logging step that prints the bytes that are sent and or received from the connection.
val sourceActor = tlsConnectionFlow.join(logging).to(sink).runWith(source) 

// send a message to the sourceActor that will be send to the Source of the stream
sourceActor ! ByteString("<message>")

TLS连接流是BidiFlow.我的第一个简单示例忽略了所有证书,并避免了管理信任和密钥库.如何完成的示例可以在上面的.java示例中找到.

def tlsStage(role: TLSRole)(implicit system: ActorSystem) = {
    val sslConfig = AkkaSSLConfig.get(system)
    val config = sslConfig.config

    // create a ssl-context that ignores self-signed certificates
    implicit val sslContext: SSLContext = {
        object WideOpenX509TrustManager extends X509TrustManager {
            override def checkClientTrusted(chain: Array[X509Certificate],authType: String) = ()
            override def checkServerTrusted(chain: Array[X509Certificate],authType: String) = ()
            override def getAcceptedIssuers = Array[X509Certificate]()
        }

        val context = SSLContext.getInstance("TLS")
        context.init(Array[KeyManager](),Array(WideOpenX509TrustManager),null)
        context
    }
    // protocols
    val defaultParams = sslContext.getDefaultSSLParameters()
    val defaultProtocols = defaultParams.getProtocols()
    val protocols = sslConfig.configureProtocols(defaultProtocols,config)
    defaultParams.setProtocols(protocols)

    // ciphers
    val defaultCiphers = defaultParams.getCipherSuites()
    val cipherSuites = sslConfig.configureCipherSuites(defaultCiphers,config)
    defaultParams.setCipherSuites(cipherSuites)

    val firstSession = new TLSProtocol.NegotiateNewSession(None,None,None)
       .withCipherSuites(cipherSuites: _*)
       .withProtocols(protocols: _*)
       .withParameters(defaultParams)

    val clientAuth = getClientAuth(config.sslParametersConfig.clientAuth)
    clientAuth map { firstSession.withClientAuth(_) }

    val tls = TLS.apply(sslContext,firstSession,role)

    val pf: PartialFunction[TLSProtocol.SslTlsInbound,ByteString] = {
      case TLSProtocol.SessionBytes(_,sb) => ByteString.fromByteBuffer(sb.asByteBuffer)
    }

    val tlsSupport = BidiFlow.fromFlows(
        Flow[ByteString].map(TLSProtocol.SendBytes),Flow[TLSProtocol.SslTlsInbound].collect(pf));

    tlsSupport.atop(tls);
  }

  def getClientAuth(auth: ClientAuth) = {
     if (auth.equals(ClientAuth.want)) {
         Some(TLSClientAuth.want)
     } else if (auth.equals(ClientAuth.need)) {
         Some(TLSClientAuth.need)
     } else if (auth.equals(ClientAuth.none)) {
         Some(TLSClientAuth.none)
     } else {
         None
     }
  }

完成后,日志记录阶段也作为BidiFlow实现.

def logging: BidiFlow[ByteString,ByteString,NotUsed] = {
    // function that takes a string,prints it with some fixed prefix in front and returns the string again
    def logger(prefix: String) = (chunk: ByteString) => {
      println(prefix + chunk.utf8String)
      chunk
    }

    val inputLogger = logger("> ")
    val outputLogger = logger("< ")

    // create BidiFlow with a separate logger function for each of both streams
    BidiFlow.fromFunctions(outputLogger,inputLogger)
 }

我将进一步尝试改进和更新答案.希望有所帮助.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!