Refine CVE check in check script for k8s version policy

Standards/scs-0210-v2-k8s-version-policy.md

@@ -55,14 +55,31 @@ window period.
 In order to keep up-to-date with the latest Kubernetes features, bug fixes and security improvements,
 the provided Kubernetes versions should be kept up-to-date with new upstream releases:
 
-- The latest minor version MUST be provided no later than 4 months after release.
-- The latest patch version MUST be provided no later than 2 weeks after release.
-- This time period MUST be even shorter for patches that fix critical CVEs.
-  In this context, a critical CVE is a CVE with a CVSS base score >= 8 according
-  to the CVSS version used in the original CVE record (e.g., CVSSv3.1).
-  It is RECOMMENDED to provide a new patch version in a 2-day time period after their release.
-- New versions MUST be tested before being rolled out on productive infrastructure;
-  at least the [CNCF E2E tests][cncf-conformance] should be passed beforehand.
+1. Minor Versions:
+   - The latest minor version MUST be provided no later than 4 months after release.
+
+2. Patch Versions:
+   - The latest patch version MUST be provided no later than 1 week after release.
+   - This time period MUST be even shorter for patches that fix critical CVEs.
+     In this context, a critical CVE is a CVE with a CVSS base score >= 8 according
+     to the CVSS version used in the original CVE record (e.g., CVSSv3.1).
+     It is RECOMMENDED to provide a new patch version in a 2-day time period after their release.
+   - New versions MUST be tested before being rolled out on productive infrastructure;
+     at least the [CNCF E2E tests][cncf-conformance] should be passed beforehand.
+
+3. CI Integration
+   * Trivy
+     - Providers should integrate Trivy into their CI pipeline to automatically scan Kubernetes cluster components,
+    including kubelet, apiserver, and others.
+     - The CI job MUST fail if critical vulnerabilities (CVSS >= 8) are detected in the cluster components.
+     - JSON reports from Trivy scans should be reviewed, and Trivy's experimental status should be monitored for changes
+     in output formats.
+   * nvdlib (Fallback):
+     - If Trivy fails or cannot meet requierements, nvdlib MUST be used as a fallback to query CVE data for Kubernetes
+     versions, laveraging CPE-based searches to track vunerabilities for specific versions. 
+     - Providers using nvdlib MUST periodically query for critical cunerabilities affecting the Kubernetes version in production.
+
+4. TBD
 
 At the same time, providers must support Kubernetes versions at least as long as the
 official sources as described in [Kubernetes Support Period][k8s-support-period]:

Tests/kaas/k8s-version-policy/k8s_version_policy.py

@@ -24,6 +24,7 @@
 (c) Hannes Baum <[email protected]>, 6/2023
 (c) Martin Morgenstern <[email protected]>, 2/2024
 (c) Matthias Büchse <[email protected]>, 3/2024
