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/17 19:19:10 UTC
(tooling-trusted-releases) branch sbp updated: Import all certificates in blocks containing multiple certificates
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 e70040af Import all certificates in blocks containing multiple certificates
e70040af is described below
commit e70040af03a48ddf789b6035c319fab61e35c868
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Mon Aug 17 20:18:18 2026 +0100
Import all certificates in blocks containing multiple certificates
---
atr/api/__init__.py | 2 +-
atr/storage/writers/keys.py | 88 ++++++++++++++++++++-----
tests/unit/pgp_fixtures.py | 44 +++++++++++++
tests/unit/test_api_key_results.py | 28 ++++++++
tests/unit/test_keys_writer.py | 129 +++++++++++++++++++++++++++++++++++++
tests/unit/test_pgp.py | 54 ++--------------
6 files changed, 279 insertions(+), 66 deletions(-)
diff --git a/atr/api/__init__.py b/atr/api/__init__.py
index 87185d9e..093e8ae6 100644
--- a/atr/api/__init__.py
+++ b/atr/api/__init__.py
@@ -728,7 +728,7 @@ async def key_add(
ocr, publications = await wafc.keys.ensure_stored_one(data.key)
try:
key = ocr.result_or_raise()
- except datatypes.UnknownApacheUidError as e:
+ except (datatypes.UnknownApacheUidError, ValueError) as e:
raise exceptions.BadRequest(str(e)) from e
for selected_committee_key in selected_committee_keys:
diff --git a/atr/storage/writers/keys.py b/atr/storage/writers/keys.py
index 9b8c2a65..05cddc25 100644
--- a/atr/storage/writers/keys.py
+++ b/atr/storage/writers/keys.py
@@ -127,6 +127,42 @@ def _block_downgrade_reason(fingerprint: str, stored_block: str, incoming_block:
return None
+def _deduplicate_keys(roes: list[datatypes.Key | Exception]) -> list[datatypes.Key | Exception]:
+ groups: dict[str, list[datatypes.Key]] = {}
+ for roe in roes:
+ if isinstance(roe, datatypes.Key):
+ groups.setdefault(roe.key_model.fingerprint, []).append(roe)
+ deduplicated: list[datatypes.Key | Exception] = []
+ seen: set[str] = set()
+ for roe in roes:
+ if not isinstance(roe, datatypes.Key):
+ deduplicated.append(roe)
+ continue
+ fingerprint = roe.key_model.fingerprint
+ if fingerprint in seen:
+ continue
+ seen.add(fingerprint)
+ keys = groups[fingerprint]
+ deduplicated.append(keys[0] if (len(keys) == 1) else _preferred_key(fingerprint, keys))
+ return deduplicated
+
+
+def _preferred_key(fingerprint: str, keys: list[datatypes.Key]) -> datatypes.Key | Exception:
+ candidates: dict[bytes, datatypes.Key] = {}
+ for key in keys:
+ parsed = pgp.certificate_for_fingerprint(key.key_model.ascii_armored_key, fingerprint)
+ if parsed is None:
+ return ValueError(f"Key {fingerprint} could not be read back from its block")
+ candidates.setdefault(parsed.to_bytes(), key)
+ if len(candidates) == 1:
+ return next(iter(candidates.values()))
+ for candidate in candidates.values():
+ others = [other for other in candidates.values() if other is not candidate]
+ if all(_supersedes(fingerprint, candidate, other) for other in others):
+ return candidate
+ return ValueError(f"Key {fingerprint} appears {len(candidates)} times with differing content")
+
+
def _signing_key_rows(certificate_fingerprint: str, block: str | bytes) -> list[dict] | None:
"""The SigningKey rows a certificate's block describes, or None if the block is for another key."""
if isinstance(block, bytes):
@@ -151,6 +187,14 @@ def _signing_key_rows(certificate_fingerprint: str, block: str | bytes) -> list[
]
+def _supersedes(fingerprint: str, candidate: datatypes.Key, other: datatypes.Key) -> bool:
+ candidate_block = candidate.key_model.ascii_armored_key
+ other_block = other.key_model.ascii_armored_key
+ if _block_downgrade_reason(fingerprint, other_block, candidate_block) is not None:
+ return False
+ return _block_downgrade_reason(fingerprint, candidate_block, other_block) is not None
+
+
async def _sync_signing_keys(data: db.Session, certificates: list[sql.SigningCertificate]) -> None:
"""Bring each certificate's SigningKey rows into line with the block it holds"""
rows = []
@@ -506,7 +550,13 @@ class FoundationCommitter(GeneralPublic):
return key
def __block_model_create(self, key_block: str, ldap_data: cache.EmailUidLookup) -> datatypes.Key:
- public_key, _ = openpgp.composed.SignedPublicKey.from_armor(key_block)
+ public_keys, _ = openpgp.composed.SignedPublicKey.from_armor_many(key_block)
+ if len(public_keys) != 1:
+ raise ValueError(
+ f"This block contains {util.plural(len(public_keys), 'key')};"
+ " add one key at a time, or import a KEYS file"
+ )
+ public_key = public_keys[0]
key_model = self.public_key_model(public_key, ldap_data, original_key_block=key_block)
_validate_key_strength(public_key)
return datatypes.Key(
@@ -928,21 +978,25 @@ class CommitteeParticipant(FoundationCommitter):
def __block_models(self, key_block: str, ldap_data: cache.EmailUidLookup) -> list[datatypes.Key | Exception]:
try:
- public_key, _ = openpgp.composed.SignedPublicKey.from_armor(key_block)
+ public_keys, _ = openpgp.composed.SignedPublicKey.from_armor_many(key_block)
except Exception as e:
raise ValueError(f"Error loading OpenPGP key block: {e}") from e
+ if not public_keys:
+ raise ValueError("Error loading OpenPGP key block: no keys found")
+ original_key_block = key_block if (len(public_keys) == 1) else None
key_list = []
- try:
- key_model = self.public_key_model(public_key, ldap_data, original_key_block=key_block)
- _validate_key_strength(public_key)
- key = datatypes.Key(
- status=datatypes.KeyStatus.PARSED,
- key_model=key_model,
- member_ids=sorted(util.openpgp_member_ids(public_key)),
- )
- key_list.append(key)
- except Exception as e:
- key_list.append(e)
+ for public_key in public_keys:
+ try:
+ key_model = self.public_key_model(public_key, ldap_data, original_key_block=original_key_block)
+ _validate_key_strength(public_key)
+ key = datatypes.Key(
+ status=datatypes.KeyStatus.PARSED,
+ key_model=key_model,
+ member_ids=sorted(util.openpgp_member_ids(public_key)),
+ )
+ key_list.append(key)
+ except Exception as e:
+ key_list.append(e)
return key_list
async def __database_add_models(
@@ -1039,7 +1093,7 @@ class CommitteeParticipant(FoundationCommitter):
outcomes.update_roes(Exception, replace_with_inserted)
- persisted_fingerprints = {v["fingerprint"] for v in key_values}
+ persisted_fingerprints = {v["fingerprint"] for v in key_values} | downgraded
# An artifact points at the SigningKey which signed it, so these rows must exist before an
# artifact can be attributed
await _sync_signing_keys(self.__data, [key.key_model for key in key_list])
@@ -1115,13 +1169,15 @@ class CommitteeParticipant(FoundationCommitter):
for key_block in key_blocks
]
key_model_batches = await asyncio.gather(*tasks, return_exceptions=True)
+ roes: list[datatypes.Key | Exception] = []
for key_model_batch in key_model_batches:
if isinstance(key_model_batch, Exception):
- outcomes.append_error(key_model_batch)
+ roes.append(key_model_batch)
elif isinstance(key_model_batch, BaseException):
raise key_model_batch
else:
- outcomes.extend_roes(Exception, key_model_batch)
+ roes.extend(key_model_batch)
+ outcomes.extend_roes(Exception, await asyncio.to_thread(_deduplicate_keys, roes))
# Try adding the keys to the database
# If not, all keys will be replaced with a PostParseError
return await self.__database_add_models(outcomes, associate=associate)
diff --git a/tests/unit/pgp_fixtures.py b/tests/unit/pgp_fixtures.py
index 9128f440..58fb7e5d 100644
--- a/tests/unit/pgp_fixtures.py
+++ b/tests/unit/pgp_fixtures.py
@@ -17,6 +17,7 @@
import io
import pathlib
+from collections.abc import Iterator
from typing import Final
import openpgp.armor
@@ -106,6 +107,49 @@ hZnIBFP9OA8A/1t15vvaSJjtPZEKGFT7POk+2Z5Q2krVupn6V9oW4oDpAQDig6j9
"""
+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.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
+ header = data[index]
+ index += 1
+ if header & 0x40:
+ tag = header & 0x3F
+ first = data[index]
+ index += 1
+ if first < 192:
+ length = first
+ elif first < 224:
+ length = ((first - 192) << 8) + data[index] + 192
+ index += 1
+ else:
+ length = int.from_bytes(data[index : index + 4], "big")
+ index += 4
+ else:
+ tag = (header >> 2) & 0x0F
+ width = (1, 2, 4)[header & 0x03]
+ length = int.from_bytes(data[index : index + width], "big")
+ index += width
+ body = index
+ index += length
+ yield tag, start, index, body
+
+
def two_certificate_block(first_armored: str, second_armored: str) -> str:
first, _ = openpgp.composed.SignedPublicKey.from_armor(first_armored)
second, _ = openpgp.composed.SignedPublicKey.from_armor(second_armored)
diff --git a/tests/unit/test_api_key_results.py b/tests/unit/test_api_key_results.py
index 83d88051..7c9225b6 100644
--- a/tests/unit/test_api_key_results.py
+++ b/tests/unit/test_api_key_results.py
@@ -15,7 +15,15 @@
# specific language governing permissions and limitations
# under the License.
+import contextlib
+import inspect
+import unittest.mock as mock
+
+import pytest
+
+import atr.api
import atr.models as models
+import atr.storage.outcome as outcome
def test_key_delete_results_carry_publication_warnings() -> None:
@@ -29,3 +37,23 @@ def test_key_delete_results_carry_publication_warnings() -> None:
assert parsed.warnings == [warning]
assert models.api.KeyDeleteResults(endpoint="/key/delete", success=True).warnings == []
+
+
+@pytest.mark.asyncio
+async def test_key_add_reports_a_multi_key_block_as_a_bad_request() -> None:
+ write = mock.MagicMock()
+ write.as_foundation_committer.return_value.keys.ensure_stored_one = mock.AsyncMock(
+ return_value=(outcome.Error(ValueError("This block contains 2 keys; add one key at a time")), {})
+ )
+
+ @contextlib.asynccontextmanager
+ async def write_context(_asf_uid):
+ yield write
+
+ args = models.api.KeyAddArgs(asfuid="alice", key="-----BEGIN PGP PUBLIC KEY BLOCK-----\n", committees=[])
+ with (
+ mock.patch.object(atr.api, "_jwt_asf_uid", return_value="alice"),
+ mock.patch.object(atr.api.storage, "write", new=write_context),
+ pytest.raises(atr.api.exceptions.BadRequest, match="contains 2 keys"),
+ ):
+ await inspect.unwrap(atr.api.key_add)("key/add", args)
diff --git a/tests/unit/test_keys_writer.py b/tests/unit/test_keys_writer.py
index 5eecad95..1ac9495b 100644
--- a/tests/unit/test_keys_writer.py
+++ b/tests/unit/test_keys_writer.py
@@ -116,6 +116,57 @@ def test_block_downgrade_reason_compares_the_named_certificate() -> None:
assert keys_writer._block_downgrade_reason(fingerprint, stored, pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC) is None
+def test_block_models_splits_a_two_certificate_block() -> None:
+ writer, _write_as = _make_committee_member(MockData(None, committees_after_commit={}), "test")
+ block = pgp_fixtures.two_certificate_block(
+ pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC, pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC
+ )
+
+ keys = writer._CommitteeParticipant__block_models(block, keys_writer.cache.EmailUidLookup({}))
+
+ fingerprints = [key.key_model.fingerprint for key in keys if isinstance(key, keys_writer.datatypes.Key)]
+ assert fingerprints == [
+ pgp_fixtures.EXPIRED_SUBKEY_PRIMARY_FINGERPRINT,
+ pgp_fixtures.REVOKED_SUBKEY_PRIMARY_FINGERPRINT,
+ ]
+ for key in keys:
+ assert isinstance(key, keys_writer.datatypes.Key)
+ own = pgp.certificate_for_fingerprint(key.key_model.ascii_armored_key, key.key_model.fingerprint)
+ assert own is not None
+ assert key.key_model.ascii_armored_key != block
+
+
+def test_block_models_keeps_the_uploaded_text_of_a_single_certificate_block() -> None:
+ writer, _write_as = _make_committee_member(MockData(None, committees_after_commit={}), "test")
+
+ keys = writer._CommitteeParticipant__block_models(
+ pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC, keys_writer.cache.EmailUidLookup({})
+ )
+
+ assert len(keys) == 1
+ assert isinstance(keys[0], keys_writer.datatypes.Key)
+ assert keys[0].key_model.ascii_armored_key == pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC
+
+
+def test_block_models_reports_a_failing_certificate_without_losing_its_siblings() -> None:
+ writer, _write_as = _make_committee_member(MockData(None, committees_after_commit={}), "test")
+ block = pgp_fixtures.two_certificate_block(
+ pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC, pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC
+ )
+ strengths = iter([None, ValueError("too weak")])
+
+ def validate(_public_key) -> None:
+ if (error := next(strengths)) is not None:
+ raise error
+
+ with mock.patch.object(keys_writer, "_validate_key_strength", side_effect=validate):
+ keys = writer._CommitteeParticipant__block_models(block, keys_writer.cache.EmailUidLookup({}))
+
+ assert isinstance(keys[0], keys_writer.datatypes.Key)
+ assert keys[0].key_model.fingerprint == pgp_fixtures.EXPIRED_SUBKEY_PRIMARY_FINGERPRINT
+ assert isinstance(keys[1], ValueError)
+
+
@pytest.mark.asyncio
async def test_database_add_model_audits_inserted_key():
data = MockData(None, committees_after_commit={})
@@ -147,6 +198,63 @@ async def test_database_add_model_audits_inserted_key():
assert audit_kwargs["key_apache_uid"] == "alice"
+def test_deduplicate_keys_keeps_identical_copies_once() -> None:
+ armored = pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC
+ re_armored = keys_writer.openpgp.composed.SignedPublicKey.from_armor(armored)[0].to_armored()
+
+ deduplicated = keys_writer._deduplicate_keys([_parsed_key(armored), _parsed_key(re_armored)])
+
+ assert len(deduplicated) == 1
+ assert isinstance(deduplicated[0], keys_writer.datatypes.Key)
+
+
+def test_deduplicate_keys_prefers_the_copy_which_supersedes_the_other() -> None:
+ full = pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC
+ stripped = pgp_fixtures.block_without_signature_type(full, 0x28)
+
+ deduplicated = keys_writer._deduplicate_keys([_parsed_key(stripped), _parsed_key(full)])
+
+ assert len(deduplicated) == 1
+ assert isinstance(deduplicated[0], keys_writer.datatypes.Key)
+ assert deduplicated[0].key_model.ascii_armored_key == full
+
+
+def test_deduplicate_keys_reports_incomparable_copies_for_that_key_only() -> None:
+ full = pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC
+ unbound = pgp_fixtures.block_without_signature_type(full, 0x18)
+ other = _parsed_key(pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC)
+
+ deduplicated = keys_writer._deduplicate_keys([_parsed_key(full), other, _parsed_key(unbound)])
+
+ assert len(deduplicated) == 2
+ assert isinstance(deduplicated[0], ValueError)
+ assert "differing content" in str(deduplicated[0])
+ assert deduplicated[1] is other
+
+
+@pytest.mark.asyncio
+async def test_database_add_models_links_a_stored_key_whose_reimport_is_a_downgrade(sqlite_data):
+ full = pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC
+ fingerprint = pgp_fixtures.REVOKED_SUBKEY_PRIMARY_FINGERPRINT
+ sqlite_data.add(keys_writer.sql.Committee(key="alpha"))
+ stored = _signing_certificate(fingerprint, apache_uid="alice")
+ stored.ascii_armored_key = full
+ sqlite_data.add(stored)
+ await sqlite_data.commit()
+ writer, _write_as = _make_committee_member(sqlite_data, "alpha")
+ outcomes = outcome.List[keys_writer.datatypes.Key]()
+ outcomes.append_result(_parsed_key(pgp_fixtures.block_without_signature_type(full, 0x28)))
+
+ with mock.patch.object(writer, "_recheck_committee_drafts", new_callable=mock.AsyncMock):
+ outcomes, _publications = await writer._CommitteeParticipant__database_add_models(outcomes)
+
+ assert outcomes.error_count == 0
+ kept = await sqlite_data.signing_certificate(fingerprint=fingerprint).get()
+ assert (kept is not None) and (kept.ascii_armored_key == full)
+ links = (await sqlite_data.execute(sqlmodel.select(keys_writer.sql.KeyLink))).all()
+ assert [(link[0].committee_key, link[0].key_fingerprint) for link in links] == [("alpha", fingerprint)]
+
+
@pytest.mark.asyncio
async def test_delete_committee_keys_audits_committed_delete_when_sync_fails(sqlite_data):
await _seed_committee_key(sqlite_data, "alpha", "fp1")
@@ -350,6 +458,20 @@ async def test_ensure_stored_one_accepts_test_key_without_email_cache() -> None:
database_add_model.assert_awaited_once()
+@pytest.mark.asyncio
+async def test_ensure_stored_one_rejects_a_block_with_two_keys() -> None:
+ data = MockData(None, committees_after_commit={})
+ writer, _write, _write_as = _make_foundation_committer_with_audit(data)
+ block = pgp_fixtures.two_certificate_block(
+ pgp_fixtures.EXPIRED_SUBKEY_PUBLIC_KEY_ASC, pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC
+ )
+
+ result, publications = await writer.ensure_stored_one(block)
+
+ assert publications == {}
+ assert "contains 2 keys" in str(result.error_or_none())
+
+
@pytest.mark.asyncio
async def test_ensure_stored_one_rejects_key_without_apache_uid() -> None:
data = MockData(None, committees_after_commit={})
@@ -693,6 +815,13 @@ def _make_foundation_committer_with_audit(data: MockData, asf_uid: str = "alice"
return keys_writer.FoundationCommitter(write, write_as, data), write, write_as
+def _parsed_key(armored: str) -> keys_writer.datatypes.Key:
+ key, _ = keys_writer.openpgp.composed.SignedPublicKey.from_armor(armored)
+ model = _signing_certificate(key.fingerprint.lower(), apache_uid=None)
+ model.ascii_armored_key = armored
+ return keys_writer.datatypes.Key(status=keys_writer.datatypes.KeyStatus.PARSED, key_model=model)
+
+
def _playwright_test_key_text() -> str:
key_path = (
pathlib.Path(__file__).resolve().parents[2] / "playwright" / "557F8D855DEF8BBE2DC5603B64C271BB87B7FE7B.asc"
diff --git a/tests/unit/test_pgp.py b/tests/unit/test_pgp.py
index a1d89c06..7f154a46 100644
--- a/tests/unit/test_pgp.py
+++ b/tests/unit/test_pgp.py
@@ -15,7 +15,6 @@
# specific language governing permissions and limitations
# under the License.
-from collections.abc import Iterator
from types import SimpleNamespace
import openpgp
@@ -32,13 +31,13 @@ def _block_with_grafted_signature(target_block: str, source_block: str, signatur
source = openpgp.composed.SignedPublicKey.from_armor(source_block)[0].to_bytes()
grafted = next(
source[start:end]
- for tag, start, end, body in _packets(source)
+ for tag, start, end, body in pgp_fixtures.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):
+ for tag, start, end, _body in pgp_fixtures.packets(target):
kept += target[start:end]
if (tag == after_tag) and not inserted:
kept += grafted
@@ -46,49 +45,6 @@ def _block_with_grafted_signature(target_block: str, source_block: str, signatur
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.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
- header = data[index]
- index += 1
- if header & 0x40:
- tag = header & 0x3F
- first = data[index]
- index += 1
- if first < 192:
- length = first
- elif first < 224:
- length = ((first - 192) << 8) + data[index] + 192
- index += 1
- else:
- length = int.from_bytes(data[index : index + 4], "big")
- index += 4
- else:
- tag = (header >> 2) & 0x0F
- width = (1, 2, 4)[header & 0x03]
- length = int.from_bytes(data[index : index + width], "big")
- index += width
- body = index
- index += length
- yield tag, start, index, body
-
-
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
@@ -303,7 +259,7 @@ 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.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)
+ pgp_fixtures.block_without_signature_type(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC, 0x20)
)
dropped = pgp.revocations_dropped(stored, incoming)
@@ -314,7 +270,7 @@ 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.composed.SignedPublicKey.from_armor(
- _block_without_signature_type(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC, 0x20)
+ pgp_fixtures.block_without_signature_type(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC, 0x20)
)
incoming, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC)
@@ -324,7 +280,7 @@ def test_revocations_dropped_permits_a_revocation_added_on_re_upload() -> None:
def test_revocations_dropped_detects_a_subkey_revocation_removed_on_re_upload() -> None:
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)
+ pgp_fixtures.block_without_signature_type(pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC, 0x28)
)
dropped = pgp.revocations_dropped(stored, incoming)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org