cli (xid): Set of commands to handle XMPP Decentralized ID (XID):

Implement commands to get, create, revoke, verify, import/export XID, and sign/verify a
file or inline data.

rel 477

libervia/cli/cmd_xid.py

@@ -0,0 +1,445 @@
+#!/usr/bin/env python3
+
+# Libervia CLI
+# 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/>.
+
+"""CLI commands for XID (XMPP Decentralized ID)."""
+
+from pathlib import Path
+import sys
+
+from rich.table import Table
+
+from libervia.backend.core.i18n import _
+from libervia.backend.plugins.plugin_misc_xid.models import ListXID, XID as XIDModel
+from libervia.backend.tools.common.date_utils import date_fmt
+from libervia.cli.constants import Const as C
+
+from . import base
+
+__commands__ = ["XID"]
+
+
+class XIDGet(base.CommandBase):
+    """Get XIDs published by an entity (self or remote)."""
+
+    def __init__(self, host) -> None:
+        extra_outputs = {"default": self.default_output}
+        super().__init__(
+            host,
+            "get",
+            help=_("get XIDs of an entity (self or remote)"),
+            use_output=C.OUTPUT_MODEL,
+            extra_outputs=extra_outputs,
+            use_verbose=True,
+        )
+
+    def add_parser_options(self) -> None:
+        self.parser.add_argument(
+            "entity",
+            nargs="?",
+            default="",
+            help=_("bare JID of the entity (default to own XIDs)"),
+        )
+        self.parser.add_argument(
+            "-a",
+            "--all",
+            action="store_true",
+            help=_("show all published XIDs instead of only the current"),
+        )
+
+    def default_output(self, xids: ListXID) -> None:
+        if not xids:
+            entity = self.args.entity or _("self")
+            self.console.print(_("No XID published by {entity}.").format(entity=entity))
+            return
+
+        if len(xids) == 1 and not self.verbosity:
+            # If we have a single XID to print, we don't use a table.
+            self.console.print(xids[0].xid)
+            return
+
+        table = Table(title=_("Published XIDs"))
+        table.add_column("🪪 " + _("XID"))
+        if self.verbosity:
+            table.add_column("📦 " + _("Item ID"))
+            table.add_column("📅 " + _("Created"))
+            table.add_column("🔒 " + _("Private Key Known"))
+
+        for xid in xids:
+            row = [str(xid.xid)]
+            if self.verbosity:
+                row.extend(
+                    [
+                        xid.item_id or "",
+                        date_fmt(xid.created, fmt="medium"),
+                        "✅" if xid.has_private_key else "❌",
+                    ]
+                )
+            table.add_row(*row)
+        self.console.print(table)
+
+    async def start(self) -> None:
+        try:
+            xids_s = await self.host.bridge.xid_get(
+                self.args.entity, self.args.all, self.host.profile
+            )
+            xids = ListXID.model_validate_json(xids_s)
+        except Exception as e:
+            self.disp(f"Failed to get XIDs: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+            return
+
+        await self.output(xids)
+        self.host.quit()
+
+
+class XIDCreate(base.CommandBase):
+    """Create a new XID key pair and publish it."""
+
+    def __init__(self, host) -> None:
+        extra_outputs = {"default": self.default_output}
+        super().__init__(
+            host,
+            "create",
+            help=_("create a new XID and publish it"),
+            use_output=C.OUTPUT_MODEL,
+            extra_outputs=extra_outputs,
+        )
+
+    def add_parser_options(self) -> None:
+        self.parser.add_argument(
+            "-i",
+            "--item-id",
+            default="current",
+            help=_("PEP item ID (default: current)"),
+        )
+
+    def default_output(self, xid: XIDModel) -> None:
+        self.console.print("[bold]XID created:[/bold]")
+        self.console.print(f"  XID:     {xid.xid}")
+        self.console.print(f"  Created: {date_fmt(xid.created, fmt='medium')}")
+
+    async def start(self) -> None:
+        try:
+            xid_s = await self.host.bridge.xid_create(
+                self.host.profile, self.args.item_id
+            )
+            xid = XIDModel.model_validate_json(xid_s)
+        except Exception as e:
+            self.disp(f"Failed to create XID: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+            return
+
+        await self.output(xid)
+        self.host.quit()
+
+
+class XIDRevoke(base.CommandBase):
+    """Revoke a XID."""
+
+    def __init__(self, host) -> None:
+        super().__init__(host, "revoke", help=_("revoke a XID"))
+
+    def add_parser_options(self) -> None:
+        self.parser.add_argument(
+            "xid",
+            help=_("XID to revoke"),
+        )
+        self.parser.add_argument(
+            "-r",
+            "--reason",
+            help=_("human-readable reason for revocation"),
+        )
+        self.parser.add_argument(
+            "-f",
+            "--force",
+            action="store_true",
+            help=_("revoke without confirmation"),
+        )
+
+    async def start(self) -> None:
+        if not self.args.force:
+            await self.host.confirm_or_quit(
+                _("Are you sure you want to revoke {xid}?").format(xid=self.args.xid)
+            )
+
+        try:
+            reason = self.args.reason or ""
+            await self.host.bridge.xid_revoke(self.args.xid, reason, self.host.profile)
+        except Exception as e:
+            self.disp(f"Failed to revoke XID: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+            return
+
+        self.disp(_("XID revoked successfully"))
+        self.host.quit()
+
+
+class XIDVerify(base.CommandBase):
+    """Verify an entity's XID via challenge/response."""
+
+    def __init__(self, host) -> None:
+        super().__init__(host, "verify", help=_("verify an entity's XID ownership"))
+
+    def add_parser_options(self) -> None:
+        self.parser.add_argument("entity", help=_("bare JID of the entity"))
+        self.parser.add_argument("xid", help=_("XID to verify"))
+
+    async def start(self) -> None:
+        try:
+            valid = await self.host.bridge.xid_verify(
+                self.args.entity, self.args.xid, self.host.profile
+            )
+        except Exception as e:
+            self.disp(f"Failed to verify XID: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+            return
+
+        if valid:
+            self.console.print(
+                f"[bold green]✅ XID {self.args.xid} verified for "
+                f"{self.args.entity}[/bold green]"
+            )
+            self.host.quit()
+        else:
+            self.console.print(
+                f"[bold red]❌ XID {self.args.xid} verification failed "
+                f"for {self.args.entity}[/bold red]"
+            )
+            self.host.quit(C.EXIT_DATA_ERROR)
+
+
+class XIDSignatureSign(base.CommandBase):
+    """Sign a file or data with a XID private key."""
+
+    def __init__(self, host) -> None:
+        super().__init__(host, "sign", help=_("sign data with a XID private key"))
+
+    def add_parser_options(self) -> None:
+        self.parser.add_argument(
+            "-x",
+            "--xid",
+            default=None,
+            help=_("XID whose private key to use (default to current)"),
+        )
+        self.parser.add_argument(
+            "file_path_or_data",
+            nargs="?",
+            default="-",
+            help=_("file to sign, or inline data with --data; defaults to stdin (-)"),
+        )
+        self.parser.add_argument(
+            "-d",
+            "--data",
+            action="store_true",
+            help=_("treat positional argument as inline data instead of file path"),
+        )
+        self.parser.add_argument(
+            "-o",
+            "--output",
+            type=Path,
+            help=_("output file for the signature (hex)"),
+        )
+
+    async def start(self) -> None:
+        xid = self.args.xid or ""
+        input_value = self.args.file_path_or_data
+
+        is_inline = self.args.data or input_value == "-"
+
+        try:
+            if is_inline and input_value != "-":
+                sig_hex = await self.host.bridge.xid_signature_sign_data(
+                    xid, input_value, self.host.profile
+                )
+            elif input_value == "-":
+                data = sys.stdin.buffer.read().decode("utf-8", errors="replace")
+                sig_hex = await self.host.bridge.xid_signature_sign_data(
+                    xid, data, self.host.profile
+                )
+            else:
+                sig_hex = await self.host.bridge.xid_signature_sign_file(
+                    xid, input_value, self.host.profile
+                )
+        except Exception as e:
+            self.disp(f"Failed to sign data: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+            return
+
+        if self.args.output:
+            self.args.output.parent.mkdir(parents=True, exist_ok=True)
+            self.args.output.write_text(sig_hex)
+            self.disp(f"Signature written to {self.args.output}")
+        else:
+            self.disp(sig_hex)
+
+        self.host.quit()
+
+
+class XIDSignatureCheck(base.CommandBase):
+    """Verify a signature against a XID's public key."""
+
+    def __init__(self, host) -> None:
+        super().__init__(host, "check", help=_("verify a signature with a XID"))
+
+    def add_parser_options(self) -> None:
+        self.parser.add_argument("xid", help=_("XID to verify against"))
+        self.parser.add_argument(
+            "file_path_or_data",
+            nargs="?",
+            default="-",
+            help=_("file to verify, or inline data with --data; defaults to stdin (-)"),
+        )
+        self.parser.add_argument(
+            "-d",
+            "--data",
+            action="store_true",
+            help=_("treat positional argument as inline data instead of file path"),
+        )
+        sig_group = self.parser.add_mutually_exclusive_group(required=True)
+        sig_group.add_argument(
+            "-s",
+            "--signature",
+            help=_("inline hex-encoded signature"),
+        )
+        sig_group.add_argument(
+            "-f",
+            "--signature-file",
+            type=Path,
+            help=_("file containing hex-encoded signature"),
+        )
+
+    async def start(self) -> None:
+        input_value = self.args.file_path_or_data
+        is_inline = self.args.data or input_value == "-"
+
+        # Read signature
+        if self.args.signature is not None:
+            sig_hex = self.args.signature
+        else:
+            sig_hex = self.args.signature_file.read_text().strip()
+
+        try:
+            if is_inline and input_value != "-":
+                valid = await self.host.bridge.xid_signature_check_data(
+                    self.args.xid, input_value, sig_hex
+                )
+            elif input_value == "-":
+                text = sys.stdin.buffer.read().decode("utf-8", errors="replace")
+                valid = await self.host.bridge.xid_signature_check_data(
+                    self.args.xid, text, sig_hex
+                )
+            else:
+                valid = await self.host.bridge.xid_signature_check_file(
+                    self.args.xid, input_value, sig_hex, self.host.profile
+                )
+        except Exception as e:
+            self.disp(f"Failed to verify signature: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+            return
+
+        if valid:
+            self.console.print("[bold green]✅ Signature is valid[/bold green]")
+            self.host.quit()
+        else:
+            self.console.print("[bold red]❌ Signature is invalid[/bold red]")
+            self.host.quit(C.EXIT_DATA_ERROR)
+
+
+class XIDSignature(base.CommandBase):
+    """Sign or verify data with a XID."""
+
+    subcommands = (
+        XIDSignatureSign,
+        XIDSignatureCheck,
+    )
+
+    def __init__(self, host) -> None:
+        super().__init__(
+            host,
+            "signature",
+            use_profile=False,
+            help=_("sign or verify data with a XID"),
+            aliases=["sign"],
+        )
+
+
+class XIDExport(base.CommandBase):
+    """Generate an xmpp: URI for private key export."""
+
+    def __init__(self, host) -> None:
+        super().__init__(host, "export", help=_("export XID private key as xmpp: URI"))
+
+    def add_parser_options(self) -> None:
+        self.parser.add_argument(
+            "xid", nargs="?", default=None, help=_("XID to export (default to current)")
+        )
+
+    async def start(self) -> None:
+        xid = self.args.xid or ""
+        try:
+            uri = await self.host.bridge.xid_export(xid, self.host.profile)
+        except Exception as e:
+            self.disp(f"Failed to export XID: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+            return
+
+        self.console.print(uri)
+        self.host.quit()
+
+
+class XIDImport(base.CommandBase):
+    """Import a XID private key from an xmpp: URI."""
+
+    def __init__(self, host) -> None:
+        super().__init__(host, "import", help=_("import XID private key from xmpp: URI"))
+
+    def add_parser_options(self) -> None:
+        self.parser.add_argument(
+            "uri",
+            help=_("xmpp: URI containing xid-private and xid-created parameters"),
+        )
+
+    async def start(self) -> None:
+        try:
+            await self.host.bridge.xid_import(self.args.uri, self.host.profile)
+        except Exception as e:
+            self.disp(f"Failed to import XID: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+            return
+
+        self.disp(_("XID private key imported successfully"))
+        self.host.quit()
+
+
+class XID(base.CommandBase):
+    """Top-level XID command."""
+
+    subcommands: tuple[type[base.CommandBase], ...] = (
+        XIDGet,
+        XIDCreate,
+        XIDRevoke,
+        XIDVerify,
+        XIDSignature,
+        XIDExport,
+        XIDImport,
+    )
+
+    def __init__(self, host) -> None:
+        super().__init__(
+            host, "xid", use_profile=False, help=_("XMPP Decentralized ID management")
+        )