erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin XEP-0358: "Publishing Available Jingle Sessions" implementation:

Browse source 
rel 453
This commit is contained in:
Goffi 2024-12-03 00:13:07 +01:00
commit 7509323137
2 changed files with 218 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

204
libervia/backend/plugins/plugin_xep_0358.py Normal file
View file

@ -0,0 +1,204 @@
#!/usr/bin/env python3
# Libervia plugin to jingle session publishing.
# 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 <http://www.gnu.org/licenses/>.
from typing import TYPE_CHECKING, Self
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 disco, iwokkel
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.types import DomishElementType, JIDType
if TYPE_CHECKING:
from libervia.backend.core.main import LiberviaBackend
log = getLogger(__name__)
PLUGIN_INFO = {
C.PI_NAME: "Publishing Available Jingle Sessions",
C.PI_IMPORT_NAME: "XEP-0358",
C.PI_TYPE: "XEP",
C.PI_MODES: C.PLUG_MODE_BOTH,
C.PI_PROTOCOLS: [],
C.PI_DEPENDENCIES: [
"XEP-0234",
],
C.PI_RECOMMENDATIONS: [],
C.PI_MAIN: "XEP_0358",
C.PI_HANDLER: "yes",
C.PI_DESCRIPTION: _("""Indicate available Jingle sessions."""),
}
NS_JINGLEPUB = "urn:xmpp:jinglepub:1"
class Meta(BaseModel):
"""Model for meta element in JinglePub element."""
lang: str | None = Field(None, description="The language of the meta data.")
title: str = Field(description="The title of the meta data.")
summary: str | None = Field(None, description="A summary of the meta data.")
def to_element(self) -> domish.Element:
"""Build the <meta> element from this instance's data.
@return: <meta> element.
"""
meta_elt = domish.Element((NS_JINGLEPUB, "meta"))
if self.lang:
meta_elt["xml:lang"] = self.lang
meta_elt["title"] = self.title
if self.summary:
meta_elt["summary"] = self.summary
return meta_elt
@classmethod
def from_element(cls, meta_elt: domish.Element) -> Self:
"""Create a Meta instance from a <meta> element.
@param meta_elt: The <meta> element.
@return: Meta instance.
"""
assert meta_elt.name == "meta" and meta_elt.uri == NS_JINGLEPUB
kwargs = {}
if meta_elt.hasAttribute("xml:lang"):
kwargs["lang"] = meta_elt["xml:lang"]
if meta_elt.hasAttribute("title"):
kwargs["title"] = meta_elt["title"]
if meta_elt.hasAttribute("summary"):
kwargs["summary"] = meta_elt["summary"]
return cls(**kwargs)
class JinglePub(BaseModel):
"""Model for JinglePub element."""
from_jid: JIDType = Field(description="The JID of the session owner.")
id: str = Field(description="The jinglepub identifier.")
uri: str | None = Field(
default=None, description="An alternate or informational URI of the content."
)
meta: list[Meta] = Field(
default_factory=list, description="Human-friendly information about the session."
)
descriptions: list[DomishElementType] = Field(
default_factory=list, description="Application format(s) for the Jingle session."
)
def set_attributes(self, jinglepub_elt: domish.Element) -> None:
"""Set <jinglepub> element attributes from this instance's data."""
jinglepub_elt["from"] = str(self.from_jid)
jinglepub_elt["id"] = self.id
def set_children(self, jinglepub_elt: domish.Element) -> None:
"""Set <jinglepub> element children from this instance's data."""
if self.uri:
uri_elt = jinglepub_elt.addElement((NS_JINGLEPUB, "uri"))
uri_elt.addContent(self.uri)
for meta in self.meta:
jinglepub_elt.addChild(meta.to_element())
for description in self.descriptions:
jinglepub_elt.addChild(description)
def to_element(self) -> domish.Element:
"""Build the <jinglepub> element from this instance's data.
@return: <jinglepub> element.
"""
jinglepub_elt = domish.Element((NS_JINGLEPUB, "jinglepub"))
self.set_attributes(jinglepub_elt)
self.set_children(jinglepub_elt)
return jinglepub_elt
@classmethod
def from_element(cls, element: domish.Element) -> Self:
"""Create a JinglePub instance from a <jinglepub> element.
@param jinglepub_elt: The <jinglepub> element.
@return: JinglePub instance.
@raise exceptions.NotFound: If the <jinglepub> element is not found.
"""
if element.uri != NS_JINGLEPUB or element.name != "jinglepub":
raise exceptions.NotFound("<jinglepub> element not found")
kwargs = {}
try:
kwargs["from_jid"] = jid.JID(element["from"])
except RuntimeError as e:
log.warning(f"Can't parse from_jid {e}: {element.toXml()}")
raise exceptions.NotFound("<jinglepub> element has invalid 'from' attribute")
if element.hasAttribute("id"):
kwargs["id"] = element["id"]
uri_elt = next(element.elements(NS_JINGLEPUB, "uri"), None)
if uri_elt:
kwargs["uri"] = str(uri_elt)
kwargs["meta"] = [
Meta.from_element(meta_elt)
for meta_elt in element.elements(NS_JINGLEPUB, "meta")
]
kwargs["descriptions"] = [
child
for child in element.elements()
if child.uri != NS_JINGLEPUB and child.name == "description"
]
return cls(**kwargs)
class XEP_0358:
namespace = NS_JINGLEPUB
def __init__(self, host: "LiberviaBackend") -> None:
log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
self.host = host
host.register_namespace("jinglepub", NS_JINGLEPUB)
def get_handler(self, client: SatXMPPEntity) -> XMPPHandler:
return JinglePubHandler(self)
@implementer(iwokkel.IDisco)
class JinglePubHandler(XMPPHandler):
def __init__(self, plugin_parent):
self.plugin_parent = plugin_parent
def getDiscoInfo(
self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = ""
) -> list[disco.DiscoFeature]:
return [
disco.DiscoFeature(NS_JINGLEPUB),
]
def getDiscoItems(
self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = ""
) -> list[disco.DiscoItems]:
return []

