Use UserWithPKey instead of pkey parameter (#144)

Fixes: #23

Signed-off-by: Lukas Bednar <[email protected]>

Author: lukas-bednar

README.rst

@@ -21,15 +21,22 @@ that means SSH server must be running there already.
     h = Host("10.11.12.13")
     h.users.append(RootUser('123456'))
 
-    # Use with ssh key. export HOST_SSH_KEY to use specific ssh key, default is ~/.ssh/id_rsa
-    host.executor_factory = rrmngmnt.ssh.RemoteExecutorFactory(use_pkey=True)
-
     exec = h.executor()
     # Run with sudo
     exec = h.executor(sudo=True)
 
     print exec.run_cmd(['echo', 'Hello World'])
 
+Using SSH key for authentication
+
+.. code:: python
+
+    from rrmngmnt import Host, UserWithPKey
+
+    h = Host("10.11.12.13")
+    user = UserWithPKey('user', '/path/to/pkey'))
+
+    h.executor(user).run_cmd(['echo', 'Use pkey for auth instead of password'])
 
 Features
 --------

rrmngmnt/__init__.py

@@ -1,6 +1,7 @@
 from rrmngmnt.host import Host
 from rrmngmnt.user import (
     User,
+    UserWithPKey,
     RootUser,
     Domain,
     InternalDomain,
@@ -12,6 +13,7 @@
 __all__ = [
     'Host',
     'User',
+    'UserWithPKey',
     'RootUser',
     'Domain',
     'InternalDomain',

rrmngmnt/host.py

@@ -228,7 +228,7 @@ def executor(self, user=None, pkey=False, sudo=False):
         if pkey:
             warnings.warn(
                 "Parameter 'pkey' is deprecated and will be removed in future."
-                "Please use ssh.RemoteExecutorFactory to set this parameter."
+                "Please use user.UserWithPKey user instead."
             )
             ef = copy.copy(self.executor_factory)
             ef.use_pkey = pkey

rrmngmnt/ssh.py

@@ -4,8 +4,10 @@
 import paramiko
 import contextlib
 import subprocess
+import warnings
 from rrmngmnt.common import normalize_string
 from rrmngmnt.executor import Executor, ExecutorFactory
+from rrmngmnt.user import UserWithPKey
 
 
 AUTHORIZED_KEYS = os.path.join("%s", ".ssh/authorized_keys")
@@ -42,7 +44,7 @@ def process(self, msg, kwargs):
                 "[%s@%s/%s] %s" % (
                     self.extra['self'].user.name,
                     self.extra['self'].address,
-                    self.extra['self'].user.password,
+                    self.extra['self'].user.credentials,
                     msg,
                 ),
                 kwargs,
@@ -59,7 +61,11 @@ def __init__(self, executor, timeout=None):
             self._timeout = timeout
             self._ssh = paramiko.SSHClient()
             self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-            if self._executor.use_pkey:
+            if isinstance(self._executor.user, UserWithPKey):
+                self.pkey = paramiko.RSAKey.from_private_key_file(
+                    self._executor.user.private_key
+                )
+            elif self._executor.use_pkey:
                 self.pkey = paramiko.RSAKey.from_private_key_file(
                     os.getenv(
                         "HOST_SSH_KEY", ID_RSA_PRV % os.path.expanduser('~')
@@ -223,6 +229,11 @@ def __init__(self, user, address, use_pkey=False, port=22, sudo=False):
         self.use_pkey = use_pkey
         self.port = port
         self.sudo = sudo
+        if use_pkey:
+            warnings.warn(
+                "Parameter 'use_pkey' is deprecated and will be removed in "
+                "future. Please use user.UserWithPKey user instead."
+            )
 
     def session(self, timeout=None):
         """
@@ -310,6 +321,11 @@ class RemoteExecutorFactory(ExecutorFactory):
     def __init__(self, use_pkey=False, port=22):
         self.use_pkey = use_pkey
         self.port = port
+        if use_pkey:
+            warnings.warn(
+                "Parameter 'use_pkey' is deprecated and will be removed in "
+                "future. Please use user.UserWithPKey user instead."
+            )
 
     def build(self, host, user, sudo=False):
         return RemoteExecutor(

rrmngmnt/user.py

@@ -1,3 +1,4 @@
+import os
 from rrmngmnt.resource import Resource
 
 
@@ -19,6 +20,27 @@ def full_name(self):
     def get_full_name(self):
         return self.name
 
+    @property
+    def credentials(self):
+        return self.get_credentials()
+
+    def get_credentials(self):
+        return self.password
+
+
+class UserWithPKey(User):
+    def __init__(self, name, private_key):
+        """
+        Args:
+            private_key (str): Path to private key
+            name (str): User name
+        """
+        super(UserWithPKey, self).__init__(name, None)
+        self.private_key = os.path.expanduser(private_key)
+
+    def get_credentials(self):
+        return self.private_key
+
 
 class RootUser(User):
     NAME = 'root'

tests/test_host.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 
 
-from rrmngmnt import Host, User, RootUser
+from rrmngmnt import Host, User, RootUser, UserWithPKey
 from rrmngmnt.executor import Executor
 import pytest
 
@@ -27,12 +27,18 @@ def test_custom_user(self):
         e = get_host().executor(user=user)
         assert e.user.name == 'lukas'
 
+    def test_pkey_user(self):
+        user = UserWithPKey('lukas', '/path/to/key')
+        e = get_host().executor(user=user)
+        assert e.user.name == 'lukas'
+        assert e.user.credentials == '/path/to/key'
+
     def test_executor_user(self):
         user = User('lukas', '123456')
         h = get_host()
         h.executor_user = user
         e = h.executor()
-        e.user.name == 'lukas'
+        assert e.user.name == 'lukas'
 
     def test_executor_with_pkey(self):
         user = User('core', '12')

tests/test_user.py

@@ -1,9 +1,11 @@
 # -*- coding: utf-8 -*-
-from rrmngmnt import User, InternalDomain, ADUser, Domain
+from rrmngmnt import User, UserWithPKey, InternalDomain, ADUser, Domain
 
 
 def test_user():
-    assert "user" == User("user", 'pass').full_name
+    user = User("user", 'pass')
+    assert "user" == user.full_name
+    assert user.credentials == user.password
 
 
 def test_indernal_domain():
@@ -16,3 +18,9 @@ def test_domain():
     assert "[email protected]" == ADUser(
         "user", "pass", Domain("example.com"),
     ).full_name
+
+
+def test_user_with_pkey():
+    user = UserWithPKey("user", "/path/to/key")
+    assert 'user' == user.full_name
+    assert '/path/to/key' == user.credentials