Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0338.py @ 4071:4b842c1fb686

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
refactoring: renamed `sat` package to `libervia.backend`
author Goffi <goffi@goffi.org>
date Fri, 02 Jun 2023 11:49:51 +0200
parents sat/plugins/plugin_xep_0338.py@08ee0e623e7e
children 52a89ddf3087
comparison
equal deleted inserted replaced
4070:d10748475025 4071:4b842c1fb686
1 #!/usr/bin/env python3
2
3 # Libervia plugin
4 # Copyright (C) 2009-2023 Jérôme Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19 from typing import List
20
21 from twisted.words.protocols.jabber.xmlstream import XMPPHandler
22 from twisted.words.xish import domish
23 from wokkel import disco, iwokkel
24 from zope.interface import implementer
25
26 from libervia.backend.core.constants import Const as C
27 from libervia.backend.core.i18n import _
28 from libervia.backend.core.log import getLogger
29 from libervia.backend.core.core_types import SatXMPPEntity
30
31 log = getLogger(__name__)
32
33 NS_JINGLE_GROUPING = "urn:xmpp:jingle:apps:grouping:0"
34 NS_RFC_5888 = "urn:ietf:rfc:5888"
35
36 PLUGIN_INFO = {
37 C.PI_NAME: "Jingle Grouping Framework",
38 C.PI_IMPORT_NAME: "XEP-0338",
39 C.PI_TYPE: "XEP",
40 C.PI_MODES: C.PLUG_MODE_BOTH,
41 C.PI_PROTOCOLS: ["XEP-0338"],
42 C.PI_DEPENDENCIES: ["XEP-0166", "XEP-0167"],
43 C.PI_RECOMMENDATIONS: [],
44 C.PI_MAIN: "XEP_0338",
45 C.PI_HANDLER: "yes",
46 C.PI_DESCRIPTION: _("""Jingle mapping of RFC 5888 SDP Grouping Framework"""),
47 }
48
49
50 class XEP_0338:
51 def __init__(self, host):
52 log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
53 self._j = host.plugins["XEP-0166"]
54 host.trigger.add("XEP-0167_parse_sdp_a", self._parse_sdp_a_trigger)
55 host.trigger.add(
56 "XEP-0167_generate_sdp_session", self._generate_sdp_session_trigger
57 )
58 host.trigger.add("XEP-0167_jingle_session_init", self._jingle_session_init_trigger)
59 host.trigger.add("XEP-0167_jingle_handler", self._jingle_handler_trigger)
60
61 def get_handler(self, client):
62 return XEP_0338_handler()
63
64 def _parse_sdp_a_trigger(
65 self,
66 attribute: str,
67 parts: List[str],
68 call_data: dict,
69 metadata: dict,
70 media_type: str,
71 application_data: dict,
72 transport_data: dict,
73 ) -> None:
74 """Parse "group" attributes"""
75 if attribute == "group":
76 semantics = parts[0]
77 content_names = parts[1:]
78 metadata.setdefault("group", {})[semantics] = content_names
79
80 def _generate_sdp_session_trigger(
81 self,
82 session: dict,
83 local: bool,
84 sdp_lines: List[str],
85 ) -> None:
86 """Generate "group" attributes"""
87 key = "metadata" if local else "peer_metadata"
88 group_data = session[key].get("group", {})
89
90 for semantics, content_names in group_data.items():
91 sdp_lines.append(f"a=group:{semantics} {' '.join(content_names)}")
92
93 def parse_group_element(
94 self, jingle_elt: domish.Element, session: dict
95 ) -> None:
96 """Parse the <group> and <content> elements"""
97 for group_elt in jingle_elt.elements(NS_JINGLE_GROUPING, "group"):
98 try:
99 metadata = session["peer_metadata"]
100 semantics = group_elt["semantics"]
101 group_content = metadata.setdefault("group", {})[semantics] = []
102 for content_elt in group_elt.elements(NS_JINGLE_GROUPING, "content"):
103 group_content.append(content_elt["name"])
104 except KeyError as e:
105 log.warning(f"Error while parsing <group>: {e}\n{group_elt.toXml()}")
106
107 def add_group_element(
108 self, jingle_elt: domish.Element, session: dict
109 ) -> None:
110 """Build the <group> and <content> elements if possible"""
111 for semantics, content_names in session["metadata"].get("group", {}).items():
112 group_elt = jingle_elt.addElement((NS_JINGLE_GROUPING, "group"))
113 group_elt["semantics"] = semantics
114 for content_name in content_names:
115 content_elt = group_elt.addElement((NS_JINGLE_GROUPING, "content"))
116 content_elt["name"] = content_name
117
118 def _jingle_session_init_trigger(
119 self,
120 client: SatXMPPEntity,
121 session: dict,
122 content_name: str,
123 media: str,
124 media_data: dict,
125 desc_elt: domish.Element,
126 ) -> None:
127 jingle_elt = session["jingle_elt"]
128 self.add_group_element(jingle_elt, session)
129
130 def _jingle_handler_trigger(
131 self,
132 client: SatXMPPEntity,
133 action: str,
134 session: dict,
135 content_name: str,
136 desc_elt: domish.Element,
137 ) -> None:
138 # this is a session metadata, so we only generate it on the first content
139 if content_name == next(iter(session["contents"])) and action in (
140 self._j.A_PREPARE_CONFIRMATION,
141 self._j.A_SESSION_INITIATE,
142 self._j.A_PREPARE_INITIATOR,
143 ):
144 jingle_elt = session["jingle_elt"]
145 self.parse_group_element(jingle_elt, session)
146 if action == self._j.A_SESSION_INITIATE:
147 self.add_group_element(jingle_elt, session)
148
149
150 @implementer(iwokkel.IDisco)
151 class XEP_0338_handler(XMPPHandler):
152 def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
153 return [disco.DiscoFeature(NS_RFC_5888)]
154
155 def getDiscoItems(self, requestor, target, nodeIdentifier=""):
156 return []