Load product metadata from ProductID certificates [RHELDST-24276] (#476)

* Load product metadata from ProductID certificates [RHELDST-24276]

python-rhsm cannot be installed with GCC 14. Given that project
has been deprecated for years, it's unlikely the actual underlying
issue will be resolved. There are several projects in the distribution
realm that used python-rhsm to read ProductID certificates and pull
product metadata from them. After some internal discussion, pushsource
seems to be an acceptable place where to put the replacement code which
can be used instead of python-rhsm.

This commit adds logic for reading metadata from Red Hat ProductID
certificate files leveraging `cryptography` and `pyasn1` modules.

It also slightly adjusts `conv.sloppylist()` so the individual string
elements are stripped of leading and trailing whitespaces after
splitting.

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

requirements.in

@@ -1,11 +1,13 @@
 attrs
-more-executors>=2.7.0
-koji>=1.18
-pushcollector
-PyYAML
+cryptography
 frozendict; python_version >= '3.6'
 frozenlist2
-python-dateutil
 kobo
+koji>=1.18
+more-executors>=2.7.0
+pushcollector
+pyasn1
+python-dateutil
 pytz; python_version < '3.9'
+PyYAML
 requests

src/pushsource/__init__.py

@@ -8,6 +8,7 @@
     CompsXmlPushItem,
     ModuleMdPushItem,
     ModuleMdSourcePushItem,
+    ProductId,
     ProductIdPushItem,
     RpmPushItem,
     ContainerImagePushItem,

src/pushsource/_impl/model/__init__.py

@@ -21,7 +21,7 @@
 )
 from .modulemd import ModuleMdPushItem, ModuleMdSourcePushItem
 from .comps import CompsXmlPushItem
-from .productid import ProductIdPushItem
+from .productid import ProductId, ProductIdPushItem
 from .ami import (
     AmiPushItem,
     AmiRelease,

src/pushsource/_impl/model/conv.py

@@ -21,7 +21,7 @@ def sloppylist(value, elem_converter=None):
     Optionally use elem_converter to convert each list element.
     """
     if isinstance(value, str):
-        value = value.split(",")
+        value = [v.strip() for v in value.split(",")]
     if elem_converter:
         value = [elem_converter(elem) for elem in value]
     return frozenlist(value)

src/pushsource/_impl/model/productid.py

@@ -1,14 +1,116 @@
+from collections import defaultdict
+
+from cryptography import x509
+from frozenlist2 import frozenlist
+from pyasn1.codec.der import decoder
+
 from .base import PushItem
+from .conv import convert_maybe, sloppylist
 from .. import compat_attr as attr
 
 
+# Red Hat OID namespace is "1.3.6.1.4.1.2312.9",
+# the trailing ".1" designates a Product Certificate.
+OID_NAMESPACE = "1.3.6.1.4.1.2312.9.1."
+
+
+@attr.s()
+class ProductId(object):
+    """A ProductID represents a group of metadata pertaining to a single product
+    contained in a ProductID certificate."""
+
+    id = attr.ib(type=int)
+    """Product Engineering ID (EngID), e.g. 72
+
+    :type: int
+    """
+
+    name = attr.ib(type=str, default=None)
+    """Human readable product name, e.g. "Red Hat Enterprise Linux for IBM z Systems"
+
+    :type: str
+    """
+
+    version = attr.ib(type=str, default=None)
+    """Human readable product version string, e.g. "9.4"
+
+    :type: str
+    """
+
+    architecture = attr.ib(type=list, default=None, converter=convert_maybe(sloppylist))
+    """List of architectures supported by the product, e.g. ["s390x"]
+
+    :type: List[str]
+    """
+
+    provided_tags = attr.ib(
+        type=list, default=None, converter=convert_maybe(sloppylist)
+    )
+    """List of tags describing the provided platforms used for pairing with other products,
+    e.g. ["rhel-9", "rhel-9-s390x"]
+
+    :type: List[str]
+    """
+
+
 @attr.s()
 class ProductIdPushItem(PushItem):
     """A :class:`~pushsource.PushItem` representing a product ID certificate.
 
     For push items of this type, the :meth:`~pushsource.PushItem.src` attribute
     refers to a file containing a PEM certificate identifying a product.
+    """
 
-    This library does not verify that the referenced file is a valid
-    certificate.
+    products = attr.ib(type=list, converter=frozenlist)
+    """List of products described by the ProductID certificate.
+
+    :type: List[ProductID]
+
+    .. versionadded:: 2.45.0
     """
+
+    @products.default
+    def _default_products(self):
+        return frozenlist(self._load_products(self.src) if self.src else [])
+
+    def _load_products(self, path):
+        """Returns a list of ProductIDs described by the ProductID X.509 certificate file
+        in PEM format. Raises ValueError if the file doesn't describe any ProductID."""
+
+        with open(path, "rb") as f:
+            x509_certificate = x509.load_pem_x509_certificate(f.read())
+        # Extensions are most commonly ASN.1 (DER) encoded UTF-8 strings.
+        # First byte is usually 0x13 = PrintableString, second byte is the length of the string
+        # However we can't rely on that and must parse the fields safely using a proper ASN.1 / DER
+        # parser. Although cryptography module does its own ASN.1 / DER parsing, it doesn't provide
+        # any public API for that yet (see https://github.com/pyca/cryptography/issues/9283),
+        # so pyasn1 module has to be used instead.
+        products_data = defaultdict(dict)
+        for extension in x509_certificate.extensions:
+            oid = extension.oid.dotted_string
+            if oid.startswith(OID_NAMESPACE):
+                # OID component with index 9 is always EngID
+                # OID component with index 10 (last) is:
+                #  1 = Product Name, e.g. "Red Hat Enterprise Linux for IBM z Systems"
+                #  2 = Product Version, e.g. "9.4"
+                #  3 = Product Architecture, e.g. "s390x"
+                #  4 = Product Tags / Provides, e.g. "rhel-9,rhel-9-s390x"
+                eng_id, attribute_id = map(int, oid.split(".")[9:11])
+                products_data[eng_id][attribute_id] = str(
+                    decoder.decode(extension.value.value)[0]
+                )
+
+        if not products_data:
+            raise ValueError("File '%s' is not a ProductID certificate." % path)
+
+        result = []
+        for eng_id, product_data in products_data.items():
+            product = ProductId(
+                id=eng_id,
+                name=product_data.get(1),
+                version=product_data.get(2),
+                architecture=product_data.get(3),
+                provided_tags=product_data.get(4),
+            )
+            result.append(product)
+        return result