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/19 14:22:11 UTC
(tooling-trusted-releases) branch sbp updated: Add a very rough OpenPGP key component browser for admins
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 6e5affaf Add a very rough OpenPGP key component browser for admins
6e5affaf is described below
commit 6e5affaf2196bec5907e80410666ed71489dc33d
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Wed Aug 19 15:21:29 2026 +0100
Add a very rough OpenPGP key component browser for admins
---
atr/admin/__init__.py | 401 ++++++++++++++++++++++++++++++++++-
atr/pgp.py | 89 +++++++-
atr/templates/includes/topnav.html | 5 +
tests/unit/test_admin_keys_survey.py | 85 ++++++++
tests/unit/test_pgp.py | 17 ++
5 files changed, 592 insertions(+), 5 deletions(-)
diff --git a/atr/admin/__init__.py b/atr/admin/__init__.py
index bf4306f5..faa314b8 100644
--- a/atr/admin/__init__.py
+++ b/atr/admin/__init__.py
@@ -27,7 +27,7 @@ import pathlib
import statistics
import sys
import time
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable, Iterable, Mapping, Sequence
from typing import Annotated, Any, Final, Literal, NamedTuple
import aiofiles.os
@@ -79,12 +79,42 @@ ROUTES_MODULE: Final[Literal[True]] = True
_CERTIFICATE_FINDING_RANK: Final[dict[str, int]] = {"signing-keys": 1, "metadata": 2}
_MAXIMUM_PERMITTED_AGE_DAYS: Final = datetime.timedelta(days=90)
+_SURVEY_KIND_CLASSES: Final[dict[str, str]] = {
+ "primary": "text-bg-dark",
+ "user-id": "text-bg-info",
+ "user-attribute": "text-bg-info",
+ "subkey": "text-bg-primary",
+}
+_SURVEY_KIND_ICONS: Final[dict[str, str]] = {
+ "primary": "key-fill",
+ "user-id": "person",
+ "user-attribute": "image",
+ "subkey": "key",
+}
+_SURVEY_STATE_CLASSES: Final[dict[str, str]] = {
+ "live": "success",
+ "expired": "warning",
+ "revoked": "danger",
+ "deleted": "secondary",
+ "unreadable": "dark",
+}
class BrowseAsUserForm(form.Form):
uid: str = form.label("ASF UID", "Enter the ASF UID to browse as.")
+@dataclasses.dataclass(frozen=True)
+class CertificateReport:
+ certificate: sql.SigningCertificate
+ committees: list[str]
+ signing_keys: list[sql.SigningKey]
+ shape: str
+ certificates_in_block: int
+ version: int | None
+ components: list[pgp.Component]
+
+
class DeleteCommitteeKeysForm(form.Form):
committee_key: str = form.label("Committee", widget=form.Widget.SELECT)
confirm_delete: Literal["DELETE KEYS"] = form.label("Confirmation", "Type DELETE KEYS to confirm.")
@@ -102,6 +132,14 @@ class DeleteReleaseForm(form.Form):
return v
+@dataclasses.dataclass(frozen=True)
+class Distribution:
+ count: int
+ mean: float
+ maximum: int
+ histogram: dict[int, int]
+
+
class EditBannerForm(form.Form):
banner_markdown: str = form.label(
"Banner Markdown",
@@ -145,6 +183,12 @@ class RotateJwtKeyForm(form.Form):
confirm_rotate: Literal["ROTATE"] = form.label("Confirmation", "Type ROTATE to confirm.")
+@dataclasses.dataclass(frozen=True)
+class Survey:
+ committees: list[sql.Committee]
+ reports: list[CertificateReport]
+
+
class ValidateJwtForm(form.Form):
token: str = form.label("JWT", "Paste the JWT to validate.", widget=form.Widget.TEXTAREA)
@@ -1650,6 +1694,65 @@ async def keys_check_post(
)
+@admin.typed
+async def keys_survey_get(_session: web.Committer, _keys_survey: Literal["keys/survey"]) -> str:
+ async with db.session() as data:
+ survey = await _survey_load(data)
+ by_committee = _survey_by_committee(survey)
+ unlinked = [certificate_report for certificate_report in survey.reports if not certificate_report.committees]
+ findings = await asyncio.to_thread(_survey_findings, survey, unlinked)
+ page = htm.Block()
+ page.h1["Keys survey"]
+ page.p(".text-muted")[
+ "Every stored certificate block, grouped by the committees it is linked to. A certificate linked to "
+ "several committees is counted under each. Follow a committee to see the components bundled in each of "
+ "its certificates."
+ ]
+ _survey_summary(page, survey)
+ _survey_legend(page)
+ for committee in survey.committees:
+ reports = by_committee.get(committee.key, [])
+ _survey_heading(page, committee.key, committee.display_name, reports, linked=True)
+ _survey_tiles(page, committee.key, reports)
+ _survey_heading(page, "unlinked", "Not linked to any committee", unlinked, linked=False)
+ _survey_tiles(page, "unlinked", unlinked)
+ for certificate_report in unlinked:
+ page.append(_survey_card("unlinked", certificate_report, findings[certificate_report.certificate.fingerprint]))
+ return await template.blank(
+ "Keys survey",
+ content=page.collect(),
+ description="Survey of stored OpenPGP certificates by committee.",
+ )
+
+
+@admin.typed
+async def keys_survey_committee_get(
+ _session: web.Committer, _keys_survey: Literal["keys/survey"], committee_key: safe.CommitteeKey
+) -> str:
+ async with db.session() as data:
+ committee = await data.committee(key=str(committee_key)).demand(
+ exceptions.NotFound(f"Committee '{committee_key}' not found.")
+ )
+ survey = await _survey_load(data)
+ reports = _survey_by_committee(survey).get(committee.key, [])
+ findings = await asyncio.to_thread(_survey_findings, survey, reports)
+ page = htm.Block()
+ page.p[htm.a(".atr-back-link", href=util.as_url(keys_survey_get))["← Back to the keys survey"]]
+ page.h1["Keys survey"]
+ _survey_legend(page)
+ _survey_heading(page, committee.key, committee.display_name, reports, linked=False)
+ _survey_tiles(page, committee.key, reports)
+ for certificate_report in reports:
+ page.append(
+ _survey_card(committee.key, certificate_report, findings[certificate_report.certificate.fingerprint])
+ )
+ return await template.blank(
+ f"Keys survey for {committee.display_name}",
+ content=page.collect(),
+ description="Survey of one committee's OpenPGP certificates, component by component.",
+ )
+
+
@admin.typed
async def keys_update_get(
_session: web.Committer, _keys_update: Literal["keys/update"]
@@ -2869,6 +2972,302 @@ async def _rotate_jwt_key_page(rendered_form: htm.Element) -> str:
)
+def _survey_algorithm_label(component: pgp.Component) -> str:
+ facts = component.facts
+ if facts is None:
+ return "none"
+ return f"{facts.algorithm} {facts.length_bits}" if facts.length_bits else facts.algorithm
+
+
+def _survey_by_committee(survey: Survey) -> dict[str, list[CertificateReport]]:
+ by_committee: dict[str, list[CertificateReport]] = collections.defaultdict(list)
+ for certificate_report in survey.reports:
+ for committee_key in certificate_report.committees:
+ by_committee[committee_key].append(certificate_report)
+ return by_committee
+
+
+def _survey_card(section: str, certificate_report: CertificateReport, findings: list[tuple[str, str]]) -> htm.Element:
+ certificate = certificate_report.certificate
+ fingerprint = certificate.fingerprint.lower()
+ state = _survey_state(certificate_report)
+ notes = [f"v{certificate_report.version}"] if (certificate_report.version is not None) else []
+ if certificate_report.shape != "single":
+ notes.append(f"{certificate_report.shape} ({certificate_report.certificates_in_block} in block)")
+ notes.append(util.plural(len(certificate_report.signing_keys), "signing key row"))
+ header = htm.div(".card-header.py-1.d-flex.flex-wrap.gap-2.align-items-center")[
+ htm.span(f".badge.text-bg-{_SURVEY_STATE_CLASSES[state]}")[state],
+ htm.a(".font-monospace.small", href=util.as_url(get.keys.details, fingerprint=fingerprint))[fingerprint],
+ htm.span[certificate.apache_uid or htm.em(".text-muted")["no ASF uid"]],
+ htm.span(".text-muted.small")[" · ".join(notes)],
+ *(htm.span(".badge.text-bg-light")[committee] for committee in certificate_report.committees),
+ ]
+ components = htm.div(".d-flex.flex-wrap.gap-1")[
+ *(_survey_component(component) for component in certificate_report.components)
+ ]
+ problems = htm.ul(".small.text-danger.mb-0.mt-2")[*(htm.li[f"{kind}: {line}"] for kind, line in findings)]
+ body = htm.div(".card-body.py-2")[components, problems if findings else None]
+ return htm.div(f".card.mb-2.border-{_SURVEY_STATE_CLASSES[state]}", id=f"{section}-{fingerprint}")[header, body]
+
+
+def _survey_component(component: pgp.Component) -> htm.Element:
+ facts = component.facts
+ now = datetime.datetime.now(datetime.UTC)
+ expired = (facts is not None) and (facts.expires is not None) and (facts.expires <= now)
+ label = component.label
+ if facts is not None:
+ flags = "no binding" if (component.flags is None) else (component.flags or "no flags")
+ label = f"{facts.algorithm} {facts.length_bits or '?'} [{flags}] {facts.created:%Y-%m}"
+ if facts.expires is not None:
+ label += f" → {facts.expires:%Y-%m}"
+ classes = f".badge.{_SURVEY_KIND_CLASSES[component.kind]}"
+ if component.revoked:
+ classes += ".text-decoration-line-through"
+ states = [state for state, flagged in (("revoked", component.revoked), ("expired", expired)) if flagged]
+ title = "\n".join(
+ [
+ f"{component.kind} {component.label}",
+ f"{component.self_signatures} self-signatures, {component.other_signatures} other signatures",
+ *states,
+ ]
+ )
+ return htm.span(classes, title=title)[
+ htm.icon(_SURVEY_KIND_ICONS[component.kind]),
+ " ",
+ label,
+ " ",
+ htm.span(".opacity-75")[f"{component.self_signatures}+{component.other_signatures}"],
+ htm.icon("x-circle", ".ms-1") if component.revoked else None,
+ htm.icon("hourglass-bottom", ".ms-1") if expired else None,
+ ]
+
+
+def _survey_distribution(values: Iterable[int]) -> Distribution:
+ histogram = collections.Counter(values)
+ count = sum(histogram.values())
+ if count == 0:
+ return Distribution(0, 0.0, 0, {})
+ total = sum(value * times for value, times in histogram.items())
+ return Distribution(count, total / count, max(histogram), dict(sorted(histogram.items())))
+
+
+def _survey_findings(survey: Survey, reports: list[CertificateReport]) -> dict[str, list[tuple[str, str]]]:
+ texts = collections.Counter(
+ certificate_report.certificate.ascii_armored_key for certificate_report in survey.reports
+ )
+ return {
+ certificate_report.certificate.fingerprint: _certificate_block_lines(
+ certificate_report.certificate,
+ {signing_key.fingerprint.lower() for signing_key in certificate_report.signing_keys},
+ texts[certificate_report.certificate.ascii_armored_key] > 1,
+ )
+ for certificate_report in reports
+ }
+
+
+def _survey_flags_label(component: pgp.Component) -> str:
+ if component.flags is None:
+ return "unsigned"
+ return component.flags or "undeclared"
+
+
+def _survey_heading(
+ page: htm.Block, section: str, name: str, reports: list[CertificateReport], *, linked: bool
+) -> None:
+ states = collections.Counter(_survey_state(certificate_report) for certificate_report in reports)
+ title = htm.a(href=util.as_url(keys_survey_committee_get, committee_key=section))[name] if linked else name
+ page.h2(".h4.mt-4.d-flex.flex-wrap.gap-2.align-items-center", id=f"committee-{section}")[
+ title,
+ htm.span(".text-muted.fs-6")[util.plural(len(reports), "certificate")],
+ *(
+ htm.span(f".badge.text-bg-{_SURVEY_STATE_CLASSES[state]}.fs-6")[f"{count} {state}"]
+ for state, count in sorted(states.items())
+ ),
+ ]
+
+
+def _survey_legend(page: htm.Block) -> None:
+ states = [htm.span(f".badge.text-bg-{classes}")[state] for state, classes in _SURVEY_STATE_CLASSES.items()]
+ kinds = [
+ htm.span(f".badge.{classes}")[htm.icon(_SURVEY_KIND_ICONS[kind]), " ", kind]
+ for kind, classes in _SURVEY_KIND_CLASSES.items()
+ ]
+ page.h2(".h5.mt-4")["Legend"]
+ page.p(".d-flex.flex-wrap.gap-1.align-items-center")[
+ "Certificate state: ",
+ *states,
+ " Component kind: ",
+ *kinds,
+ htm.span(".text-muted.small.ms-2")[
+ "Each component shows self-signatures+other signatures; struck through when revoked, ",
+ htm.icon("hourglass-bottom"),
+ " when expired. A tile shows user IDs·subkeys and links to its card.",
+ ],
+ ]
+
+
+async def _survey_load(data: db.Session) -> Survey:
+ via = sql.validate_instrumented_attribute
+ committees = await data.committee().all()
+ certificates = await data.signing_certificate(deleted=db.NOT_SET).all()
+ signing_keys = (await data.execute(sqlmodel.select(sql.SigningKey))).scalars().all()
+ links = await data.execute(sqlmodel.select(via(sql.KeyLink.key_fingerprint), via(sql.KeyLink.committee_key)))
+ keys_by_certificate: dict[str, list[sql.SigningKey]] = collections.defaultdict(list)
+ for signing_key in signing_keys:
+ keys_by_certificate[signing_key.certificate_fingerprint.lower()].append(signing_key)
+ committees_by_certificate: dict[str, list[str]] = collections.defaultdict(list)
+ for fingerprint, committee_key in links.all():
+ committees_by_certificate[fingerprint.lower()].append(committee_key)
+ reports = await asyncio.to_thread(
+ lambda: [
+ _survey_report(
+ certificate,
+ sorted(committees_by_certificate.get(certificate.fingerprint.lower(), [])),
+ keys_by_certificate.get(certificate.fingerprint.lower(), []),
+ )
+ for certificate in certificates
+ ]
+ )
+ return Survey(sorted(committees, key=lambda committee: committee.key), reports)
+
+
+def _survey_of_kind(certificate_report: CertificateReport, kind: pgp.ComponentKind) -> list[pgp.Component]:
+ return [component for component in certificate_report.components if component.kind == kind]
+
+
+def _survey_report(
+ certificate: sql.SigningCertificate, committees: list[str], signing_keys: list[sql.SigningKey]
+) -> CertificateReport:
+ fingerprint = certificate.fingerprint.lower()
+ text = certificate.ascii_armored_key
+ if isinstance(text, bytes):
+ text = text.decode("utf-8", errors="replace")
+ try:
+ keys, _ = openpgp.composed.SignedPublicKey.from_armor_many(text)
+ except Exception:
+ return CertificateReport(certificate, committees, signing_keys, "unparseable", 0, None, [])
+ shape = pgp.certificate_block_shape(keys, fingerprint)
+ own = [key for key in keys if key.fingerprint.lower() == fingerprint]
+ if len(own) != 1:
+ return CertificateReport(certificate, committees, signing_keys, shape, len(keys), None, [])
+ components = pgp.certificate_components(own[0])
+ return CertificateReport(certificate, committees, signing_keys, shape, len(keys), own[0].version, components)
+
+
+def _survey_state(certificate_report: CertificateReport) -> str:
+ if certificate_report.certificate.deleted is not None:
+ return "deleted"
+ if not certificate_report.components:
+ return "unreadable"
+ primary = certificate_report.components[0]
+ if primary.revoked:
+ return "revoked"
+ expires = primary.facts.expires if (primary.facts is not None) else None
+ if (expires is not None) and (expires <= datetime.datetime.now(datetime.UTC)):
+ return "expired"
+ return "live"
+
+
+def _survey_statistics(survey: Survey) -> dict[str, Distribution]:
+ reports = survey.reports
+ readable = [certificate_report for certificate_report in reports if certificate_report.components]
+ parsed = [certificate_report for certificate_report in reports if certificate_report.shape != "unparseable"]
+ linked = collections.Counter(
+ committee for certificate_report in reports for committee in certificate_report.committees
+ )
+ user_ids = [c for certificate_report in readable for c in _survey_of_kind(certificate_report, "user-id")]
+ return {
+ "certificates per block": _survey_distribution(r.certificates_in_block for r in parsed),
+ "user IDs per certificate": _survey_distribution(len(_survey_of_kind(r, "user-id")) for r in readable),
+ "user attributes per certificate": _survey_distribution(
+ len(_survey_of_kind(r, "user-attribute")) for r in readable
+ ),
+ "subkeys per certificate": _survey_distribution(len(_survey_of_kind(r, "subkey")) for r in readable),
+ "signing key rows per certificate": _survey_distribution(len(r.signing_keys) for r in reports),
+ "self signatures per user ID": _survey_distribution(component.self_signatures for component in user_ids),
+ "other signatures per user ID": _survey_distribution(component.other_signatures for component in user_ids),
+ "committees per certificate": _survey_distribution(len(r.committees) for r in reports),
+ "certificates per committee": _survey_distribution(
+ linked.get(committee.key, 0) for committee in survey.committees
+ ),
+ }
+
+
+def _survey_summary(page: htm.Block, survey: Survey) -> None:
+ rows = [
+ htm.tr[
+ htm.td(".text-nowrap")[name],
+ htm.td(".text-nowrap.text-end")[str(measure.count)],
+ htm.td(".text-nowrap.text-end")[f"{measure.mean:.2f}"],
+ htm.td(".text-nowrap.text-end")[str(measure.maximum)],
+ htm.td(".small")[
+ *(htm.span(".text-nowrap.me-2")[f"{value}:{times}"] for value, times in measure.histogram.items())
+ ],
+ ]
+ for name, measure in _survey_statistics(survey).items()
+ ]
+ page.h2(".h5.mt-4")["Statistics"]
+ page.table(".table.table-sm.border")[
+ htm.thead[
+ htm.tr[htm.th["Measure"], htm.th["n"], htm.th["Mean"], htm.th["Max"], htm.th["Histogram value:count"]]
+ ],
+ htm.tbody[*rows],
+ ]
+ page.h2(".h5.mt-4")["Tallies"]
+ page.table(".table.table-sm.border")[
+ htm.tbody[
+ *(
+ htm.tr[
+ htm.td(".text-nowrap")[name],
+ htm.td(".d-flex.flex-wrap.gap-1")[
+ *(htm.span(".badge.text-bg-light")[f"{label} {count}"] for label, count in tally.most_common())
+ ],
+ ]
+ for name, tally in _survey_tallies(survey).items()
+ )
+ ]
+ ]
+
+
+def _survey_tallies(survey: Survey) -> dict[str, collections.Counter[str]]:
+ reports = survey.reports
+ versions = [f"v{r.version}" for r in reports if r.version is not None]
+ components = [component for certificate_report in reports for component in certificate_report.components]
+ return {
+ "block shapes": collections.Counter(r.shape for r in reports),
+ "certificate states": collections.Counter(_survey_state(r) for r in reports),
+ "certificate versions": collections.Counter(versions),
+ "primary key flags": collections.Counter(_survey_flags_label(c) for c in components if c.kind == "primary"),
+ "subkey flags": collections.Counter(_survey_flags_label(c) for c in components if c.kind == "subkey"),
+ "key algorithms": collections.Counter(_survey_algorithm_label(c) for c in components if c.facts is not None),
+ }
+
+
+def _survey_tile(section: str, certificate_report: CertificateReport) -> htm.Element:
+ certificate = certificate_report.certificate
+ fingerprint = certificate.fingerprint.lower()
+ state = _survey_state(certificate_report)
+ identities = sum(
+ 1 for component in certificate_report.components if component.kind in ("user-id", "user-attribute")
+ )
+ subkeys = sum(1 for component in certificate_report.components if component.kind == "subkey")
+ title = f"{fingerprint} {certificate.primary_declared_uid or ''}".strip()
+ href = "" if (section == "unlinked") else util.as_url(keys_survey_committee_get, committee_key=section)
+ return htm.a(
+ f".badge.text-bg-{_SURVEY_STATE_CLASSES[state]}.text-decoration-none",
+ href=f"{href}#{section}-{fingerprint}",
+ title=title,
+ )[f"{identities}·{subkeys}"]
+
+
+def _survey_tiles(page: htm.Block, section: str, reports: list[CertificateReport]) -> None:
+ if not reports:
+ return
+ page.div(".d-flex.flex-wrap.gap-1.mb-2")[
+ *(_survey_tile(section, certificate_report) for certificate_report in reports)
+ ]
+
+
async def _update_keys(asf_uid: str) -> int:
async def _log_process(process: asyncio.subprocess.Process) -> None:
try:
diff --git a/atr/pgp.py b/atr/pgp.py
index 811752ad..e2a8008f 100644
--- a/atr/pgp.py
+++ b/atr/pgp.py
@@ -17,7 +17,7 @@
import dataclasses
import datetime
-from typing import Final
+from typing import Final, Literal
import openpgp
@@ -31,6 +31,19 @@ _KEY_REVOCATION_SIGNATURE_TYPE: Final[str] = "key-revocation"
_SUBKEY_BINDING_SIGNATURE_TYPE: Final[str] = "subkey-binding"
_SUBKEY_REVOCATION_SIGNATURE_TYPE: Final[str] = "subkey-revocation"
+ComponentKind = Literal["primary", "user-id", "user-attribute", "subkey"]
+
+
+@dataclasses.dataclass(frozen=True)
+class Component:
+ kind: ComponentKind
+ label: str
+ flags: str | None
+ revoked: bool
+ self_signatures: int
+ other_signatures: int
+ facts: "SigningKeyFacts | None"
+
@dataclasses.dataclass(frozen=True)
class SigningKeyFacts:
@@ -75,6 +88,26 @@ def certificate_block_shape(keys: list[openpgp.composed.SignedPublicKey], finger
return "multi-own-first" if (fingerprints[0] == fingerprint.lower()) else "multi-own-not-first"
+def certificate_components(key: openpgp.composed.SignedPublicKey) -> list[Component]:
+ fingerprint = key.fingerprint.lower()
+ key_id = key.key_id.lower()
+ facts = signing_key_facts(key)
+ primary_signatures = [*key.details.direct_signatures, *key.details.revocation_signatures]
+ components = [
+ _key_component("primary", facts[0], _effective_self_signature(key), primary_signatures, fingerprint, key_id)
+ ]
+ components.extend(
+ _identity_component("user-id", user.id, user.signatures, fingerprint, key_id) for user in key.details.users
+ )
+ for attribute in key.details.user_attributes:
+ label = _attribute_label(attribute.attr)
+ components.append(_identity_component("user-attribute", label, attribute.signatures, fingerprint, key_id))
+ for subkey_facts, subkey in zip(facts[1:], key.public_subkeys, strict=True):
+ binding = _latest_binding_signature(subkey)
+ components.append(_key_component("subkey", subkey_facts, binding, subkey.signatures, fingerprint, key_id))
+ return components
+
+
def certificate_for_fingerprint(armored: str, fingerprint: str) -> openpgp.composed.SignedPublicKey | None:
keys, _ = openpgp.composed.SignedPublicKey.from_armor_many(armored)
matches = [key for key in keys if key.fingerprint.lower() == fingerprint.lower()]
@@ -216,10 +249,15 @@ def signing_key_status(
return SigningKeyStatus(identified=False, fingerprint=None, expires=None, can_sign=False, revoked=False)
-def _binding_revoked(user: openpgp.types.SignedUser, fingerprint: str, key_id: str) -> bool:
+def _attribute_label(attribute: openpgp.packet.UserAttribute) -> str:
+ kind = " ".join(part for part in (attribute.kind, attribute.image_format) if part)
+ return f"{kind} ({len(attribute.data)} bytes)"
+
+
+def _binding_revoked(signatures: list[openpgp.packet.Signature], fingerprint: str, key_id: str) -> bool:
candidates = [
signature
- for signature in user.signatures
+ for signature in signatures
if _signature_is_self(signature, fingerprint, key_id)
and (
(signature.typ() in _CERTIFICATION_SIGNATURE_TYPES)
@@ -233,7 +271,9 @@ def _binding_revoked(user: openpgp.types.SignedUser, fingerprint: str, key_id: s
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_users = [user for user in key.details.users if not _binding_revoked(user, self_fingerprint, self_key_id)]
+ active_users = [
+ user for user in key.details.users if not _binding_revoked(user.signatures, 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] = []
@@ -291,6 +331,17 @@ def _effective_self_signature(key: openpgp.composed.SignedPublicKey) -> openpgp.
return _latest_signature(direct_sigs)
+def _identity_component(
+ kind: ComponentKind,
+ label: str,
+ signatures: list[openpgp.packet.Signature],
+ fingerprint: str,
+ key_id: str,
+) -> Component:
+ own, other = _signature_counts(signatures, fingerprint, key_id)
+ return Component(kind, label, None, _binding_revoked(signatures, fingerprint, key_id), own, other, None)
+
+
def _issuer_is_primary(
key: openpgp.composed.SignedPublicKey,
issuer_fingerprints: set[str],
@@ -312,6 +363,31 @@ def _issuing_subkey(
return None
+def _key_component(
+ kind: ComponentKind,
+ facts: SigningKeyFacts,
+ signature: openpgp.packet.Signature | None,
+ signatures: list[openpgp.packet.Signature],
+ fingerprint: str,
+ key_id: str,
+) -> Component:
+ own, other = _signature_counts(signatures, fingerprint, key_id)
+ return Component(kind, facts.fingerprint, _key_flags(signature), facts.revoked, own, other, facts)
+
+
+def _key_flags(signature: openpgp.packet.Signature | None) -> str | None:
+ if signature is None:
+ return None
+ flags = signature.key_flags()
+ letters = (
+ ("C", flags.certify),
+ ("S", flags.sign),
+ ("E", flags.encrypt_communications or flags.encrypt_storage),
+ ("A", flags.authenticate),
+ )
+ return "".join(letter for letter, declared in letters if declared)
+
+
def _latest_binding_signature(subkey: openpgp.composed.SignedPublicSubKey) -> openpgp.packet.Signature | None:
return _latest_signature(
[signature for signature in subkey.signatures if signature.typ() == _SUBKEY_BINDING_SIGNATURE_TYPE]
@@ -349,6 +425,11 @@ def _revocation_verifies(signature: openpgp.packet.Signature, primary: openpgp.p
return True
+def _signature_counts(signatures: list[openpgp.packet.Signature], fingerprint: str, key_id: str) -> tuple[int, int]:
+ own = sum(1 for signature in signatures if _signature_is_self(signature, fingerprint, key_id))
+ return own, len(signatures) - own
+
+
def _signature_expired(signature: openpgp.packet.Signature, now: float) -> bool:
expiration = signature.signature_expiration_time()
if not expiration:
diff --git a/atr/templates/includes/topnav.html b/atr/templates/includes/topnav.html
index 450758a8..2de538f8 100644
--- a/atr/templates/includes/topnav.html
+++ b/atr/templates/includes/topnav.html
@@ -300,6 +300,11 @@
href="{{ as_url(admin.keys_check_get) }}"
{% if request.endpoint == 'atr_admin_keys_check_get' %}class="active"{% endif %}><i class="bi bi-key"></i> Keys check</a>
</li>
+ <li>
+ <a class="dropdown-item"
+ href="{{ as_url(admin.keys_survey_get) }}"
+ {% if request.endpoint == 'atr_admin_keys_survey_get' %}class="active"{% endif %}><i class="bi bi-key"></i> Keys survey</a>
+ </li>
<li>
<a class="dropdown-item"
href="{{ as_url(admin.keys_update_get) }}"
diff --git a/tests/unit/test_admin_keys_survey.py b/tests/unit/test_admin_keys_survey.py
new file mode 100644
index 00000000..d188bc2a
--- /dev/null
+++ b/tests/unit/test_admin_keys_survey.py
@@ -0,0 +1,85 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import datetime
+
+import atr.admin as admin
+import atr.models.sql as sql
+import tests.unit.pgp_fixtures as pgp_fixtures
+
+_BROKEN_BLOCK = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n\nnotbase64\n-----END PGP PUBLIC KEY BLOCK-----\n"
+_NOW = datetime.datetime.now(datetime.UTC)
+
+
+def _certificate(fingerprint: str, armored: str, deleted: datetime.datetime | None = None) -> sql.SigningCertificate:
+ return sql.SigningCertificate(
+ fingerprint=fingerprint,
+ primary_declared_uid="Alice <al...@example.org>",
+ secondary_declared_uids=[],
+ apache_uid="alice",
+ ascii_armored_key=armored,
+ deleted=deleted,
+ )
+
+
+def test_distribution_summarises_counts() -> None:
+ assert admin._survey_distribution([]) == admin.Distribution(0, 0.0, 0, {})
+ assert admin._survey_distribution([2, 1, 2, 5]) == admin.Distribution(4, 2.5, 5, {1: 1, 2: 2, 5: 1})
+
+
+def test_report_reads_components_shape_and_state() -> None:
+ block = pgp_fixtures.two_certificate_block(
+ pgp_fixtures.REVOKED_SUBKEY_PUBLIC_KEY_ASC, pgp_fixtures.REVOKED_PRIMARY_PUBLIC_KEY_ASC
+ )
+ live = admin._survey_report(_certificate(pgp_fixtures.REVOKED_SUBKEY_PRIMARY_FINGERPRINT, block), ["alpha"], [])
+ revoked = admin._survey_report(_certificate(pgp_fixtures.REVOKED_PRIMARY_FINGERPRINT, block), [], [])
+ broken = admin._survey_report(_certificate("ab" * 20, _BROKEN_BLOCK), [], [])
+ deleted = admin._survey_report(
+ _certificate(pgp_fixtures.REVOKED_UID_FINGERPRINT, pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC, deleted=_NOW),
+ [],
+ [],
+ )
+
+ assert (live.shape, live.certificates_in_block, live.version) == ("multi-own-first", 2, 4)
+ assert [component.kind for component in live.components] == ["primary", "user-id", "subkey"]
+ assert revoked.shape == "multi-own-not-first"
+ assert (broken.shape, broken.components) == ("unparseable", [])
+ assert [admin._survey_state(r) for r in (live, revoked, broken, deleted)] == [
+ "live",
+ "revoked",
+ "unreadable",
+ "deleted",
+ ]
+
+
+def test_statistics_and_tallies_cover_every_report() -> None:
+ reports = [
+ admin._survey_report(
+ _certificate(pgp_fixtures.REVOKED_UID_FINGERPRINT, pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC), ["alpha"], []
+ ),
+ admin._survey_report(_certificate("ab" * 20, _BROKEN_BLOCK), ["alpha", "beta"], []),
+ ]
+ survey = admin.Survey([sql.Committee(key="alpha"), sql.Committee(key="beta"), sql.Committee(key="gamma")], reports)
+
+ statistics = admin._survey_statistics(survey)
+ tallies = admin._survey_tallies(survey)
+
+ assert statistics["user IDs per certificate"] == admin.Distribution(1, 2.0, 2, {2: 1})
+ assert statistics["committees per certificate"].histogram == {1: 1, 2: 1}
+ assert statistics["certificates per committee"].histogram == {0: 1, 1: 1, 2: 1}
+ assert tallies["block shapes"] == {"single": 1, "unparseable": 1}
+ assert tallies["subkey flags"] == {"S": 1}
diff --git a/tests/unit/test_pgp.py b/tests/unit/test_pgp.py
index 537dc937..a97531f4 100644
--- a/tests/unit/test_pgp.py
+++ b/tests/unit/test_pgp.py
@@ -59,6 +59,23 @@ def test_certificate_block_shape_names_every_shape() -> None:
assert pgp.certificate_block_shape([], own) == "empty"
+def test_certificate_components_lists_each_part_with_its_signatures() -> None:
+ key, _ = openpgp.composed.SignedPublicKey.from_armor(pgp_fixtures.REVOKED_UID_PUBLIC_KEY_ASC)
+
+ components = pgp.certificate_components(key)
+
+ assert [component.kind for component in components] == ["primary", "user-id", "user-id", "subkey"]
+ assert components[0].label == pgp_fixtures.REVOKED_UID_FINGERPRINT
+ assert (components[0].flags, components[3].flags) == ("C", "S")
+ assert [component.revoked for component in components] == [False, False, True, False]
+ assert [(component.self_signatures, component.other_signatures) for component in components[1:3]] == [
+ (1, 0),
+ (2, 0),
+ ]
+ assert (components[1].facts is None) and (components[3].facts is not None)
+ assert components[3].facts.fingerprint == pgp_fixtures.REVOKED_UID_SIGNING_FINGERPRINT
+
+
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
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org