openpgp: use imported key metadata for key expiration

gemato/openpgp.py

@@ -96,6 +96,20 @@ def primary_key_fingerprint(self) -> str:
         return self[0].primary_key_fingerprint
 
 
+@dataclasses.dataclass
+class OpenPGPSubKey:
+    keyid: str
+    created: datetime.datetime | None
+    fingerprint: str = ""
+    expires: datetime.datetime | None = None
+
+
+@dataclasses.dataclass
+class OpenPGPPubKey(OpenPGPSubKey):
+    uids: list[bytes] = dataclasses.field(default_factory=list)
+    subkeys: dict[str, OpenPGPSubKey] = dataclasses.field(default_factory=dict)
+
+
 ZBASE32_TRANSLATE = bytes.maketrans(
     b'ABCDEFGHIJKLMNOPQRSTUVWXYZ234567',
     b'ybndrfg8ejkmcpqxot1uwisza345h769')
@@ -141,7 +155,7 @@ def import_key(self, keyfile, trust=True):
             'import_key() is not implemented by this OpenPGP provider')
 
     def list_keys(self, key_ids: list[str] = []
-                  ) -> dict[str, list[tuple[str, str]]]:
+                  ) -> dict[str, OpenPGPPubKey]:
         """
         List fingerprints and UIDs of specified keys or all keys in keyring
 
@@ -153,57 +167,113 @@ def list_keys(self, key_ids: list[str] = []
             [GNUPG, '--batch', '--with-colons', '--list-keys', *key_ids],
             raise_on_error=OpenPGPKeyListingError)
 
-        prev_pub = None
-        fpr = None
-        ret = {}
+        pubkeys: dict[str, OpenPGPPubKey] = {}
+
+        current_pub: OpenPGPPubKey | None = None
+        current_sub: OpenPGPSubKey | None = None
+
+        def pub_fpr_waiting(pub: OpenPGPPubKey | None) -> bool:
+            return pub is not None and pub.fingerprint == ""
+
+        def set_fingerprint(ktype: str,
+                            key: OpenPGPSubKey | OpenPGPPubKey,
+                            fpr: str) -> None:
+            if not fpr.endswith(key.keyid):
+                raise OpenPGPKeyListingError(
+                    f"Incorrect fingerprint {fpr} for {ktype} key {key.keyid}"
+                )
+
+            key.fingerprint = fpr
+            LOGGER.debug("list_keys(): %s fingerprint: %r", ktype, fpr)
+
         for line in out.splitlines():
-            if line.startswith(b"fpr:"):
-                fpr = line.split(b":")[9].decode("ASCII")
-                # this can be subkey fingerprint
-                if prev_pub is None:
-                    continue
-                if not fpr.endswith(prev_pub):
+            fields = line.split(b":")
+            rec_type = fields[0]
+
+            if rec_type in (b"pub", b"sub"):
+                if pub_fpr_waiting(current_pub) or current_sub is not None:
                     raise OpenPGPKeyListingError(
-                        f"Incorrect fingerprint {fpr} for key {prev_pub}"
+                        f"New key while waiting for fingerprint: {line!r}"
                     )
-                LOGGER.debug(
-                    f"list_keys(): fingerprint: {fpr}")
-                ret[fpr] = []
-                prev_pub = None
-            elif line.startswith(b"pub:"):
-                if prev_pub is not None:
-                    raise OpenPGPKeyListingError(
-                        f"New key while waiting for fingerprint: {line} "
-                        f"({prev_pub=})"
+
+                keyid = fields[4].decode("ASCII")
+                created = (
+                    self._parse_gpg_ts(fields[5].decode('utf8'))
+                    if fields[5]
+                    else None
+                )
+                expires = (
+                    self._parse_gpg_ts(fields[6].decode('utf8'))
+                    if fields[6]
+                    else None
+                )
+
+                if rec_type == b"pub":
+                    assert current_sub is None
+
+                    current_pub = OpenPGPPubKey(
+                        keyid=keyid,
+                        created=created,
+                        expires=expires,
+                    )
+                else:  # b"sub"
+                    assert current_sub is None
+
+                    if current_pub is None:
+                        raise OpenPGPKeyListingError(
+                            f"Subkey without public key: {line!r}"
+                        )
+
+                    current_sub = OpenPGPSubKey(
+                        keyid=keyid,
+                        created=created,
+                        expires=expires,
                     )
-                # wait for the fingerprint
-                prev_pub = line.split(b":")[4].decode("ASCII")
-                LOGGER.debug(f"list_keys(): keyid: {prev_pub}")
-            elif line.startswith(b"sub:"):
-                if fpr is None:
+
+                LOGGER.debug(
+                    "list_keys(): %s keyid: %s",
+                    rec_type.decode('ASCII'),
+                    keyid,
+                )
+            elif rec_type == b"fpr":
+                fpr = fields[9].decode("ASCII")
+                if pub_fpr_waiting(current_pub):
+                    set_fingerprint("pub", current_pub, fpr)
+                    pubkeys[fpr] = current_pub
+                elif current_sub is not None:
+                    set_fingerprint("sub", current_sub, fpr)
+                    current_pub.subkeys[fpr] = current_sub
+                    current_sub = None
+                else:
                     raise OpenPGPKeyListingError(
-                        "Subkey without prior fingerprint in GPG output: "
-                        f"{line} ({prev_pub=})"
+                        f"Fingerprint without preceding key: {line!r}"
                     )
-                prev_pub = None
-            elif line.startswith(b'uid:'):
-                if fpr is None:
+            elif rec_type == b"uid":
+                if current_pub is None or pub_fpr_waiting(current_pub):
                     raise OpenPGPKeyListingError(
-                        f"UID without prior fingerprint in GPG output: {line} "
-                        f"({prev_pub=})"
+                        f"UID without public key: {line!r}"
                     )
-                uid_split = line.split(b":", 10)
-                uid = uid_split[9]
-                # no creation date means missing/broken self-sig
-                if not uid_split[5]:
+
+                uid = fields[9]
+
+                if not fields[5]:  # creation date
                     LOGGER.debug(
-                        f"list_keys(): skipping UID with missing self-sig: "
-                        f"{fpr=}, {uid=!r}")
+                        "list_keys(): skipping UID with missing self-sig: "
+                        "fingerprint=%r, uid=%r",
+                        current_pub.fingerprint,
+                        uid,
+                    )
                     continue
-                LOGGER.debug(f'list_keys(): UID: {uid}')
-                ret[fpr].append(uid)
 
-        return ret
+                LOGGER.debug("list_keys(): UID: %r", uid)
+                current_pub.uids.append(uid)
+
+        if pub_fpr_waiting(current_pub) or current_sub is not None:
+            raise OpenPGPKeyListingError(
+                "Unexpected end of GPG output while waiting for fingerprint"
+            )
+
+        return pubkeys
 
     def refresh_keys(self, allow_wkd=True, keyserver=None):
         """
