libervia-backend/tests/unit/test_plugin_misc_xid.py

#!/usr/bin/env python3

# Libervia: an XMPP client
# 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 time
from unittest.mock import AsyncMock, MagicMock, patch

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from pytest import raises
from pytest_twisted import ensureDeferred as ed
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish

from libervia.backend.core import exceptions
from libervia.backend.plugins.plugin_misc_xid import (
    ED25519_PREFIX,
    NS_XID,
    XIDPlugin,
    XID_DOMAIN,
)
from libervia.backend.plugins.plugin_misc_xid.models import Challenge, Response, XID
from libervia.backend.tools.utils import xmpp_date


class TestXIDModel:
    def test_xid_roundtrip(self):
        """XID is correctly serialized and parsed via to_element/from_element."""
        xid_jid = jid.JID("00abcdef1234567890@id.internal")
        created = time.time()
        xid = XID(xid=xid_jid, created=created)
        xid_elt = xid.to_element()
        parsed = XID.from_element(xid_elt)
        assert parsed.xid == xid_jid
        assert abs(parsed.created - created) < 1

    def test_xid_from_parent_element(self):
        """XID is correctly parsed from a parent <item> element."""
        xid_jid = jid.JID("00abcdef1234567890@id.internal")
        item_elt = domish.Element((None, "item"))
        xid_elt = item_elt.addElement((NS_XID, "xid"))
        xid_elt["created"] = xmpp_date(time.time())
        xid_elt.addContent(str(xid_jid))
        parsed = XID.from_element(item_elt)
        assert parsed.xid == xid_jid

    def test_challenge_roundtrip(self):
        """Challenge is correctly serialized and parsed via to_element/from_element."""
        xid_jid = jid.JID("00abcdef1234567890@id.internal")
        nonce = "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6"
        ts = time.time()
        challenge = Challenge(xid=xid_jid, timestamp=ts, nonce=nonce)
        challenge_elt = challenge.to_element()
        parsed = Challenge.from_element(challenge_elt)
        assert parsed.xid == xid_jid
        assert parsed.nonce == nonce
        assert abs(parsed.timestamp - ts) < 1

    def test_response_roundtrip(self):
        """Response is correctly serialized and parsed via to_element/from_element."""
        xid_jid = jid.JID("00abcdef1234567890@id.internal")
        signature = "f1e2d3c4b5a6f7e8d9c0b1a2f3e4d5c6"
        ts = time.time()
        response = Response(xid=xid_jid, timestamp=ts, signature=signature)
        response_elt = response.to_element()
        parsed = Response.from_element(response_elt)
        assert parsed.xid == xid_jid
        assert parsed.signature == signature
        assert abs(parsed.timestamp - ts) < 1


