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.effects.IO
  26 import arrow.syntax.collections.firstOption
  27 import io.netty.handler.ssl.SslHandler
  28 import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
  29 import org.onap.dcae.collectors.veshv.boundary.Metrics
  30 import org.onap.dcae.collectors.veshv.boundary.Server
  31 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.debug
  32 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.info
  33 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.withWarn
  34 import org.onap.dcae.collectors.veshv.model.ClientContext
  35 import org.onap.dcae.collectors.veshv.model.ServerConfiguration
  36 import org.onap.dcae.collectors.veshv.model.ServiceContext
  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 org.onap.dcae.collectors.veshv.utils.logging.Marker
  42 import reactor.core.publisher.Mono
  43 import reactor.netty.ByteBufFlux
  44 import reactor.netty.Connection
  45 import reactor.netty.NettyInbound
  46 import reactor.netty.NettyOutbound
  47 import reactor.netty.tcp.TcpServer
  48 import java.security.cert.X509Certificate
  49 import java.time.Duration
  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,
  60                               private val metrics: Metrics) : Server {
  61
  62     override fun start(): IO<ServerHandle> = IO {
  63         TcpServer.create()
  64                 .addressSupplier { serverConfig.serverListenAddress }
  65                 .configureSsl()
  66                 .handle(this::handleConnection)
  67                 .doOnUnbound {
  68                     logger.info(ServiceContext::mdc) { "Netty TCP Server closed" }
  69                     collectorProvider.close().unsafeRunSync()
  70                 }
  71                 .let { NettyServerHandle(it.bindNow()) }
  72     }
  73
  74     private fun TcpServer.configureSsl() =
  75             sslContextFactory
  76                     .createSslContext(serverConfig.securityConfiguration)
  77                     .map { sslContext ->
  78                         logger.info { "Collector configured with SSL enabled" }
  79                         this.secure { b -> b.sslContext(sslContext) }
  80                     }.getOrElse {
  81                         logger.info { "Collector configured with SSL disabled" }
  82                         this
  83                     }
  84
  85     private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> {
  86         metrics.notifyClientConnected()
  87         val clientContext = ClientContext(nettyOutbound.alloc())
  88         nettyInbound.withConnection {
  89             populateClientContext(clientContext, it)
  90             it.channel().pipeline().get(SslHandler::class.java)?.engine()?.session?.let { sslSession ->
  91                 sslSession.peerCertificates.firstOption().map { it as X509Certificate }.map { it.subjectDN.name }
  92             }
  93         }
  94
  95         logger.debug(clientContext::fullMdc, Marker.Entry) { "Client connection request received" }
  96         messageHandlingStream(clientContext, nettyInbound).subscribe()
  97         return nettyOutbound.neverComplete()
  98     }
  99
 100     private fun messageHandlingStream(clientContext: ClientContext, nettyInbound: NettyInbound): Mono<Void> =
 101         collectorProvider(clientContext).fold(
 102                 {
 103                     logger.warn(clientContext::fullMdc) { "Collector not ready. Closing connection..." }
 104                     Mono.empty()
 105                 },
 106                 {
 107                     logger.info(clientContext::fullMdc) { "Handling new connection" }
 108                     nettyInbound.withConnection { conn ->
 109                         conn
 110                                 .configureIdleTimeout(clientContext, serverConfig.idleTimeout)
 111                                 .logConnectionClosed(clientContext)
 112                     }
 113                     it.handleConnection(createDataStream(nettyInbound))
 114                 }
 115         )
 116
 117     private fun populateClientContext(clientContext: ClientContext, connection: Connection) {
 118         clientContext.clientAddress = try {
 119             Option.fromNullable(connection.address().address)
 120         } catch (ex: Exception) {
 121             None
 122         }
 123         clientContext.clientCert = getSslSession(connection).flatMap(::findClientCert)
 124     }
 125
 126     private fun getSslSession(connection: Connection) = Option.fromNullable(
 127             connection
 128                     .channel()
 129                     .pipeline()
 130                     .get(SslHandler::class.java)
 131                     ?.engine()
 132                     ?.session)
 133
 134     private fun findClientCert(sslSession: SSLSession): Option<X509Certificate> =
 135             sslSession
 136                     .peerCertificates
 137                     .firstOption()
 138                     .flatMap { Option.fromNullable(it as? X509Certificate) }
 139
 140     private fun createDataStream(nettyInbound: NettyInbound): ByteBufFlux = nettyInbound
 141             .receive()
 142             .retain()
 143
 144     private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection =
 145             onReadIdle(timeout.toMillis()) {
 146                 logger.info(ctx) {
 147                     "Idle timeout of ${timeout.seconds} s reached. Closing connection from ${address()}..."
 148                 }
 149                 disconnectClient(ctx)
 150             }
 151
 152
 153     private fun Connection.disconnectClient(ctx: ClientContext) {
 154         channel().close().addListener {
 155             logger.debug(ctx::fullMdc, Marker.Exit) { "Closing client channel." }
 156             if (it.isSuccess)
 157                 logger.debug(ctx) { "Channel closed successfully." }
 158             else
 159                 logger.withWarn(ctx) { log("Channel close failed", it.cause()) }
 160         }
 161     }
 162
 163     private fun Connection.logConnectionClosed(ctx: ClientContext): Connection =
 164             onDispose {
 165                 metrics.notifyClientDisconnected()
 166                 logger.info(ctx::fullMdc, Marker.Exit) { "Connection has been closed" }
 167             }
 168
 169     companion object {
 170         private val logger = Logger(NettyTcpServer::class)
 171     }
 172 }