diff --git a/libervia/backend/core/core_types.py b/libervia/backend/core/core_types.py
index 7cebf318..83b4a615 100644
--- a/libervia/backend/core/core_types.py
+++ b/libervia/backend/core/core_types.py
@@ -18,22 +18,21 @@
from abc import ABC, abstractmethod
from collections import namedtuple
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Type, TypeVar
+from typing import Any, TYPE_CHECKING, Type, TypeVar
+
from twisted.internet import defer
from twisted.python import failure
-from typing_extensions import TypedDict
-
from twisted.words.protocols.jabber import jid as t_jid
from twisted.words.protocols.jabber import xmlstream as mod_xmlstream
from twisted.words.xish import domish
from tx_xmpp import disco
+from typing_extensions import TypedDict
from libervia.backend.core import exceptions
from libervia.backend.models.core import MessageData
if TYPE_CHECKING:
- from libervia.backend.core.xmpp import LiberviaRosterProtocol
+ from libervia.backend.core.xmpp import LiberviaRosterProtocol, SatDiscoProtocol
from libervia.backend.models.tor import TorMode
from treq.client import HTTPClient
from txtorcon import Tor
@@ -87,6 +86,8 @@ class SatXMPPEntity(ABC):
is_component: bool
server_jid: t_jid.JID
port: int
+ disco: "SatDiscoProtocol"
+ discoHandler: disco.DiscoHandler
identities: list[disco.DiscoIdentity]
xmlstream: mod_xmlstream.XmlStream
tor: "Tor | None" = None
diff --git a/libervia/backend/memory/disco.py b/libervia/backend/memory/disco.py
index 60fbd0c1..b430b5df 100755
--- a/libervia/backend/memory/disco.py
+++ b/libervia/backend/memory/disco.py
@@ -17,7 +17,8 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
-from typing import Iterable, Optional, cast
+from collections.abc import Iterable
+from typing import cast
from twisted.internet.interfaces import IReactorCore
from libervia.backend.core.i18n import _
@@ -34,34 +35,22 @@ from libervia.backend.core.constants import Const as C
from libervia.backend.tools import xml_tools
from libervia.backend.memory import persistent
from tx_xmpp import disco
-from base64 import b64encode
-from hashlib import sha1
+import time
log = getLogger(__name__)
TIMEOUT = 15
-CAP_HASH_ERROR = "ERROR"
+# Cache TTL for successful disco queries (in seconds)
+DISCO_CACHE_TTL = 900
+# Delay before retrying a failed disco query (in seconds)
+DISCO_RETRY_DELAY = 300
class HashGenerationError(Exception):
pass
-class ByteIdentity(object):
- """This class manage identity as bytes (needed for i;octet sort), it is used for the hash generation"""
-
- def __init__(self, identity, lang=None):
- assert isinstance(identity, disco.DiscoIdentity)
- self.category = identity.category.encode("utf-8")
- self.idType = identity.type.encode("utf-8")
- self.name = identity.name.encode("utf-8") if identity.name else b""
- self.lang = lang.encode("utf-8") if lang is not None else b""
-
- def __bytes__(self):
- return b"%s/%s/%s/%s" % (self.category, self.idType, self.lang, self.name)
-
-
class HashManager(object):
"""map object which manage hashes
@@ -98,7 +87,8 @@ class HashManager(object):
if not disco_info.features and not disco_info.identities:
log.warning(
_(
- "no feature/identity found in disco element (hash: {cap_hash}), ignoring: {xml}"
+ "no feature/identity found in disco element (hash: "
+ "{cap_hash}), ignoring: {xml}"
).format(cap_hash=hash_, xml=xml)
)
else:
@@ -111,12 +101,19 @@ class HashManager(object):
return d
-class Discovery(object):
+class Discovery:
"""Manage capabilities of entities"""
def __init__(self, host):
self.host = host
- # TODO: remove legacy hashes
+ # JID-based disco info cache, it is used for entities without XEP-0115 support or
+ # when something goes wrong with hash validation.
+ # Cache entries are tuples: (data, expiry_timestamp)
+ # - (disco.DiscoInfo, float): successful query, expires at timestamp
+ # - (None, float): failed query, retry after DISCO_RETRY_DELAY
+ self._jid_disco_cache: dict[
+ jid.JID, tuple[disco.DiscoInfo | None, float]
+ ] = {}
def load(self):
"""Load persistent hashes"""
@@ -191,7 +188,7 @@ class Discovery(object):
client: SatXMPPEntity,
category: str,
type_: str,
- jid_: Optional[jid.JID] = None,
+ jid_: jid.JID | None = None,
node: str = "",
) -> bool:
"""Tell if an entity has the requested identity
@@ -229,19 +226,42 @@ class Discovery(object):
client, jid_, [C.ENTITY_CAP_HASH]
)[C.ENTITY_CAP_HASH]
except (KeyError, exceptions.UnknownEntityError):
- # capability hash is not available, we'll compute one
+ # Capability hash is not available, check JID-based cache first
+ if use_cache and jid_ in self._jid_disco_cache:
+ data, timestamp = self._jid_disco_cache[jid_]
+ elapsed = time.time() - timestamp
+ if data is None:
+ # Cached failure
+ if elapsed < DISCO_RETRY_DELAY:
+ log.debug(
+ f"Cached disco failure for {jid_.full()}, retry in "
+ f"{DISCO_RETRY_DELAY - int(elapsed)} seconds"
+ )
+ raise exceptions.ExternalRequestError(
+ f"Disco query failed for {jid_.full()}, retrying in "
+ f"{DISCO_RETRY_DELAY} seconds"
+ )
+ # Retry delay expired, remove from cache and retry
+ log.debug(f"Disco retry delay expired for {jid_.full()}, retrying")
+ del self._jid_disco_cache[jid_]
+ elif elapsed < DISCO_CACHE_TTL:
+ # Successful cache hit
+ log.debug(f"JID-based disco cache hit for {jid_.full()}")
+ return defer.succeed(data)
+ else:
+ # Cache expired, remove and retry
+ log.debug(f"Disco cache expired for {jid_.full()}, re-querying")
+ del self._jid_disco_cache[jid_]
+
+ # No cache is used, we will do a query to return disco infos.
+ # The query result will be cached in JID cache.
def infos_cb(disco_infos):
- cap_hash = self.generate_hash(disco_infos)
for ext_form in disco_infos.extensions.values():
# tx_xmpp doesn't call typeCheck on reception, so we do it here
- # to avoid ending up with incorrect types. We have to do it after
- # the hash has been generated (str value is needed to compute the
- # hash)
+ # to avoid ending up with incorrect types.
ext_form.typeCheck()
- self.hashes[cap_hash] = disco_infos
- self.host.memory.update_entity_data(
- client, jid_, C.ENTITY_CAP_HASH, cap_hash
- )
+ # Cache disco info per JID with expiry timestamp
+ self._jid_disco_cache[jid_] = (disco_infos, time.time())
return disco_infos
def infos_eb(fail):
@@ -254,17 +274,10 @@ class Discovery(object):
except AttributeError:
reason = str(fail)
- log.warning(
- "can't request disco infos from {jid}: {reason}".format(
- jid=jid_.full(), reason=reason
- )
- )
+ log.warning(f"can't request disco infos from {jid_.full()}: {reason}")
- # XXX we set empty disco in cache, to avoid getting an error or waiting
- # for a timeout again the next time
- self.host.memory.update_entity_data(
- client, jid_, C.ENTITY_CAP_HASH, CAP_HASH_ERROR
- )
+ # Cache the failure with timestamp to avoid repeated queries
+ self._jid_disco_cache[jid_] = (None, time.time())
raise fail
d = client.disco.requestInfo(jid_, nodeIdentifier=node)
@@ -406,58 +419,6 @@ class Discovery(object):
await defer.DeferredList(defer_list)
return found_entities
- def generate_hash(self, services):
- """Generate a unique hash for given service
-
- hash algorithm is the one described in XEP-0115
- @param services: iterable of disco.DiscoIdentity/disco.DiscoFeature, as returned by discoHandler.info
-
- """
- s = []
- # identities
- byte_identities = [
- ByteIdentity(service)
- for service in services
- if isinstance(service, disco.DiscoIdentity)
- ] # FIXME: lang must be managed here
- byte_identities.sort(key=lambda i: i.lang)
- byte_identities.sort(key=lambda i: i.idType)
- byte_identities.sort(key=lambda i: i.category)
- for identity in byte_identities:
- s.append(bytes(identity))
- s.append(b"<")
- # features
- byte_features = [
- service.encode("utf-8")
- for service in services
- if isinstance(service, disco.DiscoFeature)
- ]
- byte_features.sort() # XXX: the default sort has the same behaviour as the requested RFC 4790 i;octet sort
- for feature in byte_features:
- s.append(feature)
- s.append(b"<")
-
- # extensions
- ext = list(services.extensions.values())
- ext.sort(key=lambda f: f.formNamespace.encode("utf-8"))
- for extension in ext:
- s.append(extension.formNamespace.encode("utf-8"))
- s.append(b"<")
- fields = extension.fieldList
- fields.sort(key=lambda f: f.var.encode("utf-8"))
- for field in fields:
- s.append(field.var.encode("utf-8"))
- s.append(b"<")
- values = [v.encode("utf-8") for v in field.values]
- values.sort()
- for value in values:
- s.append(value)
- s.append(b"<")
-
- cap_hash = b64encode(sha1(b"".join(s)).digest()).decode("utf-8")
- log.debug(_("Capability hash generated: [{cap_hash}]").format(cap_hash=cap_hash))
- return cap_hash
-
@defer.inlineCallbacks
def _disco_infos(
self, entity_jid_s, node="", use_cache=True, profile_key=C.PROF_KEY_NONE
diff --git a/libervia/backend/plugins/plugin_xep_0115.py b/libervia/backend/plugins/plugin_xep_0115.py
index aeec5b52..d2d070fa 100644
--- a/libervia/backend/plugins/plugin_xep_0115.py
+++ b/libervia/backend/plugins/plugin_xep_0115.py
@@ -17,29 +17,40 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
+from base64 import b64encode
+from dataclasses import dataclass, field
+from hashlib import sha1
+from typing import TYPE_CHECKING
+
+from twisted.internet import defer, error
+from twisted.words.protocols.jabber import jid
+from twisted.words.protocols.jabber.xmlstream import XMPPHandler
+from twisted.words.xish import domish
+from tx_xmpp import disco, itx_xmpp
+from zope.interface import implementer
+
from libervia.backend import G
+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.constants import Const as C
from libervia.backend.core.log import getLogger
log = getLogger(__name__)
-from twisted.words.xish import domish
-from twisted.words.protocols.jabber import jid
-from twisted.internet import defer, error
-from zope.interface import implementer
-from tx_xmpp import disco, itx_xmpp
-from twisted.words.protocols.jabber.xmlstream import XMPPHandler
+
+if TYPE_CHECKING:
+ from libervia.backend.core.main import LiberviaBackend
PRESENCE = "/presence"
NS_ENTITY_CAPABILITY = "http://jabber.org/protocol/caps"
NS_CAPS_OPTIMIZE = "http://jabber.org/protocol/caps#optimize"
CAPABILITY_UPDATE = PRESENCE + '/c[@xmlns="' + NS_ENTITY_CAPABILITY + '"]'
+IMPORT_NAME = "XEP-0115"
+
PLUGIN_INFO = {
C.PI_NAME: "XEP 0115 Plugin",
- C.PI_IMPORT_NAME: "XEP-0115",
+ C.PI_IMPORT_NAME: IMPORT_NAME,
C.PI_TYPE: "XEP",
C.PI_MODES: C.PLUG_MODE_BOTH,
C.PI_PROTOCOLS: ["XEP-0115"],
@@ -50,35 +61,68 @@ PLUGIN_INFO = {
}
-class XEP_0115:
- cap_hash = None # capabilities hash is class variable as it is common to all profiles
+@dataclass
+class CapsClientData:
+ """Per-client data for XEP-0115 capabilities."""
- def __init__(self, host):
- log.info(_("Plugin XEP_0115 initialization"))
+ #: True if caps optimisation is available
+ caps_optimize: bool = False
+ #: True if a presence with caps has been sent
+ caps_sent: bool = False
+ #: Our own capability hash
+ caps_hash: str | None = None
+ #: The element built for this client's presence
+ caps_elt: domish.Element | None = field(default=None)
+
+
+class ByteIdentity:
+ """This class manage identity as bytes (needed for i;octet sort)
+
+ It is used for the hash generation.
+ """
+
+ def __init__(self, identity: disco.DiscoIdentity):
+ assert isinstance(identity, disco.DiscoIdentity)
+ self.category = identity.category.encode("utf-8")
+ self.idType = identity.type.encode("utf-8")
+ self.name = identity.name.encode("utf-8") if identity.name else b""
+ self.lang = identity.lang.encode("utf-8") if identity.lang else b""
+
+ def __bytes__(self):
+ return b"%s/%s/%s/%s" % (self.category, self.idType, self.lang, self.name)
+
+
+class XEP_0115:
+ def __init__(self, host: "LiberviaBackend") -> None:
self.host = host
+ log.info(_("Plugin XEP_0115 initialization"))
host.trigger.add("Presence send", self._presence_trigger)
- def get_handler(self, client):
+ def get_handler(self, client: SatXMPPEntity) -> XEP_0115_handler:
return XEP_0115_handler(self)
- async def _prepare_caps(self, client: SatXMPPEntity):
+ async def _prepare_caps(self, client: SatXMPPEntity) -> None:
# we have to calculate hash for client
# because disco infos/identities may change between clients
+ data = client.get_plugin_data(IMPORT_NAME, CapsClientData)
+
# optimize check
- client._caps_optimize = await self.host.has_feature(client, NS_CAPS_OPTIMIZE)
- if client._caps_optimize:
+ data.caps_optimize = await G.host.has_feature(client, NS_CAPS_OPTIMIZE)
+ if data.caps_optimize:
log.info(_("Caps optimisation enabled"))
- client._caps_sent = False
+ data.caps_sent = False
else:
log.warning(_("Caps optimisation not available"))
# hash generation
+ assert data.caps_hash is None, "caps_hash already set"
_infos = await client.discoHandler.info(client.jid, client.jid, "")
disco_infos = disco.DiscoInfo()
for item in _infos:
disco_infos.append(item)
- cap_hash = client._caps_hash = self.host.memory.disco.generate_hash(disco_infos)
+ cap_hash = self.generate_hash(disco_infos)
+ data.caps_hash = cap_hash
log.info(
"Our capability hash has been generated: [{cap_hash}]".format(
cap_hash=cap_hash
@@ -89,24 +133,32 @@ class XEP_0115:
c_elt["hash"] = "sha-1"
c_elt["node"] = C.APP_URL
c_elt["ver"] = cap_hash
- client._caps_elt = c_elt
- if client._caps_optimize:
- client._caps_sent = False
- if cap_hash not in self.host.memory.disco.hashes:
- self.host.memory.disco.hashes[cap_hash] = disco_infos
- self.host.memory.update_entity_data(
+ data.caps_elt = c_elt
+ if data.caps_optimize:
+ data.caps_sent = False
+ if cap_hash not in G.host.memory.disco.hashes:
+ G.host.memory.disco.hashes[cap_hash] = disco_infos
+ G.host.memory.update_entity_data(
client, client.jid, C.ENTITY_CAP_HASH, cap_hash
)
- def _presence_add_elt(self, client, obj):
- if client._caps_optimize:
- if client._caps_sent:
+ def _presence_add_elt(self, client: SatXMPPEntity, obj: domish.Element) -> None:
+ data = client.get_plugin_data(IMPORT_NAME, CapsClientData)
+ if data.caps_optimize:
+ if data.caps_sent:
return
- client.caps_sent = True
- obj.addChild(client._caps_elt)
+ data.caps_sent = True
+ if data.caps_elt is not None:
+ obj.addChild(data.caps_elt)
- def _presence_trigger(self, client, obj, presence_d):
- if not hasattr(client, "_caps_optimize"):
+ def _presence_trigger(
+ self,
+ client: SatXMPPEntity,
+ obj: domish.Element,
+ presence_d: defer.Deferred[domish.Element],
+ ) -> bool:
+ data = client.get_plugin_data(IMPORT_NAME, CapsClientData)
+ if data.caps_elt is None:
presence_d.addCallback(
lambda __: defer.ensureDeferred(self._prepare_caps(client))
)
@@ -136,15 +188,13 @@ class XEP_0115:
log.warning(_("Received invalid capabilities tag: %s") % c_elem.toXml())
return
- if c_ver in self.host.memory.disco.hashes:
+ if c_ver in G.host.memory.disco.hashes:
# we already know the hash, we update the jid entity
log.debug(
"hash [%(hash)s] already in cache, updating entity [%(jid)s]"
% {"hash": c_ver, "jid": from_jid.full()}
)
- self.host.memory.update_entity_data(
- client, from_jid, C.ENTITY_CAP_HASH, c_ver
- )
+ G.host.memory.update_entity_data(client, from_jid, C.ENTITY_CAP_HASH, c_ver)
return
if c_hash != "sha-1": # unknown hash method
@@ -154,9 +204,12 @@ class XEP_0115:
"(entity: {entity_jid}, node: {node})"
).format(hash_method=c_hash, entity_jid=from_jid, node=c_node)
)
+ return
try:
- await G.host.get_disco_infos(client, from_jid)
+ # Per XEP-0115 §6.2, include the caps node in the disco#info request
+ caps_node = f"{c_node}#{c_ver}"
+ disco_infos = await G.host.get_disco_infos(client, from_jid, node=caps_node)
except error.ConnectionDone:
return
except Exception as e:
@@ -165,49 +218,252 @@ class XEP_0115:
jid=from_jid.full(), error=str(e)
)
)
- else:
- computed_hash = self.host.memory.get_entity_datum(
- client, from_jid, C.ENTITY_CAP_HASH
+ return
+
+ # Validate disco info structure per XEP-0115 §5.4
+ if not self._validate_caps(from_jid, c_ver, c_node, disco_infos):
+ return
+
+ # disco is validated, we can compute hash
+ cap_hash = self.generate_hash(disco_infos)
+
+ # Compare computed hash with received verification string
+ if cap_hash != c_ver:
+ log.warning(
+ _(
+ "Computed hash differs from received hash for {jid}:\n"
+ "received: [{received}]\n"
+ "computed: [{computed}]\n"
+ "(node: {node})"
+ ).format(
+ jid=from_jid.full(),
+ received=c_ver,
+ computed=cap_hash,
+ node=c_node,
+ )
)
- if computed_hash != c_ver:
+ # It's invalid, we exit early and won't cache.
+ return
+
+ # We're all good, we can cache the disco info
+ G.host.memory.disco.hashes[cap_hash] = disco_infos
+ G.host.memory.update_entity_data(client, from_jid, C.ENTITY_CAP_HASH, cap_hash)
+ log.debug(f"Caps validated and cached for {from_jid}: {cap_hash}")
+
+ def generate_hash(self, services: disco.DiscoInfo) -> str:
+ """Generate a unique hash for given service
+
+ Hash algorithm is the one described in XEP-0115 §5.1 (Generation Method).
+ @param services: disco infos as returned by discoHandler.info()
+
+ """
+ # Step 1: initialize empty string S
+ s = []
+ # Step 2: sort identities by category, then type, then xml:lang (i;octet
+ # collation)
+ byte_identities = [
+ ByteIdentity(service)
+ for service in services
+ if isinstance(service, disco.DiscoIdentity)
+ ]
+ # i;octet (RFC 4790 §9.3) sorting.
+ # Python's bytes comparison is equivalent to i;octet
+ byte_identities.sort(key=lambda i: (i.category, i.idType, i.lang))
+ # Step 3: append each identity formatted as cat/type/lang/name followed by '<'
+ for identity in byte_identities:
+ s.append(bytes(identity))
+ s.append(b"<")
+ # features
+ byte_features = [
+ service.encode("utf-8")
+ for service in services
+ if isinstance(service, disco.DiscoFeature)
+ ]
+ # Step 4: sort the features
+ byte_features.sort()
+ # Step 5: append each feature followed by '<'
+ for feature in byte_features:
+ s.append(feature)
+ s.append(b"<")
+
+ # Step 6: sort forms by FORM_TYPE; Step 7: process each form's fields
+ ext = list(services.extensions.values())
+ ext.sort(key=lambda f: f.formNamespace.encode("utf-8"))
+ for extension in ext:
+ # Step 7.a: append FORM_TYPE field's value (formNamespace) followed by '<'
+ s.append(extension.formNamespace.encode("utf-8"))
+ s.append(b"<")
+ # Step 7.b: sort fields by "var" attribute
+ fields = extension.fieldList
+ fields.sort(key=lambda f: f.var.encode("utf-8"))
+ for field in fields:
+ # Step 7.c: "For each field other than FORM_TYPE"
+ if field.var == "FORM_TYPE":
+ continue
+ # Step 7.c.a: append var attribute followed by '<'
+ s.append(field.var.encode("utf-8"))
+ s.append(b"<")
+ # Step 7.c.b: sort values
+ values = [v.encode("utf-8") for v in field.values]
+ values.sort()
+ # Step 7.c.c: append each value followed by '<'
+ for value in values:
+ s.append(value)
+ s.append(b"<")
+
+ # Step 9: compute verification string (SHA-1 -> Base64)
+ cap_hash = b64encode(sha1(b"".join(s)).digest()).decode("utf-8")
+ log.debug(_("Capability hash generated: [{cap_hash}]").format(cap_hash=cap_hash))
+ return cap_hash
+
+ def _validate_caps(
+ self,
+ from_jid: jid.JID,
+ c_ver: str,
+ c_node: str,
+ disco_infos: disco.DiscoInfo,
+ ) -> bool:
+ """Validate capabilities disco info structure.
+
+ @param from_jid: The JID that sent the presence with caps.
+ @param c_ver: The verification string from the caps element.
+ @param c_node: The node attribute from the caps element.
+ @param disco_infos: The fetched disco#info response.
+ @return: True if disco info is well-formed, False otherwise.
+ """
+ # §5.4 step 3.c: duplicate identities with same category/type/lang/name
+ seen_identities = set()
+ for item in disco_infos:
+ if isinstance(item, disco.DiscoIdentity):
+ lang = item.lang or ""
+ identity_key = (item.category, item.type, lang)
+ if identity_key in seen_identities:
+ log.warning(
+ _(
+ "Duplicate identity found in caps for {jid}: category={cat},"
+ " type={typ}, lang={lang}"
+ ).format(
+ jid=from_jid.full(),
+ cat=item.category,
+ typ=item.type,
+ lang=lang,
+ )
+ )
+ return False
+ seen_identities.add(identity_key)
+
+ # §5.4 step 3.d: duplicate features
+ seen_features = set()
+ for item in disco_infos:
+ if isinstance(item, disco.DiscoFeature):
+ if item in seen_features:
+ log.warning(
+ _("Duplicate feature found in caps for {jid}: {feat}").format(
+ jid=from_jid.full(), feat=item
+ )
+ )
+ return False
+ seen_features.add(item)
+
+ # §5.4 step 3.e: duplicate FORM_TYPEs or FORM_TYPE with multiple values
+ seen_form_types = set()
+ for form_type, form in disco_infos.extensions.items():
+ if form_type in seen_form_types:
log.warning(
- _(
- "Computed hash differ from given hash:\n"
- "given: [{given}]\n"
- "computed: [{computed}]\n"
- "(entity: {entity_jid}, node: {node})"
- ).format(
- given=c_ver,
- computed=computed_hash,
- entity_jid=from_jid,
- node=c_node,
+ _("Duplicate FORM_TYPE found in caps for {jid}: {ft}").format(
+ jid=from_jid.full(), ft=form_type
)
)
+ return False
- # TODO: me must manage the full algorithm described at XEP-0115 #5.4 part 3
+ form_type_field = None
+ for field in form.fieldList:
+ if field.var == "FORM_TYPE":
+ form_type_field = field
+ break
+
+ if form_type_field is None:
+ # §5.4 step 3.f: ignore forms without FORM_TYPE field
+ log.debug(
+ _(
+ "Form without FORM_TYPE field in caps for {jid}, ignoring form"
+ ).format(jid=from_jid.full())
+ )
+ continue
+
+ if len(form_type_field.values) != 1:
+ log.warning(
+ _(
+ "FORM_TYPE field has multiple values in caps for {jid}, ignoring"
+ " form"
+ ).format(jid=from_jid.full())
+ )
+ continue
+
+ seen_form_types.add(form_type)
+
+ return True
@implementer(itx_xmpp.IDisco)
class XEP_0115_handler(XMPPHandler):
- def __init__(self, plugin_parent):
+ def __init__(self, plugin_parent: XEP_0115) -> None:
self.plugin_parent = plugin_parent
@property
- def client(self):
- return self.parent
+ def client(self) -> SatXMPPEntity:
+ return self.parent # type: ignore[return-value]
- def connectionInitialized(self):
+ def connectionInitialized(self) -> None:
assert self.xmlstream is not None
self.xmlstream.addObserver(
CAPABILITY_UPDATE, self.plugin_parent._update, client=self.client
)
- def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
+ def getDiscoInfo(
+ self,
+ requestor: SatXMPPEntity,
+ target: SatXMPPEntity,
+ nodeIdentifier: str = "",
+ ) -> list[disco.DiscoFeature | disco.DiscoIdentity | disco.DiscoExtension]:
+ """
+ Handle disco#info requests, including caps node queries.
+
+ When nodeIdentifier is in the format 'node#ver' (e.g.,
+ 'http://libervia.org#abc123'), this is a caps node query per XEP-0115 §6.2
+ (Discovering Capabilities). We return the cached disco info for that verification
+ string.
+
+ @param requestor: The entity that sent the request.
+ @param target: The entity the request was sent to.
+ @param nodeIdentifier: The optional node being queried.
+ @return: List of disco items (features, identities, extensions).
+ """
+ # Check if this is a caps node query (format: node#ver)
+ if nodeIdentifier and "#" in nodeIdentifier:
+ parts = nodeIdentifier.rsplit("#", 1)
+ if len(parts) == 2:
+ caps_node, cap_ver = parts
+ # Only respond if this matches our caps node
+ if (
+ caps_node == C.APP_URL
+ and cap_ver in self.plugin_parent.host.memory.disco.hashes
+ ):
+ disco_infos = self.plugin_parent.host.memory.disco.hashes[cap_ver]
+ # Return all items from the cached disco info
+ return list(disco_infos)
+
+ # For non-caps queries, return our caps features
return [
disco.DiscoFeature(NS_ENTITY_CAPABILITY),
disco.DiscoFeature(NS_CAPS_OPTIMIZE),
]
- def getDiscoItems(self, requestor, target, nodeIdentifier=""):
+ def getDiscoItems(
+ self,
+ requestor: SatXMPPEntity,
+ target: SatXMPPEntity,
+ nodeIdentifier: str = "",
+ ) -> list[disco.DiscoItem]:
return []
diff --git a/tests/unit/test_plugin_xep_0115.py b/tests/unit/test_plugin_xep_0115.py
new file mode 100644
index 00000000..ad8babf8
--- /dev/null
+++ b/tests/unit/test_plugin_xep_0115.py
@@ -0,0 +1,260 @@
+#!/usr/bin/env python3
+
+# Libervia: an XMPP client
+# Copyright (C) 2009-2026 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 .
+
+"""Tests for XEP-0115 Entity Capabilities implementation."""
+
+from unittest.mock import MagicMock
+from pytest import fixture
+from twisted.words.protocols.jabber import jid
+from tx_xmpp import data_form, disco
+
+from libervia.backend.plugins.plugin_xep_0115 import (
+ NS_ENTITY_CAPABILITY,
+ NS_CAPS_OPTIMIZE,
+ XEP_0115,
+)
+from libervia.backend.core.constants import Const as C
+from libervia.backend.memory.disco import Discovery
+
+
+@fixture
+def xep_0115(host) -> XEP_0115:
+ """Fixture for initializing XEP-0115 plugin."""
+ return XEP_0115(host)
+
+
+@fixture(autouse=True)
+def setup_memory_disco(host):
+ host.memory.disco = Discovery(host)
+ host.memory.disco.hashes = {}
+ return host
+
+
+class TestXEP0115HashGeneration:
+ """Tests for hash generation per XEP-0115 section 4.1."""
+
+ def test_simple_hash_generation(self, xep_0115):
+ """Test hash generation with simple identity and features (XEP-0115 example)."""
+ disco_infos = disco.DiscoInfo()
+ disco_infos.append(disco.DiscoIdentity("client", "pc", "Exodus 0.9.1"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#items"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/muc"))
+
+ cap_hash = xep_0115.generate_hash(disco_infos)
+
+ assert cap_hash == "QgayPKawpkPSDYmwT/WM94uAlu0="
+
+ def test_hash_generation_with_xml_lang(self, xep_0115):
+ """Test hash generation with xml:lang attribute on identities."""
+ disco_infos = disco.DiscoInfo()
+ disco_infos.append(disco.DiscoIdentity("client", "pc", "Psi 0.11", "en"))
+ disco_infos.append(disco.DiscoIdentity("client", "pc", "Ψ 0.11", "el"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#items"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/muc"))
+
+ cap_hash = xep_0115.generate_hash(disco_infos)
+ cap_hash2 = xep_0115.generate_hash(disco_infos)
+ # Verify that the hash is deterministic
+ assert cap_hash == cap_hash2
+
+ # Verify identities with lang are properly included in hash
+ # The hash should be different from one without lang
+ disco_infos_no_lang = disco.DiscoInfo()
+ disco_infos_no_lang.append(disco.DiscoIdentity("client", "pc", "Psi 0.11"))
+ disco_infos_no_lang.append(disco.DiscoIdentity("client", "pc", "Ψ 0.11"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#items"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/muc"))
+
+ cap_hash_no_lang = xep_0115.generate_hash(disco_infos_no_lang)
+ assert cap_hash != cap_hash_no_lang
+
+ def test_hash_generation_spec_complex_example(self, xep_0115):
+ """Test hash generation against the XEP-0115 §5.3 Complex Generation Example."""
+ disco_infos = disco.DiscoInfo()
+ disco_infos.append(disco.DiscoIdentity("client", "pc", "\u03a8 0.11", "el"))
+ disco_infos.append(disco.DiscoIdentity("client", "pc", "Psi 0.11", "en"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#items"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/muc"))
+
+ form = data_form.Form("result")
+ form.formNamespace = "urn:xmpp:dataforms:softwareinfo"
+
+ form_type_field = data_form.Field("hidden", "FORM_TYPE")
+ form_type_field.values = ["urn:xmpp:dataforms:softwareinfo"]
+ form.addField(form_type_field)
+
+ ip_version_field = data_form.Field("text-multi", "ip_version")
+ ip_version_field.values = ["ipv4", "ipv6"]
+ form.addField(ip_version_field)
+
+ os_field = data_form.Field("text", "os")
+ os_field.values = ["Mac"]
+ form.addField(os_field)
+
+ os_version_field = data_form.Field("text", "os_version")
+ os_version_field.values = ["10.5.1"]
+ form.addField(os_version_field)
+
+ software_field = data_form.Field("text", "software")
+ software_field.values = ["Psi"]
+ form.addField(software_field)
+
+ software_version_field = data_form.Field("text", "software_version")
+ software_version_field.values = ["0.11"]
+ form.addField(software_version_field)
+
+ disco_infos.append(form)
+
+ cap_hash = xep_0115.generate_hash(disco_infos)
+
+ # Expected hash from XEP-0115 §5.3
+ assert cap_hash == "q07IKJEyjvHSyhy//CH0CxmKi8w="
+
+
+class TestXEP0115Handler:
+ """Tests for XEP-0115 handler."""
+
+ def test_getDiscoInfo_returns_caps_features(self, xep_0115):
+ """getDiscoInfo returns caps features for non-caps queries."""
+ handler = xep_0115.get_handler(MagicMock())
+
+ result = handler.getDiscoInfo(MagicMock(), MagicMock(), "")
+
+ assert disco.DiscoFeature(NS_ENTITY_CAPABILITY) in result
+ assert disco.DiscoFeature(NS_CAPS_OPTIMIZE) in result
+
+ def test_getDiscoInfo_caps_node_query(self, host, xep_0115):
+ """getDiscoInfo returns cached disco info for caps node queries."""
+ handler = xep_0115.get_handler(MagicMock())
+
+ # Create and cache some disco info
+ disco_infos = disco.DiscoInfo()
+ disco_infos.append(disco.DiscoIdentity("client", "pc", "Test Client"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+
+ cap_hash = xep_0115.generate_hash(disco_infos)
+ host.memory.disco.hashes[cap_hash] = disco_infos
+
+ # Query with caps node format
+ node_identifier = f"{C.APP_URL}#{cap_hash}"
+ result = handler.getDiscoInfo(MagicMock(), MagicMock(), node_identifier)
+
+ # Should return the cached disco info items
+ assert len(result) > 0
+ assert any(isinstance(item, disco.DiscoIdentity) for item in result)
+ assert any(isinstance(item, disco.DiscoFeature) for item in result)
+
+ # Verify that the returned items match the cached disco info
+ identities = [item for item in result if isinstance(item, disco.DiscoIdentity)]
+ features = [item for item in result if isinstance(item, disco.DiscoFeature)]
+ assert len(identities) == len(disco_infos.identities)
+ assert len(features) == len(disco_infos.features)
+ assert identities[0].name == "Test Client"
+ assert disco.DiscoFeature("http://jabber.org/protocol/caps") in features
+ assert disco.DiscoFeature("http://jabber.org/protocol/disco#info") in features
+
+ def test_getDiscoInfo_unknown_caps_node(self, xep_0115):
+ """getDiscoInfo returns caps features for unknown caps nodes."""
+ handler = xep_0115.get_handler(MagicMock())
+
+ # Query with unknown hash
+ node_identifier = f"{C.APP_URL}#unknownhash123"
+ result = handler.getDiscoInfo(MagicMock(), MagicMock(), node_identifier)
+
+ # Should return just the caps features, not cached info
+ assert disco.DiscoFeature(NS_ENTITY_CAPABILITY) in result
+ assert disco.DiscoFeature(NS_CAPS_OPTIMIZE) in result
+
+ def test_getDiscoItems_empty(self, xep_0115):
+ """getDiscoItems returns empty list (caps nodes are virtual)."""
+ handler = xep_0115.get_handler(MagicMock())
+
+ result = handler.getDiscoItems(MagicMock(), MagicMock(), "")
+
+ assert result == []
+
+
+class TestXEP0115Validation:
+ """Tests for caps validation."""
+
+ def test_validate_disco_info_well_formed(self, xep_0115):
+ """Validation passes for well-formed disco info."""
+ disco_infos = disco.DiscoInfo()
+ disco_infos.append(disco.DiscoIdentity("client", "pc", "Test"))
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+
+ assert (
+ xep_0115._validate_caps(
+ jid.JID("test@example.com"),
+ "testhash",
+ "http://test.com",
+ disco_infos,
+ )
+ is True
+ )
+
+ def test_validate_disco_info_duplicate_identity(self, xep_0115, capsys):
+ """Validation fails for duplicate identities."""
+ disco_infos = disco.DiscoInfo()
+ disco_infos.append(disco.DiscoIdentity("client", "pc", "Test", "en"))
+ disco_infos.append(
+ disco.DiscoIdentity("client", "pc", "Test2", "en")
+ ) # Duplicate category/type/lang
+
+ assert (
+ xep_0115._validate_caps(
+ jid.JID("test@example.com"),
+ "testhash",
+ "http://test.com",
+ disco_infos,
+ )
+ is False
+ )
+
+ captured = capsys.readouterr()
+ assert "Duplicate identity found" in captured.out
+
+ def test_validate_disco_info_duplicate_feature(self, xep_0115, capsys):
+ """Validation fails for duplicate features."""
+ disco_infos = disco.DiscoInfo()
+ disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+ disco_infos.append(
+ disco.DiscoFeature("http://jabber.org/protocol/caps")
+ ) # Duplicate
+
+ assert (
+ xep_0115._validate_caps(
+ jid.JID("test@example.com"),
+ "testhash",
+ "http://test.com",
+ disco_infos,
+ )
+ is False
+ )
+
+ captured = capsys.readouterr()
+ assert "Duplicate feature found" in captured.out