268 lines
9.3 KiB
Python
268 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# Libervia plugin for XEP-0084
|
|
# Copyright (C) 2009-2022 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/>.
|
|
|
|
from typing import Optional, Dict, Any
|
|
from pathlib import Path
|
|
from base64 import b64decode, b64encode
|
|
|
|
from twisted.internet import defer
|
|
from twisted.words.protocols.jabber.xmlstream import XMPPHandler
|
|
from twisted.words.protocols.jabber import jid, error
|
|
from twisted.words.xish import domish
|
|
from zope.interface import implementer
|
|
from wokkel import disco, iwokkel, pubsub
|
|
|
|
from libervia.backend.core.constants import Const as C
|
|
from libervia.backend.core.i18n import _
|
|
from libervia.backend.core.log import getLogger
|
|
from libervia.backend.core.core_types import SatXMPPEntity
|
|
from libervia.backend.core import exceptions
|
|
|
|
|
|
log = getLogger(__name__)
|
|
|
|
IMPORT_NAME = "XEP-0084"
|
|
|
|
PLUGIN_INFO = {
|
|
C.PI_NAME: "User Avatar",
|
|
C.PI_IMPORT_NAME: IMPORT_NAME,
|
|
C.PI_TYPE: C.PLUG_TYPE_XEP,
|
|
C.PI_MODES: C.PLUG_MODE_BOTH,
|
|
C.PI_PROTOCOLS: ["XEP-0084"],
|
|
C.PI_DEPENDENCIES: ["IDENTITY", "XEP-0060", "XEP-0163"],
|
|
C.PI_MAIN: "XEP_0084",
|
|
C.PI_HANDLER: "yes",
|
|
C.PI_DESCRIPTION: _("""XEP-0084 (User Avatar) implementation"""),
|
|
}
|
|
|
|
NS_AVATAR = "urn:xmpp:avatar"
|
|
NS_AVATAR_METADATA = f"{NS_AVATAR}:metadata"
|
|
NS_AVATAR_DATA = f"{NS_AVATAR}:data"
|
|
|
|
|
|
class XEP_0084:
|
|
namespace_metadata = NS_AVATAR_METADATA
|
|
namespace_data = NS_AVATAR_DATA
|
|
|
|
def __init__(self, host):
|
|
log.info(_("XEP-0084 (User Avatar) plugin initialization"))
|
|
host.register_namespace("avatar_metadata", NS_AVATAR_METADATA)
|
|
host.register_namespace("avatar_data", NS_AVATAR_DATA)
|
|
self.host = host
|
|
self._p = host.plugins["XEP-0060"]
|
|
self._i = host.plugins['IDENTITY']
|
|
self._i.register(
|
|
IMPORT_NAME,
|
|
"avatar",
|
|
self.get_avatar,
|
|
self.set_avatar,
|
|
priority=2000
|
|
)
|
|
host.plugins["XEP-0163"].add_pep_event(
|
|
None, NS_AVATAR_METADATA, self._on_metadata_update
|
|
)
|
|
|
|
def get_handler(self, client):
|
|
return XEP_0084_Handler()
|
|
|
|
def _on_metadata_update(self, itemsEvent, profile):
|
|
client = self.host.get_client(profile)
|
|
defer.ensureDeferred(self.on_metadata_update(client, itemsEvent))
|
|
|
|
async def on_metadata_update(
|
|
self,
|
|
client: SatXMPPEntity,
|
|
itemsEvent: pubsub.ItemsEvent
|
|
) -> None:
|
|
entity = client.jid.userhostJID()
|
|
avatar_metadata = await self.get_avatar(client, entity)
|
|
await self._i.update(client, IMPORT_NAME, "avatar", avatar_metadata, entity)
|
|
|
|
async def get_avatar(
|
|
self,
|
|
client: SatXMPPEntity,
|
|
entity_jid: jid.JID
|
|
) -> Optional[dict]:
|
|
"""Get avatar data
|
|
|
|
@param entity: entity to get avatar from
|
|
@return: avatar metadata, or None if no avatar has been found
|
|
"""
|
|
service = entity_jid.userhostJID()
|
|
# metadata
|
|
try:
|
|
items, __ = await self._p.get_items(
|
|
client,
|
|
service,
|
|
NS_AVATAR_METADATA,
|
|
max_items=1
|
|
)
|
|
except exceptions.NotFound:
|
|
return None
|
|
|
|
if not items:
|
|
return None
|
|
|
|
item_elt = items[0]
|
|
try:
|
|
metadata_elt = next(item_elt.elements(NS_AVATAR_METADATA, "metadata"))
|
|
except StopIteration:
|
|
log.warning(f"missing metadata element: {item_elt.toXml()}")
|
|
return None
|
|
|
|
for info_elt in metadata_elt.elements(NS_AVATAR_METADATA, "info"):
|
|
try:
|
|
metadata = {
|
|
"id": str(info_elt["id"]),
|
|
"size": int(info_elt["bytes"]),
|
|
"media_type": str(info_elt["type"])
|
|
}
|
|
avatar_id = metadata["id"]
|
|
if not avatar_id:
|
|
raise ValueError
|
|
except (KeyError, ValueError):
|
|
log.warning(f"invalid <info> element: {item_elt.toXml()}")
|
|
return None
|
|
# FIXME: to simplify, we only handle image/png for now
|
|
if metadata["media_type"] == "image/png":
|
|
break
|
|
else:
|
|
# mandatory image/png is missing, or avatar is disabled
|
|
# (https://xmpp.org/extensions/xep-0084.html#pub-disable)
|
|
return None
|
|
|
|
cache_data = self.host.common_cache.get_metadata(avatar_id)
|
|
if not cache_data:
|
|
try:
|
|
data_items, __ = await self._p.get_items(
|
|
client,
|
|
service,
|
|
NS_AVATAR_DATA,
|
|
item_ids=[avatar_id]
|
|
)
|
|
data_item_elt = data_items[0]
|
|
except (error.StanzaError, IndexError) as e:
|
|
log.warning(
|
|
f"Can't retrieve avatar of {service.full()} with ID {avatar_id!r}: "
|
|
f"{e}"
|
|
)
|
|
return None
|
|
try:
|
|
avatar_buf = b64decode(
|
|
str(next(data_item_elt.elements(NS_AVATAR_DATA, "data")))
|
|
)
|
|
except Exception as e:
|
|
log.warning(
|
|
f"invalid data element for {service.full()} with avatar ID "
|
|
f"{avatar_id!r}: {e}\n{data_item_elt.toXml()}"
|
|
)
|
|
return None
|
|
with self.host.common_cache.cache_data(
|
|
IMPORT_NAME,
|
|
avatar_id,
|
|
metadata["media_type"]
|
|
) as f:
|
|
f.write(avatar_buf)
|
|
cache_data = {
|
|
"path": Path(f.name),
|
|
"mime_type": metadata["media_type"]
|
|
}
|
|
|
|
return self._i.avatar_build_metadata(
|
|
cache_data['path'], cache_data['mime_type'], avatar_id
|
|
)
|
|
|
|
def build_item_data_elt(self, avatar_data: Dict[str, Any]) -> domish.Element:
|
|
"""Generate the item for the data node
|
|
|
|
@param avatar_data: data as build by identity plugin (need to be filled with
|
|
"cache_uid" and "base64" keys)
|
|
"""
|
|
data_elt = domish.Element((NS_AVATAR_DATA, "data"))
|
|
data_elt.addContent(avatar_data["base64"])
|
|
return pubsub.Item(id=avatar_data["cache_uid"], payload=data_elt)
|
|
|
|
def build_item_metadata_elt(self, avatar_data: Dict[str, Any]) -> domish.Element:
|
|
"""Generate the item for the metadata node
|
|
|
|
@param avatar_data: data as build by identity plugin (need to be filled with
|
|
"cache_uid", "path", and "media_type" keys)
|
|
"""
|
|
metadata_elt = domish.Element((NS_AVATAR_METADATA, "metadata"))
|
|
info_elt = metadata_elt.addElement("info")
|
|
# FIXME: we only fill required elements for now (see
|
|
# https://xmpp.org/extensions/xep-0084.html#table-1)
|
|
info_elt["id"] = avatar_data["cache_uid"]
|
|
info_elt["type"] = avatar_data["media_type"]
|
|
info_elt["bytes"] = str(avatar_data["path"].stat().st_size)
|
|
return pubsub.Item(id=self._p.ID_SINGLETON, payload=metadata_elt)
|
|
|
|
async def set_avatar(
|
|
self,
|
|
client: SatXMPPEntity,
|
|
avatar_data: Dict[str, Any],
|
|
entity: jid.JID
|
|
) -> None:
|
|
"""Set avatar of the profile
|
|
|
|
@param avatar_data(dict): data of the image to use as avatar, as built by
|
|
IDENTITY plugin.
|
|
@param entity(jid.JID): entity whose avatar must be changed
|
|
"""
|
|
service = entity.userhostJID()
|
|
|
|
# Data
|
|
await self._p.create_if_new_node(
|
|
client,
|
|
service,
|
|
NS_AVATAR_DATA,
|
|
options={
|
|
self._p.OPT_ACCESS_MODEL: self._p.ACCESS_OPEN,
|
|
self._p.OPT_PERSIST_ITEMS: 1,
|
|
self._p.OPT_MAX_ITEMS: 1,
|
|
}
|
|
)
|
|
item_data_elt = self.build_item_data_elt(avatar_data)
|
|
await self._p.send_items(client, service, NS_AVATAR_DATA, [item_data_elt])
|
|
|
|
# Metadata
|
|
await self._p.create_if_new_node(
|
|
client,
|
|
service,
|
|
NS_AVATAR_METADATA,
|
|
options={
|
|
self._p.OPT_ACCESS_MODEL: self._p.ACCESS_OPEN,
|
|
self._p.OPT_PERSIST_ITEMS: 1,
|
|
self._p.OPT_MAX_ITEMS: 1,
|
|
}
|
|
)
|
|
item_metadata_elt = self.build_item_metadata_elt(avatar_data)
|
|
await self._p.send_items(client, service, NS_AVATAR_METADATA, [item_metadata_elt])
|
|
|
|
|
|
@implementer(iwokkel.IDisco)
|
|
class XEP_0084_Handler(XMPPHandler):
|
|
|
|
def getDiscoInfo(self, requestor, service, nodeIdentifier=""):
|
|
return [
|
|
disco.DiscoFeature(NS_AVATAR_METADATA),
|
|
disco.DiscoFeature(NS_AVATAR_DATA)
|
|
]
|
|
|
|
def getDiscoItems(self, requestor, service, nodeIdentifier=""):
|
|
return []
|