You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by sb...@apache.org on 2026/08/20 18:49:28 UTC

(tooling-trusted-releases) branch sbp updated: Merge OpenPGP keys as a union of their packets

This is an automated email from the ASF dual-hosted git repository.

sbp pushed a commit to branch sbp
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git


The following commit(s) were added to refs/heads/sbp by this push:
     new fdda1630 Merge OpenPGP keys as a union of their packets
fdda1630 is described below

commit fdda163042fc5138dbc81227e805fac08917981d
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Thu Aug 20 19:48:49 2026 +0100

    Merge OpenPGP keys as a union of their packets
---
 atr/pgp.py                    | 263 ++++++++++++++++++++++++++++++++++++++++--
 atr/tasks/checks/signature.py |   2 +-
 tests/unit/pgp_fixtures.py    |  52 +++++++++
 tests/unit/test_pgp.py        |  62 ++++++++++
 4 files changed, 371 insertions(+), 8 deletions(-)

diff --git a/atr/pgp.py b/atr/pgp.py
index af05b616..721151a7 100644
--- a/atr/pgp.py
+++ b/atr/pgp.py
@@ -15,8 +15,10 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import base64
 import dataclasses
 import datetime
+import itertools
 from typing import Final, Literal
 
 import openpgp
@@ -31,6 +33,15 @@ _KEY_REVOCATION_SIGNATURE_TYPE: Final[str] = "key-revocation"
 _SUBKEY_BINDING_SIGNATURE_TYPE: Final[str] = "subkey-binding"
 _SUBKEY_REVOCATION_SIGNATURE_TYPE: Final[str] = "subkey-revocation"
 
+_ARMOR_BEGIN: Final[str] = "-----BEGIN PGP PUBLIC KEY BLOCK-----"
+_ARMOR_END: Final[str] = "-----END PGP PUBLIC KEY BLOCK-----"
+_PRIMARY_KEY_TAG: Final[int] = 6
+_PUBLIC_SUBKEY_TAG: Final[int] = 14
+_SIGNATURE_TAG: Final[int] = 2
+_SKIPPABLE_PACKET_TAGS: Final[frozenset[int]] = frozenset({10, 12, 21})
+_USER_ATTRIBUTE_TAG: Final[int] = 17
+_USER_ID_TAG: Final[int] = 13
+
 ComponentKind = Literal["primary", "user-id", "user-attribute", "subkey"]
 
 
@@ -74,6 +85,16 @@ class SigningKeyStatus:
     revoked: bool
 
 
+def certificate_block_fingerprint(block: str) -> str:
+    frames = _frames(_dearmored(block))
+    if (not frames) or (frames[0][0] != _PRIMARY_KEY_TAG):
+        raise ValueError("The block does not begin with a primary key packet")
+    value = openpgp.packet.Packet.from_bytes(_framed(*frames[0])).value
+    if not isinstance(value, openpgp.packet.PublicKey):
+        raise ValueError("The primary packet is not a public key")
+    return value.fingerprint.lower()
+
+
 def certificate_block_shape(keys: list[openpgp.composed.SignedPublicKey], fingerprint: str) -> str:
     fingerprints = [key.fingerprint.lower() for key in keys]
     matches = fingerprints.count(fingerprint.lower())