class TestXIDPlugin:
    def test_get_local_part(self):
        """Public key bytes are converted to XID localpart with prefix."""
        public_key = bytes(range(32))
        localpart = XIDPlugin.get_local_part(public_key)
        assert localpart.startswith(ED25519_PREFIX)
        assert localpart[2:] == public_key.hex()

    def test_xid_to_public_key(self):
        """Public key bytes are extracted from a valid XID."""
        public_key = bytes(range(1, 33))
        localpart = ED25519_PREFIX + public_key.hex()
        xid_jid = jid.JID(f"{localpart}@{XID_DOMAIN}")
        extracted = XIDPlugin.xid_to_public_key(xid_jid)
        assert extracted == public_key

    def test_xid_to_public_key_wrong_prefix(self):
        """ValueError is raised for XID without Ed25519 prefix."""
        xid_jid = jid.JID("ff00" + "ab" * 32 + "@id.internal")
        with raises(ValueError, match="does not start with Ed25519 prefix"):
            XIDPlugin.xid_to_public_key(xid_jid)

    def test_xid_to_public_key_wrong_length(self):
        """ValueError is raised for XID with wrong key length."""
        xid_jid = jid.JID("00" + "ab" * 10 + "@id.internal")
        with raises(ValueError, match="wrong length"):
            XIDPlugin.xid_to_public_key(xid_jid)

    def test_private_key_to_public_key(self):
        """Public key derivation produces 32 raw bytes."""
        private_key = Ed25519PrivateKey.generate()
        public_key_bytes = XIDPlugin.private_key_to_public_key(private_key)
        assert len(public_key_bytes) == 32
        assert public_key_bytes == private_key.public_key().public_bytes(
            serialization.Encoding.Raw, serialization.PublicFormat.Raw
        )

    @ed
    async def test_signature_check_valid(self):
        """Valid signature returns True."""
        plugin = XIDPlugin(MagicMock())
        private_key = Ed25519PrivateKey.generate()
        public_key = XIDPlugin.private_key_to_public_key(private_key)
        localpart = XIDPlugin.get_local_part(public_key)
        xid_jid = jid.JID(f"{localpart}@{XID_DOMAIN}")

        data = "test data"
        signature = private_key.sign(data.encode())

        result = await plugin.signature_check(
            xid_jid, data=data, signature_hex=signature.hex()
        )
        assert result is True

    @ed
    async def test_signature_check_invalid(self):
        """Invalid signature returns False."""
        plugin = XIDPlugin(MagicMock())
        private_key = Ed25519PrivateKey.generate()
        public_key = XIDPlugin.private_key_to_public_key(private_key)
        localpart = XIDPlugin.get_local_part(public_key)
        xid_jid = jid.JID(f"{localpart}@{XID_DOMAIN}")

        wrong_data = "wrong data"
        signature = private_key.sign("other data".encode())

        result = await plugin.signature_check(
            xid_jid, data=wrong_data, signature_hex=signature.hex()
        )
        assert result is False

    @ed
    async def test_sign_and_verify_roundtrip(self, host, client):
        """Signing then verifying data produces a valid signature."""
        with patch.dict(
            host.plugins,
            {"XEP-0060": AsyncMock()},
        ):
            plugin = XIDPlugin(host)

            private_key = Ed25519PrivateKey.generate()
            public_key = XIDPlugin.private_key_to_public_key(private_key)
            localpart = XIDPlugin.get_local_part(public_key)
            xid_jid = jid.JID(f"{localpart}@{XID_DOMAIN}")

            private_key_hex = private_key.private_bytes(
                serialization.Encoding.Raw,
                serialization.PrivateFormat.Raw,
                serialization.NoEncryption(),
            ).hex()

            xid_data = MagicMock()
            client.get_plugin_data.return_value = xid_data
            xid_data.xids = {str(xid_jid): MagicMock(private_key="encrypted")}

            with patch.object(host.memory, "decrypt_value", return_value=private_key_hex):
                sig_hex = await plugin.sign(client, xid_jid, data="hello")

            result = await plugin.signature_check(
                xid_jid, data="hello", signature_hex=sig_hex
            )
            assert result is True

    @ed
    async def test_sign_and_verify_wrong_data(self, host, client):
        """Verifying with different data returns False."""
        with patch.dict(host.plugins, {"XEP-0060": AsyncMock()}):
            plugin = XIDPlugin(host)

            private_key = Ed25519PrivateKey.generate()
            public_key = XIDPlugin.private_key_to_public_key(private_key)
            localpart = XIDPlugin.get_local_part(public_key)
            xid_jid = jid.JID(f"{localpart}@{XID_DOMAIN}")

            private_key_hex = private_key.private_bytes(
                serialization.Encoding.Raw,
                serialization.PrivateFormat.Raw,
                serialization.NoEncryption(),
            ).hex()

            xid_data = MagicMock()
            client.get_plugin_data.return_value = xid_data
            xid_data.xids = {str(xid_jid): MagicMock(private_key="encrypted")}

            with patch.object(host.memory, "decrypt_value", return_value=private_key_hex):
                sig_hex = await plugin.sign(client, xid_jid, data="hello")

            result = await plugin.signature_check(
                xid_jid, data="wrong", signature_hex=sig_hex
            )
            assert result is False

    @ed
    async def test_sign_xor_guard(self, host, client):
        """DataError is raised when both or neither file_path and data are given."""
        with patch.dict(host.plugins, {"XEP-0060": AsyncMock()}):
            plugin = XIDPlugin(host)

            with raises(exceptions.DataError):
                await plugin.sign(client, None)
            with raises(exceptions.DataError):
                await plugin.sign(client, None, file_path="a", data="b")

    @ed
    async def test_signature_check_xor_guard(self):
        """DataError is raised when both or neither file_path and data are given."""
        plugin = XIDPlugin(MagicMock())

        with raises(exceptions.DataError):
            await plugin.signature_check(
                jid.JID("00" + "ab" * 32 + "@id.internal"), signature_hex="aa"
            )
        with raises(exceptions.DataError):
            await plugin.signature_check(
                jid.JID("00" + "ab" * 32 + "@id.internal"),
                file_path="a",
                data="b",
                signature_hex="aa",
            )