cloudvision/api: add cacert param to AsyncCVClient initialisation

Previously async client would only work with self-signed certificates or with the certificates that are CA-signed with the system's trusted CA. It got the host certificate with `ssl.get_server_certificate(host, port)` and used it for creating the connection, which could potentially be exploited if DNS entry for the host had been tampered. By adding `cacert` parameters, we at provide an option to verify if the cert received from the host is signed by the provided CA.

This change includes:
- Add `cacert` and `insecure` params to `from_token` and `from_user_credentials`
- Pass serviceCACert to AsyncCVCient when using ctx.getAsyncClient in actions
- Add couple of test cases that verify various uses of the new parameters

Change-Id: I9ea1162893913818e1a2d870f52ddaf136412647

Author: rstatkevych-arista

README.md

@@ -94,7 +94,7 @@ import asyncio
 from cloudvision.api.client import AsyncCVClient
 from cloudvision.api.arista.inventory.v1 import DeviceServiceStub, DeviceStreamRequest
 
-async def get_auditlog():
+async def get_devices():
     client = AsyncCVClient.from_token('<your service account token>', 'your-cvp.io')
 
     # get channel
@@ -107,7 +107,7 @@ async def get_auditlog():
         async for item in service.get_all(DeviceStreamRequest()):
             print(item)
 
