sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * dcaegen2-collectors-veshv
   4  * ================================================================================
   5  * Copyright (C) 2018 NOKIA
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20 package org.onap.dcae.collectors.veshv.impl.socket
  21
  22 import arrow.core.None
  23 import arrow.core.Option
  24 import arrow.core.getOrElse
  25 import arrow.core.toOption
  26 import arrow.effects.IO
  27 import arrow.syntax.collections.firstOption
  28 import io.netty.handler.ssl.SslHandler
  29 import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
  30 import org.onap.dcae.collectors.veshv.boundary.Server
  31 import org.onap.dcae.collectors.veshv.model.ClientContext
  32 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.info
  33 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.debug
  34 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.withWarn
  35 import org.onap.dcae.collectors.veshv.utils.logging.Marker
  36 import org.onap.dcae.collectors.veshv.model.ServerConfiguration
  37 import org.onap.dcae.collectors.veshv.ssl.boundary.ServerSslContextFactory
  38 import org.onap.dcae.collectors.veshv.utils.NettyServerHandle
  39 import org.onap.dcae.collectors.veshv.utils.ServerHandle
  40 import org.onap.dcae.collectors.veshv.utils.logging.Logger
  41 import reactor.core.publisher.Mono
  42 import reactor.netty.ByteBufFlux
  43 import reactor.netty.Connection
  44 import reactor.netty.NettyInbound
  45 import reactor.netty.NettyOutbound
  46 import reactor.netty.tcp.TcpServer
  47 import java.time.Duration
  48 import java.lang.Exception
  49 import java.security.cert.X509Certificate
  50 import javax.net.ssl.SSLSession
  51
  52
  53 /**
  54  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  55  * @since May 2018
  56  */
  57 internal class NettyTcpServer(private val serverConfig: ServerConfiguration,
  58                               private val sslContextFactory: ServerSslContextFactory,
  59                               private val collectorProvider: CollectorProvider) : Server {
  60
  61     override fun start(): IO<ServerHandle> = IO {
  62         val tcpServer = TcpServer.create()
  63                 .addressSupplier { serverConfig.serverListenAddress }
  64                 .configureSsl()
  65                 .handle(this::handleConnection)
  66
  67         NettyServerHandle(tcpServer.bindNow())
  68     }
  69
  70     private fun TcpServer.configureSsl() =
  71             sslContextFactory
  72                     .createSslContext(serverConfig.securityConfiguration)
  73                     .map { sslContext ->
  74                         logger.info { "Collector configured with SSL enabled" }
  75                         this.secure { b -> b.sslContext(sslContext) }
  76                     }.getOrElse {
  77                         logger.info { "Collector configured with SSL disabled" }
  78                         this
  79                     }
  80
  81     private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> {
  82         val clientContext = ClientContext(nettyOutbound.alloc())
  83         nettyInbound.withConnection {
  84             populateClientContext(clientContext, it)
  85             it.channel().pipeline().get(SslHandler::class.java)?.engine()?.session?.let { sslSession ->
  86                 sslSession.peerCertificates.firstOption().map { it as X509Certificate }.map { it.subjectDN.name }
  87             }
  88
  89         }
  90
  91         logger.debug(clientContext::fullMdc, Marker.Entry) { "Client connection request received" }
  92         return collectorProvider(clientContext).fold(
  93                 {
  94                     logger.warn(clientContext::fullMdc) { "Collector not ready. Closing connection..." }
  95                     Mono.empty()
  96                 },
  97                 {
  98                     logger.info(clientContext::fullMdc) { "Handling new connection" }
  99                     nettyInbound.withConnection { conn ->
 100                         conn.configureIdleTimeout(clientContext, serverConfig.idleTimeout)
 101                                 .logConnectionClosed(clientContext)
 102                     }
 103                     it.handleConnection(createDataStream(nettyInbound))
 104                 }
 105         )
 106     }
 107
 108     private fun populateClientContext(clientContext: ClientContext, connection: Connection) {
 109         clientContext.clientAddress = try {
 110             Option.fromNullable(connection.address().address)
 111         } catch (ex: Exception) {
 112             None
 113         }
 114         clientContext.clientCert = getSslSession(connection).flatMap(::findClientCert)
 115     }
 116
 117     private fun getSslSession(connection: Connection) = Option.fromNullable(
 118             connection
 119                     .channel()
 120                     .pipeline()
 121                     .get(SslHandler::class.java)
 122                     ?.engine()
 123                     ?.session)
 124
 125     private fun findClientCert(sslSession: SSLSession): Option<X509Certificate> =
 126             sslSession
 127                     .peerCertificates
 128                     .firstOption()
 129                     .flatMap { Option.fromNullable(it as? X509Certificate) }
 130
 131     private fun createDataStream(nettyInbound: NettyInbound): ByteBufFlux = nettyInbound
 132             .receive()
 133             .retain()
 134
 135     private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection {
 136         onReadIdle(timeout.toMillis()) {
 137             logger.info(ctx) {
 138                 "Idle timeout of ${timeout.seconds} s reached. Closing connection from ${address()}..."
 139             }
 140             disconnectClient(ctx)
 141         }
 142         return this
 143     }
 144
 145     private fun Connection.disconnectClient(ctx: ClientContext) {
 146         channel().close().addListener {
 147             logger.debug(ctx::fullMdc, Marker.Exit) { "Closing client channel." }
 148             if (it.isSuccess)
 149                 logger.debug(ctx) { "Channel closed successfully." }
 150             else
 151                 logger.withWarn(ctx) { log("Channel close failed", it.cause()) }
 152         }
 153     }
 154
 155     private fun Connection.logConnectionClosed(ctx: ClientContext): Connection {
 156         onTerminate().subscribe {
 157             // TODO: this code is never executed (at least with ssl-enabled, did not checked with ssl-disabled)
 158             logger.info(ctx::fullMdc, Marker.Exit) { "Connection has been closed" }
 159         }
 160         return this
 161     }
 162
 163     companion object {
 164         private val logger = Logger(NettyTcpServer::class)
 165     }
 166 }