@@ -88,6 +109,14 @@ def certificate_block_shape(keys: list[openpgp.composed.SignedPublicKey], finger
     return "multi-own-first" if (fingerprints[0] == fingerprint.lower()) else "multi-own-not-first"
 
 
+def certificate_blocks(text: str) -> list[str]:
+    frames = _frames(_dearmored(text))
+    starts = [index for index, frame in enumerate(frames) if frame[0] == _PRIMARY_KEY_TAG]
+    bounds = [*starts, len(frames)]
+    segments = [frames[begin:end] for begin, end in itertools.pairwise(bounds)]
+    return [_armored(b"".join(_framed(tag, body) for tag, body in segment)) for segment in segments]
+
+
 def certificate_components(
     key: openpgp.composed.SignedPublicKey, *, at: datetime.datetime | None = None
 ) -> list[Component]:
@@ -158,6 +187,35 @@ def latest_self_signature_created_at(
     return datetime.datetime.fromtimestamp(created, tz=datetime.UTC)
 
 
+def merge_certificate_blocks(blocks: list[str]) -> str:
+    if not blocks:
+        raise ValueError("No certificate blocks to merge")
+    grouped = [_certificate_groups(_frames(_dearmored(block))) for block in blocks]
+    primary_bodies = {groups[0][0][1] for groups in grouped}
+    if len(primary_bodies) > 1:
+        raise ValueError("The blocks hold different primary keys")
+    primary = openpgp.packet.Packet.from_bytes(_framed(*grouped[0][0][0])).value
+    if not isinstance(primary, openpgp.packet.PublicKey):
+        raise ValueError("The primary packet is not a public key")
+    sections: dict[tuple[int, bytes], dict[tuple[int, bytes], None]] = {}
+    for groups in grouped:
+        for head, attached in groups:
+            merged = sections.setdefault(head, {})
+            for frame in attached:
+                merged.setdefault(frame)
+    fingerprint = primary.fingerprint.lower()
+    key_id = primary.key_id.lower()
+    heads = list(sections)
+    parts = [_emitted_section(heads[0], list(sections[heads[0]]), fingerprint, key_id)]
+    for tag in (_USER_ID_TAG, _USER_ATTRIBUTE_TAG, _PUBLIC_SUBKEY_TAG):
+        section_heads = [head for head in heads[1:] if head[0] == tag]
+        section_heads.sort(key=lambda head: _section_order(head, list(sections[head]), fingerprint, key_id))
+        parts.extend(_emitted_section(head, list(sections[head]), fingerprint, key_id) for head in section_heads)
+    data = b"".join(parts)
+    openpgp.composed.SignedPublicKey.from_bytes(data)
+    return _armored(data)
+
+
 def public_params_bits(public_params: openpgp.types.PublicParams) -> int | None:
     for bits in (public_params.rsa_bits, public_params.dsa_bits, public_params.curve_bits):
         if isinstance(bits, int):
@@ -205,7 +263,7 @@ def signing_key_facts(
             created=datetime.datetime.fromtimestamp(key.created_at, tz=datetime.UTC),
             expires=primary_expires,
             revoked=primary_revoked,
-            can_sign=_declares_signing(_effective_self_signature(key, timestamp)),
+            can_sign=primary.algorithm.can_sign() and _declares_signing(_effective_self_signature(key, timestamp)),
         )
     ]
 
@@ -227,7 +285,7 @@ def signing_key_facts(
                 created=datetime.datetime.fromtimestamp(subkey.key.created_at, tz=datetime.UTC),
                 expires=min(expirations) if expirations else None,
                 revoked=primary_revoked or _subkey_is_revoked(primary, subkey),
-                can_sign=_declares_signing(binding_signature),
+                can_sign=subkey.key.algorithm.can_sign() and _declares_signing(binding_signature),
             )
         )
     return facts
@@ -258,7 +316,7 @@ def signing_key_status(
             identified=True,
             fingerprint=subkey.key.fingerprint.lower(),
             expires=min(expirations) if expirations else None,
-            can_sign=_declares_signing(binding_signature),
+            can_sign=subkey.key.algorithm.can_sign() and _declares_signing(binding_signature),
             # Revoking the primary revokes everything beneath it, so a live subkey binding isn't enough
             revoked=primary_revoked or _subkey_is_revoked(primary, subkey),
         )
@@ -268,7 +326,7 @@ def signing_key_status(
             identified=True,
             fingerprint=key.fingerprint.lower(),
             expires=key_expires_at(key, at=at),
-            can_sign=_declares_signing(_effective_self_signature(key, timestamp)),
+            can_sign=primary.algorithm.can_sign() and _declares_signing(_effective_self_signature(key, timestamp)),
             revoked=primary_revoked,
         )
 
