# HG changeset patch # User Goffi # Date 1727531764 -7200 # Node ID 6a70fcd93a7a9fff879f6922ecbf4a10ffa60623 # Parent 530f86f078ccd261a5ae95054b9ec35504c087a6 plugin XEP-0131: Stanza Headers and Internet Metadata implementation: - SHIM is now supported and put in `msg_data["extra"]["headers"]`. - `Keywords` are converted from and to list of string in `msg_data["extra"]["keywords"]` field (if present in headers on message sending, values are merged). - Python minimal version upgraded to 3.11 due to use of `StrEnum`. rel 451 diff -r 530f86f078cc -r 6a70fcd93a7a libervia/backend/plugins/plugin_xep_0131.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/libervia/backend/plugins/plugin_xep_0131.py Sat Sep 28 15:56:04 2024 +0200 @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 + +# Libervia plugin Stanza Headers and Internet Metadata (XEP-0131) +# Copyright (C) 2009-2024 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 email.utils import quote, unquote +from enum import StrEnum, auto +import re +from typing import Iterator, List, Literal, Optional, Self + +from pydantic import BaseModel, ConfigDict, Field, RootModel +from twisted.internet import defer +from twisted.words.protocols.jabber.xmlstream import XMPPHandler +from twisted.words.xish import domish +from wokkel import 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.core_types import SatXMPPEntity +from libervia.backend.core.i18n import _ +from libervia.backend.core.log import getLogger +from libervia.backend.models.core import MessageData + +log = getLogger(__name__) + +PLUGIN_INFO = { + C.PI_NAME: "Stanza Headers and Internet Metadata Plugin", + C.PI_IMPORT_NAME: "XEP-0131", + C.PI_TYPE: "XEP", + C.PI_MODES: C.PLUG_MODE_BOTH, + C.PI_PROTOCOLS: ["XEP-0131"], + C.PI_DEPENDENCIES: [], + C.PI_MAIN: "XEP_0131", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: _( + "Enables the inclusion of non-addressing header information in XMPP stanzas." + ), +} + +NS_SHIM = "http://jabber.org/protocol/shim" + +# Regex to match quoted and non-quoted values. +RE_QUOTED_VALUES = re.compile( + r""" + # Match quoted phrases + " + (?: + # Match any escaped character + \\. + # Match any character that is not a double quote or a backslash + |[^"\\] + )* + " + | + # Match non-quoted phrases + (?:[^,]+) +""", + re.VERBOSE, +) + + +class Urgency(StrEnum): + low = auto() + medium = auto() + high = auto() + + +class Priority(StrEnum): + non_urgent = "non-urgent" + normal = auto() + urgent = auto() + emergency = auto() + + +class HeadersData(BaseModel): + model_config = ConfigDict(extra="allow") + + __pydantic_extra__: dict[str, str] = Field(init=False) # type: ignore + + keywords: str | None = None + urgency: Urgency | None = None + priority: Priority | None = None + + def items(self): + return self.__pydantic_extra__.items() + + def to_element(self) -> domish.Element: + """Build the element from this instance's data.""" + headers_elt = domish.Element((NS_SHIM, "headers")) + header_names = list(self.model_fields.keys()) + list( + self.__pydantic_extra__.keys() + ) + for name in header_names: + value = getattr(self, name) + if value is None: + continue + header_elt = headers_elt.addElement("header") + header_elt["name"] = name + header_elt.addContent(value) + return headers_elt + + @classmethod + def from_element(cls, headers_elt: domish.Element) -> Self: + """Create a HeadersData instance from a element.""" + if headers_elt.uri != NS_SHIM or headers_elt.name != "headers": + child_headers_elt = next(headers_elt.elements(NS_SHIM, "headers"), None) + if child_headers_elt is None: + raise exceptions.NotFound(" element not found") + else: + headers_elt = child_headers_elt + + headers = {} + for header_elt in headers_elt.elements(NS_SHIM, "header"): + name = header_elt.getAttribute("name") + value = str(header_elt) + headers[name] = value + return cls(**headers) + + +class Keywords(RootModel): + root: list[str] + + def __iter__(self) -> Iterator[str]: # type: ignore + return iter(self.root) + + def __getitem__(self, item) -> str: + return self.root[item] + + def __len__(self) -> int: + return len(self.root) + + +class XEP_0131: + """Implementation for XEP-0131""" + + def __init__(self, host): + log.info(_("Stanza Headers and Internet Metadata plugin initialization")) + self.host = host + host.register_namespace("shim", NS_SHIM) + host.trigger.add("sendMessage", self.send_message_trigger) + host.trigger.add("sendMessageComponent", self.send_message_trigger) + host.trigger.add("message_received", self.message_received_trigger) + + def quote_value(self, value: str) -> str: + """Quote a value if it contain special characters + + @param value: Value to quote if necessary. + @return: Quoted value. + """ + if any(c in value for c in r" ,\""): + value = f'"{quote(value)}"' + return value + + def unquote_values(self, raw_header: str) -> list[str]: + """Unquote raw list of values header. + + This is raw header for potentially quoted values separated by commas, like in the + "keywords" header. + + @param raw_keywords_header: Raw Keywords header. + @return: A list of unquoted strings. + """ + unquoted_values = [] + + for match in RE_QUOTED_VALUES.finditer(raw_header): + value = match.group(0).strip() + + # Unquote the keyword if needed. + if value.startswith('"') and value.endswith('"'): + value = unquote(value) + + value = value.strip() + if value: + unquoted_values.append(value) + + return unquoted_values + + def move_keywords_to_headers(self, extra: dict) -> None: + """Check if keywords are present in extra, and move them to headers. + + The list of keywords will be converted to a header value and set in the right + location. + + @param extra: MessageData's ``extra`` field. Will be modified in place by + creating/updating the ``headers`` field. + """ + # Keywords can be in a list of strings in extra's "keywords" field. + if "keywords" in extra: + keywords = Keywords(extra["keywords"]) + if keywords: + headers = extra.setdefault("headers", {}) + quoted_kw = ",".join(self.quote_value(kw) for kw in keywords) + existing_kw = headers.get("keywords") + if existing_kw: + # We have also keywords in headers, we merge both. + quoted_kw = f"{existing_kw},{quoted_kw}" + headers["keywords"] = quoted_kw + + def send_message_trigger( + self, client, mess_data, pre_xml_treatments, post_xml_treatments + ) -> Literal[True]: + """Process the XEP-0131 related data to be sent""" + + def add_headers(mess_data: MessageData) -> MessageData: + extra = mess_data["extra"] + self.move_keywords_to_headers(extra) + # Now we parse headers, if any. + if "headers" in extra: + headers_data = HeadersData(**extra["headers"]) + message_elt = mess_data["xml"] + message_elt.addChild(headers_data.to_element()) + return mess_data + + post_xml_treatments.addCallback(add_headers) + return True + + def message_received_trigger( + self, + client: SatXMPPEntity, + message_elt: domish.Element, + post_treat: defer.Deferred, + ) -> Literal[True]: + """Parse headers information and add them to message data.""" + try: + headers = HeadersData.from_element(message_elt) + except exceptions.NotFound: + pass + else: + + def post_treat_addr(mess_data: MessageData): + """Add the headers metadata to the message data""" + if headers.keywords: + # We move keywords to a list of string in extra's "keywords" field. + mess_data["extra"]["keywords"] = self.unquote_values(headers.keywords) + headers.keywords = None + mess_data["extra"]["headers"] = headers.model_dump( + mode="json", exclude_none=True + ) + return mess_data + + post_treat.addCallback(post_treat_addr) + return True + + def get_handler(self, client): + return XEP_0131_handler() + + +@implementer(disco.IDisco) +class XEP_0131_handler(XMPPHandler): + + def getDiscoInfo( + self, requestor, target, nodeIdentifier: Optional[str] = "" + ) -> List[disco.DiscoFeature]: + return [disco.DiscoFeature(NS_SHIM)] + + def getDiscoItems( + self, requestor, target, nodeIdentifier: Optional[str] = "" + ) -> List[disco.DiscoItem]: + return [] diff -r 530f86f078cc -r 6a70fcd93a7a pyproject.toml --- a/pyproject.toml Sat Sep 28 15:52:14 2024 +0200 +++ b/pyproject.toml Sat Sep 28 15:56:04 2024 +0200 @@ -8,7 +8,7 @@ description = "Libervia: A feature-rich XMPP client showcasing diverse frontends, uniting instant messaging, blogging, file sharing, and ActivityPub-XMPP interactions seamlessly." readme = "README.md" license = "AGPL-3.0-or-later" -requires-python = ">=3.10" +requires-python = ">=3.11" authors = [ { name = "Libervia Dev Team", email = "contact@goffi.org" }, ]