Support for ssh key instead of password

Closes #55

README.md

@@ -107,7 +107,9 @@ curl -XPUT 'localhost:9200/_river/mynewriver/_meta' -d '{
 Indexing using SSH
 ------------------
 
-You can now index files remotely using SSH:
+You can now index files remotely using SSH.
+
+### Username / Password
 
 * FS URL: `/tmp3`
 * Server: `mynode.mydomain.com`
@@ -134,6 +136,34 @@ curl -XPUT 'localhost:9200/_river/mysshriver/_meta' -d '{
 }'
 ```
 
+### Using Username / PEM file
+
+
+* FS URL: `/tmp3`
+* Server: `mynode.mydomain.com`
+* Username: `username`
+* PEM File: `/path/to/private_key.pem`
+* Protocol: `ssh` (default to `local`)
+* Port: `22` (default to `22`)
+* Update Rate: every hour (60 * 60 * 1000 = 3600000 ms)
+* Get only docs like `*.doc`, `*.xls` and `*.pdf`
+
+```sh
+curl -XPUT 'localhost:9200/_river/mysshriver/_meta' -d '{
+  "type": "fs",
+  "fs": {
+	"url": "/tmp3",
+	"server": "mynode.mydomain.com",
+	"port": 22,
+	"username": "username",
+	"pem_path": "/path/to/private_key.pem",
+	"protocol": "ssh",
+	"update_rate": 3600000,
+	"includes": [ "*.doc" , "*.xls", "*.pdf" ]
+  }
+}'
+```
+
 Searching for docs
 ------------------
 
@@ -833,6 +863,7 @@ Here is a full list of existing settings:
 | `fs.username`                    | [Indexing using SSH](#indexing-using-ssh)                                         |
 | `fs.password`                    | [Indexing using SSH](#indexing-using-ssh)                                         |
 | `fs.protocol`                    | [Indexing using SSH](#indexing-using-ssh)                                         |
+| `fs.pem_path`                    | [Indexing using SSH](#indexing-using-ssh)                                         |
 | `fs.json_support`                | [Indexing JSon docs](#indexing-json-docs)                                         |
 | `fs.filename_as_id`              | [Indexing JSon docs](#indexing-json-docs)                                         |
 | `fs.add_filesize`                | [Disabling file size field](#disabling-file-size-field)                           |

src/main/java/fr/pilato/elasticsearch/river/fs/river/FileAbstractorSSH.java

@@ -85,8 +85,13 @@ public ChannelSftp openSSHConnection(FsRiverFeedDefinition fsdef) throws Excepti
         Session session = jsch.getSession(fsdef.getUsername(), fsdef.getServer(), fsdef.getPort());
         java.util.Properties config = new java.util.Properties();
         config.put("StrictHostKeyChecking", "no");
+        if (fsdef.getPemFilePath() != null) {
+            jsch.addIdentity(fsdef.getPemFilePath());
+        }
         session.setConfig(config);
-        session.setPassword(fsdef.getPassword());
+        if (fsdef.getPassword() != null) {
+            session.setPassword(fsdef.getPassword());
+        }
         session.connect();
 
         //Open a new session for SFTP.

src/main/java/fr/pilato/elasticsearch/river/fs/river/FsRiver.java

@@ -136,6 +136,7 @@ public FsRiver(RiverName riverName, RiverSettings settings, Client client)
             String server = XContentMapValues.nodeStringValue(feed.get("server"), null);
             int port = XContentMapValues.nodeIntegerValue(feed.get("port"), PROTOCOL.SSH_PORT);
             String protocol = XContentMapValues.nodeStringValue(feed.get("protocol"), PROTOCOL.LOCAL);
+            String pemPathFile = XContentMapValues.nodeStringValue(feed.get("pem_path"), null);
 
             // https://github.com/dadoonet/fsriver/issues/35 : Option to not delete documents when files are removed
             boolean removeDeleted = XContentMapValues.nodeBooleanValue(feed.get("remove_deleted"), true);
@@ -144,7 +145,7 @@ public FsRiver(RiverName riverName, RiverSettings settings, Client client)
             fsDefinition = new FsRiverFeedDefinition(riverName.getName(), url,
                     updateRate, Arrays.asList(includes), Arrays.asList(excludes),
                     jsonSupport, filenameAsId, addFilesize, indexedChars,
-                    username, password, server, port, protocol, removeDeleted, storeSource);
+                    username, password, server, port, protocol, pemPathFile, removeDeleted, storeSource);
         } else {
             String url = "/esdir";
             logger.warn(
@@ -153,7 +154,7 @@ public FsRiver(RiverName riverName, RiverSettings settings, Client client)
             int updateRate = 60 * 60 * 1000;
             fsDefinition = new FsRiverFeedDefinition(riverName.getName(), url,
                     updateRate, Arrays.asList("*.txt", "*.pdf"), Arrays.asList("*.exe"), false, false, true, 0.0,
-                    null, null, null, PROTOCOL.SSH_PORT, null, true, false);
+                    null, null, null, PROTOCOL.SSH_PORT, null, null, true, false);
         }
 
         if (settings.settings().containsKey("index")) {
@@ -191,7 +192,7 @@ public FsRiver(RiverName riverName, RiverSettings settings, Client client)
         if (PROTOCOL.SSH.equals(fsDefinition.getProtocol()) &&
                 !Strings.hasLength(fsDefinition.getUsername())) {
             // Non supported protocol
-            logger.error("When using SSH, you need to set a username and probably a password. Disabling river");
+            logger.error("When using SSH, you need to set a username and probably a password or a pem file. Disabling river");
             closed = true;
         }
     }

src/main/java/fr/pilato/elasticsearch/river/fs/river/FsRiverFeedDefinition.java

@@ -40,14 +40,15 @@ public class FsRiverFeedDefinition {
     private String password;
     private String server;
     private String protocol;
+    private String pemFilePath;
     private boolean removeDeleted;
     private boolean storeSource;
     private int port;
 
     public FsRiverFeedDefinition(String rivername, String url, int updateRate, List<String> includes,
                                  List<String> excludes, boolean jsonSupport, boolean filenameAsId,
                                  boolean addFilesize, double indexedChars, String username, String password,
-                                 String server, int port, String protocol, boolean removeDeleted, boolean storeSource) {
+                                 String server, int port, String protocol, String pemFilePath, boolean removeDeleted, boolean storeSource) {
         assert (excludes != null);
         assert (includes != null);
         this.includes = includes;
@@ -63,6 +64,7 @@ public FsRiverFeedDefinition(String rivername, String url, int updateRate, List<
         this.password = password;
         this.server = server;
         this.protocol = protocol;
+        this.pemFilePath = pemFilePath;
         this.removeDeleted = removeDeleted;
         this.storeSource = storeSource;
         this.port = port;
@@ -128,6 +130,10 @@ public String getProtocol() {
         return this.protocol;
     }
 
+    public String getPemFilePath() {
+        return pemFilePath;
+    }
+
     public boolean isRemoveDeleted() {
         return removeDeleted;
     }

src/test/java/fr/pilato/elasticsearch/river/fs/integration/FsRiverAllParametersTest.java

@@ -555,6 +555,26 @@ public void test_ssh() throws Exception {
         countTestHelper("ssh", null, 2);
     }
 
+    /**
+     * 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
+    public void test_ssh_with_key() throws Exception {
+        String username = "USERNAME";
+        String path_to_pem_file = "/path/to/private_key.pem";
+        String server = "localhost";
+
+        XContentBuilder river = startRiverDefinition("testsubdir")
+                .field("username", username)
+                .field("pem_path", path_to_pem_file)
+                .field("protocol", FsRiver.PROTOCOL.SSH)
+                .field("server", server);
+        startRiver("ssh", endRiverDefinition(river));
+
+        countTestHelper("ssh", null, 2);
+    }
+
     @Test
     public void test_stop_river_while_adding_content() throws Exception {
         String sourcedir = "test_add_new_file";