@@ -277,6 +335,12 @@ def signing_key_status(
     return SigningKeyStatus(identified=False, fingerprint=None, expires=None, can_sign=False, revoked=False)
 
 
+def _armored(data: bytes) -> str:
+    encoded = base64.b64encode(data).decode("ascii")
+    lines = [encoded[index : index + 64] for index in range(0, len(encoded), 64)]
+    return "\n".join([_ARMOR_BEGIN, "", *lines, _ARMOR_END]) + "\n"
+
+
 def _attribute_label(attribute: openpgp.packet.UserAttribute) -> str:
     kind = " ".join(part for part in (attribute.kind, attribute.image_format) if part)
     return f"{kind} ({len(attribute.data)} bytes)"
@@ -294,8 +358,11 @@ def _binding_revoked(
             or (signature.typ() == _CERTIFICATION_REVOCATION_SIGNATURE_TYPE)
         )
     ]
-    latest = _latest_signature(candidates, timestamp)
-    return (latest is not None) and (latest.typ() == _CERTIFICATION_REVOCATION_SIGNATURE_TYPE)
+    active = [signature for signature in candidates if _signature_active(signature, timestamp)]
+    if not active:
+        return False
+    latest = max(active, key=_revocation_preferring_order)
+    return latest.typ() == _CERTIFICATION_REVOCATION_SIGNATURE_TYPE
 
 
 def _binding_self_signatures(key: openpgp.composed.SignedPublicKey, timestamp: float) -> list[openpgp.packet.Signature]:
