Add function to determine RPM signing state

This adds a function that can tell whether a package is header-only
signable, header-only signed, IMA signed and signed.

Signed-off-by: Patrick Uiterwijk <[email protected]>

rpm_head_signing/__init__.py

@@ -5,3 +5,4 @@
     parse_ima_signature,
     get_rpm_ima_signature_info,
 )
+from .determine import determine_rpm_status

rpm_head_signing/determine.py

@@ -0,0 +1,63 @@
+import rpm
+from koji import (
+    find_rpm_sighdr,
+    rpm_hdr_size,
+    RawHeader,
+    get_header_fields,
+)
+
+from .extract_header import RPMTAG_PAYLOADDIGEST
+from .extract_rpm_with_filesigs import RPMSIGTAG_FILESIGNATURES
+
+
+class DetermineResult(object):
+    is_head_only_signable = None
+    is_head_only_signed = None
+    is_signed = None
+    is_ima_signed = None
+
+
+def determine_rpm_status(rpm_path):
+    (sig_start, sig_size) = find_rpm_sighdr(rpm_path)
+    hdr_start = sig_start + sig_size
+    hdr_size = rpm_hdr_size(rpm_path, hdr_start)
+
+    with open(rpm_path, "rb") as f:
+        f.seek(sig_start)
+        sighdr = f.read(sig_size)
+        sighdr = RawHeader(sighdr)
+        f.seek(hdr_start)
+        hdr = f.read(hdr_size)
+        hdr = RawHeader(hdr)
+
+    fields = get_header_fields(
+        rpm_path,
+        (
+            "name",
+            "basenames",
+            "siggpg",
+            "sigpgp",
+            "rsaheader",
+            "dsaheader",
+        ),
+    )
+
+    filesignatures = sighdr.index.get(RPMSIGTAG_FILESIGNATURES)
+    payloaddigest = hdr.index.get(RPMTAG_PAYLOADDIGEST)
+
+    result = DetermineResult()
+    result.is_head_only_signable = payloaddigest is not None
+    result.is_head_only_signed = (
+        (fields["rsaheader"] is not None or fields["dsaheader"] is not None)
+        and fields["siggpg"] is None
+        and fields["sigpgp"] is None
+    )
+    result.is_signed = (
+        fields["rsaheader"] is not None
+        or fields["dsaheader"] is not None
+        or fields["siggpg"] is not None
+        or fields["sigpgp"] is not None
+    )
+    result.is_ima_signed = filesignatures is not None
+
+    return result

test.py

@@ -92,6 +92,16 @@ def test_extract_non_hdrsign_able(self):
             )
         self.assertFalse(os.path.exists(os.path.join(self.tmpdir, "digests.out.tmp")))
 
+        result = rpm_head_signing.determine_rpm_status(
+            os.path.join(
+                self.asset_dir, "sblim-cim-client-javadoc-1.3.9.1-1.el6.noarch.rpm"
+            )
+        )
+        self.assertFalse(result.is_head_only_signable)
+        self.assertFalse(result.is_head_only_signed)
+        self.assertTrue(result.is_signed)
+        self.assertFalse(result.is_ima_signed)
+
     def test_insert_no_ima(self):
         self._add_gpg_key("gpgkey.asc")
         for pkg in self.pkg_numbers:
@@ -127,6 +137,14 @@ def test_insert_no_ima(self):
             self.assertTrue(b"Header V3 RSA" in res)
             self.assertTrue(b"15f712be: ok" in res.lower())
 
+            result = rpm_head_signing.determine_rpm_status(
+                os.path.join(self.tmpdir, "testpkg-%s.rpm" % pkg)
+            )
+            self.assertTrue(result.is_head_only_signable)
+            self.assertTrue(result.is_head_only_signed)
+            self.assertTrue(result.is_signed)
+            self.assertFalse(result.is_ima_signed)
+
     def test_insert_ima_presigned(self):
         def insert_cb(pkg):
             rpm_head_signing.insert_signature(
@@ -301,6 +319,14 @@ def _ima_insertion_test(self, insert_command, rpm_keyid, nonhdrsigned=False):
             msg = ("%s: ok" % rpm_keyid).encode("utf8")
             self.assertTrue(msg in res.lower())
 
+            result = rpm_head_signing.determine_rpm_status(
+                os.path.join(self.tmpdir, rpm_filename)
+            )
+            self.assertTrue(result.is_head_only_signable)
+            self.assertTrue(result.is_head_only_signed)
+            self.assertTrue(result.is_signed)
+            self.assertTrue(result.is_ima_signed)
+
         # Construct the arguments to verify_rpm
         args = self._get_verify_rpm_args()
         args.extend(rpm_paths)