[TACACS+] Add TACACS per-command accounting test case.

ansible/group_vars/lab/lab.yml

@@ -31,6 +31,10 @@ tacacs_ro_user_passwd: '123456'
 tacacs_jit_user: test_jituser
 tacacs_jit_user_passwd: '123456'
 tacacs_jit_user_membership: netuser
+tacacs_authorization_user: test_auuser
+tacacs_authorization_user_passwd: '123456'
+local_user: test_louser
+local_user_passwd: '123456'
 
 # tacacs grous
 tacacs_group: 'testlab'

tests/common/utilities.py

@@ -25,7 +25,6 @@
 logger = logging.getLogger(__name__)
 cache = FactsCache()
 
-
 def skip_release(duthost, release_list):
     """
     @summary: Skip current test if any given release keywords are in os_version, match sonic_release.
@@ -39,6 +38,22 @@ def skip_release(duthost, release_list):
     if any(release == duthost.sonic_release for release in release_list):
         pytest.skip("DUT is release {} and test does not support {}".format(duthost.sonic_release, ", ".join(release_list)))
 
+def check_skip_release(duthost, release_list):
+    """
+    @summary: check if need skip current test if any given release keywords are in os_version, match sonic_release.
+    @param duthost: The DUT
+    @param release_list: A list of incompatible releases
+    """
+    if any(release in duthost.os_version for release in release_list):
+        logger.info("DUT has version {} and test does not support {}".format(duthost.os_version, ", ".join(release_list)))
+        return True
+
+    if any(release == duthost.sonic_release for release in release_list):
+        logger.info("DUT is release {} and test does not support {}".format(duthost.sonic_release, ", ".join(release_list)))
+        return True
+
+    return False
+
 def skip_release_for_platform(duthost, release_list, platform_list):
     """
     @summary: Skip current test if any given release keywords are in os_version and any given platform keywords are in platform

tests/kvmtest.sh

