diff --git a/README.md b/README.md index bcf69e6..3027eae 100644 --- a/README.md +++ b/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: diff --git a/VERSION b/VERSION index ccbccc3..276cbf9 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.2.0 +2.3.0 diff --git a/cvprac/__init__.py b/cvprac/__init__.py index 3e9608a..c2dde5b 100644 --- a/cvprac/__init__.py +++ b/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.' diff --git a/cvprac/cvp_api.py b/cvprac/cvp_api.py index 93768ee..65d8077 100644 --- a/cvprac/cvp_api.py +++ b/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. diff --git a/cvprac/cvp_client.py b/cvprac/cvp_client.py index ebe4d3a..71b1895 100644 --- a/cvprac/cvp_client.py +++ b/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, diff --git a/docs/release-notes-2.3.0.rst b/docs/release-notes-2.3.0.rst new file mode 100644 index 0000000..664f19f --- /dev/null +++ b/docs/release-notes-2.3.0.rst @@ -0,0 +1,17 @@ +###### +v2.3.0 +###### + +2026-9-10 + +Enhancements +^^^^^^^^^^^^ + +* Allow cert-based login. (`323 `_) [`vaibhavshaw-star `_] +* Add support for image upload into studios repository. (`331 `_) [`mharista `_] + + +Documentation +^^^^^^^^^^^^^ + +* Add documentation on system test setup. (`319 `_) [`chetryan `_] diff --git a/test/system/SYSTEM_TEST_CVP_SETUP.md b/test/system/SYSTEM_TEST_CVP_SETUP.md new file mode 100644 index 0000000..ab216ab --- /dev/null +++ b/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_` 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 +``` diff --git a/test/system/test_cvp_client_api.py b/test/system/test_cvp_client_api.py index 54454e0..61d00dc 100644 --- a/test/system/test_cvp_client_api.py +++ b/test/system/test_cvp_client_api.py @@ -35,10 +35,16 @@ ''' System test for the CvpApi class + Refer to test/system/SYSTEM_TEST_CVP_SETUP.md for more details. + + The file test/fixtures/cvp_nodes.yaml should be modified to + point to the CVP instance under test, with the associated username & password + and device name. + Requirements for CVP Node: 1) Test has dedicated access to the CVP node. - 2) Contains at least one device in a container. - 3) Container or device has at least one configlet applied. + 2) Contains at least one device in the "Tenant" container. + 3) Device has at least one configlet applied directly, without Reconcile configlets. 4) Device has a user account and password that matches the CVP username and password. If device does not have correct username and/or password then the tests that execute tasks will fail with the following error: @@ -48,6 +54,7 @@ and in the test log is the error: Failure response received from the netElement : ' Unauthorized User ' + 5) Device needs to have at least 1 interface called Ethernet1 ''' import os import shutil diff --git a/test/unit/test_client.py b/test/unit/test_client.py index 2268600..7cfa8e0 100644 --- a/test/unit/test_client.py +++ b/test/unit/test_client.py @@ -34,12 +34,15 @@ ''' Unit tests for the CvpClient class ''' import json +import ssl import unittest from itertools import cycle -from unittest.mock import Mock +from unittest.mock import Mock, patch +from requests.cookies import cookiejar_from_dict from requests.exceptions import HTTPError, ReadTimeout, JSONDecodeError -from cvprac.cvp_client import CvpClient -from cvprac.cvp_client_errors import CvpApiError, CvpSessionLogOutError +from cvprac.cvp_client import CvpClient, url_with_port +from cvprac.cvp_client_errors import (CvpApiError, CvpRequestError, + CvpSessionLogOutError) class TestClient(unittest.TestCase): @@ -200,6 +203,185 @@ def test_create_session_no_http_fallback(self): self.assertEqual(self.clnt.url_prefix, url) self.assertEqual(self.clnt.error_msg, error) + def test_url_with_port_ipv4_and_ipv6(self): + """Test replacing the port in a URL.""" + self.assertEqual(url_with_port('https://1.1.1.1:443', 9443), + 'https://1.1.1.1:9443') + self.assertEqual(url_with_port('https://[2001:db8::1]:443', 9443), + 'https://[2001:db8::1]:9443') + + def test_validate_certificate(self): + """Test certificate validation request uses mTLS endpoint.""" + self.clnt.session = Mock() + self.clnt.url_prefix_short = 'https://1.1.1.1:443' + self.clnt.cert = False + self.clnt.cert_login_port = 9443 + self.clnt.client_cert = ('client.crt', 'client.key') + self.clnt.connect_timeout = 5 + response = Mock() + response.ok = True + response.text = '{}' + response.headers = {'Location': '/?cert_valid=true'} + response.cookies = cookiejar_from_dict({'cert_state_token': 'token'}) + self.clnt.session.get.return_value = response + self.clnt._validate_client_certificate = Mock() + + self.clnt._validate_certificate() + + self.clnt._validate_client_certificate.assert_called_once_with() + self.clnt.session.get.assert_called_once_with( + 'https://1.1.1.1:9443/aaa/v1/validateCertificate', + headers=self.clnt.headers, timeout=5, verify=False, + cert=('client.crt', 'client.key'), allow_redirects=False) + self.assertEqual(self.clnt.cookies.get('cert_state_token'), 'token') + + def test_validate_certificate_location_error(self): + """Test validate certificate raises backend redirect errors.""" + self.clnt.session = Mock() + self.clnt.url_prefix_short = 'https://1.1.1.1:443' + self.clnt.cert = False + self.clnt.cert_login_port = 9443 + self.clnt.client_cert = ('client.crt', 'client.key') + self.clnt.connect_timeout = 5 + response = Mock() + response.ok = True + response.text = '{}' + response.headers = { + 'Location': '/cv?cert_error=%7B%22errorCode%22%3A%22212498%22' + '%2C%22errorMessage%22%3A%22Failed+to+verify+' + 'client+certificate.+Please+ensure+the+issuing+CA+' + 'is+present+in+the+trusted+certificate+store.%22%7D' + } + self.clnt.session.get.return_value = response + self.clnt._validate_client_certificate = Mock() + + with self.assertRaisesRegex(CvpApiError, + 'Failed to verify client certificate'): + self.clnt._validate_certificate() + + def test_validate_certificate_close_browser_error_trimmed(self): + """Test close browser guidance is trimmed from cert errors.""" + self.clnt.session = Mock() + self.clnt.url_prefix_short = 'https://1.1.1.1:443' + self.clnt.cert = False + self.clnt.cert_login_port = 9443 + self.clnt.client_cert = ('client.crt', 'client.key') + self.clnt.connect_timeout = 5 + response = Mock() + response.ok = True + response.text = '{}' + response.headers = { + 'Location': '/cv?cert_error=%7B%22errorCode%22%3A%22212498%22' + '%2C%22errorMessage%22%3A%22x+y+z.+Please+close+' + 'the+browser+and+try+again.%22%7D' + } + self.clnt.session.get.return_value = response + self.clnt._validate_client_certificate = Mock() + + with self.assertRaisesRegex(CvpApiError, + 'Validate certificate: .*x y z$'): + self.clnt._validate_certificate() + + def test_validate_client_certificate(self): + """Test local client certificate validation loads cert chain.""" + self.clnt.client_cert = ('client.crt', 'client.key') + context = Mock() + + with patch('cvprac.cvp_client.ssl.SSLContext', + return_value=context) as mock_context: + self.clnt._validate_client_certificate() + + mock_context.assert_called_once_with(ssl.PROTOCOL_TLS_CLIENT) + context.load_cert_chain.assert_called_once_with( + certfile='client.crt', keyfile='client.key') + + def test_validate_client_certificate_single_file(self): + """Test local client certificate validation supports one file.""" + self.clnt.client_cert = 'client.pem' + context = Mock() + + with patch('cvprac.cvp_client.ssl.SSLContext', + return_value=context): + self.clnt._validate_client_certificate() + + context.load_cert_chain.assert_called_once_with( + certfile='client.pem', keyfile=None) + + def test_validate_client_certificate_invalid_file(self): + """Test bad local client certificate raises a useful error.""" + self.clnt.client_cert = ('bad.crt', 'bad.key') + context = Mock() + context.load_cert_chain.side_effect = ssl.SSLError( + 'PEM lib') + + with patch('cvprac.cvp_client.ssl.SSLContext', + return_value=context): + with self.assertRaisesRegex(CvpRequestError, 'Invalid client certificate or key'): + self.clnt._validate_client_certificate() + + def test_validate_client_certificate_bad_tuple(self): + """Test malformed client_cert values fail before requests.""" + self.clnt.client_cert = ('client.crt', 'client.key', 'extra') + + with self.assertRaisesRegex(CvpRequestError, + 'Invalid client certificate/key'): + self.clnt._validate_client_certificate() + + def test_validate_certificate_bad_client_cert_skips_request(self): + """Test bad local client cert prevents validateCertificate call.""" + self.clnt.session = Mock() + self.clnt.url_prefix_short = 'https://1.1.1.1:443' + self.clnt.cert = False + self.clnt.cert_login_port = 9443 + self.clnt.client_cert = ('bad.crt', 'bad.key') + self.clnt._validate_client_certificate = Mock( + side_effect=CvpRequestError( + 'Invalid client certificate/key: PEM lib')) + + with self.assertRaisesRegex(CvpRequestError, + 'Invalid client certificate/key'): + self.clnt._validate_certificate() + self.clnt.session.get.assert_not_called() + + def test_login_on_prem_with_certificate(self): + """Test certificate based login validates cert before auth.""" + self.clnt.session = Mock() + self.clnt.url_prefix = 'https://1.1.1.1:443/web' + self.clnt.authdata = {'userId': 'certuser', 'password': 'password'} + self.clnt.connect_timeout = 10 + self.clnt.cert = False + self.clnt.cert_login = True + self.clnt.cert_login_port = 9443 + self.clnt.client_cert = ('client.crt', 'client.key') + self.clnt.cookies = cookiejar_from_dict({'cert_state_token': 'token'}) + self.clnt._validate_certificate = Mock() + response = Mock() + response.ok = True + response.text = '{"sessionId": "SESSION"}' + response.json.return_value = {'sessionId': 'SESSION'} + response.cookies = cookiejar_from_dict({'access_token': 'ACCESS'}) + self.clnt.session.post.return_value = response + + self.clnt._login_on_prem() + + self.clnt._validate_certificate.assert_called_once_with() + self.clnt.session.post.assert_called_once_with( + 'https://1.1.1.1:443/web/login/authenticate.do', + data=json.dumps({'userId': 'certuser', 'password': 'password'}), + headers=self.clnt.headers, cookies=self.clnt.cookies, timeout=10, + verify=False, cert=('client.crt', 'client.key')) + self.assertEqual(self.clnt.headers['APP_SESSION_ID'], 'SESSION') + self.assertEqual(self.clnt.cookies.get('access_token'), 'ACCESS') + + def test_login_on_prem_with_certificate_requires_client_cert(self): + """Test certificate login requires a client cert.""" + self.clnt.url_prefix = 'https://1.1.1.1:443/web' + self.clnt.cert_login = True + self.clnt.client_cert = None + + with self.assertRaises(CvpRequestError): + self.clnt._login_on_prem() + def test_make_request_good(self): """ Test request does not raise exception and returns json. """