You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by ar...@apache.org on 2026/07/21 08:50:40 UTC

(tooling-trusted-releases) 01/02: #1417 - Reject on revoked

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

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

commit 93bab29eedeb2f7eab34ddd89f12fec24b7e1a39
Author: Alastair McFarlane <ar...@apache.org>
AuthorDate: Fri Jul 17 17:17:14 2026 +0100

    #1417 - Reject on revoked
---
 atr/pgp.py                          |  26 ++++++++-
 atr/post/keys.py                    |   4 ++
 atr/storage/datatypes.py            |   2 +
 atr/storage/writers/keys.py         |  68 +++++++++++++++++++++--
 atr/tasks/checks/__init__.py        |  18 +++++--
 atr/tasks/checks/signature.py       |  35 +++++++++++-
 tests/unit/pgp_fixtures.py          | 105 ++++++++++++++++++++++++++++++++++++
 tests/unit/test_checks_signature.py |  16 ++++++
 tests/unit/test_keys_delete.py      |  38 ++++++++++++-
 tests/unit/test_pgp.py              |  54 +++++++++++++++++++
 10 files changed, 356 insertions(+), 10 deletions(-)

diff --git a/atr/pgp.py b/atr/pgp.py
index 98587e92..14e9d850 100644
--- a/atr/pgp.py
+++ b/atr/pgp.py
@@ -22,10 +22,14 @@ from typing import Final
 import openpgp
 
 _CERTIFICATION_REVOCATION_SIGNATURE_TYPE: Final[str] = "cert-revocation"
+# The self-certifications which carry key expiry and capabilities. A revocation sits alongside these
+# on a user binding, but declares neither, so it can't stand in for one
 _CERTIFICATION_SIGNATURE_TYPES: Final[frozenset[str]] = frozenset(
     {"cert-generic", "cert-persona", "cert-casual", "cert-positive"}
 )
+_KEY_REVOCATION_SIGNATURE_TYPE: Final[str] = "key-revocation"
 _SUBKEY_BINDING_SIGNATURE_TYPE: Final[str] = "subkey-binding"
+_SUBKEY_REVOCATION_SIGNATURE_TYPE: Final[str] = "subkey-revocation"
 
 
 @dataclasses.dataclass(frozen=True)
@@ -36,6 +40,8 @@ class SigningKeyStatus:
     expires: datetime.datetime | None
     # False only when the issuing key declares capabilities which exclude signing
     can_sign: bool
+    # True where the issuing key carries a revocation, or hangs beneath a primary which does
+    revoked: bool
 
 
 def key_expires_at(key: openpgp.PublicKey) -> datetime.datetime | None:
@@ -67,6 +73,7 @@ def signing_key_status(
     issuer_key_ids: set[str],
 ) -> SigningKeyStatus:
     """Validity of the key which issued a signature, be that the primary key or one of its subkeys."""
+    primary_revoked = _primary_is_revoked(key)
     binding = _issuing_subkey_binding(key, issuer_fingerprints, issuer_key_ids)
     if binding is not None:
         # A subkey is only usable while its own binding and the primary above it both hold
@@ -80,6 +87,8 @@ def signing_key_status(
             identified=True,
             expires=min(expirations) if expirations else None,
             can_sign=_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(binding),
         )
 
     if _issuer_is_primary(key, issuer_fingerprints, issuer_key_ids):
@@ -87,11 +96,12 @@ def signing_key_status(
             identified=True,
             expires=key_expires_at(key),
             can_sign=_declares_signing(_effective_self_signature(key)),
+            revoked=primary_revoked,
         )
 
     # Naming no key we hold is not the same as naming the primary, and a signature we can't attribute
     # is one we can't judge, so report it as such rather than borrowing the primary's answer
-    return SigningKeyStatus(identified=False, expires=None, can_sign=False)
+    return SigningKeyStatus(identified=False, expires=None, can_sign=False, revoked=False)
 
 
 def _binding_revoked(binding: openpgp.UserBindingInfo, fingerprint: str, key_id: str) -> bool:
@@ -122,6 +132,8 @@ def _binding_self_signatures(key: openpgp.PublicKey) -> list[openpgp.SignatureIn
             signature
             for signature in binding.signatures
             if _signature_is_self(signature, self_fingerprint, self_key_id)
+            # A cert-revocation is self-issued too, but carries no expiry or key flags, so admitting it
+            # here lets a revoked user id erase both for the whole key
             and (signature.signature_type in _CERTIFICATION_SIGNATURE_TYPES)
         )
     return binding_sigs