-asyncio.run(get_auditlog())
+asyncio.run(get_devices())
 ```
 
 ## CloudVision Connector

cloudvision/api/client.py

@@ -2,10 +2,18 @@
 # Use of this source code is governed by the Apache License 2.0
 # that can be found in the COPYING file.
 
-import requests
+import urllib3
+import pathlib
 import ssl
 from grpclib import client, events
-import tempfile
+from base64 import b64encode
+
+
+class UnableToAuthenticateException(Exception):
+    """
+    Is thrown when unable to authenticate using username and password
+    """
+    pass
 
 
 class AsyncCVClient:
@@ -69,49 +77,115 @@ def __init__(self, token, ssl_context, host, port=443, username=None):
         self._channel_stack = []
 
     @classmethod
-    def from_token(cls, token, host, port=443, username=None):
+    def _get_ssl_context(cls, host, port=443, cacert=None, insecure=False):
+        if not insecure:
+            if not cacert:
+                # This would be the case for on prem deployments as they will have self-signed
+                # certificates. This won't save from bogus cert, but at least it verify that
+                # the cert didn't expired and the hostname is right
+                cadata = ssl.get_server_certificate((host, port))
+            else:
+                cacert = pathlib.Path(cacert)
+                with cacert.open("r") as f:
+                    cadata = f.read()
+
+            context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH,
+                                                 cadata=cadata)
+            context.set_alpn_protocols(["h2"])
+        else:
+            context = ssl.create_default_context()
+            context.check_hostname = False
+            context.verify_mode = ssl.CERT_NONE
+        return context
+
+    @classmethod
+    def from_token(cls, token, host, port=443, username=None, insecure=False, cacert=None):
         """
         If you would like to use service accounts, you can create them in CloudVision UI
         https://my-cloudvision-instance.io/cv/setting/aaa-service-accounts
 
         Generate a token for service account and pass it to this method to get an instance of
         the client.
 
-        :rtype: AsyncCVClient
-        """
-        cadata = ssl.get_server_certificate((host, port))
+        .. note::
+            With default parameters, it would assume that the host has a self-signed certificate and
+            it will fetch it and verify hostname and expiry. It is recommended that you use
+            CA signed certificate in your CloudVision deployment, so you'd either add this CA to
+            the list of trusted CAs, or provide it's certificate via `cacert` parameter
+
+        .. DANGER::
+            Avoid setting `insecure=True` as this would disable certificate check
+
+        .. versionchanged:: 1.27.2
+            Added `insecure` and `cacert` parameters
+
+        :param host: CloudVision hostname
 
-        context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH,
-                                             cadata=cadata)
-        context.set_alpn_protocols(["h2"])
+        :param insecure:
+            skip certificate verification
 
-        return cls(token=token, ssl_context=context, port=port, host=host, username=username)
+        :param cacert:
+            path to CA certificate to use for verifying the host's certificate
+        :type cacert: :class:`pathlib.Path` or str
+
+        :rtype: AsyncCVClient
+        """
+        ssl_context = cls._get_ssl_context(host, port, cacert=cacert, insecure=insecure)
+        return cls(token=token, ssl_context=ssl_context, port=port, host=host, username=username)
 
     @classmethod
-    def from_user_credentials(cls, username, password, host, port=443):
+    def from_user_credentials(cls, username, password, host, port=443,
+                              insecure=False, cacert=None):
         """
         Use usename and password to authenticate in CloudVision
 
-        :rtype: AsyncCVClient
-        """
-        cadata = ssl.get_server_certificate((host, port))
+        .. note::
+            With default parameters, it would assume that the host has a self-signed certificate and
+            it will fetch it and verify hostname and expiry. It is recommended that you use
+            CA signed certificate in your CloudVision deployment, so you'd either add this CA to
+            the list of trusted CAs, or provide it's certificate via `cacert` parameter
 
-        with tempfile.NamedTemporaryFile("a+") as fw:
-            fw.write(cadata)
-            fw.flush()
 
-            r = requests.post(
-                'https://' + host + '/cvpservice/login/authenticate.do',
-                auth=(username, password), verify=fw.name)
+        .. DANGER::
+            Avoid setting `insecure=True` as this would disable certificate check
 
-            r.raise_for_status()
-            token = r.json()['sessionId']
+        .. versionchanged:: 1.27.2
+            Added `insecure` and `cacert` parameters
 
-        context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH,
-                                             cadata=cadata)
-        context.set_alpn_protocols(["h2"])
+        :param host: CloudVision hostname
+
+        :param insecure:
+            skip certificate verification
+
+        :param cacert:
+            path to CA certificate to use for verifying the host's certificate
+        :type cacert: :class:`pathlib.Path` or str
+
+        :rtype: AsyncCVClient
+        :raises UnableToAuthenticateException:
+        """
 
-        return cls(token=token, ssl_context=context, port=port, host=host, username=username)
+        headers = {
+            "Authorization": f"Basic {b64encode(":".join((username, password)).encode()).decode()}",
+        }
+        try:
+            context = cls._get_ssl_context(host, port, cacert=cacert, insecure=insecure)
+            with urllib3.PoolManager(ssl_context=context) as pool:
+                resp = pool.request('POST',
+                                    f'https://{host}:{port}/cvpservice/login/authenticate.do',
+                                    headers=headers)
+                if resp.status != 200:
+                    raise Exception(f"Status code {resp.status}: {resp.read()}")
+                data = resp.json()
+        except Exception as e:
+            raise UnableToAuthenticateException(
+                f"Unable to authenticate using user and password. Cause: {e}") from e
+
+        # Apparently we can't reuse ssl context, so creating a new one
+        context = cls._get_ssl_context(host, port, cacert=cacert, insecure=insecure)
+
+        return cls(token=data['sessionId'], ssl_context=context, port=port,
+                   host=host, username=username)
 
     def _init_channel(self):
         """

cloudvision/cvlib/context.py

@@ -253,7 +253,7 @@ def getAsyncApiClient(self, stub):
         if not self.__asyncServiceChann:
             self.__asyncServiceChann = AsyncCVClient.from_token(
                 token=self.user.token, username=self.user.username,
-                host=host, port=port
+                host=host, port=port, cacert=self.connections.serviceCACert
             )._init_channel()
 
         return stub(self.__asyncServiceChann)

test/api/test_async_client.py

@@ -3,64 +3,161 @@
 # that can be found in the COPYING file.
 
 import ssl
-import asyncio
 import functools
+import tempfile
 
+import urllib3
 import pytest
-from grpclib import utils, server
-import pytest_asyncio
-from cloudvision.api.arista.inventory import v1 as inventory
-from cloudvision.api.client import AsyncCVClient
+from cloudvision.api.client import AsyncCVClient, UnableToAuthenticateException
 from pathlib import Path
 
-TEST_TOKEN = 'test'
-THIS_DIR = Path(__file__).parent
-TEST_DIR = THIS_DIR.parent
-TEST_DATA_DIR = Path.joinpath(TEST_DIR, "test_data")
+from . import utils
 
+pytestmark = [pytest.mark.filterwarnings(
+    "ignore:Unverified HTTPS request is being made to host 'localhost'")]
 
-class MockInventoryService(inventory.DeviceServiceBase):
 
-    async def _call_rpc_handler_server_stream(self, handler, stream, request):
-        assert stream.metadata['authorization'] == f'Bearer {TEST_TOKEN}'
-        return await super()._call_rpc_handler_server_stream(handler, stream, request)
+@pytest.fixture
+def tmp_dir_factory():
+    with tempfile.TemporaryDirectory() as td:
+        counter = 1
+        td = Path(td)
 
-    async def get_all(self, device_stream_request):
-        for i in range(3):
-            yield inventory.DeviceStreamResponse(
-                value=inventory.Device(
-                    key=inventory.DeviceKey(device_id=f'device-{i}')
-                )
-            )
+        def factory():
+            nonlocal counter
+            d = td / str(counter)
+            d.mkdir()
+            counter += 1
+            return d
 
+        yield factory
 
-@pytest_asyncio.fixture
-async def grpc_server(unused_tcp_port_factory):
-    invService = MockInventoryService()
-    srv = server.Server([invService])
 
-    context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
-    context.load_cert_chain(certfile=Path.joinpath(TEST_DATA_DIR, "cert.pem"),
-                            keyfile=Path.joinpath(TEST_DATA_DIR, "key.pem"))
-    with utils.graceful_exit([server]):
-        async with srv:
-            port = unused_tcp_port_factory()
-            await srv.start('localhost', port, ssl=context)
-            yield 'localhost', port
+@pytest.mark.asyncio
+async def test_self_signed(tmp_dir_factory, unused_tcp_port_factory):
+    certs = utils.create_self_signed_cert(tmp_dir_factory())
+    async with utils.grpc_server(unused_tcp_port_factory(), certs) as (host, port):
+        f = functools.partial(AsyncCVClient.from_token, utils.TEST_TOKEN, host=host,
+                              port=port)
+        await utils.assert_grpc_response(f)
+
+
+@pytest.mark.asyncio
+async def test_self_signed_insecure(tmp_dir_factory, unused_tcp_port_factory):
+    certs = utils.create_self_signed_cert(tmp_dir_factory())
+
+    async with utils.grpc_server(unused_tcp_port_factory(), certs) as (host, port):
+        f = functools.partial(AsyncCVClient.from_token, utils.TEST_TOKEN, host=host,
+                              port=port, insecure=True)
+        await utils.assert_grpc_response(f)
+
+
+@pytest.mark.asyncio
+async def test_self_signed_insecure_wrong_host(tmp_dir_factory, unused_tcp_port_factory):
+    certs = utils.create_self_signed_cert(tmp_dir_factory(), hostname='example.org')
+    async with utils.grpc_server(unused_tcp_port_factory(), certs) as (host, port):
+        f = functools.partial(AsyncCVClient.from_token, utils.TEST_TOKEN, host=host,
+                              port=port, insecure=True)
+        await utils.assert_grpc_response(f)
+
+
+@pytest.mark.asyncio
+async def test_self_signed_wrong_host(tmp_dir_factory, unused_tcp_port_factory):
+    certs = utils.create_self_signed_cert(tmp_dir_factory(), hostname='example.org')
+    async with utils.grpc_server(unused_tcp_port_factory(), certs) as (host, port):
+        f = functools.partial(AsyncCVClient.from_token, utils.TEST_TOKEN, host=host,
+                              port=port)
+        with pytest.raises(ssl.SSLCertVerificationError):
+            await utils.assert_grpc_response(f)
+
+
+@pytest.mark.asyncio
+async def test_ca_cert_provided(tmp_dir_factory, unused_tcp_port_factory):
+    certs = utils.create_ca_signed_certs(tmp_dir_factory())
+    async with utils.grpc_server(unused_tcp_port_factory(), certs) as (host, port):
+        f = functools.partial(AsyncCVClient.from_token, utils.TEST_TOKEN, host=host,
+                              port=port, cacert=certs.cacert)
+
+        await utils.assert_grpc_response(f)
+
+
+@pytest.mark.asyncio
+async def test_bogus_ca_cert(tmp_dir_factory, unused_tcp_port_factory):
+    realCerts = utils.create_ca_signed_certs(tmp_dir_factory())
+    bogusCerts = utils.create_ca_signed_certs(tmp_dir_factory())
+
+    async with utils.grpc_server(unused_tcp_port_factory(), bogusCerts) as (host, port):
+        f = functools.partial(AsyncCVClient.from_token, utils.TEST_TOKEN, host=host,
+                              port=port, cacert=realCerts.cacert)
+        with pytest.raises(ssl.SSLCertVerificationError):
+            await utils.assert_grpc_response(f)
 
 
 @pytest.mark.asyncio
-async def test_token_auth(grpc_server):
-    host, port = grpc_server
-    callable = functools.partial(AsyncCVClient.from_token, TEST_TOKEN, host=host, port=port)
-    # Need to run this in executor, otherwise it would block the event loop forever
-    client = await asyncio.get_running_loop().run_in_executor(None, callable)
-    with client as channel:
-        stub = inventory.DeviceServiceStub(channel)
-        result = []
-        async for device in stub.get_all(inventory.DeviceStreamRequest(), timeout=10):
-            result.append(device)
-
-        assert len(result) == 3
-        assert set([dev.value.key.device_id for dev in result]) == \
-            {'device-0', 'device-1', 'device-2'}
+async def test_insecure(tmp_dir_factory, unused_tcp_port_factory):
+    bogusCerts = utils.create_ca_signed_certs(tmp_dir_factory())
+
+    async with utils.grpc_server(unused_tcp_port_factory(), bogusCerts) as (host, port):
+        f = functools.partial(AsyncCVClient.from_token, utils.TEST_TOKEN, host=host,
+                              port=port, insecure=True)
+        await utils.assert_grpc_response(f)
+
+
+@pytest.mark.asyncio
+async def test_user_password_sefl_signed(tmp_dir_factory, unused_tcp_port_factory):
+    certs = utils.create_self_signed_cert(tmp_dir_factory())
+    port = unused_tcp_port_factory()
+    async with utils.http_server(port=port, certs=certs):
+        client = AsyncCVClient.from_user_credentials(username=utils.USERNAME,
+                                                     password=utils.PASSWORD, host='localhost',
+                                                     port=port)
+        assert client.token == utils.TEST_TOKEN
+
+
+@pytest.mark.asyncio
+async def test_user_password_with_ca(tmp_dir_factory, unused_tcp_port_factory):
+    certs = utils.create_ca_signed_certs(tmp_dir_factory())
+    port = unused_tcp_port_factory()
+    async with utils.http_server(port=port, certs=certs):
+        client = AsyncCVClient.from_user_credentials(username=utils.USERNAME,
+                                                     password=utils.PASSWORD, host='localhost',
+                                                     port=port, cacert=certs.cacert)
+        assert client.token == utils.TEST_TOKEN
+
+
+@pytest.mark.asyncio
+async def test_user_password_wrong_ca_cert(tmp_dir_factory, unused_tcp_port_factory):
+    realCert = utils.create_ca_signed_certs(tmp_dir_factory())
+    bogusCert = utils.create_ca_signed_certs(tmp_dir_factory())
+
+    port = unused_tcp_port_factory()
+    async with utils.http_server(port=port, certs=bogusCert):
+        with pytest.raises(UnableToAuthenticateException):
+            AsyncCVClient.from_user_credentials(username=utils.USERNAME,
+                                                password=utils.PASSWORD, host='localhost',
+                                                port=port, cacert=realCert.cacert)
+
+
+@pytest.mark.asyncio
+async def test_user_password_wrong_ca_cert_insecure(tmp_dir_factory, unused_tcp_port_factory):
+    bogusCert = utils.create_ca_signed_certs(tmp_dir_factory())
+
+    port = unused_tcp_port_factory()
+    async with utils.http_server(port=port, certs=bogusCert):
+        client = AsyncCVClient.from_user_credentials(username=utils.USERNAME,
+                                                     password=utils.PASSWORD, host='localhost',
+                                                     port=port,
+                                                     insecure=True)
+        assert client.token == utils.TEST_TOKEN
+
+
+@pytest.mark.asyncio
+async def test_user_password_wrong_password(tmp_dir_factory, unused_tcp_port_factory):
+    certs = utils.create_ca_signed_certs(tmp_dir_factory())
+
+    port = unused_tcp_port_factory()
+    async with utils.http_server(port=port, certs=certs):
+        with pytest.raises(UnableToAuthenticateException):
+            AsyncCVClient.from_user_credentials(username=utils.USERNAME,
+                                                password='wrong', host='localhost',
+                                                port=port, cacert=certs.cacert)