From 6e1e3b0016fa1420aa7e942d6314795a15cf0378 Mon Sep 17 00:00:00 2001 From: Peter Turcsanyi Date: Thu, 13 Aug 2026 18:25:14 +0200 Subject: [PATCH] NIFI-16053 Refactored Get/PutSmbFile to use SmbClientProviderService --- .../nifi/services/smb/SmbClientProvider.java | 61 +++ .../smb/SmbClientProviderService.java | 21 +- .../nifi/services/smb/SmbClientService.java | 15 +- .../nifi/services/smb/SmbListableEntity.java | 16 +- .../nifi/services/smb/SmbShareAccess.java | 30 ++ .../nifi-smb-processors/pom.xml | 6 +- .../apache/nifi/processors/smb/FetchSmb.java | 7 +- .../nifi/processors/smb/GetSmbFile.java | 271 ++++++-------- .../apache/nifi/processors/smb/ListSmb.java | 2 +- .../nifi/processors/smb/PutSmbFile.java | 200 +++------- .../util/HostnameAndShareFlowFileFilter.java | 77 ---- .../smb/util/LocalSmbProperties.java | 135 +++++++ .../util/ServiceLocationFlowFileFilter.java | 73 ++++ .../nifi/processors/smb/FetchSmbTest.java | 8 +- .../nifi/processors/smb/GetSmbFileTest.java | 238 +++++------- .../apache/nifi/processors/smb/ListSmbIT.java | 6 +- .../nifi/processors/smb/ListSmbTest.java | 5 +- .../nifi/processors/smb/PutSmbFileTest.java | 348 ++++++------------ .../processors/smb/SambaTestContainers.java | 12 +- .../apache/nifi/processors/smb/SmbDfsIT.java | 26 +- .../nifi-smb-smbj-client/pom.xml | 14 - .../smb/SmbjClientProviderService.java | 164 ++------- .../nifi-smb-smbj-common/pom.xml | 17 + .../nifi/services/smb/SmbjClientProvider.java | 132 +++++++ .../nifi/services/smb/SmbjClientService.java | 75 +++- .../apache/nifi/smb/common/SmbProperties.java | 58 +++ .../services/smb/SmbjClientServiceIT.java | 38 +- .../services/smb/SmbjClientServiceTest.java | 6 +- 28 files changed, 1047 insertions(+), 1014 deletions(-) create mode 100644 nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientProvider.java create mode 100644 nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbShareAccess.java delete mode 100644 nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/HostnameAndShareFlowFileFilter.java create mode 100644 nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/LocalSmbProperties.java create mode 100644 nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/ServiceLocationFlowFileFilter.java create mode 100644 nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/services/smb/SmbjClientProvider.java rename nifi-extension-bundles/nifi-smb-bundle/{nifi-smb-smbj-client => nifi-smb-smbj-common}/src/main/java/org/apache/nifi/services/smb/SmbjClientService.java (78%) rename nifi-extension-bundles/nifi-smb-bundle/{nifi-smb-smbj-client => nifi-smb-smbj-common}/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceIT.java (85%) rename nifi-extension-bundles/nifi-smb-bundle/{nifi-smb-smbj-client => nifi-smb-smbj-common}/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceTest.java (94%) diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientProvider.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientProvider.java new file mode 100644 index 000000000000..acb866545b2a --- /dev/null +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientProvider.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.services.smb; + +import org.apache.nifi.logging.ComponentLog; + +import java.io.IOException; +import java.net.URI; +import java.util.Map; + +public interface SmbClientProvider { + + /** + * Returns the identifier of the service location. + * + * @return the remote location + */ + default URI getServiceLocation() { + return getServiceLocation(Map.of()); + } + + /** + * Returns the identifier of the service location. + * + * @param attributes FlowFile attributes to evaluate connection properties + * @return the remote location + */ + URI getServiceLocation(Map attributes); + + /** + * Returns the smb client to use. + * + * @return the client. + */ + default SmbClientService getClient(ComponentLog logger) throws IOException { + return getClient(logger, Map.of()); + } + + /** + * Returns the smb client to use. + * + * @param attributes FlowFile attributes to evaluate connection properties + * @return the client. + */ + SmbClientService getClient(ComponentLog logger, Map attributes) throws IOException; + +} diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientProviderService.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientProviderService.java index aff231b6c7db..caa63842e28a 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientProviderService.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientProviderService.java @@ -17,25 +17,6 @@ package org.apache.nifi.services.smb; import org.apache.nifi.controller.ControllerService; -import org.apache.nifi.logging.ComponentLog; - -import java.io.IOException; -import java.net.URI; - -public interface SmbClientProviderService extends ControllerService { - - /** - * Returns the identifier of the service location. - * - * @return the remote location - */ - URI getServiceLocation(); - - /** - * Returns the smb client to use. - * - * @return the client. - */ - SmbClientService getClient(ComponentLog logger) throws IOException; +public interface SmbClientProviderService extends SmbClientProvider, ControllerService { } diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientService.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientService.java index c70dea133ee3..f6e246e48561 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientService.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbClientService.java @@ -16,8 +16,9 @@ */ package org.apache.nifi.services.smb; -import java.io.IOException; +import java.io.InputStream; import java.io.OutputStream; +import java.util.Set; import java.util.stream.Stream; /** @@ -25,11 +26,19 @@ */ public interface SmbClientService extends AutoCloseable { - Stream listFiles(String directoryPath); + boolean folderExists(String path); + + boolean fileExists(String path); + + Stream listFiles(String directoryPath, boolean recursive); void ensureDirectory(String directoryPath); - void readFile(String filePath, OutputStream outputStream) throws IOException; + void readFile(String filePath, OutputStream outputStream, Set shareAccesses); + + void writeFile(String filePath, InputStream inputStream, Set shareAccesses); + + void renameFile(String oldFilePath, String newFilePath, boolean override); void moveFile(String filePath, String directoryPath); diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbListableEntity.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbListableEntity.java index 1428021fda68..070f59f22e85 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbListableEntity.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbListableEntity.java @@ -49,12 +49,13 @@ public class SmbListableEntity implements ListableEntity { private final long lastAccessTime; private final long changeTime; private final boolean directory; + private final boolean hidden; private final long size; private final long allocationSize; private final URI serviceLocation; private SmbListableEntity(String name, String shortName, String path, long lastModifiedTime, long creationTime, - long lastAccessTime, long changeTime, boolean directory, + long lastAccessTime, long changeTime, boolean directory, boolean hidden, long size, long allocationSize, URI serviceLocation) { this.name = name; this.shortName = shortName; @@ -64,6 +65,7 @@ private SmbListableEntity(String name, String shortName, String path, long lastM this.lastAccessTime = lastAccessTime; this.changeTime = changeTime; this.directory = directory; + this.hidden = hidden; this.size = size; this.allocationSize = allocationSize; this.serviceLocation = serviceLocation; @@ -144,6 +146,10 @@ public boolean isDirectory() { return directory; } + public boolean isHidden() { + return hidden; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -196,6 +202,7 @@ public static class SmbListableEntityBuilder { private long lastAccessTime; private long changeTime; private boolean directory = false; + private boolean hidden = false; private long size = 0; private long allocationSize = 0; private URI serviceLocation; @@ -240,6 +247,11 @@ public SmbListableEntityBuilder setDirectory(boolean directory) { return this; } + public SmbListableEntityBuilder setHidden(boolean hidden) { + this.hidden = hidden; + return this; + } + public SmbListableEntityBuilder setSize(long size) { this.size = size; return this; @@ -257,7 +269,7 @@ public SmbListableEntityBuilder setServiceLocation(URI serviceLocation) { public SmbListableEntity build() { return new SmbListableEntity(name, shortName, path, lastModifiedTime, creationTime, lastAccessTime, changeTime, - directory, size, allocationSize, serviceLocation); + directory, hidden, size, allocationSize, serviceLocation); } } diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbShareAccess.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbShareAccess.java new file mode 100644 index 000000000000..075a94374705 --- /dev/null +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-client-api/src/main/java/org/apache/nifi/services/smb/SmbShareAccess.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.services.smb; + +import java.util.Set; + +public enum SmbShareAccess { + READ_ALLOWED, + WRITE_ALLOWED, + DELETE_ALLOWED; + + public static final Set NONE = Set.of(); + public static final Set READ = Set.of(READ_ALLOWED); + public static final Set READ_DELETE = Set.of(READ_ALLOWED, DELETE_ALLOWED); + public static final Set READ_WRITE_DELETE = Set.of(READ_ALLOWED, WRITE_ALLOWED, DELETE_ALLOWED); +} diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/pom.xml b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/pom.xml index 2af1242b805d..e867474606e2 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/pom.xml +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/pom.xml @@ -51,13 +51,11 @@ nifi-smb-smbj-common 2.12.0-SNAPSHOT - - com.hierynomus - smbj - + commons-io commons-io + test org.testcontainers diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/FetchSmb.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/FetchSmb.java index 8929ad38b0fc..7f01bcfb0e67 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/FetchSmb.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/FetchSmb.java @@ -35,6 +35,7 @@ import org.apache.nifi.services.smb.SmbClientProviderService; import org.apache.nifi.services.smb.SmbClientService; import org.apache.nifi.services.smb.SmbException; +import org.apache.nifi.services.smb.SmbShareAccess; import java.util.List; import java.util.Map; @@ -149,8 +150,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final SmbClientProviderService clientProviderService = context.getProperty(SMB_CLIENT_PROVIDER_SERVICE).asControllerService(SmbClientProviderService.class); - try (SmbClientService client = clientProviderService.getClient(getLogger())) { - flowFile = session.write(flowFile, outputStream -> client.readFile(filePath, outputStream)); + try (SmbClientService client = clientProviderService.getClient(getLogger(), attributes)) { + flowFile = session.write(flowFile, outputStream -> client.readFile(filePath, outputStream, SmbShareAccess.READ)); session.transfer(flowFile, REL_SUCCESS); } catch (Exception e) { @@ -188,7 +189,7 @@ private void performCompletionStrategy(final ProcessContext context, final Map PROPERTY_DESCRIPTORS = List.of( + CONNECTION_CONFIGURATION_STRATEGY, + SMB_CLIENT_PROVIDER_SERVICE, HOSTNAME, + PORT, SHARE, DIRECTORY, DOMAIN, @@ -248,21 +228,21 @@ public class GetSmbFile extends AbstractProcessor { REL_SUCCESS ); - private final BlockingQueue fileQueue = new LinkedBlockingQueue<>(); - private final Set inProcess = new HashSet<>(); // guarded by queueLock - private final Set recentlyProcessed = new HashSet<>(); // guarded by queueLock + private final BlockingQueue fileQueue = new LinkedBlockingQueue<>(); + private final Set inProcess = new HashSet<>(); // guarded by queueLock + private final Set recentlyProcessed = new HashSet<>(); // guarded by queueLock private final Lock queueLock = new ReentrantLock(); private final Lock listingLock = new ReentrantLock(); private final AtomicLong queueLastUpdated = new AtomicLong(0L); - private SMBClient smbClient = null; // this gets synchronized when the `connect` method is called + private SmbClientProvider clientProvider; private Pattern filePattern; private Pattern pathPattern; private boolean ignoreHidden; - private Set sharedAccess; + private Set sharedAccess; @Override public Set getRelationships() { @@ -276,33 +256,36 @@ public final List getSupportedPropertyDescriptors() { @OnScheduled public void onScheduled(final ProcessContext context) { - smbClient = initSmbClient(context); + clientProvider = switch (context.getProperty(CONNECTION_CONFIGURATION_STRATEGY).asAllowableValue(LocalSmbProperties.ConnectionConfigurationStrategy.class)) { + case CONTROLLER_SERVICE -> context.getProperty(SMB_CLIENT_PROVIDER_SERVICE).asControllerService(SmbClientProviderService.class); + case LOCAL_PROPERTIES -> new SmbjClientProvider(context, getLogger()); + }; initiateFilterFile(context); fileQueue.clear(); switch (context.getProperty(SHARE_ACCESS).getValue()) { case SHARE_ACCESS_NONE: - sharedAccess = Collections.emptySet(); + sharedAccess = SmbShareAccess.NONE; break; case SHARE_ACCESS_READ: - sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ); + sharedAccess = SmbShareAccess.READ; break; case SHARE_ACCESS_READDELETE: - sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ, SMB2ShareAccess.FILE_SHARE_DELETE); + sharedAccess = SmbShareAccess.READ_DELETE; break; case SHARE_ACCESS_READWRITEDELETE: - sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ, SMB2ShareAccess.FILE_SHARE_WRITE, SMB2ShareAccess.FILE_SHARE_DELETE); + sharedAccess = SmbShareAccess.READ_WRITE_DELETE; break; } } @OnStopped public void onStopped() { - if (smbClient != null) { - smbClient.close(); - smbClient = null; + if (clientProvider instanceof SmbjClientProvider smbjClientProvider) { + smbjClientProvider.close(); } + clientProvider = null; } @Override @@ -316,14 +299,14 @@ public void migrateProperties(PropertyConfiguration config) { @Override protected Collection customValidate(ValidationContext validationContext) { Collection set = new ArrayList<>(); - if (validationContext.getProperty(USERNAME).isSet() && !validationContext.getProperty(PASSWORD).isSet()) { - set.add(new ValidationResult.Builder().explanation("Password must be set if username is supplied.").build()); + + if (validationContext.getProperty(CONNECTION_CONFIGURATION_STRATEGY).asAllowableValue(ConnectionConfigurationStrategy.class) == ConnectionConfigurationStrategy.LOCAL_PROPERTIES) { + if (validationContext.getProperty(USERNAME).isSet() && !validationContext.getProperty(PASSWORD).isSet()) { + set.add(new ValidationResult.Builder().explanation("Password must be set if username is supplied.").build()); + } } - return set; - } - SMBClient initSmbClient(final ProcessContext context) { - return buildSmbClient(context); + return set; } private void initiateFilterFile(final ProcessContext context) { @@ -331,55 +314,35 @@ private void initiateFilterFile(final ProcessContext context) { filePattern = filePatternStr == null ? null : Pattern.compile(filePatternStr); final String pathPatternStr = context.getProperty(PATH_FILTER).getValue(); pathPattern = pathPatternStr == null ? null : Pattern.compile(pathPatternStr); - ignoreHidden = context.getProperty(IGNORE_HIDDEN_FILES).asBoolean().booleanValue(); + ignoreHidden = context.getProperty(IGNORE_HIDDEN_FILES).asBoolean(); } - private boolean filterFile(final String directory, final String filename, final long fileAttributes) { - if (pathPattern != null && !pathPattern.matcher(directory).matches()) { + private boolean filterFile(final SmbFileInfo fileInfo) { + if (pathPattern != null && !pathPattern.matcher(fileInfo.path()).matches()) { return false; } - if (filePattern != null && !filePattern.matcher(filename).matches()) { + if (filePattern != null && !filePattern.matcher(fileInfo.filename()).matches()) { return false; } - if (ignoreHidden && (fileAttributes & FileAttributes.FILE_ATTRIBUTE_HIDDEN.getValue()) != 0) { + if (ignoreHidden && fileInfo.hidden()) { return false; } return true; } - private Set performListing(final DiskShare diskShare, final String directory, final String filter, final boolean recurseSubdirectories) { - final Set queue = new HashSet<>(); - if (!diskShare.folderExists(directory)) { - return queue; - } - - final List children = diskShare.list(directory); - if (children == null) { - return queue; - } - - for (final FileIdBothDirectoryInformation child : children) { - final String filename = child.getFileName(); - if (filename.equals(".") || filename.equals("..")) { - continue; - } - String fullPath; - if (directory.isEmpty()) { - fullPath = filename; - } else { - fullPath = directory + "\\" + filename; - } - final long fileAttributes = child.getFileAttributes(); - if ((fileAttributes & FileAttributes.FILE_ATTRIBUTE_DIRECTORY.getValue()) != 0) { - if (recurseSubdirectories) { - queue.addAll(performListing(diskShare, fullPath, filter, true)); - } - } else if (filterFile(directory, filename, fileAttributes)) { - queue.add(fullPath); - } - } - - return queue; + private Set performListing(final SmbClientService client, final String directory, final boolean recurseSubdirectories) { + return client.listFiles(directory, recurseSubdirectories) + .map(e -> new SmbFileInfo( + e.getName(), + e.getPath().replace('/', '\\'), + e.getSize(), + e.isHidden(), + e.getCreationTime(), + e.getLastModifiedTime(), + e.getLastAccessTime() + )) + .filter(this::filterFile) + .collect(Collectors.toSet()); } @Override @@ -387,38 +350,22 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final ComponentLog logger = getLogger(); - final String hostname = context.getProperty(HOSTNAME).getValue(); - final String shareName = context.getProperty(SHARE).getValue(); - - final String domain = context.getProperty(DOMAIN).getValue(); - final String username = context.getProperty(USERNAME).getValue(); - final String password = context.getProperty(PASSWORD).getValue(); - - AuthenticationContext ac = null; - if (username != null && password != null) { - ac = new AuthenticationContext( - username, - password.toCharArray(), - domain); - } else { - ac = AuthenticationContext.anonymous(); - } + final URI serviceLocation = clientProvider.getServiceLocation(); + final String hostname = serviceLocation.getHost(); + final String shareName = StringUtils.removeStart(serviceLocation.getPath(), '/'); - try (Connection connection = smbClient.connect(hostname); - Session smbSession = connection.authenticate(ac); - DiskShare share = (DiskShare) smbSession.connectShare(shareName)) { + try (SmbClientService client = clientProvider.getClient(getLogger())) { String directory = context.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue(); if (directory == null) { directory = ""; } final boolean keepingSourceFile = context.getProperty(KEEP_SOURCE_FILE).asBoolean(); - final String filter = context.getProperty(FILE_FILTER).getValue(); if (fileQueue.size() < 100) { final long pollingMillis = context.getProperty(POLLING_INTERVAL).asTimePeriod(TimeUnit.MILLISECONDS); if ((queueLastUpdated.get() < System.currentTimeMillis() - pollingMillis) && listingLock.tryLock()) { try { - final Set listing = performListing(share, directory, filter, context.getProperty(RECURSE).asBoolean().booleanValue()); + final Set listing = performListing(client, directory, context.getProperty(RECURSE).asBoolean()); queueLock.lock(); try { @@ -446,7 +393,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session } final int batchSize = context.getProperty(BATCH_SIZE).asInteger(); - final List files = new ArrayList<>(batchSize); + final List files = new ArrayList<>(batchSize); queueLock.lock(); try { fileQueue.drainTo(files, batchSize); @@ -459,56 +406,43 @@ public void onTrigger(final ProcessContext context, final ProcessSession session queueLock.unlock(); } - final ListIterator itr = files.listIterator(); + final ListIterator itr = files.listIterator(); FlowFile flowFile = null; try { while (itr.hasNext()) { - final String file = itr.next(); - final String[] fileSplits = file.split("\\\\"); - final String filename = fileSplits[fileSplits.length - 1]; - final String filePath = String.join("\\", Arrays.copyOf(fileSplits, fileSplits.length - 1)); - final URI uri = new URI("smb", hostname, "/" + file.replace('\\', '/'), null); + final SmbFileInfo fileInfo = itr.next(); + final String fullPath = String.format("%s\\%s", fileInfo.path(), fileInfo.filename()); + final String transitUri = String.format("%s/%s", serviceLocation, fullPath.replace('\\', '/')); flowFile = session.create(); final long importStart = System.nanoTime(); - try (File f = share.openFile( - file, - EnumSet.of(AccessMask.GENERIC_READ), - EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL), - sharedAccess, - SMB2CreateDisposition.FILE_OPEN, - EnumSet.of(SMB2CreateOptions.FILE_SEQUENTIAL_ONLY)); - InputStream is = f.getInputStream()) { - - flowFile = session.importFrom(is, flowFile); + try { + flowFile = session.write(flowFile, outputStream -> client.readFile(fullPath, outputStream, sharedAccess)); final long importNanos = System.nanoTime() - importStart; final long importMillis = TimeUnit.MILLISECONDS.convert(importNanos, TimeUnit.NANOSECONDS); - final FileAllInformation fileInfo = f.getFileInformation(); - final FileBasicInformation fileBasicInfo = fileInfo.getBasicInformation(); - final long fileSize = fileInfo.getStandardInformation().getEndOfFile(); final Map attributes = new HashMap<>(); - attributes.put(CoreAttributes.FILENAME.key(), filename); - attributes.put(CoreAttributes.PATH.key(), filePath); - attributes.put(CoreAttributes.ABSOLUTE_PATH.key(), "\\\\" + hostname + "\\" + shareName + "\\" + file); - attributes.put(FILE_CREATION_TIME_ATTRIBUTE, dateFormatter.format(fileBasicInfo.getCreationTime().toInstant().atZone(ZoneId.systemDefault()))); - attributes.put(FILE_LAST_ACCESS_TIME_ATTRIBUTE, dateFormatter.format(fileBasicInfo.getLastAccessTime().toInstant().atZone(ZoneId.systemDefault()))); - attributes.put(FILE_LAST_MODIFY_TIME_ATTRIBUTE, dateFormatter.format(fileBasicInfo.getLastWriteTime().toInstant().atZone(ZoneId.systemDefault()))); - attributes.put(FILE_SIZE_ATTRIBUTE, String.valueOf(fileSize)); + attributes.put(CoreAttributes.FILENAME.key(), fileInfo.filename()); + attributes.put(CoreAttributes.PATH.key(), fileInfo.path()); + attributes.put(CoreAttributes.ABSOLUTE_PATH.key(), "\\\\" + hostname + "\\" + shareName + "\\" + fullPath); + attributes.put(FILE_CREATION_TIME_ATTRIBUTE, dateFormatter.format(Instant.ofEpochMilli(fileInfo.creationTime()).atZone(ZoneId.systemDefault()))); + attributes.put(FILE_LAST_ACCESS_TIME_ATTRIBUTE, dateFormatter.format(Instant.ofEpochMilli(fileInfo.lastAccessTime()).atZone(ZoneId.systemDefault()))); + attributes.put(FILE_LAST_MODIFY_TIME_ATTRIBUTE, dateFormatter.format(Instant.ofEpochMilli(fileInfo.lastModifiedTime()).atZone(ZoneId.systemDefault()))); + attributes.put(FILE_SIZE_ATTRIBUTE, String.valueOf(fileInfo.size())); attributes.put(HOSTNAME.getName(), hostname); attributes.put(SHARE.getName(), shareName); flowFile = session.putAllAttributes(flowFile, attributes); - session.getProvenanceReporter().receive(flowFile, uri.toString(), importMillis); + session.getProvenanceReporter().receive(flowFile, transitUri, importMillis); session.transfer(flowFile, REL_SUCCESS); - } catch (SMBApiException e) { + } catch (SmbException e) { // do not fail whole batch if a single file cannot be accessed - if (e.getStatus() == NtStatus.STATUS_SHARING_VIOLATION) { - logger.info("Could not acquire sharing access for file {}", file); + if (e.getErrorCode() == ERROR_CODE_SHARING_VIOLATION) { + logger.info("Could not acquire sharing access for file {}", fullPath); if (flowFile != null) { session.remove(flowFile); } @@ -520,17 +454,17 @@ public void onTrigger(final ProcessContext context, final ProcessSession session try { if (!keepingSourceFile) { - share.rm(file); + client.deleteFile(fullPath); } - } catch (SMBApiException e) { - logger.error("Could not remove file {}", file); + } catch (SmbException e) { + logger.error("Could not remove file {}", fullPath); } if (!isScheduled()) { // if processor stopped, put the rest of the files back on the queue. queueLock.lock(); try { while (itr.hasNext()) { - final String nextFile = itr.next(); + final SmbFileInfo nextFile = itr.next(); fileQueue.add(nextFile); inProcess.remove(nextFile); } @@ -560,7 +494,16 @@ public void onTrigger(final ProcessContext context, final ProcessSession session } catch (Exception e) { logger.error("Could not establish smb connection", e); context.yield(); - smbClient.getServerList().unregister(hostname); } } + + private record SmbFileInfo( + String filename, + String path, + long size, + boolean hidden, + long creationTime, + long lastModifiedTime, + long lastAccessTime + ) { } } diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/ListSmb.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/ListSmb.java index 933db256076b..19afe4b818aa 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/ListSmb.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/ListSmb.java @@ -454,7 +454,7 @@ private Stream performListing(ProcessContext context) throws context.getProperty(SMB_CLIENT_PROVIDER_SERVICE).asControllerService(SmbClientProviderService.class); final String directory = getDirectory(context); final SmbClientService clientService = clientProviderService.getClient(getLogger()); - return clientService.listFiles(directory).onClose(() -> { + return clientService.listFiles(directory, true).onClose(() -> { try { clientService.close(); } catch (Exception e) { diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java index 25b0ca18b2e0..cfda90ec2988 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java @@ -16,18 +16,6 @@ */ package org.apache.nifi.processors.smb; -import com.hierynomus.msdtyp.AccessMask; -import com.hierynomus.msfscc.FileAttributes; -import com.hierynomus.mssmb2.SMB2CreateDisposition; -import com.hierynomus.mssmb2.SMB2CreateOptions; -import com.hierynomus.mssmb2.SMB2ShareAccess; -import com.hierynomus.smbj.SMBClient; -import com.hierynomus.smbj.auth.AuthenticationContext; -import com.hierynomus.smbj.connection.Connection; -import com.hierynomus.smbj.session.Session; -import com.hierynomus.smbj.share.DiskEntry; -import com.hierynomus.smbj.share.DiskShare; -import com.hierynomus.smbj.share.File; import org.apache.nifi.annotation.behavior.InputRequirement; import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; import org.apache.nifi.annotation.behavior.ReadsAttribute; @@ -51,32 +39,43 @@ import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.util.StandardValidators; -import org.apache.nifi.processors.smb.util.HostnameAndShareFlowFileFilter; +import org.apache.nifi.processors.smb.util.ServiceLocationFlowFileFilter; +import org.apache.nifi.services.smb.SmbClientProvider; +import org.apache.nifi.services.smb.SmbClientProviderService; +import org.apache.nifi.services.smb.SmbClientService; +import org.apache.nifi.services.smb.SmbShareAccess; +import org.apache.nifi.services.smb.SmbjClientProvider; import org.apache.nifi.util.StringUtils; -import java.io.OutputStream; import java.net.URI; import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; -import java.util.EnumSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; -import static org.apache.nifi.smb.common.SmbProperties.ENABLE_DFS; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.CONNECTION_CONFIGURATION_STRATEGY; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.ConnectionConfigurationStrategy; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.DOMAIN; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.ENABLE_DFS; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.HOSTNAME; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.PASSWORD; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.PORT; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.SHARE; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.SMB_CLIENT_PROVIDER_SERVICE; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.SMB_DIALECT; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.TIMEOUT; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.USERNAME; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.USE_ENCRYPTION; import static org.apache.nifi.smb.common.SmbProperties.OLD_ENABLE_DFS_PROPERTY_NAME; import static org.apache.nifi.smb.common.SmbProperties.OLD_SMB_DIALECT_PROPERTY_NAME; import static org.apache.nifi.smb.common.SmbProperties.OLD_TIMEOUT_PROPERTY_NAME; import static org.apache.nifi.smb.common.SmbProperties.OLD_USE_ENCRYPTION_PROPERTY_NAME; -import static org.apache.nifi.smb.common.SmbProperties.SMB_DIALECT; -import static org.apache.nifi.smb.common.SmbProperties.TIMEOUT; -import static org.apache.nifi.smb.common.SmbProperties.USE_ENCRYPTION; -import static org.apache.nifi.smb.common.SmbUtils.buildSmbClient; @InputRequirement(Requirement.INPUT_REQUIRED) @Tags({"samba, smb, cifs, files, put"}) -@CapabilityDescription("Writes the contents of a FlowFile to a samba network location. " + +@CapabilityDescription("Writes the contents of a FlowFile to an SMB network location (e.g. Samba or Windows Server). " + "Use this processor instead of a cifs mounts if share access control is important." + "Configure the Hostname, Share and Directory accordingly: \\\\[Hostname]\\[Share]\\[path\\to\\Directory]") @SeeAlso({GetSmbFile.class, ListSmb.class, FetchSmb.class}) @@ -94,28 +93,6 @@ public class PutSmbFile extends AbstractProcessor { public static final String IGNORE_RESOLUTION = "ignore"; public static final String FAIL_RESOLUTION = "fail"; - public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder() - .name("Hostname") - .description("The network host to which files should be written.") - .required(true) - .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) - .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) - .build(); - public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder() - .name("Port") - .description("The port to use for the SMB connection.") - .required(true) - .addValidator(StandardValidators.PORT_VALIDATOR) - .defaultValue("445") - .build(); - public static final PropertyDescriptor SHARE = new PropertyDescriptor.Builder() - .name("Share") - .description("The network share to which files should be written. This is the \"first folder\"" + - "after the hostname: \\\\hostname\\[share]\\dir1\\dir2") - .required(true) - .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) - .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) - .build(); public static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder() .name("Directory") .description("The network folder to which files should be written. This is the remaining relative " + @@ -124,25 +101,6 @@ public class PutSmbFile extends AbstractProcessor { .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) .build(); - public static final PropertyDescriptor DOMAIN = new PropertyDescriptor.Builder() - .name("Domain") - .description("The domain used for authentication. Optional, in most cases username and password is sufficient.") - .required(false) - .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) - .build(); - public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() - .name("Username") - .description("The username used for authentication. If no username is set then anonymous authentication is attempted.") - .required(false) - .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) - .build(); - public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() - .name("Password") - .description("The password used for authentication. Required if Username is set.") - .required(false) - .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) - .sensitive(true) - .build(); public static final PropertyDescriptor CREATE_DIRS = new PropertyDescriptor.Builder() .name("Create Missing Directories") .description("If true, then missing destination directories will be created. If false, FlowFiles are penalized and sent to failure.") @@ -189,6 +147,8 @@ public class PutSmbFile extends AbstractProcessor { .build(); private static final List PROPERTY_DESCRIPTORS = List.of( + CONNECTION_CONFIGURATION_STRATEGY, + SMB_CLIENT_PROVIDER_SERVICE, HOSTNAME, PORT, SHARE, @@ -211,8 +171,8 @@ public class PutSmbFile extends AbstractProcessor { REL_FAILURE ); - private SMBClient smbClient = null; // this gets synchronized when the `connect` method is called - private Set sharedAccess; + private SmbClientProvider clientProvider; + private Set sharedAccess; @Override public Set getRelationships() { @@ -226,30 +186,33 @@ public final List getSupportedPropertyDescriptors() { @OnScheduled public void onScheduled(final ProcessContext context) { - smbClient = initSmbClient(context); + clientProvider = switch (context.getProperty(CONNECTION_CONFIGURATION_STRATEGY).asAllowableValue(ConnectionConfigurationStrategy.class)) { + case CONTROLLER_SERVICE -> context.getProperty(SMB_CLIENT_PROVIDER_SERVICE).asControllerService(SmbClientProviderService.class); + case LOCAL_PROPERTIES -> new SmbjClientProvider(context, getLogger()); + }; switch (context.getProperty(SHARE_ACCESS).getValue()) { case SHARE_ACCESS_NONE: - sharedAccess = Collections.emptySet(); + sharedAccess = SmbShareAccess.NONE; break; case SHARE_ACCESS_READ: - sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ); + sharedAccess = SmbShareAccess.READ; break; case SHARE_ACCESS_READDELETE: - sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ, SMB2ShareAccess.FILE_SHARE_DELETE); + sharedAccess = SmbShareAccess.READ_DELETE; break; case SHARE_ACCESS_READWRITEDELETE: - sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ, SMB2ShareAccess.FILE_SHARE_WRITE, SMB2ShareAccess.FILE_SHARE_DELETE); + sharedAccess = SmbShareAccess.READ_WRITE_DELETE; break; } } @OnStopped public void onStopped() { - if (smbClient != null) { - smbClient.close(); - smbClient = null; + if (clientProvider instanceof SmbjClientProvider smbjClientProvider) { + smbjClientProvider.close(); } + clientProvider = null; } @Override @@ -263,34 +226,14 @@ public void migrateProperties(PropertyConfiguration config) { @Override protected Collection customValidate(ValidationContext validationContext) { Collection set = new ArrayList<>(); - if (validationContext.getProperty(USERNAME).isSet() && !validationContext.getProperty(PASSWORD).isSet()) { - set.add(new ValidationResult.Builder().explanation("Password must be set if username is supplied.").build()); - } - return set; - } - - SMBClient initSmbClient(final ProcessContext context) { - return buildSmbClient(context); - } - - private void createMissingDirectoriesRecursively(ComponentLog logger, DiskShare share, String pathToCreate) { - int index = 0; - while (index < pathToCreate.length()) { - index = pathToCreate.indexOf(PATH_SEPARATOR, index); - - if (index == -1) { - index = pathToCreate.length(); - } - - String path = pathToCreate.substring(0, index++); - if (!share.folderExists(path)) { - logger.debug("Creating folder {}", path); - share.mkdir(path); - } else { - logger.debug("Folder already exists {}. Moving on", path); + if (validationContext.getProperty(CONNECTION_CONFIGURATION_STRATEGY).asAllowableValue(ConnectionConfigurationStrategy.class) == ConnectionConfigurationStrategy.LOCAL_PROPERTIES) { + if (validationContext.getProperty(USERNAME).isSet() && !validationContext.getProperty(PASSWORD).isSet()) { + set.add(new ValidationResult.Builder().explanation("Password must be set if username is supplied.").build()); } } + + return set; } String normalizePath(String path) { @@ -306,35 +249,19 @@ String normalizePath(String path) { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { final int batchSize = context.getProperty(BATCH_SIZE).asInteger(); - final HostnameAndShareFlowFileFilter flowFileFilter = new HostnameAndShareFlowFileFilter(context, batchSize); + final ServiceLocationFlowFileFilter flowFileFilter = new ServiceLocationFlowFileFilter(clientProvider, batchSize); final List flowFiles = session.get(flowFileFilter); if (flowFiles.isEmpty()) { return; } + final ComponentLog logger = getLogger(); logger.debug("Processing next {} FlowFiles", flowFiles.size()); - final String hostname = flowFileFilter.getHostName(); - final String shareName = flowFileFilter.getShare(); - final int port = context.getProperty(PORT).asInteger(); - final String domain = context.getProperty(DOMAIN).getValue(); - final String username = context.getProperty(USERNAME).getValue(); - String password = context.getProperty(PASSWORD).getValue(); - - AuthenticationContext ac = null; - if (username != null && password != null) { - ac = new AuthenticationContext( - username, - password.toCharArray(), - domain); - } else { - ac = AuthenticationContext.anonymous(); - } - - try (Connection connection = smbClient.connect(hostname, port); - Session smbSession = connection.authenticate(ac); - DiskShare share = (DiskShare) smbSession.connectShare(shareName)) { + final URI serviceLocation = flowFileFilter.getSelectedServiceLocation(); + final Map attributes = flowFileFilter.getSelectedAttributes(); + try (SmbClientService client = clientProvider.getClient(getLogger(), attributes)) { for (FlowFile flowFile : flowFiles) { try { final long processingStartTime = System.nanoTime(); @@ -355,7 +282,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session // handle missing directory final Boolean createMissingDirectories = context.getProperty(CREATE_DIRS).asBoolean(); - if (StringUtils.isNotBlank(destinationDirectory) && !share.folderExists(destinationDirectory)) { + if (StringUtils.isNotBlank(destinationDirectory) && !client.folderExists(destinationDirectory)) { if (!createMissingDirectories) { logger.warn("Penalizing {} and routing to failure as configured because the destination directory ({}) doesn't exist", flowFile, destinationDirectory); flowFile = session.penalize(flowFile); @@ -363,7 +290,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session continue; } else { try { - createMissingDirectoriesRecursively(logger, share, destinationDirectory); + client.ensureDirectory(destinationDirectory); } catch (Exception e) { logger.error("Penalizing {} and routing to failure because failed to create missing destination directories ({})", flowFile, destinationDirectory, e); flowFile = session.penalize(flowFile); @@ -375,7 +302,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session // handle conflict resolution final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue(); - if (share.fileExists(destinationFullPath)) { + if (client.fileExists(destinationFullPath)) { if (conflictResolution.equals(IGNORE_RESOLUTION)) { logger.info("Transferring {} to success as configured because file with same name already exists", flowFile); session.transfer(flowFile, REL_SUCCESS); @@ -399,16 +326,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session } // handle the transfer - try ( - File shareDestinationFile = share.openFile( - transferDestinationFullPath, - EnumSet.of(AccessMask.GENERIC_WRITE), - EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL), - sharedAccess, - SMB2CreateDisposition.FILE_OVERWRITE_IF, - EnumSet.of(SMB2CreateOptions.FILE_WRITE_THROUGH)); - OutputStream shareDestinationFileOutputStream = shareDestinationFile.getOutputStream()) { - session.exportTo(flowFile, shareDestinationFileOutputStream); + try { + session.read(flowFile, inputStream -> client.writeFile(transferDestinationFullPath, inputStream, sharedAccess)); } catch (Exception e) { logger.error("Cannot transfer the file. Penalizing {} and routing to 'failure'", flowFile, e); flowFile = session.penalize(flowFile); @@ -418,16 +337,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session // handle the rename if (renameSuffix) { - try (DiskEntry fileDiskEntry = share.open( - transferDestinationFullPath, - EnumSet.of(AccessMask.DELETE, AccessMask.GENERIC_WRITE), - EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL), - sharedAccess, - SMB2CreateDisposition.FILE_OPEN, - EnumSet.of(SMB2CreateOptions.FILE_WRITE_THROUGH))) { - - // rename the file on the share and replace it in case it exists - fileDiskEntry.rename(destinationFullPath, true); + try { + client.renameFile(transferDestinationFullPath, destinationFullPath, true); } catch (Exception e) { logger.error("Cannot rename the file. Penalizing {} and routing to 'failure'", flowFile, e); flowFile = session.penalize(flowFile); @@ -437,10 +348,10 @@ public void onTrigger(final ProcessContext context, final ProcessSession session } // handle the success - final URI provenanceUri = new URI("smb", hostname, "/" + destinationFullPath.replace('\\', '/'), null); + final String transitUri = String.format("%s/%s", serviceLocation, destinationFullPath.replace('\\', '/')); final long processingTimeInNano = System.nanoTime() - processingStartTime; final long processingTimeInMilli = TimeUnit.MILLISECONDS.convert(processingTimeInNano, TimeUnit.NANOSECONDS); - session.getProvenanceReporter().send(flowFile, provenanceUri.toString(), processingTimeInMilli); + session.getProvenanceReporter().send(flowFile, transitUri, processingTimeInMilli); session.transfer(flowFile, REL_SUCCESS); } catch (Exception e) { logger.error("Error processing flowfile {}", flowFile, e); @@ -451,7 +362,6 @@ public void onTrigger(final ProcessContext context, final ProcessSession session } catch (Exception e) { logger.error("Could not establish smb connection", e); session.transfer(flowFiles, REL_FAILURE); - smbClient.getServerList().unregister(hostname); } } } diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/HostnameAndShareFlowFileFilter.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/HostnameAndShareFlowFileFilter.java deleted file mode 100644 index 8473e8edab6a..000000000000 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/HostnameAndShareFlowFileFilter.java +++ /dev/null @@ -1,77 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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. - */ -package org.apache.nifi.processors.smb.util; - -import org.apache.nifi.flowfile.FlowFile; -import org.apache.nifi.processor.FlowFileFilter; -import org.apache.nifi.processor.ProcessContext; - -import static org.apache.nifi.processor.FlowFileFilter.FlowFileFilterResult.REJECT_AND_TERMINATE; -import static org.apache.nifi.processors.smb.PutSmbFile.HOSTNAME; -import static org.apache.nifi.processors.smb.PutSmbFile.SHARE; - -public class HostnameAndShareFlowFileFilter implements FlowFileFilter { - - private final ProcessContext context; - private final int batchSize; - - private HostSharePair selectedHostSharePair; - private int count = 0; - - public HostnameAndShareFlowFileFilter(ProcessContext context, int batchSize) { - this.context = context; - this.batchSize = batchSize; - } - - @Override - public FlowFileFilterResult filter(FlowFile flowFile) { - final HostSharePair hostSharePair = getFlowFileHostSharePair(flowFile); - - if (selectedHostSharePair == null) { - selectedHostSharePair = hostSharePair; - } - - if (count >= batchSize) { - return REJECT_AND_TERMINATE; - } - - if (selectedHostSharePair.hostName().equals(hostSharePair.hostName()) && selectedHostSharePair.share().equals(hostSharePair.share())) { - count += 1; - return FlowFileFilterResult.ACCEPT_AND_CONTINUE; - } else { - return FlowFileFilterResult.REJECT_AND_CONTINUE; - } - } - - private HostSharePair getFlowFileHostSharePair(final FlowFile flowFile) { - final String hostName = context.getProperty(HOSTNAME).evaluateAttributeExpressions(flowFile).getValue(); - final String share = context.getProperty(SHARE).evaluateAttributeExpressions(flowFile).getValue(); - - return new HostSharePair(hostName, share); - } - - public String getHostName() { - return selectedHostSharePair.hostName(); - } - - public String getShare() { - return selectedHostSharePair.share(); - } - - record HostSharePair(String hostName, String share) { - } -} diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/LocalSmbProperties.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/LocalSmbProperties.java new file mode 100644 index 000000000000..e8dafa5d2f94 --- /dev/null +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/LocalSmbProperties.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.smb.util; + +import org.apache.nifi.components.DescribedValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.services.smb.SmbClientProviderService; +import org.apache.nifi.smb.common.SmbProperties; + +public class LocalSmbProperties { + + private static final String DEPRECATION_TAG = " Deprecation notice: Local processor properties used to configure the SMB connection are deprecated" + + " and will be removed in the next major release. Use SMB Client Provider Service instead."; + + public enum ConnectionConfigurationStrategy implements DescribedValue { + CONTROLLER_SERVICE("Controller Service", "Use SMB Client Provider Service to configure the SMB connection."), + LOCAL_PROPERTIES("Local Properties", "Use local processor properties to configure the SMB connection." + DEPRECATION_TAG); + + private final String displayName; + private final String description; + + ConnectionConfigurationStrategy(final String displayName, final String description) { + this.displayName = displayName; + this.description = description; + } + + @Override + public String getValue() { + return displayName; + } + + @Override + public String getDisplayName() { + return displayName; + } + + @Override + public String getDescription() { + return description; + } + } + + public static final PropertyDescriptor CONNECTION_CONFIGURATION_STRATEGY = new PropertyDescriptor.Builder() + .name("Connection Configuration Strategy") + .description("Specifies whether SMB Client Provider Service or local processor properties are used to configure the SMB connection.") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .allowableValues(ConnectionConfigurationStrategy.class) + .defaultValue(ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor SMB_CLIENT_PROVIDER_SERVICE = new PropertyDescriptor.Builder() + .name("SMB Client Provider Service") + .description("Specifies the SMB client provider to use for creating SMB connections.") + .required(true) + .identifiesControllerService(SmbClientProviderService.class) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.CONTROLLER_SERVICE) + .build(); + + public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.HOSTNAME) + .description(SmbProperties.HOSTNAME.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.PORT) + .description(SmbProperties.PORT.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor SHARE = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.SHARE) + .description(SmbProperties.SHARE.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor DOMAIN = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.DOMAIN) + .description(SmbProperties.DOMAIN.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.USERNAME) + .description(SmbProperties.USERNAME.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.PASSWORD) + .description(SmbProperties.PASSWORD.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor SMB_DIALECT = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.SMB_DIALECT) + .description(SmbProperties.SMB_DIALECT.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor USE_ENCRYPTION = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.USE_ENCRYPTION) + .description(SmbProperties.USE_ENCRYPTION.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor ENABLE_DFS = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.ENABLE_DFS) + .description(SmbProperties.ENABLE_DFS.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + + public static final PropertyDescriptor TIMEOUT = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(SmbProperties.TIMEOUT) + .description(SmbProperties.TIMEOUT.getDescription() + DEPRECATION_TAG) + .dependsOn(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.LOCAL_PROPERTIES) + .build(); + +} diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/ServiceLocationFlowFileFilter.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/ServiceLocationFlowFileFilter.java new file mode 100644 index 000000000000..608ab07e4ece --- /dev/null +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/util/ServiceLocationFlowFileFilter.java @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.smb.util; + +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.FlowFileFilter; +import org.apache.nifi.services.smb.SmbClientProvider; + +import java.net.URI; +import java.util.Map; + +import static org.apache.nifi.processor.FlowFileFilter.FlowFileFilterResult.ACCEPT_AND_CONTINUE; +import static org.apache.nifi.processor.FlowFileFilter.FlowFileFilterResult.REJECT_AND_CONTINUE; +import static org.apache.nifi.processor.FlowFileFilter.FlowFileFilterResult.REJECT_AND_TERMINATE; + +public class ServiceLocationFlowFileFilter implements FlowFileFilter { + + private final SmbClientProvider clientProvider; + private final int batchSize; + + private URI selectedServiceLocation; + private Map selectedAttributes; + private int count = 0; + + public ServiceLocationFlowFileFilter(SmbClientProvider clientProvider, int batchSize) { + this.clientProvider = clientProvider; + this.batchSize = batchSize; + } + + @Override + public FlowFileFilterResult filter(FlowFile flowFile) { + final Map attributes = flowFile.getAttributes(); + final URI serviceLocation = clientProvider.getServiceLocation(attributes); + + if (selectedServiceLocation == null) { + selectedServiceLocation = serviceLocation; + selectedAttributes = attributes; + } + + if (count >= batchSize) { + return REJECT_AND_TERMINATE; + } + + if (selectedServiceLocation.equals(serviceLocation)) { + count += 1; + return ACCEPT_AND_CONTINUE; + } else { + return REJECT_AND_CONTINUE; + } + } + + public URI getSelectedServiceLocation() { + return selectedServiceLocation; + } + + public Map getSelectedAttributes() { + return selectedAttributes; + } +} diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/FetchSmbTest.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/FetchSmbTest.java index 70dcde131ff3..83ffeba5fb24 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/FetchSmbTest.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/FetchSmbTest.java @@ -43,6 +43,8 @@ import static org.apache.nifi.util.TestRunners.newTestRunner; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.ArgumentMatchers.anySet; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; @@ -64,7 +66,7 @@ class FetchSmbTest { @BeforeEach public void beforeEach() throws Exception { mockCloseable = MockitoAnnotations.openMocks(this); - when(clientProviderService.getClient(any(ComponentLog.class))).thenReturn(mockNifiSmbClientService); + when(clientProviderService.getClient(any(ComponentLog.class), anyMap())).thenReturn(mockNifiSmbClientService); when(clientProviderService.getIdentifier()).thenReturn(CLIENT_SERVICE_PROVIDER_ID); when(clientProviderService.getServiceLocation()).thenReturn(URI.create("smb://localhost:445/share")); } @@ -116,14 +118,14 @@ public void noSuchAttributeReferencedInELShouldResultInFailure() throws Exceptio private void mockNifiSmbClientService() throws IOException { doThrow(new SmbException("test exception", 1L, new RuntimeException())).when(mockNifiSmbClientService) - .readFile(anyString(), any(OutputStream.class)); + .readFile(anyString(), any(OutputStream.class), anySet()); doAnswer(invocation -> { final OutputStream o = invocation.getArgument(1); final ByteArrayInputStream bytes = new ByteArrayInputStream("content".getBytes()); IOUtils.copy(bytes, o); return true; }).when(mockNifiSmbClientService) - .readFile(eq("testDirectory/canReadThis"), any(OutputStream.class)); + .readFile(eq("testDirectory/canReadThis"), any(OutputStream.class), anySet()); } private TestRunner createRunner() throws Exception { diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java index f883dc114e30..2bdf3ce9c87a 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java @@ -16,21 +16,11 @@ */ package org.apache.nifi.processors.smb; -import com.hierynomus.msdtyp.FileTime; -import com.hierynomus.msfscc.FileAttributes; -import com.hierynomus.msfscc.fileinformation.FileAllInformation; -import com.hierynomus.msfscc.fileinformation.FileBasicInformation; -import com.hierynomus.msfscc.fileinformation.FileIdBothDirectoryInformation; -import com.hierynomus.msfscc.fileinformation.FileStandardInformation; -import com.hierynomus.mssmb2.SMB2CreateDisposition; -import com.hierynomus.smbj.SMBClient; -import com.hierynomus.smbj.auth.AuthenticationContext; -import com.hierynomus.smbj.connection.Connection; -import com.hierynomus.smbj.session.Session; -import com.hierynomus.smbj.share.DiskShare; -import com.hierynomus.smbj.share.File; import org.apache.nifi.flowfile.attributes.CoreAttributes; -import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.services.smb.SmbClientProviderService; +import org.apache.nifi.services.smb.SmbClientService; +import org.apache.nifi.services.smb.SmbListableEntity; import org.apache.nifi.util.MockFlowFile; import org.apache.nifi.util.TestRunner; import org.apache.nifi.util.TestRunners; @@ -40,163 +30,105 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import java.io.ByteArrayInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.nio.charset.StandardCharsets; -import java.util.ArrayList; +import java.net.URI; import java.util.List; +import java.util.stream.Stream; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.CONNECTION_CONFIGURATION_STRATEGY; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.ConnectionConfigurationStrategy; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.SMB_CLIENT_PROVIDER_SERVICE; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anySet; -import static org.mockito.Mockito.eq; -import static org.mockito.Mockito.lenient; -import static org.mockito.Mockito.mock; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) public class GetSmbFileTest { - private TestRunner testRunner; - - @Mock - private SMBClient smbClient; - @Mock - private Connection connection; + private TestRunner testRunner; @Mock - private Session session; + private SmbClientProviderService clientProviderService; @Mock - private DiskShare diskShare; + private SmbClientService clientService; private static final String HOSTNAME = "host"; private static final String SHARE = "share"; private static final String DIRECTORY = "nifi\\input"; - private static final String USERNAME = "user"; - private static final String PASSWORD = "pass"; - private void setupSmbProcessor() throws IOException { - when(smbClient.connect(any(String.class))).thenReturn(connection); - when(connection.authenticate(any(AuthenticationContext.class))).thenReturn(session); - when(session.connectShare(SHARE)).thenReturn(diskShare); + private static final String CLIENT_SERVICE_ID = "client-provider-service-id"; - testRunner.setProperty(GetSmbFile.HOSTNAME, HOSTNAME); - testRunner.setProperty(GetSmbFile.SHARE, SHARE); + @BeforeEach + public void init() throws Exception { + when(clientProviderService.getIdentifier()).thenReturn(CLIENT_SERVICE_ID); + when(clientProviderService.getServiceLocation()).thenReturn(URI.create("smb://" + HOSTNAME + ":445/" + SHARE)); + when(clientProviderService.getClient(any(ComponentLog.class))).thenReturn(clientService); + + testRunner = TestRunners.newTestRunner(GetSmbFile.class); + testRunner.addControllerService(CLIENT_SERVICE_ID, clientProviderService); + testRunner.enableControllerService(clientProviderService); + testRunner.setProperty(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.CONTROLLER_SERVICE.getValue()); + testRunner.setProperty(SMB_CLIENT_PROVIDER_SERVICE, CLIENT_SERVICE_ID); testRunner.setProperty(GetSmbFile.DIRECTORY, DIRECTORY); - testRunner.setProperty(GetSmbFile.USERNAME, USERNAME); - testRunner.setProperty(GetSmbFile.PASSWORD, PASSWORD); - } - - private FileIdBothDirectoryInformation mockFile(String path, String filename, String fileContent, long fileAttributes) { - File smbfile = mock(File.class); - final String fullpath = path + "\\" + filename; - lenient().when(diskShare.openFile( - eq(fullpath), - anySet(), - anySet(), - anySet(), - any(SMB2CreateDisposition.class), - anySet() - )).thenReturn(smbfile); - lenient().when(smbfile.getUncPath()).thenReturn(filename); - - if (fileContent != null) { - InputStream is = new ByteArrayInputStream(fileContent.getBytes(StandardCharsets.UTF_8)); - lenient().when(smbfile.getInputStream()).thenReturn(is); - } - - FileIdBothDirectoryInformation fdInfo = mock(FileIdBothDirectoryInformation.class); - lenient().when(fdInfo.getFileName()).thenReturn(filename); - lenient().when(fdInfo.getFileAttributes()).thenReturn(fileAttributes); - - FileAllInformation fileAllInfo = mock(FileAllInformation.class); - FileTime fileTime = FileTime.ofEpochMillis(0); - FileBasicInformation fileBasicInfo = new FileBasicInformation(fileTime, fileTime, fileTime, fileTime, 0); - FileStandardInformation fileStandardInformation = mock(FileStandardInformation.class); - - lenient().when(smbfile.getFileInformation()).thenReturn(fileAllInfo); - lenient().when(fileAllInfo.getBasicInformation()).thenReturn(fileBasicInfo); - lenient().when(fileAllInfo.getStandardInformation()).thenReturn(fileStandardInformation); - lenient().when(fileStandardInformation.getEndOfFile()).thenReturn((long) 0); - - return fdInfo; } - private FileIdBothDirectoryInformation mockFile(String path, String filename, String fileContent) { - return mockFile(path, filename, fileContent, FileAttributes.FILE_ATTRIBUTE_NORMAL.getValue()); + private SmbListableEntity fileEntity(String path, String filename, boolean hidden) { + return SmbListableEntity.builder() + .setName(filename) + .setPath(path.replace('\\', '/')) + .setHidden(hidden) + .build(); } - private void verifyOpenFile(String path, String filename, int times) { - final String fullpath = path + "\\" + filename; - verify(diskShare, times(times)).openFile( - eq(fullpath), - anySet(), - anySet(), - anySet(), - any(SMB2CreateDisposition.class), - anySet() - ); + private SmbListableEntity fileEntity(String path, String filename) { + return fileEntity(path, filename, false); } - private FileIdBothDirectoryInformation mockDir(String path, List files) { - final String[] fileSplits = path.split("\\\\"); - final String filename = fileSplits[fileSplits.length - 1]; - lenient().when(diskShare.folderExists(path)).thenReturn(true); - lenient().when(diskShare.list(path)).thenReturn(files); - - FileIdBothDirectoryInformation fdInfo = mock(FileIdBothDirectoryInformation.class); - lenient().when(fdInfo.getFileName()).thenReturn(filename); - lenient().when(fdInfo.getFileAttributes()).thenReturn(FileAttributes.FILE_ATTRIBUTE_DIRECTORY.getValue()); - return fdInfo; + private void mockListing(SmbListableEntity... entities) { + when(clientService.listFiles(anyString(), anyBoolean())).thenAnswer(inv -> Stream.of(entities)); } - @BeforeEach - public void init() throws IOException { - testRunner = TestRunners.newTestRunner(new GetSmbFile() { - @Override - SMBClient initSmbClient(ProcessContext context) { - return smbClient; - } - }); - setupSmbProcessor(); + private void verifyReadFile(String path, String filename, int times) { + verify(clientService, times(times)).readFile(eq(path + "\\" + filename), any(), any()); } @Test public void testOpenFileCalled() { - FileIdBothDirectoryInformation file1 = mockFile(DIRECTORY, "file1.txt", "abc"); - mockDir(DIRECTORY, List.of(file1)); + mockListing(fileEntity(DIRECTORY, "file1.txt")); testRunner.run(); - verifyOpenFile(DIRECTORY, "file1.txt", 1); - verifyOpenFile(DIRECTORY, "file2.txt", 0); + verifyReadFile(DIRECTORY, "file1.txt", 1); + verifyReadFile(DIRECTORY, "file2.txt", 0); } @Test public void testHiddenFile() { testRunner.setProperty(GetSmbFile.IGNORE_HIDDEN_FILES, "true"); - FileIdBothDirectoryInformation file1 = mockFile(DIRECTORY, "file1.txt", "abc", FileAttributes.FILE_ATTRIBUTE_HIDDEN.getValue()); - FileIdBothDirectoryInformation file2 = mockFile(DIRECTORY, "file2.txt", "abc", FileAttributes.FILE_ATTRIBUTE_NORMAL.getValue()); - mockDir(DIRECTORY, List.of(file1, file2)); + mockListing( + fileEntity(DIRECTORY, "file1.txt", true), + fileEntity(DIRECTORY, "file2.txt", false) + ); testRunner.run(); - verifyOpenFile(DIRECTORY, "file1.txt", 0); - verifyOpenFile(DIRECTORY, "file2.txt", 1); + verifyReadFile(DIRECTORY, "file1.txt", 0); + verifyReadFile(DIRECTORY, "file2.txt", 1); } @Test public void testFileFilter() { testRunner.setProperty(GetSmbFile.FILE_FILTER, "file[0-9]\\.txt"); - mockDir(DIRECTORY, List.of( - mockFile(DIRECTORY, "something_else.txt", "abc"), - mockFile(DIRECTORY, "file1.txt", "abc"), - mockFile(DIRECTORY, "file2.txt", "abc") - )); + mockListing( + fileEntity(DIRECTORY, "something_else.txt"), + fileEntity(DIRECTORY, "file1.txt"), + fileEntity(DIRECTORY, "file2.txt") + ); testRunner.run(); - verifyOpenFile(DIRECTORY, "something_else.txt", 0); - verifyOpenFile(DIRECTORY, "file1.txt", 1); - verifyOpenFile(DIRECTORY, "file2.txt", 1); + verifyReadFile(DIRECTORY, "something_else.txt", 0); + verifyReadFile(DIRECTORY, "file1.txt", 1); + verifyReadFile(DIRECTORY, "file2.txt", 1); testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, 2); } @@ -204,16 +136,14 @@ public void testFileFilter() { public void testNonRecurse() { testRunner.setProperty(GetSmbFile.RECURSE, "false"); String subdir = DIRECTORY + "\\subdir1"; - mockDir(DIRECTORY, List.of( - mockFile(DIRECTORY, "file1.txt", "abc"), - mockFile(DIRECTORY, "file2.txt", "abc"), - mockDir(subdir, List.of(mockFile(subdir, "file3.txt", "abc"))) - )); - + mockListing( + fileEntity(DIRECTORY, "file1.txt"), + fileEntity(DIRECTORY, "file2.txt") + ); testRunner.run(); - verifyOpenFile(DIRECTORY, "file1.txt", 1); - verifyOpenFile(DIRECTORY, "file2.txt", 1); - verifyOpenFile(subdir, "file3.txt", 0); + verifyReadFile(DIRECTORY, "file1.txt", 1); + verifyReadFile(DIRECTORY, "file2.txt", 1); + verifyReadFile(subdir, "file3.txt", 0); testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, 2); } @@ -221,17 +151,15 @@ public void testNonRecurse() { public void testRecurse() { testRunner.setProperty(GetSmbFile.RECURSE, "true"); String subdir = DIRECTORY + "\\subdir1"; - mockDir(DIRECTORY, List.of( - mockFile(DIRECTORY, "file1.txt", "abc"), - mockFile(DIRECTORY, "file2.txt", "abc"), - mockDir(subdir, List.of(mockFile(subdir, "file3.txt", "abc"))) - ) + mockListing( + fileEntity(DIRECTORY, "file1.txt"), + fileEntity(DIRECTORY, "file2.txt"), + fileEntity(subdir, "file3.txt") ); - testRunner.run(); - verifyOpenFile(DIRECTORY, "file1.txt", 1); - verifyOpenFile(DIRECTORY, "file2.txt", 1); - verifyOpenFile(subdir, "file3.txt", 1); + verifyReadFile(DIRECTORY, "file1.txt", 1); + verifyReadFile(DIRECTORY, "file2.txt", 1); + verifyReadFile(subdir, "file3.txt", 1); testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, 3); } @@ -242,16 +170,15 @@ public void testPathFilter() { String subdir1 = DIRECTORY + "\\subdir1"; String subdir2 = DIRECTORY + "\\subdir2"; String subdir3 = DIRECTORY + "\\foo"; - mockDir(DIRECTORY, List.of( - mockDir(subdir1, List.of(mockFile(subdir1, "file1.txt", "abc"))), - mockDir(subdir2, List.of(mockFile(subdir2, "file2.txt", "abc"))), - mockDir(subdir3, List.of(mockFile(subdir3, "file3.txt", "abc"))) - )); - + mockListing( + fileEntity(subdir1, "file1.txt"), + fileEntity(subdir2, "file2.txt"), + fileEntity(subdir3, "file3.txt") + ); testRunner.run(); - verifyOpenFile(subdir1, "file1.txt", 1); - verifyOpenFile(subdir2, "file2.txt", 1); - verifyOpenFile(subdir3, "file3.txt", 0); + verifyReadFile(subdir1, "file1.txt", 1); + verifyReadFile(subdir2, "file2.txt", 1); + verifyReadFile(subdir3, "file3.txt", 0); testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, 2); } @@ -261,12 +188,11 @@ public void testBatchSize() { final int totalSize = batchSize * 2; testRunner.setProperty(GetSmbFile.BATCH_SIZE, Integer.toString(batchSize)); - final List files = new ArrayList<>(); - final String fileNamePrefix = "file-"; + final SmbListableEntity[] entities = new SmbListableEntity[totalSize]; for (int i = 0; i < totalSize; i++) { - files.add(mockFile(DIRECTORY, fileNamePrefix + i, Integer.toString(i))); + entities[i] = fileEntity(DIRECTORY, "file-" + i); } - mockDir(DIRECTORY, files); + when(clientService.listFiles(anyString(), anyBoolean())).thenAnswer(inv -> Stream.of(entities)); // Avoid stopping on finish and run initialization testRunner.run(1, false, true); @@ -281,7 +207,7 @@ public void testBatchSize() { .toList(); for (int i = 0; i < totalSize; i++) { - final String flowFileName = flowFileNames.get(0); + final String flowFileName = flowFileNames.get(i); assertTrue(flowFileNames.contains(flowFileName), String.format("FlowFile Name [%s] not found", flowFileName)); } } diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/ListSmbIT.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/ListSmbIT.java index 1fbefedf04c5..3c53bde27546 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/ListSmbIT.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/ListSmbIT.java @@ -46,9 +46,9 @@ import static org.apache.nifi.processors.smb.ListSmb.MINIMUM_AGE; import static org.apache.nifi.processors.smb.ListSmb.MINIMUM_SIZE; import static org.apache.nifi.processors.smb.ListSmb.PATH_FILTER; -import static org.apache.nifi.services.smb.SmbjClientProviderService.HOSTNAME; -import static org.apache.nifi.services.smb.SmbjClientProviderService.PORT; -import static org.apache.nifi.services.smb.SmbjClientProviderService.SHARE; +import static org.apache.nifi.smb.common.SmbProperties.HOSTNAME; +import static org.apache.nifi.smb.common.SmbProperties.PORT; +import static org.apache.nifi.smb.common.SmbProperties.SHARE; import static org.apache.nifi.util.TestRunners.newTestRunner; import static org.junit.jupiter.api.Assertions.assertEquals; diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/ListSmbTest.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/ListSmbTest.java index 9535f2d96551..b939b452f98d 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/ListSmbTest.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/ListSmbTest.java @@ -63,6 +63,7 @@ import static org.apache.nifi.util.TestRunners.newTestRunner; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; @@ -248,7 +249,7 @@ public void shouldTurnSmbClientExceptionsToBulletins() throws Exception { testRunner.setProperty(LISTING_STRATEGY, "timestamps"); testRunner.setProperty(TARGET_SYSTEM_TIMESTAMP_PRECISION, "millis"); try (final SmbClientService mockNifiSmbClientService = configureTestRunnerWithMockedSmbClientService(testRunner)) { - when(mockNifiSmbClientService.listFiles(anyString())).thenThrow(new RuntimeException("test exception")); + when(mockNifiSmbClientService.listFiles(anyString(), anyBoolean())).thenThrow(new RuntimeException("test exception")); testRunner.run(); assertEquals(1, testRunner.getLogger().getErrorMessages().size()); testRunner.assertValid(); @@ -364,7 +365,7 @@ private SmbClientService configureTestRunnerWithMockedSmbClientService(TestRunne } private void mockSmbFolders(SmbClientService mockNifiSmbClientService, SmbListableEntity... entities) { - doAnswer(ignore -> stream(entities)).when(mockNifiSmbClientService).listFiles(anyString()); + doAnswer(ignore -> stream(entities)).when(mockNifiSmbClientService).listFiles(anyString(), anyBoolean()); } private SmbListableEntity listableEntity(String name, long timeStamp) { diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/PutSmbFileTest.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/PutSmbFileTest.java index 5c04b61add98..97186e404f3f 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/PutSmbFileTest.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/PutSmbFileTest.java @@ -16,18 +16,11 @@ */ package org.apache.nifi.processors.smb; -import com.hierynomus.mssmb2.SMB2CreateDisposition; -import com.hierynomus.mssmb2.SMB2ShareAccess; -import com.hierynomus.smbj.SMBClient; -import com.hierynomus.smbj.auth.AuthenticationContext; -import com.hierynomus.smbj.connection.Connection; -import com.hierynomus.smbj.server.ServerList; -import com.hierynomus.smbj.session.Session; -import com.hierynomus.smbj.share.DiskEntry; -import com.hierynomus.smbj.share.DiskShare; -import com.hierynomus.smbj.share.File; import org.apache.nifi.flowfile.FlowFile; -import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.services.smb.SmbClientProviderService; +import org.apache.nifi.services.smb.SmbClientService; +import org.apache.nifi.services.smb.SmbShareAccess; import org.apache.nifi.util.MockFlowFile; import org.apache.nifi.util.TestRunner; import org.apache.nifi.util.TestRunners; @@ -36,10 +29,11 @@ import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.mockito.Captor; +import org.mockito.Mock; import org.mockito.MockitoAnnotations; -import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.net.URI; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -47,16 +41,19 @@ import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; -import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.CONNECTION_CONFIGURATION_STRATEGY; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.ConnectionConfigurationStrategy; +import static org.apache.nifi.processors.smb.util.LocalSmbProperties.SMB_CLIENT_PROVIDER_SERVICE; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.ArgumentMatchers.anySet; -import static org.mockito.Mockito.any; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doThrow; -import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -66,95 +63,46 @@ public class PutSmbFileTest { private TestRunner testRunner; - private SMBClient smbClient; - private Connection connection; - private Session session; - private DiskShare diskShare; - private DiskEntry diskEntry; - private File smbfile; - private ServerList serverList; - private ByteArrayOutputStream baOutputStream; + @Mock + private SmbClientProviderService clientProviderService; + + @Mock + private SmbClientService clientService; + + @Captor + private ArgumentCaptor> shareAccessCaptor; private static final String HOSTNAME = "smbhostname"; private static final String SHARE = "smbshare"; private static final String DIRECTORY = "smbdirectory"; - private static final String DOMAIN = "mydomain"; - private static final String USERNAME = "myusername"; - private static final String PASSWORD = "mypassword"; + + private static final String CLIENT_SERVICE_ID = "client-provider-service-id"; private static final AtomicInteger FLOWFILE_ID_COUNTER = new AtomicInteger(0); - @Captor - private ArgumentCaptor> shareAccessSet; - @Captor - private ArgumentCaptor authenticationContext; - - private void setupSmbProcessor() throws IOException { - smbClient = mock(SMBClient.class); - connection = mock(Connection.class); - session = mock(Session.class); - diskShare = mock(DiskShare.class); - diskEntry = mock(DiskEntry.class); - smbfile = mock(File.class); - serverList = mock(ServerList.class); - baOutputStream = new ByteArrayOutputStream(); - - when(smbClient.connect(any(String.class), anyInt())).thenReturn(connection); - when(smbClient.getServerList()).thenReturn(serverList); - - when(connection.authenticate(any(AuthenticationContext.class))).thenReturn(session); - when(session.connectShare(SHARE)).thenReturn(diskShare); - when(diskShare.openFile( - any(String.class), - anySet(), - anySet(), - anySet(), - any(SMB2CreateDisposition.class), - anySet() - )).thenReturn(smbfile); - when(diskShare.open( - any(String.class), - anySet(), - anySet(), - anySet(), - any(SMB2CreateDisposition.class), - anySet() - )).thenReturn(diskEntry); - when(smbfile.getOutputStream()).thenReturn(baOutputStream); - - testRunner.setProperty(PutSmbFile.HOSTNAME, HOSTNAME); - testRunner.setProperty(PutSmbFile.SHARE, SHARE); - testRunner.setProperty(PutSmbFile.DIRECTORY, DIRECTORY); - testRunner.setProperty(PutSmbFile.DOMAIN, DOMAIN); - testRunner.setProperty(PutSmbFile.USERNAME, USERNAME); - testRunner.setProperty(PutSmbFile.PASSWORD, PASSWORD); - } + private AutoCloseable mockCloseable; - private void testDirectoryCreation(String dirFlag, int times) throws IOException { - when(diskShare.folderExists(any())).thenReturn(false); + @BeforeEach + public void init() throws Exception { + mockCloseable = MockitoAnnotations.openMocks(this); - testRunner.setProperty(PutSmbFile.DIRECTORY, "smbdirectory/subdir"); - testRunner.setProperty(PutSmbFile.CREATE_DIRS, dirFlag); - testRunner.enqueue("data"); - testRunner.run(); + when(clientProviderService.getIdentifier()).thenReturn(CLIENT_SERVICE_ID); + when(clientProviderService.getServiceLocation(anyMap())).thenReturn(URI.create("smb://" + HOSTNAME + ":445/" + SHARE)); + when(clientProviderService.getClient(any(ComponentLog.class), anyMap())).thenReturn(clientService); - verify(diskShare, times(times)).mkdir("smbdirectory"); - verify(diskShare, times(times)).mkdir("smbdirectory/subdir"); + testRunner = TestRunners.newTestRunner(PutSmbFile.class); + testRunner.addControllerService(CLIENT_SERVICE_ID, clientProviderService); + testRunner.enableControllerService(clientProviderService); + testRunner.setProperty(CONNECTION_CONFIGURATION_STRATEGY, ConnectionConfigurationStrategy.CONTROLLER_SERVICE.getValue()); + testRunner.setProperty(SMB_CLIENT_PROVIDER_SERVICE, CLIENT_SERVICE_ID); + testRunner.setProperty(PutSmbFile.DIRECTORY, DIRECTORY); } - private Set testOpenFileShareAccess() throws IOException { - testRunner.enqueue("data"); - testRunner.run(); - - verify(diskShare, times(1)).openFile( - any(String.class), - anySet(), - anySet(), - shareAccessSet.capture(), - any(SMB2CreateDisposition.class), - anySet() - ); - return shareAccessSet.getValue(); + @AfterEach + public void closeMocks() throws Exception { + if (mockCloseable != null) { + mockCloseable.close(); + } } private List generateFlowFile(int numberOfFlowFiles, Map attributes) { @@ -164,35 +112,27 @@ private List generateFlowFile(int numberOfFlowFiles, Map testWriteFileShareAccess() { + when(clientService.folderExists(any())).thenReturn(true); + testRunner.enqueue("data"); + testRunner.run(); - @AfterEach - public void closeMocks() throws Exception { - if (mockCloseable != null) { - mockCloseable.close(); - } + verify(clientService, times(1)).writeFile(anyString(), any(), shareAccessCaptor.capture()); + return shareAccessCaptor.getValue(); } @Test public void testHostnameAndShareEL() { - testRunner.setProperty(PutSmbFile.HOSTNAME, "${smb.hostname}"); - testRunner.setProperty(PutSmbFile.SHARE, "${smb.share}"); + when(clientProviderService.getServiceLocation(anyMap())).thenAnswer(invocation -> { + final Map attrs = invocation.getArgument(0); + final String host = attrs.getOrDefault("smb.hostname", HOSTNAME); + final String share = attrs.getOrDefault("smb.share", SHARE); + return URI.create("smb://" + host + ":445/" + share); + }); + testRunner.setProperty(PutSmbFile.BATCH_SIZE, "20"); // Add 10 FlowFiles with the same hostname and share property values @@ -220,8 +160,13 @@ public void testHostnameAndShareEL() { @Test public void testHostnameAndShareELWhenBatchsizeIsLowerThanAcceptableFlowFiles() { - testRunner.setProperty(PutSmbFile.HOSTNAME, "${smb.hostname}"); - testRunner.setProperty(PutSmbFile.SHARE, "${smb.share}"); + when(clientProviderService.getServiceLocation(anyMap())).thenAnswer(invocation -> { + final Map attrs = invocation.getArgument(0); + final String host = attrs.getOrDefault("smb.hostname", HOSTNAME); + final String share = attrs.getOrDefault("smb.share", SHARE); + return URI.create("smb://" + host + ":445/" + share); + }); + testRunner.setProperty(PutSmbFile.BATCH_SIZE, "10"); // Add 20 FlowFiles with the same hostname and share property values @@ -248,81 +193,46 @@ public void testHostnameAndShareELWhenBatchsizeIsLowerThanAcceptableFlowFiles() } @Test - public void testDefaultPortIsUsed() throws IOException { - testRunner.enqueue("data"); - testRunner.run(); - - verify(smbClient).connect(HOSTNAME, 445); - } - - @Test - public void testCustomPortIsUsed() throws IOException { - testRunner.setProperty(PutSmbFile.PORT, "4445"); - testRunner.enqueue("data"); - testRunner.run(); - - verify(smbClient).connect(HOSTNAME, 4445); - } - - @Test - public void testNormalAuth() throws IOException { + public void testDirExistsWithoutCreate() { + when(clientService.folderExists(any())).thenReturn(false); + testRunner.setProperty(PutSmbFile.DIRECTORY, "smbdirectory/subdir"); + testRunner.setProperty(PutSmbFile.CREATE_DIRS, "false"); testRunner.enqueue("data"); testRunner.run(); - verify(connection).authenticate(authenticationContext.capture()); - AuthenticationContext acObj = authenticationContext.getValue(); - assertEquals(USERNAME, acObj.getUsername()); - assertEquals(DOMAIN, acObj.getDomain()); - assertArrayEquals(acObj.getPassword(), PASSWORD.toCharArray()); + verify(clientService, times(0)).ensureDirectory(anyString()); + testRunner.assertAllFlowFilesTransferred(PutSmbFile.REL_FAILURE); } @Test - public void testAnonymousAuth() throws IOException { - testRunner.removeProperty(PutSmbFile.USERNAME); + public void testDirExistsWithCreate() { + when(clientService.folderExists(any())).thenReturn(false); + testRunner.setProperty(PutSmbFile.DIRECTORY, "smbdirectory/subdir"); + testRunner.setProperty(PutSmbFile.CREATE_DIRS, "true"); testRunner.enqueue("data"); testRunner.run(); - verify(connection).authenticate(authenticationContext.capture()); - AuthenticationContext acObj = authenticationContext.getValue(); - AuthenticationContext compAc = AuthenticationContext.anonymous(); - assertEquals(acObj.getUsername(), compAc.getUsername()); - assertEquals(acObj.getDomain(), compAc.getDomain()); - assertArrayEquals(acObj.getPassword(), compAc.getPassword()); - } - - @Test - public void testDirExistsWithoutCreate() throws IOException { - testDirectoryCreation("false", 0); - - testRunner.assertAllFlowFilesTransferred(PutSmbFile.REL_FAILURE); - } - - @Test - public void testDirExistsWithCreate() throws IOException { - testDirectoryCreation("true", 1); - + verify(clientService, times(1)).ensureDirectory("smbdirectory/subdir"); testRunner.assertAllFlowFilesTransferred(PutSmbFile.REL_SUCCESS); } @Test - public void testDirectoriesCreatedWhenDontExists() throws IOException { + public void testDirectoriesCreatedWhenDontExists() { final String directory = "a\\b/c/b\\e"; - when(diskShare.folderExists(any())).thenReturn(false); + when(clientService.folderExists(any())).thenReturn(false); testRunner.setProperty(PutSmbFile.CREATE_DIRS, "true"); testRunner.setProperty(PutSmbFile.DIRECTORY, directory); testRunner.enqueue("data"); testRunner.run(); - verify(diskShare, times(5)).mkdir( - any(String.class) - ); + verify(clientService, times(1)).ensureDirectory("a/b/c/b/e"); } @Test - public void testBatchCanContinueAfterDirectoryCreationFailure() throws IOException { - when(diskShare.folderExists(any())).thenReturn(false); - doThrow(new RuntimeException("Access denied")).when(diskShare).mkdir("dir2"); + public void testBatchCanContinueAfterDirectoryCreationFailure() { + when(clientService.folderExists(any())).thenReturn(false); + doThrow(new RuntimeException("Access denied")).when(clientService).ensureDirectory("dir2"); FlowFile flowFile1 = createFlowFileWithDirectoryAttribute(1, "dir1"); FlowFile flowFile2 = createFlowFileWithDirectoryAttribute(2, "dir2"); @@ -344,62 +254,58 @@ private FlowFile createFlowFileWithDirectoryAttribute(long id, String directory) } @Test - public void testFileShareNone() throws IOException { + public void testFileShareNone() { testRunner.setProperty(PutSmbFile.SHARE_ACCESS, PutSmbFile.SHARE_ACCESS_NONE); - testRunner.setProperty(PutSmbFile.CREATE_DIRS, "true"); - Set shareAccessSet = testOpenFileShareAccess(); - assertTrue(shareAccessSet.isEmpty()); + final Set shareAccess = testWriteFileShareAccess(); + assertTrue(shareAccess.isEmpty()); } @Test - public void testFileShareRead() throws IOException { + public void testFileShareRead() { testRunner.setProperty(PutSmbFile.SHARE_ACCESS, PutSmbFile.SHARE_ACCESS_READ); - testRunner.setProperty(PutSmbFile.CREATE_DIRS, "true"); - Set shareAccessSet = testOpenFileShareAccess(); - assertTrue(shareAccessSet.contains(SMB2ShareAccess.FILE_SHARE_READ)); + final Set shareAccess = testWriteFileShareAccess(); + assertTrue(shareAccess.contains(SmbShareAccess.READ_ALLOWED)); } @Test - public void testFileShareReadWriteDelete() throws IOException { + public void testFileShareReadWriteDelete() { testRunner.setProperty(PutSmbFile.SHARE_ACCESS, PutSmbFile.SHARE_ACCESS_READWRITEDELETE); - testRunner.setProperty(PutSmbFile.CREATE_DIRS, "true"); - Set shareAccessSet = testOpenFileShareAccess(); - assertTrue(shareAccessSet.contains(SMB2ShareAccess.FILE_SHARE_READ)); - assertTrue(shareAccessSet.contains(SMB2ShareAccess.FILE_SHARE_WRITE)); - assertTrue(shareAccessSet.contains(SMB2ShareAccess.FILE_SHARE_DELETE)); + final Set shareAccess = testWriteFileShareAccess(); + assertTrue(shareAccess.contains(SmbShareAccess.READ_ALLOWED)); + assertTrue(shareAccess.contains(SmbShareAccess.WRITE_ALLOWED)); + assertTrue(shareAccess.contains(SmbShareAccess.DELETE_ALLOWED)); } @Test - public void testFileExistsFail() throws IOException { + public void testFileExistsFail() { testRunner.setProperty(PutSmbFile.CONFLICT_RESOLUTION, PutSmbFile.FAIL_RESOLUTION); - when(diskShare.fileExists(any(String.class))).thenReturn(true); + when(clientService.folderExists(any())).thenReturn(true); + when(clientService.fileExists(any())).thenReturn(true); + testRunner.enqueue("data"); + testRunner.run(); testRunner.assertAllFlowFilesTransferred(PutSmbFile.REL_FAILURE); } @Test - public void testFileExistsIgnore() throws IOException { + public void testFileExistsIgnore() { testRunner.setProperty(PutSmbFile.CONFLICT_RESOLUTION, PutSmbFile.IGNORE_RESOLUTION); - when(diskShare.fileExists(any(String.class))).thenReturn(true); + when(clientService.folderExists(any())).thenReturn(true); + when(clientService.fileExists(any())).thenReturn(true); + testRunner.enqueue("data"); + testRunner.run(); testRunner.assertAllFlowFilesTransferred(PutSmbFile.REL_SUCCESS); } @Test - public void testTemporarySuffixIsUnset() throws IOException { + public void testTemporarySuffixIsUnset() { testRunner.enqueue("data"); testRunner.run(); - verify(diskShare, never()).open( - any(String.class), - anySet(), - anySet(), - anySet(), - any(SMB2CreateDisposition.class), - anySet() - ); + verify(clientService, never()).renameFile(anyString(), anyString(), anyBoolean()); } @Test - public void testTemporarySuffixIsSet() throws IOException { + public void testTemporarySuffixIsSet() { final String suffix = ".test"; testRunner.setProperty(PutSmbFile.RENAME_SUFFIX, suffix); @@ -407,22 +313,13 @@ public void testTemporarySuffixIsSet() throws IOException { testRunner.enqueue("data"); testRunner.run(); - ArgumentCaptor filename = ArgumentCaptor.forClass(String.class); - - verify(diskShare, times(1)).open( - filename.capture(), - anySet(), - anySet(), - anySet(), - any(SMB2CreateDisposition.class), - anySet() - ); - - assertTrue(filename.getValue().endsWith(suffix), "Suffix is not present"); + final ArgumentCaptor writePath = ArgumentCaptor.forClass(String.class); + verify(clientService, times(1)).writeFile(writePath.capture(), any(), anySet()); + assertTrue(writePath.getValue().endsWith(suffix), "Suffix is not present"); } @Test - public void testTemporarySuffixIsSetRenameIsCalled() throws IOException { + public void testTemporarySuffixIsSetRenameIsCalled() { final String suffix = ".test"; testRunner.setProperty(PutSmbFile.RENAME_SUFFIX, suffix); @@ -430,33 +327,22 @@ public void testTemporarySuffixIsSetRenameIsCalled() throws IOException { testRunner.enqueue("data"); testRunner.run(); - ArgumentCaptor initialFilename = ArgumentCaptor.forClass(String.class); - ArgumentCaptor finalFilename = ArgumentCaptor.forClass(String.class); - ArgumentCaptor replace = ArgumentCaptor.forClass(Boolean.class); - - verify(diskShare, times(1)).open( - initialFilename.capture(), - anySet(), - anySet(), - anySet(), - any(SMB2CreateDisposition.class), - anySet() - ); - - verify(diskEntry, times(1)).rename( - finalFilename.capture(), - replace.capture() - ); - - assertTrue(initialFilename.getValue().endsWith(suffix), "Suffix is not present and it should be"); - assertFalse(finalFilename.getValue().endsWith(suffix), "Suffix is present and it shouldn't be"); + final ArgumentCaptor writePath = ArgumentCaptor.forClass(String.class); + final ArgumentCaptor newPath = ArgumentCaptor.forClass(String.class); + final ArgumentCaptor replace = ArgumentCaptor.forClass(Boolean.class); + + verify(clientService, times(1)).writeFile(writePath.capture(), any(), anySet()); + verify(clientService, times(1)).renameFile(anyString(), newPath.capture(), replace.capture()); + + assertTrue(writePath.getValue().endsWith(suffix), "Suffix is not present and it should be"); + assertFalse(newPath.getValue().endsWith(suffix), "Suffix is present and it shouldn't be"); assertTrue(replace.getValue(), "Replace flag should be true"); } @Test public void testConnectionError() throws IOException { String emsg = "mock connection exception"; - when(smbClient.connect(any(String.class), anyInt())).thenThrow(new IOException(emsg)); + when(clientProviderService.getClient(any(ComponentLog.class), anyMap())).thenThrow(new IOException(emsg)); testRunner.enqueue("1"); testRunner.enqueue("2"); @@ -468,7 +354,7 @@ public void testConnectionError() throws IOException { @Test void testNormalizePath() { - PutSmbFile processor = new PutSmbFile(); + final PutSmbFile processor = new PutSmbFile(); assertNull(processor.normalizePath(null)); diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/SambaTestContainers.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/SambaTestContainers.java index 477a1b95fee0..da1e938ef05c 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/SambaTestContainers.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/SambaTestContainers.java @@ -30,12 +30,12 @@ import static java.util.Arrays.fill; import static org.apache.nifi.processors.smb.ListSmb.SMB_CLIENT_PROVIDER_SERVICE; -import static org.apache.nifi.services.smb.SmbjClientProviderService.DOMAIN; -import static org.apache.nifi.services.smb.SmbjClientProviderService.HOSTNAME; -import static org.apache.nifi.services.smb.SmbjClientProviderService.PASSWORD; -import static org.apache.nifi.services.smb.SmbjClientProviderService.PORT; -import static org.apache.nifi.services.smb.SmbjClientProviderService.SHARE; -import static org.apache.nifi.services.smb.SmbjClientProviderService.USERNAME; +import static org.apache.nifi.smb.common.SmbProperties.DOMAIN; +import static org.apache.nifi.smb.common.SmbProperties.HOSTNAME; +import static org.apache.nifi.smb.common.SmbProperties.PASSWORD; +import static org.apache.nifi.smb.common.SmbProperties.PORT; +import static org.apache.nifi.smb.common.SmbProperties.SHARE; +import static org.apache.nifi.smb.common.SmbProperties.USERNAME; public class SambaTestContainers { diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/SmbDfsIT.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/SmbDfsIT.java index 9b4a84f02ccf..6d9df71fc7c3 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/SmbDfsIT.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/SmbDfsIT.java @@ -43,12 +43,12 @@ import static org.apache.nifi.processor.util.list.AbstractListProcessor.NO_TRACKING; import static org.apache.nifi.processors.smb.ListSmb.SMB_CLIENT_PROVIDER_SERVICE; -import static org.apache.nifi.services.smb.SmbjClientProviderService.HOSTNAME; -import static org.apache.nifi.services.smb.SmbjClientProviderService.PASSWORD; -import static org.apache.nifi.services.smb.SmbjClientProviderService.PORT; -import static org.apache.nifi.services.smb.SmbjClientProviderService.SHARE; -import static org.apache.nifi.services.smb.SmbjClientProviderService.USERNAME; import static org.apache.nifi.smb.common.SmbProperties.ENABLE_DFS; +import static org.apache.nifi.smb.common.SmbProperties.HOSTNAME; +import static org.apache.nifi.smb.common.SmbProperties.PASSWORD; +import static org.apache.nifi.smb.common.SmbProperties.PORT; +import static org.apache.nifi.smb.common.SmbProperties.SHARE; +import static org.apache.nifi.smb.common.SmbProperties.USERNAME; import static org.apache.nifi.util.TestRunners.newTestRunner; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -161,11 +161,11 @@ private void testListSmb(String directory) throws Exception { @Test void testPutSmbFile() { TestRunner testRunner = newTestRunner(PutSmbFile.class); - testRunner.setProperty(PutSmbFile.HOSTNAME, sambaContainer.getHost()); - testRunner.setProperty(PutSmbFile.SHARE, "dfs-share"); + testRunner.setProperty(HOSTNAME, sambaContainer.getHost()); + testRunner.setProperty(SHARE, "dfs-share"); testRunner.setProperty(PutSmbFile.DIRECTORY, "dfs-link"); - testRunner.setProperty(PutSmbFile.USERNAME, "myuser"); - testRunner.setProperty(PutSmbFile.PASSWORD, "mypass"); + testRunner.setProperty(USERNAME, "myuser"); + testRunner.setProperty(PASSWORD, "mypass"); testRunner.setProperty(ENABLE_DFS, "true"); testRunner.enqueue("put_content", Map.of(CoreAttributes.FILENAME.key(), "put_file")); @@ -182,11 +182,11 @@ void testGetSmbFile() { writeFile("get_file", "get_content"); TestRunner testRunner = newTestRunner(GetSmbFile.class); - testRunner.setProperty(GetSmbFile.HOSTNAME, sambaContainer.getHost()); - testRunner.setProperty(GetSmbFile.SHARE, "dfs-share"); + testRunner.setProperty(HOSTNAME, sambaContainer.getHost()); + testRunner.setProperty(SHARE, "dfs-share"); testRunner.setProperty(GetSmbFile.DIRECTORY, "dfs-link"); - testRunner.setProperty(GetSmbFile.USERNAME, "myuser"); - testRunner.setProperty(GetSmbFile.PASSWORD, "mypass"); + testRunner.setProperty(USERNAME, "myuser"); + testRunner.setProperty(PASSWORD, "mypass"); testRunner.setProperty(ENABLE_DFS, "true"); testRunner.run(); diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/pom.xml b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/pom.xml index f54ed416ae4d..33cc502d734a 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/pom.xml +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/pom.xml @@ -46,19 +46,5 @@ nifi-smb-smbj-common 2.12.0-SNAPSHOT - - com.hierynomus - smbj - - - org.testcontainers - testcontainers - test - - - org.testcontainers - testcontainers-toxiproxy - test - diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/main/java/org/apache/nifi/services/smb/SmbjClientProviderService.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/main/java/org/apache/nifi/services/smb/SmbjClientProviderService.java index 441a3b9be565..01c1b40e1d02 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/main/java/org/apache/nifi/services/smb/SmbjClientProviderService.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/main/java/org/apache/nifi/services/smb/SmbjClientProviderService.java @@ -16,12 +16,6 @@ */ package org.apache.nifi.services.smb; -import com.hierynomus.smbj.SMBClient; -import com.hierynomus.smbj.auth.AuthenticationContext; -import com.hierynomus.smbj.connection.Connection; -import com.hierynomus.smbj.session.Session; -import com.hierynomus.smbj.share.DiskShare; -import com.hierynomus.smbj.share.Share; import org.apache.nifi.annotation.documentation.CapabilityDescription; import org.apache.nifi.annotation.documentation.Tags; import org.apache.nifi.annotation.lifecycle.OnDisabled; @@ -35,72 +29,33 @@ import java.io.IOException; import java.net.URI; import java.util.List; +import java.util.Map; -import static org.apache.nifi.processor.util.StandardValidators.NON_BLANK_VALIDATOR; -import static org.apache.nifi.processor.util.StandardValidators.NON_EMPTY_VALIDATOR; -import static org.apache.nifi.processor.util.StandardValidators.PORT_VALIDATOR; +import static org.apache.nifi.smb.common.SmbProperties.DOMAIN; import static org.apache.nifi.smb.common.SmbProperties.ENABLE_DFS; +import static org.apache.nifi.smb.common.SmbProperties.HOSTNAME; +import static org.apache.nifi.smb.common.SmbProperties.OLD_DOMAIN_PROPERTY_NAME; import static org.apache.nifi.smb.common.SmbProperties.OLD_ENABLE_DFS_PROPERTY_NAME; +import static org.apache.nifi.smb.common.SmbProperties.OLD_HOSTNAME_PROPERTY_NAME; +import static org.apache.nifi.smb.common.SmbProperties.OLD_PASSWORD_PROPERTY_NAME; +import static org.apache.nifi.smb.common.SmbProperties.OLD_PORT_PROPERTY_NAME; +import static org.apache.nifi.smb.common.SmbProperties.OLD_SHARE_PROPERTY_NAME; import static org.apache.nifi.smb.common.SmbProperties.OLD_SMB_DIALECT_PROPERTY_NAME; import static org.apache.nifi.smb.common.SmbProperties.OLD_TIMEOUT_PROPERTY_NAME; +import static org.apache.nifi.smb.common.SmbProperties.OLD_USERNAME_PROPERTY_NAME; import static org.apache.nifi.smb.common.SmbProperties.OLD_USE_ENCRYPTION_PROPERTY_NAME; +import static org.apache.nifi.smb.common.SmbProperties.PASSWORD; +import static org.apache.nifi.smb.common.SmbProperties.PORT; +import static org.apache.nifi.smb.common.SmbProperties.SHARE; import static org.apache.nifi.smb.common.SmbProperties.SMB_DIALECT; import static org.apache.nifi.smb.common.SmbProperties.TIMEOUT; +import static org.apache.nifi.smb.common.SmbProperties.USERNAME; import static org.apache.nifi.smb.common.SmbProperties.USE_ENCRYPTION; -import static org.apache.nifi.smb.common.SmbUtils.buildSmbClient; @Tags({"samba, smb, cifs, files"}) @CapabilityDescription("Provides access to SMB Sessions with shared authentication credentials.") public class SmbjClientProviderService extends AbstractControllerService implements SmbClientProviderService { - public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder() - .name("Hostname") - .description("The network host of the SMB file server.") - .required(true) - .addValidator(NON_BLANK_VALIDATOR) - .build(); - - public static final PropertyDescriptor DOMAIN = new PropertyDescriptor.Builder() - .name("Domain") - .description( - "The domain used for authentication. Optional, in most cases username and password is sufficient.") - .required(false) - .addValidator(NON_EMPTY_VALIDATOR) - .build(); - - public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() - .name("Username") - .description( - "The username used for authentication.") - .required(false) - .defaultValue("Guest") - .addValidator(NON_EMPTY_VALIDATOR) - .build(); - - public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() - .name("Password") - .description("The password used for authentication.") - .required(false) - .addValidator(NON_EMPTY_VALIDATOR) - .sensitive(true) - .build(); - - public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder() - .name("Port") - .description("Port to use for connection.") - .required(true) - .addValidator(PORT_VALIDATOR) - .defaultValue("445") - .build(); - - public static final PropertyDescriptor SHARE = new PropertyDescriptor.Builder() - .name("Share") - .description("The network share to which files should be listed from. This is the \"first folder\"" + - "after the hostname: smb://hostname:port/[share]/dir1/dir2") - .required(true) - .addValidator(NON_BLANK_VALIDATOR) - .build(); - private static final List PROPERTY_DESCRIPTORS = List.of( HOSTNAME, PORT, @@ -114,11 +69,7 @@ public class SmbjClientProviderService extends AbstractControllerService impleme TIMEOUT ); - private SMBClient smbClient; - private AuthenticationContext authenticationContext; - private String hostname; - private int port; - private String shareName; + private SmbjClientProvider delegate; @Override protected List getSupportedPropertyDescriptors() { @@ -127,90 +78,43 @@ protected List getSupportedPropertyDescriptors() { @OnEnabled public void onEnabled(final ConfigurationContext context) { - this.hostname = context.getProperty(HOSTNAME).getValue(); - this.port = context.getProperty(PORT).asInteger(); - this.shareName = context.getProperty(SHARE).getValue(); - this.smbClient = buildSmbClient(context); - createAuthenticationContext(context); + delegate = new SmbjClientProvider(context, getLogger()); } @OnDisabled public void onDisabled() { - smbClient.close(); - smbClient = null; - hostname = null; - port = 0; - shareName = null; + if (delegate != null) { + try { + delegate.close(); + } catch (Exception e) { + getLogger().error("Error while closing SMB ClientProvider", e); + } finally { + delegate = null; + } + } } @Override - public URI getServiceLocation() { - return URI.create(String.format("smb://%s:%d/%s", hostname, port, shareName)); + public URI getServiceLocation(final Map attributes) { + return delegate.getServiceLocation(attributes); } @Override - public SmbClientService getClient(final ComponentLog logger) throws IOException { - final Connection connection = smbClient.connect(hostname, port); - - final Session session; - final Share share; - - try { - session = connection.authenticate(authenticationContext); - } catch (Exception e) { - throw new IOException("Could not create session for share " + getServiceLocation(), e); - } - - try { - share = session.connectShare(shareName); - } catch (Exception e) { - closeSession(session); - throw new IOException("Could not connect to share " + getServiceLocation(), e); - } - - if (!(share instanceof DiskShare)) { - closeSession(session); - throw new IllegalArgumentException("DiskShare not found. Share " + share.getClass().getSimpleName() + " found on " + getServiceLocation()); - } - - return new SmbjClientService(session, (DiskShare) share, getServiceLocation(), logger); + public SmbClientService getClient(final ComponentLog logger, final Map attributes) throws IOException { + return delegate.getClient(logger, attributes); } @Override public void migrateProperties(PropertyConfiguration config) { - config.renameProperty("hostname", HOSTNAME.getName()); - config.renameProperty("domain", DOMAIN.getName()); - config.renameProperty("username", USERNAME.getName()); - config.renameProperty("password", PASSWORD.getName()); - config.renameProperty("port", PORT.getName()); - config.renameProperty("share", SHARE.getName()); + config.renameProperty(OLD_HOSTNAME_PROPERTY_NAME, HOSTNAME.getName()); + config.renameProperty(OLD_DOMAIN_PROPERTY_NAME, DOMAIN.getName()); + config.renameProperty(OLD_USERNAME_PROPERTY_NAME, USERNAME.getName()); + config.renameProperty(OLD_PASSWORD_PROPERTY_NAME, PASSWORD.getName()); + config.renameProperty(OLD_PORT_PROPERTY_NAME, PORT.getName()); + config.renameProperty(OLD_SHARE_PROPERTY_NAME, SHARE.getName()); config.renameProperty(OLD_ENABLE_DFS_PROPERTY_NAME, ENABLE_DFS.getName()); config.renameProperty(OLD_SMB_DIALECT_PROPERTY_NAME, SMB_DIALECT.getName()); config.renameProperty(OLD_TIMEOUT_PROPERTY_NAME, TIMEOUT.getName()); config.renameProperty(OLD_USE_ENCRYPTION_PROPERTY_NAME, USE_ENCRYPTION.getName()); } - - private void closeSession(final Session session) { - try { - if (session != null) { - session.close(); - } - } catch (Exception e) { - getLogger().error("Could not close session to {}", getServiceLocation(), e); - } - } - - private void createAuthenticationContext(final ConfigurationContext context) { - if (context.getProperty(USERNAME).isSet()) { - final String userName = context.getProperty(USERNAME).getValue(); - final String password = - context.getProperty(PASSWORD).isSet() ? context.getProperty(PASSWORD).getValue() : ""; - final String domainOrNull = - context.getProperty(DOMAIN).isSet() ? context.getProperty(DOMAIN).getValue() : null; - authenticationContext = new AuthenticationContext(userName, password.toCharArray(), domainOrNull); - } else { - authenticationContext = AuthenticationContext.anonymous(); - } - } - } diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/pom.xml b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/pom.xml index 38f64122bd06..9c305b665254 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/pom.xml +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/pom.xml @@ -25,6 +25,12 @@ nifi-smb-smbj-common + + org.apache.nifi + nifi-smb-client-api + 2.12.0-SNAPSHOT + provided + org.apache.nifi @@ -34,5 +40,16 @@ com.hierynomus smbj + + + org.testcontainers + testcontainers + test + + + org.testcontainers + testcontainers-toxiproxy + test + \ No newline at end of file diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/services/smb/SmbjClientProvider.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/services/smb/SmbjClientProvider.java new file mode 100644 index 000000000000..dc420f85c62d --- /dev/null +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/services/smb/SmbjClientProvider.java @@ -0,0 +1,132 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.services.smb; + +import com.hierynomus.smbj.SMBClient; +import com.hierynomus.smbj.auth.AuthenticationContext; +import com.hierynomus.smbj.connection.Connection; +import com.hierynomus.smbj.session.Session; +import com.hierynomus.smbj.share.DiskShare; +import com.hierynomus.smbj.share.Share; +import org.apache.nifi.context.PropertyContext; +import org.apache.nifi.logging.ComponentLog; + +import java.io.Closeable; +import java.io.IOException; +import java.net.URI; +import java.util.Map; + +import static org.apache.nifi.smb.common.SmbProperties.DOMAIN; +import static org.apache.nifi.smb.common.SmbProperties.HOSTNAME; +import static org.apache.nifi.smb.common.SmbProperties.PASSWORD; +import static org.apache.nifi.smb.common.SmbProperties.PORT; +import static org.apache.nifi.smb.common.SmbProperties.SHARE; +import static org.apache.nifi.smb.common.SmbProperties.USERNAME; +import static org.apache.nifi.smb.common.SmbUtils.buildSmbClient; + +public class SmbjClientProvider implements SmbClientProvider, Closeable { + + private final PropertyContext context; + + private final ComponentLog logger; + + private final SMBClient smbClient; + + private AuthenticationContext authenticationContext; + + public SmbjClientProvider(final PropertyContext context, final ComponentLog logger) { + this.context = context; + this.logger = logger; + this.smbClient = buildSmbClient(context); + createAuthenticationContext(context); + } + + @Override + public void close() { + smbClient.close(); + } + + private String getHostname(final Map attributes) { + return context.getProperty(HOSTNAME).evaluateAttributeExpressions(attributes).getValue(); + } + + private Integer getPort(final Map attributes) { + return context.getProperty(PORT).evaluateAttributeExpressions(attributes).asInteger(); + } + + private String getShareName(final Map attributes) { + return context.getProperty(SHARE).evaluateAttributeExpressions(attributes).getValue(); + } + + @Override + public URI getServiceLocation(final Map attributes) { + return URI.create(String.format("smb://%s:%d/%s", getHostname(attributes), getPort(attributes), getShareName(attributes))); + } + + @Override + public SmbClientService getClient(final ComponentLog logger, final Map attributes) throws IOException { + final Connection connection = smbClient.connect(getHostname(attributes), getPort(attributes)); + final URI serviceLocation = getServiceLocation(attributes); + + final Session session; + final Share share; + + try { + session = connection.authenticate(authenticationContext); + } catch (Exception e) { + throw new IOException("Could not create session for share " + serviceLocation, e); + } + + try { + share = session.connectShare(getShareName(attributes)); + } catch (Exception e) { + closeSession(session, serviceLocation); + throw new IOException("Could not connect to share " + serviceLocation, e); + } + + if (!(share instanceof DiskShare)) { + closeSession(session, serviceLocation); + throw new IllegalArgumentException("DiskShare not found. Share " + share.getClass().getSimpleName() + " found on " + serviceLocation); + } + + return new SmbjClientService(session, (DiskShare) share, getServiceLocation(attributes), logger); + } + + + private void closeSession(final Session session, URI serviceLocation) { + try { + if (session != null) { + session.close(); + } + } catch (Exception e) { + logger.error("Could not close session to {}", serviceLocation, e); + } + } + + private void createAuthenticationContext(final PropertyContext context) { + if (context.getProperty(USERNAME).isSet()) { + final String userName = context.getProperty(USERNAME).getValue(); + final String password = + context.getProperty(PASSWORD).isSet() ? context.getProperty(PASSWORD).getValue() : ""; + final String domainOrNull = + context.getProperty(DOMAIN).isSet() ? context.getProperty(DOMAIN).getValue() : null; + authenticationContext = new AuthenticationContext(userName, password.toCharArray(), domainOrNull); + } else { + authenticationContext = AuthenticationContext.anonymous(); + } + } +} diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/main/java/org/apache/nifi/services/smb/SmbjClientService.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/services/smb/SmbjClientService.java similarity index 78% rename from nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/main/java/org/apache/nifi/services/smb/SmbjClientService.java rename to nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/services/smb/SmbjClientService.java index 2b04200cad0a..2abe6ac1f8b6 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/main/java/org/apache/nifi/services/smb/SmbjClientService.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/services/smb/SmbjClientService.java @@ -24,6 +24,7 @@ import com.hierynomus.mssmb2.SMB2CreateOptions; import com.hierynomus.mssmb2.SMB2ShareAccess; import com.hierynomus.mssmb2.SMBApiException; +import com.hierynomus.smbj.io.InputStreamByteChunkProvider; import com.hierynomus.smbj.session.Session; import com.hierynomus.smbj.share.Directory; import com.hierynomus.smbj.share.DiskShare; @@ -32,12 +33,14 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; +import java.io.InputStream; import java.io.OutputStream; import java.net.URI; import java.util.EnumSet; import java.util.List; import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; import java.util.stream.Stream; import static java.util.Arrays.asList; @@ -74,7 +77,17 @@ public void close() { } @Override - public Stream listFiles(final String directoryPath) { + public boolean folderExists(final String path) { + return share.folderExists(path); + } + + @Override + public boolean fileExists(final String path) { + return share.fileExists(path); + } + + @Override + public Stream listFiles(final String directoryPath, final boolean recursive) { return Stream.of(directoryPath).flatMap(path -> { final Directory directory; try { @@ -108,7 +121,7 @@ public Stream listFiles(final String directoryPath) { return stream(directory::spliterator, 0, false) .map(entity -> buildSmbListableEntity(entity, path, serviceLocation)) .filter(entity -> !specialDirectory(entity)) - .flatMap(listable -> listable.isDirectory() ? listFiles(listable.getPathWithName()) + .flatMap(listable -> listable.isDirectory() && recursive ? listFiles(listable.getPathWithName(), recursive) : Stream.of(listable)) .onClose(directory::close); }); @@ -139,42 +152,63 @@ public void ensureDirectory(final String directoryPath) { } @Override - public void readFile(final String filePath, final OutputStream outputStream) throws IOException { - try (File file = share.openFile( + public void readFile(final String filePath, final OutputStream outputStream, final Set shareAccesses) { + try (outputStream; File file = share.openFile( filePath, EnumSet.of(AccessMask.GENERIC_READ), EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL), - EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ), + convertShareAccesses(shareAccesses), SMB2CreateDisposition.FILE_OPEN, EnumSet.of(SMB2CreateOptions.FILE_SEQUENTIAL_ONLY)) ) { file.read(outputStream); } catch (Exception e) { throw wrapException(e); - } finally { - outputStream.close(); } } @Override - public void moveFile(final String filePath, final String directoryPath) { - try (File file = share.openFile( + public void writeFile(final String filePath, final InputStream inputStream, final Set shareAccesses) { + try (inputStream; File file = share.openFile( filePath, + EnumSet.of(AccessMask.GENERIC_WRITE), + EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL), + convertShareAccesses(shareAccesses), + SMB2CreateDisposition.FILE_OVERWRITE_IF, + EnumSet.of(SMB2CreateOptions.FILE_WRITE_THROUGH)) + ) { + file.write(new InputStreamByteChunkProvider(inputStream)); + } catch (Exception e) { + throw wrapException(e); + } + } + + @Override + public void renameFile(final String oldFilePath, final String newFilePath, final boolean override) { + try (File file = share.openFile( + oldFilePath, EnumSet.of(AccessMask.GENERIC_WRITE, AccessMask.DELETE), EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL), EnumSet.noneOf(SMB2ShareAccess.class), SMB2CreateDisposition.FILE_OPEN, - EnumSet.of(SMB2CreateOptions.FILE_SEQUENTIAL_ONLY)) + EnumSet.noneOf(SMB2CreateOptions.class)) ) { - final String[] parts = filePath.split("/"); // rename operation on Windows requires \ (backslash) path separator - final String newFilePath = directoryPath.replace('/', '\\') + "\\" + parts[parts.length - 1]; - file.rename(newFilePath); + file.rename(newFilePath.replace('/', '\\'), override); } catch (Exception e) { throw wrapException(e); } } + @Override + public void moveFile(final String filePath, final String directoryPath) { + final String[] parts = filePath.split("/"); + final String fileName = parts[parts.length - 1]; + final String newFilePath = directoryPath + "/" + fileName; + + renameFile(filePath, newFilePath, false); + } + @Override public void deleteFile(final String filePath) { try { @@ -194,6 +228,7 @@ private SmbListableEntity buildSmbListableEntity(final FileIdBothDirectoryInform .setChangeTime(info.getChangeTime().toEpochMillis()) .setLastAccessTime(info.getLastAccessTime().toEpochMillis()) .setDirectory((info.getFileAttributes() & FileAttributes.FILE_ATTRIBUTE_DIRECTORY.getValue()) != 0) + .setHidden((info.getFileAttributes() & FileAttributes.FILE_ATTRIBUTE_HIDDEN.getValue()) != 0) .setSize(info.getEndOfFile()) .setAllocationSize(info.getAllocationSize()) .setServiceLocation(serviceLocation) @@ -224,5 +259,17 @@ private SmbException wrapException(final Exception e) { } } + private Set convertShareAccesses(final Set shareAccesses) { + return shareAccesses.stream() + .map(shareAccess -> + switch (shareAccess) { + case READ_ALLOWED -> SMB2ShareAccess.FILE_SHARE_READ; + case WRITE_ALLOWED -> SMB2ShareAccess.FILE_SHARE_WRITE; + case DELETE_ALLOWED -> SMB2ShareAccess.FILE_SHARE_DELETE; + } + ) + .collect(Collectors.toSet()); + } + } diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/smb/common/SmbProperties.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/smb/common/SmbProperties.java index 4d89326d4315..781ff56d9beb 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/smb/common/SmbProperties.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/main/java/org/apache/nifi/smb/common/SmbProperties.java @@ -17,15 +17,73 @@ package org.apache.nifi.smb.common; import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.expression.ExpressionLanguageScope; +import static org.apache.nifi.processor.util.StandardValidators.NON_BLANK_VALIDATOR; +import static org.apache.nifi.processor.util.StandardValidators.NON_EMPTY_VALIDATOR; +import static org.apache.nifi.processor.util.StandardValidators.PORT_VALIDATOR; import static org.apache.nifi.processor.util.StandardValidators.TIME_PERIOD_VALIDATOR; public class SmbProperties { + public static final String OLD_HOSTNAME_PROPERTY_NAME = "hostname"; + public static final String OLD_PORT_PROPERTY_NAME = "port"; + public static final String OLD_SHARE_PROPERTY_NAME = "share"; + public static final String OLD_DOMAIN_PROPERTY_NAME = "domain"; + public static final String OLD_USERNAME_PROPERTY_NAME = "username"; + public static final String OLD_PASSWORD_PROPERTY_NAME = "password"; public static final String OLD_SMB_DIALECT_PROPERTY_NAME = "smb-dialect"; public static final String OLD_USE_ENCRYPTION_PROPERTY_NAME = "use-encryption"; public static final String OLD_ENABLE_DFS_PROPERTY_NAME = "enable-dfs"; public static final String OLD_TIMEOUT_PROPERTY_NAME = "timeout"; + public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder() + .name("Hostname") + .description("The network host of the SMB file server.") + .required(true) + .addValidator(NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .build(); + + public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder() + .name("Port") + .description("Port to use for connection.") + .required(true) + .addValidator(PORT_VALIDATOR) + .defaultValue("445") + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .build(); + + public static final PropertyDescriptor SHARE = new PropertyDescriptor.Builder() + .name("Share") + .description("The network share that hosts the files. This is the \"first folder\"" + + "after the hostname: smb://hostname:port/[share]/dir1/dir2") + .required(true) + .addValidator(NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .build(); + + public static final PropertyDescriptor DOMAIN = new PropertyDescriptor.Builder() + .name("Domain") + .description("The domain used for authentication. Optional, in most cases username and password is sufficient.") + .required(false) + .addValidator(NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("The username used for authentication. If no username is set then anonymous authentication is attempted.") + .required(false) + .addValidator(NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .description("The password used for authentication.") + .required(false) + .addValidator(NON_EMPTY_VALIDATOR) + .sensitive(true) + .build(); + public static final PropertyDescriptor SMB_DIALECT = new PropertyDescriptor.Builder() .name("SMB Dialect") .description("The SMB dialect is negotiated between the client and the server by default to the highest common version supported by both end. " + diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceIT.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceIT.java similarity index 85% rename from nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceIT.java rename to nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceIT.java index 4e9e70f3c1b8..d7fc2460aba5 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceIT.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceIT.java @@ -21,7 +21,7 @@ import eu.rekawek.toxiproxy.model.ToxicDirection; import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.logging.ComponentLog; -import org.apache.nifi.util.MockConfigurationContext; +import org.apache.nifi.util.MockPropertyContext; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -46,13 +46,13 @@ import java.util.concurrent.atomic.AtomicInteger; import static java.util.stream.Collectors.toSet; -import static org.apache.nifi.services.smb.SmbjClientProviderService.DOMAIN; -import static org.apache.nifi.services.smb.SmbjClientProviderService.HOSTNAME; -import static org.apache.nifi.services.smb.SmbjClientProviderService.PASSWORD; -import static org.apache.nifi.services.smb.SmbjClientProviderService.PORT; -import static org.apache.nifi.services.smb.SmbjClientProviderService.SHARE; -import static org.apache.nifi.services.smb.SmbjClientProviderService.USERNAME; +import static org.apache.nifi.smb.common.SmbProperties.DOMAIN; +import static org.apache.nifi.smb.common.SmbProperties.HOSTNAME; +import static org.apache.nifi.smb.common.SmbProperties.PASSWORD; +import static org.apache.nifi.smb.common.SmbProperties.PORT; +import static org.apache.nifi.smb.common.SmbProperties.SHARE; import static org.apache.nifi.smb.common.SmbProperties.TIMEOUT; +import static org.apache.nifi.smb.common.SmbProperties.USERNAME; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -100,7 +100,6 @@ public void shouldRescueAfterConnectionFailure() throws Exception { final Proxy proxy = toxiproxyClient.createProxy("samba", "0.0.0.0:8666", "samba:445"); final String ipAddressViaToxiproxy = toxiproxy.getHost(); final int portViaToxiproxy = toxiproxy.getMappedPort(8666); - SmbjClientProviderService smbjClientProviderService = new SmbjClientProviderService(); Map properties = new HashMap<>(); properties.put(HOSTNAME, ipAddressViaToxiproxy); @@ -110,9 +109,9 @@ public void shouldRescueAfterConnectionFailure() throws Exception { properties.put(PASSWORD, "password"); properties.put(DOMAIN, "domain"); properties.put(TIMEOUT, "0.5 sec"); - MockConfigurationContext mockConfigurationContext = new MockConfigurationContext(properties, null, null); + MockPropertyContext mockPropertyContext = new MockPropertyContext(properties); - smbjClientProviderService.onEnabled(mockConfigurationContext); + SmbjClientProvider smbjClientProvider = new SmbjClientProvider(mockPropertyContext, mock(ComponentLog.class)); proxy.toxics().latency("slow", ToxicDirection.DOWNSTREAM, 300); @@ -133,13 +132,13 @@ public void shouldRescueAfterConnectionFailure() throws Exception { SmbClientService s = null; try { - s = smbjClientProviderService.getClient(mock(ComponentLog.class)); + s = smbjClientProvider.getClient(mock(ComponentLog.class)); if (iteration == 25) { proxy.toxics().bandwidth("CUT_CONNECTION_DOWNSTREAM", ToxicDirection.DOWNSTREAM, 0L); proxy.toxics().bandwidth("CUT_CONNECTION_UPSTREAM", ToxicDirection.UPSTREAM, 0L); } - final Set actual = s.listFiles("testDirectory") + final Set actual = s.listFiles("testDirectory", true) .map(SmbListableEntity::getIdentifier) .collect(toSet()); @@ -175,7 +174,7 @@ public void shouldRescueAfterConnectionFailure() throws Exception { latch.await(); executorService.shutdown(); - smbjClientProviderService.onDisabled(); + smbjClientProvider.close(); } @Test @@ -186,8 +185,6 @@ public void shouldContinueListingAfterPermissionDenied() throws Exception { sambaContainer.execInContainer("bash", "-c", "chmod 000 /folder/testDirectory/directory2"); - SmbjClientProviderService smbjClientProviderService = new SmbjClientProviderService(); - Map properties = new HashMap<>(); properties.put(HOSTNAME, sambaContainer.getHost()); properties.put(PORT, String.valueOf(sambaContainer.getMappedPort(445))); @@ -197,12 +194,13 @@ public void shouldContinueListingAfterPermissionDenied() throws Exception { properties.put(DOMAIN, "domain"); properties.put(TIMEOUT, "0.5 sec"); - MockConfigurationContext mockConfigurationContext = new MockConfigurationContext(properties, null, null); - smbjClientProviderService.onEnabled(mockConfigurationContext); + MockPropertyContext mockPropertyContext = new MockPropertyContext(properties); + + SmbjClientProvider smbjClientProvider = new SmbjClientProvider(mockPropertyContext, mock(ComponentLog.class)); - SmbClientService smbClientService = smbjClientProviderService.getClient(mock(ComponentLog.class)); + SmbClientService smbClientService = smbjClientProvider.getClient(mock(ComponentLog.class)); - final Set actual = smbClientService.listFiles("testDirectory") + final Set actual = smbClientService.listFiles("testDirectory", true) .map(SmbListableEntity::getIdentifier) .collect(toSet()); @@ -210,7 +208,7 @@ public void shouldContinueListingAfterPermissionDenied() throws Exception { assertTrue(actual.contains("testDirectory/directory1/file1")); assertTrue(actual.contains("testDirectory/directory3/file3")); - smbjClientProviderService.onDisabled(); + smbjClientProvider.close(); } private void writeFile(String path, String content) { diff --git a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceTest.java b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceTest.java similarity index 94% rename from nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceTest.java rename to nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceTest.java index 4431c9c20566..aed6f3e34e7c 100644 --- a/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-client/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceTest.java +++ b/nifi-extension-bundles/nifi-smb-bundle/nifi-smb-smbj-common/src/test/java/org/apache/nifi/services/smb/SmbjClientServiceTest.java @@ -76,9 +76,9 @@ void listFilesShouldHandlePermissionErrors() { mockOpenDirectory("dir2", NtStatus.STATUS_BAD_NETWORK_NAME); mockOpenDirectory("dir3", NtStatus.STATUS_OTHER); - assertEquals(0, underTest.listFiles("dir1").count()); - assertEquals(0, underTest.listFiles("dir2").count()); - assertThrows(SMBApiException.class, () -> underTest.listFiles("dir3").count()); + assertEquals(0, underTest.listFiles("dir1", true).count()); + assertEquals(0, underTest.listFiles("dir2", true).count()); + assertThrows(SMBApiException.class, () -> underTest.listFiles("dir3", true).count()); } private void mockOpenDirectory(String directoryName, NtStatus responseStatus) {