@@ -212,6 +224,12 @@ def _signature_expired(signature: openpgp.SignatureInfo, now: float) -> bool:
     return (creation + expiration) <= now
 
 
+def _primary_is_revoked(key: openpgp.PublicKey) -> bool:
+    return any(
+        signature.signature_type == _KEY_REVOCATION_SIGNATURE_TYPE for signature in key.revocation_signature_infos()
+    )
+
+
 def _signature_is_self(signature: openpgp.SignatureInfo, self_fingerprint: str, self_key_id: str) -> bool:
     fingerprints = {fingerprint.lower() for fingerprint in signature.issuer_fingerprints}
     key_ids = {key_id.lower() for key_id in signature.issuer_key_ids}
@@ -229,3 +247,9 @@ def _subkey_expires_at(
         return None
     # Subkey expiration counts from the subkey's own creation, not the primary's
     return datetime.datetime.fromtimestamp(binding.created_at + key_expiration_seconds, tz=datetime.UTC)
+
+
+def _subkey_is_revoked(binding: openpgp.SubkeyBindingInfo) -> bool:
+    # The revocation can share a timestamp with the binding it kills, so this asks whether one exists
+    # at all rather than which of the two is more recent
+    return any(signature.signature_type == _SUBKEY_REVOCATION_SIGNATURE_TYPE for signature in binding.signatures)
diff --git a/atr/post/keys.py b/atr/post/keys.py
index 10475e9e..6f4b3f26 100644
--- a/atr/post/keys.py
+++ b/atr/post/keys.py
@@ -90,6 +90,10 @@ async def add(
                     ".",
                 ]
                 await quart.flash(str(p), "warning")
+            elif key.status == datatypes.KeyStatus.REFRESHED:
+                await quart.flash(
+                    f"OpenPGP key {fingerprint_upper} was updated to the version you uploaded.", "success"
+                )
             else:
                 await quart.flash(f"OpenPGP key {fingerprint_upper} added successfully.", "success")
             await _flash_openpgp_key_uid_warning(key.key_model, session.asf_uid)
diff --git a/atr/storage/datatypes.py b/atr/storage/datatypes.py
index faf7b55c..fc39a5d9 100644
--- a/atr/storage/datatypes.py
+++ b/atr/storage/datatypes.py
@@ -46,6 +46,8 @@ class KeyStatus(enum.Flag):
     PARSED = 0
     INSERTED = enum.auto()
     LINKED = enum.auto()
+    # The fingerprint was already stored, but its key block was replaced with a newer one
+    REFRESHED = enum.auto()
     INSERTED_AND_LINKED = INSERTED | LINKED
 
 
diff --git a/atr/storage/writers/keys.py b/atr/storage/writers/keys.py
index 7839193e..4a244427 100644
--- a/atr/storage/writers/keys.py
+++ b/atr/storage/writers/keys.py
@@ -479,11 +479,16 @@ class FoundationCommitter(GeneralPublic):
         )
         inserted = key_insert_result.one_or_none() is not None
         undeleted = []
+        refreshed = False
         if inserted:
             await self.__signature_hints_consume([key])
         else:
+            # The fingerprint only covers the primary packet, so an existing row can hold a stale
+            # block - a since-revoked, re-signed, or re-subkeyed export of the same key. Bring it up
+            # to date, which also carries the apache_uid the undelete below would otherwise set
+            refreshed = await self.__refresh_stored_key(key.key_model)
             undeleted = await self.__undelete_keys([key.key_model.fingerprint])
