dd79a42931
- the XEP-0115 implementation has been completed with validation, caps node handling in disco requests, hash generation fixes, and real support for xml:lang. - the `generate_hash` method has been moved from disco where it's not used anymore to XEP-0115. - Disco doesn't hash systematically, it nos use a per-jid cache for non XEP-0115 supporting client. - code modernisation, with use of `get_plugin_data` for Client data in XEP-0115, type hints, etc. - More comments have been added to link algorithms steps to the relevant part in specification.
260 lines
11 KiB
Python
260 lines
11 KiB
Python
#!/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 <http://www.gnu.org/licenses/>.
|
|
|
|
"""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
|