erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin XEP-0085: Modernisation:

Browse source 
- Async coroutines, type hints
- Use client plugin data instead of per-profile dict
- Detect chat state availability with disco queries
- Better spec compliance
This commit is contained in:
Goffi 2026-07-02 15:35:23 +02:00
commit 5ed83ae9e3
1 changed files with 352 additions and 169 deletions
Download patch file Download diff file Expand all files Collapse all files

529
libervia/backend/plugins/plugin_xep_0085.py
View file

@ -1,7 +1,7 @@
#!/usr/bin/env python3
# SAT plugin for Chat State Notifications Protocol (xep-0085)
# Libervia plugin for Chat State Notifications Protocol (XEP-0085)
# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)
# Copyright (C) 2009-2016 Adrien Cossa (souliane@mailoo.org)
# This program is free software: you can redistribute it and/or modify
@ -17,20 +17,32 @@
# 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/>.
from libervia.backend.core.i18n import _
from libervia.backend.core.constants import Const as C
from libervia.backend.core import exceptions
from libervia.backend.core.log import getLogger
from dataclasses import dataclass, field
from typing import Any, TYPE_CHECKING, cast
log = getLogger(__name__)
from tx_xmpp import disco, itx_xmpp
from zope.interface import implementer
from twisted.words.protocols.jabber.jid import JID
from twisted.words.protocols.jabber.xmlstream import XMPPHandler
from twisted.words.xish import domish
from twisted.internet import defer
from twisted.internet import reactor
from twisted.internet import error as internet_error
from twisted.internet.base import DelayedCall
from twisted.internet.interfaces import IReactorTime
from twisted.words.protocols.jabber.jid 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.core import exceptions
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.log import getLogger
if TYPE_CHECKING:
from libervia.backend.core.main import LiberviaBackend
log = getLogger(__name__)
NS_XMPP_CLIENT = "jabber:client"
NS_CHAT_STATES = "http://jabber.org/protocol/chatstates"
@ -52,6 +64,18 @@ PLUGIN_INFO = {
C.PI_DESCRIPTION: _("""Implementation of Chat State Notifications Protocol"""),
}
IMPORT_NAME = PLUGIN_INFO[C.PI_IMPORT_NAME]
@dataclass
class ChatStatesData:
"""Per-client data for XEP-0085 chat state notifications."""
# peer_jid -> Allowed (True)/Denied (False)
disco_cache: dict[JID, bool] = field(default_factory=dict)
# Chat state machines, keyed by peer_jid
machines: dict[JID, "ChatStateMachine"] = field(default_factory=dict)
# Describe the internal transitions that are triggered
# by a timer. Beside that, external transitions can be
@ -74,7 +98,7 @@ class UnknownChatStateException(Exception):
pass
class XEP_0085(object):
class XEP_0085:
"""
Implementation for XEP 0085
"""
@ -94,13 +118,9 @@ class XEP_0085(object):
"param_label": _("Enable chat state notifications"),
}
def __init__(self, host):
def __init__(self, host: "LiberviaBackend") -> None:
log.info(_("Chat State Notifications plugin initialization"))
self.host = host
self.map = (
{}
) # FIXME: would be better to use client instead of mapping profile to data
# parameter value is retrieved before each use
host.memory.update_params(self.params)
@ -121,26 +141,104 @@ class XEP_0085(object):
# args: from (jid as string), state in CHAT_STATES, profile
host.bridge.add_signal("chat_state_received", ".plugin", signature="sss")
def get_handler(self, client):
def get_handler(self, client: SatXMPPEntity) -> "XEP_0085_handler":
return XEP_0085_handler(self, client.profile)
def profile_disconnected(self, client):
def profile_disconnected(self, client: SatXMPPEntity) -> None:
"""Eventually send a 'gone' state to all one2one contacts."""
profile = client.profile
if profile not in self.map:
return
for to_jid in self.map[profile]:
chat_states_data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
for machine in list(chat_states_data.machines.values()):
# FIXME: the "unavailable" presence stanza is received by to_jid
# before the chat state, so it will be ignored... find a way to
# actually defer the disconnection
self.map[profile][to_jid]._onEvent("gone")
del self.map[profile]
machine._onEvent("gone")
chat_states_data.machines.clear()
def update_cache(self, entity_jid, value, profile):
async def _check_disco_support(
self, client: SatXMPPEntity, peer_jid: JID
) -> bool | None:
"""Check if a contact supports XEP-0085 via service discovery.
For 1:1 (bare JID): checks all known resources.
For MUC DM (full JID): checks the occupant directly.
@param client: the client instance
@param peer_jid: bare JID for 1:1, full JID for MUC DM
@return: True if confirmed, False if denied, None if no resources known
"""
data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
if peer_jid in data.disco_cache:
return data.disco_cache[peer_jid]
if not peer_jid.resource:
# 1:1 (we check all known resources)
try:
resources = self.host.memory.get_all_resources(client, peer_jid)
except exceptions.UnknownEntityError:
# entity not in cache
return None
if not resources:
# no resources known yet
return None
for resource in resources:
full_jid = JID(None, (peer_jid.user, peer_jid.host, resource))
try:
has_feature = await self.host.memory.disco.has_feature(
client, NS_CHAT_STATES, full_jid
)
if has_feature:
data.disco_cache[peer_jid] = True
return True
except (
exceptions.ExternalRequestError,
exceptions.TimeOutError,
) as e:
log.warning(
f"Can't check chat state availability for {full_jid}: {e}"
)
continue
data.disco_cache[peer_jid] = False
return False
# We have a full JID -> This is a MUC DM
try:
has_feature = await self.host.memory.disco.has_feature(
client, NS_CHAT_STATES, peer_jid
)
except (
exceptions.ExternalRequestError,
exceptions.TimeOutError,
):
log.debug(
f"Disco query for chat states support failed for {peer_jid.full()}, "
"falling back to implicit discovery"
)
has_feature = False
data.disco_cache[peer_jid] = has_feature
return has_feature
async def _ensure_disco_check(self, client: SatXMPPEntity, to_jid: JID) -> None:
"""Check disco support for chat states if not already done.
Skipped for MUC groupchat.
For 1:1: passes bare JID, checks all known resources.
For MUC DM: passes full JID, checks occupant.
@param client: the client instance
@param to_jid: JID of the contact
"""
if self._is_muc(to_jid, client.profile) and not to_jid.resource:
# This is a MUC groupchat, we skip
return
data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
if to_jid not in data.disco_cache:
await self._check_disco_support(client, to_jid)
def update_cache(self, entity_jid: JID, value: bool | str, profile: str) -> None:
"""Update the entity data of the given profile for one or all contacts.
Reset the chat state(s) display if the notification has been disabled.
@param entity_jid: contact's JID, or C.ENTITY_ALL to update all contacts.
@param entity_jid: contact's JID
@param value: True, False or DELETE_VALUE to delete the entity data
@param profile: current profile
"""
@ -150,108 +248,164 @@ class XEP_0085(object):
else:
self.host.memory.update_entity_data(client, entity_jid, ENTITY_KEY, value)
if not value or value == DELETE_VALUE:
# reinit chat state UI for this or these contact(s)
# reinit chat state UI for this contact
self.host.bridge.chat_state_received(entity_jid.full(), "", profile)
def param_update_trigger(self, name, value, category, type_, profile):
"""Reset all the existing chat state entity data associated with this profile after a parameter modification.
def param_update_trigger(
self, name: str, value: str, category: str, type_: str, profile: str
) -> bool:
"""Update Chat State data according to parameter.
@param name: parameter name
@param value: "true" to activate the notifications, or any other value to delete it
@param category: parameter category
@param type_: parameter type
@param name: Parameter name.
@param value: "true" to activate the notifications, or any other value to delete
it.
@param category: Parameter category.
@param type_: Parameter type.
"""
if (category, name) == (PARAM_KEY, PARAM_NAME):
self.update_cache(
C.ENTITY_ALL, True if C.bool(value) else DELETE_VALUE, profile=profile
)
client = self.host.get_client(profile)
if C.bool(value):
self.host.memory.update_entity_data(
client, C.ENTITY_ALL, ENTITY_KEY, True
)
else:
self.host.memory.del_entity_datum(client, C.ENTITY_ALL, ENTITY_KEY)
return False
return True
def message_received_trigger(self, client, message, post_treat):
"""
Update the entity cache when we receive a message with body.
Check for a chat state in the message and signal frontends.
async def message_received_trigger(
self, client: SatXMPPEntity, message: domish.Element, post_treat: bool
) -> bool:
"""Handle an incoming message.
Update entity cache, init state machine, and signal frontends about
received chat states.
"""
profile = client.profile
if not self.host.memory.param_get_a(PARAM_NAME, PARAM_KEY, profile_key=profile):
return True
from_jid = JID(message.getAttribute("from"))
if self._is_muc(from_jid, profile):
msg_type = message.getAttribute("type")
# MUC: for groupchat we remove resource to get room bare JID.
# for DM (type="chat") we keep full JID to track occupant.
is_groupchat = self._is_muc(from_jid, profile) and msg_type == "groupchat"
if is_groupchat:
from_jid = from_jid.userhostJID()
else: # update entity data for one2one chat
# assert from_jid.resource # FIXME: assert doesn't work on normal message from server (e.g. server announce), because there is no resource
try:
next(domish.generateElementsNamed(message.elements(), name="body"))
try:
next(domish.generateElementsNamed(message.elements(), name="active"))
# contact enabled Chat State Notifications
self.update_cache(from_jid, True, profile=profile)
except StopIteration:
if message.getAttribute("type") == "chat":
# contact didn't enable Chat State Notifications
self.update_cache(from_jid, False, profile=profile)
return True
except StopIteration:
pass
# send our next "composing" states to any MUC and to the contacts who enabled the feature
self._chat_state_init(from_jid, message.getAttribute("type"), profile)
# Extract any chat state element from the message (for all message types)
chat_state = next(
(
child.name
for child in message.elements()
if child.name in CHAT_STATES and child.defaultUri == NS_CHAT_STATES
),
None,
)
state_list = [
child.name
for child in message.elements()
if message.getAttribute("type") in MESSAGE_TYPES
and child.name in CHAT_STATES
and child.defaultUri == NS_CHAT_STATES
]
for state in state_list:
# there must be only one state according to the XEP
if state != "gone" or message.getAttribute("type") != "groupchat":
self.host.bridge.chat_state_received(
message.getAttribute("from"), state, profile
if not is_groupchat:
# one2one or MUC DM: implicit discovery (XEP-0085 §5.1)
# check if this is a content message (has <body/>)
has_body = (
next(
domish.generateElementsNamed(message.elements(), name="body"),
None,
)
is not None
)
if chat_state is not None:
# XEP-0085 §5.1 Rule 3: any chat state proves support
self.update_cache(from_jid, True, profile=profile)
elif has_body and message.getAttribute("type") == "chat":
# XEP-0085 §5.1 Rule 2: content message without chat state
# only deactivate if not already activated
try:
current = self.host.memory.get_entity_datum(
client, from_jid, ENTITY_KEY
)
except (exceptions.UnknownEntityError, KeyError):
current = None
if not current:
self.update_cache(from_jid, False, profile=profile)
return True
# fire explicit disco check for this contact
await self._ensure_disco_check(client, from_jid)
# init state machine so composing states can be sent to this contact
# XEP-0085 §5.4: only use with chat/groupchat message types
if msg_type in MESSAGE_TYPES:
self._chat_state_init(client, from_jid, msg_type)
# Signal frontends about received chat states
if chat_state is not None:
# XEP-0085 §5.5: ignore <gone/> in groupchat
if chat_state != "gone" or message.getAttribute("type") != "groupchat":
self.host.bridge.chat_state_received(
message.getAttribute("from"), chat_state, profile
)
break
return True
def send_message_trigger(
self, client, mess_data, pre_xml_treatments, post_xml_treatments
):
"""
Eventually add the chat state to the message and initiate
the state machine when sending an "active" state.
"""
self,
client: SatXMPPEntity,
mess_data: dict,
pre_xml_treatments: "defer.Deferred",
post_xml_treatments: "defer.Deferred",
) -> bool:
"""Add chat state to outgoing messages and initiate the state machine."""
profile = client.profile
def treatment(mess_data):
async def treatment(mess_data):
message = mess_data["xml"]
to_jid = JID(message.getAttribute("to"))
if not self._check_activation(to_jid, forceEntityData=True, profile=profile):
# XEP-0085 §4: fire disco check for this contact
await self._ensure_disco_check(client, to_jid)
if not await self._check_activation(client, to_jid, forceEntityData=True):
return mess_data
try:
# message with a body always mean active state
next(domish.generateElementsNamed(message.elements(), name="body"))
# XEP-0085 §2: check if this is a content message (has <body/>)
has_body = (
next(
domish.generateElementsNamed(message.elements(), name="body"),
None,
)
is not None
)
if has_body and mess_data["type"] in MESSAGE_TYPES:
# XEP-0085 §5.4: only use with chat/groupchat message types
# XEP-0085 §5.6: content message -> attach <active/>
message.addElement((NS_CHAT_STATES, "active"))
# launch the chat state machine (init the timer)
if self._is_muc(to_jid, profile):
to_jid = to_jid.userhostJID()
self._chat_state_active(to_jid, mess_data["type"], profile)
except StopIteration:
if "chat_state" in mess_data["extra"]:
state = mess_data["extra"].pop("chat_state")
assert state in CHAT_STATES
message.addElement((NS_CHAT_STATES, state))
# Launch the chat state machine (init the timer)
# MUC groupchat: strip resource to get room bare JID.
# MUC DM (type="chat"): keep full JID to track occupant.
target_jid = (
to_jid.userhostJID()
if self._is_muc(to_jid, profile) and mess_data["type"] == "groupchat"
else to_jid
)
self._chat_state_active(client, target_jid, mess_data["type"])
elif "chat_state" in mess_data["extra"]:
# Standalone notification
state = mess_data["extra"].pop("chat_state")
assert state in CHAT_STATES
message.addElement((NS_CHAT_STATES, state))
return mess_data
post_xml_treatments.addCallback(treatment)
post_xml_treatments.addCallback(
lambda _: defer.ensureDeferred(treatment(mess_data))
)
return True
def _is_muc(self, to_jid, profile):
def _is_muc(self, to_jid: JID, profile: str) -> bool:
"""Tell if that JID is a MUC or not
@param to_jid (JID): full or bare JID to check
@param profile (str): %(doc_profile)s
@param to_jid: full or bare JID to check
@param profile: %(doc_profile)s
@return: bool
"""
client = self.host.get_client(profile)
@ -265,92 +419,114 @@ class XEP_0085(object):
pass
return False
def _check_activation(self, to_jid, forceEntityData, profile):
"""
async def _check_activation(
self, client: SatXMPPEntity, to_jid: JID, forceEntityData: bool
) -> bool:
"""Check whether chat state notifications should be sent to a contact.
Activation follows a priority order:
1. User config
2. MUC
3. Disco cache (preferred)
4. Implicit discovery cache (fallback)
@param client: the client instance
@param to_jid: the contact's full JID (or bare if you know it's a MUC)
@param forceEntityData: if set to True, a non-existing
entity data will be considered to be True (and initialized)
@param: current profile
@return: True if the notifications should be sent to this JID.
@param forceEntityData: if True, a missing implicit cache entry is
treated as True and initialized (used for initial discovery request)
@return: True if notifications should be sent to this JID.
"""
client = self.host.get_client(profile)
# check if the parameter is active
profile = client.profile
# XEP-0085 §5.2: user must be able to disable globally
if not self.host.memory.param_get_a(PARAM_NAME, PARAM_KEY, profile_key=profile):
return False
# check if notifications should be sent to this contact
if self._is_muc(to_jid, profile):
# XEP-0085 §5.5: MUC groupchat, no discovery required.
# MUC DMs (to_jid has resource) go through normal checks.
if self._is_muc(to_jid, profile) and not to_jid.resource:
return True
# FIXME: this assertion crash when we want to send a message to an online bare jid
# assert to_jid.resource or not self.host.memory.is_entity_available(to_jid, profile) # must either have a resource, or talk to an offline contact
# XEP-0085 §4: explicit disco discovery (preferred over implicit)
disco_result = await self._check_disco_support(client, to_jid)
if disco_result is True:
# disco confirmed support, activate unconditionally
self.update_cache(to_jid, True, profile=profile)
return True
if disco_result is False:
# disco confirmed no support, deactivate
return False
# disco_result is None: no resources known, fall through to implicit discovery
# XEP-0085 §5.1: implicit discovery (fallback)
try:
return self.host.memory.get_entity_datum(client, to_jid, ENTITY_KEY)
except (exceptions.UnknownEntityError, KeyError):
if forceEntityData:
# enable it for the first time
# XEP-0085 §5.1 Rule 1: enable for initial discovery request
self.update_cache(to_jid, True, profile=profile)
return True
# wait for the first message before sending states
return False
# no discovery completed yet, wait for first message
return False
def _chat_state_init(self, to_jid, mess_type, profile):
def _chat_state_init(
self, client: SatXMPPEntity, to_jid: JID, mess_type: str
) -> None:
"""Data initialization for the chat state machine.
@param client: the client instance
@param to_jid: full JID for one2one, bare JID for MUC
@param mess_type: "chat" or "groupchat"
"""
Data initialization for the chat state machine.
chat_states_data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
machines = chat_states_data.machines
if to_jid not in machines:
machines[to_jid] = ChatStateMachine(
self.host, to_jid, mess_type, client.profile
)
@param to_jid (JID): full JID for one2one, bare JID for MUC
@param mess_type (str): "one2one" or "groupchat"
@param profile (str): %(doc_profile)s
def _chat_state_active(
self, client: SatXMPPEntity, to_jid: JID, mess_type: str
) -> None:
"""Launch the chat state machine on "active" state.
@param client: the client instance
@param to_jid: full JID for one2one, bare JID for MUC
@param mess_type: "chat" or "groupchat"
"""
if mess_type is None:
return
profile_map = self.map.setdefault(profile, {})
if to_jid not in profile_map:
machine = ChatStateMachine(self.host, to_jid, mess_type, profile)
self.map[profile][to_jid] = machine
self._chat_state_init(client, to_jid, mess_type)
chat_states_data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
chat_states_data.machines[to_jid]._onEvent("active")
def _chat_state_active(self, to_jid, mess_type, profile_key):
"""
Launch the chat state machine on "active" state.
@param to_jid (JID): full JID for one2one, bare JID for MUC
@param mess_type (str): "one2one" or "groupchat"
@param profile (str): %(doc_profile)s
"""
profile = self.host.memory.get_profile_name(profile_key)
if profile is None:
raise exceptions.ProfileUnknownError
self._chat_state_init(to_jid, mess_type, profile)
self.map[profile][to_jid]._onEvent("active")
def chat_state_composing(self, to_jid_s, profile_key):
async def chat_state_composing(self, to_jid_s: str, profile_key: str) -> None:
"""Move to the "composing" state when required.
Since this method is called from the front-end, it needs to check the
values of the parameter "Send chat state notifications" and the entity
data associated to the target JID.
Called from the frontend. Checks user config, disco support,
and implicit discovery state before sending.
@param to_jid_s (str): contact full JID as a string
@param profile_key (str): %(doc_profile_key)s
@param to_jid_s: contact full JID as a string
@param profile_key: %(doc_profile_key)s
"""
# TODO: try to optimize this method which is called often
client = self.host.get_client(profile_key)
to_jid = JID(to_jid_s)
if self._is_muc(to_jid, client.profile):
to_jid = to_jid.userhostJID()
elif not to_jid.resource:
to_jid.resource = self.host.memory.main_resource_get(client, to_jid)
if not self._check_activation(
to_jid, forceEntityData=False, profile=client.profile
):
profile = client.profile
# XEP-0085 §4: fire disco check if not already done
await self._ensure_disco_check(client, to_jid)
if not await self._check_activation(client, to_jid, forceEntityData=False):
return
try:
self.map[client.profile][to_jid]._onEvent("composing")
except (KeyError, AttributeError):
# no message has been sent/received since the notifications
# have been enabled, it's better to wait for a first one
pass
# Ensure state machine exists (may not be initialised if no message
# has been exchanged yet but disco confirmed support)
# groupchat (bare JID) vs MUC DM (full JID with resource)
is_groupchat = self._is_muc(to_jid, profile) and not to_jid.resource
self._chat_state_init(client, to_jid, "groupchat" if is_groupchat else "chat")
chat_states_data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
chat_states_data.machines[to_jid]._onEvent("composing")
class ChatStateMachine(object):
class ChatStateMachine:
"""
This class represents a chat state, between one profile and
one target contact. A timer is used to move from one state
@ -360,7 +536,9 @@ class ChatStateMachine(object):
frontend). Other states are automatically set with the timer.
"""
def __init__(self, host, to_jid, mess_type, profile):
def __init__(
self, host: "LiberviaBackend", to_jid: JID, mess_type: str, profile: str
) -> None:
"""
Initialization need to store the target, message type
and a profile for sending later messages.
@ -370,9 +548,9 @@ class ChatStateMachine(object):
self.mess_type = mess_type
self.profile = profile
self.state = None
self.timer = None
self.timer: DelayedCall | None = None
def _onEvent(self, state):
def _onEvent(self, state: str) -> None:
"""
Move to the specified state, eventually send the
notification to the contact (the "active" state is
@ -400,18 +578,21 @@ class ChatStateMachine(object):
"subject": {},
"extra": {},
}
client.generate_message_xml(mess_data)
mess_data["xml"].addElement((NS_CHAT_STATES, state))
client.send(mess_data["xml"])
client.generate_message_xml(cast(Any, mess_data))
message_elt = cast(domish.Element, mess_data["xml"])
message_elt.addElement((NS_CHAT_STATES, state))
client.send(message_elt)
self.state = state
try:
self.timer.cancel()
except (internet_error.AlreadyCalled, AttributeError):
pass
if self.timer is not None:
try:
self.timer.cancel()
except internet_error.AlreadyCalled:
pass
if transition["next_state"] and transition["delay"] > 0:
self.timer = reactor.callLater(
reactor_time = cast(IReactorTime, reactor)
self.timer = reactor_time.callLater(
transition["delay"], self._onEvent, transition["next_state"]
)
@ -419,13 +600,15 @@ class ChatStateMachine(object):
@implementer(itx_xmpp.IDisco)
class XEP_0085_handler(XMPPHandler):
def __init__(self, plugin_parent, profile):
def __init__(self, plugin_parent: "XEP_0085", profile: str) -> None:
self.plugin_parent = plugin_parent
self.host = plugin_parent.host
self.profile = profile
def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
def getDiscoInfo(self, requestor: JID, target: JID, nodeIdentifier: str = "") -> list:
return [disco.DiscoFeature(NS_CHAT_STATES)]
def getDiscoItems(self, requestor, target, nodeIdentifier=""):
def getDiscoItems(
self, requestor: JID, target: JID, nodeIdentifier: str = ""
) -> list:
return []