feat(anta.tests): Added test case for SSL certificate status as Verif…

…yCertificateStatus (#472)

* issue-470: Added testcase for SSL certificate status as VerifyCertificateStatus

* issue-470: Updated the testcase with more optimized code

* issue-470: Added the support for optional encryption type and size

* issue-472: Updated unit tests

* issue-470: fix the linting issues

* issue-470: fix unit tests for tox

* issue-470: fix black issue

* issue-470: fix utils liniting

* issue-470: fix all linting issues

* issue-470: improve code and unit tests

* issue-470: Added few more unit tests

* issue-470: fix the indetation for tests.yaml

* issue-470: added the support of verify multiple certificate

* issue-470: updated code for rsa and ecdsa key size

* issue-470: fixed the pipeline

---------

Co-authored-by: arista <atd-help@arista.com>

anta/custom_types.py

@@ -63,3 +63,6 @@ def interface_case_sensitivity(v: str) -> str:
 ]
 Afi = Literal["ipv4", "ipv6", "vpn-ipv4", "vpn-ipv6", "evpn", "rt-membership"]
 Safi = Literal["unicast", "multicast", "labeled-unicast"]
+EncryptionAlgorithm = Literal["RSA", "ECDSA"]
+RsaKeySize = Literal[2048, 3072, 4096]
+EcdsaKeySize = Literal[256, 384, 521]

anta/tests/security.py

@@ -4,13 +4,19 @@
 """
 Test functions related to the EOS various security settings
 """
+from __future__ import annotations
+
 # Mypy does not understand AntaTest.Input typing
 # mypy: disable-error-code=attr-defined
-from __future__ import annotations
+from datetime import datetime
+from typing import List, Union
 
-from pydantic import conint
+from pydantic import BaseModel, conint, model_validator
 
+from anta.custom_types import EcdsaKeySize, EncryptionAlgorithm, RsaKeySize
 from anta.models import AntaCommand, AntaTest
+from anta.tools.get_value import get_value
+from anta.tools.utils import get_failed_logs
 
 
 class VerifySSHStatus(AntaTest):
@@ -268,3 +274,106 @@ def test(self) -> None:
             self.result.is_failure(f"eAPI IPv6 ACL(s) not configured or active in vrf {self.inputs.vrf}: {not_configured_acl_list}")
         else:
             self.result.is_success()
