Release 2.3.0

README.md

@@ -247,6 +247,17 @@ Same example as above using the API method:
     {u'version': u'2016.1.0'}
     >>>
 
+Example using CVP On Prem client cert login:
+
+    >>> from cvprac.cvp_client import CvpClient
+    >>> clnt = CvpClient()
+    >>> cert, key = "/cert/client.crt", "/cert/client.key" # cert_file_path, private_key_file_path
+    >>> clnt.connect(['cvp1', 'cvp2', 'cvp3'], 'cvp_user', 'cvp_word', cert_login=True, client_cert=(cert, key))
+    >>> result = clnt.get('/cvpInfo/getCvpInfo.do')
+    >>> print result
+    {u'version': u'2016.1.0'}
+    >>>
+
 Same example as above but connecting to CVaaS with a token: Note that
 the username and password parameters are required by the connect
 function but will be ignored when using api\_token:
@@ -287,28 +298,8 @@ requiring a manual refresh.
 
 ## Testing
 
-The cvprac module provides system tests. To run the system tests, you
-will need to update the `cvp_nodes.yaml` file found in test/fixtures.
-
-Requirements for running the system tests:
-
-- Need one CVP node for test with a test user account. Create the same
-  account on the switch used for testing. The user account information
-  follows:
-
-        username: CvpRacTest
-        password: AristaInnovates
-
-        If switch does not have correct username and/or password then the tests that
-        execute tasks will fail with the following error:
-
-        AssertionError: Execution for task id 220 failed and in the test log is the error:
-
-        Failure response received from the netElement : ' Unauthorized User '
-
-- Test has dedicated access to the CVP node.
-- CVP node contains at least one device in a container.
-- Container or device has at least one configlet applied.
+The cvprac module provides both unit tests and system tests. 
+To run the system tests, read the [detailed system test documentation](test/system/SYSTEM_TEST_CVP_SETUP.md) for more information on how to set up the environment.
 
 To run the system tests:
 

VERSION

@@ -1 +1 @@
-2.2.0
+2.3.0

cvprac/__init__.py

@@ -32,5 +32,5 @@
 ''' RESTful API Client class for Cloudvision(R) Portal
 '''
 
-__version__ = '2.2.0'
+__version__ = '2.3.0'
 __author__ = 'Arista Networks, Inc.'

cvprac/cvp_api.py

@@ -2312,6 +2312,34 @@ def add_image(self, filepath):
                                       files={'file': image_data})
         return response
 
+    def image_upload(self, name, filepath, rebootRequired=False):
+            ''' Add an image to a CVP cluster studios image repository.
+
+                Args:
+                    name (str): The name that will be used to identify the image
+                        or extension in CloudVision. Must have supported extension
+                        type such as swi or swix.
+                    filepath (str): Local path to the image to upload.
+                    rebootRequired (bool): Specifies if the image or extension requires
+                        a device reboot. Reboots are always required for .swi files.
+
+                Returns:
+                    data (dict): Dictionary of image add data.
+            '''
+            # Get the absolute file path to be uploaded
+            image_path = os.path.abspath(filepath)
+            data = {
+                'name': name,
+                'rebootRequired': rebootRequired,
+            }
+            with open(image_path, 'rb') as image_data:
+                response = self.clnt.post(
+                    '/cvpservice/softwaremanagement/v1/uploads',
+                    data=data,
+                    files={'file': image_data},
+                )
+            return response
+
     def cancel_image(self, image_name):
         ''' Discard/cancel the uploaded image/image bundle before save.
 

cvprac/cvp_client.py

@@ -96,10 +96,12 @@
 import os
 import re
 import json
+import ssl
 import logging
 from logging.handlers import SysLogHandler
 from itertools import cycle
 from packaging.version import parse
+from urllib.parse import parse_qs, urlparse, urlunparse
 
 import requests
 from requests.exceptions import ( # pylint: disable=redefined-builtin
@@ -116,6 +118,22 @@
     CvpRequestError, CvpSessionLogOutError
 
 
+def url_with_port(url, port):
+    '''Return url with netloc port replaced by port.
+
+    For example, url_with_port('https://cvp.example.com:443/web', 9443)
+    returns 'https://cvp.example.com:9443/web'.
+    '''
+    parsed = urlparse(url)
+    host = parsed.hostname
+    if ':' in host and not host.startswith('['):
+        host = f'[{host}]'
+    netloc = host
+    if port:
+        netloc = f'{host}:{port}'
+    return urlunparse(parsed._replace(netloc=netloc))
+
+
 class CvpClient():
     ''' Use this class to create a persistent connection to CVP.
     '''
@@ -142,6 +160,9 @@ def __init__(self, logger='cvprac', syslog=False, filename=None,
         self.apiversion = None
         self.authdata = None
         self.cert = False
+        self.cert_login = False
+        self.cert_login_port = None
+        self.client_cert = None
         self.connect_timeout = None
         self.cookies = None
         self.error_msg = ''
@@ -325,7 +346,8 @@ def set_version(self, version):
     def connect(self, nodes, username, password, connect_timeout=10,
                 request_timeout=30, protocol='https', port=None, cert=False,
                 is_cvaas=False, tenant=None, api_token=None, cvaas_token=None,
-                proxies=None):
+                proxies=None, cert_login=False, client_cert=None,
+                cert_login_port=9443):
         ''' Login to CVP and get a session ID and cookie.  Currently
             certificates are not verified if the https protocol is specified. A
             warning may be printed out from the requests module for this case.
