diff --git a/libervia/cli/cmd_xid.py b/libervia/cli/cmd_xid.py new file mode 100644 index 00000000..3259b1a6 --- /dev/null +++ b/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 . + +"""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") + )