use rhel9 release graph

Author: jasinner

pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     "packageurl-python>=0.16.0",
     "pkce>=1.0.3",
     "pyjwt>=2.10.1",
+    "pyyaml>=6.0",
     "rich>=14.0.0",
     "univers>=30.12.1",
 ]

src/trustshell/product_definitions.py

@@ -9,6 +9,7 @@
 
 from anytree import Node, NodeMixin, LevelOrderGroupIter
 from trustshell import CONFIG_DIR, console
+from trustshell.rhel_releases import EnhancedProdDefs
 
 logger = logging.getLogger(__name__)
 
@@ -47,39 +48,20 @@ class ProductStream(ProductBase, NodeMixin):
     def __init__(self, name: str, cpes: list[str] = [], active: bool = False) -> None:
         super().__init__(name)
         # In rhel 10 we don't use mainline CPEs, so we need to filter them out
-        if name.startswith("rhel-") and not name.startswith("rhel-10"):
-            self.cpes = self._filter_rhel_mainline_cpes(cpes)
+        if name.startswith("rhel-"):
+            # Remove any single digit like cpe:/a:redhat:enterprise_linux:9::appstream
+            self.cpes = [
+                cpe
+                for cpe in cpes
+                if not re.search(r":redhat:enterprise_linux:\d:", cpe)
+            ]
         else:
             self.cpes = cpes
         self.active = active
 
     def set_active(self, active: bool) -> None:
         self.active = active
 
-    @staticmethod
-    def _filter_rhel_mainline_cpes(cpes: list[str]) -> list[str]:
-        """Special logic for RHEL streams is required because during the lifetime of a RHEL
-        release it will have the main line CPE (redhat:enterprise_linux:),
-        and also EUS/AUS/TUS CPEs depending on it's lifecycle phase. The rule here is that
-        if a stream has both main line and EUS/AUS/TUS ignore the main line one
-        (which is less specific)"""
-        cpe_set = set(cpes)
-        if len(cpe_set) <= 1:
-            return cpes
-        has_eatus = False
-        for cpe in cpes:
-            for us in "eus", "aus", "tus", "e4s":
-                if f":rhel_{us}:" in cpe:
-                    has_eatus = True
-                    break
-            if has_eatus:
-                break
-        if not has_eatus:
-            return cpes
-        return [
-            cpe for cpe in cpes if not re.search(r":redhat:enterprise_linux:\d:", cpe)
-        ]
-
 
 class ProdDefs:
     ETAG_FILE = os.path.join(CONFIG_DIR, "etag.txt")
@@ -141,11 +123,32 @@ def get_product_definitions_service(cls) -> dict[str, Any]:
                 product_definitions = json.load(f)
         return product_definitions
 
-    def __init__(self, active_only: bool = True) -> None:
+    def __init__(
+        self,
+        active_only: bool = True,
+        rhel_git_branch: str = "main",
+        rhel_releases_path: str = "",
+    ) -> None:
         self.stream_nodes_by_cpe: dict[str, list[ProductStream]] = defaultdict(list)
         product_streams_by_name: dict[str, list[ProductStream]] = defaultdict(list)
         self.product_trees: list[NodeMixin] = []
 
+        # Initialize enhanced RHEL release data if requested
+        self.enhanced_proddefs: Optional[EnhancedProdDefs] = None
+        if rhel_releases_path:
+            # Use local file for testing
+            try:
+                self.enhanced_proddefs = EnhancedProdDefs(
+                    git_branch=rhel_git_branch, rhel_releases_path=rhel_releases_path
+                )
+            except Exception as e:
+                logger.warning(
+                    f"Could not initialize enhanced product definitions: {e}"
+                )
+                self.enhanced_proddefs = None
+        else:
+            self.enhanced_proddefs = self._enable_rhel_release_data(rhel_git_branch)
+
         data = self.get_product_definitions_service()
 
         if not data:
@@ -201,6 +204,36 @@ def match_module_pattern(self, cpe: str) -> list[ProductModule]:
                         module_matches.append(module)
         return module_matches
 
