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/08/05 13:50:41 UTC
(tooling-trusted-releases) 01/01: #1420 - revocation now verified
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 b4155bc83cd4474fe8ac6d1b8ef67c9ddfa3ac2c
Author: Alastair McFarlane <ar...@apache.org>
AuthorDate: Wed Aug 5 14:50:01 2026 +0100
#1420 - revocation now verified
---
atr/api/__init__.py | 20 ++--
atr/pgp.py | 183 +++++++++++++++++++-------------
atr/storage/writers/keys.py | 17 ++-
atr/tasks/checks/signature.py | 82 +++++++-------
atr/util.py | 8 +-
pyproject.toml | 2 +-
scripts/hints_import.py | 2 +-
tests/unit/test_checks_signature.py | 10 +-
tests/unit/test_keys_writer.py | 20 ++--
tests/unit/test_pgp.py | 120 ++++++++++++++++-----
tests/unit/test_signature_provenance.py | 24 ++---
uv.lock | 22 ++--
12 files changed, 303 insertions(+), 207 deletions(-)
diff --git a/atr/api/__init__.py b/atr/api/__init__.py
index 2137b2f6..23efdfdd 100644
--- a/atr/api/__init__.py
+++ b/atr/api/__init__.py
@@ -1618,10 +1618,10 @@ async def signature_provenance(
signing_keys: list[models.api.SignatureProvenanceKey] = []
signature_asc_data = data.signature_asc_text
- sig, _ = openpgp.DetachedSignature.from_armor(signature_asc_data)
- signature_info = sig.signature_info()
- issuer_fingerprints = {fingerprint.lower() for fingerprint in signature_info.issuer_fingerprints}
- issuer_key_ids = {key_id.lower() for key_id in signature_info.issuer_key_ids}
+ sig, _ = openpgp.composed.DetachedSignature.from_armor(signature_asc_data)
+ signature_info = sig.signature
+ issuer_fingerprints = {fingerprint.lower() for fingerprint in signature_info.issuer_fingerprint()}
+ issuer_key_ids = {key_id.lower() for key_id in signature_info.issuer_key_id()}
if (not issuer_fingerprints) and (not issuer_key_ids):
raise exceptions.NotFound("No signer fingerprint or key id found")
async with db.session() as db_data:
@@ -2276,16 +2276,18 @@ async def _match_release(release_directory: safe.StatePath, data: models.api.Sig
def _matched_issuer_fingerprint(
- key: openpgp.PublicKey,
+ key: openpgp.composed.SignedPublicKey,
issuer_fingerprints: set[str],
issuer_key_ids: set[str],
) -> str | None:
key_fingerprint = key.fingerprint.lower()
if _issuer_matches_component(key_fingerprint, key.key_id.lower(), issuer_fingerprints, issuer_key_ids):
return key_fingerprint
- for subkey in key.subkey_bindings():
- subkey_fingerprint = subkey.fingerprint.lower()
- if _issuer_matches_component(subkey_fingerprint, subkey.key_id.lower(), issuer_fingerprints, issuer_key_ids):
+ for subkey in key.public_subkeys:
+ subkey_fingerprint = subkey.key.fingerprint.lower()
+ if _issuer_matches_component(
+ subkey_fingerprint, subkey.key.key_id.lower(), issuer_fingerprints, issuer_key_ids
+ ):
return subkey_fingerprint
return None
@@ -2335,7 +2337,7 @@ async def _resolve_signing_key_from_signature(
if isinstance(armored, bytes):
armored = armored.decode("utf-8", errors="replace")
try:
- parsed, _ = openpgp.PublicKey.from_armor(armored)
+ parsed, _ = openpgp.composed.SignedPublicKey.from_armor(armored)
except Exception as e:
log.info(f"Failed to parse key {stored.fingerprint}: {e}")
continue
diff --git a/atr/pgp.py b/atr/pgp.py
index a291e88a..3d4fd019 100644
--- a/atr/pgp.py
+++ b/atr/pgp.py
@@ -61,37 +61,42 @@ class SigningKeyStatus:
revoked: bool
-def key_expires_at(key: openpgp.PublicKey) -> datetime.datetime | None:
+def key_expires_at(key: openpgp.composed.SignedPublicKey) -> datetime.datetime | None:
effective = _effective_self_signature(key)
if effective is None:
return None
- key_expiration_seconds = effective.key_expiration_seconds
+ key_expiration_seconds = effective.key_expiration_time()
if not key_expiration_seconds:
return None
return datetime.datetime.fromtimestamp(key.created_at + key_expiration_seconds, tz=datetime.UTC)
-def latest_self_signature(key: openpgp.PublicKey) -> openpgp.SignatureInfo | None:
+def latest_self_signature(key: openpgp.composed.SignedPublicKey) -> openpgp.packet.Signature | None:
signatures = _direct_self_signatures(key)
signatures.extend(_binding_self_signatures(key))
return _latest_signature(signatures)
-def latest_self_signature_created_at(key: openpgp.PublicKey) -> datetime.datetime | None:
+def latest_self_signature_created_at(key: openpgp.composed.SignedPublicKey) -> datetime.datetime | None:
signature = latest_self_signature(key)
- if signature is None or signature.creation_time is None:
+ if signature is None:
+ return None
+ created = signature.created()
+ if created is None:
return None
- return datetime.datetime.fromtimestamp(signature.creation_time, tz=datetime.UTC)
+ return datetime.datetime.fromtimestamp(created, tz=datetime.UTC)
-def public_params_bits(public_params: openpgp.PublicParamsInfo) -> int | None:
+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):
return bits
return None
-def revocations_dropped(previous: openpgp.PublicKey, incoming: openpgp.PublicKey) -> set[str]:
+def revocations_dropped(
+ previous: openpgp.composed.SignedPublicKey, incoming: openpgp.composed.SignedPublicKey
+) -> set[str]:
"""Signing keys revoked in `previous` but still present and no longer revoked in `incoming`."""
# A revocation is irreversible and survives a minimised re-export, so a signing key coming back from
# revoked to live is a stripped block. A key dropped altogether is left to the caller
@@ -106,8 +111,9 @@ def revocations_dropped(previous: openpgp.PublicKey, incoming: openpgp.PublicKey
return dropped
-def signing_key_facts(key: openpgp.PublicKey) -> list[SigningKeyFacts]:
+def signing_key_facts(key: openpgp.composed.SignedPublicKey) -> list[SigningKeyFacts]:
"""Every key in a certificate which can carry a signature, the primary first."""
+ primary = key.primary_key
primary_revoked = _primary_is_revoked(key)
primary_expires = key_expires_at(key)
facts = [
@@ -124,24 +130,24 @@ def signing_key_facts(key: openpgp.PublicKey) -> list[SigningKeyFacts]:
)
]
- for binding in key.subkey_bindings():
- binding_signature = _latest_binding_signature(binding)
+ for subkey in key.public_subkeys:
+ binding_signature = _latest_binding_signature(subkey)
# A subkey is only usable while its own binding and the primary above it both hold
expirations = [
expiration
- for expiration in (primary_expires, _subkey_expires_at(binding, binding_signature))
+ for expiration in (primary_expires, _subkey_expires_at(subkey, binding_signature))
if expiration is not None
]
facts.append(
SigningKeyFacts(
- fingerprint=binding.fingerprint.lower(),
- key_id=binding.key_id.lower(),
+ fingerprint=subkey.key.fingerprint.lower(),
+ key_id=subkey.key.key_id.lower(),
is_primary=False,
- algorithm=binding.public_key_algorithm,
- length_bits=public_params_bits(binding.public_params),
- created=datetime.datetime.fromtimestamp(binding.created_at, tz=datetime.UTC),
+ algorithm=subkey.key.public_key_algorithm,
+ length_bits=public_params_bits(subkey.key.public_params),
+ 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(binding),
+ revoked=primary_revoked or _subkey_is_revoked(primary, subkey),
can_sign=_declares_signing(binding_signature),
)
)
@@ -149,28 +155,29 @@ def signing_key_facts(key: openpgp.PublicKey) -> list[SigningKeyFacts]:
def signing_key_status(
- key: openpgp.PublicKey,
+ key: openpgp.composed.SignedPublicKey,
issuer_fingerprints: set[str],
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 = key.primary_key
primary_revoked = _primary_is_revoked(key)
- binding = _issuing_subkey_binding(key, issuer_fingerprints, issuer_key_ids)
- if binding is not None:
+ subkey = _issuing_subkey(key, issuer_fingerprints, issuer_key_ids)
+ if subkey is not None:
# A subkey is only usable while its own binding and the primary above it both hold
- binding_signature = _latest_binding_signature(binding)
+ binding_signature = _latest_binding_signature(subkey)
expirations = [
expiration
- for expiration in (key_expires_at(key), _subkey_expires_at(binding, binding_signature))
+ for expiration in (key_expires_at(key), _subkey_expires_at(subkey, binding_signature))
if expiration is not None
]
return SigningKeyStatus(
identified=True,
- fingerprint=binding.fingerprint.lower(),
+ fingerprint=subkey.key.fingerprint.lower(),
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),
+ revoked=primary_revoked or _subkey_is_revoked(primary, subkey),
)
if _issuer_is_primary(key, issuer_fingerprints, issuer_key_ids):
@@ -187,47 +194,45 @@ def signing_key_status(
return SigningKeyStatus(identified=False, fingerprint=None, expires=None, can_sign=False, revoked=False)
-def _binding_revoked(binding: openpgp.UserBindingInfo, fingerprint: str, key_id: str) -> bool:
+def _binding_revoked(user: openpgp.types.SignedUser, fingerprint: str, key_id: str) -> bool:
candidates = [
signature
- for signature in binding.signatures
+ for signature in user.signatures
if _signature_is_self(signature, fingerprint, key_id)
and (
- (signature.signature_type in _CERTIFICATION_SIGNATURE_TYPES)
- or (signature.signature_type == _CERTIFICATION_REVOCATION_SIGNATURE_TYPE)
+ (signature.typ() in _CERTIFICATION_SIGNATURE_TYPES)
+ or (signature.typ() == _CERTIFICATION_REVOCATION_SIGNATURE_TYPE)
)
]
latest = _latest_signature(candidates)
- return (latest is not None) and (latest.signature_type == _CERTIFICATION_REVOCATION_SIGNATURE_TYPE)
+ return (latest is not None) and (latest.typ() == _CERTIFICATION_REVOCATION_SIGNATURE_TYPE)
-def _binding_self_signatures(key: openpgp.PublicKey) -> list[openpgp.SignatureInfo]:
+def _binding_self_signatures(key: openpgp.composed.SignedPublicKey) -> list[openpgp.packet.Signature]:
self_fingerprint = key.fingerprint.lower()
self_key_id = key.key_id.lower()
- active_bindings = [
- binding for binding in key.user_bindings() if not _binding_revoked(binding, self_fingerprint, self_key_id)
- ]
- primary_bindings = [binding for binding in active_bindings if binding.is_primary]
- chosen_bindings = primary_bindings or active_bindings
- binding_sigs: list[openpgp.SignatureInfo] = []
- for binding in chosen_bindings:
+ active_users = [user for user in key.details.users if not _binding_revoked(user, self_fingerprint, self_key_id)]
+ primary_users = [user for user in active_users if user.is_primary]
+ chosen_users = primary_users or active_users
+ binding_sigs: list[openpgp.packet.Signature] = []
+ for user in chosen_users:
binding_sigs.extend(
signature
- for signature in binding.signatures
+ for signature in user.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)
+ and (signature.typ() in _CERTIFICATION_SIGNATURE_TYPES)
)
return binding_sigs
-def _declares_signing(signature: openpgp.SignatureInfo | None) -> bool:
+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
if signature is None:
return False
- flags = signature.key_flags
+ flags = signature.key_flags()
# An absent key flags subpacket reads as every flag unset, which we can't tell apart from a key
# declaring no capabilities at all. Older keys often declare nothing, so treat silence as
# permission and only refuse a key which positively declares capabilities excluding signing
@@ -244,17 +249,17 @@ def _declares_signing(signature: openpgp.SignatureInfo | None) -> bool:
return flags.sign
-def _direct_self_signatures(key: openpgp.PublicKey) -> list[openpgp.SignatureInfo]:
+def _direct_self_signatures(key: openpgp.composed.SignedPublicKey) -> list[openpgp.packet.Signature]:
self_fingerprint = key.fingerprint.lower()
self_key_id = key.key_id.lower()
return [
signature
- for signature in key.direct_signature_infos()
+ for signature in key.details.direct_signatures
if _signature_is_self(signature, self_fingerprint, self_key_id)
]
-def _effective_self_signature(key: openpgp.PublicKey) -> openpgp.SignatureInfo | None:
+def _effective_self_signature(key: openpgp.composed.SignedPublicKey) -> openpgp.packet.Signature | None:
direct_sigs = _direct_self_signatures(key)
binding_sigs = _binding_self_signatures(key)
if key.version >= 6:
@@ -265,74 +270,100 @@ def _effective_self_signature(key: openpgp.PublicKey) -> openpgp.SignatureInfo |
def _issuer_is_primary(
- key: openpgp.PublicKey,
+ key: openpgp.composed.SignedPublicKey,
issuer_fingerprints: set[str],
issuer_key_ids: set[str],
) -> bool:
return (key.fingerprint.lower() in issuer_fingerprints) or (key.key_id.lower() in issuer_key_ids)
-def _issuing_subkey_binding(
- key: openpgp.PublicKey,
+def _issuing_subkey(
+ key: openpgp.composed.SignedPublicKey,
issuer_fingerprints: set[str],
issuer_key_ids: set[str],
-) -> openpgp.SubkeyBindingInfo | None:
- for binding in key.subkey_bindings():
- if binding.fingerprint.lower() in issuer_fingerprints:
- return binding
- if binding.key_id.lower() in issuer_key_ids:
- return binding
+) -> openpgp.composed.SignedPublicSubKey | None:
+ for subkey in key.public_subkeys:
+ if subkey.key.fingerprint.lower() in issuer_fingerprints:
+ return subkey
+ if subkey.key.key_id.lower() in issuer_key_ids:
+ return subkey
return None
-def _latest_binding_signature(binding: openpgp.SubkeyBindingInfo) -> openpgp.SignatureInfo | None:
+def _latest_binding_signature(subkey: openpgp.composed.SignedPublicSubKey) -> openpgp.packet.Signature | None:
return _latest_signature(
- [signature for signature in binding.signatures if signature.signature_type == _SUBKEY_BINDING_SIGNATURE_TYPE]
+ [signature for signature in subkey.signatures if signature.typ() == _SUBKEY_BINDING_SIGNATURE_TYPE]
)
-def _latest_signature(signatures: list[openpgp.SignatureInfo]) -> openpgp.SignatureInfo | None:
+def _latest_signature(signatures: list[openpgp.packet.Signature]) -> openpgp.packet.Signature | None:
now = datetime.datetime.now(datetime.UTC).timestamp()
valid = [signature for signature in signatures if not _signature_expired(signature, now)]
if not valid:
return None
- return max(valid, key=lambda signature: signature.creation_time or 0)
+ return max(valid, key=lambda signature: signature.created() or 0)
-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 _primary_is_revoked(key: openpgp.composed.SignedPublicKey) -> bool:
+ primary = key.primary_key
+ for signature in key.details.revocation_signatures:
+ if signature.typ() != _KEY_REVOCATION_SIGNATURE_TYPE:
+ continue
+ if _revocation_verifies(signature, primary):
+ return True
+ return False
+
+
+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
+ # by dropping one into a KEYS file. A designated-revoker revocation is signed by someone other than
+ # the primary, so it won't verify here and reads as not revoked, which is conservative and rare
+ # enough to leave for now
+ try:
+ signature.verify_key(primary)
+ except openpgp.errors.Error:
+ return False
+ return True
-def _signature_expired(signature: openpgp.SignatureInfo, now: float) -> bool:
- expiration = signature.signature_expiration_seconds
+def _signature_expired(signature: openpgp.packet.Signature, now: float) -> bool:
+ expiration = signature.signature_expiration_time()
if not expiration:
return False
- creation = signature.creation_time or 0
+ creation = signature.created() or 0
return (creation + expiration) <= now
-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}
+def _signature_is_self(signature: openpgp.packet.Signature, self_fingerprint: str, self_key_id: str) -> bool:
+ fingerprints = {fingerprint.lower() for fingerprint in signature.issuer_fingerprint()}
+ key_ids = {key_id.lower() for key_id in signature.issuer_key_id()}
return (self_fingerprint in fingerprints) or (self_key_id in key_ids)
def _subkey_expires_at(
- binding: openpgp.SubkeyBindingInfo,
- signature: openpgp.SignatureInfo | None,
+ subkey: openpgp.composed.SignedPublicSubKey,
+ signature: openpgp.packet.Signature | None,
) -> datetime.datetime | None:
if signature is None:
return None
- key_expiration_seconds = signature.key_expiration_seconds
+ key_expiration_seconds = signature.key_expiration_time()
if not key_expiration_seconds:
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)
+ return datetime.datetime.fromtimestamp(subkey.key.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)
+def _subkey_is_revoked(primary: openpgp.packet.PublicKey, subkey: openpgp.composed.SignedPublicSubKey) -> bool:
+ # As with the primary, only a revocation the primary actually made counts, so a forged
+ # subkey-revocation can't be used to block a key. The revocation can share a timestamp with the
+ # binding it kills, so a verifying one is enough on its own without weighing the two dates
+ for signature in subkey.signatures:
+ if signature.typ() != _SUBKEY_REVOCATION_SIGNATURE_TYPE:
+ continue
+ try:
+ signature.verify_subkey_binding(primary, subkey.key)
+ except openpgp.errors.Error:
+ continue
+ return True
+ return False
diff --git a/atr/storage/writers/keys.py b/atr/storage/writers/keys.py
index 9941fe1e..00fab7e3 100644
--- a/atr/storage/writers/keys.py
+++ b/atr/storage/writers/keys.py
@@ -99,7 +99,7 @@ def _algorithm_name(algorithm: int) -> str:
return _ALGORITHM_NAMES.get(algorithm, str(algorithm))
-def _key_length(key: openpgp.PublicKey) -> int:
+def _key_length(key: openpgp.composed.SignedPublicKey) -> int:
bits = pgp.public_params_bits(key.public_params)
if bits is None:
raise ValueError(f"Key size is not available for algorithm {key.public_key_algorithm}")
@@ -111,8 +111,8 @@ def _block_downgrade_reason(stored_block: str, incoming_block: str) -> str | Non
# effective state back: no revocation dropped, no regression to an older self-signature. A minimised
# re-export keeps both and passes. A block we can't parse can't be judged, so allow it through
try:
- stored_key, _ = openpgp.PublicKey.from_armor(stored_block)
- incoming_key, _ = openpgp.PublicKey.from_armor(incoming_block)
+ stored_key, _ = openpgp.composed.SignedPublicKey.from_armor(stored_block)
+ incoming_key, _ = openpgp.composed.SignedPublicKey.from_armor(incoming_block)
except Exception:
return None
dropped = pgp.revocations_dropped(stored_key, incoming_key)
@@ -129,7 +129,7 @@ def _signing_key_rows(certificate_fingerprint: str, block: str | bytes) -> list[
"""The SigningKey rows a certificate's block describes, or None if the block is for another key."""
if isinstance(block, bytes):
block = block.decode("utf-8", errors="replace")
- key, _ = openpgp.PublicKey.from_armor(block)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(block)
if key.fingerprint.lower() != certificate_fingerprint.lower():
return None
return [
@@ -257,7 +257,7 @@ class FoundationCommitter(GeneralPublic):
def public_key_model(
self,
- key: openpgp.PublicKey,
+ key: openpgp.composed.SignedPublicKey,
ldap_data: cache.EmailUidLookup,
original_key_block: str | None = None,
) -> sql.SigningCertificate:
@@ -265,7 +265,6 @@ class FoundationCommitter(GeneralPublic):
asf_uid = self.__uids_asf_uid(uids, ldap_data)
if not uids:
raise ValueError("No UIDs found in key")
-
# Use the original key block if available
ascii_armored = original_key_block if original_key_block else key.to_armored()
@@ -505,7 +504,7 @@ class FoundationCommitter(GeneralPublic):
return key
def __block_model_create(self, key_block: str, ldap_data: cache.EmailUidLookup) -> datatypes.Key:
- public_key, _ = openpgp.PublicKey.from_armor(key_block)
+ public_key, _ = openpgp.composed.SignedPublicKey.from_armor(key_block)
key_model = self.public_key_model(public_key, ldap_data, original_key_block=key_block)
_validate_key_strength(public_key)
return datatypes.Key(
@@ -927,7 +926,7 @@ class CommitteeParticipant(FoundationCommitter):
def __block_models(self, key_block: str, ldap_data: cache.EmailUidLookup) -> list[datatypes.Key | Exception]:
try:
- public_key, _ = openpgp.PublicKey.from_armor(key_block)
+ public_key, _ = openpgp.composed.SignedPublicKey.from_armor(key_block)
except Exception as e:
raise ValueError(f"Error loading OpenPGP key block: {e}") from e
key_list = []
@@ -1304,7 +1303,7 @@ class FoundationAdmin(CommitteeMember):
return (num_unlinked, num_deleted, publication)
-def _validate_key_strength(key: openpgp.PublicKey) -> None:
+def _validate_key_strength(key: openpgp.composed.SignedPublicKey) -> None:
"""Raise ValueError if the key is recently-generated and does not meet the minimum cryptographic strength."""
algorithm = _algorithm_id(key.public_key_algorithm)
length = _key_length(key)
diff --git a/atr/tasks/checks/signature.py b/atr/tasks/checks/signature.py
index 08313ef4..db1928e0 100644
--- a/atr/tasks/checks/signature.py
+++ b/atr/tasks/checks/signature.py
@@ -171,7 +171,7 @@ def _check_core_logic_verify_signature(
except OSError as e:
raise task.CheckRetryableError("Unable to read the signature file", {"error": str(e)}) from e
try:
- signature, _ = openpgp.DetachedSignature.from_armor(armored.decode("utf-8"))
+ signature, _ = openpgp.composed.DetachedSignature.from_armor(armored.decode("utf-8"))
except Exception as e:
return {
"verified": False,
@@ -186,7 +186,7 @@ def _check_core_logic_verify_signature(
key_has_apache_uid=False,
),
}
- signature_info = signature.signature_info()
+ signature_info = signature.signature
try:
with open(artifact_path, "rb"):
pass
@@ -206,8 +206,8 @@ def _check_core_logic_verify_signature(
}
except OSError as e:
raise task.CheckRetryableError("Unable to read the artifact file", {"error": str(e)}) from e
- issuer_fingerprints = {fingerprint.lower() for fingerprint in signature_info.issuer_fingerprints}
- issuer_key_ids = {key_id.lower() for key_id in signature_info.issuer_key_ids}
+ issuer_fingerprints = {fingerprint.lower() for fingerprint in signature_info.issuer_fingerprint()}
+ issuer_key_ids = {key_id.lower() for key_id in signature_info.issuer_key_id()}
candidate_keys = [key for key in public_keys if _key_matches_signature(key, issuer_fingerprints, issuer_key_ids)]
matched_key = _first_verifying_key(signature, candidate_keys, artifact_path)
@@ -258,9 +258,9 @@ def _check_core_logic_verify_signature(
"debug_info": debug_info,
}
- key_id = signature_info.issuer_key_ids[0] if signature_info.issuer_key_ids else matched_key.key_id
+ key_id = signature_info.issuer_key_id()[0] if signature_info.issuer_key_id() else matched_key.key_id
username = _signer_username(matched_key, signature_info) or "Unknown"
- timestamp = signature_info.creation_time
+ timestamp = signature_info.created()
return {
"verified": True,
"key_id": key_id,
@@ -278,8 +278,8 @@ def _check_core_logic_verify_signature(
def _debug_info(
*,
- key: openpgp.PublicKey | None,
- signature_info: openpgp.SignatureInfo | None,
+ key: openpgp.composed.SignedPublicKey | None,
+ signature_info: openpgp.packet.Signature | None,
status: str,
valid: bool,
num_committee_keys: int,
@@ -290,10 +290,12 @@ def _debug_info(
# key_id below names the issuing key, so the fingerprint beside it has to name the same one
fingerprint = signing_key_fingerprint or certificate_fingerprint
key_id = key.key_id if (key is not None) else "Not available"
- creation_time = signature_info.creation_time if (signature_info is not None) else None
+ creation_time = signature_info.created() if (signature_info is not None) else None
username = _signer_username(key, signature_info) if ((key is not None) and (signature_info is not None)) else None
return {
- "key_id": (signature_info.issuer_key_ids[0] if (signature_info and signature_info.issuer_key_ids) else key_id),
+ "key_id": (
+ signature_info.issuer_key_id()[0] if (signature_info and signature_info.issuer_key_id()) else key_id
+ ),
"fingerprint": fingerprint,
"pubkey_fingerprint": certificate_fingerprint,
"creation_date": creation_time if (creation_time is not None) else "Not available",
@@ -306,18 +308,17 @@ def _debug_info(
"stderr": "Not available",
"num_committee_keys": num_committee_keys,
"key_has_apache_uid": key_has_apache_uid,
- "hash_algorithm": signature_info.hash_algorithm if (signature_info is not None) else "Not available",
- "signature_type": signature_info.signature_type if (signature_info is not None) else "Not available",
- "public_key_algorithm": (
- signature_info.public_key_algorithm if (signature_info is not None) else "Not available"
- ),
+ "hash_algorithm": signature_info.hash_alg() if (signature_info is not None) else "Not available",
+ "signature_type": signature_info.typ() if (signature_info is not None) else "Not available",
+ # The signature no longer carries its own algorithm, so this reports the certificate's instead
+ "public_key_algorithm": (key.public_key_algorithm if (key is not None) else "Not available"),
}
def _expired_key_result(
*,
- matched_key: openpgp.PublicKey,
- signature_info: openpgp.SignatureInfo,
+ matched_key: openpgp.composed.SignedPublicKey,
+ signature_info: openpgp.packet.Signature,
expires: datetime.datetime,
num_committee_keys: int,
key_has_apache_uid: bool,
@@ -342,10 +343,10 @@ def _expired_key_result(
def _first_verifying_key(
- signature: openpgp.DetachedSignature,
- candidate_keys: list[openpgp.PublicKey],
+ signature: openpgp.composed.DetachedSignature,
+ candidate_keys: list[openpgp.composed.SignedPublicKey],
artifact_path: str,
-) -> openpgp.PublicKey | None:
+) -> openpgp.composed.SignedPublicKey | None:
for candidate_key in candidate_keys:
try:
signature.verify_file(candidate_key, artifact_path)
@@ -357,7 +358,7 @@ def _first_verifying_key(
def _key_matches_signature(
- key: openpgp.PublicKey,
+ key: openpgp.composed.SignedPublicKey,
issuer_fingerprints: set[str],
issuer_key_ids: set[str],
) -> bool:
@@ -365,20 +366,20 @@ def _key_matches_signature(
return True
key_fingerprints = {key.fingerprint.lower()}
- key_fingerprints.update(subkey.fingerprint.lower() for subkey in key.subkey_bindings())
+ key_fingerprints.update(subkey.key.fingerprint.lower() for subkey in key.public_subkeys)
if key_fingerprints.intersection(issuer_fingerprints):
return True
key_ids = {key.key_id.lower()}
- key_ids.update(subkey.key_id.lower() for subkey in key.subkey_bindings())
+ key_ids.update(subkey.key.key_id.lower() for subkey in key.public_subkeys)
return bool(key_ids.intersection(issuer_key_ids))
def _key_refusal_result(
*,
status: pgp.SigningKeyStatus,
- matched_key: openpgp.PublicKey,
- signature_info: openpgp.SignatureInfo,
+ matched_key: openpgp.composed.SignedPublicKey,
+ signature_info: openpgp.packet.Signature,
num_committee_keys: int,
key_has_apache_uid: bool,
) -> dict[str, Any] | None:
@@ -417,8 +418,8 @@ def _key_refusal_result(
def _not_signing_key_result(
*,
- matched_key: openpgp.PublicKey,
- signature_info: openpgp.SignatureInfo,
+ matched_key: openpgp.composed.SignedPublicKey,
+ signature_info: openpgp.packet.Signature,
num_committee_keys: int,
key_has_apache_uid: bool,
) -> dict[str, Any]:
@@ -441,11 +442,11 @@ def _not_signing_key_result(
}
-def _parse_public_keys(ascii_armored_keys: list[str]) -> list[openpgp.PublicKey]:
- public_keys: list[openpgp.PublicKey] = []
+def _parse_public_keys(ascii_armored_keys: list[str]) -> list[openpgp.composed.SignedPublicKey]:
+ public_keys: list[openpgp.composed.SignedPublicKey] = []
for ascii_armored_key in ascii_armored_keys:
try:
- public_key, _ = openpgp.PublicKey.from_armor(ascii_armored_key)
+ public_key, _ = openpgp.composed.SignedPublicKey.from_armor(ascii_armored_key)
except Exception as e:
log.warning(f"Failed to parse committee public key: {e}")
continue
@@ -477,8 +478,8 @@ async def _record_uploader_concern(
def _revoked_key_result(
*,
- matched_key: openpgp.PublicKey,
- signature_info: openpgp.SignatureInfo,
+ matched_key: openpgp.composed.SignedPublicKey,
+ signature_info: openpgp.packet.Signature,
num_committee_keys: int,
key_has_apache_uid: bool,
) -> dict[str, Any]:
@@ -501,12 +502,13 @@ def _revoked_key_result(
}
-def _signer_username(key: openpgp.PublicKey, signature_info: openpgp.SignatureInfo) -> str | None:
- if signature_info.signer_user_id:
- return signature_info.signer_user_id
- for binding in key.user_bindings():
- if binding.is_primary:
- return binding.user_id
+def _signer_username(key: openpgp.composed.SignedPublicKey, signature_info: openpgp.packet.Signature) -> str | None:
+ signer_user_id = signature_info.signers_userid()
+ if signer_user_id:
+ return signer_user_id
+ for user in key.details.users:
+ if user.is_primary:
+ return user.user_id
if key.user_ids:
return key.user_ids[0]
return None
@@ -514,8 +516,8 @@ def _signer_username(key: openpgp.PublicKey, signature_info: openpgp.SignatureIn
def _unidentified_key_result(
*,
- matched_key: openpgp.PublicKey,
- signature_info: openpgp.SignatureInfo,
+ matched_key: openpgp.composed.SignedPublicKey,
+ signature_info: openpgp.packet.Signature,
num_committee_keys: int,
key_has_apache_uid: bool,
) -> dict[str, Any]:
diff --git a/atr/util.py b/atr/util.py
index 93af6cb6..120db22a 100644
--- a/atr/util.py
+++ b/atr/util.py
@@ -1095,11 +1095,11 @@ async def number_of_release_files(release: sql.Release) -> int:
return count
-def openpgp_member_ids(key: openpgp.PublicKey) -> set[str]:
+def openpgp_member_ids(key: openpgp.composed.SignedPublicKey) -> set[str]:
member_ids = {key.fingerprint.lower(), key.key_id.lower()}
- for binding in key.subkey_bindings():
- member_ids.add(binding.fingerprint.lower())
- member_ids.add(binding.key_id.lower())
+ for subkey in key.public_subkeys:
+ member_ids.add(subkey.key.fingerprint.lower())
+ member_ids.add(subkey.key.key_id.lower())
return member_ids
diff --git a/pyproject.toml b/pyproject.toml
index 7012434b..b135b9a3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -50,7 +50,7 @@ dependencies = [
"quart-wtforms>=1.0.3,<2.0.0",
"requests>=2.33.0",
"rich>=14.0.0",
- "rpgp-py==0.19.8",
+ "rpgp-py==0.20.0",
"semver>=3.0.4",
"sqlmodel>=0.0.24,<0.1.0",
"ssh-audit>=3.3.0",
diff --git a/scripts/hints_import.py b/scripts/hints_import.py
index 591d0a7b..fe811aa9 100755
--- a/scripts/hints_import.py
+++ b/scripts/hints_import.py
@@ -96,7 +96,7 @@ def key_packet_ids(body: bytes) -> set[str]:
def lenient_member_ids(armored: str) -> set[str]:
- key, _ = openpgp.PublicKey.from_armor(armored)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(armored)
return util.openpgp_member_ids(key)
diff --git a/tests/unit/test_checks_signature.py b/tests/unit/test_checks_signature.py
index 783803fb..7c9b0984 100644
--- a/tests/unit/test_checks_signature.py
+++ b/tests/unit/test_checks_signature.py
@@ -394,11 +394,11 @@ async def test_check_records_a_concern_when_the_signer_did_not_upload(
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)
- signature_info = signature.signature_info()
- issuer_fingerprints = {fingerprint.lower() for fingerprint in signature_info.issuer_fingerprints}
- issuer_key_ids = {key_id.lower() for key_id in signature_info.issuer_key_ids}
+ public_key, _ = openpgp.composed.SignedPublicKey.from_armor(_EMBEDDED_PUBLIC_KEY_ASC)
+ signature, _ = openpgp.composed.DetachedSignature.from_armor(_EMBEDDED_DETACHED_SIGNATURE_ASC)
+ signature_info = signature.signature
+ issuer_fingerprints = {fingerprint.lower() for fingerprint in signature_info.issuer_fingerprint()}
+ issuer_key_ids = {key_id.lower() for key_id in signature_info.issuer_key_id()}
assert signature_check._key_matches_signature(public_key, issuer_fingerprints, issuer_key_ids)
diff --git a/tests/unit/test_keys_writer.py b/tests/unit/test_keys_writer.py
index 6799949c..a577319a 100644
--- a/tests/unit/test_keys_writer.py
+++ b/tests/unit/test_keys_writer.py
@@ -418,14 +418,14 @@ async def test_ensure_uses_cached_email_lookup_for_bulk_import() -> None:
def test_key_expires_at_uses_v4_user_binding_expiration() -> None:
- key, _ = keys_writer.openpgp.PublicKey.from_armor(_EMBEDDED_V4_EXPIRING_KEY_ASC)
- binding = next(iter(key.user_bindings()))
- binding_signature = binding.signatures[0]
+ key, _ = keys_writer.openpgp.composed.SignedPublicKey.from_armor(_EMBEDDED_V4_EXPIRING_KEY_ASC)
+ user = next(iter(key.details.users))
+ binding_signature = user.signatures[0]
expires = pgp.key_expires_at(key)
assert expires == datetime.datetime.fromtimestamp(
- key.created_at + binding_signature.key_expiration_seconds,
+ key.created_at + binding_signature.key_expiration_time(),
datetime.UTC,
)
@@ -442,7 +442,7 @@ def test_key_length_returns_dsa_bits() -> None:
def test_certificate_records_its_latest_self_signature() -> None:
- key, _ = keys_writer.openpgp.PublicKey.from_armor(_EMBEDDED_V4_EXPIRING_KEY_ASC)
+ key, _ = keys_writer.openpgp.composed.SignedPublicKey.from_armor(_EMBEDDED_V4_EXPIRING_KEY_ASC)
data = MockData(None, committees_after_commit={})
writer, _write, _write_as = _make_foundation_committer_with_audit(data)
@@ -453,21 +453,21 @@ def test_certificate_records_its_latest_self_signature() -> None:
assert latest_self_signature is not None
assert key_model.latest_self_signature == datetime.datetime.fromtimestamp(
- latest_self_signature.creation_time, datetime.UTC
+ latest_self_signature.created(), datetime.UTC
)
def test_expiry_is_recorded_against_the_signing_key_not_the_certificate() -> None:
- key, _ = keys_writer.openpgp.PublicKey.from_armor(_EMBEDDED_V4_EXPIRING_KEY_ASC)
- binding = next(iter(key.user_bindings()))
- binding_signature = binding.signatures[0]
+ key, _ = keys_writer.openpgp.composed.SignedPublicKey.from_armor(_EMBEDDED_V4_EXPIRING_KEY_ASC)
+ user = next(iter(key.details.users))
+ binding_signature = user.signatures[0]
# Expiry is per key, so it belongs to the SigningKey rows and not to the certificate above them
assert not hasattr(sql.SigningCertificate, "expires")
primary = next(facts for facts in pgp.signing_key_facts(key) if facts.is_primary)
assert primary.expires == datetime.datetime.fromtimestamp(
- key.created_at + binding_signature.key_expiration_seconds, datetime.UTC
+ key.created_at + binding_signature.key_expiration_time(), datetime.UTC
)
diff --git a/tests/unit/test_pgp.py b/tests/unit/test_pgp.py
index 77740ecb..1decaeea 100644
--- a/tests/unit/test_pgp.py
+++ b/tests/unit/test_pgp.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+from collections.abc import Iterator
from types import SimpleNamespace
import openpgp
@@ -23,12 +24,43 @@ import atr.pgp as pgp
import tests.unit.pgp_fixtures as pgp_fixtures
+def _block_with_grafted_signature(target_block: str, source_block: str, signature_type: int, after_tag: int) -> str:
+ # Lift a signature packet out of one certificate and splice it into another, just after the first
+ # packet of `after_tag`, which is what a forged revocation dropped into a KEYS file looks like. The
+ # signature was made by a different key, so it can't verify against the target
+ source = openpgp.composed.SignedPublicKey.from_armor(source_block)[0].to_bytes()
+ grafted = next(
+ source[start:end]
+ for tag, start, end, body in _packets(source)
+ if (tag == 2) and (source[body] == 4) and (source[body + 1] == signature_type)
+ )
+ target = openpgp.composed.SignedPublicKey.from_armor(target_block)[0].to_bytes()
+ kept = bytearray()
+ inserted = False
+ for tag, start, end, _body in _packets(target):
+ kept += target[start:end]
+ if (tag == after_tag) and not inserted:
+ kept += grafted
+ inserted = True
+ return openpgp.composed.SignedPublicKey.from_bytes(bytes(kept)).to_armored()
+
+
def _block_without_signature_type(block: str, signature_type: int) -> str:
# Re-armour a certificate with every self-signature of a given type removed, which is what a
# stripped re-upload looks like at the packet level
- key, _ = openpgp.PublicKey.from_armor(block)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(block)
data = key.to_bytes()
kept = bytearray()
+ for tag, start, end, body in _packets(data):
+ if not ((tag == 2) and (data[body] == 4) and (data[body + 1] == signature_type)):
+ kept += data[start:end]
+ stripped = openpgp.composed.SignedPublicKey.from_bytes(bytes(kept))
+ return stripped.to_armored()
+
+
+def _packets(data: bytes) -> Iterator[tuple[int, int, int, int]]:
+ # Walk an OpenPGP byte stream, yielding (tag, start, end, body) per packet so a test can rewrite a
+ # certificate at the packet level. Covers both old and new-format framing
index = 0
while index < len(data):
start = index
@@ -53,32 +85,29 @@ def _block_without_signature_type(block: str, signature_type: int) -> str:
index += width
body = index
index += length
- if not ((tag == 2) and (data[body] == 4) and (data[body + 1] == signature_type)):
- kept += data[start:index]
- stripped = openpgp.PublicKey.from_bytes(bytes(kept))
- return (stripped[0] if isinstance(stripped, tuple) else stripped).to_armored()
+ yield tag, start, index, body
def test_latest_self_signature_skips_uid_revocations() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
latest = pgp.latest_self_signature(key)
assert latest is not None
- assert latest.signature_type == "cert-positive"
+ assert latest.typ() == "cert-positive"
def test_latest_self_signature_survives_a_revoked_primary_uid() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_UID_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_UID_PUBLIC_KEY_ASC)
latest = pgp.latest_self_signature(key)
assert latest is not None
- assert latest.signature_type == "cert-positive"
+ assert latest.typ() == "cert-positive"
def test_signing_key_status_expiry_follows_the_issuing_subkey() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
status = pgp.signing_key_status(key, {pgp_fixtures.EXPIRED_SUBKEY_SIGNING_FINGERPRINT}, set())
@@ -87,7 +116,7 @@ def test_signing_key_status_expiry_follows_the_issuing_subkey() -> None:
def test_signing_key_status_expiry_ignores_a_later_primary_expiry() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
primary_expires = pgp.key_expires_at(key)
status = pgp.signing_key_status(key, {pgp_fixtures.EXPIRED_SUBKEY_SIGNING_FINGERPRINT}, set())
@@ -98,7 +127,7 @@ def test_signing_key_status_expiry_ignores_a_later_primary_expiry() -> None:
def test_signing_key_status_reports_an_issuer_naming_no_held_key_as_unidentified() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
status = pgp.signing_key_status(key, set(), set())
@@ -109,7 +138,7 @@ def test_signing_key_status_reports_an_issuer_naming_no_held_key_as_unidentified
def test_signing_key_status_reads_the_primary_when_the_primary_issued_the_signature() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
status = pgp.signing_key_status(key, {pgp_fixtures.EXPIRED_SUBKEY_PRIMARY_FINGERPRINT}, set())
@@ -118,7 +147,7 @@ def test_signing_key_status_reads_the_primary_when_the_primary_issued_the_signat
def test_signing_key_status_refuses_a_certify_only_primary_which_issued_the_signature() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
status = pgp.signing_key_status(key, {pgp_fixtures.EXPIRED_SUBKEY_PRIMARY_FINGERPRINT}, set())
@@ -126,7 +155,7 @@ def test_signing_key_status_refuses_a_certify_only_primary_which_issued_the_sign
def test_signing_key_status_permits_a_key_declaring_signing() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
status = pgp.signing_key_status(key, {pgp_fixtures.EXPIRED_SUBKEY_SIGNING_FINGERPRINT}, set())
@@ -134,7 +163,7 @@ def test_signing_key_status_permits_a_key_declaring_signing() -> None:
def test_signing_key_status_refuses_a_key_declaring_capabilities_without_signing() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
status = pgp.signing_key_status(key, {pgp_fixtures.EXPIRED_SUBKEY_ENCRYPTION_FINGERPRINT}, set())
@@ -145,7 +174,7 @@ def test_declares_signing_permits_a_key_declaring_no_capabilities() -> None:
# An absent key flags subpacket is indistinguishable from one declaring nothing, and older keys
# often declare nothing at all, so silence must not be read as a refusal to sign
signature = SimpleNamespace(
- key_flags=SimpleNamespace(
+ key_flags=lambda: SimpleNamespace(
certify=False,
sign=False,
encrypt_communications=False,
@@ -163,7 +192,7 @@ def test_declares_signing_refuses_a_key_whose_self_signature_cannot_be_read() ->
def test_signing_key_status_flags_a_revoked_signing_subkey() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC)
status = pgp.signing_key_status(key, {pgp_fixtures.REVOKED_SUBKEY_SIGNING_FINGERPRINT}, set())
@@ -171,7 +200,7 @@ def test_signing_key_status_flags_a_revoked_signing_subkey() -> None:
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)
+ key, _ = openpgp.composed.SignedPublicKey.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())
@@ -180,15 +209,48 @@ def test_signing_key_status_flags_a_subkey_beneath_a_revoked_primary() -> None:
def test_signing_key_status_flags_a_revoked_primary() -> None:
- key, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC)
+ key, _ = openpgp.composed.SignedPublicKey.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_ignores_a_forged_primary_revocation() -> None:
+ # A key-revocation lifted from another key and dropped in front of this one verifies against
+ # nothing here, so its mere presence must not block the key. Otherwise anyone could deny a project
+ # its releases just by planting a revocation packet in a KEYS file
+ forged = _block_with_grafted_signature(
+ pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC, pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC, 0x20, 6
+ )
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(forged)
+
+ # The forged packet really is present, so this shows verification rejects it rather than absence
+ assert any(signature.typ() == "key-revocation" for signature in key.details.revocation_signatures)
+ status = pgp.signing_key_status(key, {pgp_fixtures.EXPIRED_SUBKEY_PRIMARY_FINGERPRINT}, set())
+ assert status.revoked is False
+
+
+def test_signing_key_status_ignores_a_forged_subkey_revocation() -> None:
+ # The same forgery aimed at a subkey: a subkey-revocation made by another key can't verify as a
+ # binding on this primary, so it must not revoke the subkey it sits beneath
+ forged = _block_with_grafted_signature(
+ pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC, pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC, 0x28, 14
+ )
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(forged)
+
+ signing_subkey = next(
+ subkey
+ for subkey in key.public_subkeys
+ if subkey.key.fingerprint.lower() == pgp_fixtures.EXPIRED_SUBKEY_SIGNING_FINGERPRINT
+ )
+ assert any(signature.typ() == "subkey-revocation" for signature in signing_subkey.signatures)
+ status = pgp.signing_key_status(key, {pgp_fixtures.EXPIRED_SUBKEY_SIGNING_FINGERPRINT}, set())
+ assert status.revoked is False
+
+
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)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
status = pgp.signing_key_status(key, {pgp_fixtures.REVOKED_UID_FINGERPRINT}, set())
@@ -198,7 +260,7 @@ def test_signing_key_status_does_not_treat_a_revoked_user_id_as_a_revoked_key()
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)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
expires = pgp.key_expires_at(key)
@@ -209,7 +271,7 @@ def test_key_expires_at_ignores_a_revoked_user_id_self_signature() -> None:
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)
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
status = pgp.signing_key_status(key, {pgp_fixtures.REVOKED_UID_FINGERPRINT}, set())
@@ -217,8 +279,8 @@ def test_signing_key_status_reads_capabilities_past_a_revoked_user_id() -> None:
def test_revocations_dropped_detects_a_revocation_removed_on_re_upload() -> None:
- stored, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC)
- incoming, _ = openpgp.PublicKey.from_armor(
+ stored, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC)
+ incoming, _ = openpgp.composed.SignedPublicKey.from_armor(
_block_without_signature_type(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC, 0x20)
)
@@ -229,17 +291,17 @@ def test_revocations_dropped_detects_a_revocation_removed_on_re_upload() -> None
def test_revocations_dropped_permits_a_revocation_added_on_re_upload() -> None:
# Gap A relies on a genuine new revocation still being able to arrive, so growth is never a drop
- stored, _ = openpgp.PublicKey.from_armor(
+ stored, _ = openpgp.composed.SignedPublicKey.from_armor(
_block_without_signature_type(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC, 0x20)
)
- incoming, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC)
+ incoming, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC)
assert pgp.revocations_dropped(stored, incoming) == set()
def test_revocations_dropped_detects_a_subkey_revocation_removed_on_re_upload() -> None:
- stored, _ = openpgp.PublicKey.from_armor(pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC)
- incoming, _ = openpgp.PublicKey.from_armor(
+ stored, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC)
+ incoming, _ = openpgp.composed.SignedPublicKey.from_armor(
_block_without_signature_type(pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC, 0x28)
)
diff --git a/tests/unit/test_signature_provenance.py b/tests/unit/test_signature_provenance.py
index 2613be1c..6e617414 100644
--- a/tests/unit/test_signature_provenance.py
+++ b/tests/unit/test_signature_provenance.py
@@ -169,8 +169,8 @@ async def test_match_release_no_match_wrong_hash(tmp_path: pathlib.Path) -> None
@pytest.mark.asyncio
async def test_resolve_signing_key_from_signature_rejects_mismatched_issuer_metadata() -> None:
- parsed_key, _ = atr.api.openpgp.PublicKey.from_armor(_EMBEDDED_SUBKEY_PUBLIC_KEY_ASC)
- subkey = next(iter(parsed_key.subkey_bindings()))
+ parsed_key, _ = atr.api.openpgp.composed.SignedPublicKey.from_armor(_EMBEDDED_SUBKEY_PUBLIC_KEY_ASC)
+ subkey = next(iter(parsed_key.public_subkeys))
stored = SimpleNamespace(
fingerprint=parsed_key.fingerprint.lower(),
ascii_armored_key=_EMBEDDED_SUBKEY_PUBLIC_KEY_ASC,
@@ -182,14 +182,14 @@ async def test_resolve_signing_key_from_signature_rejects_mismatched_issuer_meta
await atr.api._resolve_signing_key_from_signature(
db_data,
issuer_fingerprints={parsed_key.fingerprint.lower()},
- issuer_key_ids={subkey.key_id.lower()},
+ issuer_key_ids={subkey.key.key_id.lower()},
)
@pytest.mark.asyncio
async def test_resolve_signing_key_from_signature_returns_subkey_fingerprint() -> None:
- parsed_key, _ = atr.api.openpgp.PublicKey.from_armor(_EMBEDDED_SUBKEY_PUBLIC_KEY_ASC)
- subkey = next(iter(parsed_key.subkey_bindings()))
+ parsed_key, _ = atr.api.openpgp.composed.SignedPublicKey.from_armor(_EMBEDDED_SUBKEY_PUBLIC_KEY_ASC)
+ subkey = next(iter(parsed_key.public_subkeys))
stored = SimpleNamespace(
fingerprint=parsed_key.fingerprint.lower(),
ascii_armored_key=_EMBEDDED_SUBKEY_PUBLIC_KEY_ASC,
@@ -200,11 +200,11 @@ async def test_resolve_signing_key_from_signature_returns_subkey_fingerprint() -
resolved, signer_fingerprint = await atr.api._resolve_signing_key_from_signature(
db_data,
issuer_fingerprints=set(),
- issuer_key_ids={subkey.key_id.lower()},
+ issuer_key_ids={subkey.key.key_id.lower()},
)
assert resolved is stored
- assert signer_fingerprint == subkey.fingerprint.lower()
+ assert signer_fingerprint == subkey.key.fingerprint.lower()
@pytest.mark.asyncio
@@ -407,7 +407,7 @@ async def _call_signature_provenance(
monkeypatch: pytest.MonkeyPatch,
matched: bool,
) -> tuple[dict, int]:
- parsed_key, _ = atr.api.openpgp.PublicKey.from_armor(_EMBEDDED_SUBKEY_PUBLIC_KEY_ASC)
+ parsed_key, _ = atr.api.openpgp.composed.SignedPublicKey.from_armor(_EMBEDDED_SUBKEY_PUBLIC_KEY_ASC)
committee = SimpleNamespace(key="example", is_podling=False)
stored = SimpleNamespace(
fingerprint=parsed_key.fingerprint.lower(),
@@ -415,16 +415,16 @@ async def _call_signature_provenance(
committees=[committee],
)
signature = SimpleNamespace(
- signature_info=lambda: SimpleNamespace(
- issuer_fingerprints={parsed_key.fingerprint.lower()},
- issuer_key_ids=set(),
+ signature=SimpleNamespace(
+ issuer_fingerprint=lambda: {parsed_key.fingerprint.lower()},
+ issuer_key_id=lambda: set(),
)
)
matched_committees = [committee] if matched else []
monkeypatch.setattr(atr.api.db, "session", _mock_session_factory(MockKeyDBSession([stored])))
monkeypatch.setattr(
- atr.api.openpgp,
+ atr.api.openpgp.composed,
"DetachedSignature",
SimpleNamespace(from_armor=lambda _text: (signature, "")),
)
diff --git a/uv.lock b/uv.lock
index 5e4c9928..3c1c9f20 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1911,18 +1911,18 @@ wheels = [
[[package]]
name = "rpgp-py"
-version = "0.19.8"
+version = "0.20.0"
source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e9/3b/7d627a66b12066fc38edf517cba7558e6c1daa141a5a00a4735529e160aa/rpgp_py-0.19.8.tar.gz", hash = "sha256:968b73a9c4a6dff06c106945201b147413111f0cce48b8e5adf4ab9e86142d14", size = 131745, upload-time = "2026-06-13T10:34:07.892Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/d1/07/761c1c03da215f1db06f63af56a10f700e7158b1676669324b236cbef935/rpgp_py-0.20.0.tar.gz", hash = "sha256:2a072023dfd034f914266e34b91d54b415565a12a2a177c23e720714c1ebf8a1", size = 144237, upload-time = "2026-07-23T22:48:29.735Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/bb/cb/3d049b94d36c75f10fc6ff30aaa7a242c9d4ba8fc4b9064eb1427534916d/rpgp_py-0.19.8-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:3800b63bf0a7899f9f531a60d943902d56f8775f49a7c7af8c5dfa5b58f36c0b", size = 3542161, upload-time = "2026-06-13T10:33:56.639Z" },
- { url = "https://files.pythonhosted.org/packages/1d/80/e227c688d212a7f7f86e86957b204501c836ba7196c9f2d43607bb4b8f4e/rpgp_py-0.19.8-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:097b10c6b31c4fba71f71415723168f8dc73f7df7e756626ccea9f2e2e6d4bc7", size = 3301523, upload-time = "2026-06-13T10:33:58.035Z" },
- { url = "https://files.pythonhosted.org/packages/64/d8/5248a999469edc8a67fcf35a4d7efcc40246412347179f2b8e8b3a54d2b8/rpgp_py-0.19.8-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:3dbf89c162cc50e1b4a0ce2411978cdf79f64cfb2df72494ee41ad9a5552c28e", size = 3621544, upload-time = "2026-06-13T10:33:59.509Z" },
- { url = "https://files.pythonhosted.org/packages/f7/c2/c96e894b4986a9b49aa28ed99b54cda6eae7cb2675f6536ad7a54d863da9/rpgp_py-0.19.8-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f89a2dfa972840b5997a6246774e9c385d3f81a04d64eccb1ed479e280d137e3", size = 3750464, upload-time = "2026-06-13T10:34:00.834Z" },
- { url = "https://files.pythonhosted.org/packages/66/b2/2341b1248d93c50130732d933031dacd474d990e8545c9f1c8826034bad1/rpgp_py-0.19.8-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:979fda6b9d33af2406a3341c377716dd488edb1ab391f30c24dcd245c9927b8d", size = 3819997, upload-time = "2026-06-13T10:34:02.852Z" },
- { url = "https://files.pythonhosted.org/packages/56/81/2844751c10718bd773d00fbe653f037bf88ef78289b6f56483488c452373/rpgp_py-0.19.8-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:0d2152067d02ddcc623288f688953ba6a9d6a95818e3e72b5913b0f2e52b3170", size = 4031482, upload-time = "2026-06-13T10:34:04.213Z" },
- { url = "https://files.pythonhosted.org/packages/79/26/06caaee58db0fcb16fdfdc16070ef535eb333d0219fa33def27df3f2d81b/rpgp_py-0.19.8-cp310-abi3-win32.whl", hash = "sha256:035ab197298cbb979ea784c27262bfcde90f544b8d8bd2cf9ce86ef8e235cde5", size = 3125027, upload-time = "2026-06-13T10:34:05.429Z" },
- { url = "https://files.pythonhosted.org/packages/82/f5/cb03284487e6d14a7d4df280fde001e0effef6be1c8ca479d007c977ac37/rpgp_py-0.19.8-cp310-abi3-win_amd64.whl", hash = "sha256:6ca6d22f98baa175faad7bc92c21c6ebcd64be6fd499524320dcb560ea31001f", size = 3391454, upload-time = "2026-06-13T10:34:06.683Z" },
+ { url = "https://files.pythonhosted.org/packages/0d/10/e14898866810360e5394d10997660500e080224a57b8df1a95a9b30d54d7/rpgp_py-0.20.0-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:618354b938c629661f9a19e290d06dbd41c20ca908cf47058c25aa629c7ccb82", size = 3125475, upload-time = "2026-07-23T22:48:18.913Z" },
+ { url = "https://files.pythonhosted.org/packages/6c/bd/45bbe117b7a56e1ab8cd6742becaadffc879f460be6a89d16d0fd9765ecc/rpgp_py-0.20.0-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:63286e891a0910f9d2e407b94cb21b274dcdc5d3f1e60d101b884667ce0a3c70", size = 2965414, upload-time = "2026-07-23T22:48:20.392Z" },
+ { url = "https://files.pythonhosted.org/packages/4e/5a/480757bf4495abd7649c35580aa59ada149c57ae386349ad8eab08d8ad6d/rpgp_py-0.20.0-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:fca326cbcbe0102bba1c2628ccaee2d75229fe50195c90bb646d79a55a330b0e", size = 3284497, upload-time = "2026-07-23T22:48:21.714Z" },
+ { url = "https://files.pythonhosted.org/packages/e6/f9/d53041547713dfec33f035d029e15b4127c4d02dcab8fbeb4af8795fd16e/rpgp_py-0.20.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:499c28ba70b2e2da214e9463369a6d929062e8e7037447b6d45a9d2663f3ba43", size = 3322931, upload-time = "2026-07-23T22:48:23.104Z" },
+ { url = "https://files.pythonhosted.org/packages/8c/a1/b122189c65bce7bdd60d1e2861b663c32d583f6c259c1ddd1da56d0aee50/rpgp_py-0.20.0-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:95881d16420949c6b893bff60276aa5da0683ae86e4cab98506b54f7b293c8f2", size = 3456963, upload-time = "2026-07-23T22:48:24.652Z" },
+ { url = "https://files.pythonhosted.org/packages/f2/02/7f953e4c2b681c917f1d27aa204042b9552a9eb674d7cefa0233b2302eac/rpgp_py-0.20.0-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:659475b9ba64cc42596e2bfa953f9cb7fd271486b2e2a957d3a7660baa6712d5", size = 3633922, upload-time = "2026-07-23T22:48:26.004Z" },
+ { url = "https://files.pythonhosted.org/packages/bf/7c/ad594a2810bb7093958d0ca9a4693db773c775f833510435e3a49ab33274/rpgp_py-0.20.0-cp310-abi3-win32.whl", hash = "sha256:f4802be4265103aa3ac5ac7882ac7388fe4b3bd3ad2257500f6ab5a8f2879637", size = 2832232, upload-time = "2026-07-23T22:48:27.321Z" },
+ { url = "https://files.pythonhosted.org/packages/bf/3f/868a9f090ecec6ec26d52522aab0f33e25184ec48a4ce260cf834fda48be/rpgp_py-0.20.0-cp310-abi3-win_amd64.whl", hash = "sha256:278d002db1a91a709f5fdc423e06adba0852e1ed2ad363ecb1b70decc9e80516", size = 3052667, upload-time = "2026-07-23T22:48:28.669Z" },
]
[[package]]
@@ -2178,7 +2178,7 @@ requires-dist = [
{ name = "quart-wtforms", specifier = ">=1.0.3,<2.0.0" },
{ name = "requests", specifier = ">=2.33.0" },
{ name = "rich", specifier = ">=14.0.0" },
- { name = "rpgp-py", specifier = "==0.19.8" },
+ { name = "rpgp-py", specifier = "==0.20.0" },
{ name = "semver", specifier = ">=3.0.4" },
{ name = "sqlmodel", specifier = ">=0.0.24,<0.1.0" },
{ name = "ssh-audit", specifier = ">=3.3.0" },
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org