@@ -129,6 +129,8 @@ test_t0() {
     tacacs/test_ro_user.py \
     tacacs/test_ro_disk.py \
     tacacs/test_jit_user.py \
+    tacacs/test_authorization.py \
+    tacacs/test_accounting.py \
     telemetry/test_telemetry.py \
     test_features.py \
     test_procdockerstatsd.py \

tests/tacacs/tac_plus.conf.j2

@@ -10,30 +10,56 @@ accounting file = /var/log/tac_plus.acct
 key = {{ tacacs_passkey }}
 
 group = netadmin {
-	default service = permit
-	service = exec {
-		priv-lvl = 15
-	}
+    default service = permit
+    service = exec {
+        priv-lvl = 15
+    }
 }
 
 user = {{ tacacs_rw_user }} {
-	login = des {{ tacacs_rw_user_passwd }}
-	member = netadmin
+    login = des {{ tacacs_rw_user_passwd }}
+    member = netadmin
 }
 
 group = netuser {
-	default service = permit
-	service = exec {
-		priv-lvl = 1
-	}
+    default service = permit
+    service = exec {
+        priv-lvl = 1
+    }
 }
 
 user = {{ tacacs_ro_user }} {
-	login = des {{ tacacs_ro_user_passwd }}
-	member = netuser
+    login = des {{ tacacs_ro_user_passwd }}
+    member = netuser
 }
 
 user = {{ tacacs_jit_user }} {
-	login = des {{ tacacs_jit_user_passwd }}
-	member = {{ tacacs_jit_user_membership }}
+    login = des {{ tacacs_jit_user_passwd }}
+    member = {{ tacacs_jit_user_membership }}
+}
+
+user = {{ tacacs_authorization_user }} {
+    login = des {{ tacacs_authorization_user_passwd }}
+    member = netuser
+    cmd = /usr/bin/cat {
+        deny .*
+    }
+    # disable following command for UT test_bypass_authorization
+    cmd = /usr/local/bin/config {
+        deny   tacacs
+        permit .*
+    }
+    cmd = /usr/bin/python {
+        deny .*
+    }
+    cmd = /usr/bin/find {
+        deny   -exec
+        permit .*
+    }
+    cmd = /lib/x86_64-linux-gnu/ld-linux-x86-64.so.2 {
+        deny .*
+    }
+    cmd = /usr/bin/dash {
+        deny .*
+    }
 }

tests/tacacs/test_accounting.py

@@ -0,0 +1,220 @@
+import crypt
+import paramiko
+import pytest
+
+from .test_authorization import ssh_connect_remote, ssh_run_command, per_command_check_skip_versions, remove_all_tacacs_server
+from .utils import stop_tacacs_server, start_tacacs_server
+from tests.common.errors import RunAnsibleModuleFail
+from tests.common.helpers.assertions import pytest_assert
+from tests.common.utilities import skip_release
+
+pytestmark = [
+    pytest.mark.disable_loganalyzer,
+    pytest.mark.topology('any'),
+    pytest.mark.device_type('vs')
+]
+
+logger = logging.getLogger(__name__)
+
+def cleanup_tacacs_log(ptfhost, rw_user_client):
+    try:
+        ptfhost.command('rm /var/log/tac_plus.acct')
+    except RunAnsibleModuleFail:
+        logger.info("/var/log/tac_plus.acct does not exist.")
+
+    res = ptfhost.command('touch /var/log/tac_plus.acct')
+    logger.info(res["stdout_lines"])
+
+    ssh_run_command(rw_user_client, 'sudo truncate -s 0 /var/log/syslog')
+
+def check_tacacs_server_log_exist(ptfhost, duthost, creds_all_duts, command):
+    username = creds_all_duts[duthost]['tacacs_rw_user']
+    """
+        Find logs run by tacacs_rw_user from tac_plus.acct:
+            Find logs match following format: "tacacs_rw_user ... cmd=command"
+            Print matched logs with /P command.
+    """
+    sed_command = "sed -nE '/	{0}	.*	cmd=.*{1}/P' /var/log/tac_plus.acct".format(username, command)
+    res = ptfhost.command(sed_command)
+    logger.info(sed_command)
+    logger.info(res["stdout_lines"])
+    pytest_assert(len(res["stdout_lines"]) > 0)
+
+def check_tacacs_server_no_other_user_log(ptfhost, duthost, creds_all_duts):
+    username = creds_all_duts[duthost]['tacacs_rw_user']
+    """
+        Find logs not run by tacacs_rw_user from tac_plus.acct:
+            Remove all tacacs_rw_user's log with /D command.
+            Print logs not removed by /D command, which are not run by tacacs_rw_user.
+    """
+    sed_command = "sed -nE '/	{0}	/D;/.*/P' /var/log/tac_plus.acct".format(username)
+    res = ptfhost.command(sed_command)
+    logger.info(sed_command)
+    logger.info(res["stdout_lines"])
+    pytest_assert(len(res["stdout_lines"]) == 0)
+
+def check_local_log_exist(rw_user_client, duthost, creds_all_duts, command):
+    username = creds_all_duts[duthost]['tacacs_rw_user']
+    """
+        Find logs run by tacacs_rw_user from syslog:
+            Find logs match following format: "INFO audisp-tacplus: Accounting: user: tacacs_rw_user,.*, command: .*command,"
+            Print matched logs with /P command.
+    """
+    sed_command = "sudo sed -nE '/INFO audisp-tacplus: Accounting: user: {0},.*, command: .*{1},/P' /var/log/syslog".format(username, command)
+    exit_code, stdout, stderr = ssh_run_command(rw_user_client, sed_command)
+    pytest_assert(exit_code == 0)
+    logger.info(sed_command)
+    logger.info(stdout)
+    pytest_assert(len(stdout) > 0)
+
+def check_local_no_other_user_log(rw_user_client, duthost, creds_all_duts):
+    username = creds_all_duts[duthost]['tacacs_rw_user']
+    """
+        Find logs not run by tacacs_rw_user from syslog:
+            Remove all tacacs_rw_user's log with /D command, which will match following format: "INFO audisp-tacplus: Accounting: user: tacacs_rw_user"
+            Find all other user's log, which will match following format: "INFO audisp-tacplus: Accounting: user:" 
+            Print matched logs with /P command, which are not run by tacacs_rw_user.
+    """
+    sed_command = "sudo sed -nE '/INFO audisp-tacplus: Accounting: user: {0},/D;/INFO audisp-tacplus: Accounting: user:/P' /var/log/syslog".format(username)
+    exit_code, stdout, stderr = ssh_run_command(rw_user_client, sed_command)
+    pytest_assert(exit_code == 0)
+    logger.info(sed_command)
+    logger.info(stdout)
+    pytest_assert(len(stdout) == 0)
+
+@pytest.fixture
+def rw_user_client(duthosts, enum_rand_one_per_hwsku_hostname, creds_all_duts):
+    duthost = duthosts[enum_rand_one_per_hwsku_hostname]
+    dutip = duthost.host.options['inventory_manager'].get_host(duthost.hostname).vars['ansible_host']
+    ssh_client = ssh_connect_remote(dutip, creds_all_duts[duthost]['tacacs_rw_user'],
+                         creds_all_duts[duthost]['tacacs_rw_user_passwd'])
+    yield ssh_client
+    ssh_client.close()
+
+@pytest.fixture(scope="module", autouse=True)
+def check_image_version(duthost):
+    """Skips this test if the SONiC image installed on DUT is older than 202112
+    Args:
+        duthost: Hostname of DUT.
+    Returns:
+        None.
+    """
+    skip_release(duthost, per_command_check_skip_versions)
+
+def test_accounting_tacacs_only(localhost, ptfhost, duthosts, enum_rand_one_per_hwsku_hostname, creds_all_duts, check_tacacs, rw_user_client):
+    duthost = duthosts[enum_rand_one_per_hwsku_hostname]
+    duthost.shell("sudo config aaa accounting tacacs+")
+    cleanup_tacacs_log(ptfhost, rw_user_client)
+
+    ssh_run_command(rw_user_client, "grep")
+
+    # Verify TACACS+ server side have user command record.
+    check_tacacs_server_log_exist(ptfhost, duthost, creds_all_duts, "grep")
+    # Verify TACACS+ server side not have any command record which not run by user.
+    check_tacacs_server_no_other_user_log(ptfhost, duthost, creds_all_duts)
+
+
+def test_accounting_tacacs_only_all_tacacs_server_down(localhost, ptfhost, duthosts, enum_rand_one_per_hwsku_hostname, creds_all_duts, check_tacacs, rw_user_client):
+    duthost = duthosts[enum_rand_one_per_hwsku_hostname]
+    duthost.shell("sudo config aaa accounting tacacs+")
+    cleanup_tacacs_log(ptfhost, rw_user_client)
+
+    """
+        when user login server are accessible.
+        user run some command in whitelist and server are accessible.
+    """
+    ssh_run_command(rw_user_client, "grep")
+
+    # Verify TACACS+ server side have user command record.
+    check_tacacs_server_log_exist(ptfhost, duthost, creds_all_duts, "grep")
+    # Verify TACACS+ server side not have any command record which not run by user.
+    check_tacacs_server_no_other_user_log(ptfhost, duthost, creds_all_duts)
+
+    cleanup_tacacs_log(ptfhost, rw_user_client)
+
+    # Shutdown tacacs server
+    stop_tacacs_server(ptfhost)
+
+    """
+        then all server not accessible, and run some command
+        Verify local user still can run command without any issue.
+    """
+    ssh_run_command(rw_user_client, "grep")
+
+    #  Cleanup UT.
+    start_tacacs_server(ptfhost)
+
+def test_accounting_tacacs_only_some_tacacs_server_down(localhost, ptfhost, duthosts, enum_rand_one_per_hwsku_hostname, creds_all_duts, check_tacacs, rw_user_client):
+    """
+        Setup multiple tacacs server for this UT.
+        Tacacs server 127.0.0.1 not accessible.
+    """
+    invalid_tacacs_server_ip = "127.0.0.1"
+    duthost = duthosts[enum_rand_one_per_hwsku_hostname]
+    tacacs_server_ip = ptfhost.host.options['inventory_manager'].get_host(ptfhost.hostname).vars['ansible_host']
+    config_facts = duthost.config_facts(host=duthost.hostname, source="running")['ansible_facts']
+    duthost.shell("sudo config tacacs timeout 1")
+    remove_all_tacacs_server(duthost)
+    duthost.shell("sudo config tacacs add %s" % invalid_tacacs_server_ip)
+    duthost.shell("sudo config tacacs add %s" % tacacs_server_ip)
+    duthost.shell("sudo config aaa accounting tacacs+")
+
+    cleanup_tacacs_log(ptfhost, rw_user_client)
+
+    ssh_run_command(rw_user_client, "grep")
+
+    # Verify TACACS+ server side have user command record.
+    check_tacacs_server_log_exist(ptfhost, duthost, creds_all_duts, "grep")
+    # Verify TACACS+ server side not have any command record which not run by user.
+    check_tacacs_server_no_other_user_log(ptfhost, duthost, creds_all_duts)
+
+    # Cleanup
+    duthost.shell("sudo config tacacs delete %s" % invalid_tacacs_server_ip)
+
+def test_accounting_local_only(localhost, ptfhost, duthosts, enum_rand_one_per_hwsku_hostname, creds_all_duts, check_tacacs, rw_user_client):
+    duthost = duthosts[enum_rand_one_per_hwsku_hostname]
+    duthost.shell("sudo config aaa accounting local")
+    cleanup_tacacs_log(ptfhost, rw_user_client)
+
+    ssh_run_command(rw_user_client, "grep")
+
+    # Verify syslog have user command record.
+    check_local_log_exist(rw_user_client, duthost, creds_all_duts, "grep")
+    # Verify syslog not have any command record which not run by user.
+    check_local_no_other_user_log(rw_user_client, duthost, creds_all_duts)
+
+def test_accounting_tacacs_and_local(localhost, ptfhost, duthosts, enum_rand_one_per_hwsku_hostname, creds_all_duts, check_tacacs, rw_user_client):
+    duthost = duthosts[enum_rand_one_per_hwsku_hostname]
+    duthost.shell('sudo config aaa accounting "tacacs+ local"')
+    cleanup_tacacs_log(ptfhost, rw_user_client)
+
+    ssh_run_command(rw_user_client, "grep")
+
+    # Verify TACACS+ server and syslog have user command record.
+    check_tacacs_server_log_exist(ptfhost, duthost, creds_all_duts, "grep")
+    check_local_log_exist(rw_user_client, duthost, creds_all_duts, "grep")
+    # Verify TACACS+ server and syslog not have any command record which not run by user.
+    check_tacacs_server_no_other_user_log(ptfhost, duthost, creds_all_duts)
+    check_local_no_other_user_log(rw_user_client, duthost, creds_all_duts)
+
+def test_accounting_tacacs_and_local_all_tacacs_server_down(localhost, ptfhost, duthosts, enum_rand_one_per_hwsku_hostname, creds_all_duts, check_tacacs, rw_user_client):
+    duthost = duthosts[enum_rand_one_per_hwsku_hostname]
+    duthost.shell('sudo config aaa accounting "tacacs+ local"')
+    cleanup_tacacs_log(ptfhost, rw_user_client)
+
+    # Shutdown tacacs server
+    stop_tacacs_server(ptfhost)
+
+    """
+        After all server not accessible, run some command
+        Verify local user still can run command without any issue.
+    """
+    ssh_run_command(rw_user_client, "grep")
+
+    # Verify syslog have user command record.
+    check_local_log_exist(rw_user_client, duthost, creds_all_duts, "grep")
+    # Verify syslog not have any command record which not run by user.
+    check_local_no_other_user_log(rw_user_client, duthost, creds_all_duts)
+
+    #  Cleanup UT.
+    start_tacacs_server(ptfhost)