Skip to content

Commit

Permalink
Switch from jsch to sshd-sftp fo SSH impl and tests
Browse files Browse the repository at this point in the history 
We switch to https://github.com/apache/mina-sshd for the implementation of the SSH crawler.
It also helps us to now run integration tests.
  • Loading branch information
@dadoonet
dadoonet committed Aug 30, 2024
1 parent 19ef29c commit 66e7bd7
Show file tree
Hide file tree
Showing 9 changed files with 177 additions and 103 deletions.
6 changes: 3 additions & 3 deletions crawler/crawler-ssh/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,10 @@
<artifactId>fscrawler-crawler-abstract</artifactId>
</dependency>

<!--Dependency for parsing remote ssh directory [http://www.jcraft.com/jsch/]-->
<!-- Dependency for parsing remote ssh directory https://github.com/apache/mina-sshd/ -->
<dependency>
<groupId>com.jcraft</groupId>
<artifactId>jsch</artifactId>
<groupId>org.apache.sshd</groupId>
<artifactId>sshd-sftp</artifactId>
</dependency>
</dependencies>

Expand Down
135 changes: 75 additions & 60 deletions ...r-ssh/src/main/java/fr/pilato/elasticsearch/crawler/fs/crawler/ssh/FileAbstractorSSH.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,6 @@

package fr.pilato.elasticsearch.crawler.fs.crawler.ssh;

import com.jcraft.jsch.Channel;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import fr.pilato.elasticsearch.crawler.fs.crawler.FileAbstractModel;
import fr.pilato.elasticsearch.crawler.fs.crawler.FileAbstractor;
import fr.pilato.elasticsearch.crawler.fs.settings.FsSettings;
Expand All @@ -32,71 +27,80 @@
import org.apache.commons.io.FilenameUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.sshd.client.SshClient;
import org.apache.sshd.client.keyverifier.AcceptAllServerKeyVerifier;
import org.apache.sshd.client.session.ClientSession;
import org.apache.sshd.common.keyprovider.FileKeyPairProvider;
import org.apache.sshd.sftp.client.SftpClient;
import org.apache.sshd.sftp.client.SftpClientFactory;

import java.io.InputStream;
import java.time.Instant;
import java.nio.file.Paths;
import java.security.KeyPair;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Vector;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

public class FileAbstractorSSH extends FileAbstractor<ChannelSftp.LsEntry> {
public class FileAbstractorSSH extends FileAbstractor<SftpClient.DirEntry> {
private final Logger logger = LogManager.getLogger(FileAbstractorSSH.class);

private ChannelSftp sftp;
private ClientSession session;
private SshClient sshClient;
private SftpClient sftpClient;

public FileAbstractorSSH(FsSettings fsSettings) {
super(fsSettings);
}

@Override
public FileAbstractModel toFileAbstractModel(String path, ChannelSftp.LsEntry file) {
public FileAbstractModel toFileAbstractModel(String path, SftpClient.DirEntry file) {
return new FileAbstractModel(
file.getFilename(),
!file.getAttrs().isDir(),
!file.getAttributes().isDirectory(),
// We are using here the local TimeZone as a reference. If the remote system is under another TZ, this might cause issues
LocalDateTime.ofInstant(Instant.ofEpochMilli(file.getAttrs().getMTime()*1000L), ZoneId.systemDefault()),
LocalDateTime.ofInstant(file.getAttributes().getModifyTime().toInstant(), ZoneId.systemDefault()),
// We don't have the creation date
null,
// We are using here the local TimeZone as a reference. If the remote system is under another TZ, this might cause issues
LocalDateTime.ofInstant(Instant.ofEpochMilli(file.getAttrs().getATime()*1000L), ZoneId.systemDefault()),
LocalDateTime.ofInstant(file.getAttributes().getAccessTime().toInstant(), ZoneId.systemDefault()),
FilenameUtils.getExtension(file.getFilename()),
path,
path.equals("/") ? path.concat(file.getFilename()) : path.concat("/").concat(file.getFilename()),
file.getAttrs().getSize(),
Integer.toString(file.getAttrs().getUId()),
Integer.toString(file.getAttrs().getGId()),
file.getAttrs().getPermissions());
file.getAttributes().getSize(),
Integer.toString(file.getAttributes().getUserId()),
Integer.toString(file.getAttributes().getGroupId()),
file.getAttributes().getPermissions());
}

@Override
public InputStream getInputStream(FileAbstractModel file) throws Exception {
return sftp.get(file.getFullpath());
return sftpClient.read(file.getFullpath());
}

@Override
public void closeInputStream(InputStream inputStream) throws IOException {
inputStream.close();
}

@SuppressWarnings("unchecked")
@Override
public Collection<FileAbstractModel> getFiles(String dir) throws Exception {
logger.debug("Listing local files from {}", dir);
Vector<ChannelSftp.LsEntry> ls;

ls = sftp.ls(dir);
if (ls == null) return null;
Iterable<SftpClient.DirEntry> ls;

Collection<FileAbstractModel> result = new ArrayList<>(ls.size());
// Iterate other files
// We ignore here all files like . and ..
result.addAll(ls.stream().filter(file -> !".".equals(file.getFilename()) &&
!"..".equals(file.getFilename()))
ls = sftpClient.readDir(dir);

/*
Iterate other files
We ignore here all files like "." and ".."
*/
Collection<FileAbstractModel> result = StreamSupport.stream(ls.spliterator(), false)
.filter(file -> !".".equals(file.getFilename()) &&
!"..".equals(file.getFilename()))
.map(file -> toFileAbstractModel(dir, file))
.collect(Collectors.toList()));
.collect(Collectors.toList());

logger.debug("{} local files found", result.size());
return result;
Expand All @@ -105,7 +109,7 @@ public Collection<FileAbstractModel> getFiles(String dir) throws Exception {
@Override
public boolean exists(String dir) {
try {
sftp.ls(dir);
sftpClient.readDir(dir);
} catch (Exception e) {
return false;
}
Expand All @@ -114,51 +118,62 @@ public boolean exists(String dir) {

@Override
public void open() throws Exception {
sftp = openSSHConnection(fsSettings.getServer());
sshClient = createSshClient();
session = openSshSession(sshClient, fsSettings.getServer());
sftpClient = createSftpClient(session);
}

@Override
public void close() throws Exception {
if (sftp != null) {
sftp.getSession().disconnect();
sftp.disconnect();
if (sshClient != null) {
if (session != null) {
if (sftpClient != null) {
logger.debug("Closing SFTP Client");
sftpClient.close();
}
logger.debug("Closing SSH Session");
session.close();
}
logger.debug("Closing SSH Client");
sshClient.close();
}
}

private ChannelSftp openSSHConnection(Server server) throws Exception {
private SshClient createSshClient() {
logger.debug("Create and start SSH client");

SshClient client = SshClient.setUpDefaultClient();
client.setServerKeyVerifier(AcceptAllServerKeyVerifier.INSTANCE);
client.start();
return client;
}

private ClientSession openSshSession(SshClient sshClient, Server server) throws Exception {
logger.debug("Opening SSH connection to {}@{}", server.getUsername(), server.getHostname());

JSch jsch = new JSch();
Session session = jsch.getSession(server.getUsername(), server.getHostname(), server.getPort());
java.util.Properties config = new java.util.Properties();
config.put("StrictHostKeyChecking", "no");
if (server.getPemPath() != null) {
jsch.addIdentity(server.getPemPath());
}
session.setConfig(config);
ClientSession session = sshClient.connect(server.getUsername(), server.getHostname(), server.getPort()).verify().getSession();

if (server.getPassword() != null) {
session.setPassword(server.getPassword());
session.addPasswordIdentity(server.getPassword()); // for password-based authentication
}

try {
session.connect();
} catch (JSchException e) {
logger.warn("Cannot connect with SSH to {}@{}: {}", server.getUsername(),
server.getHostname(), e.getMessage());
throw e;
if (server.getPemPath() != null) {
// for password-less authentication
FileKeyPairProvider fileKeyPairProvider = new FileKeyPairProvider(Paths.get(server.getPemPath()));
Iterable<KeyPair> keyPairs = fileKeyPairProvider.loadKeys(null);
for (KeyPair keyPair : keyPairs) {
session.addPublicKeyIdentity(keyPair);
}
}

//Open a new session for SFTP.
Channel channel = session.openChannel("sftp");
channel.connect();
session.auth().verify();

//checking SSH client connection.
if (!channel.isConnected()) {
logger.warn("Cannot connect with SSH to {}@{}", server.getUsername(),
server.getHostname());
throw new RuntimeException("Can not connect to " + server.getUsername() + "@" + server.getHostname());
}
logger.debug("SSH connection successful");
return (ChannelSftp) channel;
return session;
}

private SftpClient createSftpClient(ClientSession session) throws Exception {
logger.debug("Create SFTP client");
return SftpClientFactory.instance().createSftpClient(session);
}
}
1 change: 0 additions & 1 deletion ...ticsearch-client/src/main/java/fr/pilato/elasticsearch/crawler/fs/client/ESBoolQuery.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/

package fr.pilato.elasticsearch.crawler.fs.client;

import java.util.ArrayList;
Expand Down
1 change: 0 additions & 1 deletion ...sts/src/test/java/fr/pilato/elasticsearch/crawler/fs/test/integration/AbstractITCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -508,5 +508,4 @@ public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOEx
}
});
}

}
118 changes: 91 additions & 27 deletions ...fr/pilato/elasticsearch/crawler/fs/test/integration/elasticsearch/FsCrawlerTestSshIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,51 +23,115 @@
import fr.pilato.elasticsearch.crawler.fs.settings.Fs;
import fr.pilato.elasticsearch.crawler.fs.settings.Server;
import fr.pilato.elasticsearch.crawler.fs.test.integration.AbstractFsCrawlerITCase;
import org.junit.Ignore;
import org.apache.sshd.common.config.keys.writer.openssh.OpenSSHKeyPairResourceWriter;
import org.apache.sshd.server.SshServer;
import org.apache.sshd.server.config.keys.AuthorizedKeysAuthenticator;
import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
import org.apache.sshd.sftp.server.SftpFileSystemAccessor;
import org.apache.sshd.sftp.server.SftpSubsystemFactory;
import org.apache.sshd.sftp.server.SftpSubsystemProxy;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.InvalidPathException;
import java.nio.file.Path;
import java.security.*;
import java.util.Collections;

/**
* Test crawler with SSH
*/
public class FsCrawlerTestSshIT extends AbstractFsCrawlerITCase {

/**
* You have to adapt this test to your own system (login / password and SSH connexion)
* So this test is disabled by default
*/
@Test @Ignore
public void test_ssh() throws Exception {
String username = "USERNAME";
String password = "PASSWORD";
String hostname = "localhost";
private final static String SSH_USERNAME = "USERNAME";
private final static String SSH_PASSWORD = "PASSWORD";

private SshServer sshd = null;

@Before
public void setup() throws IOException, NoSuchAlgorithmException {
SftpSubsystemFactory factory = new SftpSubsystemFactory.Builder()
.withFileSystemAccessor(new SftpFileSystemAccessor() {
@Override
public Path resolveLocalFilePath(SftpSubsystemProxy subsystem, Path rootDir, String remotePath) throws InvalidPathException {
String path = remotePath;
if (remotePath.startsWith("/")) {
path = remotePath.substring(1);
}
return currentTestResourceDir.resolve(path);
}
})
.build();

// Generate the key files for our SSH tests
KeyPair keyPair = KeyPairGenerator.getInstance("RSA").generateKeyPair();
saveKeyPair(rootTmpDir, keyPair);

sshd = SshServer.setUpDefaultServer();
sshd.setHost("localhost");
sshd.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(rootTmpDir.resolve("host.ser")));
sshd.setPasswordAuthenticator((username, password, session) ->
SSH_USERNAME.equals(username) && SSH_PASSWORD.equals(password));
sshd.setPublickeyAuthenticator(new AuthorizedKeysAuthenticator(rootTmpDir.resolve("public.key")));

sshd.setSubsystemFactories(Collections.singletonList(factory));
sshd.start();

logger.info(" -> Started fake SSHD service on {}:{}", sshd.getHost(), sshd.getPort());
}

private void saveKeyPair(Path path, KeyPair keyPair) {
OpenSSHKeyPairResourceWriter writer = new OpenSSHKeyPairResourceWriter();

// Store Public Key.
try (FileOutputStream fos = new FileOutputStream(path.resolve("public.key").toFile())) {
writer.writePublicKey(keyPair.getPublic(), "Public Key for tests", fos);
} catch (GeneralSecurityException | IOException e) {
logger.error("Failed to save public key", e);
}

// Store Private Key.
try (FileOutputStream fos = new FileOutputStream(path.resolve("private.key").toFile())) {
writer.writePrivateKey(keyPair, "Private Key for tests", null, fos);
} catch (GeneralSecurityException | IOException e) {
logger.error("Failed to save public key", e);
}
}

Fs fs = startCrawlerDefinition().build();
@After
public void shutDown() throws IOException {
if (sshd != null) {
sshd.stop(true);
logger.info(" -> Stopped fake SSHD service on {}:{}", sshd.getHost(), sshd.getPort());
}
}

@Test
public void test_ssh() throws Exception {
Fs fs = startCrawlerDefinition("/").build();
Server server = Server.builder()
.setHostname(hostname)
.setUsername(username)
.setPassword(password)
.setHostname(sshd.getHost())
.setPort(sshd.getPort())
.setUsername(SSH_USERNAME)
.setPassword(SSH_PASSWORD)
.setProtocol(Server.PROTOCOL.SSH)
.build();
crawler = startCrawler(getCrawlerName(), fs, endCrawlerDefinition(getCrawlerName()), server);

countTestHelper(new ESSearchRequest().withIndex(getCrawlerName()), 2L, null);
}

/**
* You have to adapt this test to your own system (login / pem file and SSH connexion)
* So this test is disabled by default
*/
@Test @Ignore
@Test
public void test_ssh_with_key() throws Exception {
String username = "USERNAME";
String path_to_pem_file = "/path/to/private_key.pem";
String hostname = "localhost";

Fs fs = startCrawlerDefinition().build();
Fs fs = startCrawlerDefinition("/").build();
Server server = Server.builder()
.setHostname(hostname)
.setUsername(username)
.setPemPath(path_to_pem_file)
.setHostname(sshd.getHost())
.setPort(sshd.getPort())
.setUsername(SSH_USERNAME)
.setPemPath(rootTmpDir.resolve("private.key").toFile().getAbsolutePath())
.setProtocol(Server.PROTOCOL.SSH)
.build();
crawler = startCrawler(getCrawlerName(), fs, endCrawlerDefinition(getCrawlerName()), server);
Expand Down
3 changes: 3 additions & 0 deletions integration-tests/src/test/resources/log4j2.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,9 @@
<Logger name="org.mockftpserver" level="error" additivity="false">
<AppenderRef ref="CONSOLE"/>
</Logger>
<Logger name="org.apache.sshd" level="info" additivity="false">
<AppenderRef ref="CONSOLE"/>
</Logger>
<Logger name="org.apache.tika.parser.ocr.TesseractOCRParser" level="error" additivity="false">
<AppenderRef ref="CONSOLE"/>
</Logger>
Expand Down
Loading

0 comments on commit 66e7bd7

Please sign in to comment.