+    def _enable_rhel_release_data(
+        self, git_branch: str = "main", rhel_releases_path: str = ""
+    ) -> Optional[EnhancedProdDefs]:
+        """
+        Enable RHEL release data fetching from GitLab repository or local file.
+
+        Args:
+            git_branch: Git branch to use for fetching data
+            rhel_releases_path: Local file path (for testing)
+
+        Returns:
+            EnhancedProdDefs instance or None if failed
+        """
+        try:
+            enhanced_proddefs = EnhancedProdDefs(
+                git_branch=git_branch, rhel_releases_path=rhel_releases_path
+            )
+            if rhel_releases_path:
+                logger.info(
+                    f"Enabled RHEL release data from local file: {rhel_releases_path}"
+                )
+            else:
+                logger.info(
+                    f"Enabled RHEL release data from GitLab (branch: {git_branch})"
+                )
+            return enhanced_proddefs
+        except Exception as e:
+            logger.error(f"Could not enable RHEL release data: {e}")
+            return None
+
     @staticmethod
     def _clean_cpe(cpe: str) -> str:
         """CPEs from SBOMs have extra characters added to them, clean them up here
@@ -226,6 +259,13 @@ def extend_with_product_mappings(
         for tree in ancestor_trees:
             for leaf in tree.leaves:
                 cleaned_leaf_name = self._clean_cpe(leaf.name)
+                # Don't try and match single digit enterprise_linux CPEs
+                # Filter out single digit enterprise_linux CPEs (only when not keeping CPEs)
+                if not keep_cpes and re.search(
+                    r":redhat:enterprise_linux:\d:", cleaned_leaf_name
+                ):
+                    leaf.parent = None
+                    continue
                 leaf_with_products = self._check_streams(
                     leaf, cleaned_leaf_name, keep_cpes
                 )
@@ -248,6 +288,13 @@ def extend_with_product_mappings(
     def _check_streams(self, leaf: Node, cpe: str, keep_cpes: bool) -> list[Node]:
         """Check if cpe matches exactly to any ProductStreams, if it does add the CPE as a parent
         of the stream. If more than one stream matches, create copies of the stream and leaf"""
+        # First try enhanced matching if RHEL release data is available
+        enhanced_streams = self._check_enhanced_streams(cpe)
+        if enhanced_streams:
+            return self._duplicate_leaves_and_set_parents(
+                leaf, enhanced_streams, keep_cpes
+            )
+        # Fallback to original direct matching
         if cpe not in self.stream_nodes_by_cpe:
             return []
         stream_nodes = self.stream_nodes_by_cpe[cpe]
@@ -259,6 +306,39 @@ def _check_streams(self, leaf: Node, cpe: str, keep_cpes: bool) -> list[Node]:
             leaf, copy_of_stream_nodes, keep_cpes
         )
 
+    def _check_enhanced_streams(self, cpe: str) -> list[ProductStream]:
+        """Check if CPE matches using enhanced RHEL release data logic."""
+        if not self.enhanced_proddefs:
+            return []
+
+        # Get active streams and their CPEs
+        active_streams = set()
+        stream_cpes = {}
+
+        for stream_name, stream_nodes in self.stream_nodes_by_cpe.items():
+            for stream_node in stream_nodes:
+                if stream_node.active:
+                    active_streams.add(stream_node.name)
+                    if stream_node.name not in stream_cpes:
+                        stream_cpes[stream_node.name] = []
+                    stream_cpes[stream_node.name].extend(stream_node.cpes)
+
+        # Use enhanced matching to find relevant active streams
+        matching_stream_names = self.enhanced_proddefs.enhance_cpe_matching(
+            cpe, active_streams, stream_cpes
+        )
+
+        # Return the actual ProductStream objects for the matching streams
+        result_streams = []
+        for stream_name in matching_stream_names:
+            # Find ProductStream objects with this name
+            for stream_nodes in self.stream_nodes_by_cpe.values():
+                for stream_node in stream_nodes:
+                    if stream_node.name == stream_name and stream_node.active:
+                        result_streams.append(copy.deepcopy(stream_node))
+
+        return result_streams
+
     def _check_modules(self, leaf: Node, cpe: str, keep_cpes: bool) -> list[Node]:
         """Check if the cpe matches any ProductModule"""
         module_nodes = self.match_module_pattern(cpe)
@@ -280,7 +360,11 @@ def _duplicate_leaves_and_set_parents(
             If keep_cpes=False: Returns the list of unique streams that replaced the leaf.
         """
         if not product_nodes:
-            return []
+            if keep_cpes:
+                # Keep the CPE node even if no products match
+                return [leaf]
+            else:
+                return []
 
         # Convert modules to their parent streams
         streams_to_attach: list[Any] = []
@@ -313,6 +397,37 @@ def _duplicate_leaves_and_set_parents(
                 stream.parent = leaf.parent
             return unique_streams
 
+    def get_all_cpes_for_rhel_stream(self, stream_name: str) -> set[str]:
+        """
+        Get all CPEs that should be associated with a given RHEL stream by traversing
+        the RHEL release graph to include related releases.
+
+        Args:
+            stream_name: The ps_update_stream name (e.g., "rhel-9.2.0.z")
+
+        Returns:
+            Set of all CPEs that should be associated with this stream
+        """
+        if not self.enhanced_proddefs:
+            # Fallback to direct stream CPEs only
+            result = set()
+            for cpe, stream_nodes in self.stream_nodes_by_cpe.items():
+                for stream_node in stream_nodes:
+                    if stream_node.name == stream_name:
+                        result.update(stream_node.cpes)
+            return result
+
+        # Get stream CPEs mapping
+        stream_cpes = {}
+        for cpe, stream_nodes in self.stream_nodes_by_cpe.items():
+            for stream_node in stream_nodes:
+                if stream_node.name not in stream_cpes:
+                    stream_cpes[stream_node.name] = []
+                stream_cpes[stream_node.name].extend(stream_node.cpes)
+
+        # Use enhanced matching to get all related CPEs
+        return self.enhanced_proddefs.get_all_cpes_for_stream(stream_name, stream_cpes)
+
     def _add_ancestor(self, leaf: Node, product: Any) -> None:
         if product.parent:
             product.parent.parent = leaf

src/trustshell/products.py

@@ -106,7 +106,9 @@ def search(
         console.print("No results")
         return
 
-    prod_defs = ProdDefs()
+    prod_defs = ProdDefs(
+        rhel_releases_path="tests/testdata/products/rhel9-releases.yml"
+    )
     prod_defs.extend_with_product_mappings(ancestor_trees, keep_cpes=cpes)
 
     seen_trees = set()