Add new test to verify gnmi role authentication (#18050)

tests/common/plugins/conditional_mark/tests_mark_conditions.yaml

@@ -1126,6 +1126,10 @@ gnmi/test_gnmi_configdb.py::test_gnmi_configdb_full_01:
     conditions:
       - "https://github.com/sonic-net/sonic-mgmt/issues/17436"
 
+gnmi/test_gnoi_killprocess.py::test_gnoi_killprocess_restart:
+  skip:
+    reason: "Test noisy due to restart issue not relevant to GNOI. Disabling them to rewrite."
+
 gnmi/test_gnoi_killprocess.py::test_gnoi_killprocess_then_restart:
   skip:
     reason: "Test noisy due to restart issue not relevant to GNOI. Disabling them to rewrite."

tests/gnmi/helper.py

@@ -65,9 +65,9 @@ def verify_tcp_port(localhost, ip, port):
     logger.info("TCP: " + res['stdout'] + res['stderr'])
 
 
-def add_gnmi_client_common_name(duthost, cname):
-    duthost.shell('sudo sonic-db-cli CONFIG_DB hset "GNMI_CLIENT_CERT|{}" "role" "role1"'.format(cname),
-                  module_ignore_errors=True)
+def add_gnmi_client_common_name(duthost, cname, role="gnmi_readwrite"):
+    command = 'sudo sonic-db-cli CONFIG_DB hset "GNMI_CLIENT_CERT|{}" "role@" "{}"'.format(cname, role)
+    duthost.shell(command, module_ignore_errors=True)
 
 
 def del_gnmi_client_common_name(duthost, cname):
@@ -100,8 +100,9 @@ def apply_cert_config(duthost):
     duthost.shell(dut_command)
 
     # Setup gnmi client cert common name
-    add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic")
-    add_gnmi_client_common_name(duthost, "test.client.revoked.gnmi.sonic")
+    role = "gnmi_readwrite,gnmi_config_db_readwrite,gnmi_appl_db_readwrite,gnmi_dpu_appl_db_readwrite,gnoi_readwrite"
+    add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+    add_gnmi_client_common_name(duthost, "test.client.revoked.gnmi.sonic", role)
 
     time.sleep(GNMI_SERVER_START_WAIT_TIME)
     dut_command = "sudo netstat -nap | grep %d" % env.gnmi_port

tests/gnmi/test_gnmi.py

@@ -3,8 +3,10 @@
 
 from .helper import gnmi_capabilities, gnmi_set, add_gnmi_client_common_name, del_gnmi_client_common_name, dump_gnmi_log
 from tests.common.utilities import wait_until
+from tests.common.plugins.allure_wrapper import allure_step_wrapper as allure
 
 logger = logging.getLogger(__name__)
+allure.logger = logger
 
 pytestmark = [
     pytest.mark.topology('any'),
@@ -23,6 +25,47 @@ def test_gnmi_capabilities(duthosts, rand_one_dut_hostname, localhost):
     assert "JSON_IETF" in msg, msg
 
 
+def test_gnmi_capabilities_authenticate(duthosts, rand_one_dut_hostname, localhost):
+    '''
+    Verify GNMI capabilities with different roles
+    '''
+    duthost = duthosts[rand_one_dut_hostname]
+
+    with allure.step("Verify GNMI capabilities with noaccess role"):
+        role = "gnmi_noaccess"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        ret, msg = gnmi_capabilities(duthost, localhost)
+        assert ret != 0, msg
+        assert role in msg, msg
+
+    with allure.step("Verify GNMI capabilities with readonly role"):
+        role = "gnmi_readonly"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        ret, msg = gnmi_capabilities(duthost, localhost)
+        assert ret == 0, msg
+        assert "sonic-db" in msg, msg
+        assert "JSON_IETF" in msg, msg
+
+    with allure.step("Verify GNMI capabilities with readwrite role"):
+        role = "gnmi_readwrite"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        ret, msg = gnmi_capabilities(duthost, localhost)
+        assert ret == 0, msg
+        assert "sonic-db" in msg, msg
+        assert "JSON_IETF" in msg, msg
+
+    with allure.step("Verify GNMI capabilities with empty role"):
+        role = ""
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        ret, msg = gnmi_capabilities(duthost, localhost)
+        assert ret == 0, msg
+        assert "sonic-db" in msg, msg
+        assert "JSON_IETF" in msg, msg
+
+    # Restore default role
+    add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic")
+
+
 @pytest.fixture(scope="function")
 def setup_invalid_client_cert_cname(duthosts, rand_one_dut_hostname):
     duthost = duthosts[rand_one_dut_hostname]

tests/gnmi/test_gnmi_configdb.py

@@ -8,12 +8,15 @@
 from .helper import gnmi_set, gnmi_get, gnoi_reboot
 from .helper import gnmi_subscribe_polling
 from .helper import gnmi_subscribe_streaming_sample, gnmi_subscribe_streaming_onchange
+from .helper import add_gnmi_client_common_name
 from tests.common.helpers.assertions import pytest_assert
 from tests.common.utilities import wait_until
 from tests.common.platform.processes_utils import wait_critical_processes
 from tests.common.platform.interface_utils import check_interface_status_of_up_ports
+from tests.common.plugins.allure_wrapper import allure_step_wrapper as allure
 
 logger = logging.getLogger(__name__)
+allure.logger = logger
 
 pytestmark = [
     pytest.mark.topology('any'),
@@ -304,3 +307,142 @@ def test_gnmi_configdb_full_01(duthosts, rand_one_dut_hostname, ptfhost):
     duthost.shell("config interface startup %s" % interface)
     # Wait for BGP neighbor to be up
     wait_bgp_neighbor(duthost)
+
+
+def test_gnmi_configdb_set_authenticate(duthosts, rand_one_dut_hostname, ptfhost):
+    '''
+    Verify GNMI native write with authentication
+    '''
+    duthost = duthosts[rand_one_dut_hostname]
+    file_name = "cloud.txt"
+    text = "\"Public\""
+    with open(file_name, 'w') as file:
+        file.write(text)
+    ptfhost.copy(src=file_name, dest='/root')
+    update_list = ["/sonic-db:CONFIG_DB/localhost/DEVICE_METADATA/localhost/cloudtype:@/root/%s" % (file_name)]
+
+    with allure.step("Verify GNMI set with noaccess role"):
+        role = "gnmi_config_db_noaccess"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        try:
+            gnmi_set(duthost, ptfhost, [], update_list, [])
+        except Exception as e:
+            logger.info("Failed to set: " + str(e))
+            assert role in str(e), str(e)
+
+    with allure.step("Verify GNMI set with readwrite role"):
+        role = "gnmi_config_db_readwrite"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        try:
+            gnmi_set(duthost, ptfhost, [], update_list, [])
+        except Exception as e:
+            logger.info("Failed to set: " + str(e))
+            pytest.fail("Set request failed: " + str(e))
+
+    with allure.step("Verify GNMI set with readonly role"):
+        role = "gnmi_config_db_readonly"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        try:
+            gnmi_set(duthost, ptfhost, [], update_list, [])
+        except Exception as e:
+            logger.info("Failed to set: " + str(e))
+            assert role in str(e), str(e)
+
+    with allure.step("Verify GNMI set with empty role"):
+        role = ""
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        try:
+            gnmi_set(duthost, ptfhost, [], update_list, [])
+        except Exception as e:
+            logger.info("Failed to set: " + str(e))
+            assert "write access" in str(e), str(e)
+
+    # Restore default role
+    add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic")
+
+
+def test_gnmi_configdb_get_authenticate(duthosts, rand_one_dut_hostname, ptfhost):
+    '''
+    Verify GNMI native read with authentication
+    '''
+    duthost = duthosts[rand_one_dut_hostname]
+    path_list = ["/sonic-db:CONFIG_DB/localhost/DEVICE_METADATA/localhost"]
+
+    with allure.step("Verify GNMI get with noaccess role"):
+        role = "gnmi_config_db_noaccess"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        try:
+            gnmi_get(duthost, ptfhost, path_list)
+        except Exception as e:
+            logger.info("Failed to get: " + str(e))
+            assert role in str(e), str(e)
+
+    with allure.step("Verify GNMI get with readwrite role"):
+        role = "gnmi_config_db_readwrite"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        try:
+            gnmi_get(duthost, ptfhost, path_list)
+        except Exception as e:
+            logger.info("Failed to get: " + str(e))
+            pytest.fail("Get request failed: " + str(e))
+
+    with allure.step("Verify GNMI get with readonly role"):
+        role = "gnmi_config_db_readonly"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        try:
+            gnmi_get(duthost, ptfhost, path_list)
+        except Exception as e:
+            logger.info("Failed to get: " + str(e))
+            pytest.fail("Get request failed: " + str(e))
+
+    with allure.step("Verify GNMI get with empty role"):
+        role = ""
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        try:
+            gnmi_get(duthost, ptfhost, path_list)
+        except Exception as e:
+            logger.info("Failed to get: " + str(e))
+            pytest.fail("Get request failed: " + str(e))
+
+    # Restore default role
+    add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic")
+
+
+def test_gnmi_configdb_subscribe_authenticate(duthosts, rand_one_dut_hostname, ptfhost):
+    '''
+    Verify GNMI native read with authentication
+    '''
+    duthost = duthosts[rand_one_dut_hostname]
+    path_list = ["/sonic-db:CONFIG_DB/localhost/DEVICE_METADATA/localhost"]
+
+    with allure.step("Verify GNMI subscribe with noaccess role"):
+        role = "gnmi_config_db_noaccess"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        output, _ = gnmi_subscribe_streaming_sample(duthost, ptfhost, path_list, 0, 1)
+        logger.info("GNMI subscribe output: " + output)
+        assert "GRPC error" in output, output
+        assert role in output, output
+
+    with allure.step("Verify GNMI subscribe with readwrite role"):
+        role = "gnmi_config_db_readwrite"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        output, _ = gnmi_subscribe_streaming_sample(duthost, ptfhost, path_list, 0, 1)
+        assert "GRPC error" not in output, output
+        assert "cloudtype" in output, output
+
+    with allure.step("Verify GNMI subscribe with readonly role"):
+        role = "gnmi_config_db_readonly"
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        output, _ = gnmi_subscribe_streaming_sample(duthost, ptfhost, path_list, 0, 1)
+        assert "GRPC error" not in output, output
+        assert "cloudtype" in output, output
+
+    with allure.step("Verify GNMI subscribe with empty role"):
+        role = ""
+        add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic", role)
+        output, _ = gnmi_subscribe_streaming_sample(duthost, ptfhost, path_list, 0, 1)
+        assert "GRPC error" not in output, output
+        assert "cloudtype" in output, output
+
+    # Restore default role
+    add_gnmi_client_common_name(duthost, "test.client.gnmi.sonic")