Last version from demo.
Change-Id: Ib41581bf6f9eb92a03edf8434261d3674b6e3e39
Issue-ID: DCAEGEN2-860
Signed-off-by: elinuxhenrik <henrik.b.andersson@est.tech>
Signed-off-by: Vijay Venkatesh Kumar <vv770d@att.com>
"dmaapPortNumber": 2222,
"dmaapTopicName": "/events/unauthenticated.VES_NOTIFICATION_OUTPUT",
"dmaapProtocol": "http",
- "dmaapUserName": "admin",
- "dmaapUserPassword": "admin",
+ "dmaapUserName": "",
+ "dmaapUserPassword": "",
"dmaapContentType": "application/json",
"consumerId": "C12",
"consumerGroup": "OpenDcae-c12",
"dmaapUserPassword": "dradmin",
"dmaapContentType": "application/octet-stream"
}
+ },
+ "ftp": {
+ "ftpesConfiguration": {
+ "keyCert": "config/ftpKey.jks",
+ "keyPassword": "secret",
+ "trustedCA": "config/cacerts",
+ "trustedCAPassword": "secret"
+ }
}
}
}
+
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
<parent>
<groupId>org.onap.dcaegen2.collectors</groupId>
<artifactId>datafile</artifactId>
<version>1.0.0-SNAPSHOT</version>
</parent>
- <modelVersion>4.0.0</modelVersion>
<groupId>org.onap.dcaegen2.collectors.datafile</groupId>
<artifactId>datafile-app-server</artifactId>
+ <version>1.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<properties>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
- <dependency>
- <groupId>commons-net</groupId>
- <artifactId>commons-net</artifactId>
- </dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<artifactId>testng</artifactId>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.mockftpserver</groupId>
- <artifactId>MockFtpServer</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>com.github.stefanbirkner</groupId>
- <artifactId>fake-sftp-server-rule</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>com.jcraft</groupId>
- <artifactId>jsch</artifactId>
- <version>0.1.53</version>
- </dependency>
<!--REQUIRED TO GENERATE DOCUMENTATION -->
<dependency>
@Value("${dmaap.dmaapProducerConfiguration.dmaapContentType:}")
public String producerDmaapContentType;
+ @Value("${ftp.ftpesConfiguration.keyCert:}")
+ public String keyCert;
+
+ @Value("${ftp.ftpesConfiguration.keyPassword:}")
+ public String keyPassword;
+
+ @Value("${ftp.ftpesConfiguration.trustedCA:}")
+ public String trustedCA;
+
+ @Value("${ftp.ftpesConfiguration.trustedCAPassword:}")
+ public String trustedCAPassword;
+
@Override
public DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
return new ImmutableDmaapConsumerConfiguration.Builder()
.build();
}
+ @Override
+ public FtpesConfig getFtpesConfiguration() {
+ return new ImmutableFtpesConfig.Builder()
+ .keyCert(
+ Optional.ofNullable(keyCert).filter(p -> !p.isEmpty())
+ .orElse(ftpesConfig.keyCert()))
+ .keyPassword(
+ Optional.ofNullable(keyPassword).filter(p -> !p.isEmpty())
+ .orElse(ftpesConfig.keyPassword()))
+ .trustedCA(
+ Optional.ofNullable(trustedCA).filter(p -> !p.isEmpty())
+ .orElse(ftpesConfig.trustedCA()))
+ .trustedCAPassword(
+ Optional.ofNullable(trustedCAPassword).filter(p -> !p.isEmpty())
+ .orElse(ftpesConfig.trustedCAPassword()))
+ .build();
+ }
}
package org.onap.dcaegen2.collectors.datafile.configuration;
import com.google.gson.JsonObject;
+
import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
import org.onap.dcaegen2.collectors.datafile.config.ImmutableDmaapConsumerConfiguration;
.consumerGroup(jsonObject.get("dmaap.dmaapConsumerConfiguration.consumerGroup").getAsString())
.build();
}
+
+ public FtpesConfig getFtpesConfig() {
+ return new ImmutableFtpesConfig.Builder()
+ .keyCert(jsonObject.get("dmaap.ftpesConfig.keyCert").getAsString())
+ .keyPassword(jsonObject.get("dmaap.ftpesConfig.keyPassword").getAsString())
+ .trustedCA(jsonObject.get("dmaap.ftpesConfig.trustedCA").getAsString())
+ .trustedCAPassword(jsonObject.get("dmaap.ftpesConfig.trustedCAPassword").getAsString())
+ .build();
+ }
}
\ No newline at end of file
DmaapPublisherConfiguration getDmaapPublisherConfiguration();
+ FtpesConfig getFtpesConfiguration();
+
void initFileStreamReader();
+
}
* ============LICENSE_START======================================================================
* Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
* ===============================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
* ============LICENSE_END========================================================================
*/
private static final String DMAAP = "dmaap";
private static final String DMAAP_PRODUCER = "dmaapProducerConfiguration";
private static final String DMAAP_CONSUMER = "dmaapConsumerConfiguration";
+ private static final String FTP = "ftp";
+ private static final String FTPES_CONFIGURATION = "ftpesConfiguration";
+
private static final Logger logger = LoggerFactory.getLogger(DatafileAppConfig.class);
DmaapPublisherConfiguration dmaapPublisherConfiguration;
+ FtpesConfig ftpesConfig;
+
@NotEmpty
private String filepath;
return dmaapPublisherConfiguration;
}
+ @Override
+ public FtpesConfig getFtpesConfiguration() {
+ return ftpesConfig;
+ }
+
@Override
public void initFileStreamReader() {
JsonElement rootElement = getJsonElement(parser, inputStream);
if (rootElement.isJsonObject()) {
jsonObject = rootElement.getAsJsonObject();
+ ftpesConfig = deserializeType(gsonBuilder,
+ jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(FTP).getAsJsonObject(FTPES_CONFIGURATION),
+ FtpesConfig.class);
dmaapConsumerConfiguration = deserializeType(gsonBuilder,
- jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_CONSUMER),
- DmaapConsumerConfiguration.class);
+ jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_CONSUMER),
+ DmaapConsumerConfiguration.class);
dmaapPublisherConfiguration = deserializeType(gsonBuilder,
- jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_PRODUCER),
- DmaapPublisherConfiguration.class);
+ jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_PRODUCER),
+ DmaapPublisherConfiguration.class);
}
} catch (IOException e) {
logger.error("Problem with file loading, file: {}", filepath, e);
}
private <T> T deserializeType(@NotNull GsonBuilder gsonBuilder, @NotNull JsonObject jsonObject,
- @NotNull Class<T> type) {
+ @NotNull Class<T> type) {
return gsonBuilder.create().fromJson(jsonObject, type);
}
this.filepath = filepath;
}
-}
\ No newline at end of file
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcaegen2.collectors.datafile.configuration;
+
+/**
+ * @author
+ *
+ */
+import java.io.Serializable;
+
+import org.immutables.gson.Gson;
+import org.immutables.value.Value;
+import org.springframework.stereotype.Component;
+
+
+@Component
+@Value.Immutable
+@Value.Style(builder = "new")
+@Gson.TypeAdapters
+public abstract class FtpesConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ @Value.Parameter
+ public abstract String keyCert();
+
+ @Value.Parameter
+ public abstract String keyPassword();
+
+ @Value.Parameter
+ public abstract String trustedCA();
+
+ @Value.Parameter
+ public abstract String trustedCAPassword();
+}
\ No newline at end of file
* @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
*/
@RestController
-@Api(value = "HeartbeatController", description = "Check liveness of DATAFILE service")
+@Api(value = "HeartbeatController")
public class HeartbeatController {
private static final Logger logger = LoggerFactory.getLogger(HeartbeatController.class);
*/
public class DatafileTaskException extends Exception {
+ private static final long serialVersionUID = 1L;
+
public DatafileTaskException() {
super();
}
*/
public class DmaapEmptyResponseException extends DatafileTaskException {
+ private static final long serialVersionUID = 1L;
+
public DmaapEmptyResponseException() {
super();
}
*/
public class DmaapNotFoundException extends DatafileTaskException {
+ private static final long serialVersionUID = 1L;
+
public DmaapNotFoundException(String message) {
super(message);
}
*/
public class EnvironmentLoaderException extends Exception {
+ private static final long serialVersionUID = 1L;
+
public EnvironmentLoaderException(String message) {
super(message);
}
+++ /dev/null
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2018 Nordix Foundation. All rights reserved.
- * ===============================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
- * in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License
- * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
- * or implied. See the License for the specific language governing permissions and limitations under
- * the License.
- * ============LICENSE_END========================================================================
- */
-
-package org.onap.dcaegen2.collectors.datafile.ftp;
-
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-
-import org.apache.commons.net.ftp.FTPReply;
-import org.apache.commons.net.ftp.FTPSClient;
-import org.apache.commons.net.util.TrustManagerUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Component;
-
-/**
- * Gets file from xNF with FTPS protocol.
- *
- * TODO: Refactor for better test.
- *
- * @author <a href="mailto:martin.c.yan@est.tech">Martin Yan</a>
- *
- */
-@Component
-public class FtpsClient { // TODO: Should be final but needs PowerMock or Mockito 2.x to be able to
- // mock then, so this will be done as an improvement after first version
- // committed.
- private static final Logger logger = LoggerFactory.getLogger(FtpsClient.class);
-
- public boolean collectFile(FileServerData fileServerData, String remoteFile, String localFile) {
- logger.trace("collectFile called with fileServerData: {}, remoteFile: {}, localFile: {}", fileServerData,
- remoteFile, localFile);
- boolean result = true;
- try {
- FTPSClient ftps = new FTPSClient("TLS");
-
- result = setUpConnection(fileServerData, ftps);
-
- if (result) {
- getFile(remoteFile, localFile, ftps);
-
- closeDownConnection(ftps);
- }
- } catch (IOException ex) {
- logger.error("Unable to collect file from xNF. Data: {}", fileServerData, ex);
- result = false;
- }
- logger.trace("collectFile left with result: {}", result);
- return result;
- }
-
- private boolean setUpConnection(FileServerData fileServerData, FTPSClient ftps) {
- boolean success = true;
- ftps.setTrustManager(TrustManagerUtils.getAcceptAllTrustManager());
-
- try {
- ftps.connect(fileServerData.serverAddress(), fileServerData.port());
-
- if (!ftps.login(fileServerData.userId(), fileServerData.password())) {
- ftps.logout();
- logger.error("Unable to log in to xNF. {}", fileServerData);
- success = false;
- }
-
- if (success) {
- int reply = ftps.getReplyCode();
- if (!FTPReply.isPositiveCompletion(reply)) {
- ftps.disconnect();
- logger.error("Unable to connect in to xNF. {}", fileServerData);
- success = false;
- }
- ftps.enterLocalPassiveMode();
- }
- } catch (Exception ex) {
- logger.error("Unable to connect to xNF. Data: {}", fileServerData, ex);
- success = false;
- }
-
- return success;
- }
-
- private void getFile(String remoteFile, String localFile, FTPSClient ftps)
- throws IOException {
- OutputStream output;
- File outfile = new File(localFile);
- outfile.createNewFile();
-
- output = new FileOutputStream(outfile);
-
- ftps.retrieveFile(remoteFile, output);
-
- output.close();
- logger.debug("File {} Download Successfull from xNF", outfile.getName());
- }
-
- private void closeDownConnection(FTPSClient ftps) {
- try {
- ftps.logout();
- ftps.disconnect();
- } catch (Exception e) {
- // Do nothing, file has been collected.
- }
- }
-}
* @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
*/
public class DmaapConsumerJsonParser {
-
private static final Logger logger = LoggerFactory.getLogger(DmaapConsumerJsonParser.class);
private static final String EVENT = "event";
}
private Mono<JsonElement> getJsonParserMessage(String message) {
+ logger.trace("original message from message router: {}", message);
return StringUtils.isEmpty(message) ? Mono.error(new DmaapEmptyResponseException())
: Mono.fromSupplier(() -> new JsonParser().parse(message));
}
}
public Optional<JsonObject> getJsonObjectFromAnArray(JsonElement element) {
+ logger.trace("starting to getJsonObjectFromAnArray!");
+
return Optional.of(new JsonParser().parse(element.getAsString()).getAsJsonObject());
}
String changeType = getValueFromJson(notificationFields, CHANGE_TYPE);
String notificationFieldsVersion = getValueFromJson(notificationFields, NOTIFICATION_FIELDS_VERSION);
JsonArray arrayOfNamedHashMap = notificationFields.getAsJsonArray(ARRAY_OF_NAMED_HASH_MAP);
-
if (isNotificationFieldsHeaderNotEmpty(changeIdentifier, changeType, notificationFieldsVersion)
&& arrayOfNamedHashMap != null) {
return getAllFileDataFromJson(changeIdentifier, changeType, arrayOfNamedHashMap);
}
return Flux.error(
new DmaapNotFoundException("FileReady event has incorrect JsonObject - missing header. " + jsonObject));
-
}
private Flux<FileData> getAllFileDataFromJson(String changeIdentifier, String changeType,
}
private FileData getFileDataFromJson(JsonObject fileInfo, String changeIdentifier, String changeType) {
+ logger.trace("starting to getFileDataFromJson!");
+
FileData fileData = null;
String name = getValueFromJson(fileInfo, NAME);
@Override
Flux<FileData> consume(Mono<String> message) {
- logger.trace("consume called with arg {}", message.toString());
+ logger.trace("consume called with arg {}", message);
return dmaapConsumerJsonParser.getJsonObject(message);
}
package org.onap.dcaegen2.collectors.datafile.tasks;
+import org.onap.dcaegen2.collectors.datafile.configuration.FtpesConfig;
import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
import org.onap.dcaegen2.collectors.datafile.model.FileData;
* @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
*/
public interface XnfCollectorTask {
+ abstract FtpesConfig resolveConfiguration();
Flux<ConsumerDmaapModel> execute(FileData fileData);
}
import java.io.File;
import java.net.URI;
+import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
+import org.onap.dcaegen2.collectors.datafile.configuration.Config;
+import org.onap.dcaegen2.collectors.datafile.configuration.FtpesConfig;
import org.onap.dcaegen2.collectors.datafile.ftp.FileServerData;
import org.onap.dcaegen2.collectors.datafile.ftp.FtpsClient;
import org.onap.dcaegen2.collectors.datafile.ftp.ImmutableFileServerData;
private static final String SFTP = "sftp";
private static final Logger logger = LoggerFactory.getLogger(XnfCollectorTaskImpl.class);
+ private Config datafileAppConfig;
private final FtpsClient ftpsClient;
private final SftpClient sftpClient;
@Autowired
- protected XnfCollectorTaskImpl(FtpsClient ftpsCleint, SftpClient sftpClient) {
+ protected XnfCollectorTaskImpl(AppConfig datafileAppConfig, FtpsClient ftpsCleint, SftpClient sftpClient) {
+ this.datafileAppConfig = datafileAppConfig;
this.ftpsClient = ftpsCleint;
this.sftpClient = sftpClient;
}
@Override
public Flux<ConsumerDmaapModel> execute(FileData fileData) {
logger.trace("Entering execute with {}", fileData);
+ resolveKeyStore();
+
String localFile = collectFile(fileData);
if (localFile != null) {
return Flux.empty();
}
+ @Override
+ public FtpesConfig resolveConfiguration() {
+ return datafileAppConfig.getFtpesConfiguration();
+ }
+
+ private void resolveKeyStore() {
+ FtpesConfig ftpesConfig = resolveConfiguration();
+ ftpsClient.setKeyCertPath(ftpesConfig.keyCert());
+ ftpsClient.setKeyCertPassword(ftpesConfig.keyPassword());
+ ftpsClient.setTrustedCAPath(ftpesConfig.trustedCA());
+ ftpsClient.setTrustedCAPassword(ftpesConfig.trustedCAPassword());
+ }
+
private String collectFile(FileData fileData) {
+ logger.trace("starting to collectFile");
String location = fileData.location();
URI uri = URI.create(location);
- String[] userInfo = getUserNameAndPasswordIfGiven(uri.getUserInfo());
- FileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(uri.getHost())
- .userId(userInfo != null ? userInfo[0] : "").password(userInfo != null ? userInfo[1] : "")
- .port(uri.getPort()).build();
+ FileServerData fileServerData = getFileServerData(uri);
String remoteFile = uri.getPath();
String localFile = "target" + File.separator + fileData.name();
String scheme = uri.getScheme();
-
boolean fileDownloaded = false;
if (FTPES.equals(scheme) || FTPS.equals(scheme)) {
fileDownloaded = ftpsClient.collectFile(fileServerData, remoteFile, localFile);
return localFile;
}
+ private FileServerData getFileServerData(URI uri) {
+ String[] userInfo = getUserNameAndPasswordIfGiven(uri.getUserInfo());
+ return ImmutableFileServerData.builder().serverAddress(uri.getHost())
+ .userId(userInfo != null ? userInfo[0] : "").password(userInfo != null ? userInfo[1] : "")
+ .port(uri.getPort()).build();
+ }
+
private String[] getUserNameAndPasswordIfGiven(String userInfoString) {
String[] userInfo = null;
if (userInfoString != null && !userInfoString.isEmpty()) {
/*
- * ============LICENSE_START=======================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
*/
package org.onap.dcaegen2.collectors.datafile.configuration;
-
import static org.assertj.core.api.Assertions.assertThat;
-import com.google.gson.Gson;
import com.google.gson.JsonObject;
+
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
class CloudConfigParserTest {
- private static final String CORRECT_JSON =
- "{\"dmaap.dmaapProducerConfiguration.dmaapTopicName\": \"/events/unauthenticated.VES_NOTIFICATION_OUTPUT\", "
- + "\"dmaap.dmaapConsumerConfiguration.timeoutMS\": -1,"
- + " \"dmaap.dmaapConsumerConfiguration.dmaapHostName\": \"message-router.onap.svc.cluster.local\","
- + "\"dmaap.dmaapConsumerConfiguration.dmaapUserName\": \"admin\", "
- + "\"dmaap.dmaapProducerConfiguration.dmaapPortNumber\": 3904, "
- + "\"dmaap.dmaapConsumerConfiguration.dmaapUserPassword\": \"admin\", "
- + "\"dmaap.dmaapProducerConfiguration.dmaapProtocol\": \"http\", "
- + "\"dmaap.dmaapProducerConfiguration.dmaapContentType\": \"application/json\", "
- + "\"dmaap.dmaapConsumerConfiguration.dmaapTopicName\": \"/events/unauthenticated.VES_NOTIFICATION_OUTPUT\", "
- + "\"dmaap.dmaapConsumerConfiguration.dmaapPortNumber\": 3904, "
- + "\"dmaap.dmaapConsumerConfiguration.dmaapContentType\": \"application/json\", "
- + "\"dmaap.dmaapConsumerConfiguration.messageLimit\": -1, "
- + "\"dmaap.dmaapConsumerConfiguration.dmaapProtocol\": \"http\", "
- + "\"dmaap.dmaapConsumerConfiguration.consumerId\": \"c12\","
- + "\"dmaap.dmaapProducerConfiguration.dmaapHostName\": \"message-router.onap.svc.cluster.local\", "
- + "\"dmaap.dmaapConsumerConfiguration.consumerGroup\": \"OpenDCAE-c12\", "
- + "\"dmaap.dmaapProducerConfiguration.dmaapUserName\": \"admin\", "
- + "\"dmaap.dmaapProducerConfiguration.dmaapUserPassword\": \"admin\"}";
-
- private static final ImmutableDmaapConsumerConfiguration correctDmaapConsumerConfig =
- new ImmutableDmaapConsumerConfiguration.Builder()
- .timeoutMS(-1)
- .dmaapHostName("message-router.onap.svc.cluster.local")
- .dmaapUserName("admin")
- .dmaapUserPassword("admin")
- .dmaapTopicName("/events/unauthenticated.VES_NOTIFICATION_OUTPUT")
- .dmaapPortNumber(3904)
- .dmaapContentType("application/json")
- .messageLimit(-1)
- .dmaapProtocol("http")
- .consumerId("c12")
- .consumerGroup("OpenDCAE-c12")
- .build();
-
- private static final ImmutableDmaapPublisherConfiguration correctDmaapPublisherConfig =
- new ImmutableDmaapPublisherConfiguration.Builder()
- .dmaapTopicName("/events/unauthenticated.VES_NOTIFICATION_OUTPUT")
- .dmaapUserPassword("admin")
- .dmaapPortNumber(3904)
- .dmaapProtocol("http")
- .dmaapContentType("application/json")
- .dmaapHostName("message-router.onap.svc.cluster.local")
- .dmaapUserName("admin")
- .build();
-
- private CloudConfigParser cloudConfigParser = new CloudConfigParser(
- new Gson().fromJson(CORRECT_JSON, JsonObject.class));
+ private static final ImmutableDmaapConsumerConfiguration CORRECT_DMAAP_CONSUMER_CONFIG =
+ new ImmutableDmaapConsumerConfiguration.Builder().timeoutMS(-1)
+ .dmaapHostName("message-router.onap.svc.cluster.local").dmaapUserName("admin")
+ .dmaapUserPassword("admin").dmaapTopicName("/events/unauthenticated.VES_NOTIFICATION_OUTPUT")
+ .dmaapPortNumber(2222).dmaapContentType("application/json").messageLimit(-1).dmaapProtocol("http")
+ .consumerId("C12").consumerGroup("OpenDCAE-c12").build();
+
+ private static final ImmutableDmaapPublisherConfiguration CORRECT_DMAAP_PUBLISHER_CONFIG =
+ new ImmutableDmaapPublisherConfiguration.Builder().dmaapTopicName("publish").dmaapUserPassword("dradmin")
+ .dmaapPortNumber(3907).dmaapProtocol("https").dmaapContentType("application/json")
+ .dmaapHostName("message-router.onap.svc.cluster.local").dmaapUserName("dradmin").build();
+ private static final ImmutableFtpesConfig CORRECT_FTPES_CONFIGURATION =
+ new ImmutableFtpesConfig.Builder().keyCert("/config/ftpKey.jks").keyPassword("secret")
+ .trustedCA("config/cacerts").trustedCAPassword("secret").build();
+
+ private CloudConfigParser cloudConfigParser = new CloudConfigParser(getCloudConfigJsonObject());
@Test
public void shouldCreateDmaapConsumerConfigurationCorrectly() {
- // when
DmaapConsumerConfiguration dmaapConsumerConfig = cloudConfigParser.getDmaapConsumerConfig();
- // then
assertThat(dmaapConsumerConfig).isNotNull();
- assertThat(dmaapConsumerConfig).isEqualToComparingFieldByField(correctDmaapConsumerConfig);
+ assertThat(dmaapConsumerConfig).isEqualToComparingFieldByField(CORRECT_DMAAP_CONSUMER_CONFIG);
}
-
@Test
public void shouldCreateDmaapPublisherConfigurationCorrectly() {
- // when
DmaapPublisherConfiguration dmaapPublisherConfig = cloudConfigParser.getDmaapPublisherConfig();
- // then
assertThat(dmaapPublisherConfig).isNotNull();
- assertThat(dmaapPublisherConfig).isEqualToComparingFieldByField(correctDmaapPublisherConfig);
+ assertThat(dmaapPublisherConfig).isEqualToComparingFieldByField(CORRECT_DMAAP_PUBLISHER_CONFIG);
+ }
+
+ @Test
+ public void shouldCreateFtpesConfigurationCorrectly() {
+ FtpesConfig ftpesConfig = cloudConfigParser.getFtpesConfig();
+
+ assertThat(ftpesConfig).isNotNull();
+ assertThat(ftpesConfig).isEqualToComparingFieldByField(CORRECT_FTPES_CONFIGURATION);
+ }
+
+ public JsonObject getCloudConfigJsonObject() {
+ JsonObject config = new JsonObject();
+ config.addProperty("dmaap.dmaapConsumerConfiguration.timeoutMS", -1);
+ config.addProperty("dmaap.dmaapConsumerConfiguration.dmaapHostName", "message-router.onap.svc.cluster.local");
+ config.addProperty("dmaap.dmaapConsumerConfiguration.dmaapUserName", "admin");
+ config.addProperty("dmaap.dmaapConsumerConfiguration.dmaapUserPassword", "admin");
+ config.addProperty("dmaap.dmaapConsumerConfiguration.dmaapTopicName",
+ "/events/unauthenticated.VES_NOTIFICATION_OUTPUT");
+ config.addProperty("dmaap.dmaapConsumerConfiguration.dmaapPortNumber", 2222);
+ config.addProperty("dmaap.dmaapConsumerConfiguration.dmaapContentType", "application/json");
+ config.addProperty("dmaap.dmaapConsumerConfiguration.messageLimit", -1);
+ config.addProperty("dmaap.dmaapConsumerConfiguration.dmaapProtocol", "http");
+ config.addProperty("dmaap.dmaapConsumerConfiguration.consumerId", "C12");
+ config.addProperty("dmaap.dmaapConsumerConfiguration.consumerGroup", "OpenDCAE-c12");
+ config.addProperty("dmaap.dmaapProducerConfiguration.dmaapTopicName", "publish");
+ config.addProperty("dmaap.dmaapProducerConfiguration.dmaapProtocol", "https");
+ config.addProperty("dmaap.dmaapProducerConfiguration.dmaapContentType", "application/json");
+ config.addProperty("dmaap.dmaapProducerConfiguration.dmaapHostName", "message-router.onap.svc.cluster.local");
+ config.addProperty("dmaap.dmaapProducerConfiguration.dmaapPortNumber", 3907);
+ config.addProperty("dmaap.dmaapProducerConfiguration.dmaapUserName", "dradmin");
+ config.addProperty("dmaap.dmaapProducerConfiguration.dmaapUserPassword", "dradmin");
+ config.addProperty("dmaap.ftpesConfig.keyCert", "/config/ftpKey.jks");
+ config.addProperty("dmaap.ftpesConfig.keyPassword", "secret");
+ config.addProperty("dmaap.ftpesConfig.trustedCA", "config/cacerts");
+ config.addProperty("dmaap.ftpesConfig.trustedCAPassword", "secret");
+
+ return config;
}
-}
\ No newline at end of file
+}
import static org.mockito.Mockito.when;
import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import java.io.ByteArrayInputStream;
class DatafileAppConfigTest {
private static final String DATAFILE_ENDPOINTS = "datafile_endpoints.json";
- private static final String JSON_STRING =
- "{\"configs\":{\"dmaap\":{\"dmaapConsumerConfiguration\":{\"consumerGroup\":\"other\",\"consumerId\":\"1\","
- + "\"dmaapContentType\":\"application/json\",\"dmaapHostName\":\"localhost\","
- + "\"dmaapPortNumber\":2222,\"dmaapProtocol\":\"http\",\"dmaapTopicName\":\"temp\","
- + "\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\",\"messageLimit\":1000,"
- + "\"timeoutMS\":1000},\"dmaapProducerConfiguration\":{\"dmaapContentType\":\"application/json\","
- + "\"dmaapHostName\":\"localhost\",\"dmaapPortNumber\":2223,\"dmaapProtocol\":\"http\","
- + "\"dmaapTopicName\":\"temp\",\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\"}}}}";
- private static final String INCORRECT_JSON_STRING =
- "{\"configs\":{\"dmaap\":{\"dmaapConsumerConfiguration\":{\"consumerGroup\":\"other\",\"consumerId\":\"1\","
- + "\"dmaapContentType\":\"application/json\",\"dmaapHostName\":\"localhost\","
- + "\"dmaapPortNumber\":2222,\"dmaapProtocol\":\"http\",\"dmaapTopicName\":\"temp\","
- + "\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\",\"messageLimit\":1000,"
- + "\"timeoutMS\":1000},\"dmaapProducerConfiguration\":{\"dmaapContentType\":\"application/json\","
- + "\"dmaapHostName\":\"localhost\",\"dmaapPortNumber\":2223,\"dmaapProtocol\":\"http\","
- + "\"FAULTY_PARAMETER_NAME\":\"temp\","
- + "\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\"}}}}";
+ private static final boolean CORRECT_JSON = true;
+ private static final boolean INCORRECT_JSON = false;
private static DatafileAppConfig datafileAppConfig;
private static AppConfig appConfig;
}
@Test
- public void whenTheConfigurationFits_GetAaiAndDmaapObjectRepresentationConfiguration() throws IOException {
+ public void whenTheConfigurationFits_GetFtpsAndDmaapObjectRepresentationConfiguration() throws IOException {
// Given
- InputStream inputStream = new ByteArrayInputStream((JSON_STRING.getBytes(StandardCharsets.UTF_8)));
+ InputStream inputStream =
+ new ByteArrayInputStream((getJsonConfig(CORRECT_JSON).getBytes(StandardCharsets.UTF_8)));
// When
datafileAppConfig.setFilepath(filePath);
datafileAppConfig.initFileStreamReader();
appConfig.dmaapConsumerConfiguration = datafileAppConfig.getDmaapConsumerConfiguration();
appConfig.dmaapPublisherConfiguration = datafileAppConfig.getDmaapPublisherConfiguration();
+ appConfig.ftpesConfig = datafileAppConfig.getFtpesConfiguration();
// Then
verify(datafileAppConfig, times(1)).setFilepath(anyString());
datafileAppConfig.getDmaapPublisherConfiguration());
Assertions.assertEquals(appConfig.getDmaapConsumerConfiguration(),
datafileAppConfig.getDmaapConsumerConfiguration());
+ Assertions.assertEquals(appConfig.getFtpesConfiguration(), datafileAppConfig.getFtpesConfiguration());
}
verify(datafileAppConfig, times(1)).initFileStreamReader();
Assertions.assertNull(datafileAppConfig.getDmaapConsumerConfiguration());
Assertions.assertNull(datafileAppConfig.getDmaapPublisherConfiguration());
+ Assertions.assertNull(datafileAppConfig.getFtpesConfiguration());
}
@Test
public void whenFileIsExistsButJsonIsIncorrect() throws IOException {
// Given
- InputStream inputStream = new ByteArrayInputStream((INCORRECT_JSON_STRING.getBytes(StandardCharsets.UTF_8)));
+ InputStream inputStream =
+ new ByteArrayInputStream((getJsonConfig(INCORRECT_JSON).getBytes(StandardCharsets.UTF_8)));
// When
datafileAppConfig.setFilepath(filePath);
verify(datafileAppConfig, times(1)).initFileStreamReader();
Assertions.assertNotNull(datafileAppConfig.getDmaapConsumerConfiguration());
Assertions.assertNull(datafileAppConfig.getDmaapPublisherConfiguration());
+ Assertions.assertNotNull(datafileAppConfig.getFtpesConfiguration());
}
@Test
public void whenTheConfigurationFits_ButRootElementIsNotAJsonObject() throws IOException {
// Given
- InputStream inputStream = new ByteArrayInputStream((JSON_STRING.getBytes(StandardCharsets.UTF_8)));
+ InputStream inputStream =
+ new ByteArrayInputStream((getJsonConfig(CORRECT_JSON).getBytes(StandardCharsets.UTF_8)));
// When
datafileAppConfig.setFilepath(filePath);
doReturn(inputStream).when(datafileAppConfig).getInputStream(any());
datafileAppConfig.initFileStreamReader();
appConfig.dmaapConsumerConfiguration = datafileAppConfig.getDmaapConsumerConfiguration();
appConfig.dmaapPublisherConfiguration = datafileAppConfig.getDmaapPublisherConfiguration();
+ appConfig.ftpesConfig = datafileAppConfig.getFtpesConfiguration();
// Then
verify(datafileAppConfig, times(1)).setFilepath(anyString());
verify(datafileAppConfig, times(1)).initFileStreamReader();
Assertions.assertNull(datafileAppConfig.getDmaapConsumerConfiguration());
Assertions.assertNull(datafileAppConfig.getDmaapPublisherConfiguration());
+ Assertions.assertNull(datafileAppConfig.getFtpesConfiguration());
+ }
+
+ private String getJsonConfig(boolean correct) {
+ JsonObject dmaapConsumerConfigData = new JsonObject();
+ dmaapConsumerConfigData.addProperty("dmaapHostName", "localhost");
+ dmaapConsumerConfigData.addProperty("dmaapPortNumber", 2222);
+ dmaapConsumerConfigData.addProperty("dmaapTopicName", "/events/unauthenticated.VES_NOTIFICATION_OUTPUT");
+ dmaapConsumerConfigData.addProperty("dmaapProtocol", "http");
+ dmaapConsumerConfigData.addProperty("dmaapUserName", "admin");
+ dmaapConsumerConfigData.addProperty("dmaapUserPassword", "admin");
+ dmaapConsumerConfigData.addProperty("dmaapContentType", "application/json");
+ dmaapConsumerConfigData.addProperty("consumerId", "C12");
+ dmaapConsumerConfigData.addProperty("consumerGroup", "OpenDcae-c12");
+ dmaapConsumerConfigData.addProperty("timeoutMS", -1);
+ dmaapConsumerConfigData.addProperty("messageLimit", 1);
+
+ JsonObject dmaapProducerConfigData = new JsonObject();
+ dmaapProducerConfigData.addProperty("dmaapHostName", "localhost");
+ dmaapProducerConfigData.addProperty("dmaapPortNumber", 3907);
+ dmaapProducerConfigData.addProperty("dmaapTopicName", "publish");
+ dmaapProducerConfigData.addProperty("dmaapProtocol", "https");
+ if (correct) {
+ dmaapProducerConfigData.addProperty("dmaapUserName", "dradmin");
+ dmaapProducerConfigData.addProperty("dmaapUserPassword", "dradmin");
+ dmaapProducerConfigData.addProperty("dmaapContentType", "application/octet-stream");
+ }
+
+ JsonObject dmaapConfigs = new JsonObject();
+ dmaapConfigs.add("dmaapConsumerConfiguration", dmaapConsumerConfigData);
+ dmaapConfigs.add("dmaapProducerConfiguration", dmaapProducerConfigData);
+
+ JsonObject ftpesConfigData = new JsonObject();
+ ftpesConfigData.addProperty("keyCert", "config/ftpKey.jks");
+ ftpesConfigData.addProperty("keyPassword", "secret");
+ ftpesConfigData.addProperty("trustedCA", "config/cacerts");
+ ftpesConfigData.addProperty("trustedCAPassword", "secret");
+
+ JsonObject ftpesConfiguration = new JsonObject();
+ ftpesConfiguration.add("ftpesConfiguration", ftpesConfigData);
+
+ JsonObject configs = new JsonObject();
+ configs.add("dmaap", dmaapConfigs);
+ configs.add("ftp", ftpesConfiguration);
+
+ JsonObject completeJson = new JsonObject();
+ completeJson.add("configs", configs);
+
+ return completeJson.toString();
}
}
+++ /dev/null
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2018 Nordix Foundation. All rights reserved.
- * ===============================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
- * in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License
- * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
- * or implied. See the License for the specific language governing permissions and limitations under
- * the License.
- * ============LICENSE_END========================================================================
- */
-
-package org.onap.dcaegen2.collectors.datafile.ftp;
-
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.mockftpserver.fake.FakeFtpServer;
-import org.mockftpserver.fake.UserAccount;
-import org.mockftpserver.fake.filesystem.FileEntry;
-import org.mockftpserver.fake.filesystem.FileSystem;
-import org.mockftpserver.fake.filesystem.UnixFakeFileSystem;
-
-public class FtpClientTest {
-
- private static final String HOME_DIR = "/";
- private static final String FILE = "/dir/sample.txt";
- private static final String CONTENTS = "abcdef 1234567890";
- private static final int PORT = 8021;
-
- private static final String USERNAME = "bob";
- private static final String PASSWORD = "123";
-
- private FakeFtpServer fakeFtpServer;
-
- @BeforeAll
- protected void setUp() throws Exception {
- fakeFtpServer = new FakeFtpServer();
- fakeFtpServer.setServerControlPort(PORT);
-
- FileSystem fileSystem = new UnixFakeFileSystem();
- fileSystem.add(new FileEntry(FILE, CONTENTS));
- fakeFtpServer.setFileSystem(fileSystem);
- UserAccount userAccount = new UserAccount(USERNAME, PASSWORD, HOME_DIR);
- fakeFtpServer.addUserAccount(userAccount);
-
- fakeFtpServer.start();
- }
-
- @AfterAll
- protected void tearDown() throws Exception {
- fakeFtpServer.stop();
- }
-}
import java.io.File;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
+import org.onap.dcaegen2.collectors.datafile.configuration.FtpesConfig;
import org.onap.dcaegen2.collectors.datafile.ftp.FileServerData;
import org.onap.dcaegen2.collectors.datafile.ftp.FtpsClient;
import org.onap.dcaegen2.collectors.datafile.ftp.ImmutableFileServerData;
private static final String MEAS_COLLECT_FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
private static final String FILE_FORMAT_VERSION = "V10";
+ private static final String FTP_KEY_PATH = "ftpKeyPath";
+ private static final String FTP_KEY_PASSWORD = "ftpKeyPassword";
+ private static final String TRUSTED_CA_PATH = "trustedCAPath";
+ private static final String TRUSTED_CA_PASSWORD = "trustedCAPassword";
+
+ private static AppConfig appConfigMock = mock(AppConfig.class);
+ private static FtpesConfig ftpesConfigMock = mock(FtpesConfig.class);
+
private FtpsClient ftpsClientMock = mock(FtpsClient.class);
private SftpClient sftpClientMock = mock(SftpClient.class);
- private XnfCollectorTask collectorUndetTest = new XnfCollectorTaskImpl(ftpsClientMock, sftpClientMock);
+ private XnfCollectorTask collectorUndetTest = new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
+
+ @BeforeAll
+ public static void setUpConfiguration() {
+ when(appConfigMock.getFtpesConfiguration()).thenReturn(ftpesConfigMock);
+ when(ftpesConfigMock.keyCert()).thenReturn(FTP_KEY_PATH);
+ when(ftpesConfigMock.keyPassword()).thenReturn(FTP_KEY_PASSWORD);
+ when(ftpesConfigMock.trustedCA()).thenReturn(TRUSTED_CA_PATH);
+ when(ftpesConfigMock.trustedCAPassword()).thenReturn(TRUSTED_CA_PASSWORD);
+ }
@Test
- public void whenSingleFtpesFile_returnCorrectResponse() {
+ public void whenFtpesFile_returnCorrectResponse() {
FileData fileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
.changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(FTPES_LOCATION)
.compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
.verifyComplete();
verify(ftpsClientMock, times(1)).collectFile(fileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
+ verify(ftpsClientMock).setKeyCertPath(FTP_KEY_PATH);
+ verify(ftpsClientMock).setKeyCertPassword(FTP_KEY_PASSWORD);
+ verify(ftpsClientMock).setTrustedCAPath(TRUSTED_CA_PATH);
+ verify(ftpsClientMock).setTrustedCAPassword(TRUSTED_CA_PASSWORD);
verifyNoMoreInteractions(ftpsClientMock);
}
@Test
- public void whenSingleSftpFile_returnCorrectResponse() {
+ public void whenSftpFile_returnCorrectResponse() {
FileData fileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
.changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(SFTP_LOCATION)
.compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
.verifyComplete();
verify(sftpClientMock, times(1)).collectFile(fileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
+ verify(ftpsClientMock).setKeyCertPath(FTP_KEY_PATH);
+ verify(ftpsClientMock).setKeyCertPassword(FTP_KEY_PASSWORD);
+ verify(ftpsClientMock).setTrustedCAPath(TRUSTED_CA_PATH);
+ verify(ftpsClientMock).setTrustedCAPassword(TRUSTED_CA_PASSWORD);
verifyNoMoreInteractions(ftpsClientMock);
}
StepVerifier.create(collectorUndetTest.execute(fileData)).expectNextCount(0).verifyComplete();
+ verify(ftpsClientMock).setKeyCertPath(FTP_KEY_PATH);
+ verify(ftpsClientMock).setKeyCertPassword(FTP_KEY_PASSWORD);
+ verify(ftpsClientMock).setTrustedCAPath(TRUSTED_CA_PATH);
+ verify(ftpsClientMock).setTrustedCAPassword(TRUSTED_CA_PASSWORD);
verifyNoMoreInteractions(ftpsClientMock);
}
}
"configs": {
"dmaap": {
"dmaapConsumerConfiguration": {
- "consumerGroup": "notification",
"consumerId": "C12",
- "dmaapContentType": "application/json",
"dmaapHostName": "localhost",
"dmaapPortNumber": 2222,
- "dmaapProtocol": "http",
"dmaapTopicName": "/events/unauthenticated.VES_NOTIFICATION_OUTPUT",
+ "dmaapProtocol": "http",
"dmaapUserName": "admin",
"dmaapUserPassword": "admin",
- "messageLimit": 1000,
- "timeoutMS": 1000
+ "dmaapContentType": "application/json",
+ "consumerId": "C12",
+ "consumerGroup": "OpenDcae-c12",
+ "timeoutMS": -1,
+ "messageLimit": 1
},
"dmaapProducerConfiguration": {
- "dmaapContentType": "application/octet-stream",
"dmaapHostName": "localhost",
"dmaapPortNumber": 3907,
"dmaapProtocol": "https",
"dmaapTopicName": "publish",
"dmaapUserName": "dradmin",
- "dmaapUserPassword": "dradmin"
+ "dmaapUserPassword": "dradmin",
+ "dmaapContentType": "application/octet-stream"
+ }
+ },
+ "ftp": {
+ "ftpesConfiguration": {
+ "keyCert": "/config/ftpKey.jks",
+ "keyPassword": "secret",
+ "trustedCA": "/config/ftpKey.jks",
+ "trustedCAPassword": "secret"
}
}
}
}
+
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
- <version>5.0.5.RELEASE</version>
- </dependency>
+ </dependency>
</dependencies>
</project>
public class CommonFunctions {
- private static Gson gson = new GsonBuilder().create();
+ private static Gson gson = new GsonBuilder().serializeNulls().create();
private CommonFunctions() {}
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
<parent>
<groupId>org.onap.dcaegen2.collectors</groupId>
<artifactId>datafile</artifactId>
<version>1.0.0-SNAPSHOT</version>
</parent>
- <modelVersion>4.0.0</modelVersion>
<groupId>org.onap.dcaegen2.collectors.datafile</groupId>
<artifactId>datafile-dmaap-client</artifactId>
<dependencies>
<!-- DEVELOPMENT DEPENDENCIES -->
+ <dependency>
+ <groupId>org.onap.dcaegen2.collectors.datafile</groupId>
+ <artifactId>datafile-commons</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
<dependency>
<groupId>org.immutables</groupId>
<artifactId>value</artifactId>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</dependency>
- <dependency>
- <groupId>org.onap.dcaegen2.collectors.datafile</groupId>
- <artifactId>datafile-commons</artifactId>
- <version>1.0.0-SNAPSHOT</version>
- </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-reactor-netty</artifactId>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.jcraft</groupId>
+ <artifactId>jsch</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>commons-net</groupId>
+ <artifactId>commons-net</artifactId>
+ </dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.github.stefanbirkner</groupId>
+ <artifactId>fake-sftp-server-rule</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-test</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>com.google.code.findbugs</groupId>
<artifactId>jsr305</artifactId>
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+
+import org.apache.commons.net.ftp.FTPSClient;
+
+public class FTPSClientWrapper implements IFTPSClient {
+ private FTPSClient ftpsClient = new FTPSClient();
+
+ @Override
+ public void setNeedClientAuth(boolean isNeedClientAuth) {
+ ftpsClient.setNeedClientAuth(isNeedClientAuth);
+ }
+
+ @Override
+ public void setKeyManager(KeyManager keyManager) {
+ ftpsClient.setKeyManager(keyManager);
+ }
+
+ @Override
+ public void setTrustManager(TrustManager trustManager) {
+ ftpsClient.setTrustManager(trustManager);
+ }
+
+ @Override
+ public void connect(String hostName, int port) throws IOException {
+ ftpsClient.connect(hostName, port);
+ }
+
+ @Override
+ public boolean login(String username, String password) throws IOException {
+ return ftpsClient.login(username, password);
+ }
+
+ @Override
+ public boolean logout() throws IOException {
+ return ftpsClient.logout();
+ }
+
+ @Override
+ public int getReplyCode() {
+ return ftpsClient.getReplyCode();
+ }
+
+ @Override
+ public void disconnect() throws IOException {
+ ftpsClient.disconnect();
+ }
+
+ @Override
+ public void enterLocalPassiveMode() {
+ ftpsClient.enterLocalPassiveMode();
+ }
+
+ @Override
+ public void execPBSZ(int psbz) throws IOException {
+ ftpsClient.execPBSZ(psbz);
+ }
+
+ @Override
+ public void execPROT(String prot) throws IOException {
+ ftpsClient.execPROT(prot);
+ }
+
+ @Override
+ public boolean retrieveFile(String remote, OutputStream local) throws IOException {
+ return ftpsClient.retrieveFile(remote, local);
+ }
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.security.GeneralSecurityException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+
+import org.apache.commons.net.ftp.FTPReply;
+import org.onap.dcaegen2.collectors.datafile.io.FileSystemResourceWrapper;
+import org.onap.dcaegen2.collectors.datafile.io.FileWrapper;
+import org.onap.dcaegen2.collectors.datafile.io.IFile;
+import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
+import org.onap.dcaegen2.collectors.datafile.io.IOutputStream;
+import org.onap.dcaegen2.collectors.datafile.io.OutputStreamWrapper;
+import org.onap.dcaegen2.collectors.datafile.ssl.IKeyManagerUtils;
+import org.onap.dcaegen2.collectors.datafile.ssl.IKeyStore;
+import org.onap.dcaegen2.collectors.datafile.ssl.ITrustManagerFactory;
+import org.onap.dcaegen2.collectors.datafile.ssl.KeyManagerUtilsWrapper;
+import org.onap.dcaegen2.collectors.datafile.ssl.KeyStoreWrapper;
+import org.onap.dcaegen2.collectors.datafile.ssl.TrustManagerFactoryWrapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+/**
+ * Gets file from xNF with FTPS protocol.
+ *
+ * @author <a href="mailto:martin.c.yan@est.tech">Martin Yan</a>
+ */
+@Component
+public class FtpsClient {
+ private static final Logger logger = LoggerFactory.getLogger(FtpsClient.class);
+
+ private String keyCertPath;
+ private String keyCertPassword;
+ private String trustedCAPath;
+ private String trustedCAPassword;
+
+ private IFTPSClient realFtpsClient;
+ private IKeyManagerUtils kmu;
+ private IKeyStore keyStore;
+ private ITrustManagerFactory trustManagerFactory;
+ private IFile localFile;
+ private IFileSystemResource fileResource;
+ private IOutputStream os;
+
+ public boolean collectFile(FileServerData fileServerData, String remoteFile, String localFile) {
+ logger.trace("collectFile called with fileServerData: {}, remoteFile: {}, localFile: {}", fileServerData,
+ remoteFile, localFile);
+ boolean result = true;
+ IFTPSClient ftps = getFtpsClient();
+
+ ftps.setNeedClientAuth(true);
+
+ if (setUpKeyManager(ftps) && setUpTrustedCA(ftps) && setUpConnection(fileServerData, ftps)) {
+ result = getFileFromxNF(remoteFile, localFile, ftps, fileServerData);
+
+ closeDownConnection(ftps);
+ } else {
+ result = false;
+ }
+ logger.trace("collectFile left with result: {}", result);
+ return result;
+ }
+
+ private boolean setUpKeyManager(IFTPSClient ftps) {
+ boolean result = true;
+ try {
+ IKeyManagerUtils keyManagerUtils = getKeyManagerUtils();
+ keyManagerUtils.setCredentials(keyCertPath, keyCertPassword);
+ ftps.setKeyManager(keyManagerUtils.getClientKeyManager());
+ } catch (GeneralSecurityException | IOException e) {
+ logger.error("Unable to use own key store {}", keyCertPath, e);
+ result = false;
+ }
+ return result;
+ }
+
+ private boolean setUpTrustedCA(IFTPSClient ftps) {
+ boolean result = true;
+ try {
+ IFileSystemResource fileSystemResource = getFileSystemResource();
+ fileSystemResource.setPath(trustedCAPath);
+ InputStream fis = fileSystemResource.getInputStream();
+ IKeyStore ks = getKeyStore();
+ ks.load(fis, trustedCAPassword.toCharArray());
+ fis.close();
+ ITrustManagerFactory tmf = getTrustManagerFactory();
+ tmf.init(ks.getKeyStore());
+ ftps.setTrustManager(tmf.getTrustManagers()[0]);
+
+ } catch (Exception e) {
+ logger.error("Unable to trust xNF's CA, {}", trustedCAPath, e);
+ result = false;
+ }
+ return result;
+ }
+
+ private boolean setUpConnection(FileServerData fileServerData, IFTPSClient ftps) {
+ boolean result = true;
+ try {
+ ftps.connect(fileServerData.serverAddress(), fileServerData.port());
+
+ boolean loginSuccesful = ftps.login(fileServerData.userId(), fileServerData.password());
+ if (!loginSuccesful) {
+ ftps.logout();
+ logger.error("Unable to log in to xNF. {}", fileServerData);
+ result = false;
+ }
+
+ if (loginSuccesful && FTPReply.isPositiveCompletion(ftps.getReplyCode())) {
+ ftps.enterLocalPassiveMode();
+ // Set protection buffer size
+ ftps.execPBSZ(0);
+ // Set data channel protection to private
+ ftps.execPROT("P");
+ } else {
+ ftps.disconnect();
+ logger.error("Unable to connect to xNF. {}", fileServerData);
+ result = false;
+ }
+ } catch (Exception ex) {
+ logger.error("Unable to connect to xNF. Data: {}", fileServerData, ex);
+ result = false;
+ }
+ logger.trace("setUpConnection return value: {}", result);
+ return result;
+ }
+
+ private boolean getFileFromxNF(String remoteFile, String localFilePath, IFTPSClient ftps,
+ FileServerData fileServerData) {
+ logger.trace("starting to getFile");
+ boolean result = true;
+ try {
+ IFile outfile = getFile();
+ outfile.setPath(localFilePath);
+ outfile.createNewFile();
+
+ IOutputStream outputStream = getOutputStream();
+ OutputStream output = outputStream.getOutputStream(outfile.getFile());
+
+ ftps.retrieveFile(remoteFile, output);
+
+ output.close();
+ logger.debug("File {} Download Successfull from xNF", localFilePath);
+ } catch (IOException ex) {
+ logger.error("Unable to collect file from xNF. Data: {}", fileServerData, ex);
+ result = false;
+ }
+ return result;
+ }
+
+ private void closeDownConnection(IFTPSClient ftps) {
+ logger.trace("starting to closeDownConnection");
+ try {
+ if (ftps != null) {
+ ftps.logout();
+ ftps.disconnect();
+ }
+ } catch (Exception e) {
+ // Do nothing, file has been collected.
+ }
+ }
+
+ public void setKeyCertPath(String keyCertPath) {
+ this.keyCertPath = keyCertPath;
+ }
+
+ public void setKeyCertPassword(String keyCertPassword) {
+ this.keyCertPassword = keyCertPassword;
+ }
+
+ public void setTrustedCAPath(String trustedCAPath) {
+ this.trustedCAPath = trustedCAPath;
+ }
+
+ public void setTrustedCAPassword(String trustedCAPassword) {
+ this.trustedCAPassword = trustedCAPassword;
+ }
+
+ private ITrustManagerFactory getTrustManagerFactory() throws NoSuchAlgorithmException {
+ if (trustManagerFactory == null) {
+ trustManagerFactory = new TrustManagerFactoryWrapper();
+ }
+ return trustManagerFactory;
+ }
+
+ private IFTPSClient getFtpsClient() {
+ if (realFtpsClient == null) {
+ realFtpsClient = new FTPSClientWrapper();
+ }
+ return realFtpsClient;
+ }
+
+ private IKeyManagerUtils getKeyManagerUtils() {
+ if (kmu == null) {
+ kmu = new KeyManagerUtilsWrapper();
+ }
+
+ return kmu;
+ }
+
+ private IKeyStore getKeyStore() throws KeyStoreException {
+ if (keyStore == null) {
+ keyStore = new KeyStoreWrapper();
+ }
+
+ return keyStore;
+ }
+
+ private IFile getFile() {
+ if (localFile == null) {
+ localFile = new FileWrapper();
+ }
+
+ return localFile;
+ }
+
+ private IOutputStream getOutputStream() {
+ if (os == null) {
+ os = new OutputStreamWrapper();
+ }
+
+ return os;
+ }
+
+ private IFileSystemResource getFileSystemResource() {
+ if (fileResource == null) {
+ fileResource = new FileSystemResourceWrapper();
+ }
+ return fileResource;
+ }
+
+ protected void setFtpsClient(IFTPSClient ftpsClient) {
+ this.realFtpsClient = ftpsClient;
+ }
+
+ protected void setKeyManagerUtils(IKeyManagerUtils keyManagerUtils) {
+ this.kmu = keyManagerUtils;
+ }
+
+ protected void setKeyStore(IKeyStore keyStore) {
+ this.keyStore = keyStore;
+ }
+
+ protected void setTrustManagerFactory(ITrustManagerFactory tmf) {
+ trustManagerFactory = tmf;
+ }
+
+ protected void setFile(IFile file) {
+ localFile = file;
+ }
+
+ protected void setOutputStream(IOutputStream outputStream) {
+ os = outputStream;
+ }
+
+ protected void setFileSystemResource(IFileSystemResource fileSystemResource) {
+ fileResource = fileSystemResource;
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+
+public interface IFTPSClient {
+ public void setNeedClientAuth(boolean isNeedClientAuth);
+
+ public void setKeyManager(KeyManager keyManager);
+
+ public void setTrustManager(TrustManager trustManager);
+
+ public void connect(String hostname, int port) throws IOException;
+
+ public boolean login(String username, String password) throws IOException;
+
+ public boolean logout() throws IOException;
+
+ public int getReplyCode();
+
+ public void disconnect() throws IOException;
+
+ public void enterLocalPassiveMode();
+
+ public void execPBSZ(int newParam) throws IOException;
+
+ public void execPROT(String prot) throws IOException;
+
+ public boolean retrieveFile(String remote, OutputStream local) throws IOException;
+}
/**
* Gets file from xNF with SFTP protocol.
*
- * TODO: Refactor for better test.
- *
* @author <a href="mailto:martin.c.yan@est.tech">Martin Yan</a>
*
*/
@Component
-public class SftpClient { // TODO: Should be final but needs PowerMock to be able to mock then, so
- // this will be done as an improvement after first version committed.
+public class SftpClient {
private static final Logger logger = LoggerFactory.getLogger(SftpClient.class);
public boolean collectFile(FileServerData fileServerData, String remoteFile, String localFile) {
if (sftpChannel != null) {
try {
sftpChannel.get(remoteFile, localFile);
- logger.debug("File " + FilenameUtils.getName(localFile) + " Download Successfull from xNF");
+ logger.debug("File {} Download Successfull from xNF", FilenameUtils.getName(localFile));
} catch (SftpException e) {
- logger.error("Unable to get file from xNF. " + fileServerData, e);
+ logger.error("Unable to get file from xNF. Data: {}", fileServerData, e);
result = false;
}
}
private Session setUpSession(FileServerData fileServerData) {
- JSch jsch = new JSch(); // TODO: Might be changed to use Spring as an improvement after
- // first version committed.
+ JSch jsch = new JSch();
Session session = null;
try {
session.setPassword(fileServerData.password());
session.connect();
} catch (JSchException e) {
- logger.error("Unable to set up SFTP connection to xNF. " + fileServerData, e);
+ logger.error("Unable to set up SFTP connection to xNF. Data: {}", fileServerData, e);
}
return session;
}
channel.connect();
sftpChannel = (ChannelSftp) channel;
} catch (JSchException e) {
- logger.error("Unable to get sftp channel to xNF. " + fileServerData, e);
+ logger.error("Unable to get sftp channel to xNF. Data: {}", fileServerData, e);
}
return sftpChannel;
}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.springframework.core.io.FileSystemResource;
+
+/**
+ * @author
+ *
+ */
+public class FileSystemResourceWrapper implements IFileSystemResource {
+ private FileSystemResource realResource;
+
+ public void setPath(String path) {
+ realResource = new FileSystemResource(path);
+ }
+ @Override
+ public InputStream getInputStream() throws IOException {
+ return realResource.getInputStream();
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.File;
+import java.io.IOException;
+
+public class FileWrapper implements IFile {
+ private File file;
+
+ @Override
+ public void setPath(String path) {
+ file = new File(path);
+ }
+
+ @Override
+ public boolean createNewFile() throws IOException {
+ if (file == null) {
+ throw new IOException("Path to file not set.");
+ }
+ return file.createNewFile();
+ }
+
+ @Override
+ public File getFile() {
+ return file;
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.File;
+import java.io.IOException;
+
+public interface IFile {
+ public void setPath(String path);
+
+ public boolean createNewFile() throws IOException;
+
+ public File getFile();
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * @author
+ *
+ */
+public interface IFileSystemResource {
+
+ public void setPath(String filePath);
+
+ public InputStream getInputStream() throws IOException;
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.OutputStream;
+
+public interface IOutputStream {
+ public OutputStream getOutputStream(File file) throws FileNotFoundException;
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.OutputStream;
+
+public class OutputStreamWrapper implements IOutputStream {
+
+ @Override
+ public OutputStream getOutputStream(File file) throws FileNotFoundException {
+ return new FileOutputStream(file);
+ }
+
+}
* ============LICENSE_START======================================================================
* Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
* ===============================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
* ============LICENSE_END========================================================================
*/
package org.onap.dcaegen2.collectors.datafile.service;
+import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
+
import org.onap.dcaegen2.collectors.datafile.config.DmaapCustomConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClient.Builder;
import reactor.core.publisher.Mono;
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private String dmaaPContentType;
+ private String dmaaPUserName;
+ private String dmaaPUserPassword;
/**
* Creating DmaapReactiveWebClient passing to them basic DmaapConfig.
* @return WebClient
*/
public WebClient build() {
- return WebClient.builder()
- .defaultHeader(HttpHeaders.CONTENT_TYPE, dmaaPContentType)
- .filter(logRequest())
- .filter(logResponse())
- .build();
+ Builder webClientBuilder = WebClient.builder().defaultHeader(HttpHeaders.CONTENT_TYPE, dmaaPContentType)
+ .filter(logRequest()).filter(logResponse());
+ if (dmaaPUserName != null && !dmaaPUserName.isEmpty() && dmaaPUserPassword != null
+ && !dmaaPUserPassword.isEmpty()) {
+ webClientBuilder.filter(basicAuthentication(dmaaPUserName, dmaaPUserPassword));
+
+ }
+ return webClientBuilder.build();
}
private ExchangeFilterFunction logResponse() {
return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
- logger.info("Response Status {}", clientResponse.statusCode());
+ logger.trace("Response Status {}", clientResponse.statusCode());
return Mono.just(clientResponse);
});
}
private ExchangeFilterFunction logRequest() {
return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
- logger.info("Request: {} {}", clientRequest.method(), clientRequest.url());
+ logger.trace("Request: {} {}", clientRequest.method(), clientRequest.url());
clientRequest.headers()
- .forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));
+ .forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));
+ logger.trace("HTTP request headers: {}", clientRequest.headers());
return Mono.just(clientRequest);
});
}
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.io.IOUtils;
import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
+import org.onap.dcaegen2.collectors.datafile.io.FileSystemResourceWrapper;
+import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions;
import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.web.IRestTemplate;
+import org.onap.dcaegen2.collectors.datafile.web.RestTemplateWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpEntity;
addUserCredentialsToHead(headers);
- InputStream fileInputStream = getInputStream(consumerDmaapModel.getLocation());
+ IFileSystemResource fileSystemResource = getFileSystemResource();
+ fileSystemResource.setPath(consumerDmaapModel.getLocation());
+ InputStream fileInputStream = fileSystemResource.getInputStream();
HttpEntity<byte[]> request = addFileToRequest(fileInputStream, headers);
metaData.getAsJsonObject().remove(LOCATION_JSON_TAG);
headers.set(X_ATT_DR_META, metaData.toString());
}
-
private HttpEntity<byte[]> addFileToRequest(InputStream inputStream, HttpHeaders headers)
throws IOException {
return new HttpEntity<>(IOUtils.toByteArray(inputStream), headers);
}
- private InputStream getInputStream(String filePath) throws IOException {
- if (fileResource == null) {
- fileResource = new FileSystemResourceWrapper(filePath);
- }
- return fileResource.getInputStream();
- }
-
private IRestTemplate getRestTemplate() throws NoSuchAlgorithmException, KeyManagementException, KeyStoreException {
if (restTemplate == null) {
restTemplate = new RestTemplateWrapper();
.path(path).build();
}
+ private IFileSystemResource getFileSystemResource() {
+ if (fileResource == null) {
+ fileResource = new FileSystemResourceWrapper();
+ }
+ return fileResource;
+ }
+
protected void setFileSystemResource(IFileSystemResource fileSystemResource) {
fileResource = fileSystemResource;
}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+
+import javax.net.ssl.KeyManager;
+
+/**
+ * @author
+ *
+ */
+public interface IKeyManagerUtils {
+ public void setCredentials(String keyStorePath, String keyStorePass) throws IOException, GeneralSecurityException;
+
+ public KeyManager getClientKeyManager();
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+
+public interface IKeyStore {
+ public void load(InputStream arg0, char[] arg1)
+ throws IOException, NoSuchAlgorithmException, CertificateException, KeyStoreException;
+
+ public KeyStore getKeyStore();
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+
+import javax.net.ssl.TrustManager;
+
+public interface ITrustManagerFactory {
+ public void init(KeyStore ks) throws KeyStoreException;
+
+ public TrustManager[] getTrustManagers();
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.io.File;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+
+import javax.net.ssl.KeyManager;
+
+import org.apache.commons.net.util.KeyManagerUtils;
+
+/**
+ * @author
+ *
+ */
+public class KeyManagerUtilsWrapper implements IKeyManagerUtils {
+ private KeyManager keyManager;
+
+ @Override
+ public void setCredentials(String keyStorePath, String keyStorePass) throws IOException, GeneralSecurityException {
+ keyManager = KeyManagerUtils.createClientKeyManager(new File(keyStorePath), keyStorePass);
+ }
+
+ @Override
+ public KeyManager getClientKeyManager() {
+ return keyManager;
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+
+public class KeyStoreWrapper implements IKeyStore {
+ private KeyStore keyStore;
+
+ public KeyStoreWrapper() throws KeyStoreException {
+ keyStore = KeyStore.getInstance("JKS");
+ }
+
+ @Override
+ public void load(InputStream stream, char[] password)
+ throws IOException, NoSuchAlgorithmException, CertificateException, KeyStoreException {
+ keyStore.load(stream, password);
+ }
+
+ @Override
+ public KeyStore getKeyStore() {
+ return keyStore;
+ }
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+
+public class TrustManagerFactoryWrapper implements ITrustManagerFactory {
+ TrustManagerFactory tmf;
+
+ public TrustManagerFactoryWrapper() throws NoSuchAlgorithmException {
+ tmf = TrustManagerFactory.getInstance("SunX509");
+ }
+
+ @Override
+ public void init(KeyStore ks) throws KeyStoreException {
+ tmf.init(ks);
+ }
+
+ @Override
+ public TrustManager[] getTrustManagers() {
+ return tmf.getTrustManagers();
+ }
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.web;
+
+import java.net.URI;
+
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+
+/**
+ * @author
+ *
+ */
+public interface IRestTemplate {
+ public ResponseEntity<String> exchange(URI url, HttpMethod method, HttpEntity<byte[]> requestEntity,
+ Class<String> responseType);
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+package org.onap.dcaegen2.collectors.datafile.web;
+
+import java.net.URI;
+
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.ProtocolException;
+import org.apache.http.annotation.Contract;
+import org.apache.http.annotation.ThreadingBehavior;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpHead;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.client.methods.RequestBuilder;
+import org.apache.http.impl.client.DefaultRedirectStrategy;
+import org.apache.http.protocol.HttpContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * PublishRedirectStrategy implementation
+ * that automatically redirects all HEAD, GET, POST, PUT, and DELETE requests.
+ * This strategy relaxes restrictions on automatic redirection of
+ * POST methods imposed by the HTTP specification.
+ *
+ */
+@Contract(threading = ThreadingBehavior.IMMUTABLE)
+public class PublishRedirectStrategy extends DefaultRedirectStrategy {
+
+ public static final PublishRedirectStrategy INSTANCE = new PublishRedirectStrategy();
+ private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ /**
+ * Redirectable methods.
+ */
+ private static final String[] REDIRECT_METHODS = new String[] {
+ HttpPut.METHOD_NAME,
+ HttpGet.METHOD_NAME,
+ HttpPost.METHOD_NAME,
+ HttpHead.METHOD_NAME,
+ HttpDelete.METHOD_NAME
+ };
+
+ @Override
+ protected boolean isRedirectable(final String method) {
+ for (final String m: REDIRECT_METHODS) {
+ if (m.equalsIgnoreCase(method)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public HttpUriRequest getRedirect(
+ final HttpRequest request,
+ final HttpResponse response,
+ final HttpContext context) throws ProtocolException {
+ final URI uri = getLocationURI(request, response, context);
+ logger.trace("getRedirect...: {}", request);
+ return RequestBuilder.copy(request).setUri(uri).build();
+ }
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+package org.onap.dcaegen2.collectors.datafile.web;
+import java.io.IOException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpRequest;
+import org.springframework.http.client.ClientHttpRequestExecution;
+import org.springframework.http.client.ClientHttpRequestInterceptor;
+import org.springframework.http.client.ClientHttpResponse;
+
+public class RequestResponseLoggingInterceptor implements ClientHttpRequestInterceptor {
+
+ private final Logger log = LoggerFactory.getLogger(this.getClass());
+
+ @Override
+ public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) throws IOException {
+ logRequest(request, body);
+ ClientHttpResponse response = execution.execute(request, body);
+ logResponse(response);
+ return response;
+ }
+
+ private void logRequest(HttpRequest request, byte[] body) throws IOException {
+ if (log.isDebugEnabled()) {
+ log.debug("===========================request begin================================================");
+ log.debug("URI : {}", request.getURI());
+ log.debug("Method : {}", request.getMethod());
+ log.debug("Headers : {}", request.getHeaders());
+ log.debug("Request body: {}", new String(body, "UTF-8"));
+ log.debug("==========================request end================================================");
+ }
+ }
+
+ private void logResponse(ClientHttpResponse response) throws IOException {
+ if (log.isDebugEnabled()) {
+ log.debug("============================response begin==========================================");
+ log.debug("Status code : {}", response.getStatusCode());
+ log.debug("Status text : {}", response.getStatusText());
+ log.debug("Headers : {}", response.getHeaders());
+ log.debug("=======================response end=================================================");
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.web;
+
+import java.net.URI;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Collections;
+
+import javax.net.ssl.SSLContext;
+
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * @author
+ *
+ */
+public class RestTemplateWrapper implements IRestTemplate {
+ private RestTemplate restTemplate;
+
+ public RestTemplateWrapper() throws KeyManagementException, NoSuchAlgorithmException, KeyStoreException {
+ SSLContext sslContext =
+ new SSLContextBuilder().loadTrustMaterial(null, (certificate, authType) -> true).build();
+ CloseableHttpClient httpClient =
+ HttpClients.custom().setSSLContext(sslContext).setSSLHostnameVerifier(new NoopHostnameVerifier())
+ .setRedirectStrategy(new PublishRedirectStrategy()).build();
+
+ HttpComponentsClientHttpRequestFactory requestFactory = new HttpComponentsClientHttpRequestFactory();
+ requestFactory.setHttpClient(httpClient);
+
+ restTemplate = new RestTemplate(requestFactory);
+ restTemplate.setInterceptors(Collections.singletonList(new RequestResponseLoggingInterceptor()));
+
+ }
+
+ @Override
+ public ResponseEntity<String> exchange(URI url, HttpMethod method, HttpEntity<byte[]> requestEntity,
+ Class<String> responseType) {
+ return restTemplate.exchange(url, method, requestEntity, responseType);
+ }
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.io.IFile;
+import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
+import org.onap.dcaegen2.collectors.datafile.io.IOutputStream;
+import org.onap.dcaegen2.collectors.datafile.ssl.IKeyManagerUtils;
+import org.onap.dcaegen2.collectors.datafile.ssl.IKeyStore;
+import org.onap.dcaegen2.collectors.datafile.ssl.ITrustManagerFactory;
+import org.springframework.http.HttpStatus;
+
+public class FtpsClientTest {
+
+ private static final String REMOTE_FILE_PATH = "/dir/sample.txt";
+ private static final String LOCAL_FILE_PATH = "target/sample.txt";
+ private static final String XNF_ADDRESS = "127.0.0.1";
+ private static final int PORT = 8021;
+ private static final String FTP_KEY_PATH = "ftpKeyPath";
+ private static final String FTP_KEY_PASSWORD = "ftpKeyPassword";
+ private static final String TRUSTED_CA_PATH = "trustedCAPath";
+ private static final String TRUSTED_CA_PASSWORD = "trustedCAPassword";
+
+ private static final String USERNAME = "bob";
+ private static final String PASSWORD = "123";
+
+ private IFTPSClient ftpsClientMock = mock(IFTPSClient.class);
+ private IKeyManagerUtils keyManagerUtilsMock = mock(IKeyManagerUtils.class);
+ private KeyManager keyManagerMock = mock(KeyManager.class);
+ private IKeyStore keyStoreWrapperMock = mock(IKeyStore.class);
+ private KeyStore keyStoreMock = mock(KeyStore.class);
+ private ITrustManagerFactory trustManagerFactoryMock = mock(ITrustManagerFactory.class);
+ private TrustManager trustManagerMock = mock(TrustManager.class);
+ private IFile localFileMock = mock(IFile.class);
+ private IFileSystemResource fileResourceMock = mock(IFileSystemResource.class);
+ private IOutputStream outputStreamMock = mock(IOutputStream.class);
+ private InputStream inputStreamMock = mock(InputStream.class);
+
+ FtpsClient clientUnderTest = new FtpsClient();
+
+ @BeforeEach
+ protected void setUp() throws Exception {
+ clientUnderTest.setFtpsClient(ftpsClientMock);
+ clientUnderTest.setKeyManagerUtils(keyManagerUtilsMock);
+ clientUnderTest.setKeyStore(keyStoreWrapperMock);
+ clientUnderTest.setTrustManagerFactory(trustManagerFactoryMock);
+ clientUnderTest.setFile(localFileMock);
+ clientUnderTest.setFileSystemResource(fileResourceMock);
+ clientUnderTest.setOutputStream(outputStreamMock);
+
+ clientUnderTest.setKeyCertPath(FTP_KEY_PATH);
+ clientUnderTest.setKeyCertPassword(FTP_KEY_PASSWORD);
+ clientUnderTest.setTrustedCAPath(TRUSTED_CA_PATH);
+ clientUnderTest.setTrustedCAPassword(TRUSTED_CA_PASSWORD);
+}
+
+ @Test
+ public void collectFile_allOk() throws Exception {
+ when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+ when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+ when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+ when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+ when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
+ when(ftpsClientMock.getReplyCode()).thenReturn(HttpStatus.OK.value());
+ File fileMock = mock(File.class);
+ when(localFileMock.getFile()).thenReturn(fileMock);
+ OutputStream osMock = mock(OutputStream.class);
+ when(outputStreamMock.getOutputStream(fileMock)).thenReturn(osMock);
+
+ ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+ .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+ boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+ assertTrue(result);
+ verify(ftpsClientMock).setNeedClientAuth(true);
+ verify(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+ verify(ftpsClientMock).setKeyManager(keyManagerMock);
+ verify(fileResourceMock).setPath(TRUSTED_CA_PATH);
+ verify(keyStoreWrapperMock).load(inputStreamMock, TRUSTED_CA_PASSWORD.toCharArray());
+ verify(inputStreamMock, times(1)).close();
+ verify(trustManagerFactoryMock).init(keyStoreMock);
+ verify(ftpsClientMock).setTrustManager(trustManagerMock);
+ verify(ftpsClientMock).connect(XNF_ADDRESS, PORT);
+ verify(ftpsClientMock).login(USERNAME, PASSWORD);
+ verify(ftpsClientMock).getReplyCode();
+ verify(ftpsClientMock, times(1)).enterLocalPassiveMode();
+ verify(ftpsClientMock).execPBSZ(0);
+ verify(ftpsClientMock).execPROT("P");
+ verify(localFileMock).setPath(LOCAL_FILE_PATH);
+ verify(localFileMock, times(1)).createNewFile();
+ verify(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, osMock);
+ verify(osMock, times(1)).close();
+ verify(ftpsClientMock, times(1)).logout();
+ verify(ftpsClientMock, times(1)).disconnect();
+ verifyNoMoreInteractions(ftpsClientMock);
+ }
+
+ @Test
+ public void collectFileFaultyOwnKey_shouldFail() throws Exception {
+ doThrow(new GeneralSecurityException())
+ .when(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+
+ ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+ .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+ boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+ assertFalse(result);
+ }
+
+ @Test
+ public void collectFileFaultTrustedCA_shouldFail() throws Exception {
+ when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+ when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+ when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+
+ doThrow(new KeyStoreException()).when(trustManagerFactoryMock).init(keyStoreMock);
+
+ ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+ .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+ boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+ assertFalse(result);
+ }
+
+ @Test
+ public void collectFileFaultyLogin_shouldFail() throws Exception {
+ when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+ when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+ when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+ when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+ when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(false);
+
+ ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+ .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+ boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+ verify(ftpsClientMock, times(1)).logout();
+ assertFalse(result);
+ }
+
+ @Test
+ public void collectFileBadRequestResponse_shouldFail() throws Exception {
+ when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+ when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+ when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+ when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+ when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
+ when(ftpsClientMock.getReplyCode()).thenReturn(HttpStatus.BAD_REQUEST.value());
+
+ ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+ .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+ boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+ verify(ftpsClientMock, times(1)).disconnect();
+ assertFalse(result);
+ }
+
+ @Test
+ public void collectFileFaultyConnection_shouldFail() throws Exception {
+ when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+ when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+ when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+ when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+
+ doThrow(new IOException()).when(ftpsClientMock).connect(XNF_ADDRESS, PORT);
+
+ ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+ .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+ boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+ assertFalse(result);
+ }
+
+ @Test
+ public void collectFileFailingFileCollect_shouldFail() throws Exception {
+ when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+ when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+ when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+ when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+ when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
+ when(ftpsClientMock.getReplyCode()).thenReturn(HttpStatus.OK.value());
+
+ doThrow(new IOException()).when(localFileMock).createNewFile();
+
+ ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+ .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+ boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+ assertFalse(result);
+ }
+}
\ No newline at end of file
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
+import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions;
import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModelForUnitTest;
+import org.onap.dcaegen2.collectors.datafile.web.IRestTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
fileStream.reset();
HttpEntity<byte[]> requestEntity = new HttpEntity<>(IOUtils.toByteArray(fileStream), headers);
+ verify(fileSystemResourceMock).setPath("target/" + FILE_NAME);
verify(restTemplateMock).exchange(expectedUri, HttpMethod.PUT, requestEntity, String.class);
verifyNoMoreInteractions(restTemplateMock);
}
--dmaap.dmaapProducerConfiguration.dmaapUserName=dradmin
--dmaap.dmaapProducerConfiguration.dmaapUserPassword=dradmin
--dmaap.dmaapProducerConfiguration.dmaapContentType=application/octet-stream
+ --dmaap.ftpesConfig.keyCert=config/ftpKey.jks
+ --dmaap.ftpesConfig.keyPassword=secret
+ --dmaap.ftpesConfig.trustedCA=config/cacerts
+ --dmaap.ftpesConfig.trustedCAPassword=secret
entrypoint:
- java
- -Dspring.profiles.active=dev
</properties>
<modules>
- <module>datafile-app-server</module>
- <module>datafile-dmaap-client</module>
<module>datafile-commons</module>
+ <module>datafile-dmaap-client</module>
+ <module>datafile-app-server</module>
</modules>
<build>
<artifactId>fake-sftp-server-rule</artifactId>
<version>2.0.1</version>
</dependency>
+ <dependency>
+ <groupId>com.jcraft</groupId>
+ <artifactId>jsch</artifactId>
+ <version>0.1.53</version>
+ </dependency>
<!--REQUIRED TO GENERATE DOCUMENTATION -->
<dependency>
Code Review / dcaegen2 / collectors / datafile.git / commitdiff