Skip to content

Commit

Permalink
NIFI-9990 This closes apache#6030. Improved FTP 550 file unavailable …
Browse files Browse the repository at this point in the history 
…handling

- Improved File Not Found reply detection
- Added Permission Denied reply handling

Signed-off-by: Joe Witt <[email protected]>
  • Loading branch information
@exceptionfactory @joewitt
exceptionfactory authored and joewitt committed May 13, 2022
1 parent 313d705 commit 30f7c1b
Show file tree
Hide file tree
Showing 2 changed files with 66 additions and 18 deletions.
26 changes: 19 additions & 7 deletions ...andard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,10 @@ public class FTPTransfer implements FileTransfer {
.addValidator(StandardValidators.BOOLEAN_VALIDATOR)
.build();

private static final int REPLY_CODE_FILE_UNAVAILABLE = 550;

private static final Pattern NOT_FOUND_MESSAGE_PATTERN = Pattern.compile("(no such)|(not exist)|(not found)", Pattern.CASE_INSENSITIVE);

private static final FTPClientProvider FTP_CLIENT_PROVIDER = new StandardFTPClientProvider();

private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP_AUTH, ProxySpec.SOCKS_AUTH};
Expand Down Expand Up @@ -321,13 +325,22 @@ public FlowFile getRemoteFile(final String remoteFileName, final FlowFile origFl
FlowFile resultFlowFile;
try (InputStream in = client.retrieveFileStream(remoteFileName)) {
if (in == null) {
final String response = client.getReplyString();
// FTPClient doesn't throw exception if file not found.
// Instead, response string will contain: "550 Can't open <absolute_path>: No such file or directory"
if (response != null && response.trim().endsWith("No such file or directory")) {
throw new FileNotFoundException(response);
final String reply = client.getReplyString();
if (reply == null) {
throw new IOException("Retrieve File Failed: FTP server response not found");
}
throw new IOException(response);

// Get reply code after checking for reply string
final int replyCode = client.getReplyCode();
if (REPLY_CODE_FILE_UNAVAILABLE == replyCode) {
if (NOT_FOUND_MESSAGE_PATTERN.matcher(reply).find()) {
throw new FileNotFoundException(reply);
} else {
throw new PermissionDeniedException(reply);
}
}

throw new IOException(reply);
}
resultFlowFile = session.write(origFlowFile, out -> StreamUtils.copy(in, out));
client.completePendingCommand();
Expand Down Expand Up @@ -606,5 +619,4 @@ public static Supplier<ProxyConfiguration> createComponentProxyConfigSupplier(fi
return componentProxyConfig;
};
}

}
58 changes: 47 additions & 11 deletions ...e/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestFTP.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/
package org.apache.nifi.processors.standard;

import static org.junit.Assert.assertEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;

import java.io.FileInputStream;
import java.io.IOException;
Expand Down Expand Up @@ -45,6 +45,7 @@
import org.mockftpserver.fake.filesystem.DirectoryEntry;
import org.mockftpserver.fake.filesystem.FileEntry;
import org.mockftpserver.fake.filesystem.FileSystem;
import org.mockftpserver.fake.filesystem.Permissions;
import org.mockftpserver.fake.filesystem.WindowsFakeFileSystem;

public class TestFTP {
Expand Down Expand Up @@ -80,7 +81,6 @@ public void testValidators() {
ProcessContext pc;

/* Set the basic required values */
results = new HashSet<>();
runner.setProperty(FTPTransfer.USERNAME, "${el-username}");
runner.setProperty(FTPTransfer.HOSTNAME, "static-hostname");
runner.setProperty(FTPTransfer.PORT, "${el-portNumber}");
Expand Down Expand Up @@ -132,8 +132,8 @@ public void basicFileUpload() throws IOException {
runner.setProperty(FTPTransfer.USERNAME, username);
runner.setProperty(FTPTransfer.PASSWORD, password);
runner.setProperty(FTPTransfer.PORT, Integer.toString(ftpPort));
try (FileInputStream fis = new FileInputStream("src/test/resources/randombytes-1");) {
Map<String, String> attributes = new HashMap<String, String>();
try (FileInputStream fis = new FileInputStream("src/test/resources/randombytes-1")) {
Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.FILENAME.key(), "randombytes-1");
runner.enqueue(fis, attributes);
runner.run();
Expand All @@ -157,14 +157,14 @@ public void basicProvenanceEventTest() throws IOException {

// Get two flowfiles to test by running data
try (FileInputStream fis = new FileInputStream("src/test/resources/randombytes-1")) {
Map<String, String> attributes = new HashMap<String, String>();
Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.FILENAME.key(), "randombytes-1");
attributes.put("transfer-host", "localhost");
runner.enqueue(fis, attributes);
runner.run();
}
try (FileInputStream fis = new FileInputStream("src/test/resources/hello.txt")) {
Map<String, String> attributes = new HashMap<String, String>();
Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.FILENAME.key(), "hello.txt");
attributes.put("transfer-host", "127.0.0.1");
runner.enqueue(fis, attributes);
Expand Down Expand Up @@ -198,7 +198,7 @@ public void basicProvenanceEventTest() throws IOException {
}

@Test
public void basicFileGet() throws IOException {
public void basicFileGet() {
FileSystem results = fakeFtpServer.getFileSystem();

FileEntry sampleFile = new FileEntry("c:\\data\\randombytes-2");
Expand All @@ -222,7 +222,7 @@ public void basicFileGet() throws IOException {
}

@Test
public void basicFileFetch() throws IOException {
public void basicFileFetch() {
FileSystem results = fakeFtpServer.getFileSystem();

FileEntry sampleFile = new FileEntry("c:\\data\\randombytes-2");
Expand All @@ -242,7 +242,7 @@ public void basicFileFetch() throws IOException {
runner.setProperty(FetchFTP.MOVE_DESTINATION_DIR, "data");


Map<String, String> attrs = new HashMap<String, String>();
Map<String, String> attrs = new HashMap<>();
attrs.put("host", "localhost");
attrs.put("username", username);
attrs.put("port", Integer.toString(ftpPort));
Expand All @@ -254,10 +254,46 @@ public void basicFileFetch() throws IOException {
retrievedFile.assertContentEquals("Just some random test test test chocolate");
}

@Test
public void testFetchFileNotFound() {
final TestRunner runner = TestRunners.newTestRunner(FetchFTP.class);
runner.setProperty(FetchFTP.HOSTNAME, "127.0.0.1");
runner.setProperty(FetchFTP.USERNAME, username);
runner.setProperty(FTPTransfer.PASSWORD, password);
runner.setProperty(FTPTransfer.PORT, Integer.toString(ftpPort));
runner.setProperty(FetchFTP.REMOTE_FILENAME, "remote-file-not-found");

runner.enqueue(new byte[0]);
runner.run();

runner.assertAllFlowFilesTransferred(FetchFTP.REL_NOT_FOUND);
}

@Test
public void testFetchFilePermissionDenied() {
final FileSystem fs = fakeFtpServer.getFileSystem();

final FileEntry restrictedFileEntry = new FileEntry("c:\\data\\restricted");
restrictedFileEntry.setPermissions(Permissions.NONE);
fs.add(restrictedFileEntry);

final TestRunner runner = TestRunners.newTestRunner(FetchFTP.class);
runner.setProperty(FetchFTP.HOSTNAME, "127.0.0.1");
runner.setProperty(FetchFTP.USERNAME, username);
runner.setProperty(FTPTransfer.PASSWORD, password);
runner.setProperty(FTPTransfer.PORT, Integer.toString(ftpPort));
runner.setProperty(FetchFTP.REMOTE_FILENAME, "restricted");

runner.enqueue(new byte[0]);
runner.run();

runner.assertAllFlowFilesTransferred(FetchFTP.REL_PERMISSION_DENIED);
}

@Test
@EnabledIfSystemProperty(named = "file.encoding", matches = "UTF-8",
disabledReason = "org.mockftpserver does not support specification of charset")
public void basicFileFetchWithUTF8FileName() throws IOException {
public void basicFileFetchWithUTF8FileName() {
FileSystem fs = fakeFtpServer.getFileSystem();

FileEntry sampleFile = new FileEntry("c:\\data\\őűőű.txt");
Expand All @@ -284,7 +320,7 @@ public void basicFileFetchWithUTF8FileName() throws IOException {
}

@Test
public void basicFileList() throws IOException, InterruptedException {
public void basicFileList() throws InterruptedException {
FileSystem results = fakeFtpServer.getFileSystem();

FileEntry sampleFile = new FileEntry("c:\\data\\randombytes-2");
Expand Down

0 comments on commit 30f7c1b

Please sign in to comment.