+(c) Piotr Bigos <[email protected]>, 10/2024
 SPDX-License-Identifier: CC-BY-SA-4.0
 """
 
@@ -188,7 +189,6 @@ class K8sBranch:
 
     def previous(self):
         if self.minor == 0:
-            # FIXME: this is ugly
             return self
         return K8sBranch(self.major, self.minor - 1)
 
@@ -257,19 +257,26 @@ class VersionRange:
     upper_version: K8sVersion = None
     inclusive: bool = False
 
-    def __post_init__(self):
-        if self.lower_version is None:
-            raise ValueError("lower_version must not be None")
-        if self.upper_version and self.upper_version < self.lower_version:
-            raise ValueError("lower_version must be lower than upper_version")
-
     def __contains__(self, version: K8sVersion) -> bool:
         if self.upper_version is None:
             return self.lower_version == version
         if self.inclusive:
             return self.lower_version <= version <= self.upper_version
         return self.lower_version <= version < self.upper_version
 
+    # def __post_init__(self):
+    #     if self.lower_version is None:
+    #         raise ValueError("lower_version must not be None")
+    #     if self.upper_version and self.upper_version < self.lower_version:
+    #         raise ValueError("lower_version must be lower than upper_version")
+    #
+    # def __contains__(self, version: K8sVersion) -> bool:
+    #     if self.upper_version is None:
+    #         return self.lower_version == version
+    #     if self.inclusive:
+    #         return self.lower_version <= version <= self.upper_version
+    #     return self.lower_version <= version < self.upper_version
+
 
 @dataclass
 class ClusterInfo:
@@ -337,19 +344,32 @@ async def collect_cve_versions(session: aiohttp.ClientSession) -> set:
 
     # CVE fix versions
     cfvs = set()
+    cve_patch_data = dict()
+
+    # # Request latest version
+    # async with session.get(
+    #     "https://kubernetes.io/docs/reference/issues-security/official-cve-feed/index.json",
+    #     headers={"Accept": "application/json"}
+    # ) as resp:
+    #     cve_list = await resp.json()
 
-    # Request latest version
     async with session.get(
-        "https://kubernetes.io/docs/reference/issues-security/official-cve-feed/index.json",
-        headers={"Accept": "application/json"}
+            "https://kubernetes.io/docs/reference/issues-security/official-cve-feed/index.json",
+            headers={"Accept": "application/json"}
     ) as resp:
+        if resp.status != 200:
+            logger.error(f"Failed to fetch CVE data, status code: {resp.status}")
+            return cve_patch_data
+
         cve_list = await resp.json()
 
     tasks = [request_cve_data(session=session, cveid=cve['id'])
-             for cve in cve_list['items']]
+             for cve in cve_list.get('items', [])]
 
     cve_data_list = await asyncio.gather(*tasks, return_exceptions=True)
 
+    cve_data_list = [data for data in cve_data_list if not isinstance(data, Exception)]
+
     for cve_data in cve_data_list:
         try:
             cve_cna = cve_data['containers']['cna']
@@ -381,6 +401,44 @@ async def collect_cve_versions(session: aiohttp.ClientSession) -> set:
     return cfvs
 
 
+def is_critical_cve(cve_metrics: list) -> bool:
+    """Checks if the CVE is considered critical based on CVSS score."""
+    for metric in cve_metrics:
+        if metric.get('cvssV3', {}).get('baseScore', 0) >= CVE_SEVERITY:
+            return True
+        return False
+
+
+def parse_cve_version_information_new(version_info: dict) -> str:
+    """Extracts the affected Kubernetes version from CVE data."""
+    return version_info.get('version')
+
+
+def parse_patch_release_date(version_info: dict) -> datetime:
+    """Extracts the release date of the patch from the CVE version info."""
+    patch_release_str = version_info.get('patchReleaseDate', None)
+    if patch_release_str:
+        return datetime.strptime(patch_release_str, "%Y-%m-%d")
+    return None
+
+
+async def check_patch_deployment(session: aiohttp.ClientSession, current_version: str, deployed_date: datetime) -> None:
+    """Check if the latest patch targeting a critical CVE was deployed within the allowed time."""
+    cve_patch_data = await collect_cve_versions(session)
+
+    for cve_id, version_data in cve_patch_data.items():
+        for version, patch_release_date in version_data:
+            if version == current_version:
+                if patch_release_date:
+                    allowed_timeframe = patch_release_date + CVE_VERSION_CADENCE
+                    if deployed_date > allowed_timeframe:
+                        logger.error(f"Patch for {cve_id} affecting version {version} was not deployed in time!")
+                    else:
+                        logger.info(f"Patch for {cve_id} affecting version {version} deployed in time.")
+                else:
+                    logger.warning(f"Patch release date for {cve_id} affecting version {version} is missing.")
+
+
 async def get_k8s_cluster_info(kubeconfig, context=None) -> ClusterInfo:
     """Get the k8s version of the cluster under test."""
     cluster_config = await kubernetes_asyncio.config.load_kube_config(kubeconfig, context)
@@ -416,9 +474,16 @@ def check_k8s_version_recency(
         if my_version.patch >= release.version.patch:
             continue
         # at this point `release` has the same major.minor, but higher patch than `my_version`
-        if release.age > PATCH_VERSION_CADENCE:
-            # whoops, the cluster should have been updated to this (or a higher version) already!
-            return False
+        if my_version == release.version:
+            if release.age <= MINOR_VERSION_CADENCE:
+                logger.info(f"Version {my_version} is recent (within cadence).")
+                return True
+            else:
+                logger.error(f"Version {my_version} is too old.")
+                return False
+        # if release.age > PATCH_VERSION_CADENCE:
+        #     # whoops, the cluster should have been updated to this (or a higher version) already!
+        #     return False
         ranges = [_range for _range in cve_affected_ranges if my_version in _range]
         if ranges and release.age > CVE_VERSION_CADENCE:
             # -- two FIXMEs: