* ========================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.
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;
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<ResponseEntity<String>> postForEntity(String uri, @Nullable String body) {
final Class<String> 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<String> entity) {
+ logger.trace("{} Received: {} {}", traceTag, entity.getBody(), entity.getHeaders().getContentType());
+ }
+
private static Object createTraceTag() {
return sequenceNumber.incrementAndGet();
}
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 boolean isCertificateEntry(KeyStore trustStore, String alias) {
- try {
- return trustStore.isCertificateEntry(alias);
- } catch (KeyStoreException e) {
- logger.error("Error reading truststore {}", e.getMessage());
- return false;
- }
- }
-
- 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 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 boolean isHttpProxyConfigured() {
+ return httpProxyConfig != null && httpProxyConfig.httpProxyPort() > 0
+ && !httpProxyConfig.httpProxyHost().isEmpty();
}
- private SslContext createSslContextRejectingUntrustedPeers(String trustStorePath, String trustStorePass,
- KeyManagerFactory keyManager)
- throws NoSuchAlgorithmException, CertificateException, IOException, KeyStoreException {
-
- final KeyStore trustStore = getTrustStore(trustStorePath, trustStorePass);
- List<Certificate> 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<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 Mono.just(buildWebClient(baseUrl));
}
}
Code Review / nonrtric.git / blobdiff