+
+
+class VerifyAPISSLCertificate(AntaTest):
+    """
+    Verifies the eAPI SSL certificate expiry, common subject name, encryption algorithm and key size.
+
+    Expected Results:
+        * success: The test will pass if the certificate's expiry date is greater than the threshold,
+                   and the certificate has the correct name, encryption algorithm, and key size.
+        * failure: The test will fail if the certificate is expired or is going to expire,
+                   or if the certificate has an incorrect name, encryption algorithm, or key size.
+    """
+
+    name = "VerifyAPISSLCertificate"
+    description = "Verifies the eAPI SSL certificate expiry, common subject name, encryption algorithm and key size."
+    categories = ["security"]
+    commands = [AntaCommand(command="show management security ssl certificate"), AntaCommand(command="show clock")]
+
+    class Input(AntaTest.Input):
+        """
+        Input parameters for the VerifyAPISSLCertificate test.
+        """
+
+        certificates: List[APISSLCertificates]
+        """List of API SSL certificates"""
+
+        class APISSLCertificates(BaseModel):
+            """
+            This class defines the details of an API SSL certificate.
+            """
+
+            certificate_name: str
+            """The name of the certificate to be verified."""
+            expiry_threshold: int
+            """The expiry threshold of the certificate in days."""
+            common_name: str
+            """The common subject name of the certificate."""
+            encryption_algorithm: EncryptionAlgorithm
+            """The encryption algorithm of the certificate."""
+            key_size: Union[RsaKeySize, EcdsaKeySize]
+            """The encryption algorithm key size of the certificate."""
+
+            @model_validator(mode="after")
+            def validate_inputs(self: BaseModel) -> BaseModel:
+                """
+                Validate the key size provided to the APISSLCertificates class.
+
+                If encryption_algorithm is RSA then key_size should be in {2048, 3072, 4096}.
+
+                If encryption_algorithm is ECDSA then key_size should be in {256, 384, 521}.
+                """
+
+                if self.encryption_algorithm == "RSA" and self.key_size not in RsaKeySize.__args__:
+                    raise ValueError(f"`{self.certificate_name}` key size {self.key_size} is invalid for RSA encryption. Allowed sizes are {RsaKeySize.__args__}.")
+
+                if self.encryption_algorithm == "ECDSA" and self.key_size not in EcdsaKeySize.__args__:
+                    raise ValueError(
+                        f"`{self.certificate_name}` key size {self.key_size} is invalid for ECDSA encryption. Allowed sizes are {EcdsaKeySize.__args__}."
+                    )
+
+                return self
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        # Mark the result as success by default
+        self.result.is_success()
+
+        # Extract certificate and clock output
+        certificate_output = self.instance_commands[0].json_output
+        clock_output = self.instance_commands[1].json_output
+        current_timestamp = clock_output["utcTime"]
+
+        # Iterate over each API SSL certificate
+        for certificate in self.inputs.certificates:
+            # Collecting certificate expiry time and current EOS time.
+            # These times are used to calculate the number of days until the certificate expires.
+            if not (certificate_data := get_value(certificate_output, f"certificates..{certificate.certificate_name}", separator="..")):
+                self.result.is_failure(f"SSL certificate '{certificate.certificate_name}', is not configured.\n")
+                continue
+
+            expiry_time = certificate_data["notAfter"]
+            day_difference = (datetime.fromtimestamp(expiry_time) - datetime.fromtimestamp(current_timestamp)).days
+
+            # Verify certificate expiry
+            if 0 < day_difference < certificate.expiry_threshold:
+                self.result.is_failure(f"SSL certificate `{certificate.certificate_name}` is about to expire in {day_difference} days.\n")
+            elif day_difference < 0:
+                self.result.is_failure(f"SSL certificate `{certificate.certificate_name}` is expired.\n")
+
+            # Verify certificate common subject name, encryption algorithm and key size
+            keys_to_verify = ["subject.commonName", "publicKey.encryptionAlgorithm", "publicKey.size"]
+            actual_certificate_details = {key: get_value(certificate_data, key) for key in keys_to_verify}
+
+            expected_certificate_details = {
+                "subject.commonName": certificate.common_name,
+                "publicKey.encryptionAlgorithm": certificate.encryption_algorithm,
+                "publicKey.size": certificate.key_size,
+            }
+
+            if actual_certificate_details != expected_certificate_details:
+                failed_log = f"SSL certificate `{certificate.certificate_name}` is not configured properly:"
+                failed_log += get_failed_logs(expected_certificate_details, actual_certificate_details)
+                self.result.is_failure(f"{failed_log}\n")

anta/tools/utils.py

@@ -0,0 +1,34 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""
+Toolkit for ANTA.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+
+def get_failed_logs(expected_output: dict[Any, Any], actual_output: dict[Any, Any]) -> str:
+    """
+    Get the failed log for a test.
+    Returns the failed log or an empty string if there is no difference between the expected and actual output.
+
+    Parameters:
+    expected_output (dict): Expected output of a test.
+    actual_output (dict): Actual output of a test
+
+    Returns:
+    str: Failed log of a test.
+    """
+    failed_logs = []
+
+    for element, expected_data in expected_output.items():
+        actual_data = actual_output.get(element)
+
+        if actual_data is None:
+            failed_logs.append(f"\nExpected `{expected_data}` as the {element}, but it was not found in the actual output.")
+        elif actual_data != expected_data:
+            failed_logs.append(f"\nExpected `{expected_data}` as the {element}, but found `{actual_data}` instead.")
+
+    return "".join(failed_logs)

examples/tests.yaml

@@ -193,6 +193,18 @@ anta.tests.security:
   - VerifyAPIIPv6Acl:
       number: 3
       vrf: default
+  - VerifyAPISSLCertificate:
+      certificates:
+        - certificate_name: ARISTA_SIGNING_CA.crt
+          expiry_threshold: 30
+          common_name: AristaIT-ICA ECDSA Issuing Cert Authority
+          encryption_algorithm: ECDSA
+          key_size: 256
+        - certificate_name: ARISTA_ROOT_CA.crt
+          expiry_threshold: 30
+          common_name: Arista Networks Internal IT Root Cert Authority
+          encryption_algorithm: RSA
+          key_size: 4096
 
 anta.tests.snmp:
   - VerifySnmpStatus: