gerrit.onap Code Review - dcaegen2/collectors/hv-ves.git/blob

   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         return collectorProvider(clientContext).fold(
  97                 {
  98                     logger.warn(clientContext::fullMdc) { "Collector not ready. Closing connection..." }
  99                     Mono.empty()
 100                 },
 101                 {
 102                     logger.info(clientContext::fullMdc) { "Handling new connection" }
 103                     nettyInbound.withConnection { conn ->
 104                         conn
 105                                 .configureIdleTimeout(clientContext, serverConfig.idleTimeout)
 106                                 .logConnectionClosed(clientContext)
 107                     }
 108                     it.handleConnection(createDataStream(nettyInbound))
 109                 }
 110         )
 111     }
 112
 113     private fun populateClientContext(clientContext: ClientContext, connection: Connection) {
 114         clientContext.clientAddress = try {
 115             Option.fromNullable(connection.address().address)
 116         } catch (ex: Exception) {
 117             None
 118         }
 119         clientContext.clientCert = getSslSession(connection).flatMap(::findClientCert)
 120     }
 121
 122     private fun getSslSession(connection: Connection) = Option.fromNullable(
 123             connection
 124                     .channel()
 125                     .pipeline()
 126                     .get(SslHandler::class.java)
 127                     ?.engine()
 128                     ?.session)
 129
 130     private fun findClientCert(sslSession: SSLSession): Option<X509Certificate> =
 131             sslSession
 132                     .peerCertificates
 133                     .firstOption()
 134                     .flatMap { Option.fromNullable(it as? X509Certificate) }
 135
 136     private fun createDataStream(nettyInbound: NettyInbound): ByteBufFlux = nettyInbound
 137             .receive()
 138             .retain()
 139
 140     private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection =
 141             onReadIdle(timeout.toMillis()) {
 142                 logger.info(ctx) {
 143                     "Idle timeout of ${timeout.seconds} s reached. Closing connection from ${address()}..."
 144                 }
 145                 disconnectClient(ctx)
 146             }
 147
 148
 149     private fun Connection.disconnectClient(ctx: ClientContext) {
 150         channel().close().addListener {
 151             logger.debug(ctx::fullMdc, Marker.Exit) { "Closing client channel." }
 152             if (it.isSuccess)
 153                 logger.debug(ctx) { "Channel closed successfully." }
 154             else
 155                 logger.withWarn(ctx) { log("Channel close failed", it.cause()) }
 156         }
 157     }
 158
 159     private fun Connection.logConnectionClosed(ctx: ClientContext): Connection =
 160             onDispose {
 161                 metrics.notifyClientDisconnected()
 162                 logger.info(ctx::fullMdc, Marker.Exit) { "Connection has been closed" }
 163             }
 164
 165     companion object {
 166         private val logger = Logger(NettyTcpServer::class)
 167     }
 168 }