@@ -366,6 +388,13 @@ def connect(self, nodes, username, password, connect_timeout=10,
                      Proxies can also be set via environment variables.
                      Please reference the below link for details of precedence.
                      https://requests.readthedocs.io/en/latest/user/advanced/#proxies
+                cert_login (boolean): Use CVP certificate based login flow.
+                    Supported on CVP 2026.3.0 and later. When True,
+                    client_cert must also be provided.
+                client_cert (str or tuple): Path to client certificate, or
+                    (cert, key) tuple as accepted by requests.
+                cert_login_port (int): CVP mTLS endpoint port. Default is
+                    9443.
 
             Raises:
                 CvpLoginError: A CvpLoginError is raised if a connection
@@ -388,6 +417,9 @@ def connect(self, nodes, username, password, connect_timeout=10,
                 nodes[idx] = os.environ.get('CURRENT_NODE_IP')
 
         self.cert = cert
+        self.cert_login = cert_login
+        self.cert_login_port = cert_login_port
+        self.client_cert = client_cert
         self.nodes = nodes
         self.node_cnt = len(nodes)
         self.node_pool = cycle(nodes)
@@ -538,6 +570,86 @@ def _check_response_status(self, response, prefix):
             self.log.error(msg)
             raise CvpRequestError(msg)
 
+    def _check_validate_certificate_result(self, response, prefix):
+        '''Check certificate validation details returned in redirect metadata.
+        '''
+        headers = getattr(response, 'headers', {}) or {}
+        location = headers.get('Location') if hasattr(headers, 'get') else ''
+        if not isinstance(location, str) or not location:
+            return
+        query = parse_qs(urlparse(location).query)
+        cert_valid = query.get('cert_valid', [None])[0]
+        err_msg = query.get('cert_error', [None])[0]
+        if err_msg:
+            try:
+                err_data = json.loads(err_msg)
+            except ValueError:
+                pass
+            else:
+                if isinstance(err_data, dict) and err_data.get('errorMessage'):
+                    err_msg = err_data['errorMessage']
+            if not isinstance(err_msg, str):
+                err_msg = str(err_msg)
+            # Backend returns a generic error message only for browser use case
+            # that is not helpful here. If the error message contains the string
+            # "close the browser and try again" then remove that part of the
+            # message to make it more useful.
+            # e.g "x y z. Please close the browser and try again. a b." becomes "x y z"
+            # e.g "x y z, then close the browser and try again." becomes "x y z"
+            close_browser_msg = 'close the browser and try again'
+            err_msg_lower = err_msg.lower()
+            close_browser_idx = err_msg_lower.find(close_browser_msg)
+            if close_browser_idx != -1:
+                end_idx = max(err_msg.rfind('.', 0, close_browser_idx),
+                              err_msg.rfind(',', 0, close_browser_idx))
+                if end_idx != -1:
+                    err_msg = err_msg[:end_idx]
+        err_msg = err_msg.strip() if err_msg else err_msg
+        if err_msg or (cert_valid and cert_valid.lower() != 'true'):
+            msg = f"{prefix}: Request Error: {err_msg or location}"
+            self.log.error(msg)
+            raise CvpApiError(msg)
+
+    def _validate_client_certificate(self):
+        '''Validate the local client certificate/key before calling CVP.
+        '''
+        cert_file = self.client_cert
+        key_file = None
+        if isinstance(self.client_cert, (tuple)):
+            if len(self.client_cert) != 2:
+                msg = ('Invalid client certificate/key: client_cert must be a '
+                       '(certificate, key) pair')
+                self.log.error(msg)
+                raise CvpRequestError(msg)
+            cert_file, key_file = self.client_cert
+
+        try:
+            context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+            context.load_cert_chain(certfile=cert_file, keyfile=key_file)
+        except (OSError, ssl.SSLError, TypeError, ValueError) as error:
+            msg = "Invalid client certificate or key"
+            self.log.error(msg)
+            raise CvpRequestError(msg) from error
+
+    def _validate_certificate(self):
+        '''Validate a client certificate for certificate based login.
+        '''
+        self._validate_client_certificate()
+        url = (url_with_port(self.url_prefix_short, self.cert_login_port) +
+               '/aaa/v1/validateCertificate')
+        response = self.session.get(url, headers=self.headers,
+                                    timeout=self.connect_timeout, verify=self.cert,
+                                    cert=self.client_cert,
+                                    allow_redirects=False)
+        prefix = f"Validate certificate: {url}"
+        self._is_good_response(response, prefix)
+        self._check_validate_certificate_result(response, prefix)
+
+        if self.cookies is None:
+            self.cookies = response.cookies
+        else:
+            self.cookies.update(response.cookies)
+
     def _login(self):
         ''' Make a POST request to CVP login authentication.
             An error can be raised from the post method call or the
@@ -602,14 +714,24 @@ def _login_on_prem(self):
                     CVP node.  Destroy the class and re-instantiate.
         '''
         url = self.url_prefix + '/login/authenticate.do'
+        if self.cert_login:
+            if self.client_cert is None:
+                raise CvpRequestError(
+                    'client_cert is required when cert_login is True')
+            self._validate_certificate()
         response = self.session.post(url,
                                      data=json.dumps(self.authdata),
                                      headers=self.headers,
+                                     cookies=self.cookies,
                                      timeout=self.connect_timeout,
-                                     verify=self.cert)
+                                     verify=self.cert,
+                                     cert=self.client_cert)
         self._is_good_response(response, f"Authenticate: {url}")
 
-        self.cookies = response.cookies
+        if self.cookies is None:
+            self.cookies = response.cookies
+        else:
+            self.cookies.update(response.cookies)
         self.headers['APP_SESSION_ID'] = response.json()['sessionId']
 
     def _set_headers_api_token(self):
@@ -859,12 +981,18 @@ def _send_request(self, req_type, full_url, timeout, data=None,
                         if 'Authorization' in self.headers:
                             fhs['Authorization'] = self.headers[
                                 'Authorization']
+                        # Data must be None or a dict so the key-value pairs can be extracted
+                        # as individual multipart/form-data fields right alongside the file
+                        # boundary parts. Hence when both data= and file= are present we do not
+                        # want to dump data as a JSON encoded string. Requests cannot merge a
+                        # raw string body with files and will raise a ValueError.
                         response = self.session.post(full_url,
                                                      cookies=self.cookies,
                                                      headers=fhs,
                                                      timeout=timeout,
                                                      verify=self.cert,
-                                                     files=files)
+                                                     files=files,
+                                                     data=data)
                 elif req_type == 'DELETE':
                     response = self.session.delete(full_url,
                                                    cookies=self.cookies,

docs/release-notes-2.3.0.rst

@@ -0,0 +1,17 @@
+######
+v2.3.0
+######
+
+2026-9-10
+
+Enhancements
+^^^^^^^^^^^^
+
+* Allow cert-based login. (`323 <https://github.com/aristanetworks/cvprac/pull/323>`_) [`vaibhavshaw-star <https://github.com/vaibhavshaw-star>`_]
+* Add support for image upload into studios repository. (`331 <https://github.com/aristanetworks/cvprac/pull/331>`_) [`mharista <https://github.com/mharista>`_]
+
+
+Documentation
+^^^^^^^^^^^^^
+
+* Add documentation on system test setup. (`319 <https://github.com/aristanetworks/cvprac/pull/319>`_) [`chetryan <https://github.com/chetryan>`_]

test/system/SYSTEM_TEST_CVP_SETUP.md

@@ -0,0 +1,53 @@
+# CVP System Test Setup
+
+Prepare a dedicated CloudVision/CVP instance before running `test/system`.
+The tests create, edit, execute, cancel, and delete CVP objects, so do not run
+them against a shared or production CVP.
+
+## CVP Requirements
+
+- Create a CVP user with `network-admin` permissions.
+- Create the same username and password on the test switch. Task execution
+  tests fail with `Unauthorized User` if the switch credentials do not match
+  the CVP login. **_Note: on Arista Cloud Test, these two steps are already done during lab deployment._**
+- Add at least one test device to CVP inventory.
+- Move the test device into the `Tenant` container, not `Undefined`.
+- Apply at least one normal configlet directly to the test device. Use a
+  configlet assigned only to that device when possible, because tests edit
+  configlet contents to create tasks.
+- Make sure there is no existing `RECONCILE_<device-name>` configlet for the
+  test device.
+- Make sure the test device has an `Ethernet1` interface. Tag resource tests
+  assign and remove tags on `Ethernet1`.
+- If you want image tests to exercise apply/remove image bundle paths, create
+  at least one image bundle. Otherwise those paths are skipped.
+
+## Fixture
+
+Update `test/fixtures/cvp_nodes.yaml` before running system tests.
+- Change the username and password to the user with `network-admin` permissions that
+was created during CVP configuration.
+- Change `device` to the device hostname that was added to the **Tenant** Container.
+
+```yaml
+- node: cvp-hostname-or-ip
+  username: CvpRacTest
+  password: AristaInnovates
+  device: test-device-hostname
+  # api_token: optional-valid-token
+  # api_token_expired: optional-expired-token
+  # is_cvaas: false
+  # connect_timeout: 10
+  # request_timeout: 30
+```
+
+Multiple entries may be listed, but most system tests use the first entry.
+Token tests are skipped unless `api_token` or `api_token_expired` is provided.
+
+## Run
+
+Run the full suite of tests.
+
+```bash
+make tests
+```