From 5bfcc3c06ebe6d6694221c25f07a4901572f97f5 Mon Sep 17 00:00:00 2001 From: Dennis Fink Date: Thu, 20 Aug 2026 23:01:59 +0200 Subject: docs(core): add docstrings and type annotations Document forms, validators, models, and views with PEP 287-style docstrings and complete missing type annotations. Add developer-readable representations for Webmention models and mark the user password property as write-only in its return type. --- webmentions_ssg/forms/__init__.py | 15 ++++++ webmentions_ssg/forms/validators.py | 90 ++++++++++++++++++++++++--------- webmentions_ssg/models.py | 99 ++++++++++++++++++++++++++++++++++++- webmentions_ssg/views.py | 72 ++++++++++++++++++++++++++- 4 files changed, 250 insertions(+), 26 deletions(-) (limited to 'webmentions_ssg') diff --git a/webmentions_ssg/forms/__init__.py b/webmentions_ssg/forms/__init__.py index 7ab82e3..5d2f827 100644 --- a/webmentions_ssg/forms/__init__.py +++ b/webmentions_ssg/forms/__init__.py @@ -12,12 +12,23 @@ from .validators import AllowedHostname, NotEqualTo, PublicURL class LoginForm(FlaskForm): + """ + Form for authenticating a user. + """ + username = StringField("Username", validators=[InputRequired()]) password = PasswordField("Password", validators=[InputRequired()]) submit = SubmitField("Sign In") class EndpointForm(FlaskForm): + """ + Form for receiving a Webmention. + + The source must be a public HTTP or HTTPS URL distinct from the target. + The target must be an HTTP or HTTPS URL using an allowed hostname. + """ + source = StringField( "source", validators=[ @@ -48,4 +59,8 @@ class EndpointForm(FlaskForm): class AdminActionForm(FlaskForm): + """ + Form for confirming an administrative action. + """ + submit = SubmitField("Confirm") diff --git a/webmentions_ssg/forms/validators.py b/webmentions_ssg/forms/validators.py index fd54e0f..761dc46 100644 --- a/webmentions_ssg/forms/validators.py +++ b/webmentions_ssg/forms/validators.py @@ -5,28 +5,39 @@ from urllib.parse import urlsplit from flask import current_app -from wtforms import ValidationError +from wtforms import Field, ValidationError +from wtforms.form import BaseForm from ..url_security import AddressResolutionError, is_public_url class NotEqualTo: """ - Compares the values of two fields. - - :param fieldname: - The name of the other field to compare to. - :param message: - Error message to raise in case of a validation error. Can be - interpolated with `%(other_label)s` and `%(other_name)s` to provide a - more helpful error. + Validate that a field does not equal another field. + + :param fieldname: Name of the field to compare against. + :param message: Optional validation error message. """ - def __init__(self, fieldname, message=None): + def __init__(self, fieldname: str, message: str | None = None) -> None: + """ + Initialize the validator. + + :param fieldname: Name of the field to compare against. + :param message: Optional validation error message. + """ self.fieldname = fieldname self.message = message - def __call__(self, form, field): + def __call__(self, form: BaseForm, field: Field) -> None: + """ + Validate that two fields do not contain equal values. + + :param form: Form containing the fields to compare. + :param field: Field being validated. + :raises ValidationError: If the comparison field does not exist or both + fields contain equal values. + """ try: other = form[self.fieldname] except KeyError as exc: @@ -43,36 +54,67 @@ class NotEqualTo: or self.fieldname, "other_name": self.fieldname, } - message = self.message - if message is None: - message = field.gettext("Field must not be equal to %(other_name)s.") - + message = self.message or field.gettext( + "Field must not be equal to %(other_name)s." + ) raise ValidationError(message % d) class AllowedHostname: - def __init__(self, message=None): + """ + Validate that a URL uses an allowed hostname. + + :param message: Optional validation error message. + """ + + def __init__(self, message: str | None = None) -> None: + """ + Initialize the validator. + + :param message: Optional validation error message. + """ self.message = message - def __call__(self, form, field): + def __call__(self, form: BaseForm, field: Field) -> None: + """ + Validate that a URL uses an allowed hostname. + + :param form: Form containing the field. + :param field: Field containing the URL to validate. + :raises ValidationError: If the URL hostname is not configured as allowed. + """ if ( urlsplit(field.data).hostname in current_app.config["WEBMENTIONS_SSG_ALLOWED_HOSTNAMES"] ): return + raise ValidationError(self.message or field.gettext("Invalid input.")) - message = self.message - if self.message is None: - message = field.gettext("Invalid input.") - raise ValidationError(message) +class PublicURL: + """ + Validate that a URL resolves exclusively to public IP addresses. + :param message: Optional validation error message. + """ -class PublicURL: - def __init__(self, message=None): + def __init__(self, message: str | None = None) -> None: + """ + Initialize the validator. + + :param message: Optional validation error message. + """ self.message = message - def __call__(self, form, field): + def __call__(self, form: BaseForm, field: Field) -> None: + """ + Validate that a URL resolves exclusively to public IP addresses. + + :param form: Form containing the field. + :param field: Field containing the URL to validate. + :raises ValidationError: If the URL cannot be resolved, has no hostname, or + resolves to a non-public address. + """ message = self.message if message is None: diff --git a/webmentions_ssg/models.py b/webmentions_ssg/models.py index e7ff0be..5d22395 100644 --- a/webmentions_ssg/models.py +++ b/webmentions_ssg/models.py @@ -7,6 +7,7 @@ from __future__ import annotations import uuid from datetime import UTC, datetime from enum import StrEnum +from typing import Never from flask_login import UserMixin from sqlalchemy import ( @@ -27,12 +28,20 @@ from . import Base class SentWebmentionStatus(StrEnum): + """ + Represent the outcome of a Webmention send attempt. + """ + SENT = "sent" UNSUPPORTED = "unsupported" FAILED = "failed" class User(UserMixin, Base): + """ + Represent an authenticated application user. + """ + __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) @@ -40,21 +49,46 @@ class User(UserMixin, Base): password_hash: Mapped[str] = mapped_column(Text(), nullable=False) def __repr__(self) -> str: + """ + Return a developer-readable representation of the user. + + :return: Representation containing the username. + """ return f"" @property - def password(self) -> None: + def password(self) -> Never: + """ + Prevent access to the user's plain-text password. + + :raises AttributeError: Always, because the password is write-only. + """ raise AttributeError("Password is write-only") @password.setter def password(self, password: str) -> None: + """ + Hash and store a new password. + + :param password: Plain-text password to hash. + """ self.password_hash = generate_password_hash(password) def check_password(self, password: str) -> bool: + """ + Check a plain-text password against the stored password hash. + + :param password: Plain-text password to check. + :return: Whether the password matches the stored hash. + """ return check_password_hash(self.password_hash, password) class ReceivedWebmention(Base): + """ + Represent a Webmention received for verification. + """ + __tablename__ = "received_webmentions" uuid: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True) @@ -78,16 +112,38 @@ class ReceivedWebmention(Base): UniqueConstraint("source", "target", name="uq_webmention_source_target"), ) + def __repr__(self) -> str: + """ + Return a developer-readable representation of the received Webmention. + + :return: Representation containing the identifier, source, and target. + """ + return f" {self.target!r}>" + @property def verified(self) -> bool: + """ + Return whether the Webmention has been successfully verified. + + :return: Whether the Webmention status is ``verified``. + """ return self.status == "verified" @property def created_at(self) -> datetime: + """ + Return the creation time encoded in the Webmention UUID. + + :return: Creation time as a timezone-aware datetime. + """ return uuid7_to_datetime(self.uuid) class Source(Base): + """ + Represent a generated source document containing outgoing Webmentions. + """ + __tablename__ = "sources" uuid: Mapped[uuid.UUID] = mapped_column( @@ -114,12 +170,29 @@ class Source(Base): back_populates="source", cascade="all, delete-orphan", lazy="selectin" ) + def __repr__(self) -> str: + """ + Return a developer-readable representation of the source. + + :return: Representation containing the identifier and public URL. + """ + return f"" + @property def created_at(self) -> datetime: + """ + Return the creation time encoded in the source UUID. + + :return: Creation time as a timezone-aware datetime. + """ return uuid7_to_datetime(self.uuid) class SentWebmention(Base): + """ + Represent an outgoing Webmention associated with a source. + """ + __tablename__ = "sent_webmentions" uuid: Mapped[uuid.UUID] = mapped_column( @@ -181,8 +254,21 @@ class SentWebmention(Base): ), ) + def __repr__(self) -> str: + """ + Return a developer-readable representation of the sent Webmention. + + :return: Representation containing the identifier and target URL. + """ + return f"" + @property def pending(self) -> bool: + """ + Return whether the Webmention has an unprocessed source revision. + + :return: Whether the desired revision still requires processing. + """ return ( self.processed_revision is None or self.processed_revision < self.desired_revision @@ -190,8 +276,19 @@ class SentWebmention(Base): @property def created_at(self) -> datetime: + """ + Return the creation time encoded in the Webmention UUID. + + :return: Creation time as a timezone-aware datetime. + """ return uuid7_to_datetime(self.uuid) def uuid7_to_datetime(identifier: uuid.UUID) -> datetime: + """ + Convert a UUID version 7 timestamp to a datetime. + + :param identifier: UUID whose embedded timestamp should be converted. + :return: Timestamp as a timezone-aware UTC datetime. + """ return datetime.fromtimestamp(identifier.time / 1000, tz=UTC) diff --git a/webmentions_ssg/views.py b/webmentions_ssg/views.py index eb2b372..3430cea 100644 --- a/webmentions_ssg/views.py +++ b/webmentions_ssg/views.py @@ -33,11 +33,25 @@ root_page = Blueprint("root", __name__) @root_page.route("/") def index() -> ResponseReturnValue: + """ + Redirect to the received Webmentions view. + + :return: Redirect response to the received Webmentions view. + """ return redirect(url_for("root.received")) @root_page.route("/login", methods=["GET", "POST"]) def login() -> ResponseReturnValue: + """ + Authenticate a user and start a login session. + + Authenticated users are redirected to the application index. After a + successful login, the user is redirected to the requested local URL when + provided. + + :return: Rendered login page or redirect response. + """ if current_user.is_authenticated: return redirect(url_for("root.index")) @@ -64,6 +78,11 @@ def login() -> ResponseReturnValue: @root_page.route("/logout") def logout() -> ResponseReturnValue: + """ + End the current user's login session. + + :return: Redirect response to the application index. + """ logout_user() return redirect(url_for("root.index")) @@ -71,6 +90,11 @@ def logout() -> ResponseReturnValue: @root_page.route("/received") @login_required def received() -> ResponseReturnValue: + """ + Display received Webmentions. + + :return: Rendered page containing the paginated received Webmentions. + """ webmentions = db.paginate( sa.select(ReceivedWebmention).order_by(ReceivedWebmention.uuid.desc()), per_page=25, @@ -87,6 +111,12 @@ def received() -> ResponseReturnValue: @root_page.post("/received//delete") @login_required def delete_received_webmention(identifier: uuid.UUID) -> ResponseReturnValue: + """ + Delete a received Webmention. + + :param identifier: Identifier of the Webmention to delete. + :return: Redirect response to the received Webmentions view. + """ form = forms.AdminActionForm() if not form.validate_on_submit(): @@ -110,6 +140,14 @@ def delete_received_webmention(identifier: uuid.UUID) -> ResponseReturnValue: @root_page.post("/received//reverify") @login_required def reverify_received_webmention(identifier: uuid.UUID) -> ResponseReturnValue: + """ + Queue a received Webmention for reverification. + + The Webmention status is reset before a new verification task is queued. + + :param identifier: Identifier of the Webmention to reverify. + :return: Redirect response to the received Webmentions view. + """ form = forms.AdminActionForm() if not form.validate_on_submit(): @@ -137,6 +175,11 @@ def reverify_received_webmention(identifier: uuid.UUID) -> ResponseReturnValue: @root_page.route("/sent") @login_required def sent() -> ResponseReturnValue: + """ + Display sources with sent Webmentions. + + :return: Rendered page containing the paginated source list. + """ sources = db.paginate( sa.select(Source) .options(selectinload(Source.sent_webmentions)) @@ -153,7 +196,12 @@ def sent() -> ResponseReturnValue: @root_page.post("/sent/rescan") @login_required -def rescan_sent_sources(): +def rescan_sent_sources() -> ResponseReturnValue: + """ + Queue a manual scan of sent Webmention sources. + + :return: Redirect response to the sent Webmentions view. + """ form = forms.AdminActionForm() if form.validate_on_submit(): manual_scan_sources() @@ -164,6 +212,12 @@ def rescan_sent_sources(): @root_page.route("/sent/") @login_required def sent_source(identifier: uuid.UUID) -> ResponseReturnValue: + """ + Display the sent Webmentions associated with a source. + + :param identifier: Identifier of the source to display. + :return: Rendered source details page. + """ source = db.session.scalar( sa.select(Source) .options(selectinload(Source.sent_webmentions)) @@ -186,6 +240,16 @@ def sent_source(identifier: uuid.UUID) -> ResponseReturnValue: @root_page.post("/endpoint") @CSRF.exempt def endpoint() -> ResponseReturnValue: + """ + Receive and queue a Webmention for verification. + + Existing Webmentions with the same source and target are reset for + reverification. Concurrent insertion of the same Webmention is handled by + retrieving and updating the row created by the competing request. + + :return: HTTP 201 response with the Webmention status URL in the + ``Location`` header, or validation errors with HTTP 400. + """ form = forms.EndpointForm(meta={"csrf": False}) if not form.validate_on_submit(): @@ -245,6 +309,12 @@ def endpoint() -> ResponseReturnValue: @root_page.route("/status/") def status(identifier: uuid.UUID) -> ResponseReturnValue: + """ + Display the verification status of a received Webmention. + + :param identifier: Identifier of the Webmention to display. + :return: Rendered Webmention status page. + """ webmention = db.session.get(ReceivedWebmention, identifier) if webmention is None: -- cgit v1.3.1