Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0298.py @ 4292:dd0891d0b22b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
plugin XEP-0298: Delivering Conference Information to Jingle Participants (Coin) implementation: This is the first draft of XEP-0298 implementation. The focus is to implement elements needed for A/V Conferences protoXEP. rel 447
author Goffi <goffi@goffi.org>
date Mon, 29 Jul 2024 03:31:13 +0200
parents
children a0ed5c976bf8
comparison
equal deleted inserted replaced
4291:39ac821ebbdb 4292:dd0891d0b22b
1 #!/usr/bin/env python3
2
3
4 # Libervia plugin to handle events
5 # Copyright (C) 2009-2022 Jérôme Poisson (goffi@goffi.org)
6
7 # This program is free software: you can redistribute it and/or modify
8 # it under the terms of the GNU Affero General Public License as published by
9 # the Free Software Foundation, either version 3 of the License, or
10 # (at your option) any later version.
11
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU Affero General Public License for more details.
16
17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20 from urllib.parse import quote, unquote
21 from twisted.words.protocols.jabber import jid
22 from twisted.words.xish import domish
23 from libervia.backend.core.i18n import _
24 from libervia.backend.core.constants import Const as C
25 from libervia.backend.core.log import getLogger
26 from wokkel import disco, iwokkel
27 from zope.interface import implementer
28 from twisted.words.protocols.jabber.xmlstream import XMPPHandler
29
30 log = getLogger(__name__)
31
32
33 PLUGIN_INFO = {
34 C.PI_NAME: "Events",
35 C.PI_IMPORT_NAME: "XEP-0298",
36 C.PI_TYPE: "XEP",
37 C.PI_MODES: C.PLUG_MODE_BOTH,
38 C.PI_PROTOCOLS: [],
39 C.PI_DEPENDENCIES: [],
40 C.PI_MAIN: "XEP_0298",
41 C.PI_HANDLER: "yes",
42 C.PI_DESCRIPTION: _(
43 "Delivering Conference Information to Jingle Participants (Coin). This plugin "
44 "is used to associate metadata about uses an call states in multi-party calls."
45 ),
46 }
47
48 NS_COIN = "urn:xmpp:coin:1"
49 NS_CONFERENCE_INFO = "urn:ietf:params:xml:ns:conference-info"
50
51
52 class XEP_0298:
53 namespace = NS_COIN
54
55 def __init__(self, host):
56 log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
57 self.host = host
58
59 def get_handler(self, client):
60 return CoinHandler(self)
61
62 def add_conference_info(
63 self, jingle_elt: domish.Element, is_focus: bool = False, **kwargs
64 ) -> domish.Element:
65 """Create and return a <conference_info> element
66
67 @param jingle_elt: parent element
68 @param kwargs: attributes of the element.
69 @return: created <conference-info> element.
70 """
71 conference_info_elt = jingle_elt.addElement(
72 (NS_CONFERENCE_INFO, "conference-info"),
73 )
74 if is_focus:
75 conference_info_elt["isfocus"] = C.BOOL_TRUE
76 conference_info_elt.attributes.update(kwargs)
77 return conference_info_elt
78
79 def add_user(
80 self, conference_info_elt: domish.Element, entity: jid.JID
81 ) -> domish.Element:
82 """Add an user to a cconference info element.
83
84 If the parent <users> doesn't exist, it will be created.
85 @param conference_info_elt: <conference-info> element where the <user> element
86 need to be added
87 @param entity: XMPP JID to use as entity.
88 @return: created <user> element.
89 """
90 try:
91 users_elt = next(conference_info_elt.elements(NS_CONFERENCE_INFO, "users"))
92 except StopIteration:
93 users_elt = conference_info_elt.addElement("users")
94
95 user_elt = users_elt.addElement("user")
96 user_elt["entity"] = f"xmpp:{quote(entity.userhost())}"
97 return user_elt
98
99 def parse(self, jingle_elt: domish.Element) -> dict:
100 """Parse a Jingle element and return a dictionary with conference info if present.
101
102 @param jingle_elt: Jingle element to parse.
103 @return: Dictionary with "info" key if conference info is found.
104 """
105 try:
106 conference_info_elt = next(
107 jingle_elt.elements((NS_CONFERENCE_INFO, "conference-info"))
108 )
109 except StopIteration:
110 return {}
111
112 users = []
113 try:
114 users_elt = next(conference_info_elt.elements(NS_CONFERENCE_INFO, "users"))
115 for user_elt in users_elt.elements(NS_CONFERENCE_INFO, "user"):
116 entity = user_elt.getAttribute("entity")
117 if entity.startswith("xmpp:"):
118 try:
119 entity = jid.JID(unquote(entity[5:]))
120 users.append(entity)
121 except Exception as e:
122 log.warning(f"Failed to parse entity {entity!r}: {e}")
123 else:
124 log.warning(f"Ignoring non-XMPP entity {entity!r}")
125 except StopIteration:
126 pass
127
128 return {"info": {"users": users}}
129
130
131 @implementer(iwokkel.IDisco)
132 class CoinHandler(XMPPHandler):
133
134 def __init__(self, plugin_parent):
135 self.plugin_parent = plugin_parent
136
137 def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
138 return [
139 disco.DiscoFeature(NS_COIN),
140 ]
141
142 def getDiscoItems(self, requestor, target, nodeIdentifier=""):
143 return []