@@ -321,6 +388,60 @@ def _binding_self_signatures(key: openpgp.composed.SignedPublicKey, timestamp: f
     return binding_sigs
 
 
+def _certificate_groups(
+    frames: list[tuple[int, bytes]],
+) -> list[tuple[tuple[int, bytes], list[tuple[int, bytes]]]]:
+    if (not frames) or (frames[0][0] != _PRIMARY_KEY_TAG):
+        raise ValueError("The block does not begin with a primary key packet")
+    groups = [(frames[0], [])]
+    for frame in frames[1:]:
+        if frame[0] == _PRIMARY_KEY_TAG:
+            raise ValueError("The block holds more than one certificate")
+        if frame[0] in (_USER_ID_TAG, _USER_ATTRIBUTE_TAG, _PUBLIC_SUBKEY_TAG):
+            groups.append((frame, []))
+        elif frame[0] not in _SKIPPABLE_PACKET_TAGS:
+            groups[-1][1].append(frame)
+    return groups
+
+
+def _crc24(data: bytes) -> int:
+    crc = 0xB704CE
+    for octet in data:
+        crc ^= octet << 16
+        for _ in range(8):
+            crc <<= 1
+            if crc & 0x1000000:
+                crc ^= 0x1864CFB
+    return crc & 0xFFFFFF
+
+
+def _dearmored(text: str) -> bytes:
+    lines = []
+    checksum = None
+    in_data = False
+    for line in text.splitlines()[1:]:
+        stripped = line.strip()
+        if not in_data:
+            if stripped == "":
+                in_data = True
+            elif ":" not in stripped:
+                in_data = True
+                lines.append(stripped)
+            continue
+        if stripped.startswith("-----END"):
+            break
+        if stripped.startswith("="):
+            checksum = stripped[1:5]
+            continue
+        lines.append(stripped)
+    data = base64.b64decode("".join(lines), validate=True)
+    if checksum is not None:
+        declared = int.from_bytes(base64.b64decode(checksum, validate=True))
+        if declared != _crc24(data):
+            raise ValueError("Armor checksum mismatch")
+    return data
+
+
 def _declares_signing(signature: openpgp.packet.Signature | None) -> bool:
     # Every usable key carries a self-signature, so its absence means a key we can't read rather than
     # one which permits everything
@@ -353,6 +474,20 @@ def _direct_self_signatures(key: openpgp.composed.SignedPublicKey) -> list[openp
     ]
 
 
+def _earliest_self_certification(attached: list[tuple[int, bytes]], fingerprint: str, key_id: str) -> float:
+    created_values = []
+    for tag, body in attached:
+        if tag != _SIGNATURE_TAG:
+            continue
+        signature = _parsed_signature(body)
+        if (signature is None) or (signature.typ() not in _CERTIFICATION_SIGNATURE_TYPES):
+            continue
+        if not _signature_is_self(signature, fingerprint, key_id):
+            continue
+        created_values.append(signature.created() or 0)
+    return min(created_values) if created_values else float("inf")
+
+
 def _effective_self_signature(
     key: openpgp.composed.SignedPublicKey, timestamp: float
 ) -> openpgp.packet.Signature | None:
@@ -363,6 +498,73 @@ def _effective_self_signature(
     return binding if (binding is not None) else direct
 
 
+def _emitted_section(
+    head: tuple[int, bytes], attached: list[tuple[int, bytes]], fingerprint: str, key_id: str
+) -> bytes:
+    signatures: list[tuple[bytes, openpgp.packet.Signature]] = []
+    opaques: list[tuple[int, bytes]] = []
+    for tag, body in attached:
+        signature = _parsed_signature(body) if (tag == _SIGNATURE_TAG) else None
+        if signature is None:
+            opaques.append((tag, body))
+            continue
+        if _local_certification(signature):
+            continue
+        signatures.append((body, signature))
+    if head[0] == _PRIMARY_KEY_TAG:
+        revocations = [pair for pair in signatures if pair[1].typ() == _KEY_REVOCATION_SIGNATURE_TYPE]
+        others = [pair for pair in signatures if pair[1].typ() != _KEY_REVOCATION_SIGNATURE_TYPE]
+        ordered = sorted(revocations, key=_signature_order) + sorted(others, key=_signature_order)
+    else:
+        ordered = sorted(signatures, key=_signature_order)
+    frames = [head, *[(_SIGNATURE_TAG, body) for body, _ in ordered], *sorted(opaques)]
+    return b"".join(_framed(tag, body) for tag, body in frames)
+
+
+def _framed(tag: int, body: bytes) -> bytes:
+    length = len(body)
+    if length < 192:
+        header = bytes((0xC0 | tag, length))
+    elif length < 8384:
+        header = bytes((0xC0 | tag, ((length - 192) >> 8) + 192, (length - 192) & 0xFF))
+    else:
+        header = bytes((0xC0 | tag, 0xFF)) + length.to_bytes(4)
+    return header + body
+
+
+def _frames(data: bytes) -> list[tuple[int, bytes]]:
+    frames = []
+    offset = 0
+    while offset < len(data):
+        ctb = data[offset]
+        if not ctb & 0x80:
+            raise ValueError(f"Invalid packet marker {ctb:#x} at offset {offset}")
+        if ctb & 0x40:
+            tag = ctb & 0x3F
+            first = _octets(data, offset + 1, 1)[0]
+            if first < 192:
+                length, header = first, 2
+            elif first < 224:
+                length, header = ((first - 192) << 8) + _octets(data, offset + 2, 1)[0] + 192, 3
+            elif first == 255:
+                length, header = int.from_bytes(_octets(data, offset + 2, 4)), 6
+            else:
+                raise ValueError(f"Partial packet length at offset {offset}")
+        else:
+            tag = (ctb >> 2) & 0x0F
+            kind = ctb & 0x03
+            if kind == 3:
+                raise ValueError(f"Indeterminate packet length at offset {offset}")
+            sizes = {0: 1, 1: 2, 2: 4}[kind]
+            length, header = int.from_bytes(_octets(data, offset + 1, sizes)), 1 + sizes
+        body = data[offset + header : offset + header + length]
+        if len(body) != length:
+            raise ValueError(f"Truncated packet at offset {offset}")
+        frames.append((tag, bytes(body)))
+        offset += header + length
+    return frames
+
+
 def _identity_component(
     kind: ComponentKind,
     label: str,
@@ -433,7 +635,29 @@ def _latest_signature(signatures: list[openpgp.packet.Signature], timestamp: flo
     valid = [signature for signature in signatures if _signature_active(signature, timestamp)]
     if not valid:
         return None
-    return max(valid, key=lambda signature: signature.created() or 0)
+    return max(valid, key=lambda signature: (signature.created() or 0, signature.to_bytes()))
+
+
+def _local_certification(signature: openpgp.packet.Signature) -> bool:
+    typ = signature.typ()
+    if (typ not in _CERTIFICATION_SIGNATURE_TYPES) and (typ != _CERTIFICATION_REVOCATION_SIGNATURE_TYPE):
+        return False
+    return signature.exportable_certification() is False
+
+
+def _octets(data: bytes, offset: int, count: int) -> bytes:
+    taken = data[offset : offset + count]
+    if len(taken) != count:
+        raise ValueError(f"Truncated packet header at offset {offset}")
+    return taken
+
+
+def _parsed_signature(body: bytes) -> openpgp.packet.Signature | None:
+    try:
+        value = openpgp.packet.Packet.from_bytes(_framed(_SIGNATURE_TAG, body)).value
+    except Exception:
+        return None
+    return value if isinstance(value, openpgp.packet.Signature) else None
 
 
 def _primary_is_revoked(key: openpgp.composed.SignedPublicKey) -> bool:
@@ -446,6 +670,11 @@ def _primary_is_revoked(key: openpgp.composed.SignedPublicKey) -> bool:
     return False
 
 
+def _revocation_preferring_order(signature: openpgp.packet.Signature) -> tuple[float, bool, bytes]:
+    is_revocation = signature.typ() == _CERTIFICATION_REVOCATION_SIGNATURE_TYPE
+    return (signature.created() or 0, is_revocation, signature.to_bytes())
+
+
 def _revocation_verifies(signature: openpgp.packet.Signature, primary: openpgp.packet.PublicKey) -> bool:
     # A revocation only counts once we've checked the primary key actually made it. An unsigned or
     # forged revocation packet must not block a key, or anyone could deny a project its releases just
@@ -459,6 +688,21 @@ def _revocation_verifies(signature: openpgp.packet.Signature, primary: openpgp.p
     return True
 
 
+def _section_order(
+    head: tuple[int, bytes], attached: list[tuple[int, bytes]], fingerprint: str, key_id: str
+) -> tuple[float, bytes]:
+    tag, body = head
+    if tag != _PUBLIC_SUBKEY_TAG:
+        return (_earliest_self_certification(attached, fingerprint, key_id), body)
+    try:
+        value = openpgp.packet.Packet.from_bytes(_framed(tag, body)).value
+    except Exception:
+        return (float("inf"), body)
+    if not isinstance(value, openpgp.packet.PublicSubkey):
+        return (float("inf"), body)
+    return (value.created_at, body)
+
+
 def _signature_active(signature: openpgp.packet.Signature, timestamp: float) -> bool:
     created = signature.created() or 0
     if created > timestamp:
@@ -480,6 +724,11 @@ def _signature_is_self(signature: openpgp.packet.Signature, self_fingerprint: st
     return (self_fingerprint in fingerprints) or (self_key_id in key_ids)
 
 
+def _signature_order(pair: tuple[bytes, openpgp.packet.Signature]) -> tuple[float, str, bytes]:
+    body, signature = pair
+    return (signature.created() or 0, signature.typ() or "", body)
+
+
 def _subkey_expires_at(
     subkey: openpgp.composed.SignedPublicSubKey,
     signature: openpgp.packet.Signature | None,
diff --git a/atr/tasks/checks/signature.py b/atr/tasks/checks/signature.py
index a483c56e..026f6780 100644
--- a/atr/tasks/checks/signature.py
+++ b/atr/tasks/checks/signature.py
@@ -41,7 +41,7 @@ INPUT_EXTRA_ARGS: Final[list[str]] = [
     "unsuffixed_file_hash",
     "unsuffixed_file_uploaders",
 ]
-CHECK_VERSION: Final[str] = "11"
+CHECK_VERSION: Final[str] = "12"
 
 
 async def check(args: checks.FunctionArguments) -> results.Results | None:
diff --git a/tests/unit/pgp_fixtures.py b/tests/unit/pgp_fixtures.py
index 58fb7e5d..d41c982c 100644
--- a/tests/unit/pgp_fixtures.py
+++ b/tests/unit/pgp_fixtures.py
@@ -233,6 +233,58 @@ ixdipMf9A5HgGyWO0mNWIGk+ZRIH
 
 # A certificate with a revoked secondary user id. The key itself stays entirely valid: reading the
 # uid revocation as a key-level self-signature would erase its expiry and its declared capabilities
+ALL_UIDS_REVOKED_PUBLIC_KEY_ASC: Final[str] = """-----BEGIN PGP PUBLIC KEY BLOCK-----
+
+xjMEXgvhABYJKwYBBAHaRw8BAQdAmbGDg3AS7oD3hJHRMiqyl66UyEtAAhkFDuZR
++I8CY/7NPUFUUiBSZWNlcnQgVHdvIChGb3IgdGVzdCB1c2Ugb25seSkgPHJlY2Vy
+dC0yQGV4YW1wbGUuaW52YWxpZD7CkwQTFgoAOxYhBLbgBR68+9U0Pk2D37Vu607g
+vPwlBQJeC+EAAhsBBQsJCAcCAiICBhUKCQgLAgQWAgMBAh4HAheAAAoJELVu607g
+vPwlep4A/1efZVjLbaTYw33bfHVAc1XRiF/I2WD7jjZNmT3D42HEAP4ulBMFnq0+
+SBRtf05r+PFN6qo64l4ssyR22SqeAD6zBsJ4BDAWCgAgFiEEtuAFHrz71TQ+TYPf
+tW7rTuC8/CUFAl/uZgACHSAACgkQtW7rTuC8/CXjJwEAooBNfv0rpklQL0Qg3GGL
+bAULW8LDtveiOg6rh5o7gEoA/R3f6oCT9wbJKlmq4xal8bc9SSpqtoVlcW6iPONl
+gn0O
+-----END PGP PUBLIC KEY BLOCK-----
+"""
+
+RECERTIFIED_UID_PUBLIC_KEY_ASC: Final[str] = """-----BEGIN PGP PUBLIC KEY BLOCK-----
+
+xjMEXgvhABYJKwYBBAHaRw8BAQdAmbGDg3AS7oD3hJHRMiqyl66UyEtAAhkFDuZR
++I8CY/7NPUFUUiBSZWNlcnQgT25lIChGb3IgdGVzdCB1c2Ugb25seSkgPHJlY2Vy
+dC0xQGV4YW1wbGUuaW52YWxpZD7CkwQTFgoAOxYhBLbgBR68+9U0Pk2D37Vu607g
+vPwlBQJeC+EAAhsBBQsJCAcCAiICBhUKCQgLAgQWAgMBAh4HAheAAAoJELVu607g
+vPwlh04BAODwhPX8utN5+pKDDt5QYES3hzPdY9c63Qsu3K6F+yr9AQDPVFCOwrXW
+yos8KFwpfcCKdaR2Smbg4KK3Be65Qf23Cs09QVRSIFJlY2VydCBUd28gKEZvciB0
+ZXN0IHVzZSBvbmx5KSA8cmVjZXJ0LTJAZXhhbXBsZS5pbnZhbGlkPsKTBBMWCgA7
+FiEEtuAFHrz71TQ+TYPftW7rTuC8/CUFAl4L4QACGwEFCwkIBwICIgIGFQoJCAsC
+BBYCAwECHgcCF4AACgkQtW7rTuC8/CV6ngD/V59lWMttpNjDfdt8dUBzVdGIX8jZ
+YPuONk2ZPcPjYcQA/i6UEwWerT5IFG1/Tmv48U3qqjriXiyzJHbZKp4APrMGwngE
+MBYKACAWIQS24AUevPvVND5Ng9+1butO4Lz8JQUCX+5mAAIdIAAKCRC1butO4Lz8
+JeMnAQCigE1+/SumSVAvRCDcYYtsBQtbwsO296I6DquHmjuASgD9Hd/qgJP3Bskq
+WarjFqXxtz1JKmq2hWVxbqI842WCfQ7CkwQTFgoAOxYhBLbgBR68+9U0Pk2D37Vu
+607gvPwlBQJhz5mAAhsBBQsJCAcCAiICBhUKCQgLAgQWAgMBAh4HAheAAAoJELVu
+607gvPwlz7wA/R5oMWr85w1+aPq/2lWgMX8gPjx+0uugqSzlTFpMGf3/AQDzgPjC
+v/kThixelh2qISyZgqlnxyHzf8K8nAvHHSobBA==
+-----END PGP PUBLIC KEY BLOCK-----
+"""
+
+LOCAL_CERTIFICATION_FINGERPRINT: Final[str] = "34a7704ef4a3760d556e53867362b716c7acd9f6"
+
+LOCAL_CERTIFICATION_PUBLIC_KEY_ASC: Final[str] = """-----BEGIN PGP PUBLIC KEY BLOCK-----
+
+mDMEaoccdBYJKwYBBAHaRw8BAQdAz8SrbRVLTsholjimiU4SQeEeSq4jSsu8wrEh
+QaKTQWC0REFUUiBMb2NhbCBTaWcgVGFyZ2V0IChGb3IgdGVzdCB1c2Ugb25seSkg
+PGxvY2FsLXNpZ0BleGFtcGxlLmludmFsaWQ+iJMEExYKADsWIQQ0p3BO9KN2DVVu
+U4ZzYrcWx6zZ9gUCaoccdAIbAQULCQgHAgIiAgYVCgkICwIEFgIDAQIeBwIXgAAK
+CRBzYrcWx6zZ9i9sAP48Z7BlHvq5g516KBkXnl21I+7hcRUdGZaVn8fJYTiZpQEA
+8tRs+szgi5RcQVSuw5/N0f0KwM3FGC+GB9XHT+SeiwOIeAQQFgoAIBYhBCv+BPRX
+tnN+CLbd4I99cc259PeTBQJqhxx1AgQAAAoJEI99cc259PeT57cA/1tWd7RyPVyY
+Vx2MB/LkcbDoOUE+qqmlawoFvK2d3kxRAQDvfCK3GyFsgaU1IBrSPCIXACV8m8he
+JxoCla67csUuAA==
+=SanX
+-----END PGP PUBLIC KEY BLOCK-----
+"""
+
 REVOKED_UID_FINGERPRINT: Final[str] = "346c2e8cd37b4f7c0bde03d9fdefe91a648ddd76"
 REVOKED_UID_SIGNING_FINGERPRINT: Final[str] = "5aec39f36f29b0d02ac67eb6dd6e33ceb5d275a6"
 REVOKED_UID_PRIMARY_EXPIRES_YEAR: Final[int] = 2030
diff --git a/tests/unit/test_pgp.py b/tests/unit/test_pgp.py
index 2d1069ec..eb86f597 100644
--- a/tests/unit/test_pgp.py
+++ b/tests/unit/test_pgp.py
@@ -77,6 +77,18 @@ def test_certificate_components_lists_each_part_with_its_signatures() -> None:
     assert components[3].facts.fingerprint == pgp_fixtures.REVOKED_UID_SIGNING_FINGERPRINT
 
 
+def test_certificate_components_revives_a_recertified_uid() -> None:
+    key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.RECERTIFIED_UID_PUBLIC_KEY_ASC)
+    while_revoked = datetime.datetime(2021, 6, 1, tzinfo=datetime.UTC)
+    after_recertification = datetime.datetime(2023, 1, 1, tzinfo=datetime.UTC)
+
+    revoked_then = [c.revoked for c in pgp.certificate_components(key, at=while_revoked) if c.kind == "user-id"]
+    revoked_now = [c.revoked for c in pgp.certificate_components(key, at=after_recertification) if c.kind == "user-id"]
+
+    assert revoked_then == [False, True]
+    assert revoked_now == [False, False]
+
+
 def test_certificate_for_fingerprint_selects_the_named_certificate() -> None:
     block = pgp_fixtures.two_certificate_block(
         pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC, pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC
@@ -116,6 +128,48 @@ def test_latest_self_signature_survives_a_revoked_primary_uid() -> None:
     assert latest.typ() == "cert-positive"
 
 
+def test_merge_certificate_blocks_drops_local_certifications() -> None:
+    merged = pgp.merge_certificate_blocks([pgp_fixtures.LOCAL_CERTIFICATION_PUBLIC_KEY_ASC])
+
+    signatures = [pgp._parsed_signature(body) for tag, body in pgp._frames(pgp._dearmored(merged)) if tag == 2]
+
+    assert [signature.typ() for signature in signatures if signature is not None] == ["cert-positive"]
+
+
+def test_merge_certificate_blocks_keeps_unknown_packets() -> None:
+    block = pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC
+    frames = pgp._frames(pgp._dearmored(block))
+    spliced = [*frames[:2], (60, b"experimental"), *frames[2:]]
+    variant = pgp._armored(b"".join(pgp._framed(tag, body) for tag, body in spliced))
+
+    merged = pgp.merge_certificate_blocks([variant, block])
+
+    assert (60, b"experimental") in pgp._frames(pgp._dearmored(merged))
+    assert pgp.merge_certificate_blocks([merged]) == merged
+
+
+def test_merge_certificate_blocks_refuses_different_primaries() -> None:
+    blocks = [pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC, pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC]
+
+    with pytest.raises(ValueError, match="different primary keys"):
+        pgp.merge_certificate_blocks(blocks)
+
+
+def test_merge_certificate_blocks_unions_and_commutes() -> None:
+    block = pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC
+    canonical = pgp.merge_certificate_blocks([block])
+    frames = pgp._frames(pgp._dearmored(block))
+    without_revocation = pgp._armored(b"".join(pgp._framed(tag, body) for tag, body in frames[:4] + frames[5:]))
+    without_subkey = pgp._armored(b"".join(pgp._framed(tag, body) for tag, body in frames[:6]))
+
+    forward = pgp.merge_certificate_blocks([without_revocation, without_subkey])
+    reverse = pgp.merge_certificate_blocks([without_subkey, without_revocation])
+    folded = pgp.merge_certificate_blocks([forward, block])
+
+    assert pgp.merge_certificate_blocks([canonical]) == canonical
+    assert (forward == reverse) and (folded == canonical)
+
+
 def test_reference_time_excludes_future_signatures() -> None:
     key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
     before = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC)
@@ -132,6 +186,14 @@ def test_reference_time_excludes_future_signatures() -> None:
     assert (expires is not None) and (expires.year == pgp_fixtures.REVOKED_UID_PRIMARY_EXPIRES_YEAR)
 
 
+def test_signing_key_facts_conservative_when_every_uid_is_revoked() -> None:
+    key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.ALL_UIDS_REVOKED_PUBLIC_KEY_ASC)
+
+    facts = pgp.signing_key_facts(key, at=datetime.datetime(2023, 1, 1, tzinfo=datetime.UTC))
+
+    assert (facts[0].can_sign, facts[0].expires) == (False, None)
+
+
 def test_signing_key_status_expiry_follows_the_issuing_subkey() -> None:
     key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
 


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org