libervia-backend/libervia/cli/cmd_xid.py

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