diff --git a/libervia/backend/plugins/plugin_misc_xid/__init__.py b/libervia/backend/plugins/plugin_misc_xid/__init__.py
new file mode 100644
index 00000000..ac7edb6a
--- /dev/null
+++ b/libervia/backend/plugins/plugin_misc_xid/__init__.py
@@ -0,0 +1,919 @@
+#!/usr/bin/env python3
+
+# Libervia plugin for XMPP Decentralized ID (XID)
+# 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/>.
+
+import os
+from pathlib import Path
+import time
+from typing import TYPE_CHECKING, cast
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey,
+    Ed25519PublicKey,
+)
+from twisted.internet import defer, reactor
+from twisted.internet.interfaces import IReactorCore
+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, pubsub
+from zope.interface import implementer
+
+from libervia.backend import G
+from libervia.backend.core import exceptions
+from libervia.backend.core.constants import Const as C
+from libervia.backend.core.core_types import SatXMPPClient, SatXMPPEntity
+from libervia.backend.core.i18n import _
+from libervia.backend.core.log import getLogger
+from libervia.backend.memory import persistent
+from libervia.backend.plugins.plugin_xep_0060 import XEP_0060
+from libervia.backend.tools.common.uri import build_xmpp_uri, parse_xmpp_uri
+from libervia.backend.tools.utils import xmpp_date
+
+from .models import Challenge, ListXID, Response, XID, XIDClientData, XIDStorage
+
+if TYPE_CHECKING:
+    from libervia.backend.core.main import LiberviaBackend
+
+log = getLogger(__name__)
+
+IMPORT_NAME = "XID"
+
+PLUGIN_INFO = {
+    C.PI_NAME: "XMPP Decentralized ID",
+    C.PI_IMPORT_NAME: IMPORT_NAME,
+    C.PI_TYPE: C.PLUG_TYPE_MISC,
+    C.PI_MODES: C.PLUG_MODE_BOTH,
+    C.PI_PROTOCOLS: [],
+    C.PI_DEPENDENCIES: ["XEP-0060", "XEP-0163"],
+    C.PI_RECOMMENDATIONS: ["XEP-0384"],
+    C.PI_MAIN: "XIDPlugin",
+    C.PI_HANDLER: "yes",
+    C.PI_DESCRIPTION: _("""DNS-independent XMPP entity identifier."""),
+}
+
+NS_XID = "urn:xmpp:xid:0"
+PEP_NODE = "urn:xmpp:xid"
+PEP_NODE_REVOKED = "urn:xmpp:xid:revoked"
+XID_DOMAIN = "id.internal"
+
+ED25519_PREFIX = "00"
+CHALLENGE_NONCE_BYTES = 16
+
+
+class XIDPlugin:
+    """Main plugin class for XID support."""
+
+    # TODO: implement private-key automatic synchronisation.
+
+    def get_handler(self, client: SatXMPPEntity) -> "XIDHandler":
+        return XIDHandler()
+
+    def __init__(self, host: "LiberviaBackend") -> None:
+        log.info(_("Plugin XID initialization"))
+        self._ps = cast(XEP_0060, host.plugins["XEP-0060"])
+
+        host.bridge.add_method(
+            "xid_create",
+            ".plugin",
+            in_sign="ss",
+            out_sign="s",
+            method=self._create,
+            async_=True,
+        )
+        host.bridge.add_method(
+            "xid_get",
+            ".plugin",
+            in_sign="sbs",
+            out_sign="s",
+            method=self._get,
+            async_=True,
+        )
+        host.bridge.add_method(
+            "xid_revoke",
+            ".plugin",
+            in_sign="sss",
+            out_sign="",
+            method=self._revoke,
+            async_=True,
+        )
+        host.bridge.add_method(
+            "xid_verify",
+            ".plugin",
+            in_sign="sss",
+            out_sign="s",
+            method=self._verify,
+            async_=True,
+        )
+        host.bridge.add_method(
+            "xid_signature_sign_file",
+            ".plugin",
+            in_sign="sss",
+            out_sign="s",
+            method=self._sign_file,
+            async_=True,
+        )
+        host.bridge.add_method(
+            "xid_signature_sign_data",
+            ".plugin",
+            in_sign="sss",
+            out_sign="s",
+            method=self._sign_data,
+            async_=True,
+        )
+        host.bridge.add_method(
+            "xid_signature_check_file",
+            ".plugin",
+            in_sign="ssss",
+            out_sign="b",
+            method=self._signature_check_file,
+            async_=True,
+        )
+        host.bridge.add_method(
+            "xid_signature_check_data",
+            ".plugin",
+            in_sign="sss",
+            out_sign="b",
+            method=self._signature_check_data,
+            async_=True,
+        )
+        host.bridge.add_method(
+            "xid_export",
+            ".plugin",
+            in_sign="ss",
+            out_sign="s",
+            method=self._export,
+            async_=True,
+        )
+        host.bridge.add_method(
+            "xid_import",
+            ".plugin",
+            in_sign="ss",
+            out_sign="",
+            method=self._import,
+            async_=True,
+        )
+        host.register_namespace("xid", NS_XID)
+        host.trigger.add("message_received", self._message_received_trigger)
+
+    async def profile_connecting(self, client: SatXMPPEntity) -> None:
+        """Initialize XID storage before client is ready."""
+        if client.is_component:
+            return
+        storage = persistent.PersistentBinaryDict(IMPORT_NAME, client.profile)
+        await storage.load()
+
+        xids = {}
+        for xid_s, value in storage.items():
+            xids[xid_s] = XIDStorage.model_validate(value)
+
+        xid_data = XIDClientData(_storage=storage, xids=xids)
+        client.set_plugin_data(C.PI_IMPORT_NAME, xid_data)
+
+    async def profile_connected(self, client: SatXMPPEntity) -> None:
+        """Ensure a current XID exists."""
+        if client.is_component:
+            return
+
+        try:
+            items, __ = await self._ps.get_items(
+                client, client.jid.userhostJID(), PEP_NODE, item_ids=["current"]
+            )
+        except exceptions.NotFound:
+            items = None
+
+        if items:
+            try:
+                xid_model = XID.from_element(items[0], item_id="current")
+            except exceptions.NotFound:
+                xid_model = None
+            if xid_model:
+                xid_data = client.get_plugin_data(C.PI_IMPORT_NAME, XIDClientData)
+                xid_storage = xid_data.xids.get(str(xid_model.xid))
+                if xid_storage is not None and xid_storage.private_key is not None:
+                    log.debug(f"XID {xid_model.xid} found with private key.")
+                    return
+                log.warning(
+                    f"XID {xid_model.xid} published but private key not found in storage."
+                )
+                # TODO: Auto sync private key.
+                return
+
+        log.info(_("No current XID found, creating one…"))
+        try:
+            await self.create(client)
+        except Exception:
+            log.exception("Failed to create XID")
+        else:
+            log.info("XID created and published successfully.")
+
+    @staticmethod
+    def get_local_part(public_key: bytes) -> str:
+        """Convert Ed25519 public key to a XID localpart.
+
+        @param public_key: Ed25519 public key.
+        @return: XID local part.
+        """
+        return ED25519_PREFIX + public_key.hex()
+
+    @staticmethod
+    def xid_to_public_key(xid: jid.JID) -> bytes:
+        """Extract public key bytes from a XID.
+
+        @param xid: XID to extract the public key from.
+        @return: 32 bytes of public key.
+        @raise ValueError: if XID is malformed.
+        """
+        localpart = xid.user
+        if localpart is None:
+            raise ValueError("XID has no localpart")
+        if not localpart.startswith(ED25519_PREFIX):
+            raise ValueError(
+                f"XID localpart {localpart!r} does not start with Ed25519 prefix "
+                f"({ED25519_PREFIX!r})"
+            )
+        key_hex = localpart[2:]
+        if len(key_hex) != 64:
+            raise ValueError(f"XID public key hex has wrong length: {len(key_hex)}")
+        return bytes.fromhex(key_hex)
+
+    @staticmethod
+    def private_key_to_public_key(private_key: Ed25519PrivateKey) -> bytes:
+        """Derive public key bytes from a private key."""
+        public_key = private_key.public_key()
+        return public_key.public_bytes(
+            serialization.Encoding.Raw, serialization.PublicFormat.Raw
+        )
+
+    async def _get_private_key(
+        self, client: SatXMPPEntity, xid: jid.JID
+    ) -> Ed25519PrivateKey | None:
+        """Retrieve the private key for a XID from storage.
+
+        @return: Private key or None if not found.
+        """
+        xid_data = client.get_plugin_data(C.PI_IMPORT_NAME, XIDClientData)
+        xid_s = str(xid)
+        xid_storage = xid_data.xids.get(xid_s)
+        if xid_storage is None or xid_storage.private_key is None:
+            return None
+        private_key_hex = G.host.memory.decrypt_value(client, xid_storage.private_key)
+        return Ed25519PrivateKey.from_private_bytes(bytes.fromhex(private_key_hex))
+
+    async def _store_private_key(
+        self, client: SatXMPPEntity, xid: jid.JID, private_key: Ed25519PrivateKey
+    ) -> None:
+        """Store a private key associated with a XID.
+
+        @param client: client instance.
+        @param xid: XID.
+        @param private_key: Ed25519 private key.
+        """
+        xid_data = client.get_plugin_data(C.PI_IMPORT_NAME, XIDClientData)
+        private_key_hex = private_key.private_bytes(
+            serialization.Encoding.Raw,
+            serialization.PrivateFormat.Raw,
+            serialization.NoEncryption(),
+        ).hex()
+        encrypted = G.host.memory.encrypt_value(client, private_key_hex)
+        xid_s = str(xid)
+        xid_storage = XIDStorage(private_key=encrypted)
+        xid_data.xids[xid_s] = xid_storage
+        await xid_data._storage.aset(xid_s, xid_storage.model_dump())
+
+    def _create(self, profile: str, item_id: str) -> defer.Deferred:
+        client = G.host.get_client(profile)
+        d = defer.ensureDeferred(self.create(client, item_id))
+        d.addCallback(lambda result: result.model_dump_json())
+        return d
+
+    async def create(self, client: SatXMPPEntity, item_id: str = "current") -> XID:
+        """Generate a new XID key pair and publish it.
+
+        @param client: client instance.
+        @param item_id: PEP item ID ("current" for main XID).
+        @return: XID model.
+        @raise exceptions.ConflictError: if item_id is "current" and already published.
+        """
+        if item_id == "current":
+            try:
+                items, __ = await self._ps.get_items(
+                    client, client.jid.userhostJID(), PEP_NODE, item_ids=["current"]
+                )
+            except exceptions.NotFound:
+                items = None
+            if items:
+                raise exceptions.ConflictError(_("A current XID is already published"))
+
+        private_key = Ed25519PrivateKey.generate()
+        public_key = self.private_key_to_public_key(private_key)
+        localpart = self.get_local_part(public_key)
+        xid = jid.JID(f"{localpart}@{XID_DOMAIN}")
+        created = time.time()
+
+        await self._store_private_key(client, xid, private_key)
+
+        xid_model = XID(xid=xid, created=created, has_private_key=True, item_id=item_id)
+        item = pubsub.Item(id=item_id, payload=xid_model.to_element())
+        await self._ps.publish(
+            client,
+            client.jid.userhostJID(),
+            PEP_NODE,
+            [item],
+            extra={
+                self._ps.EXTRA_PUBLISH_OPTIONS: {
+                    "pubsub#access_model": "presence",
+                },
+                self._ps.EXTRA_AUTOCREATE: True,
+            },
+        )
+
+        return xid_model
+
+    def _revoke(self, xid: str, reason: str, profile: str) -> defer.Deferred:
+        client = G.host.get_client(profile)
+        try:
+            xid_jid = jid.JID(xid)
+        except Exception:
+            raise exceptions.DataError(f"Invalid XID: {xid}")
+        return defer.ensureDeferred(self.revoke(client, xid_jid, reason or None))
+
+    async def revoke(
+        self, client: SatXMPPEntity, xid: jid.JID, reason: str | None
+    ) -> None:
+        """Revoke a XID.
+
+        @param client: client instance.
+        @param xid: XID to revoke.
+        @param reason: optional human-readable reason for revocation.
+        """
+        if xid.host != XID_DOMAIN:
+            raise exceptions.DataError(
+                f'XID domain mismatch: expected "{XID_DOMAIN}", got "{xid.host}"'
+            )
+        items, __ = await self._ps.get_items(client, client.jid.userhostJID(), PEP_NODE)
+        created = None
+        for candidate in items:
+            try:
+                xid_in_item = XID.from_element(candidate)
+            except exceptions.NotFound:
+                continue
+            if xid_in_item.xid == xid:
+                item = candidate
+                created = xid_in_item.created
+                break
+        else:
+            raise exceptions.NotFound(f"XID {xid} not found in published XIDs")
+
+        item_id = item["id"]
+
+        await self._ps.retract_items(
+            client, client.jid.userhostJID(), PEP_NODE, [item_id]
+        )
+        revoked_elt = domish.Element((NS_XID, "revoked"))
+        revoked_elt["created"] = xmpp_date(created)
+        revoked_elt["revoked"] = xmpp_date()
+        revoked_elt.addContent(str(xid))
+        if reason:
+            reason_elt = revoked_elt.addElement((NS_XID, "reason"))
+            reason_elt.addContent(reason)
+        revoked_item = pubsub.Item(payload=revoked_elt)
+        await self._ps.publish(
+            client,
+            client.jid.userhostJID(),
+            PEP_NODE_REVOKED,
+            [revoked_item],
+            extra={
+                self._ps.EXTRA_PUBLISH_OPTIONS: {
+                    "pubsub#access_model": "presence",
+                },
+                self._ps.EXTRA_AUTOCREATE: True,
+            },
+        )
+
+    def _get(self, entity: str, all: bool, profile: str) -> defer.Deferred:
+        client = G.host.get_client(profile)
+        d = defer.ensureDeferred(self.get(client, entity or None, all))
+        d.addCallback(lambda result: ListXID(result).model_dump_json())
+        return d
+
+    async def get(
+        self, client: SatXMPPEntity, entity: str | None, all: bool
+    ) -> list[XID]:
+        """Get XIDs published by an entity.
+
+        @param client: client instance.
+        @param entity: bare JID of the entity.
+            None to use profile own JID.
+        @param all: If True, return all XIDs; otherwise only current.
+        @return: List of XID entries.
+        """
+        if entity:
+            target = jid.JID(entity).userhostJID()
+        else:
+            target = client.jid.userhostJID()
+
+        try:
+            items, __ = await self._ps.get_items(
+                client, target, PEP_NODE, item_ids=None if all else ["current"]
+            )
+        except exceptions.NotFound:
+            return []
+
+        if not items:
+            return []
+
+        result = []
+        for item_elt in items:
+            try:
+                xid_model = XID.from_element(item_elt, item_id=item_elt["id"] or None)
+            except exceptions.NotFound:
+                continue
+            has_private_key = (
+                await self._get_private_key(client, xid_model.xid) is not None
+            )
+            result.append(
+                XID(
+                    xid=xid_model.xid,
+                    created=xid_model.created,
+                    has_private_key=has_private_key,
+                    item_id=xid_model.item_id,
+                )
+            )
+
+        return result
+
+    def _verify(self, entity_jid: str, xid: str, profile: str) -> defer.Deferred:
+        client = G.host.get_client(profile)
+        d = defer.ensureDeferred(self.verify(client, entity_jid, jid.JID(xid)))
+        d.addCallback(lambda result: "true" if result else "false")
+        return d
+
+    async def verify(self, client: SatXMPPEntity, entity_jid: str, xid: jid.JID) -> bool:
+        """Verify an entity's XID via challenge/response.
+
+        Sends a challenge to the entity as specified in the XEP, and waits for a response.
+
+        @param client: client instance.
+        @param entity_jid: bare JID of the entity to verify.
+        @param xid: XID to verify.
+        @return: True if verified successfully.
+        @raise exceptions.TimeOutError: if no response is received.
+        """
+        nonce = os.urandom(CHALLENGE_NONCE_BYTES).hex()
+        timestamp = time.time()
+        challenge = Challenge(xid=xid, timestamp=timestamp, nonce=nonce)
+
+        d = defer.Deferred()
+        xid_data = client.get_plugin_data(C.PI_IMPORT_NAME, XIDClientData)
+        xid_data.pending_challenges[nonce] = d
+
+        msg = domish.Element((None, "message"))
+        msg["to"] = entity_jid
+        msg["type"] = "chat"
+        msg.addChild(challenge.to_element())
+        client.send(msg)
+
+        d.addTimeout(30, cast(IReactorCore, reactor))
+        try:
+            return await d
+        except defer.TimeoutError:
+            raise exceptions.TimeOutError(
+                f"No response received for challenge to {entity_jid}"
+            )
+        finally:
+            xid_data.pending_challenges.pop(nonce, None)
+
+    def _message_received_trigger(
+        self,
+        client: SatXMPPClient,
+        message_elt: domish.Element,
+        post_treat: defer.Deferred,
+    ) -> bool:
+        """Trigger handler for incoming messages containing XID challenge/response.
+
+        @param client: client instance.
+        @param message_elt: the <message/> element.
+        @param post_treat: not used here
+        @return: False to stop propagation if a XID element was found, True otherwise.
+        """
+        for element in message_elt.elements(NS_XID):
+            if element.name == "challenge":
+                self._handle_challenge(client, message_elt, element)
+                return False
+            elif element.name == "response":
+                self._handle_response(client, message_elt, element)
+                return False
+        if "challenge" in message_elt.toXml():
+            from pudb.remote import set_trace
+
+            set_trace(term_size=(253, 69))
+        return True
+
+    def _handle_challenge(
+        self,
+        client: SatXMPPClient,
+        message_elt: domish.Element,
+        challenge_elt: domish.Element,
+    ) -> None:
+        """Handle incoming challenge from another entity.
+
+        Signs the nonce with the corresponding private key and sends a response.
+
+        @param client: client entity that received the challenge.
+        @param message_elt: the <message/> element.
+        @param challenge_elt: <challenge/> element.
+        """
+        challenge = Challenge.from_element(challenge_elt)
+        from_jid = jid.JID(message_elt["from"])
+
+        # Check if the requesting entity is our own server or someone from our roster with
+        # presence access
+        if from_jid.host != client.jid.host:
+            if not client.roster.is_subscribed_from(from_jid.userhostJID()):
+                log.warning(
+                    f"Ignoring challenge from {from_jid!r}, "
+                    "not our server and not subscribed to our presence"
+                )
+                return
+
+        log.debug(f"Received challenge for XID {challenge.xid} from {from_jid!r}")
+
+        private_key = self._get_private_key_sync(client, challenge.xid)
+        if private_key is None:
+            log.warning(f"No private key for XID {challenge.xid}, ignoring challenge")
+            return
+
+        signature = private_key.sign(challenge.nonce.encode())
+        signature_hex = signature.hex()
+
+        response = Response(
+            xid=challenge.xid,
+            timestamp=challenge.timestamp,
+            signature=signature_hex,
+        )
+
+        msg = domish.Element((None, "message"))
+        msg["to"] = message_elt["from"]
+        msg["type"] = "chat"
+        msg.addChild(response.to_element())
+        client.send(msg)
+        log.debug(f"Sent response to challenge for XID {challenge.xid}")
+
+    @staticmethod
+    def _get_private_key_sync(
+        client: SatXMPPClient, xid: jid.JID
+    ) -> Ed25519PrivateKey | None:
+        """Synchronously retrieve the private key for a XID from storage.
+
+        @return: Private key or None if not found.
+        """
+        xid_data = client.get_plugin_data(C.PI_IMPORT_NAME, XIDClientData)
+        xid_storage = xid_data.xids.get(str(xid))
+        if xid_storage is None or xid_storage.private_key is None:
+            return None
+        private_key_hex = G.host.memory.decrypt_value(client, xid_storage.private_key)
+        return Ed25519PrivateKey.from_private_bytes(bytes.fromhex(private_key_hex))
+
+    def _handle_response(
+        self,
+        client: SatXMPPClient,
+        message_elt: domish.Element,
+        response_elt: domish.Element,
+    ) -> None:
+        """Handle incoming response to a challenge we sent.
+
+        @param client: client entity that sent the challenge.
+        @param message_elt: the <message/> element.
+        @param response_elt: <response/> element.
+        """
+        response = Response.from_element(response_elt)
+        log.debug(f"Received response for XID {response.xid}")
+
+        xid_data = client.get_plugin_data(C.PI_IMPORT_NAME, XIDClientData)
+
+        for nonce, deferred in xid_data.pending_challenges.items():
+            if deferred.called:
+                continue
+            public_key_bytes = self.xid_to_public_key(response.xid)
+            public_key = Ed25519PublicKey.from_public_bytes(public_key_bytes)
+            try:
+                public_key.verify(bytes.fromhex(response.signature), nonce.encode())
+                deferred.callback(True)
+                return
+            except Exception:
+                continue
+
+        log.warning(f"No matching pending challenge for response to XID {response.xid}")
+
+    def _sign_file(self, xid: str, file_path: str, profile: str) -> defer.Deferred:
+        client = G.host.get_client(profile)
+        d = defer.ensureDeferred(
+            self.sign(client, jid.JID(xid) if xid else None, file_path=file_path)
+        )
+        return d
+
+    def _sign_data(self, xid: str, data: str, profile: str) -> defer.Deferred:
+        client = G.host.get_client(profile)
+        d = defer.ensureDeferred(
+            self.sign(client, jid.JID(xid) if xid else None, data=data)
+        )
+        return d
+
+    async def sign(
+        self,
+        client: SatXMPPEntity,
+        xid: jid.JID | None,
+        file_path: str | None = None,
+        data: str | None = None,
+    ) -> str:
+        """Sign a file or inline data.
+
+        @param client: client instance.
+        @param xid: XID whose private key to use (None for current).
+        @param file_path: path to file to sign.
+            mutually exclusive with ``data``.
+        @param data: inline data to sign.
+            mutually exclusive with ``file_path``.
+        @return: hex-encoded signature.
+        @raise exceptions.DataError: if both or neither file_path and data are given.
+        """
+        if (file_path is None) == (data is None):
+            raise exceptions.DataError(
+                '"file_path" and "data" are mutually exclusive, exactly one must be '
+                "provided"
+            )
+
+        if file_path is not None:
+            normalized = Path(file_path).expanduser().resolve()
+            text = normalized.read_text(errors="replace")
+        else:
+            assert data is not None
+            text = data
+
+        xid = xid or await self.get_current_xid(client)
+        private_key = await self._get_private_key(client, xid)
+        if private_key is None:
+            raise exceptions.NotFound(f"No private key found for XID {xid}")
+
+        signature = private_key.sign(text.encode())
+        return signature.hex()
+
+    async def get_current_xid(self, client: SatXMPPEntity) -> jid.JID:
+        """Get the current published XID.
+
+        @param client: client instance.
+        @return: XID as JID.
+        @raise exceptions.NotFound: if no current XID is published.
+        """
+        items, __ = await self._ps.get_items(
+            client,
+            client.jid.userhostJID(),
+            PEP_NODE,
+            item_ids=["current"],
+        )
+        if not items:
+            raise exceptions.NotFound("No current XID published")
+        try:
+            xid_item = XID.from_element(items[0])
+        except exceptions.NotFound:
+            raise exceptions.DataError("Invalid XID item payload")
+        return xid_item.xid
+
+    def _signature_check_file(
+        self, xid: str, file_path: str, signature_hex: str, profile: str
+    ) -> defer.Deferred:
+        return defer.ensureDeferred(
+            self.signature_check(
+                jid.JID(xid), file_path=file_path, signature_hex=signature_hex
+            )
+        )
+
+    def _signature_check_data(
+        self, xid: str, data: str, signature_hex: str
+    ) -> defer.Deferred:
+        return defer.ensureDeferred(
+            self.signature_check(jid.JID(xid), data=data, signature_hex=signature_hex)
+        )
+
+    async def signature_check(
+        self,
+        xid: jid.JID,
+        *,
+        file_path: str | None = None,
+        data: str | None = None,
+        signature_hex: str,
+    ) -> bool:
+        """Verify file or inline data signature.
+
+        @param xid: XID whose public key is used for verification.
+        @param file_path: path to file with original data.
+            mutually exclusive with ``data``.
+        @param data: inline original data.
+            mutually exclusive with ``file_path``.
+        @param signature_hex: hex-encoded signature.
+        @return: True if signature is valid.
+        """
+        if (file_path is None) == (data is None):
+            raise exceptions.DataError(
+                _(
+                    "file_path and data are mutually exclusive, exactly one must be "
+                    "provided"
+                )
+            )
+
+        if file_path is not None:
+            normalized = Path(file_path).expanduser().resolve()
+            text = normalized.read_text(errors="replace")
+        else:
+            assert data is not None
+            text = data
+
+        public_key_bytes = self.xid_to_public_key(xid)
+        from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+            Ed25519PublicKey,
+        )
+
+        public_key = Ed25519PublicKey.from_public_bytes(public_key_bytes)
+        signature = bytes.fromhex(signature_hex)
+
+        try:
+            public_key.verify(signature, text.encode())
+            return True
+        except Exception:
+            return False
+
+    def _export(self, xid: str, profile: str) -> defer.Deferred:
+        client = G.host.get_client(profile)
+        return defer.ensureDeferred(self.export(client, jid.JID(xid) if xid else None))
+
+    async def export(self, client: SatXMPPEntity, xid: jid.JID | None) -> str:
+        """Generate an xmpp: URI for private key export.
+
+        @param client: client instance.
+        @param xid: XID to export, None for current.
+        @return: xmpp: URI as specified in the XEP.
+        """
+        if xid is None:
+            items, __ = await self._ps.get_items(
+                client,
+                client.jid.userhostJID(),
+                PEP_NODE,
+                item_ids=["current"],
+            )
+            if not items:
+                raise exceptions.NotFound("No current XID published")
+            try:
+                xid_item = XID.from_element(items[0])
+            except exceptions.NotFound:
+                raise exceptions.DataError("Invalid XID item payload")
+        else:
+            items, __ = await self._ps.get_items(
+                client,
+                client.jid.userhostJID(),
+                PEP_NODE,
+            )
+            xid_item = None
+            for item in items:
+                try:
+                    candidate = XID.from_element(item)
+                except exceptions.NotFound:
+                    continue
+                if candidate.xid == xid:
+                    xid_item = candidate
+                    break
+            else:
+                raise exceptions.NotFound(f"No published item for XID {xid}")
+
+        xid = xid_item.xid
+        created = xmpp_date(xid_item.created)
+
+        private_key = await self._get_private_key(client, xid)
+        if private_key is None:
+            raise exceptions.NotFound(f"No private key found for XID {xid}")
+
+        private_key_hex = private_key.private_bytes(
+            serialization.Encoding.Raw,
+            serialization.PrivateFormat.Raw,
+            serialization.NoEncryption(),
+        ).hex()
+
+        return build_xmpp_uri(
+            path=str(xid),
+            xid_private=private_key_hex,
+            xid_created=created,
+        )
+
+    def _import(self, uri_str: str, profile: str) -> defer.Deferred:
+        client = G.host.get_client(profile)
+        return defer.ensureDeferred(self.import_xid(client, uri_str))
+
+    async def import_xid(self, client: SatXMPPEntity, uri_str: str) -> None:
+        """Import a XID private key from an xmpp: URI.
+
+        @param client: client instance.
+        @param uri_str: xmpp: URI containing xid-private and xid-created parameters.
+        @raise ValueError: if URI is malformed or verifications fail.
+        """
+        try:
+            parsed = parse_xmpp_uri(uri_str)
+        except ValueError as e:
+            raise ValueError(f"Invalid XMPP URI: {e}")
+
+        xid_from_uri = parsed.get("path", "")
+        if not xid_from_uri:
+            raise ValueError("URI missing XID (public key)")
+        try:
+            xid_jid = jid.JID(xid_from_uri)
+        except (ValueError, RuntimeError):
+            raise ValueError(f"Invalid XID in URI path: {xid_from_uri!r}")
+        if xid_jid.host != XID_DOMAIN:
+            raise ValueError(
+                f'XID domain mismatch: expected "{XID_DOMAIN}", '
+                f'got {xid_jid.host if xid_jid.host else "no domain"}'
+            )
+
+        public_key_hex = xid_jid.user
+
+        private_key_hex = parsed.get("xid-private")
+        created_xep = parsed.get("xid-created")
+
+        if not private_key_hex:
+            raise ValueError("URI missing 'xid-private' parameter")
+        if not created_xep:
+            raise ValueError("URI missing 'xid-created' parameter")
+
+        try:
+            private_key = Ed25519PrivateKey.from_private_bytes(
+                bytes.fromhex(private_key_hex)
+            )
+        except Exception as e:
+            raise ValueError(f"Invalid private key in URI: {e}")
+
+        derived_public_key_bytes = self.private_key_to_public_key(private_key)
+        derived_public_key_hex = ED25519_PREFIX + derived_public_key_bytes.hex()
+
+        if derived_public_key_hex != public_key_hex:
+            raise ValueError(
+                f"Public key mismatch: derived {derived_public_key_hex!r} "
+                f"does not match URI {public_key_hex!r}"
+            )
+
+        try:
+            items, __ = await self._ps.get_items(
+                client,
+                client.jid.userhostJID(),
+                PEP_NODE,
+            )
+        except exceptions.NotFound:
+            raise ValueError(
+                "No XID published on own PEP node; cannot verify imported key"
+            )
+
+        if not items:
+            raise ValueError(
+                f"XID {xid_jid!r} not found in own PEP node; "
+                "cannot import private key for unknown XID"
+            )
+        for item in items:
+            try:
+                xid_in_item = XID.from_element(item)
+            except exceptions.NotFound:
+                continue
+            if xid_in_item.xid == xid_jid:
+                break
+        else:
+            raise ValueError(
+                f"XID {xid_jid!r} not found in own PEP node; cannot import private key "
+                "for unknown XID"
+            )
+
+        await self._store_private_key(client, xid_jid, private_key)
+        log.info(f"XID {xid_jid!r} private key imported successfully")
+
+
+@implementer(itx_xmpp.IDisco)
+class XIDHandler(XMPPHandler):
+
+    def getDiscoInfo(self, requestor: str, target: str, nodeIdentifier: str = "") -> list:
+        return [disco.DiscoFeature(NS_XID)]
+
+    def getDiscoItems(
+        self, requestor: str, target: str, nodeIdentifier: str = ""
+    ) -> list:
+        return []
diff --git a/libervia/backend/plugins/plugin_misc_xid/models.py b/libervia/backend/plugins/plugin_misc_xid/models.py
new file mode 100644
index 00000000..111b0a83
--- /dev/null
+++ b/libervia/backend/plugins/plugin_misc_xid/models.py
@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+
+# Libervia plugin for XMPP Decentralized ID (XID)
+# 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/>.
+
+from collections.abc import Iterator
+from dataclasses import dataclass, field
+from typing import Self
+
+from pydantic import BaseModel, Field, RootModel
+from twisted.internet import defer
+from twisted.words.protocols.jabber import jid
+from twisted.words.xish import domish
+
+from libervia.backend.core import exceptions
+from libervia.backend.core.log import getLogger
+from libervia.backend.memory import persistent
+from libervia.backend.models.types import JIDType
+from libervia.backend.tools.utils import parse_xmpp_date, xmpp_date
+
+log = getLogger(__name__)
+
+NS_XID = "urn:xmpp:xid:0"
+
+
+class XID(BaseModel):
+    xid: JIDType = Field(description='XID (i.e., JID in the form "XXX@id.internal")')
+    created: float = Field(description="POSIX timestamp of XID creation")
+    has_private_key: bool = Field(
+        default=False, description="True if the private key is known and stored"
+    )
+    item_id: str | None = Field(default=None, description="PEP item ID")
+
+    @classmethod
+    def from_element(cls, xid_elt: domish.Element, item_id: str | None = None) -> Self:
+        """Create a XID instance from a <xid> element or its parent.
+
+        @param xid_elt: The <xid> element in NS_XID namespace, or a parent element.
+        @param item_id: PEP item ID where the XID is published.
+        @return: XID instance.
+        @raise exceptions.NotFound: If the <xid> element is not found.
+        """
+        if xid_elt.uri != NS_XID or xid_elt.name != "xid":
+            child_xid_elt = next(xid_elt.elements(NS_XID, "xid"), None)
+            if child_xid_elt is None:
+                raise exceptions.NotFound("<xid> element not found")
+            xid_elt = child_xid_elt
+        xid = jid.JID(str(xid_elt))
+        created = parse_xmpp_date(xid_elt["created"])
+        return cls(xid=xid, created=created, item_id=item_id)
+
+    def to_element(self) -> domish.Element:
+        """Build the <xid> element from this instance's data.
+
+        @return: <xid> element in NS_XID namespace.
+        """
+        xid_elt = domish.Element((NS_XID, "xid"))
+        xid_elt["created"] = xmpp_date(self.created)
+        xid_elt.addContent(str(self.xid))
+        return xid_elt
+
+
+class ListXID(RootModel):
+    root: list[XID]
+
+    def __iter__(self) -> Iterator[XID]:  # type: ignore
+        return iter(self.root)
+
+    def __len__(self) -> int:
+        return len(self.root)
+
+    def __getitem__(self, index: int) -> XID:
+        return self.root[index]
+
+
+class Challenge(BaseModel):
+    """Model for a XID challenge element."""
+
+    xid: JIDType = Field(description="XID being challenged")
+    timestamp: float = Field(description="POSIX timestamp when challenge was issued")
+    nonce: str = Field(description="hex-encoded random nonce")
+
+    @classmethod
+    def from_element(cls, elt: domish.Element) -> Self:
+        """Create a Challenge from a <challenge/> element.
+
+        @param elt: <challenge/> element in NS_XID namespace.
+        @return: Challenge instance.
+        """
+        xid_attr = elt.getAttribute("xid")
+        timestamp_attr = elt.getAttribute("timestamp")
+        nonce = str(elt)
+        if xid_attr is None or timestamp_attr is None:
+            raise ValueError(f"Missing required attribute on <challenge/>")
+        return cls(
+            xid=jid.JID(xid_attr),
+            timestamp=parse_xmpp_date(timestamp_attr),
+            nonce=nonce,
+        )
+
+    def to_element(self) -> domish.Element:
+        """Build a <challenge/> element.
+
+        @return: <challenge/> element in NS_XID namespace.
+        """
+        challenge_elt = domish.Element((NS_XID, "challenge"))
+        challenge_elt["xid"] = str(self.xid)
+        challenge_elt["timestamp"] = xmpp_date(self.timestamp)
+        challenge_elt.addContent(self.nonce)
+        return challenge_elt
+
+
+class Response(BaseModel):
+    """Model for a XID response element."""
+
+    xid: JIDType = Field(description="XID being verified")
+    timestamp: float = Field(description="POSIX timestamp from the challenge")
+    signature: str = Field(description="hex-encoded Ed25519 signature of the nonce")
+
+    @classmethod
+    def from_element(cls, elt: domish.Element) -> Self:
+        """Create a Response from a <response/> element.
+
+        @param elt: <response/> element in NS_XID namespace.
+        @return: Response instance.
+        """
+        xid_attr = elt.getAttribute("xid")
+        timestamp_attr = elt.getAttribute("timestamp")
+        signature = str(elt)
+        if xid_attr is None or timestamp_attr is None:
+            raise ValueError(f"Missing required attribute on <response/>")
+        return cls(
+            xid=jid.JID(xid_attr),
+            timestamp=parse_xmpp_date(timestamp_attr),
+            signature=signature,
+        )
+
+    def to_element(self) -> domish.Element:
+        """Build a <response/> element.
+
+        @return: <response/> element in NS_XID namespace.
+        """
+        response_elt = domish.Element((NS_XID, "response"))
+        response_elt["xid"] = str(self.xid)
+        response_elt["timestamp"] = xmpp_date(self.timestamp)
+        response_elt.addContent(self.signature)
+        return response_elt
+
+
+class XIDStorage(BaseModel):
+    """Persistent storage for a XID.
+
+    Only ``private_key`` is stored for now, but other values can easily added as needed in
+    the future by extending this model with default values.
+    """
+
+    private_key: str | None = Field(
+        default=None,
+        description=(
+            "Encrypted Ed25519 private key (hex-encoded raw key, encrypted with "
+            "profile password via [G.host.memory.encrypt_value])"
+        ),
+    )
+
+
+@dataclass
+class XIDClientData:
+    """Per-client data for XID plugin."""
+
+    _storage: persistent.PersistentBinaryDict
+    xids: dict[str, XIDStorage] = field(default_factory=dict)
+    pending_challenges: dict[str, defer.Deferred] = field(default_factory=dict)