Merge pull request #2038 from c-po/t141-tacacs

T141: add TACACS support

3 files changed, 57 insertions, 19 deletions

diff --git a/smoketest/scripts/cli/base_vyostest_shim.py b/smoketest/scripts/cli/base_vyostest_shim.py
index 7cfb53045..d56a8eb33 100644
--- a/smoketest/scripts/cli/base_vyostest_shim.py
+++ b/smoketest/scripts/cli/base_vyostest_shim.py
@@ -14,6 +14,7 @@
 
 import os
 import unittest
+import paramiko
 
 from time import sleep
 from typing import Type
@@ -87,6 +88,19 @@ class VyOSUnitTestSHIM:
                 pprint.pprint(out)
             return out
 
+        def ssh_send_cmd(command, username, password, hostname='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=hostname, username=username, password=password)
+            print(host, username, password)
+            _, stdout, stderr = ssh_client.exec_command(command)
+            output = stdout.read().decode().strip()
+            error = stderr.read().decode().strip()
+            ssh_client.close()
+            return output, error
+
 # standard construction; typing suggestion: https://stackoverflow.com/a/70292317
 def ignore_warning(warning: Type[Warning]):
     import warnings
diff --git a/smoketest/scripts/cli/test_service_ssh.py b/smoketest/scripts/cli/test_service_ssh.py
index 8de98f34f..e03907dd8 100755
--- a/smoketest/scripts/cli/test_service_ssh.py
+++ b/smoketest/scripts/cli/test_service_ssh.py
@@ -174,18 +174,6 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase):
         #
         # 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'
@@ -197,14 +185,14 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase):
         self.cli_commit()
 
         # Login with proper credentials
-        output, error = ssh_send_cmd(test_command, test_user, test_pass)
+        output, error = self.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')
+            output, error = self.ssh_send_cmd(test_command, 'invalid_user', 'invalid_password')
 
         self.cli_delete(['system', 'login', 'user', test_user])
         self.cli_commit()
diff --git a/smoketest/scripts/cli/test_system_login.py b/smoketest/scripts/cli/test_system_login.py
index a1d2ba2ad..8a4f5fdd1 100755
--- a/smoketest/scripts/cli/test_system_login.py
+++ b/smoketest/scripts/cli/test_system_login.py
@@ -17,13 +17,13 @@
 import re
 import platform
 import unittest
+import paramiko
 
 from base_vyostest_shim import VyOSUnitTestSHIM
 
-from distutils.version import LooseVersion
-from platform import release as kernel_version
 from subprocess import Popen, PIPE
 from pwd import getpwall
+from time import sleep
 
 from vyos.configsession import ConfigSessionError
 from vyos.util import cmd
@@ -53,12 +53,16 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase):
         # ensure we can also run this test on a live system - so lets clean
         # out the current configuration which will break this test
         cls.cli_delete(cls, base_path + ['radius'])
+        cls.cli_delete(cls, base_path + ['tacacs'])
 
     def tearDown(self):
         # Delete individual users from configuration
         for user in users:
             self.cli_delete(base_path + ['user', user])
 
+        self.cli_delete(base_path + ['radius'])
+        self.cli_delete(base_path + ['tacacs'])
+
         self.cli_commit()
 
         # After deletion, a user is not allowed to remain in /etc/passwd
@@ -149,9 +153,6 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase):
 
         # T2886 - RADIUS authentication - check for statically compiled options
         options = ['CONFIG_AUDIT', 'CONFIG_AUDITSYSCALL', 'CONFIG_AUDIT_ARCH']
-        if LooseVersion(kernel_version()) < LooseVersion('5.0'):
-            options.append('CONFIG_AUDIT_WATCH')
-            options.append('CONFIG_AUDIT_TREE')
 
         for option in options:
             self.assertIn(f'{option}=y', kernel_config)
@@ -284,6 +285,41 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase):
         self.cli_delete(base_path + ['timeout'])
         self.cli_delete(base_path + ['max-login-session'])
 
+    def test_system_login_tacacs(self):
+        tacacs_secret = 'tac_plus_key'
+        tacacs_servers = ['100.64.0.11', '100.64.0.12']
+
+        # Enable TACACS
+        for server in tacacs_servers:
+            self.cli_set(base_path + ['tacacs', 'server', server, 'key', tacacs_secret])
+
+        self.cli_commit()
+
+        # NSS
+        nsswitch_conf = read_file('/etc/nsswitch.conf')
+        tmp = re.findall(r'passwd:\s+tacplus\s+files', nsswitch_conf)
+        self.assertTrue(tmp)
+
+        tmp = re.findall(r'group:\s+tacplus\s+files', nsswitch_conf)
+        self.assertTrue(tmp)
+
+        # PAM TACACS configuration
+        pam_tacacs_conf = read_file('/etc/tacplus_servers')
+        # NSS TACACS configuration
+        nss_tacacs_conf = read_file('/etc/tacplus_nss.conf')
+        # Users have individual home directories
+        self.assertIn('user_homedir=1', pam_tacacs_conf)
+
+        # specify services
+        self.assertIn('service=shell', pam_tacacs_conf)
+        self.assertIn('protocol=ssh', pam_tacacs_conf)
+
+        for server in tacacs_servers:
+            self.assertIn(f'secret={tacacs_secret}', pam_tacacs_conf)
+            self.assertIn(f'server={server}', pam_tacacs_conf)
+
+            self.assertIn(f'secret={tacacs_secret}', nss_tacacs_conf)
+            self.assertIn(f'server={server}', nss_tacacs_conf)
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)