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 misc XID: XMPP Decentralized ID implementation:

Browse source 
XMPP Decentralized ID (XID) implementation. All the spec but private key synchronisation
is implemented, and private key can be synchronised with import/export method which
through a URI usable with QR Code as explained in the spec.

rel 477
This commit is contained in:
Goffi 2026-06-06 12:23:39 +02:00
commit 9d4c2bace9
2 changed files with 1104 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

919
libervia/backend/plugins/plugin_misc_xid/__init__.py Normal file
View file

@ -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 []

185
libervia/backend/plugins/plugin_misc_xid/models.py Normal file
View file

@ -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)