-            if undeleted:
+            if undeleted and not refreshed:
                 await self.__data.execute(
                     sqlmodel.update(sql.PublicSigningKey)
                     .where(via(sql.PublicSigningKey.fingerprint) == key.key_model.fingerprint)
@@ -502,6 +507,19 @@ class FoundationCommitter(GeneralPublic):
             undeleted_key = datatypes.Key(status=datatypes.KeyStatus.INSERTED, key_model=key.key_model)
             return outcome.Result(undeleted_key), publications
 
+        if refreshed:
+            self.__write_as.append_to_audit_log(
+                action="key_refresh",
+                asf_uid=self.__asf_uid,
+                fingerprint=key.key_model.fingerprint,
+                key_apache_uid=key.key_model.apache_uid,
+            )
+            # A changed block can flip an existing signature check, so let its committees catch up
+            publications = await self.__sync_committees_for_keys([key.key_model.fingerprint])
+            log.info(f"Refreshed stored key {key.key_model.fingerprint}")
+            refreshed_key = datatypes.Key(status=datatypes.KeyStatus.REFRESHED, key_model=key.key_model)
+            return outcome.Result(refreshed_key), publications
+
         if not inserted:
             log.info(f"Key {key.key_model.fingerprint} already exists in database")
             return outcome.Result(datatypes.Key(status=datatypes.KeyStatus.PARSED, key_model=key.key_model)), {}
@@ -591,6 +609,21 @@ and was published by the committee.\
         full_keys_file_content = header_content + key_blocks_str
         return full_keys_file_content
 
+    async def __refresh_stored_key(self, model: sql.PublicSigningKey) -> bool:
+        existing = await self.__data.public_signing_key(fingerprint=model.fingerprint, deleted=db.NOT_SET).get()
+        if (existing is None) or (existing.ascii_armored_key == model.ascii_armored_key):
+            return False
+        existing.ascii_armored_key = model.ascii_armored_key
+        existing.algorithm = model.algorithm
+        existing.length = model.length
+        existing.latest_self_signature = model.latest_self_signature
+        existing.expires = model.expires
+        existing.primary_declared_uid = model.primary_declared_uid
+        existing.secondary_declared_uids = model.secondary_declared_uids
+        existing.apache_uid = model.apache_uid
+        self.__data.add(existing)
+        return True
+
     async def __signature_hints_consume(self, keys: list[datatypes.Key]) -> list[str]:
         via = sql.validate_instrumented_attribute
         member_map = {key.key_model.fingerprint: set(key.member_ids) for key in keys}
@@ -886,11 +919,40 @@ class CommitteeParticipant(FoundationCommitter):
         committee = await self.committee()
 
         key_values = [key.key_model.model_dump(exclude={"committees"}) for key in key_list]
+        incoming_blocks = {v["fingerprint"]: v["ascii_armored_key"] for v in key_values}
+        stored_blocks: dict[str, str] = {}
+        if incoming_blocks:
+            stored_result = await self.__data.execute(
+                sqlmodel.select(
+                    via(sql.PublicSigningKey.fingerprint),
+                    via(sql.PublicSigningKey.ascii_armored_key),
+                ).where(via(sql.PublicSigningKey.fingerprint).in_(sorted(incoming_blocks)))
+            )
+            stored_blocks = {fingerprint: block for fingerprint, block in stored_result.all()}
+        # A key we already hold whose incoming block differs has been re-imported with a newer state,
+        # which can flip an existing signature check even when nothing about the association changed
+        refreshed_fingerprints = {
+            fingerprint
+            for fingerprint, block in incoming_blocks.items()
+            if (fingerprint in stored_blocks) and (stored_blocks[fingerprint] != block)
+        }
         if key_values:
             stmt = sqlite.insert(sql.PublicSigningKey).values(key_values)
+            # The fingerprint only covers the primary packet, so a re-import can carry a newer block
+            # for a key we already hold - a revocation, an extended expiry, a fresh subkey. Refresh
+            # the whole derived row, not just the apache_uid
             stmt = stmt.on_conflict_do_update(
                 index_elements=["fingerprint"],
-                set_={"apache_uid": stmt.excluded.apache_uid},
+                set_={
+                    "apache_uid": stmt.excluded.apache_uid,
+                    "ascii_armored_key": stmt.excluded.ascii_armored_key,
+                    "algorithm": stmt.excluded.algorithm,
+                    "length": stmt.excluded.length,
+                    "latest_self_signature": stmt.excluded.latest_self_signature,
+                    "expires": stmt.excluded.expires,
+                    "primary_declared_uid": stmt.excluded.primary_declared_uid,
+                    "secondary_declared_uids": stmt.excluded.secondary_declared_uids,
+                },
             )
             key_insert_result = await self.__data.execute(
                 stmt.returning(via(sql.PublicSigningKey.fingerprint)),
@@ -960,7 +1022,7 @@ class CommitteeParticipant(FoundationCommitter):
                 fingerprints=[f for f in undeleted],
             )
             publications = await self.__sync_committees_for_keys(undeleted)
-        if link_inserts:
+        if link_inserts or refreshed_fingerprints:
             await self._recheck_committee_drafts(self.__committee_key)
         return outcomes, publications
 
diff --git a/atr/tasks/checks/__init__.py b/atr/tasks/checks/__init__.py
index 99b410d8..835769a5 100644
--- a/atr/tasks/checks/__init__.py
+++ b/atr/tasks/checks/__init__.py
@@ -480,7 +480,10 @@ async def _resolve_committee_signing_keys(release: sql.Release, rel_path: str |
     committee_key = release.committee.key
     async with db.session() as data:
         statement = (
-            sqlmodel.select(via(sql.KeyLink.key_fingerprint))
+            sqlmodel.select(
+                via(sql.KeyLink.key_fingerprint),
+                via(sql.PublicSigningKey.ascii_armored_key),
+            )
             .join(sql.PublicSigningKey)
             .where(
                 via(sql.KeyLink.committee_key) == committee_key,
@@ -488,8 +491,17 @@ async def _resolve_committee_signing_keys(release: sql.Release, rel_path: str |
             )
         )
         result = await data.execute(statement)
-        fingerprints = result.scalars().all()
-    return sorted(fp for fp in fingerprints if fp)
+        rows = result.all()
+
+    # A revocation or expiry change leaves the fingerprint alone but rewrites the key block, so the
+    # block content is what the cache has to track to know the check needs re-running
+    entries = []
+    for fingerprint, block in rows:
+        if not fingerprint:
+            continue
+        text = block if isinstance(block, str) else block.decode("utf-8", errors="replace")
+        entries.append(f"{fingerprint}:{text}")
+    return sorted(entries)
 
 
 async def _resolve_cross_format_sibling_swhids(release: sql.Release, rel_path: str | None = None) -> list[str]:
diff --git a/atr/tasks/checks/signature.py b/atr/tasks/checks/signature.py
index f47902fb..2c9a22be 100644
--- a/atr/tasks/checks/signature.py
+++ b/atr/tasks/checks/signature.py
@@ -34,7 +34,7 @@ import atr.util as util
 # Release policy fields which this check relies on - used for result caching
 INPUT_POLICY_KEYS: Final[list[str]] = []
 INPUT_EXTRA_ARGS: Final[list[str]] = ["committee_key", "committee_signing_keys", "unsuffixed_file_hash"]
-CHECK_VERSION: Final[str] = "5"
+CHECK_VERSION: Final[str] = "6"
 
 
 async def check(args: checks.FunctionArguments) -> results.Results | None:
@@ -342,6 +342,13 @@ def _key_refusal_result(
             num_committee_keys=num_committee_keys,
             key_has_apache_uid=key_has_apache_uid,
         )
+    if status.revoked:
+        return _revoked_key_result(
+            matched_key=matched_key,
+            signature_info=signature_info,
+            num_committee_keys=num_committee_keys,
+            key_has_apache_uid=key_has_apache_uid,
+        )
     if (status.expires is not None) and (status.expires <= datetime.datetime.now(datetime.UTC)):
         return _expired_key_result(
             matched_key=matched_key,
@@ -398,6 +405,32 @@ def _parse_public_keys(ascii_armored_keys: list[str]) -> list[openpgp.PublicKey]
     return public_keys
 
 
+def _revoked_key_result(
+    *,
+    matched_key: openpgp.PublicKey,
+    signature_info: openpgp.SignatureInfo,
+    num_committee_keys: int,
+    key_has_apache_uid: bool,
+) -> dict[str, Any]:
+    debug_info = _debug_info(
+        key=matched_key,
+        signature_info=signature_info,
+        status="Invalid: Signing key revoked",
+        valid=False,
+        num_committee_keys=num_committee_keys,
+        key_has_apache_uid=key_has_apache_uid,
+    )
+    return {
+        "verified": False,
+        "error": "Signature was made by a revoked key",
+        "error_kind": "revoked_key",
+        "hint": "Sign the artifact with a key which has not been revoked, then",
+        "hint_link_text": "check your keys",
+        "hint_link": "/keys",
+        "debug_info": debug_info,
+    }
+
+
 def _signer_username(key: openpgp.PublicKey, signature_info: openpgp.SignatureInfo) -> str | None:
     if signature_info.signer_user_id:
         return signature_info.signer_user_id
diff --git a/tests/unit/pgp_fixtures.py b/tests/unit/pgp_fixtures.py
index 38afa213..01b521e8 100644
--- a/tests/unit/pgp_fixtures.py
+++ b/tests/unit/pgp_fixtures.py
@@ -108,3 +108,108 @@ def write_expired_subkey_fixture(tmp_path: pathlib.Path) -> tuple[str, str]:
     artifact_path.write_bytes(EXPIRED_SUBKEY_ARTIFACT_BYTES)
     signature_path.write_text(EXPIRED_SUBKEY_DETACHED_SIGNATURE_ASC, encoding="utf-8")
     return str(signature_path), str(artifact_path)
+
+
+# A certificate whose signing subkey was revoked after it had signed. The signature below was made
+# while the subkey was still good, so only a check which reads the revocation will reject it
+REVOKED_SUBKEY_PRIMARY_FINGERPRINT: Final[str] = "627e587f1f3f8cd99cc8c6781e02faf30b9b2d9d"
+REVOKED_SUBKEY_SIGNING_FINGERPRINT: Final[str] = "5b024ae4e7690e808c1c3a18c13114edca77d92f"
+
+REVOKED_SUBKEY_ARTIFACT_BYTES: Final[bytes] = b"atr revocation fixture\n"
+
+REVOKED_SUBKEY_PUBLIC_KEY_ASC: Final[str] = """-----BEGIN PGP PUBLIC KEY BLOCK-----
+
+mDMEalpBFhYJKwYBBAHaRw8BAQdAjDlZ6Lr4uf3ylu8ozNd5ZeyG9dPPeG/vHH6l
+QVRSW+q0R0FUUiBSZXZva2VkIFN1YmtleSAoRm9yIHRlc3QgdXNlIG9ubHkpIDxy
+ZXZva2VkLXN1YmtleUBleGFtcGxlLmludmFsaWQ+iJMEExYKADsWIQRiflh/Hz+M
+2ZzIxngeAvrzC5stnQUCalpBFgIbAQULCQgHAgIiAgYVCgkICwIEFgIDAQIeBwIX
+gAAKCRAeAvrzC5stnfdAAQCDqDG6MbixdM9Vxsouxg3IeM3xSPH00uv8zs7ds9je
+NgD+OuBJ81xTQF2Nud+aZO9QPbr8JVRebjDi517pMiE+kA64MwRqWkEWFgkrBgEE
+AdpHDwEBB0DYrVxiE6+rIZt5WhehU+Y1ZgjSUmS4LUjQBeBxcs78u4h4BCgWCgAg
+FiEEYn5Yfx8/jNmcyMZ4HgL68wubLZ0FAmpaQRYCHQIACgkQHgL68wubLZ21oQEA
+2POrR1Ff0xJik+0M17V8nu8zCagAauhiM9S7nKD/5bgBAI/pD0d08UMLGqDFzIO7
+GlxonRgJsv+LcRh0Mw5yyVUEiO8EGBYKACAWIQRiflh/Hz+M2ZzIxngeAvrzC5st
+nQUCalpBFgIbAgCBCRAeAvrzC5stnXYgBBkWCgAdFiEEWwJK5OdpDoCMHDoYwTEU
+7cp32S8FAmpaQRYACgkQwTEU7cp32S8PsQD+P1bArcgoslNBVjKS1WvxmO7LQv+1
+/xymuXAvvVyIARsA/2bv6hKQMYhqrGERZ7ptUh9z6x15tgLGiS3UMr5FSW0LepYB
+AJOv44nvZhSS7FEFKEVUDA8wFXMkN8b0vONjttnOSVqQAP9aONCFDVFZHDwxhXe2
+Ai4vJkztXwDtM9aamALxT+OrDw==
+=2c/y
+-----END PGP PUBLIC KEY BLOCK-----
+"""
+
+REVOKED_SUBKEY_DETACHED_SIGNATURE_ASC: Final[str] = """-----BEGIN PGP SIGNATURE-----
+
+iHUEABYKAB0WIQRbAkrk52kOgIwcOhjBMRTtynfZLwUCalpBFgAKCRDBMRTtynfZ
+L96xAP9O/POM9QlwStFscv/Ylj2AcGmnRIipcdYtG63grTTgWwD8C0d+wPULT/oY
+9gkQVkwcs5X4x8x+qKKvLLQHAgX3jgg=
+=wOC7
+-----END PGP SIGNATURE-----
+"""
+
+
+# A certificate whose primary key is revoked. Revoking the primary revokes everything beneath it, so
+# its still-bound signing subkey is unusable too
+REVOKED_PRIMARY_FINGERPRINT: Final[str] = "894e4b57dca9772b6b692b6a1d725675acbdb724"
+REVOKED_PRIMARY_SIGNING_FINGERPRINT: Final[str] = "d52d750d3e80bede5553b2edaf0a1d93538cd91e"
+
+REVOKED_PRIMARY_PUBLIC_KEY_ASC: Final[str] = """-----BEGIN PGP PUBLIC KEY BLOCK-----
+
+mDMEalpBFhYJKwYBBAHaRw8BAQdAKUdQxIFXyT1Nn153YCA/poF2sSjombcngmqF
+0z8FAxyIeAQgFgoAIBYhBIlOS1fcqXcra2krah1yVnWsvbckBQJqWkEWAh0AAAoJ
+EB1yVnWsvbck1/sBAJPxFkfg+Cz8+RLXieMId3W7z1ySbAY0ZXdW0+zIXTs6AP0V
+lbtW9cFIu8fpMPQxw7ghJAiT/NcLewQTSdc1QfZ8DbRJQVRSIFJldm9rZWQgUHJp
+bWFyeSAoRm9yIHRlc3QgdXNlIG9ubHkpIDxyZXZva2VkLXByaW1hcnlAZXhhbXBs
+ZS5pbnZhbGlkPoiTBBMWCgA7FiEEiU5LV9ypdytraStqHXJWday9tyQFAmpaQRYC
+GwEFCwkIBwICIgIGFQoJCAsCBBYCAwECHgcCF4AACgkQHXJWday9tyRzUAEA8dxQ
+eXeLraUvbKmTwArK4EBez/kfb/iAw3SjhcNzlQgBAK7rYpI/2O7lCEJ/Ibhom9cI
+CcY0CEWd4CB148F+bEQMuDMEalpBFhYJKwYBBAHaRw8BAQdAWTKqGwULwFLUhP2f
+plzat/55UCABJStYrUeeCwbbXYGI7wQYFgoAIBYhBIlOS1fcqXcra2krah1yVnWs
+vbckBQJqWkEWAhsCAIEJEB1yVnWsvbckdiAEGRYKAB0WIQTVLXUNPoC+3lVTsu2v
+Ch2TU4zZHgUCalpBFgAKCRCvCh2TU4zZHovHAP46C9cmUu4ep4HuMhsdALUH7V2F
+i0zZxlh06TsoCyjqegD/Wc+lxUjExmc6Nwth8iib5BqB8NUxYNO42AHQRW1y6Aq6
+9wD+NEFQrTDnM7165KMXcprJwmixQAS8sVKsTuuQRo642hABAOoifUNE9r4/6Bzm
+ixdipMf9A5HgGyWO0mNWIGk+ZRIH
+=Sivq
+-----END PGP PUBLIC KEY BLOCK-----
+"""
+
+
+# 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
+REVOKED_UID_FINGERPRINT: Final[str] = "346c2e8cd37b4f7c0bde03d9fdefe91a648ddd76"
+REVOKED_UID_SIGNING_FINGERPRINT: Final[str] = "5aec39f36f29b0d02ac67eb6dd6e33ceb5d275a6"
+REVOKED_UID_PRIMARY_EXPIRES_YEAR: Final[int] = 2030
+
+REVOKED_UID_PUBLIC_KEY_ASC: Final[str] = """-----BEGIN PGP PUBLIC KEY BLOCK-----
+
+mDMEalpBFhYJKwYBBAHaRw8BAQdAo/CQbE40pWaoJHDJ+zqNttN6+Op4Nb/7cFFw
+xev24XO0QUFUUiBSZXZva2VkIFVpZCAoRm9yIHRlc3QgdXNlIG9ubHkpIDxyZXZv
+a2VkLXVpZEBleGFtcGxlLmludmFsaWQ+iJkEExYKAEEWIQQ0bC6M03tPfAveA9n9
+7+kaZI3ddgUCalpBFgIbAQUJCGIiKgULCQgHAgIiAgYVCgkICwIEFgIDAQIeBwIX
+gAAKCRD97+kaZI3ddoKJAP0Qj4S7sl729rTc0UxbGeQur/GyuY/38H9LOCZUqsQ3
+UQEA+fneJoJtrT/TCCe9F+3xKDGLRrFzXbM5j6jwRTaILgK0NkFUUiBSZXZva2Vk
+IFVpZCBTZWNvbmQgPHJldm9rZWQtdWlkLTJAZXhhbXBsZS5pbnZhbGlkPoh4BDAW
+CgAgFiEENGwujNN7T3wL3gPZ/e/pGmSN3XYFAmpaQRcCHSAACgkQ/e/pGmSN3XYT
+EQD/UjqNftLvxzE8ah+nHWZZ3SRTSCcdZC6B/Hak4dd92gkA/3HPMo3BNSky+mml
+KSFbjOCembdIL9cWhsk/60T70ZADiJkEExYKAEEWIQQ0bC6M03tPfAveA9n97+ka
+ZI3ddgUCalpBFgIbAQUJCGIiKgULCQgHAgIiAgYVCgkICwIEFgIDAQIeBwIXgAAK
+CRD97+kaZI3ddv1AAP9nXCJtCPupIhiFj8MJmf5zs1iB8mvB8EExA62MEHFLMgD/
+Rfj3KhAGm8D6+e+T11dtP4RSeAev6OA8smDj1WBldQO4MwRqWkEWFgkrBgEEAdpH
+DwEBB0DfkeUI+ub7ARbR7L1mpf7IWypAnCO5dOzjcnJ6dwlSQ4jvBBgWCgAgFiEE
+NGwujNN7T3wL3gPZ/e/pGmSN3XYFAmpaQRYCGwIAgQkQ/e/pGmSN3XZ2IAQZFgoA
+HRYhBFrsOfNvKbDQKsZ+tt1uM8610nWmBQJqWkEWAAoJEN1uM8610nWmYHwA/1ng
+VdOTntQA+bbo+vHhqEGdpDvw/a1r6M2lxtBT87+2AQC1UhAkMxN0vdAZczqlBMFC
+i05PRe7RtkmV0875EUYpAz1fAP4yeSCHoNNOtOZKj3+uZuUeoDFyC6afH/LAmU+w
+rkJ3LQEA8l4Luk7tZdHZyq+w5q+hbbkRz4X5RJh3091rYGqBtAQ=
+=P9R1
+-----END PGP PUBLIC KEY BLOCK-----
+"""
+
+
+def write_revoked_subkey_fixture(tmp_path: pathlib.Path) -> tuple[str, str]:
+    artifact_path = tmp_path / "apache-test-1.0.txt"
+    signature_path = tmp_path / "apache-test-1.0.txt.asc"
+    artifact_path.write_bytes(REVOKED_SUBKEY_ARTIFACT_BYTES)
+    signature_path.write_text(REVOKED_SUBKEY_DETACHED_SIGNATURE_ASC, encoding="utf-8")
+    return str(signature_path), str(artifact_path)
diff --git a/tests/unit/test_checks_signature.py b/tests/unit/test_checks_signature.py
index f48b8c2b..c1d53161 100644
--- a/tests/unit/test_checks_signature.py
+++ b/tests/unit/test_checks_signature.py
@@ -288,6 +288,22 @@ def test_check_core_logic_rejects_signature_from_an_expired_subkey(tmp_path: pat
     assert result["error_kind"] == "key_expired"
 
 
+def test_check_core_logic_rejects_signature_from_a_revoked_subkey(tmp_path: pathlib.Path) -> None:
+    # The signature was made while the subkey was still good, so it verifies cryptographically and
+    # only a revocation check can reject it
+    signature_path, artifact_path = pgp_fixtures.write_revoked_subkey_fixture(tmp_path)
+
+    result = signature_check._check_core_logic_verify_signature(
+        signature_path=signature_path,
+        artifact_path=artifact_path,
+        ascii_armored_keys=[pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC],
+        apache_uid_map={pgp_fixtures.REVOKED_SUBKEY_PRIMARY_FINGERPRINT: True},
+    )
+
+    assert result["verified"] is False
+    assert result["error_kind"] == "revoked_key"
+
+
 def test_key_matches_signature_accepts_subkey_issuer_metadata() -> None:
     public_key, _ = openpgp.PublicKey.from_armor(_EMBEDDED_PUBLIC_KEY_ASC)
     signature, _ = openpgp.DetachedSignature.from_armor(_EMBEDDED_DETACHED_SIGNATURE_ASC)
diff --git a/tests/unit/test_keys_delete.py b/tests/unit/test_keys_delete.py
index a002fedd..c46a3113 100644
--- a/tests/unit/test_keys_delete.py
+++ b/tests/unit/test_keys_delete.py
@@ -59,9 +59,9 @@ async def test_committee_signing_keys_exclude_deleted(sqlite_sessionmaker, monke
     monkeypatch.setattr(checks.db, "session", lambda: sqlite_sessionmaker())
     release = SimpleNamespace(committee=SimpleNamespace(key="tooling"))
 
-    fingerprints = await checks._resolve_committee_signing_keys(release)
+    signing_keys = await checks._resolve_committee_signing_keys(release)
 
-    assert fingerprints == [BETA_FINGERPRINT]
+    assert signing_keys == [f"{BETA_FINGERPRINT}:armored beta"]
 
 
 async def test_delete_committee_keys_deletes_orphans(sqlite_sessionmaker) -> None:
@@ -188,6 +188,40 @@ async def test_reupload_undeletes_and_restores_associations(sqlite_sessionmaker)
         assert sorted(k.fingerprint for k in committee.public_signing_keys) == [ALPHA_FINGERPRINT, BETA_FINGERPRINT]
 
 
+async def test_reupload_with_a_changed_block_refreshes_it(sqlite_sessionmaker) -> None:
+    async with sqlite_sessionmaker() as data:
+        await _seed_keys(data)
+        writer = _committer_writer(data)
+        key = datatypes.Key(
+            status=datatypes.KeyStatus.PARSED,
+            key_model=_key(ALPHA_FINGERPRINT, "alice", "armored alpha revoked"),
+        )
+
+        oc, _publications = await writer._FoundationCommitter__database_add_model(key)
+
+        assert oc.result_or_raise().status == datatypes.KeyStatus.REFRESHED
+        row = await data.public_signing_key(fingerprint=ALPHA_FINGERPRINT).get()
+        assert row is not None
+        assert row.ascii_armored_key == "armored alpha revoked"
+        # A changed block can flip a signature check, so the key's committees must be rechecked
+        writer._recheck_committee_drafts.assert_awaited()
+
+
+async def test_reupload_with_an_unchanged_block_is_a_noop(sqlite_sessionmaker) -> None:
+    async with sqlite_sessionmaker() as data:
+        await _seed_keys(data)
+        writer = _committer_writer(data)
+        key = datatypes.Key(
+            status=datatypes.KeyStatus.PARSED,
+            key_model=_key(ALPHA_FINGERPRINT, "alice", "armored alpha"),
+        )
+
+        oc, _publications = await writer._FoundationCommitter__database_add_model(key)
+
+        assert oc.result_or_raise().status == datatypes.KeyStatus.PARSED
+        writer._recheck_committee_drafts.assert_not_awaited()
+
+
 async def test_signature_hints_consume_flags_and_deletes(sqlite_sessionmaker) -> None:
     async with sqlite_sessionmaker() as data:
         await _seed_keys(data)
diff --git a/tests/unit/test_pgp.py b/tests/unit/test_pgp.py
index 291e961b..02beb687 100644
--- a/tests/unit/test_pgp.py
+++ b/tests/unit/test_pgp.py
@@ -124,3 +124,57 @@ def test_declares_signing_permits_a_key_declaring_no_capabilities() -> None:
 
 def test_declares_signing_refuses_a_key_whose_self_signature_cannot_be_read() -> None:
     assert pgp._declares_signing(None) is False
+
+
+def test_signing_key_status_flags_a_revoked_signing_subkey() -> None:
+    key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC)
+
+    status = pgp.signing_key_status(key, {pgp_fixtures.REVOKED_SUBKEY_SIGNING_FINGERPRINT}, set())
+
+    assert status.revoked is True
+
+
+def test_signing_key_status_flags_a_subkey_beneath_a_revoked_primary() -> None:
+    key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC)
+
+    # The subkey binding is intact, so only the primary's revocation cascading down can reject it
+    status = pgp.signing_key_status(key, {pgp_fixtures.REVOKED_PRIMARY_SIGNING_FINGERPRINT}, set())
+
+    assert status.revoked is True
+
+
+def test_signing_key_status_flags_a_revoked_primary() -> None:
+    key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC)
+
+    status = pgp.signing_key_status(key, {pgp_fixtures.REVOKED_PRIMARY_FINGERPRINT}, set())
+
+    assert status.revoked is True
+
+
+def test_signing_key_status_does_not_treat_a_revoked_user_id_as_a_revoked_key() -> None:
+    key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
+
+    status = pgp.signing_key_status(key, {pgp_fixtures.REVOKED_UID_FINGERPRINT}, set())
+
+    assert status.revoked is False
+
+
+def test_key_expires_at_ignores_a_revoked_user_id_self_signature() -> None:
+    # A revocation is self-issued but declares no expiry, so reading it as the effective self-signature
+    # would erase the expiry the certification actually carries
+    key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
+
+    expires = pgp.key_expires_at(key)
+
+    assert expires is not None
+    assert expires.year == pgp_fixtures.REVOKED_UID_PRIMARY_EXPIRES_YEAR
+
+
+def test_signing_key_status_reads_capabilities_past_a_revoked_user_id() -> None:
+    # The same revocation carries no key flags, so admitting it would flip a certify-only primary to
+    # can_sign purely because a secondary address was revoked
+    key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
+
+    status = pgp.signing_key_status(key, {pgp_fixtures.REVOKED_UID_FINGERPRINT}, set())
+
+    assert status.can_sign is False


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