14
libervia/backend/plugins/plugin_xep_0447.py
View file

@ -47,6 +47,7 @@ from libervia.backend.core.core_types import SatXMPPEntity
from libervia.backend.core.i18n import _
from libervia.backend.core.log import getLogger
from libervia.backend.plugins.plugin_xep_0103 import URLData, XEP_0103
from libervia.backend.plugins.plugin_xep_0358 import JinglePub, XEP_0358
from libervia.backend.plugins.plugin_xep_0446 import FileMetadata, XEP_0446
from libervia.backend.tools import stream
from libervia.backend.tools.web import treq_client_no_ssl
@ -63,6 +64,7 @@ PLUGIN_INFO = {
C.PI_DEPENDENCIES: [
"XEP-0103",
"XEP-0334",
"XEP-0358",
"XEP-0446",
"ATTACH",
"DOWNLOAD",
@ -186,6 +188,17 @@ class URLDataSource(URLData, Source):
return super().to_element()
class JinglePubSource(JinglePub, Source):
type = "jingle"
@classmethod
def from_element(cls, element: domish.Element) -> Self:
return super().from_element(element)
def to_element(self) -> domish.Element:
return super().to_element()
class XEP_0447:
namespace = NS_SFS
@ -196,6 +209,7 @@ class XEP_0447:
FileSharing._sfs = self
self._sources_handlers: dict[tuple[str, str], type[Source]] = {}
self._u = cast(XEP_0103, host.plugins["XEP-0103"])
self._jp = cast(XEP_0358, host.plugins["XEP-0358"])
self._hints = host.plugins["XEP-0334"]
self._m = cast(XEP_0446, host.plugins["XEP-0446"])
self._http_upload = host.plugins.get("XEP-0363")