# Copyright (c) 2025 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the COPYING file.
import urllib3
import pathlib
import ssl
from grpclib import client, events
from base64 import b64encode
[docs]
class UnableToAuthenticateException(Exception):
"""
Is thrown when unable to authenticate using username and password
"""
pass
[docs]
class AsyncCVClient:
"""
This class enables access to CloudVision GRPC API using grpclib stubs.
Use one of classmethods :py:func:`from_token` :py:func:`from_user_credentials`
to intantialize the client, and use it with stubs from
:py:mod:`cloudvision.api.arista`
It is implemented as contextmanager, that provides instance of grpclib.client.Channel_.
See example below
.. _grpclib.client.Channel: https://grpclib.readthedocs.io/en/latest/client.html
.. versionadded:: 1.26.1
.. code-block:: python
import asyncio
from cloudvision.api.client import AsyncCVClient
from cloudvision.api.arista.inventory.v1 import DeviceServiceStub, DeviceStreamRequest
async def get_devices():
client = AsyncCVClient.from_token('<your service account token>', 'your-cvp.io')
# get channel
with client as channel:
# pass it to the stub
service = DeviceServiceStub(channel)
# execute one of stub's methods
async for item in service.get_all(DeviceStreamRequest()):
print(item)
asyncio.run(get_devices())
.. note::
If for some reason multiple context managers with the same client are
used, each will produce a new channel that would be closed accordingly (see below)
.. code-block:: python
client = AsyncCVCClient(...)
with client as channel1:
# channel1 is open
with client as channel2:
... # both channel1 and channel2 are open
# channel2 is closed while channel 1 is still open
# channel1 is closed
"""
def __init__(self, token, ssl_context, host, port=443, username=None):
self.token = token
self.ssl_context = ssl_context
self._channel = None
self.host = host
self.port = port
self.username = username
self._channel_stack = []
@classmethod
def _get_ssl_context(cls, host, port=443, cacert=None, insecure=False):
if insecure:
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
elif cacert:
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)
else:
# Use the system CA bundle for publicly-signed certificates.
# Private and self-signed CAs must be supplied explicitly.
context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
context.set_alpn_protocols(["h2"])
return context
[docs]
@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/settings/aaa-service-accounts
Generate a token for service account and pass it to this method to get an instance of
the client.
.. note::
By default, certificates are verified using the system CA bundle.
For deployments using a private or self-signed CA, provide its
certificate using ``cacert``.
.. DANGER::
Avoid setting `insecure=True` as this would disable certificate check
.. versionchanged:: 1.27.2
Added `insecure` and `cacert` parameters
: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
"""
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)
[docs]
@classmethod
def from_user_credentials(cls, username, password, host, port=443,
insecure=False, cacert=None):
"""
Use usename and password to authenticate in CloudVision
.. note::
By default, certificates are verified using the system CA bundle.
For deployments using a private or self-signed CA, provide its
certificate using ``cacert``.
.. DANGER::
Avoid setting `insecure=True` as this would disable certificate check
.. versionchanged:: 1.27.2
Added `insecure` and `cacert` parameters
: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:
"""
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):
"""
Initializes an authenticated Channel for sending GRPC requests. When using this method,
ensure that Channel is duly closed (`channel.close()`) after it's being used. Using context
manager `with client as channel: ...` is more recommended for this reason.
:rtype: grpclib.client.Channel
"""
async def _auth_middleware(event: events.SendRequest):
if self.username:
event.metadata['username'] = str(self.username)
event.metadata['authorization'] = f'Bearer {self.token}'
channel = client.Channel(
host=self.host,
port=self.port,
ssl=self.ssl_context
)
events.listen(channel, events.SendRequest, _auth_middleware)
return channel
def __enter__(self):
"""
Returns an authenticated GRPC channel, that would be closed upon exitting
the context manager.
:rtype: grpclib.client.Channel
"""
channel = self._init_channel()
self._channel_stack.append(channel)
return channel
def __exit__(self, *args):
channel = self._channel_stack.pop()
channel.close()