@@ -244,6 +314,8 @@ def _process_gpg_verify_output(self,
                                    ) -> OpenPGPSignatureList:
         """Process the output of gpg --verify and return a siglist"""
 
+        key_expiration_cache: dict[str, datetime.datetime | None] | None = None
+
         sig_list = OpenPGPSignatureList()
         for line in out.splitlines():
             if line.startswith(b'[GNUPG:] NEWSIG'):
@@ -277,12 +349,23 @@ def _process_gpg_verify_output(self,
                 spl = line.split(b' ')
                 assert len(spl) >= 12
                 sig_list[-1].valid_sig = True
-                sig_list[-1].fingerprint = spl[2].decode('utf8')
+                fingerprint = spl[2].decode('utf8')
+                sig_list[-1].fingerprint = fingerprint
                 sig_list[-1].timestamp = (
                     self._parse_gpg_ts(spl[4].decode('utf8')))
                 sig_list[-1].expire_timestamp = (
                     self._parse_gpg_ts(spl[5].decode('utf8')))
                 sig_list[-1].primary_key_fingerprint = spl[11].decode('utf8')
+                if key_expiration_cache is None:
+                    key_expiration_cache = {}
+                    for pub in self.list_keys().values():
+                        key_expiration_cache[pub.fingerprint] = pub.expires
+                        for sub in pub.subkeys.values():
+                            key_expiration_cache[sub.fingerprint] = sub.expires
+
+                sig_list[-1].key_expiration = (
+                    key_expiration_cache.get(fingerprint)
+                )
             elif line.startswith(b'[GNUPG:] TRUST_'):
                 assert sig_list
                 spl = line.split(b' ', 2)
@@ -291,12 +374,10 @@ def _process_gpg_verify_output(self,
                               b'TRUST_ULTIMATE'):
                     sig_list[-1].trusted_sig = True
             elif line.startswith(b"[GNUPG:] KEYEXPIRED"):
-                # TODO: will the "correct" key be emitted last?
-                assert sig_list
-                spl = line.split(b" ", 3)
-                assert len(spl) >= 3
-                sig_list[-1].key_expiration = (
-                    self._parse_gpg_ts(spl[2].decode("utf8")))
+                # Ignore KEYEXPIRED because it does not identify
+                # which key expired. Use expiration from imported
+                # key metadata instead.
+                continue
 
         if not sig_list:
             raise OpenPGPUnknownSigFailure(
@@ -560,7 +641,11 @@ def import_key(self, keyfile, trust=True):
             raise OpenPGPKeyImportError("No keys imported")
 
         imported = self.list_keys(list(fprs))
-        missing = fprs - {fpr for fpr, uids in imported.items() if uids}
+        missing = fprs - {
+            key.fingerprint
+            for key in imported.values()
+            if key.uids
+        }
         if missing:
             raise OpenPGPKeyImportError(
                 "Import succeeded but no valid key for fingerprints: "
@@ -591,9 +676,9 @@ def refresh_keys_wkd(self):
             LOGGER.debug('refresh_keys_wkd(): no keys found')
             return False
         addrs = set()
-        for key, uids in keys.items():
+        for key, pub in keys.items():
             key_addrs = {
-                addr for uid in uids
+                addr for uid in pub.uids
                 if "@" in (addr := email.utils.parseaddr(
                     uid.decode("utf8", errors="replace")
                 )[1])

tests/test_openpgp.py

@@ -43,6 +43,8 @@
     OpenPGPSignatureList,
     OpenPGPSignatureData,
     OpenPGPSignatureStatus,
+    OpenPGPPubKey,
+    OpenPGPSubKey,
     )
 from gemato.recursiveloader import ManifestRecursiveLoader
 
@@ -761,19 +763,76 @@ def test_recursive_manifest_loader_save_submanifest(tmp_path, privkey_env):
 
 @pytest.mark.parametrize(
     'key_var,expected',
-    [('VALID_PUBLIC_KEY',
-      {KEY_FINGERPRINT: [b"gemato test key <gemato@example.com>"]}),
-     ('OTHER_VALID_PUBLIC_KEY',
-      {OTHER_KEY_FINGERPRINT: [b"gemato test key <gemato@example.com>"]}),
-     ('VALID_KEY_SUBKEY',
-      {KEY_FINGERPRINT: [b"gemato test key <gemato@example.com>"]}),
-     ('VALID_KEY_NOEMAIL',
-      {KEY_FINGERPRINT: [b"gemato test key"]}),
-     ('VALID_KEY_NONUTF',
-      {KEY_FINGERPRINT: [
-           b"gemat\\xf6 test key <gemato@example.com>" if is_sequoia()
-           else b"gemat\xf6 test key <gemato@example.com>"
-       ]}),
+    [
+        (
+            'VALID_PUBLIC_KEY', {
+                KEY_FINGERPRINT: OpenPGPPubKey(
+                    fingerprint=KEY_FINGERPRINT,
+                    keyid=KEY_FINGERPRINT[-16:],
+                    created=datetime.datetime(2020, 1, 1),
+                    uids=[
+                        b'gemato test key <gemato@example.com>',
+                    ],
+                ),
+            },
+        ),
+        (
+            'OTHER_VALID_PUBLIC_KEY', {
+                OTHER_KEY_FINGERPRINT: OpenPGPPubKey(
+                    fingerprint=OTHER_KEY_FINGERPRINT,
+                    keyid=OTHER_KEY_FINGERPRINT[-16:],
+                    created=datetime.datetime(2020, 1, 1),
+                    uids=[
+                        b'gemato test key <gemato@example.com>',
+                    ],
+                ),
+            },
+        ),
+        (
+            'VALID_KEY_SUBKEY', {
+                KEY_FINGERPRINT: OpenPGPPubKey(
+                    fingerprint=KEY_FINGERPRINT,
+                    keyid=KEY_FINGERPRINT[-16:],
+                    created=datetime.datetime(2020, 1, 1),
+                    uids=[
+                        b'gemato test key <gemato@example.com>',
+                    ],
+                    subkeys={
+                        SUBKEY_FINGERPRINT: OpenPGPSubKey(
+                            fingerprint=SUBKEY_FINGERPRINT,
+                            keyid=SUBKEY_FINGERPRINT[-16:],
+                            created=datetime.datetime(2020, 1, 1),
+                        ),
+                    },
+                ),
+            },
+        ),
+        (
+            'VALID_KEY_NOEMAIL', {
+                KEY_FINGERPRINT: OpenPGPPubKey(
+                    fingerprint=KEY_FINGERPRINT,
+                    keyid=KEY_FINGERPRINT[-16:],
+                    created=datetime.datetime(2020, 1, 1),
+                    uids=[
+                        b'gemato test key',
+                    ],
+                ),
+                },
+        ),
+        (
+            'VALID_KEY_NONUTF', {
+                KEY_FINGERPRINT: OpenPGPPubKey(
+                    fingerprint=KEY_FINGERPRINT,
+                    keyid=KEY_FINGERPRINT[-16:],
+                    created=datetime.datetime(2020, 1, 1),
+                    uids=[
+                        b"gemat\\xf6 test key <gemato@example.com>"
+                        if is_sequoia() else
+                        b"gemat\xf6 test key <gemato@example.com>",
+                    ],
+                ),
+            },
+        ),
      ])
 def test_list_keys(openpgp_env, key_var, expected):
     try: