diff --git a/libervia/backend/core/main.py b/libervia/backend/core/main.py index 459f4c1e..874a6bde 100644 --- a/libervia/backend/core/main.py +++ b/libervia/backend/core/main.py @@ -25,7 +25,6 @@ from pathlib import Path import shutil from typing import Optional, List, Tuple, Dict, cast -from wokkel.data_form import Option from libervia import backend from libervia.backend.core.i18n import _, D_, language_switch from libervia.backend.core import patches @@ -82,7 +81,6 @@ class LiberviaBackend(service.Service): # map for short name to whole namespace, # extended by plugins with register_namespace self.ns_map = { - "x-data": xmpp.NS_X_DATA, "disco#info": xmpp.NS_DISCO_INFO, } diff --git a/libervia/backend/core/xmpp.py b/libervia/backend/core/xmpp.py index 89d3e480..0b4e067f 100644 --- a/libervia/backend/core/xmpp.py +++ b/libervia/backend/core/xmpp.py @@ -58,9 +58,7 @@ from libervia.backend.tools.common import data_format log = getLogger(__name__) -NS_X_DATA = "jabber:x:data" NS_DISCO_INFO = "http://jabber.org/protocol/disco#info" -NS_XML_ELEMENT = "urn:xmpp:xml-element" NS_ROSTER_VER = "urn:xmpp:features:rosterver" # we use 2 "@" which is illegal in a jid, to be sure we are not mixing keys # with roster jids @@ -2067,8 +2065,6 @@ class SatDiscoProtocol(disco.DiscoClientProtocol): # those features are implemented in Wokkel (or sat_tmp.wokkel) # and thus are always available return [ - disco.DiscoFeature(NS_X_DATA), - disco.DiscoFeature(NS_XML_ELEMENT), disco.DiscoFeature(NS_DISCO_INFO), ] diff --git a/libervia/backend/plugins/plugin_xep_0004.py b/libervia/backend/plugins/plugin_xep_0004.py new file mode 100644 index 00000000..93d51174 --- /dev/null +++ b/libervia/backend/plugins/plugin_xep_0004.py @@ -0,0 +1,350 @@ +#!/usr/bin/env python3 + +# Libervia plugin for Data Form (XEP-0004) +# Copyright (C) 2009-2025 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +from typing import Self, TYPE_CHECKING +from typing import Annotated, Literal + +from pydantic import BaseModel, Field +from twisted.words.protocols.jabber import jid +from twisted.words.protocols.jabber.xmlstream import XMPPHandler +from twisted.words.xish import domish +from wokkel import data_form, disco +from zope.interface import implementer + +from libervia.backend.core import exceptions +from libervia.backend.core.constants import Const as C +from libervia.backend.core.i18n import _ +from libervia.backend.core.log import getLogger +from libervia.backend.models.types import DomishElementType, JIDType + +if TYPE_CHECKING: + from libervia.backend.core.main import LiberviaBackend + +log = getLogger(__name__) + +PLUGIN_INFO = { + C.PI_NAME: "Data Forms", + C.PI_IMPORT_NAME: "XEP-0004", + C.PI_MODES: C.PLUG_MODE_BOTH, + C.PI_TYPE: "XEP", + C.PI_PROTOCOLS: ["XEP-0004", "XEP-0315"], + C.PI_MAIN: "XEP_0004", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: _( + "Implementation of data forms.\n" + "This plugins extends Wokkel implementation with Pydantic models.\n" + # XXX: XEP-0315 is currently implemented in "sat_tmp.wokkel.data_form" + "XEP-0315 (Data Forms XML Element) is also supported here." + ), +} + +NS_XML_ELEMENT = "urn:xmpp:xml-element" + + +# Map from value to optional label. +class Option(BaseModel): + value: str + label: str | None = None + + +class BaseField(BaseModel): + var: str + label: str | None = None + description: str | None = Field(alias="desc", default=None) + required: bool = False + + +class Text(BaseField): + type: Literal["text-single", "text-private", "text-multi", "fixed"] = "text-single" + value: str | None = None + + +class Boolean(BaseField): + type: Literal["boolean"] = "boolean" + value: bool | None = None + + +class Hidden(BaseField): + type: Literal["hidden"] = "hidden" + value: str | None = None + + +class JIDSingle(BaseField): + type: Literal["jid-single"] = "jid-single" + value: JIDType | None = None + + +class JIDMulti(BaseField): + type: Literal["jid-multi"] = "jid-multi" + values: list[JIDType] = Field(default_factory=list) + + +class ListSingle(BaseField): + type: Literal["list-single"] = "list-single" + value: str | None = None + options: list[Option] = Field(default_factory=list) + + +class ListMulti(BaseField): + type: Literal["list-multi"] = "list-multi" + values: list[str] = Field(default_factory=list) + options: list[Option] = Field(default_factory=list) + + +class XML(BaseField): + type: Literal["xml"] = "xml" + value: DomishElementType | None = None + + +DataField = Annotated[ + Text | Boolean | Hidden | JIDSingle | JIDMulti | ListSingle | ListMulti | XML, + Field(discriminator="type"), +] + + +class DataForm(BaseModel): + type: Literal["form", "submit", "result", "cancel"] = "form" + title: str | None = None + instructions: str | None = None + form_namespace: str | None = None + fields: list[DataField] = Field(default_factory=list) + + def get_values(self) -> dict[str, list[None | str | bool | jid.JID | domish.Element]]: + """Return a dict mapping from ``var`` to a list of values. + + This is typically used to retrieve submitted values. + """ + return { + field.var: (getattr(field, "values", None) or [field.value]) #  type: ignore + for field in self.fields + if field.var + } + + def __getitem__(self, field_var: str) -> DataField | None: + for field in self.fields: + if field.var == field_var: + return field + + @classmethod + def from_wokkel_form( + cls, form: data_form.Form, source_form: Self | None = None + ) -> Self: + """Convert Wokkel Form to DataForm. + + @param form: Wokkel Form to convert. + @param source_form: Source DataForm, used to get field types on submitted form. + This is the original form the sending entity is expected to have filled, its + field type are used to correctly set the one from submitted form (if they are + not specified already). + @return: Instance of DataForm. + + """ + fields = [] + + for field in form.fieldList: + if field.var == "FORM_TYPE" and field.fieldType == "hidden": + # We skip FORM_TYPE field which is handled separately. + continue + + if ( + field.fieldType is None + and field.var is not None + and form.formType == "submit" + ): + # If on a submitted form the field types are not specified, we try to + # retrieve them from source form, if any. + if source_form is not None: + for source_field in source_form.fields: + if source_field.var == field.var: + field.fieldType = source_field.type + break + + # If we have no source field, we default to ``list-multi`` because it has + # multiple text values, which can then be retrieved with + # ``get_fields_map``. + if field.fieldType is None: + field.fieldType = "list-multi" + + field_dict = { + "var": field.var or "", + "label": field.label, + "desc": field.desc, + "required": field.required, + } + + # Handle options for list fields. + if field.fieldType in ("list-single", "list-multi") and field.options: + field_dict["options"] = [ + Option(value=opt.value, label=opt.label) for opt in field.options + ] + + try: + match field.fieldType: + case "text-single" | "text-private" | "fixed": + fields.append( + Text(type=field.fieldType, value=field.value, **field_dict) + ) + case "text-multi": + fields.append( + Text( + type="text-multi", + value="\n".join(field.values), + **field_dict, + ) + ) + case "boolean": + fields.append(Boolean(value=field.value, **field_dict)) + case "hidden": + fields.append(Hidden(value=field.value, **field_dict)) + case "jid-single": + fields.append(JIDSingle(value=field.value, **field_dict)) + case "jid-multi": + fields.append(JIDMulti(values=field.values, **field_dict)) + case "list-single": + fields.append(ListSingle(value=field.value, **field_dict)) + case "list-multi": + fields.append(ListMulti(values=field.values, **field_dict)) + case None if field.ext_type == "xml": + fields.append(XML(value=field.value, **field_dict)) + case _: + # Fallback to text-single for unknown types. + fields.append( + Text( + value="\n".join(field.values) if field.values else "", + **field_dict, + ) + ) + except Exception: + # Fallback for any validation issues. + fields.append( + Text( + value="\n".join(field.values) if field.values else "", + **field_dict, + ) + ) + + return cls( + type=form.formType, + title=form.title, + instructions="\n".join(form.instructions) if form.instructions else None, + form_namespace=form.formNamespace, + fields=fields, + ) + + def to_wokkel_form(self) -> data_form.Form: + """Convert DataForm to Wokkel Form.""" + form = data_form.Form( + formType=self.type, + title=self.title, + instructions=self.instructions.splitlines() if self.instructions else None, + formNamespace=self.form_namespace, + ) + + for field in self.fields: + kwargs = field.model_dump(by_alias=True) + # We don't want to alias each indivial type with "fieldType", so we change the + # name of the key here. + + kwargs["fieldType"] = kwargs.pop("type") + if kwargs["fieldType"] == "text-multi": + if value := kwargs.pop("value"): + kwargs["values"] = value.splitlines() + if "options" in kwargs: + kwargs["options"] = [ + data_form.Option(value=opt["value"], label=opt["label"]) + for opt in kwargs["options"] + ] + + form.addField(data_form.Field(**kwargs)) + + return form + + @classmethod + def from_element( + cls, element: domish.Element, source_form: Self | None = None + ) -> Self: + """Parse a data form to create an instance of DataForm. + + If the element itself is not a data form, a data form will be looked after in its + descendant. + @param element: Element to parse, it must be a data form, or have a child which is + a data form. + @param source_form: Original form for submitted forms. See [from_wokkel_form]. + @return: Instance of DataForm. + @raise exceptions.NotFound: No data form found in ``element`` or one of its + descendants. + """ + if (element.uri, element.name) != ((data_form.NS_X_DATA, "x")): + try: + element = next(element.elements(data_form.NS_X_DATA, "x")) + except StopIteration: + raise exceptions.NotFound( + f"No data form found in {element} or its descendants." + ) + form = data_form.Form.fromElement(element) + return cls.from_wokkel_form(form, source_form=source_form) + + def to_element(self) -> domish.Element: + """Generate the element corresponding to this form.""" + return self.to_wokkel_form().toElement() + + +class XEP_0004: + + def __init__(self, host: "LiberviaBackend") -> None: + log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") + self.host = host + host.register_namespace("x-data", data_form.NS_X_DATA) + + def get_handler(self, client): + return XEP_0004_handler(self) + + +@implementer(disco.IDisco) +class XEP_0004_handler(XMPPHandler): + """Handler for pubsub extended discovery requests.""" + + def __init__(self, plugin_parent: XEP_0004) -> None: + self.plugin_parent = plugin_parent + + def getDiscoInfo( + self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" + ) -> list[disco.DiscoFeature]: + """Get disco info for pubsub extended discovery + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: optional node identifier + @return: list of disco features + """ + return [ + disco.DiscoFeature(data_form.NS_X_DATA), + disco.DiscoFeature(NS_XML_ELEMENT), + ] + + def getDiscoItems( + self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" + ) -> list[disco.DiscoItem]: + """Get disco items with extended discovery support + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: optional node identifier + @return: list of disco items + """ + return []