smoketest: ssh: verify login of valid and invalid test user

In order to test for proper system authentication and security setup a new
testcase is added which performs an SSH login and command execution with a
predefined user. The result (output of uname -a) must match the output if the
command is run natively.

We also try to login as an invalid user - this is not allowed to work.

1 files changed, 49 insertions, 0 deletions

diff --git a/smoketest/scripts/cli/test_service_ssh.py b/smoketest/scripts/cli/test_service_ssh.py
index a54c03919..8c23b1b62 100755
--- a/smoketest/scripts/cli/test_service_ssh.py
+++ b/smoketest/scripts/cli/test_service_ssh.py
@@ -15,9 +15,12 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
+import paramiko
 import re
 import unittest
 
+from pwd import getpwall
+
 from base_vyostest_shim import VyOSUnitTestSHIM
 
 from vyos.configsession import ConfigSessionError
@@ -167,5 +170,51 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase):
         # delete VRF
         self.cli_delete(['vrf', 'name', vrf])
 
+    def test_ssh_login(self):
+        # Perform SSH login and command execution with a predefined user. The
+        # result (output of uname -a) must match the output if the command is
+        # run natively.
+        #
+        # We also try to login as an invalid user - this is not allowed to work.
+
+        def ssh_send_cmd(command, username, password, host='localhost'):
+            """ SSH command execution helper """
+            # Try to login via SSH
+            ssh_client = paramiko.SSHClient()
+            ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+            ssh_client.connect(hostname='localhost', username=username, password=password)
+            _, stdout, stderr = ssh_client.exec_command(command)
+            output = stdout.read().decode().strip()
+            error = stderr.read().decode().strip()
+            ssh_client.close()
+            return output, error
+
+        test_user = 'ssh_test'
+        test_pass = 'v2i57DZs8idUwMN3VC92'
+        test_command = 'uname -a'
+
+        self.cli_set(base_path)
+        self.cli_set(['system', 'login', 'user', test_user, 'authentication', 'plaintext-password', test_pass])
+
+        # commit changes
+        self.cli_commit()
+
+        # Login with proper credentials
+        output, error = ssh_send_cmd(test_command, test_user, test_pass)
+        # verify login
+        self.assertFalse(error)
+        self.assertEqual(output, cmd(test_command))
+
+        # Login with invalid credentials
+        with self.assertRaises(paramiko.ssh_exception.AuthenticationException):
+            output, error = ssh_send_cmd(test_command, 'invalid_user', 'invalid_password')
+
+        self.cli_delete(['system', 'login', 'user', test_user])
+        self.cli_commit()
+
+        # After deletion the test user is not allowed to remain in /etc/passwd
+        usernames = [x[0] for x in getpwall()]
+        self.assertNotIn(test_user, usernames)
+
 if __name__ == '__main__':
     unittest.main(verbosity=2)