libervia-backend/libervia/backend/plugins/plugin_misc_xid/models.py

#!/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)