| /* |
| * 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.standard.util; |
| |
| import net.schmizz.keepalive.KeepAlive; |
| import net.schmizz.keepalive.KeepAliveProvider; |
| import net.schmizz.sshj.Config; |
| import net.schmizz.sshj.DefaultConfig; |
| import net.schmizz.sshj.SSHClient; |
| import net.schmizz.sshj.common.Factory; |
| import net.schmizz.sshj.connection.ConnectionException; |
| import net.schmizz.sshj.connection.ConnectionImpl; |
| import net.schmizz.sshj.sftp.FileAttributes; |
| import net.schmizz.sshj.sftp.FileMode; |
| import net.schmizz.sshj.sftp.RemoteFile; |
| import net.schmizz.sshj.sftp.RemoteResourceFilter; |
| import net.schmizz.sshj.sftp.RemoteResourceInfo; |
| import net.schmizz.sshj.sftp.Response; |
| import net.schmizz.sshj.sftp.SFTPClient; |
| import net.schmizz.sshj.sftp.SFTPException; |
| import net.schmizz.sshj.transport.TransportException; |
| import net.schmizz.sshj.transport.verification.PromiscuousVerifier; |
| import net.schmizz.sshj.userauth.keyprovider.KeyProvider; |
| import net.schmizz.sshj.userauth.method.AuthMethod; |
| import net.schmizz.sshj.userauth.method.AuthPassword; |
| import net.schmizz.sshj.userauth.method.AuthPublickey; |
| import net.schmizz.sshj.userauth.password.PasswordFinder; |
| import net.schmizz.sshj.userauth.password.PasswordUtils; |
| import net.schmizz.sshj.xfer.FilePermission; |
| import net.schmizz.sshj.xfer.LocalSourceFile; |
| import org.apache.nifi.components.PropertyDescriptor; |
| import org.apache.nifi.components.PropertyValue; |
| import org.apache.nifi.components.ValidationContext; |
| import org.apache.nifi.components.ValidationResult; |
| import org.apache.nifi.components.resource.ResourceCardinality; |
| import org.apache.nifi.components.resource.ResourceType; |
| import org.apache.nifi.context.PropertyContext; |
| import org.apache.nifi.expression.ExpressionLanguageScope; |
| import org.apache.nifi.flowfile.FlowFile; |
| import org.apache.nifi.logging.ComponentLog; |
| import org.apache.nifi.processor.ProcessSession; |
| import org.apache.nifi.processor.exception.ProcessException; |
| import org.apache.nifi.processor.io.OutputStreamCallback; |
| import org.apache.nifi.processor.util.StandardValidators; |
| import org.apache.nifi.proxy.ProxyConfiguration; |
| import org.apache.nifi.proxy.ProxySpec; |
| import org.apache.nifi.stream.io.StreamUtils; |
| |
| import javax.net.SocketFactory; |
| import java.io.File; |
| import java.io.FileNotFoundException; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.OutputStream; |
| import java.net.InetAddress; |
| import java.net.Proxy; |
| import java.net.Socket; |
| import java.nio.file.Path; |
| import java.nio.file.Paths; |
| import java.text.DateFormat; |
| import java.text.SimpleDateFormat; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Date; |
| import java.util.List; |
| import java.util.Locale; |
| import java.util.Set; |
| import java.util.concurrent.TimeUnit; |
| import java.util.regex.Pattern; |
| import java.util.stream.Collectors; |
| |
| import static org.apache.nifi.processors.standard.util.FTPTransfer.createComponentProxyConfigSupplier; |
| |
| public class SFTPTransfer implements FileTransfer { |
| private static final Set<String> DEFAULT_KEY_ALGORITHM_NAMES; |
| private static final Set<String> DEFAULT_CIPHER_NAMES; |
| private static final Set<String> DEFAULT_MESSAGE_AUTHENTICATION_CODE_NAMES; |
| private static final Set<String> DEFAULT_KEY_EXCHANGE_ALGORITHM_NAMES; |
| |
| static { |
| DefaultConfig defaultConfig = new DefaultConfig(); |
| |
| DEFAULT_KEY_ALGORITHM_NAMES = Collections.unmodifiableSet(defaultConfig.getKeyAlgorithms().stream() |
| .map(Factory.Named::getName).collect(Collectors.toSet())); |
| DEFAULT_CIPHER_NAMES = Collections.unmodifiableSet(defaultConfig.getCipherFactories().stream() |
| .map(Factory.Named::getName).collect(Collectors.toSet())); |
| DEFAULT_MESSAGE_AUTHENTICATION_CODE_NAMES = Collections.unmodifiableSet(defaultConfig.getMACFactories().stream() |
| .map(Factory.Named::getName).collect(Collectors.toSet())); |
| DEFAULT_KEY_EXCHANGE_ALGORITHM_NAMES = Collections.unmodifiableSet(defaultConfig.getKeyExchangeFactories().stream() |
| .map(Factory.Named::getName).collect(Collectors.toSet())); |
| } |
| |
| /** |
| * Converts a set of names into an alphabetically ordered comma separated value list. |
| * |
| * @param factorySetNames The set of names |
| * @return An alphabetically ordered comma separated value list of names |
| */ |
| private static String convertFactorySetToString(Set<String> factorySetNames) { |
| return factorySetNames |
| .stream() |
| .sorted() |
| .collect(Collectors.joining(", ")); |
| } |
| |
| public static final PropertyDescriptor PRIVATE_KEY_PATH = new PropertyDescriptor.Builder() |
| .name("Private Key Path") |
| .description("The fully qualified path to the Private Key file") |
| .required(false) |
| .identifiesExternalResource(ResourceCardinality.SINGLE, ResourceType.FILE) |
| .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) |
| .build(); |
| public static final PropertyDescriptor PRIVATE_KEY_PASSPHRASE = new PropertyDescriptor.Builder() |
| .name("Private Key Passphrase") |
| .description("Password for the private key") |
| .required(false) |
| .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) |
| .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) |
| .sensitive(true) |
| .build(); |
| public static final PropertyDescriptor HOST_KEY_FILE = new PropertyDescriptor.Builder() |
| .name("Host Key File") |
| .description("If supplied, the given file will be used as the Host Key;" + |
| " otherwise, if 'Strict Host Key Checking' property is applied (set to true)" + |
| " then uses the 'known_hosts' and 'known_hosts2' files from ~/.ssh directory" + |
| " else no host key file will be used") |
| .identifiesExternalResource(ResourceCardinality.SINGLE, ResourceType.FILE) |
| .required(false) |
| .build(); |
| public static final PropertyDescriptor STRICT_HOST_KEY_CHECKING = new PropertyDescriptor.Builder() |
| .name("Strict Host Key Checking") |
| .description("Indicates whether or not strict enforcement of hosts keys should be applied") |
| .allowableValues("true", "false") |
| .defaultValue("false") |
| .required(true) |
| .build(); |
| public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder() |
| .name("Port") |
| .description("The port that the remote system is listening on for file transfers") |
| .addValidator(StandardValidators.PORT_VALIDATOR) |
| .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) |
| .required(true) |
| .defaultValue("22") |
| .build(); |
| public static final PropertyDescriptor USE_KEEPALIVE_ON_TIMEOUT = new PropertyDescriptor.Builder() |
| .name("Send Keep Alive On Timeout") |
| .description("Indicates whether or not to send a single Keep Alive message when SSH socket times out") |
| .allowableValues("true", "false") |
| .defaultValue("true") |
| .required(true) |
| .build(); |
| |
| public static final PropertyDescriptor KEY_ALGORITHMS_ALLOWED = new PropertyDescriptor.Builder() |
| .name("Key Algorithms Allowed") |
| .displayName("Key Algorithms Allowed") |
| .description("A comma-separated list of Key Algorithms allowed for SFTP connections. Leave unset to allow all. Available options are: " |
| + convertFactorySetToString(DEFAULT_KEY_ALGORITHM_NAMES)) |
| .required(false) |
| .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) |
| .addValidator(StandardValidators.NON_BLANK_VALIDATOR) |
| .build(); |
| |
| public static final PropertyDescriptor CIPHERS_ALLOWED = new PropertyDescriptor.Builder() |
| .name("Ciphers Allowed") |
| .displayName("Ciphers Allowed") |
| .description("A comma-separated list of Ciphers allowed for SFTP connections. Leave unset to allow all. Available options are: " + convertFactorySetToString(DEFAULT_CIPHER_NAMES)) |
| .required(false) |
| .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) |
| .addValidator(StandardValidators.NON_BLANK_VALIDATOR) |
| .build(); |
| |
| public static final PropertyDescriptor MESSAGE_AUTHENTICATION_CODES_ALLOWED = new PropertyDescriptor.Builder() |
| .name("Message Authentication Codes Allowed") |
| .displayName("Message Authentication Codes Allowed") |
| .description("A comma-separated list of Message Authentication Codes allowed for SFTP connections. Leave unset to allow all. Available options are: " |
| + convertFactorySetToString(DEFAULT_MESSAGE_AUTHENTICATION_CODE_NAMES)) |
| .required(false) |
| .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) |
| .addValidator(StandardValidators.NON_BLANK_VALIDATOR) |
| .build(); |
| |
| public static final PropertyDescriptor KEY_EXCHANGE_ALGORITHMS_ALLOWED = new PropertyDescriptor.Builder() |
| .name("Key Exchange Algorithms Allowed") |
| .displayName("Key Exchange Algorithms Allowed") |
| .description("A comma-separated list of Key Exchange Algorithms allowed for SFTP connections. Leave unset to allow all. Available options are: " |
| + convertFactorySetToString(DEFAULT_KEY_EXCHANGE_ALGORITHM_NAMES)) |
| .required(false) |
| .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) |
| .addValidator(StandardValidators.NON_BLANK_VALIDATOR) |
| .build(); |
| |
| /** |
| * Property which is used to decide if the {@link #ensureDirectoryExists(FlowFile, File)} method should perform a {@link SFTPClient#ls(String)} before calling |
| * {@link SFTPClient#mkdir(String)}. In most cases, the code should call ls before mkdir, but some weird permission setups (chmod 100) on a directory would cause the 'ls' to throw a permission |
| * exception. |
| */ |
| public static final PropertyDescriptor DISABLE_DIRECTORY_LISTING = new PropertyDescriptor.Builder() |
| .name("Disable Directory Listing") |
| .description("If set to 'true', directory listing is not performed prior to create missing directories." + |
| " By default, this processor executes a directory listing command" + |
| " to see target directory existence before creating missing directories." + |
| " However, there are situations that you might need to disable the directory listing such as the following." + |
| " Directory listing might fail with some permission setups (e.g. chmod 100) on a directory." + |
| " Also, if any other SFTP client created the directory after this processor performed a listing" + |
| " and before a directory creation request by this processor is finished," + |
| " then an error is returned because the directory already exists.") |
| .addValidator(StandardValidators.BOOLEAN_VALIDATOR) |
| .allowableValues("true", "false") |
| .defaultValue("false") |
| .build(); |
| |
| private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP_AUTH, ProxySpec.SOCKS_AUTH}; |
| public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE |
| = ProxyConfiguration.createProxyConfigPropertyDescriptor(true, PROXY_SPECS); |
| |
| private final ComponentLog logger; |
| |
| private final PropertyContext ctx; |
| |
| private SSHClient sshClient; |
| private SFTPClient sftpClient; |
| |
| private volatile boolean closed = false; |
| private String homeDir; |
| |
| private final boolean disableDirectoryListing; |
| |
| public SFTPTransfer(final PropertyContext propertyContext, final ComponentLog logger) { |
| this.ctx = propertyContext; |
| this.logger = logger; |
| |
| final PropertyValue disableListing = propertyContext.getProperty(DISABLE_DIRECTORY_LISTING); |
| disableDirectoryListing = disableListing == null ? false : Boolean.TRUE.equals(disableListing.asBoolean()); |
| } |
| |
| public static void validateProxySpec(ValidationContext context, Collection<ValidationResult> results) { |
| ProxyConfiguration.validateProxySpec(context, results, PROXY_SPECS); |
| } |
| |
| @Override |
| public String getProtocolName() { |
| return "sftp"; |
| } |
| |
| @Override |
| public List<FileInfo> getListing() throws IOException { |
| final String path = ctx.getProperty(FileTransfer.REMOTE_PATH).evaluateAttributeExpressions().getValue(); |
| final int depth = 0; |
| |
| final int maxResults; |
| final PropertyValue batchSizeValue = ctx.getProperty(FileTransfer.REMOTE_POLL_BATCH_SIZE); |
| if (batchSizeValue == null) { |
| maxResults = Integer.MAX_VALUE; |
| } else { |
| final Integer configuredValue = batchSizeValue.asInteger(); |
| maxResults = configuredValue == null ? Integer.MAX_VALUE : configuredValue; |
| } |
| |
| final List<FileInfo> listing = new ArrayList<>(1000); |
| getListing(path, depth, maxResults, listing); |
| return listing; |
| } |
| |
| protected void getListing(final String path, final int depth, final int maxResults, final List<FileInfo> listing) throws IOException { |
| if (maxResults < 1 || listing.size() >= maxResults) { |
| return; |
| } |
| |
| if (depth >= 100) { |
| logger.warn(this + " had to stop recursively searching directories at a recursive depth of " + depth + " to avoid memory issues"); |
| return; |
| } |
| |
| final boolean ignoreDottedFiles = ctx.getProperty(FileTransfer.IGNORE_DOTTED_FILES).asBoolean(); |
| final boolean recurse = ctx.getProperty(FileTransfer.RECURSIVE_SEARCH).asBoolean(); |
| final boolean symlink = ctx.getProperty(FileTransfer.FOLLOW_SYMLINK).asBoolean(); |
| final String fileFilterRegex = ctx.getProperty(FileTransfer.FILE_FILTER_REGEX).getValue(); |
| final Pattern pattern = (fileFilterRegex == null) ? null : Pattern.compile(fileFilterRegex); |
| final String pathFilterRegex = ctx.getProperty(FileTransfer.PATH_FILTER_REGEX).getValue(); |
| final Pattern pathPattern = (!recurse || pathFilterRegex == null) ? null : Pattern.compile(pathFilterRegex); |
| final String remotePath = ctx.getProperty(FileTransfer.REMOTE_PATH).evaluateAttributeExpressions().getValue(); |
| |
| // check if this directory path matches the PATH_FILTER_REGEX |
| boolean pathFilterMatches = true; |
| if (pathPattern != null) { |
| Path reldir = path == null ? Paths.get(".") : Paths.get(path); |
| if (remotePath != null) { |
| reldir = Paths.get(remotePath).relativize(reldir); |
| } |
| if (reldir != null && !reldir.toString().isEmpty()) { |
| if (!pathPattern.matcher(reldir.toString().replace("\\", "/")).matches()) { |
| pathFilterMatches = false; |
| } |
| } |
| } |
| |
| final SFTPClient sftpClient = getSFTPClient(null); |
| final boolean isPathMatch = pathFilterMatches; |
| |
| //subDirs list is used for both 'sub directories' and 'symlinks' |
| final List<RemoteResourceInfo> subDirs = new ArrayList<>(); |
| try { |
| final RemoteResourceFilter filter = (entry) -> { |
| final String entryFilename = entry.getName(); |
| |
| // skip over 'this directory' and 'parent directory' special files regardless of ignoring dot files |
| if (entryFilename.equals(".") || entryFilename.equals("..")) { |
| return false; |
| } |
| |
| // skip files and directories that begin with a dot if we're ignoring them |
| if (ignoreDottedFiles && entryFilename.startsWith(".")) { |
| return false; |
| } |
| |
| // if is a directory and we're supposed to recurse OR if is a link and we're supposed to follow symlink |
| if ((recurse && entry.isDirectory()) || (symlink && (entry.getAttributes().getType() == FileMode.Type.SYMLINK))){ |
| subDirs.add(entry); |
| return false; |
| } |
| |
| // Since SSHJ does not have the concept of BREAK that JSCH had, we need to move this before the call to listing.add |
| // below, otherwise we would keep adding to the listings since returning false here doesn't break |
| if (listing.size() >= maxResults) { |
| return false; |
| } |
| |
| // if is not a directory and is not a link and it matches FILE_FILTER_REGEX - then let's add it |
| if (!entry.isDirectory() && !(entry.getAttributes().getType() == FileMode.Type.SYMLINK) && isPathMatch) { |
| if (pattern == null || pattern.matcher(entryFilename).matches()) { |
| listing.add(newFileInfo(entry, path)); |
| } |
| } |
| |
| return false; |
| }; |
| |
| if (path == null || path.trim().isEmpty()) { |
| sftpClient.ls(".", filter); |
| } else { |
| sftpClient.ls(path, filter); |
| } |
| } catch (final SFTPException e) { |
| final String pathDesc = path == null ? "current directory" : path; |
| switch (e.getStatusCode()) { |
| case NO_SUCH_FILE: |
| throw new FileNotFoundException("Could not perform listing on " + pathDesc + " because could not find the file on the remote server"); |
| case PERMISSION_DENIED: |
| throw new PermissionDeniedException("Could not perform listing on " + pathDesc + " due to insufficient permissions"); |
| default: |
| throw new IOException(String.format("Failed to obtain file listing for %s due to unexpected SSH_FXP_STATUS (%d)", |
| pathDesc, e.getStatusCode().getCode()), e); |
| } |
| } |
| |
| for (final RemoteResourceInfo entry : subDirs) { |
| final String entryFilename = entry.getName(); |
| final File newFullPath = new File(path, entryFilename); |
| final String newFullForwardPath = newFullPath.getPath().replace("\\", "/"); |
| |
| try { |
| getListing(newFullForwardPath, depth + 1, maxResults, listing); |
| } catch (final IOException e) { |
| logger.error("Unable to get listing from " + newFullForwardPath + "; skipping", e); |
| } |
| } |
| |
| } |
| |
| private FileInfo newFileInfo(final RemoteResourceInfo entry, String path) { |
| if (entry == null) { |
| return null; |
| } |
| final File newFullPath = new File(path, entry.getName()); |
| final String newFullForwardPath = newFullPath.getPath().replace("\\", "/"); |
| |
| final StringBuilder permsBuilder = new StringBuilder(); |
| final Set<FilePermission> permissions = entry.getAttributes().getPermissions(); |
| |
| appendPermission(permsBuilder, permissions, FilePermission.USR_R, "r"); |
| appendPermission(permsBuilder, permissions, FilePermission.USR_W, "w"); |
| appendPermission(permsBuilder, permissions, FilePermission.USR_X, "x"); |
| |
| appendPermission(permsBuilder, permissions, FilePermission.GRP_R, "r"); |
| appendPermission(permsBuilder, permissions, FilePermission.GRP_W, "w"); |
| appendPermission(permsBuilder, permissions, FilePermission.GRP_X, "x"); |
| |
| appendPermission(permsBuilder, permissions, FilePermission.OTH_R, "r"); |
| appendPermission(permsBuilder, permissions, FilePermission.OTH_W, "w"); |
| appendPermission(permsBuilder, permissions, FilePermission.OTH_X, "x"); |
| |
| final FileInfo.Builder builder = new FileInfo.Builder() |
| .filename(entry.getName()) |
| .fullPathFileName(newFullForwardPath) |
| .directory(entry.isDirectory()) |
| .size(entry.getAttributes().getSize()) |
| .lastModifiedTime(entry.getAttributes().getMtime() * 1000L) |
| .permissions(permsBuilder.toString()) |
| .owner(Integer.toString(entry.getAttributes().getUID())) |
| .group(Integer.toString(entry.getAttributes().getGID())); |
| return builder.build(); |
| } |
| |
| private void appendPermission(final StringBuilder builder, final Set<FilePermission> permissions, final FilePermission filePermission, final String permString) { |
| if (permissions.contains(filePermission)) { |
| builder.append(permString); |
| } else { |
| builder.append("-"); |
| } |
| } |
| |
| @Override |
| public FlowFile getRemoteFile(final String remoteFileName, final FlowFile origFlowFile, final ProcessSession session) throws ProcessException, IOException { |
| final SFTPClient sftpClient = getSFTPClient(origFlowFile); |
| RemoteFile rf = null; |
| RemoteFile.ReadAheadRemoteFileInputStream rfis = null; |
| FlowFile resultFlowFile = null; |
| try { |
| rf = sftpClient.open(remoteFileName); |
| rfis = rf.new ReadAheadRemoteFileInputStream(16); |
| final InputStream in = rfis; |
| resultFlowFile = session.write(origFlowFile, new OutputStreamCallback() { |
| @Override |
| public void process(final OutputStream out) throws IOException { |
| StreamUtils.copy(in, out); |
| } |
| }); |
| return resultFlowFile; |
| } catch (final SFTPException e) { |
| switch (e.getStatusCode()) { |
| case NO_SUCH_FILE: |
| throw new FileNotFoundException("Could not find file " + remoteFileName + " on remote SFTP Server"); |
| case PERMISSION_DENIED: |
| throw new PermissionDeniedException("Insufficient permissions to read file " + remoteFileName + " from remote SFTP Server", e); |
| default: |
| throw new IOException("Failed to obtain file content for " + remoteFileName, e); |
| } |
| } finally { |
| if(rf != null){ |
| try{ |
| rf.close(); |
| }catch(final IOException ioe){ |
| //do nothing |
| } |
| } |
| if(rfis != null){ |
| try{ |
| rfis.close(); |
| }catch(final IOException ioe){ |
| //do nothing |
| } |
| } |
| } |
| } |
| |
| @Override |
| public void deleteFile(final FlowFile flowFile, final String path, final String remoteFileName) throws IOException { |
| final SFTPClient sftpClient = getSFTPClient(flowFile); |
| final String fullPath = (path == null) ? remoteFileName : (path.endsWith("/")) ? path + remoteFileName : path + "/" + remoteFileName; |
| try { |
| sftpClient.rm(fullPath); |
| } catch (final SFTPException e) { |
| switch (e.getStatusCode()) { |
| case NO_SUCH_FILE: |
| throw new FileNotFoundException("Could not find file " + remoteFileName + " to remove from remote SFTP Server"); |
| case PERMISSION_DENIED: |
| throw new PermissionDeniedException("Insufficient permissions to delete file " + remoteFileName + " from remote SFTP Server", e); |
| default: |
| throw new IOException("Failed to delete remote file " + fullPath, e); |
| } |
| } |
| } |
| |
| @Override |
| public void deleteDirectory(final FlowFile flowFile, final String remoteDirectoryName) throws IOException { |
| final SFTPClient sftpClient = getSFTPClient(flowFile); |
| try { |
| sftpClient.rmdir(remoteDirectoryName); |
| } catch (final SFTPException e) { |
| throw new IOException("Failed to delete remote directory " + remoteDirectoryName, e); |
| } |
| } |
| |
| @Override |
| public void ensureDirectoryExists(final FlowFile flowFile, final File directoryName) throws IOException { |
| final SFTPClient sftpClient = getSFTPClient(flowFile); |
| final String remoteDirectory = directoryName.getAbsolutePath().replace("\\", "/").replaceAll("^.\\:", ""); |
| |
| // if we disable the directory listing, we just want to blindly perform the mkdir command, |
| // eating failure exceptions thrown (like if the directory already exists). |
| if (disableDirectoryListing) { |
| try { |
| // Blindly create the dir. |
| sftpClient.mkdir(remoteDirectory); |
| // The remote directory did not exist, and was created successfully. |
| return; |
| } catch (SFTPException e) { |
| if (e.getStatusCode() == Response.StatusCode.NO_SUCH_FILE) { |
| // No Such File. This happens when parent directory was not found. |
| logger.debug(String.format("Could not create %s due to 'No such file'. Will try to create the parent dir.", remoteDirectory)); |
| } else if (e.getStatusCode() == Response.StatusCode.FAILURE) { |
| // Swallow '4: Failure' including the remote directory already exists. |
| logger.debug("Could not blindly create remote directory due to " + getMessage(e), e); |
| return; |
| } else { |
| throw new IOException("Could not blindly create remote directory due to " + e.getMessage(), e); |
| } |
| } |
| } else { |
| try { |
| // Check dir existence. |
| sftpClient.stat(remoteDirectory); |
| // The remote directory already exists. |
| return; |
| } catch (final SFTPException e) { |
| if (e.getStatusCode() != Response.StatusCode.NO_SUCH_FILE) { |
| throw new IOException("Failed to determine if remote directory exists at " + remoteDirectory + " due to " + getMessage(e), e); |
| } |
| } |
| } |
| |
| // first ensure parent directories exist before creating this one |
| if (directoryName.getParent() != null && !directoryName.getParentFile().equals(new File(File.separator))) { |
| ensureDirectoryExists(flowFile, directoryName.getParentFile()); |
| } |
| logger.debug("Remote Directory {} does not exist; creating it", new Object[] {remoteDirectory}); |
| try { |
| sftpClient.mkdir(remoteDirectory); |
| logger.debug("Created {}", new Object[] {remoteDirectory}); |
| } catch (final SFTPException e) { |
| throw new IOException("Failed to create remote directory " + remoteDirectory + " due to " + getMessage(e), e); |
| } |
| } |
| |
| private String getMessage(final SFTPException e) { |
| if (e.getStatusCode() != null) { |
| return e.getStatusCode().getCode() + ": " + e.getMessage(); |
| } else { |
| return e.getMessage(); |
| } |
| } |
| |
| private static KeepAliveProvider NO_OP_KEEP_ALIVE = new KeepAliveProvider() { |
| @Override |
| public KeepAlive provide(final ConnectionImpl connection) { |
| return new KeepAlive(connection, "no-op-keep-alive") { |
| @Override |
| protected void doKeepAlive() throws TransportException, ConnectionException { |
| // do nothing; |
| } |
| }; |
| } |
| }; |
| |
| protected SFTPClient getSFTPClient(final FlowFile flowFile) throws IOException { |
| // If the client is already initialized then compare the host that the client is connected to with the current |
| // host from the properties/flow-file, and if different then we need to close and reinitialize, if same we can reuse |
| if (sftpClient != null) { |
| final String clientHost = sshClient.getRemoteHostname(); |
| final String propertiesHost = ctx.getProperty(HOSTNAME).evaluateAttributeExpressions(flowFile).getValue(); |
| if (clientHost.equals(propertiesHost)) { |
| // destination matches so we can keep our current session |
| return sftpClient; |
| } else { |
| // this flowFile is going to a different destination, reset session |
| close(); |
| } |
| } |
| |
| // Initialize a new SSHClient... |
| |
| // If use keep-alive is set then set the provider which sends max of 5 keep-alives, otherwise set the no-op provider |
| final DefaultConfig sshClientConfig = new DefaultConfig(); |
| final boolean useKeepAliveOnTimeout = ctx.getProperty(USE_KEEPALIVE_ON_TIMEOUT).asBoolean(); |
| if (useKeepAliveOnTimeout) { |
| sshClientConfig.setKeepAliveProvider(KeepAliveProvider.KEEP_ALIVE); |
| } else { |
| sshClientConfig.setKeepAliveProvider(NO_OP_KEEP_ALIVE); |
| } |
| |
| updateConfigAlgorithms(sshClientConfig); |
| |
| final SSHClient sshClient = new SSHClient(sshClientConfig); |
| |
| // Create a Proxy if the config was specified, proxy will be null if type was NO_PROXY |
| final Proxy proxy; |
| final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(ctx, createComponentProxyConfigSupplier(ctx)); |
| switch (proxyConfig.getProxyType()) { |
| case HTTP: |
| case SOCKS: |
| proxy = proxyConfig.createProxy(); |
| break; |
| default: |
| proxy = null; |
| break; |
| } |
| |
| // If a proxy was specified, configure the client to use a SocketFactory that creates Sockets using the proxy |
| if (proxy != null) { |
| sshClient.setSocketFactory(new SocketFactory() { |
| @Override |
| public Socket createSocket() { |
| return new Socket(proxy); |
| } |
| |
| @Override |
| public Socket createSocket(String s, int i) { |
| return new Socket(proxy); |
| } |
| |
| @Override |
| public Socket createSocket(String s, int i, InetAddress inetAddress, int i1) { |
| return new Socket(proxy); |
| } |
| |
| @Override |
| public Socket createSocket(InetAddress inetAddress, int i) { |
| return new Socket(proxy); |
| } |
| |
| @Override |
| public Socket createSocket(InetAddress inetAddress, int i, InetAddress inetAddress1, int i1) { |
| return new Socket(proxy); |
| } |
| }); |
| } |
| |
| // If strict host key checking is false, add a HostKeyVerifier that always returns true |
| final boolean strictHostKeyChecking = ctx.getProperty(STRICT_HOST_KEY_CHECKING).asBoolean(); |
| if (!strictHostKeyChecking) { |
| sshClient.addHostKeyVerifier(new PromiscuousVerifier()); |
| } |
| |
| // Load known hosts file if specified, otherwise load default |
| final String hostKeyVal = ctx.getProperty(HOST_KEY_FILE).getValue(); |
| if (hostKeyVal != null) { |
| sshClient.loadKnownHosts(new File(hostKeyVal)); |
| // Load default known_hosts file only when 'Strict Host Key Checking' property is enabled |
| } else if (strictHostKeyChecking) { |
| sshClient.loadKnownHosts(); |
| } |
| |
| // Enable compression on the client if specified in properties |
| final PropertyValue compressionValue = ctx.getProperty(FileTransfer.USE_COMPRESSION); |
| if (compressionValue != null && "true".equalsIgnoreCase(compressionValue.getValue())) { |
| sshClient.useCompression(); |
| } |
| |
| // Configure connection timeout |
| final int connectionTimeoutMillis = ctx.getProperty(FileTransfer.CONNECTION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(); |
| sshClient.setTimeout(connectionTimeoutMillis); |
| |
| // Connect to the host and port |
| final String hostname = ctx.getProperty(HOSTNAME).evaluateAttributeExpressions(flowFile).getValue(); |
| final int port = ctx.getProperty(PORT).evaluateAttributeExpressions(flowFile).asInteger().intValue(); |
| sshClient.connect(hostname, port); |
| |
| // Setup authentication methods... |
| final List<AuthMethod> authMethods = new ArrayList<>(); |
| |
| // Add public-key auth if a private key is specified |
| final String privateKeyFile = ctx.getProperty(PRIVATE_KEY_PATH).evaluateAttributeExpressions(flowFile).getValue(); |
| if (privateKeyFile != null) { |
| final String privateKeyPassphrase = ctx.getProperty(PRIVATE_KEY_PASSPHRASE).evaluateAttributeExpressions(flowFile).getValue(); |
| final KeyProvider keyProvider = privateKeyPassphrase == null ? sshClient.loadKeys(privateKeyFile) : sshClient.loadKeys(privateKeyFile, privateKeyPassphrase); |
| final AuthMethod publicKeyAuth = new AuthPublickey(keyProvider); |
| authMethods.add(publicKeyAuth); |
| } |
| |
| // Add password auth if a password is specified |
| final String password = ctx.getProperty(FileTransfer.PASSWORD).evaluateAttributeExpressions(flowFile).getValue(); |
| if (password != null) { |
| final PasswordFinder passwordFinder = PasswordUtils.createOneOff(password.toCharArray()); |
| final AuthMethod passwordAuth = new AuthPassword(passwordFinder); |
| authMethods.add(passwordAuth); |
| } |
| |
| // Authenticate... |
| final String username = ctx.getProperty(USERNAME).evaluateAttributeExpressions(flowFile).getValue(); |
| sshClient.auth(username, authMethods); |
| |
| // At this point we are connected and can create a new SFTPClient which means everything is good |
| this.sshClient = sshClient; |
| this.sftpClient = sshClient.newSFTPClient(); |
| this.closed = false; |
| |
| // Configure timeout for sftp operations |
| final int dataTimeout = ctx.getProperty(FileTransfer.DATA_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(); |
| this.sftpClient.getSFTPEngine().setTimeoutMs(dataTimeout); |
| |
| // Attempt to get the home dir |
| try { |
| this.homeDir = sftpClient.canonicalize(""); |
| } catch (IOException e) { |
| this.homeDir = ""; |
| // For some combination of server configuration and user home directory, getHome() can fail with "2: File not found" |
| // Since homeDir is only used tor SEND provenance event transit uri, this is harmless. Log and continue. |
| logger.debug("Failed to retrieve {} home directory due to {}", new Object[]{username, e.getMessage()}); |
| } |
| |
| return sftpClient; |
| } |
| |
| void updateConfigAlgorithms(final Config config) { |
| if (ctx.getProperty(CIPHERS_ALLOWED).isSet()) { |
| Set<String> allowedCiphers = Arrays.stream(ctx.getProperty(CIPHERS_ALLOWED).evaluateAttributeExpressions().getValue().split(",")) |
| .map(String::trim) |
| .collect(Collectors.toSet()); |
| config.setCipherFactories(config.getCipherFactories().stream() |
| .filter(cipherNamed -> allowedCiphers.contains(cipherNamed.getName())) |
| .collect(Collectors.toList())); |
| } |
| |
| if (ctx.getProperty(KEY_ALGORITHMS_ALLOWED).isSet()) { |
| Set<String> allowedKeyAlgorithms = Arrays.stream(ctx.getProperty(KEY_ALGORITHMS_ALLOWED).evaluateAttributeExpressions().getValue().split(",")) |
| .map(String::trim) |
| .collect(Collectors.toSet()); |
| config.setKeyAlgorithms(config.getKeyAlgorithms().stream() |
| .filter(keyAlgorithmNamed -> allowedKeyAlgorithms.contains(keyAlgorithmNamed.getName())) |
| .collect(Collectors.toList())); |
| } |
| |
| if (ctx.getProperty(KEY_EXCHANGE_ALGORITHMS_ALLOWED).isSet()) { |
| Set<String> allowedKeyExchangeAlgorithms = Arrays.stream(ctx.getProperty(KEY_EXCHANGE_ALGORITHMS_ALLOWED).evaluateAttributeExpressions().getValue().split(",")) |
| .map(String::trim) |
| .collect(Collectors.toSet()); |
| config.setKeyExchangeFactories(config.getKeyExchangeFactories().stream() |
| .filter(keyExchangeNamed -> allowedKeyExchangeAlgorithms.contains(keyExchangeNamed.getName())) |
| .collect(Collectors.toList())); |
| } |
| |
| if (ctx.getProperty(MESSAGE_AUTHENTICATION_CODES_ALLOWED).isSet()) { |
| Set<String> allowedMessageAuthenticationCodes = Arrays.stream(ctx.getProperty(MESSAGE_AUTHENTICATION_CODES_ALLOWED).evaluateAttributeExpressions().getValue().split(",")) |
| .map(String::trim) |
| .collect(Collectors.toSet()); |
| config.setMACFactories(config.getMACFactories().stream() |
| .filter(macNamed -> allowedMessageAuthenticationCodes.contains(macNamed.getName())) |
| .collect(Collectors.toList())); |
| } |
| } |
| |
| @Override |
| public String getHomeDirectory(final FlowFile flowFile) throws IOException { |
| getSFTPClient(flowFile); |
| return this.homeDir; |
| } |
| |
| @Override |
| public void close() throws IOException { |
| if (closed) { |
| return; |
| } |
| closed = true; |
| |
| try { |
| if (null != sftpClient) { |
| sftpClient.close(); |
| } |
| } catch (final Exception ex) { |
| logger.warn("Failed to close SFTPClient due to {}", new Object[] {ex.toString()}, ex); |
| } |
| sftpClient = null; |
| |
| try { |
| if (null != sshClient) { |
| sshClient.disconnect(); |
| } |
| } catch (final Exception ex) { |
| logger.warn("Failed to close SSHClient due to {}", new Object[] {ex.toString()}, ex); |
| } |
| sshClient = null; |
| } |
| |
| @Override |
| public boolean isClosed() { |
| return closed; |
| } |
| |
| @Override |
| @SuppressWarnings("unchecked") |
| public FileInfo getRemoteFileInfo(final FlowFile flowFile, final String path, String filename) throws IOException { |
| final SFTPClient sftpClient = getSFTPClient(flowFile); |
| |
| final List<RemoteResourceInfo> remoteResources; |
| try { |
| remoteResources = sftpClient.ls(path); |
| } catch (final SFTPException e) { |
| if (e.getStatusCode() == Response.StatusCode.NO_SUCH_FILE) { |
| return null; |
| } else { |
| throw new IOException("Failed to obtain file listing for " + path, e); |
| } |
| } |
| |
| RemoteResourceInfo matchingEntry = null; |
| for (final RemoteResourceInfo entry : remoteResources) { |
| if (entry.getName().equalsIgnoreCase(filename)) { |
| matchingEntry = entry; |
| break; |
| } |
| } |
| |
| // Previously JSCH would perform a listing on the full path (path + filename) and would get an exception when it wasn't |
| // a file and then return null, so to preserve that behavior we return null if the matchingEntry is a directory |
| if (matchingEntry != null && matchingEntry.isDirectory()) { |
| return null; |
| } else { |
| return newFileInfo(matchingEntry, path); |
| } |
| } |
| |
| @Override |
| public String put(final FlowFile flowFile, final String path, final String filename, final InputStream content) throws IOException { |
| final SFTPClient sftpClient = getSFTPClient(flowFile); |
| |
| // destination path + filename |
| final String fullPath = (path == null) ? filename : (path.endsWith("/")) ? path + filename : path + "/" + filename; |
| |
| // temporary path + filename |
| String tempFilename = ctx.getProperty(TEMP_FILENAME).evaluateAttributeExpressions(flowFile).getValue(); |
| if (tempFilename == null) { |
| final boolean dotRename = ctx.getProperty(DOT_RENAME).asBoolean(); |
| tempFilename = dotRename ? "." + filename : filename; |
| } |
| final String tempPath = (path == null) ? tempFilename : (path.endsWith("/")) ? path + tempFilename : path + "/" + tempFilename; |
| |
| int perms = 0; |
| final String permissions = ctx.getProperty(PERMISSIONS).evaluateAttributeExpressions(flowFile).getValue(); |
| if (permissions == null || permissions.trim().isEmpty()) { |
| sftpClient.getFileTransfer().setPreserveAttributes(false); //We will accept whatever the default permissions are of the destination |
| perms = 0; |
| } else { |
| sftpClient.getFileTransfer().setPreserveAttributes(true); //We will use the permissions supplied by evaluating processor property expression |
| perms = numberPermissions(permissions); |
| } |
| |
| try { |
| final LocalSourceFile sourceFile = new SFTPFlowFileSourceFile(filename, content, perms); |
| sftpClient.put(sourceFile, tempPath); |
| } catch (final SFTPException e) { |
| throw new IOException("Unable to put content to " + fullPath + " due to " + getMessage(e), e); |
| } |
| |
| final String lastModifiedTime = ctx.getProperty(LAST_MODIFIED_TIME).evaluateAttributeExpressions(flowFile).getValue(); |
| if (lastModifiedTime != null && !lastModifiedTime.trim().isEmpty()) { |
| try { |
| final DateFormat formatter = new SimpleDateFormat(FILE_MODIFY_DATE_ATTR_FORMAT, Locale.US); |
| final Date fileModifyTime = formatter.parse(lastModifiedTime); |
| int time = (int) (fileModifyTime.getTime() / 1000L); |
| |
| final FileAttributes tempAttributes = sftpClient.stat(tempPath); |
| |
| final FileAttributes modifiedAttributes = new FileAttributes.Builder() |
| .withAtimeMtime(tempAttributes.getAtime(), time) |
| .build(); |
| |
| sftpClient.setattr(tempPath, modifiedAttributes); |
| } catch (final Exception e) { |
| logger.error("Failed to set lastModifiedTime on {} to {} due to {}", new Object[] {tempPath, lastModifiedTime, e}); |
| } |
| } |
| |
| final String owner = ctx.getProperty(REMOTE_OWNER).evaluateAttributeExpressions(flowFile).getValue(); |
| if (owner != null && !owner.trim().isEmpty()) { |
| try { |
| sftpClient.chown(tempPath, Integer.parseInt(owner)); |
| } catch (final Exception e) { |
| logger.error("Failed to set owner on {} to {} due to {}", new Object[] {tempPath, owner, e}); |
| } |
| } |
| |
| final String group = ctx.getProperty(REMOTE_GROUP).evaluateAttributeExpressions(flowFile).getValue(); |
| if (group != null && !group.trim().isEmpty()) { |
| try { |
| sftpClient.chgrp(tempPath, Integer.parseInt(group)); |
| } catch (final Exception e) { |
| logger.error("Failed to set group on {} to {} due to {}", new Object[] {tempPath, group, e}); |
| } |
| } |
| |
| if (!filename.equals(tempFilename)) { |
| try { |
| sftpClient.rename(tempPath, fullPath); |
| } catch (final SFTPException e) { |
| try { |
| sftpClient.rm(tempPath); |
| throw new IOException("Failed to rename dot-file to " + fullPath + " due to " + getMessage(e), e); |
| } catch (final SFTPException e1) { |
| throw new IOException("Failed to rename dot-file to " + fullPath + " and failed to delete it when attempting to clean up", e1); |
| } |
| } |
| } |
| |
| return fullPath; |
| } |
| |
| @Override |
| public void rename(final FlowFile flowFile, final String source, final String target) throws IOException { |
| final SFTPClient sftpClient = getSFTPClient(flowFile); |
| try { |
| sftpClient.rename(source, target); |
| } catch (final SFTPException e) { |
| switch (e.getStatusCode()) { |
| case NO_SUCH_FILE: |
| throw new FileNotFoundException("No such file or directory"); |
| case PERMISSION_DENIED: |
| throw new PermissionDeniedException("Could not rename remote file " + source + " to " + target + " due to insufficient permissions"); |
| default: |
| throw new IOException(e); |
| } |
| } |
| } |
| |
| protected int numberPermissions(String perms) { |
| int number = -1; |
| final Pattern rwxPattern = Pattern.compile("^[rwx-]{9}$"); |
| final Pattern numPattern = Pattern.compile("\\d+"); |
| if (rwxPattern.matcher(perms).matches()) { |
| number = 0; |
| if (perms.charAt(0) == 'r') { |
| number |= 0x100; |
| } |
| if (perms.charAt(1) == 'w') { |
| number |= 0x80; |
| } |
| if (perms.charAt(2) == 'x') { |
| number |= 0x40; |
| } |
| if (perms.charAt(3) == 'r') { |
| number |= 0x20; |
| } |
| if (perms.charAt(4) == 'w') { |
| number |= 0x10; |
| } |
| if (perms.charAt(5) == 'x') { |
| number |= 0x8; |
| } |
| if (perms.charAt(6) == 'r') { |
| number |= 0x4; |
| } |
| if (perms.charAt(7) == 'w') { |
| number |= 0x2; |
| } |
| if (perms.charAt(8) == 'x') { |
| number |= 0x1; |
| } |
| } else if (numPattern.matcher(perms).matches()) { |
| try { |
| number = Integer.parseInt(perms, 8); |
| } catch (NumberFormatException ignore) { |
| } |
| } |
| return number; |
| } |
| |
| } |
