This is an automated email from the ASF dual-hosted git repository.
sbp pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
The following commit(s) were added to refs/heads/main by this push:
new 4c3a067 Make the form to add an OpenPGP key more type safe
4c3a067 is described below
commit 4c3a06735c2cd629d3b4c5149ab423f353152a6a
Author: Sean B. Palmer <[email protected]>
AuthorDate: Tue Nov 11 18:33:14 2025 +0000
Make the form to add an OpenPGP key more type safe
---
atr/form.py | 1 +
atr/get/keys.py | 31 ++++++++++++++++++-
atr/post/keys.py | 42 ++++++++++++++++++++++++--
atr/shared/keys.py | 89 +++++++++---------------------------------------------
4 files changed, 85 insertions(+), 78 deletions(-)
diff --git a/atr/form.py b/atr/form.py
index d8fa7d2..0a2e5da 100644
--- a/atr/form.py
+++ b/atr/form.py
@@ -382,6 +382,7 @@ def to_int(v: Any) -> int:
def to_str_list(v: Any) -> list[str]:
+ # TODO: Might need to handle the empty case
if isinstance(v, list):
return [str(item) for item in v]
if isinstance(v, str):
diff --git a/atr/get/keys.py b/atr/get/keys.py
index 1dc725a..a9eb88d 100644
--- a/atr/get/keys.py
+++ b/atr/get/keys.py
@@ -21,6 +21,9 @@ import quart
import atr.blueprints.get as get
import atr.db as db
+import atr.form as form
+import atr.htm as htm
+import atr.post as post
import atr.shared as shared
import atr.storage as storage
import atr.template as template
@@ -31,7 +34,33 @@ import atr.web as web
@get.committer("/keys/add")
async def add(session: web.Committer) -> str:
"""Add a new public signing key to the user's account."""
- return await shared.keys.add(session)
+ async with storage.write() as write:
+ participant_of_committees = await write.participant_of_committees()
+
+ committee_choices = [(c.name, c.display_name or c.name) for c in
participant_of_committees]
+
+ page = htm.Block()
+ page.p[htm.a(href=util.as_url(keys), class_="atr-back-link")["← Back to
Manage keys"],]
+ page.div(class_="my-4")[
+ htm.h1(class_="mb-4")["Add your OpenPGP key"],
+ htm.p["Add your public key to use for signing release artifacts."],
+ ]
+ form.render_block(
+ page,
+ model_cls=shared.keys.AddOpenPGPKeyForm,
+ action=util.as_url(post.keys.add),
+ submit_label="Add OpenPGP key",
+ cancel_url=util.as_url(keys),
+ defaults={
+ "selected_committees": committee_choices,
+ },
+ )
+
+ return await template.blank(
+ "Add your OpenPGP key",
+ content=page.collect(),
+ description="Add your public signing key to your ATR account.",
+ )
@get.committer("/keys/details/<fingerprint>")
diff --git a/atr/post/keys.py b/atr/post/keys.py
index 0fcad5e..36054cb 100644
--- a/atr/post/keys.py
+++ b/atr/post/keys.py
@@ -20,6 +20,8 @@ import quart
import atr.blueprints.post as post
import atr.get as get
+import atr.htm as htm
+import atr.log as log
import atr.models.sql as sql
import atr.shared as shared
import atr.storage as storage
@@ -30,9 +32,45 @@ import atr.web as web
@post.committer("/keys/add")
-async def add(session: web.Committer) -> str:
[email protected](shared.keys.AddOpenPGPKeyForm)
+async def add(session: web.Committer, add_openpgp_key_form:
shared.keys.AddOpenPGPKeyForm) -> web.WerkzeugResponse:
"""Add a new public signing key to the user's account."""
- return await shared.keys.add(session)
+ try:
+ key_text = add_openpgp_key_form.public_key
+ selected_committee_names = add_openpgp_key_form.selected_committees
+
+ async with storage.write() as write:
+ wafc = write.as_foundation_committer()
+ ocr: outcome.Outcome[types.Key] = await
wafc.keys.ensure_stored_one(key_text)
+ key = ocr.result_or_raise()
+
+ for selected_committee_name in selected_committee_names:
+ wacp = write.as_committee_participant(selected_committee_name)
+ oc: outcome.Outcome[types.LinkedCommittee] = await
wacp.keys.associate_fingerprint(
+ key.key_model.fingerprint
+ )
+ oc.result_or_raise()
+
+ fingerprint_upper = key.key_model.fingerprint.upper()
+ if key.status == types.KeyStatus.PARSED:
+ details_url = util.as_url(get.keys.details,
fingerprint=key.key_model.fingerprint)
+ p = htm.p[
+ f"OpenPGP key {fingerprint_upper} was already in the
database. ",
+ htm.a(href=details_url)["View key details"],
+ ".",
+ ]
+ await quart.flash(str(p), "warning")
+ else:
+ await quart.flash(f"OpenPGP key {fingerprint_upper} added
successfully.", "success")
+
+ except web.FlashError as e:
+ log.warning("FlashError adding OpenPGP key: %s", e)
+ await quart.flash(str(e), "error")
+ except Exception as e:
+ log.exception("Error adding OpenPGP key:")
+ await quart.flash(f"An unexpected error occurred: {e!s}", "error")
+
+ return await session.redirect(get.keys.keys)
@post.committer("/keys/delete")
diff --git a/atr/shared/keys.py b/atr/shared/keys.py
index b576eca..8da536f 100644
--- a/atr/shared/keys.py
+++ b/atr/shared/keys.py
@@ -23,15 +23,15 @@ from collections.abc import Awaitable, Callable, Sequence
import aiohttp
import asfquart.base as base
+import pydantic
import quart
import werkzeug.datastructures as datastructures
import wtforms
import atr.db as db
+import atr.form as form
import atr.forms as forms
import atr.get as get
-import atr.htm as htm
-import atr.log as log
import atr.models.sql as sql
import atr.shared as shared
import atr.storage as storage
@@ -43,18 +43,22 @@ import atr.util as util
import atr.web as web
-class AddOpenPGPKeyForm(forms.Typed):
- public_key = forms.textarea(
+class AddOpenPGPKeyForm(form.Form):
+ public_key: str = form.label(
"Public OpenPGP key",
- placeholder="Paste your ASCII-armored public OpenPGP key here...",
- description="Your public key should be in ASCII-armored format,
starting with"
- ' "-----BEGIN PGP PUBLIC KEY BLOCK-----"',
+ 'Your public key should be in ASCII-armored format, starting with
"-----BEGIN PGP PUBLIC KEY BLOCK-----"',
+ widget=form.Widget.TEXTAREA,
)
- selected_committees = forms.checkboxes(
+ selected_committees: form.StrList = form.label(
"Associate key with committees",
- description="Select the committees with which to associate your key.",
+ "Select the committees with which to associate your key.",
)
- submit = forms.submit("Add OpenPGP key")
+
+ @pydantic.model_validator(mode="after")
+ def validate_at_least_one_committee(self) -> "AddOpenPGPKeyForm":
+ if not self.selected_committees:
+ raise ValueError("You must select at least one committee to
associate with this key")
+ return self
class AddSSHKeyForm(forms.Typed):
@@ -134,71 +138,6 @@ class UploadKeyFormBase(forms.Typed):
return True
-async def add(session: web.Committer) -> str:
- """Add a new public signing key to the user's account."""
- key_info = None
-
- async with storage.write() as write:
- participant_of_committees = await write.participant_of_committees()
-
- committee_choices: forms.Choices = [(c.name, c.display_name or c.name) for
c in participant_of_committees]
-
- form = await AddOpenPGPKeyForm.create_form(
- data=(await quart.request.form) if (quart.request.method == "POST")
else None
- )
- forms.choices(form.selected_committees, committee_choices)
-
- if await form.validate_on_submit():
- try:
- key_text: str = util.unwrap(form.public_key.data)
- selected_committee_names: list[str] =
util.unwrap(form.selected_committees.data)
-
- async with storage.write() as write:
- wafc = write.as_foundation_committer()
- ocr: outcome.Outcome[types.Key] = await
wafc.keys.ensure_stored_one(key_text)
- key = ocr.result_or_raise()
-
- for selected_committee_name in selected_committee_names:
- # TODO: Should this be committee member or committee
participant?
- # Also, should we emit warnings and continue here?
- wacp =
write.as_committee_participant(selected_committee_name)
- oc: outcome.Outcome[types.LinkedCommittee] = await
wacp.keys.associate_fingerprint(
- key.key_model.fingerprint
- )
- oc.result_or_raise()
-
- fingerprint_upper = key.key_model.fingerprint.upper()
- if key.status == types.KeyStatus.PARSED:
- details_url = util.as_url(get.keys.details,
fingerprint=key.key_model.fingerprint)
- p = htm.p[
- f"OpenPGP key {fingerprint_upper} was already in the
database. ",
- htm.a(href=details_url)["View key details"],
- ".",
- ]
- await quart.flash(str(p), "warning")
- else:
- await quart.flash(f"OpenPGP key {fingerprint_upper} added
successfully.", "success")
- # Clear form data on success by creating a new empty form instance
- form = await AddOpenPGPKeyForm.create_form()
- forms.choices(form.selected_committees, committee_choices)
-
- except web.FlashError as e:
- log.warning("FlashError adding OpenPGP key: %s", e)
- await quart.flash(str(e), "error")
- except Exception as e:
- log.exception("Error adding OpenPGP key:")
- await quart.flash(f"An unexpected error occurred: {e!s}", "error")
-
- return await template.render(
- "keys-add.html",
- asf_id=session.uid,
- user_committees=participant_of_committees,
- form=form,
- key_info=key_info,
- algorithms=shared.algorithms,
- )
-
-
async def details(session: web.Committer, fingerprint: str) -> str |
web.WerkzeugResponse:
"""Display details for a specific OpenPGP key."""
fingerprint = fingerprint.lower()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
