X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=enrichment-coordinator-service%2Fsrc%2Fmain%2Fjava%2Forg%2Foransc%2Fenrichment%2Fclients%2FAsyncRestClient.java;h=b7f23b1f5d20df269b8a8e36dc83fee317c5e7ba;hb=5ee9fd987436011e7130eb05126858cfe54ca545;hp=2782f94acf869d0eca57a08c983b33e7cdefba80;hpb=6f86ab364ac739951556bf2d5bf70429b518de47;p=nonrtric.git diff --git a/enrichment-coordinator-service/src/main/java/org/oransc/enrichment/clients/AsyncRestClient.java b/enrichment-coordinator-service/src/main/java/org/oransc/enrichment/clients/AsyncRestClient.java index 2782f94a..b7f23b1f 100644 --- a/enrichment-coordinator-service/src/main/java/org/oransc/enrichment/clients/AsyncRestClient.java +++ b/enrichment-coordinator-service/src/main/java/org/oransc/enrichment/clients/AsyncRestClient.java @@ -2,7 +2,7 @@ * ========================LICENSE_START================================= * O-RAN-SC * %% - * Copyright (C) 2019 Nordix Foundation + * Copyright (C) 2020 Nordix Foundation * %% * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,36 +22,19 @@ package org.oransc.enrichment.clients; import io.netty.channel.ChannelOption; import io.netty.handler.ssl.SslContext; -import io.netty.handler.ssl.SslContextBuilder; -import io.netty.handler.ssl.util.InsecureTrustManagerFactory; import io.netty.handler.timeout.ReadTimeoutHandler; import io.netty.handler.timeout.WriteTimeoutHandler; -import java.io.FileInputStream; -import java.io.IOException; -import java.io.InputStream; import java.lang.invoke.MethodHandles; -import java.security.KeyStore; -import java.security.KeyStoreException; -import java.security.NoSuchAlgorithmException; -import java.security.cert.Certificate; -import java.security.cert.CertificateException; -import java.security.cert.X509Certificate; -import java.util.Collections; -import java.util.List; import java.util.concurrent.atomic.AtomicInteger; -import java.util.stream.Collectors; -import javax.net.ssl.KeyManagerFactory; - -import org.oransc.enrichment.configuration.WebClientConfig; +import org.oransc.enrichment.configuration.WebClientConfig.HttpProxyConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.http.client.reactive.ReactorClientHttpConnector; import org.springframework.lang.Nullable; -import org.springframework.util.ResourceUtils; import org.springframework.web.reactive.function.client.ExchangeStrategies; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.client.WebClient.RequestHeadersSpec; @@ -59,29 +42,24 @@ import org.springframework.web.reactive.function.client.WebClientResponseExcepti import reactor.core.publisher.Mono; import reactor.netty.http.client.HttpClient; -import reactor.netty.resources.ConnectionProvider; -import reactor.netty.tcp.TcpClient; +import reactor.netty.transport.ProxyProvider; /** * Generic reactive REST client. */ public class AsyncRestClient { + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); private WebClient webClient = null; private final String baseUrl; private static final AtomicInteger sequenceNumber = new AtomicInteger(); - private final WebClientConfig clientConfig; - static KeyStore clientTrustStore = null; - private boolean sslEnabled = true; - - public AsyncRestClient(String baseUrl) { - this(baseUrl, null); - this.sslEnabled = false; - } + private final SslContext sslContext; + private final HttpProxyConfig httpProxyConfig; - public AsyncRestClient(String baseUrl, WebClientConfig config) { + public AsyncRestClient(String baseUrl, @Nullable SslContext sslContext, @Nullable HttpProxyConfig httpProxyConfig) { this.baseUrl = baseUrl; - this.clientConfig = config; + this.sslContext = sslContext; + this.httpProxyConfig = httpProxyConfig; } public Mono> postForEntity(String uri, @Nullable String body) { @@ -89,106 +67,99 @@ public class AsyncRestClient { logger.debug("{} POST uri = '{}{}''", traceTag, baseUrl, uri); logger.trace("{} POST body: {}", traceTag, body); Mono bodyProducer = body != null ? Mono.just(body) : Mono.empty(); - return getWebClient() // - .flatMap(client -> { - RequestHeadersSpec request = client.post() // - .uri(uri) // - .contentType(MediaType.APPLICATION_JSON) // - .body(bodyProducer, String.class); - return retrieve(traceTag, request); - }); + + RequestHeadersSpec request = getWebClient() // + .post() // + .uri(uri) // + .contentType(MediaType.APPLICATION_JSON) // + .body(bodyProducer, String.class); + return retrieve(traceTag, request); } public Mono post(String uri, @Nullable String body) { return postForEntity(uri, body) // - .flatMap(this::toBody); + .map(this::toBody); } public Mono postWithAuthHeader(String uri, String body, String username, String password) { Object traceTag = createTraceTag(); logger.debug("{} POST (auth) uri = '{}{}''", traceTag, baseUrl, uri); logger.trace("{} POST body: {}", traceTag, body); - return getWebClient() // - .flatMap(client -> { - RequestHeadersSpec request = client.post() // - .uri(uri) // - .headers(headers -> headers.setBasicAuth(username, password)) // - .contentType(MediaType.APPLICATION_JSON) // - .bodyValue(body); - return retrieve(traceTag, request) // - .flatMap(this::toBody); - }); + + RequestHeadersSpec request = getWebClient() // + .post() // + .uri(uri) // + .headers(headers -> headers.setBasicAuth(username, password)) // + .contentType(MediaType.APPLICATION_JSON) // + .bodyValue(body); + return retrieve(traceTag, request) // + .map(this::toBody); } public Mono> putForEntity(String uri, String body) { Object traceTag = createTraceTag(); logger.debug("{} PUT uri = '{}{}''", traceTag, baseUrl, uri); logger.trace("{} PUT body: {}", traceTag, body); - return getWebClient() // - .flatMap(client -> { - RequestHeadersSpec request = client.put() // - .uri(uri) // - .contentType(MediaType.APPLICATION_JSON) // - .bodyValue(body); - return retrieve(traceTag, request); - }); + + RequestHeadersSpec request = getWebClient() // + .put() // + .uri(uri) // + .contentType(MediaType.APPLICATION_JSON) // + .bodyValue(body); + return retrieve(traceTag, request); } public Mono> putForEntity(String uri) { Object traceTag = createTraceTag(); logger.debug("{} PUT uri = '{}{}''", traceTag, baseUrl, uri); logger.trace("{} PUT body: ", traceTag); - return getWebClient() // - .flatMap(client -> { - RequestHeadersSpec request = client.put() // - .uri(uri); - return retrieve(traceTag, request); - }); + RequestHeadersSpec request = getWebClient() // + .put() // + .uri(uri); + return retrieve(traceTag, request); } public Mono put(String uri, String body) { return putForEntity(uri, body) // - .flatMap(this::toBody); + .map(this::toBody); } public Mono> getForEntity(String uri) { Object traceTag = createTraceTag(); logger.debug("{} GET uri = '{}{}''", traceTag, baseUrl, uri); - return getWebClient() // - .flatMap(client -> { - RequestHeadersSpec request = client.get().uri(uri); - return retrieve(traceTag, request); - }); + RequestHeadersSpec request = getWebClient().get().uri(uri); + return retrieve(traceTag, request); } public Mono get(String uri) { return getForEntity(uri) // - .flatMap(this::toBody); + .map(this::toBody); } public Mono> deleteForEntity(String uri) { Object traceTag = createTraceTag(); logger.debug("{} DELETE uri = '{}{}''", traceTag, baseUrl, uri); - return getWebClient() // - .flatMap(client -> { - RequestHeadersSpec request = client.delete().uri(uri); - return retrieve(traceTag, request); - }); + RequestHeadersSpec request = getWebClient().delete().uri(uri); + return retrieve(traceTag, request); } public Mono delete(String uri) { return deleteForEntity(uri) // - .flatMap(this::toBody); + .map(this::toBody); } private Mono> retrieve(Object traceTag, RequestHeadersSpec request) { final Class clazz = String.class; return request.retrieve() // .toEntity(clazz) // - .doOnNext(entity -> logger.trace("{} Received: {}", traceTag, entity.getBody())) // + .doOnNext(entity -> logReceivedData(traceTag, entity)) // .doOnError(throwable -> onHttpError(traceTag, throwable)); } + private void logReceivedData(Object traceTag, ResponseEntity entity) { + logger.trace("{} Received: {} {}", traceTag, entity.getBody(), entity.getHeaders().getContentType()); + } + private static Object createTraceTag() { return sequenceNumber.incrementAndGet(); } @@ -199,136 +170,58 @@ public class AsyncRestClient { logger.debug("{} HTTP error status = '{}', body '{}'", traceTag, exception.getStatusCode(), exception.getResponseBodyAsString()); } else { - logger.debug("{} HTTP error", traceTag, t); + logger.debug("{} HTTP error {}", traceTag, t.getMessage()); } } - private Mono toBody(ResponseEntity entity) { + private String toBody(ResponseEntity entity) { if (entity.getBody() == null) { - return Mono.just(""); + return ""; } else { - return Mono.just(entity.getBody()); - } - } - - private boolean isCertificateEntry(KeyStore trustStore, String alias) { - try { - return trustStore.isCertificateEntry(alias); - } catch (KeyStoreException e) { - logger.error("Error reading truststore {}", e.getMessage()); - return false; + return entity.getBody(); } } - private Certificate getCertificate(KeyStore trustStore, String alias) { - try { - return trustStore.getCertificate(alias); - } catch (KeyStoreException e) { - logger.error("Error reading truststore {}", e.getMessage()); - return null; - } + private boolean isHttpProxyConfigured() { + return httpProxyConfig != null && httpProxyConfig.httpProxyPort() > 0 + && !httpProxyConfig.httpProxyHost().isEmpty(); } - private static synchronized KeyStore getTrustStore(String trustStorePath, String trustStorePass) - throws NoSuchAlgorithmException, CertificateException, IOException, KeyStoreException { - if (clientTrustStore == null) { - KeyStore store = KeyStore.getInstance(KeyStore.getDefaultType()); - store.load(new FileInputStream(ResourceUtils.getFile(trustStorePath)), trustStorePass.toCharArray()); - clientTrustStore = store; - } - return clientTrustStore; - } - - private SslContext createSslContextRejectingUntrustedPeers(String trustStorePath, String trustStorePass, - KeyManagerFactory keyManager) - throws NoSuchAlgorithmException, CertificateException, IOException, KeyStoreException { - - final KeyStore trustStore = getTrustStore(trustStorePath, trustStorePass); - List certificateList = Collections.list(trustStore.aliases()).stream() // - .filter(alias -> isCertificateEntry(trustStore, alias)) // - .map(alias -> getCertificate(trustStore, alias)) // - .collect(Collectors.toList()); - final X509Certificate[] certificates = certificateList.toArray(new X509Certificate[certificateList.size()]); - - return SslContextBuilder.forClient() // - .keyManager(keyManager) // - .trustManager(certificates) // - .build(); - } - - private SslContext createSslContext(KeyManagerFactory keyManager) - throws NoSuchAlgorithmException, CertificateException, KeyStoreException, IOException { - if (this.clientConfig.isTrustStoreUsed()) { - return createSslContextRejectingUntrustedPeers(this.clientConfig.trustStore(), - this.clientConfig.trustStorePassword(), keyManager); - } else { - // Trust anyone - return SslContextBuilder.forClient() // - .keyManager(keyManager) // - .trustManager(InsecureTrustManagerFactory.INSTANCE) // - .build(); - } - } - - private TcpClient createTcpClientSecure(SslContext sslContext) { - return TcpClient.create(ConnectionProvider.newConnection()) // + private HttpClient buildHttpClient() { + HttpClient httpClient = HttpClient.create() // .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10_000) // - .secure(c -> c.sslContext(sslContext)) // .doOnConnected(connection -> { connection.addHandlerLast(new ReadTimeoutHandler(30)); connection.addHandlerLast(new WriteTimeoutHandler(30)); }); - } - private TcpClient createTcpClientInsecure() { - return TcpClient.create(ConnectionProvider.newConnection()) // - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10_000) // - .doOnConnected(connection -> { - connection.addHandlerLast(new ReadTimeoutHandler(30)); - connection.addHandlerLast(new WriteTimeoutHandler(30)); - }); + if (this.sslContext != null) { + httpClient = httpClient.secure(ssl -> ssl.sslContext(sslContext)); + } + + if (isHttpProxyConfigured()) { + httpClient = httpClient.proxy(proxy -> proxy.type(ProxyProvider.Proxy.HTTP) + .host(httpProxyConfig.httpProxyHost()).port(httpProxyConfig.httpProxyPort())); + } + return httpClient; } - private WebClient createWebClient(String baseUrl, TcpClient tcpClient) { - HttpClient httpClient = HttpClient.from(tcpClient); - ReactorClientHttpConnector connector = new ReactorClientHttpConnector(httpClient); + private WebClient buildWebClient(String baseUrl) { + final HttpClient httpClient = buildHttpClient(); ExchangeStrategies exchangeStrategies = ExchangeStrategies.builder() // .codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(-1)) // .build(); return WebClient.builder() // - .clientConnector(connector) // + .clientConnector(new ReactorClientHttpConnector(httpClient)) // .baseUrl(baseUrl) // .exchangeStrategies(exchangeStrategies) // .build(); } - private Mono getWebClient() { + private WebClient getWebClient() { if (this.webClient == null) { - try { - if (this.sslEnabled) { - final KeyManagerFactory keyManager = - KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); - final KeyStore keyStore = KeyStore.getInstance(this.clientConfig.keyStoreType()); - final String keyStoreFile = this.clientConfig.keyStore(); - final String keyStorePassword = this.clientConfig.keyStorePassword(); - final String keyPassword = this.clientConfig.keyPassword(); - try (final InputStream inputStream = new FileInputStream(keyStoreFile)) { - keyStore.load(inputStream, keyStorePassword.toCharArray()); - } - keyManager.init(keyStore, keyPassword.toCharArray()); - SslContext sslContext = createSslContext(keyManager); - TcpClient tcpClient = createTcpClientSecure(sslContext); - this.webClient = createWebClient(this.baseUrl, tcpClient); - } else { - TcpClient tcpClient = createTcpClientInsecure(); - this.webClient = createWebClient(this.baseUrl, tcpClient); - } - } catch (Exception e) { - logger.error("Could not create WebClient {}", e.getMessage()); - return Mono.error(e); - } + this.webClient = buildWebClient(baseUrl); } - return Mono.just(this.webClient); + return this.webClient; } - }