1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
|
########################################################################
# File name: service.py
# This file is part of: aioxmpp
#
# LICENSE
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program. If not, see
# <http://www.gnu.org/licenses/>.
#
########################################################################
import asyncio
import typing
import aioxmpp
import aioxmpp.disco
import aioxmpp.errors
import aioxmpp.service
from . import xso as version_xso
class VersionServer(aioxmpp.service.Service):
"""
:class:`~aioxmpp.service.Service` which handles inbound :xep:`92` Software
Version requests.
.. warning::
Do **not** depend on this service in another service. This service
exposes possibily private or sensitive information over the XMPP
network without any filtering. Implicitly summoning this service via
a dependency is thus discouraged.
*If* you absolutely need to do this for the implementation of another
published XEP, please file an issue against :mod:`aioxmpp` so that we
can work out a good solution.
.. warning::
This service does answer version queries, no matter who asks. This may
not be desirable, in which case this service is not for you.
.. seealso::
:func:`~.version.query_version`
for a function to obtain another entities software version.
.. note::
By default, this service does not reply to version queries. The
:attr:`name` attribute needs to be set first.
The response can be configured with the following attributes:
.. autoattribute:: name
:annotation: = None
.. autoattribute:: version
:annotation: = None
.. autoattribute:: os
:annotation: = distro.name() or platform.system()
"""
ORDER_AFTER = [
aioxmpp.disco.DiscoServer,
]
disco_feature = aioxmpp.disco.register_feature(
"jabber:iq:version",
)
def __init__(self, client, **kwargs):
super().__init__(client, **kwargs)
try:
import distro
except ImportError:
import platform
self._os = platform.system()
else:
self._os = distro.name()
self._name = None
self._version = None
@property
def os(self) -> typing.Optional[str]:
"""
The operating system of this entity.
Defaults to :func:`distro.name` or :func:`platform.system` (if
:mod:`distro` is not available).
This attribute can be set to :data:`None` or deleted to prevent
inclusion of the OS element in the reply.
"""
return self._os
@os.setter
def os(self, value: typing.Optional[str]):
if value is None:
self._os = None
else:
self._os = str(value)
@os.deleter
def os(self):
self._os = None
@property
def name(self) -> typing.Optional[str]:
"""
The software name of this entity.
Defaults to :data:`None`.
If this attribute is :data:`None`, version requests are not answered
but fail with a ``service-unavailable`` error.
"""
return self._name
@name.setter
def name(self, value: typing.Optional[str]):
if value is None:
self._name = None
else:
self._name = str(value)
@name.deleter
def name(self):
self._name = None
@property
def version(self) -> typing.Optional[str]:
"""
The software version of this entity.
Defaults to :data:`None`.
If this attribute is :data:`None` or the empty string, the version will
be shown as ``"unspecified"`` to other entities. This can be used to
avoid disclosing the specific version of the software.
"""
return self._version
@version.setter
def version(self, value: typing.Optional[str]):
if value is None:
self._version = None
else:
self._version = str(value)
@version.deleter
def version(self):
self._version = None
@aioxmpp.service.iq_handler(aioxmpp.IQType.GET,
version_xso.Query)
async def handle_query(self, iq: aioxmpp.IQ) -> version_xso.Query:
if self._name is None:
raise aioxmpp.errors.XMPPCancelError(
aioxmpp.errors.ErrorCondition.SERVICE_UNAVAILABLE,
)
result = version_xso.Query()
result.name = self._name
result.os = self._os
result.version = self._version or "unspecified"
return result
async def query_version(stream: aioxmpp.stream.StanzaStream,
target: aioxmpp.JID) -> version_xso.Query:
"""
Query the software version of an entity.
:param stream: A stanza stream to send the query on.
:type stream: :class:`aioxmpp.stream.StanzaStream`
:param target: The address of the entity to query.
:type target: :class:`aioxmpp.JID`
:raises OSError: if a connection issue occured before a reply was received
:raises aioxmpp.errors.XMPPError: if an XMPP error was returned instead
of a reply.
:rtype: :class:`aioxmpp.version.xso.Query`
:return: The response from the peer.
The response is returned as :class:`~aioxmpp.version.xso.Query` object. The
attributes hold the data returned by the peer. Each attribute may be
:data:`None` if the peer chose to omit that information. In an extreme
case, all attributes are :data:`None`.
"""
return await stream.send(aioxmpp.IQ(
type_=aioxmpp.IQType.GET,
to=target,
payload=version_xso.Query(),
))
|
