Merge pull request #157 from rohit04saluja/rohit04saluja/skip_verify_gnmi

Skip verification of certificate

Author: ThomasJRyan

connector/docs/changelog/undistributed/skip_verify_gnmi.rst

@@ -0,0 +1,8 @@
+--------------------------------------------------------------------------------
+                                New
+--------------------------------------------------------------------------------
+* yang
+    * Modified Gnmi:
+        * Added `skip_verify` property for `gnmi` connection
+        * if `skip_verify` is set to `true` pyats will establishes a secure connection, using the server credentials.
+        * Verification of certificate is skipped with this option.

connector/src/yang/connector/gnmi.py

@@ -6,6 +6,9 @@
 from google.protobuf import json_format
 import grpc
 from . import proto
+import ssl
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
 
 
 try:
@@ -228,6 +231,7 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.device = kwargs.get('device')
         self.dev_args = self.connection_info
+        self.skip_verify = kwargs.get('skip_verify')
         if self.dev_args.get('protocol', '') != 'gnmi':
             msg = 'Invalid protocol {0}'.format(self.dev_args.get('protocol', ''))
             raise TypeError(msg)
@@ -266,6 +270,7 @@ def connect(self):
         dev_args = self.dev_args
         username = dev_args.get('username', '')
         password = dev_args.get('password', '')
+        skip_verify = dev_args.get('skip_verify')
 
         if dev_args.get('custom_log', ''):
             self.log = dev_args.get('custom_log')
@@ -331,6 +336,28 @@ def connect(self):
         if private_key and os.path.isfile(private_key):
             private_key = open(private_key, 'rb').read()
 
+        if skip_verify and not root:
+            try:
+                ssl_cert = ssl.get_server_certificate((str(host), port)).encode("utf-8")
+            except Exception as e:
+                self.log.error(f'The SSH certificate cannot be retrieved from {target}')
+                raise gNMIException(f'The SSH certificate cannot be retrieved from {target}', e)
+
+            ssl_cert_deserialized = x509.load_pem_x509_certificate(
+                ssl_cert, default_backend()
+            )
+
+            try:
+                ssl_cert_common_name = ssl_cert_deserialized.subject.get_attributes_for_oid(
+                        (x509.oid.NameOID.COMMON_NAME)
+                    )[0].value
+                options.append(
+                    ('grpc.ssl_target_name_override', ssl_cert_common_name),
+                )
+                root = ssl_cert
+            except BaseException as err:
+                self.log.warning(f'Unable to get common name: {err}')
+
         if any((root, chain, private_key)):
             override_name = dev_args.get('ssl_name_override', '')
             if override_name:
@@ -359,7 +386,6 @@ def connect(self):
                 target, channel_creds, options
             )
         else:
-            self.channel = grpc.insecure_channel(target)
             self.channel = grpc.insecure_channel(target, options)
             self.metadata = [
                 ("username", username),

connector/src/yang/connector/tests/test_gnmi.py

@@ -8,6 +8,8 @@
 from pyats.topology import loader
 from pyats.datastructures import AttrDict
 
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
 
 class TestGnmi(unittest.TestCase):
 
@@ -133,6 +135,105 @@ def test_connect_proxy(self):
         ],
     }
 
+    @patch('yang.connector.gnmi.grpc.secure_channel')
+    @patch('yang.connector.gnmi.grpc.composite_channel_credentials')
+    @patch('yang.connector.gnmi.grpc.metadata_call_credentials')
+    @patch('yang.connector.gnmi.grpc.ssl_channel_credentials')
+    @patch('yang.connector.gnmi.x509.load_pem_x509_certificate')
+    @patch('yang.connector.gnmi.ssl.get_server_certificate')
+    def test_connect_with_skip_verify(
+        self, mock_get_server_certificate, mock_load_pem_x509_certificate,
+        mock_ssl_channel_credentials, mock_metadata_call_credentials,
+        mock_composite_channel_credentials, mock_secure_channel, *_
+    ):
+        yaml = """
+devices:
+    dummy:
+        type: dummy_device
+        connections:
+            Gnmi:
+                class:  yang.connector.Gnmi
+                protocol: gnmi
+                ip : 1.2.3.4
+                port: 830
+                username: admin
+                password: admin
+                skip_verify: true
+"""
+        testbed = loader.load(yaml)
+        device = testbed.devices['dummy']
+        device.connect(alias='gnmi', via='Gnmi')
+        mock_get_server_certificate.assert_called_once_with(('1.2.3.4', '830'))
+        mock_load_pem_x509_certificate.assert_called_once_with(
+            mock_get_server_certificate.return_value.encode('utf-8'),
+            default_backend()
+        )
+        mock_load_pem_x509_certificate.return_value.subject.get_attributes_for_oid.assert_called_once_with(
+            (x509.NameOID.COMMON_NAME)
+        )
+        mock_ssl_channel_credentials.assert_called_once_with(
+            mock_get_server_certificate.return_value.encode('utf-8'), None, None
+        )
+        self.assertEqual(
+            mock_secure_channel.call_args[0][2][-1],
+            (
+                'grpc.ssl_target_name_override',
+                mock_load_pem_x509_certificate.return_value.subject.get_attributes_for_oid.return_value[0].value
+            )
+        )
+
+    @patch('yang.connector.gnmi.ssl.get_server_certificate', side_effect=Exception)
+    def test_connect_with_skip_verify_get_server_certificate_exception(self, *_):
+        yaml = """
+devices:
+    dummy:
+        type: dummy_device
+        connections:
+            Gnmi:
+                class:  yang.connector.Gnmi
+                protocol: gnmi
+                ip: 1.2.3.4
+                port: 830
+                username: admin
+                password: admin
+                skip_verify: true
+"""
+        testbed = loader.load(yaml)
+        device = testbed.devices['dummy']
+        with self.assertRaises(Exception):
+            device.connect(alias='gnmi', via='Gnmi')
+
+    @patch('yang.connector.gnmi.ssl.get_server_certificate')
+    @patch('yang.connector.gnmi.grpc.insecure_channel')
+    @patch('yang.connector.gnmi.x509.load_pem_x509_certificate')
+    def test_connect_with_skip_verify_get_attributes_for_oid_exception(
+        self, mock_load_pem_x509_certificate, mock_insecure_channel, *_
+    ):
+        yaml = """
+devices:
+    dummy:
+        type: dummy_device
+        connections:
+            Gnmi:
+                class:  yang.connector.Gnmi
+                protocol: gnmi
+                ip: 1.2.3.4
+                port: 830
+                username: admin
+                password: admin
+                skip_verify: true
+"""
+        mock_load_pem_x509_certificate.return_value.subject.get_attributes_for_oid.side_effect = BaseException
+        testbed = loader.load(yaml)
+        device = testbed.devices['dummy']
+        device.connect(alias='gnmi', via='Gnmi')
+        mock_insecure_channel.assert_called_once_with(
+            '1.2.3.4:830', [
+                ('grpc.max_receive_message_length', 1000000000),
+                ('grpc.max_send_message_length', 1000000000)
+            ]
+        )
+
     def test_xpath_to_path_elem(self):
         """Test converting Genie content data to cisco_gnmi format."""
         modules, message, origin = xpath_util.xml_path_to_path_elem(self.request)