diff --git a/README.md b/README.md index e4277ad..bb5e8ef 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,18 @@ pipx install "git+https://github.com/QortiumDev/Qortium-Python-CLI.git" pipx upgrade qortium-cli ``` +### Startup Update Checks + +On interactive startup, the CLI checks the GitHub releases page at most once per +day and lets you know when a newer release is available. + +- The newest stable release is shown when it is newer than the installed CLI. +- A prerelease is shown first only when it is newer than both the installed CLI + and the newest stable release. +- A prerelease is not offered when the newest stable release is newer. + +Set `QORTIUM_CLI_NO_UPDATE_CHECK=1` before launch to skip startup update checks. + ## 3) Start The App ```bash @@ -42,7 +54,8 @@ On first launch, setup asks for: 1. Endpoint URL (default: `http://127.0.0.1:24891`) 2. Timeout seconds 3. API key -4. Wallet key: Base58 private key or seed phrase +4. Wallet key: Base58 private key, seed phrase, existing wallet file, or new + encrypted wallet file Then it auto-fills: @@ -50,6 +63,25 @@ Then it auto-fills: - account address - display name (primary name if available) +For an existing Qortium Home wallet file, choose `Use wallet file`, enter the JSON +backup path, then enter the wallet password. Version 2 master-seed wallets use +address `0`; version 3 private-key wallets use the stored private key directly. + +To create a new account, choose `New wallet file`, enter and confirm a +wallet password, then choose where to save the JSON backup. The CLI creates an +encrypted version 2 master-seed wallet compatible with Qortium Home, uses address +`0` for CLI setup, and keeps the password out of the CLI config. Keep the wallet +file and password; the master-seed wallet can derive additional addresses later. + +The endpoint URL is checked with `/admin/status` when selected. If the node is not +connected or does not return node status, setup lets you enter a different endpoint +or continue with the selected endpoint anyway. + +When the selected endpoint is local, the CLI tries to detect a running +Qortium/Qortal Core process and read its `apikey.txt`. If exactly one matching +local Core API key is found, setup offers it as the default without printing the +key; press Enter at the API-key prompt to use it. + ## 5) Runtime Settings Files The app writes runtime settings here: @@ -67,10 +99,16 @@ Optional custom location: - set `QORTIUM_CLI_HOME=/path/to/folder` before launch -## 6) Reconfigure Or Reset +## 6) Reconfigure Or Run Setup -- Choose menu option `9` to change the endpoint URL, request timeout, API key, - or wallet key independently. +- Choose main menu option `9` to change the endpoint URL, request timeout, + API key, or wallet key independently. +- Inside reconfigure, choose option `9` (`Run initial setup`) to run the + initial setup flow again, then return to the main menu. +- Endpoint URL changes run the same connection check and can be retried or kept + anyway if the node is offline. +- API key changes also look for a matching local Core `apikey.txt`; when one is + found, press Enter to use it or type a different key. - Changing the API key does not ask for, replace, or re-derive your private key. - Delete the runtime settings files above to run first-time setup again. @@ -78,7 +116,9 @@ Optional custom location: - `qortium-cli: command not found`: open a new terminal after `pipx ensurepath`, or add pipx bin dir to `PATH`. - Cannot reach node/API: verify your node is running and listening on `127.0.0.1:24891`. -- API key prompt blocks setup: enter a valid `X-API-KEY`; in the reconfigure menu, press Enter to cancel without changing it. +- API key prompt blocks setup: enter a valid `X-API-KEY`; if local Core detection + finds a key, press Enter to use it. In the reconfigure menu, type `/cancel` to + cancel a detected-key prompt without changing it. ## 8) Sign And Submit Raw Transactions @@ -185,8 +225,15 @@ The main menu is organized as: 4. `Register Name` 5. `Wallet` 6. `QDN Resources` +8. `Help/Info` 9. `Reconfigure endpoint/config` +`Register Name` checks the configured wallet for existing registered names first. +If none are found, it opens the normal registration prompt. If names already +exist, it lists them and lets you choose `Update name` or `New name`; choose `0` +to go back. At the register-name prompt, press Enter without typing a name to +cancel. + The Groups menu contains: - `Join group (/groups/join)` diff --git a/pyproject.toml b/pyproject.toml index 0b3c8a0..2b0358d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "qortium-cli" -version = "0.2.0" +version = "0.3.0" description = "Qortium/Qortal terminal toolbox" readme = "README.md" requires-python = ">=3.10" diff --git a/qortium_cli/app.py b/qortium_cli/app.py index ad35010..8c46928 100644 --- a/qortium_cli/app.py +++ b/qortium_cli/app.py @@ -20,6 +20,7 @@ startup_splash, warn, ) +from qortium_cli.update_checker import maybe_notify_available_updates from qortium_cli.utils import pretty_exception @@ -77,4 +78,5 @@ def run() -> None: settings_dir = resolve_settings_dir(project_root) ctx = create_context(settings_dir) configure_first_run_files(ctx) + maybe_notify_available_updates(settings_dir) run_main_menu(ctx) diff --git a/qortium_cli/chat_format.py b/qortium_cli/chat_format.py new file mode 100644 index 0000000..75277e8 --- /dev/null +++ b/qortium_cli/chat_format.py @@ -0,0 +1,428 @@ +from __future__ import annotations + +import base64 +import json +from dataclasses import dataclass +from typing import Any, Mapping + +from qortium_cli.crypto import b58decode, is_base58 + +MAX_REACTION_CONTENT_LENGTH = 32 +DEFAULT_REACTION_CATEGORIES = ( + ( + "Quick", + ( + "\U0001f44d", + "\u2764\ufe0f", + "\U0001f602", + "\U0001f62e", + "\U0001f622", + "\U0001f64f", + ), + ), + ( + "Faces", + ( + "\U0001f600", + "\U0001f604", + "\U0001f60a", + "\U0001f60e", + "\U0001f914", + "\U0001f60d", + "\U0001f62d", + "\U0001f621", + ), + ), + ( + "Hands", + ( + "\U0001f44e", + "\U0001f44f", + "\U0001f64c", + "\U0001f4aa", + "\U0001f91d", + "\U0001f440", + "\U0001faf6", + "\U0001f91e", + ), + ), + ( + "Hearts", + ( + "\U0001f9e1", + "\U0001f49b", + "\U0001f49a", + "\U0001f499", + "\U0001f49c", + "\U0001f5a4", + "\U0001f90d", + "\U0001f494", + ), + ), + ( + "Symbols", + ( + "\u2705", + "\u274c", + "\U0001f4af", + "\u2b50", + "\U0001f525", + "\U0001f389", + "\U0001f680", + "\U0001f4a1", + ), + ), +) +DEFAULT_REACTION_OPTIONS = tuple( + reaction + for _, reactions in DEFAULT_REACTION_CATEGORIES + for reaction in reactions +) +DEFAULT_REACTION_ORDER = { + reaction: index for index, reaction in enumerate(DEFAULT_REACTION_OPTIONS) +} + + +@dataclass(frozen=True) +class ChatReaction: + content: str + content_state: bool + + +@dataclass(frozen=True) +class DecodedChatMessage: + body: str + kind: str + reaction: ChatReaction | None = None + replied_to: str | None = None + + +@dataclass(frozen=True) +class MessageThread: + latest: Mapping[str, Any] + original: Mapping[str, Any] + revisions: tuple[Mapping[str, Any], ...] + + +@dataclass(frozen=True) +class MessageReactionParticipant: + sender: str + timestamp: int + + +@dataclass(frozen=True) +class MessageReactionSummary: + content: str + count: int + latest_timestamp: int + reacted_by_self: bool + reactors: tuple[MessageReactionParticipant, ...] + + +def _extract_doc_text(node: Any) -> str: + if isinstance(node, dict): + node_type = str(node.get("type", "")) + if node_type == "text": + return str(node.get("text", "")) + + parts = [_extract_doc_text(child) for child in node.get("content", [])] + if node_type == "paragraph": + return "".join(parts) + "\n" + return "".join(parts) + + if isinstance(node, list): + return "".join(_extract_doc_text(child) for child in node) + + return "" + + +def _truthy_bool(value: Any, default: bool) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"true", "1", "yes"}: + return True + if normalized in {"false", "0", "no"}: + return False + return default + + +def _has_readable_encrypted_payload(message: Mapping[str, Any]) -> bool: + return ( + str(message.get("decryptionStatus", "") or "").upper() == "DECRYPTED" + or str(message.get("status", "") or "").upper() == "DECRYPTED" + ) + + +def _decode_text_payload(data: Any, encoding: Any) -> str | None: + if not isinstance(data, str) or not data: + return "" + + enc = str(encoding or "").upper() + decoded_bytes: bytes | None = None + + if enc == "BASE64": + try: + decoded_bytes = base64.b64decode(data) + except Exception: + return data + elif enc == "BASE58" or is_base58(data): + try: + decoded_bytes = b58decode(data) + except Exception: + return data + + if decoded_bytes is None: + return data + + return decoded_bytes.decode("utf-8", errors="replace") + + +def _normalize_reaction_content(value: Any) -> str | None: + if not isinstance(value, str): + return None + content = value.strip() + if 0 < len(content) <= MAX_REACTION_CONTENT_LENGTH: + return content + return None + + +def _get_envelope_reaction(envelope: Mapping[str, Any]) -> ChatReaction | None: + if envelope.get("type") != "reaction": + return None + + content = _normalize_reaction_content(envelope.get("content")) + if not content: + return None + + return ChatReaction( + content=content, + content_state=envelope.get("contentState") is not False, + ) + + +def _unwrap_chat_text_envelope(value: str) -> DecodedChatMessage: + body = value + reaction: ChatReaction | None = None + replied_to: str | None = None + + for _ in range(3): + try: + parsed = json.loads(body) + except Exception: + break + + if not isinstance(parsed, dict): + break + + message_doc = parsed.get("messageText") + if isinstance(message_doc, dict): + doc_text = _extract_doc_text(message_doc).strip() + if doc_text: + return DecodedChatMessage(body=doc_text, kind="text") + + envelope = parsed + if not isinstance(envelope.get("message"), str): + break + + reaction = _get_envelope_reaction(envelope) + body = str(envelope["message"]) + if reaction: + return DecodedChatMessage(body=body, kind="reaction", reaction=reaction) + + if replied_to is None: + envelope_reply = envelope.get("repliedTo") + if isinstance(envelope_reply, str) and envelope_reply: + replied_to = envelope_reply + + return DecodedChatMessage(body=body, kind="text", replied_to=replied_to) + + +def decode_chat_message(message: Mapping[str, Any]) -> DecodedChatMessage: + is_encrypted = _truthy_bool(message.get("isEncrypted"), default=False) + is_text = _truthy_bool(message.get("isText"), default=True) + + if is_encrypted and (not _has_readable_encrypted_payload(message) or not message.get("data")): + return DecodedChatMessage(body="Encrypted message", kind="encrypted") + + if not is_text: + return DecodedChatMessage(body="Binary message", kind="binary") + + data = message.get("data") + if not data: + return DecodedChatMessage(body="", kind="empty") + + encoding = str(message.get("encoding") or "").upper() + if encoding and encoding not in {"BASE58", "BASE64"}: + return DecodedChatMessage(body="Unsupported message encoding", kind="unsupported") + + text = _decode_text_payload(data, encoding) + if text is None: + return DecodedChatMessage(body="Unable to decode message", kind="unsupported") + + try: + return _unwrap_chat_text_envelope(text) + except Exception: + return DecodedChatMessage(body="Unable to decode message", kind="unsupported") + + +def build_chat_message_text(text: str, replied_to: str | None = None) -> str: + if not replied_to: + return text + return json.dumps( + {"message": text, "repliedTo": replied_to}, + separators=(",", ":"), + ) + + +def build_reaction_message_text(content: str, content_state: bool) -> str: + normalized_content = _normalize_reaction_content(content) + if not normalized_content: + raise ValueError("Reaction content must be a short emoji string.") + return json.dumps( + { + "message": "", + "type": "reaction", + "content": normalized_content, + "contentState": bool(content_state), + }, + separators=(",", ":"), + ) + + +def is_reaction_chat_message(message: Mapping[str, Any]) -> bool: + return decode_chat_message(message).kind == "reaction" + + +def _message_signature(message: Mapping[str, Any]) -> str: + return str(message.get("signature") or "").strip() + + +def _message_chat_reference(message: Mapping[str, Any]) -> str: + return str(message.get("chatReference") or "").strip() + + +def _message_sender(message: Mapping[str, Any]) -> str: + return str(message.get("sender") or "").strip() + + +def sort_messages_by_timestamp(messages: list[Mapping[str, Any]]) -> list[Mapping[str, Any]]: + return sorted(messages, key=lambda item: int(item.get("timestamp") or 0)) + + +def build_message_threads(messages: list[Mapping[str, Any]]) -> list[MessageThread]: + originals_by_signature: dict[str, Mapping[str, Any]] = {} + revisions_by_reference: dict[str, list[Mapping[str, Any]]] = {} + + for message in messages: + if is_reaction_chat_message(message): + continue + + signature = _message_signature(message) + if signature and not _message_chat_reference(message): + originals_by_signature[signature] = message + + for message in messages: + if is_reaction_chat_message(message): + continue + + reference = _message_chat_reference(message) + if not reference: + continue + + revisions_by_reference.setdefault(reference, []).append(message) + + threads: list[MessageThread] = [] + for message in messages: + if is_reaction_chat_message(message): + continue + + reference = _message_chat_reference(message) + referenced_original = originals_by_signature.get(reference) if reference else None + if referenced_original and _message_sender(referenced_original) == _message_sender(message): + continue + + signature = _message_signature(message) + revisions = sort_messages_by_timestamp( + [ + revision + for revision in revisions_by_reference.get(signature, []) + if _message_sender(revision) == _message_sender(message) + ] + ) + threads.append( + MessageThread( + latest=revisions[-1] if revisions else message, + original=message, + revisions=tuple(revisions), + ) + ) + + return threads + + +def build_message_reaction_index( + messages: list[Mapping[str, Any]], + self_address: str | None = None, +) -> dict[str, tuple[MessageReactionSummary, ...]]: + reactions_by_reference: dict[str, dict[str, dict[str, MessageReactionParticipant]]] = {} + + for message in sort_messages_by_timestamp(messages): + reference = _message_chat_reference(message) + sender = _message_sender(message) + if not reference or not sender: + continue + + decoded = decode_chat_message(message) + if decoded.kind != "reaction" or not decoded.reaction: + continue + + content_map = reactions_by_reference.setdefault(reference, {}) + sender_map = content_map.setdefault(decoded.reaction.content, {}) + if decoded.reaction.content_state: + sender_map[sender] = MessageReactionParticipant( + sender=sender, + timestamp=int(message.get("timestamp") or 0), + ) + else: + sender_map.pop(sender, None) + + reaction_index: dict[str, tuple[MessageReactionSummary, ...]] = {} + for signature, content_map in reactions_by_reference.items(): + summaries: list[MessageReactionSummary] = [] + for content, sender_map in content_map.items(): + reactors = tuple( + sorted( + sender_map.values(), + key=lambda reaction: (-reaction.timestamp, reaction.sender), + ) + ) + if not reactors: + continue + + latest_timestamp = max(reaction.timestamp for reaction in reactors) + summaries.append( + MessageReactionSummary( + content=content, + count=len(reactors), + latest_timestamp=latest_timestamp, + reacted_by_self=bool(self_address and self_address in sender_map), + reactors=reactors, + ) + ) + + summaries.sort( + key=lambda reaction: ( + DEFAULT_REACTION_ORDER.get(reaction.content, len(DEFAULT_REACTION_OPTIONS)), + -reaction.latest_timestamp, + reaction.content, + ) + ) + reaction_index[signature] = tuple(summaries) + + return reaction_index diff --git a/qortium_cli/constants.py b/qortium_cli/constants.py index 2f19b51..25ef8d0 100644 --- a/qortium_cli/constants.py +++ b/qortium_cli/constants.py @@ -1,6 +1,7 @@ from colorama import Fore, Style -APP_TITLE = "Qortium CLI 0.2.0" +APP_VERSION = "0.3.0" +APP_TITLE = f"Qortium CLI {APP_VERSION}" SETUP_TITLE = "Qortium Setup" DEFAULT_BASE_URL = "http://127.0.0.1:24891" diff --git a/qortium_cli/core_detection.py b/qortium_cli/core_detection.py new file mode 100644 index 0000000..c295afc --- /dev/null +++ b/qortium_cli/core_detection.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +from dataclasses import dataclass +import ipaddress +import json +import os +from pathlib import Path +from urllib.parse import urlparse + + +API_KEY_FILE = "apikey.txt" +MAINNET_API_PORT = 14891 +TESTNET_API_PORT = 24891 + + +@dataclass(frozen=True) +class LocalCoreApiKey: + api_key: str + api_key_path: Path + api_key_directory: Path + cwd: Path + jar_path: Path + pid: int + settings_path: Path + + +def _is_loopback_host(host: str | None) -> bool: + if not host: + return False + + normalized = host.strip().lower() + if normalized == "localhost": + return True + + try: + return ipaddress.ip_address(normalized).is_loopback + except ValueError: + return False + + +def _endpoint_port(base_url: str) -> int | None: + parsed = urlparse(base_url) + if not _is_loopback_host(parsed.hostname): + return None + + if parsed.port is not None: + return parsed.port + if parsed.scheme == "https": + return 443 + if parsed.scheme == "http": + return 80 + return None + + +def _resolve_process_path(raw_path: str, cwd: Path) -> Path: + path = Path(raw_path) + if path.is_absolute(): + return path.resolve() + return (cwd / path).resolve() + + +def _read_json_file(path: Path) -> dict: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except Exception: + return {} + return data if isinstance(data, dict) else {} + + +def _effective_settings_path(settings_path: Path, cwd: Path) -> Path: + settings = _read_json_file(settings_path) + user_path = settings.get("userPath") + if not isinstance(user_path, str) or not user_path.strip(): + return settings_path + + user_dir = Path(user_path.strip()) + if not user_dir.is_absolute(): + user_dir = cwd / user_dir + redirected = (user_dir / settings_path.name).resolve() + return redirected if redirected.exists() else settings_path + + +def _settings_api_port(settings: dict) -> int | None: + raw_api_port = settings.get("apiPort") + try: + if raw_api_port is not None: + return int(raw_api_port) + except (TypeError, ValueError): + pass + + if "isTestNet" in settings: + return TESTNET_API_PORT if bool(settings.get("isTestNet", False)) else MAINNET_API_PORT + return None + + +def _settings_api_key_directory(settings_path: Path, cwd: Path) -> Path: + settings = _read_json_file(settings_path) + raw_api_key_path = settings.get("apiKeyPath") + if isinstance(raw_api_key_path, str) and raw_api_key_path.strip(): + return _resolve_process_path(raw_api_key_path.strip(), cwd) + return cwd.resolve() + + +def _read_api_key(api_key_directory: Path) -> tuple[str, Path] | None: + api_key_path = api_key_directory / API_KEY_FILE + try: + api_key = api_key_path.read_text(encoding="utf-8").strip() + except OSError: + return None + + if not api_key: + return None + + return api_key, api_key_path.resolve() + + +def _core_process_paths(args: list[str], cwd: Path) -> tuple[Path, Path] | None: + try: + jar_index = args.index("-jar") + except ValueError: + return None + + if len(args) <= jar_index + 2: + return None + + jar_path = _resolve_process_path(args[jar_index + 1], cwd) + jar_name = jar_path.name.lower() + if not ( + jar_name.endswith(".jar") + and (jar_name.startswith("qortium") or jar_name.startswith("qortal")) + ): + return None + + settings_path = _resolve_process_path(args[jar_index + 2], cwd) + return jar_path, settings_path + + +def _iter_proc_dirs(proc_root: Path) -> list[Path]: + try: + entries = list(proc_root.iterdir()) + except OSError: + return [] + return [entry for entry in entries if entry.is_dir() and entry.name.isdigit()] + + +def _read_process_args(proc_dir: Path) -> list[str]: + raw = proc_dir.joinpath("cmdline").read_bytes() + return [part.decode("utf-8", errors="replace") for part in raw.split(b"\0") if part] + + +def _read_process_cwd(proc_dir: Path) -> Path: + return Path(os.readlink(proc_dir / "cwd")).resolve() + + +def _dedupe_candidates(candidates: list[LocalCoreApiKey]) -> list[LocalCoreApiKey]: + deduped: dict[Path, LocalCoreApiKey] = {} + for candidate in candidates: + deduped[candidate.api_key_path] = candidate + return list(deduped.values()) + + +def detect_local_core_api_key( + base_url: str, + proc_root: Path = Path("/proc"), +) -> LocalCoreApiKey | None: + target_port = _endpoint_port(base_url) + if target_port is None: + return None + + matched_candidates: list[LocalCoreApiKey] = [] + fallback_candidates: list[LocalCoreApiKey] = [] + + for proc_dir in _iter_proc_dirs(proc_root): + try: + pid = int(proc_dir.name) + args = _read_process_args(proc_dir) + cwd = _read_process_cwd(proc_dir) + process_paths = _core_process_paths(args, cwd) + if process_paths is None: + continue + + jar_path, raw_settings_path = process_paths + settings_path = _effective_settings_path(raw_settings_path, cwd) + settings = _read_json_file(settings_path) + api_key_directory = _settings_api_key_directory(settings_path, cwd) + api_key = _read_api_key(api_key_directory) + if api_key is None: + continue + + candidate = LocalCoreApiKey( + api_key=api_key[0], + api_key_path=api_key[1], + api_key_directory=api_key_directory.resolve(), + cwd=cwd, + jar_path=jar_path, + pid=pid, + settings_path=settings_path, + ) + + api_port = _settings_api_port(settings) + if api_port == target_port: + matched_candidates.append(candidate) + elif api_port is None: + fallback_candidates.append(candidate) + except Exception: + continue + + matched_candidates = _dedupe_candidates(matched_candidates) + if len(matched_candidates) == 1: + return matched_candidates[0] + if matched_candidates: + return None + + fallback_candidates = _dedupe_candidates(fallback_candidates) + return fallback_candidates[0] if len(fallback_candidates) == 1 else None diff --git a/qortium_cli/services.py b/qortium_cli/services.py index b405336..7a8af83 100644 --- a/qortium_cli/services.py +++ b/qortium_cli/services.py @@ -64,6 +64,39 @@ def generate_api_key_via_node(base_url: str, timeout_seconds: int, existing_api_ return api_key +def check_node_connection(base_url: str, timeout_seconds: int) -> tuple[bool, str]: + timeout = min(max(1, int(timeout_seconds)), 5) + try: + response = requests.get( + f"{base_url}/admin/status", + headers={"Accept": "application/json,text/plain"}, + timeout=timeout, + ) + response.raise_for_status() + except requests.exceptions.Timeout: + return False, f"Timed out after {timeout} seconds while checking /admin/status." + except requests.exceptions.ConnectionError as exc: + return False, f"Connection failed while checking /admin/status: {exc}" + except requests.exceptions.HTTPError as exc: + detail = http_error_detail(exc) + status_code = exc.response.status_code if exc.response is not None else 0 + return False, f"/admin/status returned HTTP {status_code}: {detail}" + except requests.exceptions.RequestException as exc: + return False, f"Connection check failed while checking /admin/status: {exc}" + + try: + status = response.json() + except Exception: + return False, "/admin/status responded, but did not return node status JSON." + + if not isinstance(status, dict): + return False, "/admin/status responded, but did not return node status JSON." + + height = status.get("height", "Unknown") + sync_percent = status.get("syncPercent", "Unknown") + return True, f"Node API responded. Height: {height}; sync: {sync_percent}%." + + def qortal_public_key_from_private(base_url: str, private_key: str, timeout_seconds: int) -> str: response = requests.post( f"{base_url}/utils/publickey", diff --git a/qortium_cli/setup_wizard.py b/qortium_cli/setup_wizard.py index 091fe25..8613384 100644 --- a/qortium_cli/setup_wizard.py +++ b/qortium_cli/setup_wizard.py @@ -3,9 +3,11 @@ from dataclasses import replace from qortium_cli.constants import C_TEXT, RESET +from qortium_cli.core_detection import detect_local_core_api_key from qortium_cli.crypto import derive_private_key_from_seed_phrase from qortium_cli.models import AccountSettings, AppContext, EndpointSettings from qortium_cli.services import ( + check_node_connection, qortal_address_from_public, qortal_primary_name_for_address, qortal_public_key_from_private, @@ -25,6 +27,13 @@ warn, ) from qortium_cli.validators import is_placeholder, normalize_node_url +from qortium_cli.wallet_backup import ( + default_wallet_backup_path, + generate_new_wallet_backup, + normalize_wallet_file_path, + private_key_from_wallet_file, + write_wallet_backup, +) def current_endpoint_values_ready(ctx: AppContext) -> bool: @@ -52,16 +61,49 @@ def _prompt_private_key() -> str: print() print_option("1", "Use private key") print_option("2", "Use seed phrase") - mode = read_menu_choice("Choose key input mode [1/2]: ").strip() or "1" - if mode not in {"1", "2"}: + print_option("3", "Use wallet file") + print_option("4", "New wallet file") + mode = read_menu_choice("Choose key input mode [1/2/3/4]: ").strip() or "1" + if mode not in {"1", "2", "3", "4"}: mode = "1" if mode == "1": private_key = prompt_secret("Private key: ").strip() - else: + elif mode == "2": seed_phrase = prompt_secret("Seed phrase: ").strip() private_key = derive_private_key_from_seed_phrase(seed_phrase) ok("Derived private key from seed phrase.") + elif mode == "3": + raw_wallet_path = prompt_str("Wallet file path: ") + if not raw_wallet_path.strip(): + raise RuntimeError("Wallet file path is empty.") + wallet_path = normalize_wallet_file_path(raw_wallet_path) + wallet_password = prompt_secret("Wallet password: ") + private_key = private_key_from_wallet_file(wallet_path, wallet_password) + ok("Unlocked Qortium Home wallet file.") + elif mode == "4": + wallet_password = prompt_secret("New wallet password: ") + confirm_password = prompt_secret("Confirm wallet password: ") + if wallet_password != confirm_password: + raise RuntimeError("Wallet passwords do not match.") + generated_wallet = generate_new_wallet_backup(wallet_password) + wallet_name = prompt_str("Wallet name [wallet]: ", "wallet").strip() or "wallet" + default_path = default_wallet_backup_path( + generated_wallet.address, + wallet_name=wallet_name, + ) + raw_output_path = prompt_str( + f"Wallet backup path [{default_path}]: ", + str(default_path), + ) + output_path = write_wallet_backup( + normalize_wallet_file_path(raw_output_path), + generated_wallet.wallet, + ) + ok("Created encrypted Qortium Home wallet file.") + print(C_TEXT + f"Wallet file: {output_path}" + RESET) + print(C_TEXT + f"Account: {generated_wallet.address}" + RESET) + private_key = generated_wallet.private_key if not private_key: raise RuntimeError("Private key is empty.") @@ -93,6 +135,71 @@ def _account_from_private_key(ctx: AppContext, private_key: str, api_key: str) - ) +def _confirm_endpoint_connection(base_url: str, timeout_seconds: int) -> bool: + connected, detail = check_node_connection(base_url, timeout_seconds) + if connected: + ok(detail or "Connected to node API.") + return True + + warn(f"Node is not connected at {base_url}.") + if detail: + warn(detail) + + while True: + print() + print_option("1", "Enter a different endpoint URL") + print_option("2", "Continue with this endpoint anyway") + choice = read_menu_choice("Choose an option: ").strip() + + if choice == "1": + return False + if choice == "2": + warn("Continuing with endpoint even though the connection check failed.") + return True + + warn("Unknown option.") + + +def _prompt_endpoint_url_with_connection_check( + prompt: str, + default_url: str, + timeout_seconds: int, +) -> str: + while True: + raw_url = prompt_str(prompt, default_url) + try: + base_url = normalize_node_url(raw_url) + except ValueError as exc: + warn(str(exc)) + continue + + if _confirm_endpoint_connection(base_url, timeout_seconds): + return base_url + + +def _detect_local_core_api_key(ctx: AppContext): + suggestion = detect_local_core_api_key(ctx.endpoint.base_url) + if suggestion: + ok(f"Detected local Core API key at: {suggestion.api_key_path}") + return suggestion + + +def _prompt_required_api_key(ctx: AppContext, existing_api_key: str = "") -> str: + suggestion = _detect_local_core_api_key(ctx) + if suggestion: + api_key = prompt_secret( + "API key (X-API-KEY) [detected local Core key] (press Enter to use): " + ).strip() + return api_key or suggestion.api_key + + has_existing_api_key = bool(existing_api_key) and (not is_placeholder(existing_api_key)) + if has_existing_api_key: + api_key = prompt_secret("API key (X-API-KEY) (press Enter to keep current): ").strip() + return api_key or existing_api_key + + return prompt_secret("API key (X-API-KEY): ").strip() + + def configure_endpoint_url(ctx: AppContext) -> None: current_url = ctx.endpoint.base_url while True: @@ -102,13 +209,16 @@ def configure_endpoint_url(ctx: AppContext) -> None: ) try: base_url = normalize_node_url(raw_url) - break except ValueError as exc: warn(str(exc)) + continue - if base_url == current_url: - warn("Endpoint URL unchanged.") - return + if base_url == current_url: + warn("Endpoint URL unchanged.") + return + + if _confirm_endpoint_connection(base_url, ctx.endpoint.timeout_seconds): + break ctx.endpoint = replace(ctx.endpoint, base_url=base_url) write_endpoint_file(ctx.settings_dir, ctx.endpoint) @@ -132,7 +242,19 @@ def configure_timeout(ctx: AppContext) -> None: def configure_api_key(ctx: AppContext) -> None: - api_key = prompt_secret("New API key (press Enter to cancel): ").strip() + suggestion = _detect_local_core_api_key(ctx) + if suggestion: + api_key = prompt_secret( + "New API key [detected local Core key] " + "(press Enter to use, type /cancel to cancel): " + ).strip() + if not api_key: + api_key = suggestion.api_key + elif api_key.lower() == "/cancel": + api_key = "" + else: + api_key = prompt_secret("New API key (press Enter to cancel): ").strip() + if not api_key: warn("API key unchanged.") return @@ -152,6 +274,46 @@ def configure_wallet_identity(ctx: AppContext) -> None: print(C_TEXT + f"Account: {ctx.account.account_address}" + RESET) +def run_initial_setup(ctx: AppContext) -> None: + print_setup_banner("First Run Setup") + print(C_TEXT + "Let's create endpoint.py and config.py in:" + RESET) + print(C_TEXT + f" {ctx.settings_dir}" + RESET) + print() + + current_url = ctx.endpoint.base_url + current_timeout = ctx.endpoint.timeout_seconds + + base_url = _prompt_endpoint_url_with_connection_check( + f"Endpoint URL [{current_url}] (press Enter to use default): ", + current_url, + current_timeout, + ) + + timeout = prompt_int( + f"Timeout seconds [{current_timeout}] (press Enter to use default): ", + default=current_timeout, + minimum=1, + ) + endpoint = EndpointSettings(base_url=base_url, timeout_seconds=timeout) + write_endpoint_file(ctx.settings_dir, endpoint) + ctx.endpoint = endpoint + + existing_api_key = (ctx.account.api_key or "").strip() + api_key = _prompt_required_api_key(ctx, existing_api_key) + + if not api_key: + raise RuntimeError("API key is required to continue setup.") + + private_key = _prompt_private_key() + account = _account_from_private_key(ctx, private_key, api_key) + write_config_file(ctx.settings_dir, account) + ctx.account = account + + ok("Setup complete. endpoint.py and config.py have been created.") + print(C_TEXT + f"Settings directory: {ctx.settings_dir}" + RESET) + pause() + + def run_reconfigure_menu(ctx: AppContext) -> None: while True: print_setup_banner("Reconfigure") @@ -167,6 +329,7 @@ def run_reconfigure_menu(ctx: AppContext) -> None: print_option("2", "Change request timeout") print_option("3", "Change API key") print_option("4", "Change wallet / account") + print_option("9", "Run initial setup") print_option("0", "Back") choice = read_menu_choice("Choose an option: ") @@ -190,6 +353,9 @@ def run_reconfigure_menu(ctx: AppContext) -> None: configure_wallet_identity(ctx) pause() continue + if choice == "9": + run_initial_setup(ctx) + return except Exception as exc: error("Reconfiguration failed:") print(str(exc)) @@ -208,51 +374,4 @@ def configure_first_run_files(ctx: AppContext, force: bool = False) -> None: if current_endpoint_values_ready(ctx) and current_config_values_ready(ctx): return - print_setup_banner("First Run Setup") - print(C_TEXT + "Let's create endpoint.py and config.py in:" + RESET) - print(C_TEXT + f" {ctx.settings_dir}" + RESET) - print() - - current_url = ctx.endpoint.base_url - current_timeout = ctx.endpoint.timeout_seconds - - while True: - raw_url = prompt_str( - f"Endpoint URL [{current_url}] (press Enter to use default): ", - current_url, - ) - try: - base_url = normalize_node_url(raw_url) - break - except ValueError as exc: - warn(str(exc)) - - timeout = prompt_int( - f"Timeout seconds [{current_timeout}] (press Enter to use default): ", - default=current_timeout, - minimum=1, - ) - endpoint = EndpointSettings(base_url=base_url, timeout_seconds=timeout) - write_endpoint_file(ctx.settings_dir, endpoint) - ctx.endpoint = endpoint - - existing_api_key = (ctx.account.api_key or "").strip() - has_existing_api_key = bool(existing_api_key) and (not is_placeholder(existing_api_key)) - if has_existing_api_key: - api_key = prompt_secret("API key (X-API-KEY) (press Enter to keep current): ").strip() - if not api_key: - api_key = existing_api_key - else: - api_key = prompt_secret("API key (X-API-KEY): ").strip() - - if not api_key: - raise RuntimeError("API key is required to continue setup.") - - private_key = _prompt_private_key() - account = _account_from_private_key(ctx, private_key, api_key) - write_config_file(ctx.settings_dir, account) - ctx.account = account - - ok("Setup complete. endpoint.py and config.py have been created.") - print(C_TEXT + f"Settings directory: {ctx.settings_dir}" + RESET) - pause() + run_initial_setup(ctx) diff --git a/qortium_cli/tools.py b/qortium_cli/tools.py index 640ece6..4fcf98b 100644 --- a/qortium_cli/tools.py +++ b/qortium_cli/tools.py @@ -1,9 +1,7 @@ from __future__ import annotations -import base64 import datetime import hashlib -import json import traceback from decimal import Decimal, InvalidOperation from pathlib import Path @@ -11,8 +9,19 @@ import requests -from qortium_cli.constants import BOLD, CHAT_USER_COLORS, QDN_SERVICES, RESET -from qortium_cli.crypto import b58decode, b58encode, is_base58, to_base58_pubkey +from qortium_cli.chat_format import ( + DEFAULT_REACTION_CATEGORIES, + DEFAULT_REACTION_OPTIONS, + MessageReactionSummary, + MessageThread, + build_chat_message_text, + build_message_reaction_index, + build_message_threads, + build_reaction_message_text, + decode_chat_message, +) +from qortium_cli.constants import BOLD, CHAT_USER_COLORS, C_TEXT, QDN_SERVICES, RESET +from qortium_cli.crypto import b58encode, to_base58_pubkey from qortium_cli.models import AppContext, ToolPlugin from qortium_cli.services import ( build_arbitrary_delete, @@ -78,6 +87,21 @@ APPROVAL_THRESHOLDS = ("NONE", "ONE", "PCT20", "PCT40", "PCT60", "PCT80", "PCT100") ATOMIC_UNITS = Decimal("100000000") QDN_RESOURCE_PAGE_SIZE = 10 +WHATS_NEW_ENTRIES = ( + ( + "v0.3.0", + ( + "Chat timeline now understands Qortium Chat reply, edit, and reaction envelopes.", + "Chat commands added: /reply, /edit, /react, /help, and /?.", + "Reply and reaction selection groups messages by sender to reduce long lists.", + "Reaction picker supports add/remove flows and emoji categories.", + "Setup can check endpoints, detect local Core API keys, import wallet files, " + "and create encrypted wallet files.", + "Register Name can list owned names and update an existing name.", + "Startup checks GitHub releases and lists newer stable or qualifying prerelease builds.", + ), + ), +) def _chat_user_color(identity: str) -> str: @@ -172,63 +196,452 @@ def _format_invite_expiry(expiry_ms: Any) -> str: return _format_chat_timestamp(expiry) -def _extract_doc_text(node: Any) -> str: - if isinstance(node, dict): - node_type = str(node.get("type", "")) - if node_type == "text": - return str(node.get("text", "")) +def _chat_message_signature(message: Dict[str, Any] | MessageThread) -> str: + if isinstance(message, MessageThread): + message = dict(message.original) + return str(message.get("signature") or "").strip() - parts = [] - for child in node.get("content", []): - parts.append(_extract_doc_text(child)) - if node_type == "paragraph": - return "".join(parts) + "\n" - return "".join(parts) +def _chat_message_sender(message: Dict[str, Any]) -> str: + return str(message.get("sender") or "").strip() - if isinstance(node, list): - return "".join(_extract_doc_text(child) for child in node) - return "" +def _chat_sender_label(message: Dict[str, Any]) -> str: + sender_address = _chat_message_sender(message) + return str(message.get("senderName") or sender_address or "Unknown").strip() + + +def _chat_identity_display(message: Dict[str, Any]) -> str: + sender_address = _chat_message_sender(message) + sender_label = _chat_sender_label(message) + return _colorize_chat_identity(sender_label, sender_address or sender_label) + + +def _chat_message_snippet(thread: MessageThread) -> str: + decoded = decode_chat_message(thread.latest) + for line in decoded.body.splitlines(): + clean = line.strip() + if not clean: + continue + if len(clean) > 96: + return clean[:93].rstrip() + "..." + return clean + return "[no text]" + + +def _chat_reply_reference(thread: MessageThread, threads_by_signature: Dict[str, MessageThread]) -> str: + decoded = decode_chat_message(thread.latest) + if decoded.replied_to: + return decoded.replied_to + original_decoded = decode_chat_message(thread.original) + if original_decoded.replied_to: + return original_decoded.replied_to -def _decode_chat_data(data: Any, encoding: Any) -> str: - if not isinstance(data, str) or not data: + reference = str(thread.original.get("chatReference") or "").strip() + if not reference: return "" - enc = str(encoding or "").upper() - decoded_bytes = None + referenced_thread = threads_by_signature.get(reference) + if not referenced_thread: + return "" + + if _chat_message_sender(dict(referenced_thread.original)) == _chat_message_sender(dict(thread.original)): + return "" + + return reference + + +def _format_chat_reactions(reactions: tuple[MessageReactionSummary, ...]) -> str: + return " Reactions: " + " ".join( + f"{reaction.content} {reaction.count}" for reaction in reactions + ) + + +def _normalize_chat_message_input(raw: str) -> str: + if raw.startswith("//"): + return raw[1:] + return raw + + +def _print_chat_command_help() -> None: + print_section("Chat Commands") + print("/reply Reply to a recent message.") + print("/react React to a recent message.") + print("/edit Edit one of your own recent text messages.") + print("/help Show this help.") + print("/? Show this help.") + print("/quit Leave chat.") + print("//text Send a message that starts with /.") + + +def _replyable_chat_threads(messages: List[Dict[str, Any]]) -> List[MessageThread]: + replyable: List[MessageThread] = [] + for thread in build_message_threads(messages): + original = dict(thread.original) + if not _chat_message_signature(original): + continue + if bool(original.get("_unconfirmed", False)): + continue + replyable.append(thread) + return replyable + + +def _chat_thread_timestamp(thread: MessageThread) -> int: + try: + return int(thread.original.get("timestamp") or 0) + except (TypeError, ValueError): + return 0 + + +def _replyable_chat_user_groups( + messages: List[Dict[str, Any]], +) -> List[tuple[str, str, List[MessageThread]]]: + groups: Dict[str, List[MessageThread]] = {} + labels: Dict[str, str] = {} + + for thread in _replyable_chat_threads(messages): + original = dict(thread.original) + key = _chat_message_sender(original) or _chat_sender_label(original) + groups.setdefault(key, []).append(thread) + labels.setdefault(key, _chat_sender_label(original)) + + user_groups: List[tuple[str, str, List[MessageThread]]] = [] + for key, threads in groups.items(): + threads.sort(key=_chat_thread_timestamp, reverse=True) + user_groups.append((key, labels[key], threads)) + + user_groups.sort(key=lambda group: _chat_thread_timestamp(group[2][0]), reverse=True) + return user_groups + + +def _format_replyable_chat_user(label: str, threads: List[MessageThread]) -> str: + count = len(threads) + count_label = "1 message" if count == 1 else f"{count} messages" + return f"{label} ({count_label}) - latest: {_chat_message_snippet(threads[0])}" + - if enc == "BASE64": +def _format_replyable_chat_thread(thread: MessageThread) -> str: + timestamp = _format_chat_timestamp(thread.original.get("timestamp")) + snippet = _chat_message_snippet(thread) + edited_label = " [edited]" if thread.revisions else "" + return f"{timestamp} - {snippet}{edited_label}" + + +def _select_chat_thread_by_sender( + messages: List[Dict[str, Any]], + *, + sender_section_title: str, + sender_prompt: str, + message_section_prefix: str, + message_prompt: str, + empty_warning: str, +) -> MessageThread | None: + user_groups = _replyable_chat_user_groups(messages) + if not user_groups: + warn(empty_warning) + return None + + while True: + print_section(sender_section_title) + for index, (_, label, threads) in enumerate(user_groups, start=1): + print_option(str(index), _format_replyable_chat_user(label, threads)) + print_option("0", "Cancel") + + choice = read_menu_choice(sender_prompt) + if choice == "0": + return None try: - decoded_bytes = base64.b64decode(data) - except Exception: - return data - elif enc == "BASE58" or is_base58(data): + selected_index = int(choice) - 1 + if selected_index < 0: + raise IndexError + _, label, sender_threads = user_groups[selected_index] + except (ValueError, IndexError): + warn("Unknown option.") + continue + + print_section(f"{message_section_prefix} {label}") + for index, thread in enumerate(sender_threads, start=1): + print_option(str(index), _format_replyable_chat_thread(thread)) + print_option("0", "Back") + + while True: + choice = read_menu_choice(message_prompt) + if choice == "0": + break + try: + selected_index = int(choice) - 1 + if selected_index < 0: + raise IndexError + return sender_threads[selected_index] + except (ValueError, IndexError): + warn("Unknown option.") + + +def _select_replyable_chat_thread(messages: List[Dict[str, Any]]) -> MessageThread | None: + return _select_chat_thread_by_sender( + messages, + sender_section_title="Reply Sender", + sender_prompt="Choose sender to reply to: ", + message_section_prefix="Messages From", + message_prompt="Choose message to reply to: ", + empty_warning="No replyable messages found in the current chat history.", + ) + + +def _select_reactable_chat_thread(messages: List[Dict[str, Any]]) -> MessageThread | None: + return _select_chat_thread_by_sender( + messages, + sender_section_title="Reaction Sender", + sender_prompt="Choose sender to react to: ", + message_section_prefix="Messages From", + message_prompt="Choose message to react to: ", + empty_warning="No reactable messages found in the current chat history.", + ) + + +def _select_chat_reaction( + reactions: tuple[MessageReactionSummary, ...], +) -> tuple[str, bool] | None: + self_reactions = _self_reaction_contents(reactions) + if self_reactions: + action = _select_reaction_action(self_reactions) + if action is None: + return None + else: + action = "add" + + if action == "remove": + reaction = _select_reaction_to_remove(self_reactions) + return (reaction, False) if reaction else None + + reaction = _select_reaction_to_add(self_reactions) + return (reaction, True) if reaction else None + + +def _self_reaction_contents(reactions: tuple[MessageReactionSummary, ...]) -> set[str]: + return {reaction.content for reaction in reactions if reaction.reacted_by_self} + + +def _format_reaction_list(reactions: set[str]) -> str: + return " ".join(sorted(reactions, key=_reaction_sort_key)) + + +def _reaction_sort_key(reaction: str) -> tuple[int, str]: + try: + index = DEFAULT_REACTION_OPTIONS.index(reaction) + except ValueError: + index = len(DEFAULT_REACTION_OPTIONS) + return index, reaction + + +def _select_reaction_action(self_reactions: set[str]) -> str | None: + print_section("Reaction Action") + print(f"Current reactions: {_format_reaction_list(self_reactions)}") + print_option("1", "Add reaction") + print_option("2", "Remove reaction") + print_option("0", "Cancel") + + while True: + choice = read_menu_choice("Choose reaction action: ") + if choice == "0": + return None + if choice == "1": + return "add" + if choice == "2": + return "remove" + warn("Unknown option.") + + +def _reaction_categories_for_add(self_reactions: set[str]) -> list[tuple[str, tuple[str, ...]]]: + return [ + (label, tuple(reaction for reaction in reactions if reaction not in self_reactions)) + for label, reactions in DEFAULT_REACTION_CATEGORIES + if any(reaction not in self_reactions for reaction in reactions) + ] + + +def _select_reaction_to_add(self_reactions: set[str]) -> str | None: + categories = _reaction_categories_for_add(self_reactions) + if not categories: + warn("No additional reactions are available.") + return None + + while True: + print_section("Reaction Category") + for index, (label, reactions) in enumerate(categories, start=1): + print_option(str(index), f"{label} ({len(reactions)})") + print_option("0", "Cancel") + + choice = read_menu_choice("Choose reaction category: ") + if choice == "0": + return None try: - decoded_bytes = b58decode(data) - except Exception: - return data + selected_index = int(choice) - 1 + if selected_index < 0: + raise IndexError + label, category_reactions = categories[selected_index] + except (ValueError, IndexError): + warn("Unknown option.") + continue + + print_section(label) + for index, reaction in enumerate(category_reactions, start=1): + print_option(str(index), reaction) + print_option("0", "Back") - if decoded_bytes is None: - return data + while True: + choice = read_menu_choice("Choose reaction: ") + if choice == "0": + break + try: + selected_index = int(choice) - 1 + if selected_index < 0: + raise IndexError + return category_reactions[selected_index] + except (ValueError, IndexError): + warn("Unknown option.") + + +def _select_reaction_to_remove(self_reactions: set[str]) -> str | None: + removable_reactions = tuple(sorted(self_reactions, key=_reaction_sort_key)) + if not removable_reactions: + warn("No reactions to remove.") + return None - text = decoded_bytes.decode("utf-8", errors="replace") - stripped = text.strip() + print_section("Remove Reaction") + for index, reaction in enumerate(removable_reactions, start=1): + print_option(str(index), reaction) + print_option("0", "Cancel") - if stripped.startswith("{") and stripped.endswith("}"): + while True: + choice = read_menu_choice("Choose reaction to remove: ") + if choice == "0": + return None try: - payload = json.loads(stripped) - if isinstance(payload, dict): - message_doc = payload.get("messageText") - if isinstance(message_doc, dict): - doc_text = _extract_doc_text(message_doc).strip() - if doc_text: - return doc_text - except Exception: - pass + selected_index = int(choice) - 1 + if selected_index < 0: + raise IndexError + return removable_reactions[selected_index] + except (ValueError, IndexError): + warn("Unknown option.") + + +def _editable_chat_threads(ctx: AppContext, messages: List[Dict[str, Any]]) -> List[MessageThread]: + editable: List[MessageThread] = [] + for thread in build_message_threads(messages): + original = dict(thread.original) + latest = dict(thread.latest) + if _chat_message_sender(original) != ctx.account.account_address: + continue + if not _chat_message_signature(original): + continue + if bool(original.get("_unconfirmed", False)): + continue + if decode_chat_message(latest).kind != "text": + continue + editable.append(thread) + return editable + - return text +def _format_editable_chat_thread(thread: MessageThread) -> str: + timestamp = _format_chat_timestamp(thread.original.get("timestamp")) + snippet = _chat_message_snippet(thread) + edited_label = " [edited]" if thread.revisions else "" + return f"{timestamp} - {snippet}{edited_label}" + + +def _select_editable_chat_thread(ctx: AppContext, messages: List[Dict[str, Any]]) -> MessageThread | None: + editable = _editable_chat_threads(ctx, messages) + if not editable: + warn("No editable messages found in the current chat history.") + return None + + print_section("Editable Messages") + for index, thread in enumerate(editable, start=1): + print_option(str(index), _format_editable_chat_thread(thread)) + print_option("0", "Cancel") + + while True: + choice = read_menu_choice("Choose message to edit: ") + if choice == "0": + return None + try: + selected_index = int(choice) - 1 + if selected_index < 0: + raise IndexError + return editable[selected_index] + except (ValueError, IndexError): + warn("Unknown option.") + + +def _run_chat_reply_command(ctx: AppContext, messages: List[Dict[str, Any]]) -> Any | None: + thread = _select_replyable_chat_thread(messages) + if not thread: + return None + + print() + print("Replying to:") + print(f" {_chat_sender_label(dict(thread.original))}: {_chat_message_snippet(thread)}") + reply = prompt_str("Reply message (Enter to cancel): ", "").strip() + if not reply: + warn("Reply cancelled.") + return None + + target_signature = _chat_message_signature(dict(thread.original)) + message_text = build_chat_message_text( + _normalize_chat_message_input(reply), + target_signature, + ) + return _send_chat_message(ctx, message_text) + + +def _run_chat_reaction_command(ctx: AppContext, messages: List[Dict[str, Any]]) -> Any | None: + thread = _select_reactable_chat_thread(messages) + if not thread: + return None + + target_signature = _chat_message_signature(dict(thread.original)) + reactions_by_signature = build_message_reaction_index( + messages, + self_address=ctx.account.account_address, + ) + + print() + print("Reacting to:") + print(f" {_chat_sender_label(dict(thread.original))}: {_chat_message_snippet(thread)}") + + selection = _select_chat_reaction(reactions_by_signature.get(target_signature, ())) + if not selection: + warn("Reaction cancelled.") + return None + + reaction, content_state = selection + message_text = build_reaction_message_text(reaction, content_state) + return _send_chat_message(ctx, message_text, chat_reference=target_signature) + + +def _run_chat_edit_command(ctx: AppContext, messages: List[Dict[str, Any]]) -> Any | None: + thread = _select_editable_chat_thread(ctx, messages) + if not thread: + return None + + print() + print("Current message:") + print(f" {_chat_message_snippet(thread)}") + replacement = prompt_str("New message (Enter to cancel): ", "").strip() + if not replacement: + warn("Edit cancelled.") + return None + + original = dict(thread.original) + original_signature = _chat_message_signature(original) + replied_to = decode_chat_message(original).replied_to + message_text = build_chat_message_text( + _normalize_chat_message_input(replacement), + replied_to, + ) + return _send_chat_message(ctx, message_text, chat_reference=original_signature) def _get_chat_fee_decimal(ctx: AppContext) -> Decimal: @@ -599,37 +1012,67 @@ def _print_chat_timeline(messages: List[Dict[str, Any]]) -> None: warn("No messages found for this group.") return - print_section(f"Chat Timeline ({len(messages)} messages)") + threads = build_message_threads(messages) + reactions_by_signature = build_message_reaction_index(messages) + if not threads: + warn("No displayable chat messages found for this group.") + return + + threads_by_signature = { + _chat_message_signature(thread): thread + for thread in threads + if _chat_message_signature(thread) + } + + print_section(f"Chat Timeline ({len(threads)} messages)") print() - for message in messages: - timestamp = _format_chat_timestamp(message.get("timestamp")) - sender_address = str(message.get("sender") or "").strip() - sender_label = str(message.get("senderName") or sender_address or "Unknown").strip() - sender = _colorize_chat_identity(sender_label, sender_address or sender_label) + for thread in threads: + original = dict(thread.original) + latest = dict(thread.latest) + decoded = decode_chat_message(latest) - recipient = str(message.get("recipient") or "").strip() + timestamp = _format_chat_timestamp(original.get("timestamp")) + sender = _chat_identity_display(original) + + recipient = str(original.get("recipient") or "").strip() recipient_label = _colorize_chat_identity(recipient, recipient) if recipient else "" - body = _decode_chat_data(message.get("data"), message.get("encoding")) - is_encrypted = bool(message.get("isEncrypted", False)) - is_unconfirmed = bool(message.get("_unconfirmed", False)) + is_encrypted = bool(latest.get("isEncrypted", False)) or decoded.kind == "encrypted" + is_unconfirmed = bool(latest.get("_unconfirmed", False) or original.get("_unconfirmed", False)) target_label = f" -> {recipient_label}" if recipient else "" encryption_label = " [enc]" if is_encrypted else "" + edited_label = " [edited]" if thread.revisions else "" unconfirmed_label = " [mempool]" if is_unconfirmed else "" - print(f"[{timestamp}] {sender}{target_label}{encryption_label}{unconfirmed_label}") + print( + f"[{timestamp}] {sender}{target_label}" + f"{encryption_label}{edited_label}{unconfirmed_label}" + ) - if body.strip(): - for line in body.splitlines(): + replied_to = _chat_reply_reference(thread, threads_by_signature) + referenced_thread = threads_by_signature.get(replied_to) + if referenced_thread: + reply_sender = _chat_sender_label(dict(referenced_thread.original)) + print(f" > reply to {reply_sender}: {_chat_message_snippet(referenced_thread)}") + elif replied_to: + print(" > reply to: unavailable") + + if decoded.body.strip(): + for line in decoded.body.splitlines(): print(f" {line}") else: print(" [no text payload]") + signature = _chat_message_signature(thread) + reactions = reactions_by_signature.get(signature, ()) + if reactions: + print(_format_chat_reactions(reactions)) + print() -def _send_chat_message(ctx: AppContext, message: str) -> Any: +def _send_chat_message(ctx: AppContext, message: str, *, chat_reference: str = "") -> Any: ensure_wallet_config_ready(ctx) sender_pub = to_base58_pubkey(ctx.account.public_key) @@ -649,6 +1092,9 @@ def _send_chat_message(ctx: AppContext, message: str) -> Any: "isText": 1, "isEncrypted": 0, } + if chat_reference: + payload["chatReference"] = chat_reference + unsigned_tx = build_chat(ctx, payload, session) print("[2/4] Computing nonce (can take 10s-180s on lower-balance accounts)...", flush=True) @@ -680,7 +1126,6 @@ def run_chat_room(ctx: AppContext) -> None: print_stat("Fee", d8(_get_chat_fee_decimal(ctx))) print() print("Type a message and press Enter to send.") - print("Use /quit to leave chat. Empty input refreshes.") print() try: @@ -696,14 +1141,82 @@ def run_chat_room(ctx: AppContext) -> None: pause() return + print("/? for help") raw = prompt_str("message > ", "") - if raw.strip() == "/quit": + stripped = raw.strip() + command = stripped.lower() + if stripped == "": + continue + if command == "/quit": return - if raw.strip() == "": + if command in {"/help", "/?"}: + _print_chat_command_help() + pause() + continue + if command == "/reply": + try: + result = _run_chat_reply_command(ctx, messages) + if result is None: + pause() + continue + ok("Chat reply submitted.") + if ctx.debug: + print("Process response: " + str(result)) + input("Press Enter to refresh chat...") + except Exception as exc: + error("Failed to reply to chat message:") + if _is_node_unreachable_error(exc): + _print_node_unreachable_hint(ctx) + else: + print(pretty_exception(exc)) + _print_debug_traceback(ctx, exc) + pause() + continue + if command == "/react": + try: + result = _run_chat_reaction_command(ctx, messages) + if result is None: + pause() + continue + ok("Chat reaction submitted.") + if ctx.debug: + print("Process response: " + str(result)) + input("Press Enter to refresh chat...") + except Exception as exc: + error("Failed to react to chat message:") + if _is_node_unreachable_error(exc): + _print_node_unreachable_hint(ctx) + else: + print(pretty_exception(exc)) + _print_debug_traceback(ctx, exc) + pause() + continue + if command == "/edit": + try: + result = _run_chat_edit_command(ctx, messages) + if result is None: + pause() + continue + ok("Chat edit submitted.") + if ctx.debug: + print("Process response: " + str(result)) + input("Press Enter to refresh chat...") + except Exception as exc: + error("Failed to edit chat message:") + if _is_node_unreachable_error(exc): + _print_node_unreachable_hint(ctx) + else: + print(pretty_exception(exc)) + _print_debug_traceback(ctx, exc) + pause() + continue + if stripped.startswith("/") and not stripped.startswith("//"): + warn("Unknown chat command. Type /help for commands, or use // to send a leading slash.") + pause() continue try: - result = _send_chat_message(ctx, raw) + result = _send_chat_message(ctx, _normalize_chat_message_input(stripped)) ok("Chat message submitted.") if ctx.debug: print("Process response: " + str(result)) @@ -963,9 +1476,9 @@ def tx_group_create(ctx: AppContext) -> None: def tx_name_register(ctx: AppContext) -> None: ensure_wallet_config_ready(ctx) - name = prompt_str("Name to register: ").strip() + name = prompt_str("Name to register (Enter to cancel): ").strip() if not name: - warn("Name cannot be empty.") + warn("Name registration cancelled.") return name_data = prompt_str("Name data [{}]: ", "{}") @@ -986,6 +1499,59 @@ def tx_name_register(ctx: AppContext) -> None: ) +def _select_owned_name_for_update(owned_names: List[str]) -> str | None: + print() + print_section("Choose Name") + for index, name in enumerate(owned_names, start=1): + print_option(str(index), name) + print_option("0", "Cancel") + choice = read_menu_choice("Choose a name: ") + if choice == "0": + warn("Name update cancelled.") + return None + + try: + selected_index = int(choice) - 1 + return owned_names[selected_index] + except (ValueError, IndexError): + warn("Unknown option.") + return None + + +def tx_name_update(ctx: AppContext, owned_names: List[str]) -> None: + ensure_wallet_config_ready(ctx) + selected_name = _select_owned_name_for_update(owned_names) + if not selected_name: + return + + new_name = prompt_str( + f"New name for '{selected_name}' [keep current]: ", + "", + ) + + new_data = prompt_str( + "New name data [keep current]: ", + "", + ) + + fee, tx_group_id = _prompt_tx_common_inputs(default_fee=Decimal("0")) + sender_pub = to_base58_pubkey(ctx.account.public_key) + + _submit_builder_transaction( + ctx, + "/names/update", + "UPDATE_NAME", + { + "ownerPublicKey": sender_pub, + "name": selected_name, + "newName": new_name.strip(), + "newData": new_data, + }, + fee, + tx_group_id, + ) + + def tool_chat_settings(ctx: AppContext) -> None: while True: print_banner(ctx.endpoint.base_url, "Chat Settings") @@ -1320,10 +1886,45 @@ def tool_groups(ctx: AppContext) -> None: def tool_register_name(ctx: AppContext) -> None: + ensure_wallet_config_ready(ctx) print_banner(ctx.endpoint.base_url, "Register Name") print_stat("Account", ctx.account.account_address) print() - tx_name_register(ctx) + + with make_session(ctx, include_api_key=False) as session: + owned_names = get_account_names( + ctx, + ctx.account.account_address, + session, + limit=500, + ) + + if not owned_names: + tx_name_register(ctx) + pause() + return + + print_section("Registered Names") + for name in owned_names: + print(C_TEXT + f"- {name}" + RESET) + print() + print_option("1", "Update name") + print_option("2", "New name") + print_option("0", "Back") + + choice = read_menu_choice("Choose an option: ") + if choice == "0": + return + if choice == "1": + tx_name_update(ctx, owned_names) + pause() + return + if choice == "2": + tx_name_register(ctx) + pause() + return + + warn("Unknown option.") pause() @@ -1922,6 +2523,55 @@ def tool_wallet(ctx: AppContext) -> None: pause() +def _print_whats_new_entry(version: str, bullets: tuple[str, ...]) -> None: + print_section(version) + for bullet in bullets: + print(C_TEXT + f"- {bullet}" + RESET) + + +def _tool_whats_new(ctx: AppContext) -> None: + while True: + print_banner(ctx.endpoint.base_url, "What's New?") + for index, (version, _) in enumerate(WHATS_NEW_ENTRIES, start=1): + print_option(str(index), version) + print_option("0", "Back") + + choice = read_menu_choice("Choose a version: ") + if choice == "0": + return + + try: + selected_index = int(choice) - 1 + if selected_index < 0: + raise IndexError + version, bullets = WHATS_NEW_ENTRIES[selected_index] + except (ValueError, IndexError): + warn("Unknown option.") + pause() + continue + + print_banner(ctx.endpoint.base_url, f"What's New? {version}") + _print_whats_new_entry(version, bullets) + pause() + + +def tool_help_info(ctx: AppContext) -> None: + while True: + print_banner(ctx.endpoint.base_url, "Help/Info") + print_option("1", "What's New?") + print_option("0", "Back") + + choice = read_menu_choice("Choose an option: ") + if choice == "0": + return + if choice == "1": + _tool_whats_new(ctx) + continue + + warn("Unknown option.") + pause() + + def build_tool_plugins() -> List[ToolPlugin]: tools = [ ToolPlugin("1", "Node", "Node status and admin controls", tool_node), @@ -1939,4 +2589,12 @@ def build_tool_plugins() -> List[ToolPlugin]: tool_qdn_resources, ) ) + tools.append( + ToolPlugin( + "8", + "Help/Info", + "Documentation and changelog", + tool_help_info, + ) + ) return tools diff --git a/qortium_cli/update_checker.py b/qortium_cli/update_checker.py new file mode 100644 index 0000000..8e00e99 --- /dev/null +++ b/qortium_cli/update_checker.py @@ -0,0 +1,274 @@ +from __future__ import annotations + +import json +import os +import re +import sys +import time +from dataclasses import dataclass +from functools import total_ordering +from pathlib import Path +from typing import Any, Iterable + +import requests + +from qortium_cli.constants import APP_VERSION, C_TEXT, DIM, RESET +from qortium_cli.ui import pause, print_option, print_section, warn + +GITHUB_RELEASES_URL = "https://api.github.com/repos/QortiumDev/Qortium-Python-CLI/releases" +GITHUB_REPO_URL = "https://github.com/QortiumDev/Qortium-Python-CLI" +UPDATE_STATE_FILENAME = "update_state.json" +UPDATE_CHECK_INTERVAL_SECONDS = 24 * 60 * 60 +UPDATE_CHECK_TIMEOUT_SECONDS = 3 + +_VERSION_RE = re.compile( + r"^v?(\d+)\.(\d+)\.(\d+)(?:[-_]?([A-Za-z][A-Za-z0-9]*)(?:[._-]?(\d+))?)?$" +) +_PRERELEASE_RANK = { + "dev": 0, + "alpha": 1, + "a": 1, + "beta": 2, + "b": 2, + "preview": 3, + "pre": 3, + "rc": 4, +} + + +@total_ordering +@dataclass(frozen=True) +class ParsedVersion: + major: int + minor: int + patch: int + prerelease_label: str = "" + prerelease_number: int = 0 + + def __lt__(self, other: object) -> bool: + if not isinstance(other, ParsedVersion): + return NotImplemented + + base = (self.major, self.minor, self.patch) + other_base = (other.major, other.minor, other.patch) + if base != other_base: + return base < other_base + + if self.is_prerelease != other.is_prerelease: + return self.is_prerelease + + return self._prerelease_key() < other._prerelease_key() + + @property + def is_prerelease(self) -> bool: + return bool(self.prerelease_label) + + def _prerelease_key(self) -> tuple[int, str, int]: + label = self.prerelease_label.lower() + return ( + _PRERELEASE_RANK.get(label, 99), + label, + self.prerelease_number, + ) + + +@dataclass(frozen=True) +class ReleaseInfo: + tag_name: str + version: ParsedVersion + prerelease: bool + name: str = "" + html_url: str = "" + published_at: str = "" + + @property + def is_prerelease(self) -> bool: + return self.prerelease or self.version.is_prerelease + + @property + def label(self) -> str: + kind = "prerelease" if self.is_prerelease else "stable" + return f"{self.tag_name} ({kind})" + + +def parse_version_tag(tag_name: str) -> ParsedVersion | None: + match = _VERSION_RE.match(tag_name.strip()) + if not match: + return None + + prerelease_label = (match.group(4) or "").lower() + prerelease_number = int(match.group(5) or "0") + return ParsedVersion( + major=int(match.group(1)), + minor=int(match.group(2)), + patch=int(match.group(3)), + prerelease_label=prerelease_label, + prerelease_number=prerelease_number, + ) + + +def release_from_github_payload(payload: dict[str, Any]) -> ReleaseInfo | None: + if payload.get("draft"): + return None + + tag_name = str(payload.get("tag_name") or "").strip() + version = parse_version_tag(tag_name) + if version is None: + return None + + return ReleaseInfo( + tag_name=tag_name, + version=version, + prerelease=bool(payload.get("prerelease")) or version.is_prerelease, + name=str(payload.get("name") or tag_name), + html_url=str(payload.get("html_url") or f"{GITHUB_REPO_URL}/releases/tag/{tag_name}"), + published_at=str(payload.get("published_at") or ""), + ) + + +def releases_from_github_payload(payloads: Iterable[dict[str, Any]]) -> list[ReleaseInfo]: + releases: list[ReleaseInfo] = [] + for payload in payloads: + release = release_from_github_payload(payload) + if release is not None: + releases.append(release) + return releases + + +def fetch_github_releases(timeout_seconds: int = UPDATE_CHECK_TIMEOUT_SECONDS) -> list[ReleaseInfo]: + response = requests.get( + GITHUB_RELEASES_URL, + headers={ + "Accept": "application/vnd.github+json", + "User-Agent": "qortium-cli-update-checker", + }, + timeout=timeout_seconds, + ) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, list): + return [] + return releases_from_github_payload( + item for item in payload if isinstance(item, dict) + ) + + +def select_update_offers( + current_version: str, + releases: Iterable[ReleaseInfo], +) -> list[ReleaseInfo]: + current = parse_version_tag(current_version) + if current is None: + return [] + + release_list = list(releases) + stable_releases = [release for release in release_list if not release.is_prerelease] + prereleases = [release for release in release_list if release.is_prerelease] + + newest_stable = max(stable_releases, key=lambda release: release.version, default=None) + newest_eligible_prerelease = max( + ( + release + for release in prereleases + if release.version > current + and (newest_stable is None or release.version > newest_stable.version) + ), + key=lambda release: release.version, + default=None, + ) + + offers: list[ReleaseInfo] = [] + if newest_eligible_prerelease is not None: + offers.append(newest_eligible_prerelease) + if newest_stable is not None and newest_stable.version > current: + offers.append(newest_stable) + return offers + + +def _state_path(settings_dir: Path) -> Path: + return settings_dir / UPDATE_STATE_FILENAME + + +def _read_update_state(settings_dir: Path) -> dict[str, Any]: + try: + raw = _state_path(settings_dir).read_text(encoding="utf-8") + state = json.loads(raw) + return state if isinstance(state, dict) else {} + except (OSError, json.JSONDecodeError): + return {} + + +def _write_update_state(settings_dir: Path, state: dict[str, Any]) -> None: + try: + settings_dir.mkdir(parents=True, exist_ok=True) + _state_path(settings_dir).write_text( + json.dumps(state, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + except OSError: + return + + +def should_check_for_updates(settings_dir: Path, now: float | None = None) -> bool: + state = _read_update_state(settings_dir) + last_checked_at = state.get("last_checked_at") + try: + last_checked = float(last_checked_at) + except (TypeError, ValueError): + return True + + now = time.time() if now is None else now + return now - last_checked >= UPDATE_CHECK_INTERVAL_SECONDS + + +def record_update_check(settings_dir: Path, status: str, now: float | None = None) -> None: + _write_update_state( + settings_dir, + { + "last_checked_at": time.time() if now is None else now, + "status": status, + }, + ) + + +def update_checks_enabled() -> bool: + disabled = str(os.environ.get("QORTIUM_CLI_NO_UPDATE_CHECK", "") or "").strip() + if disabled == "1": + return False + return bool(getattr(sys.stdout, "isatty", lambda: False)()) + + +def _print_update_notice(offers: list[ReleaseInfo]) -> None: + print() + print_section("Updates") + warn("A newer Qortium CLI release is available.") + for index, release in enumerate(offers, start=1): + print_option(str(index), release.label) + print(C_TEXT + f" {release.html_url}" + RESET) + print( + C_TEXT + + f' pipx install --force "git+{GITHUB_REPO_URL}.git@{release.tag_name}"' + + RESET + ) + print(DIM + "Restart qortium-cli after updating." + RESET) + + +def maybe_notify_available_updates(settings_dir: Path) -> None: + if not update_checks_enabled(): + return + if not should_check_for_updates(settings_dir): + return + + try: + releases = fetch_github_releases() + except requests.RequestException: + record_update_check(settings_dir, "failed") + return + + offers = select_update_offers(APP_VERSION, releases) + record_update_check(settings_dir, "ok" if offers else "current") + if not offers: + return + + _print_update_notice(offers) + pause() diff --git a/qortium_cli/wallet_backup.py b/qortium_cli/wallet_backup.py index 7d6ee5c..3d94e01 100644 --- a/qortium_cli/wallet_backup.py +++ b/qortium_cli/wallet_backup.py @@ -1,10 +1,12 @@ from __future__ import annotations +from dataclasses import dataclass import hashlib import hmac import json import os import re +import shlex from pathlib import Path from typing import Any, Dict @@ -16,7 +18,16 @@ from qortium_cli.crypto import b58decode, b58encode, qortal_hub_kdf QORTIUM_PRIVATE_KEY_WALLET_VERSION = 3 +QORTIUM_WALLET_VERSION = 2 PRIVATE_KEY_BYTES = 32 +MASTER_SEED_BYTES = 64 + + +@dataclass(frozen=True) +class GeneratedWalletBackup: + address: str + private_key: str + wallet: Dict[str, Any] def qortal_address_from_private_seed(seed: bytes) -> str: @@ -35,6 +46,18 @@ def qortal_address_from_private_seed(seed: bytes) -> str: return b58encode(address_without_checksum + checksum) +def derive_address_seed(master_seed: bytes, address_index: int = 0) -> bytes: + if len(master_seed) != MASTER_SEED_BYTES: + raise ValueError("Qortium Home master seed must be exactly 64 bytes.") + if address_index < 0: + raise ValueError("Wallet address index must be >= 0.") + + nonce = int(address_index).to_bytes(4, "big", signed=False) + nonce_seed = nonce + master_seed + nonce + first_hash = hashlib.sha512(nonce_seed).digest() + return hashlib.sha512(first_hash + nonce_seed).digest()[:PRIVATE_KEY_BYTES] + + def decode_private_key_input(private_key: str) -> bytes: try: decoded = b58decode(str(private_key or "").strip()) @@ -49,24 +72,138 @@ def decode_private_key_input(private_key: str) -> bytes: return decoded -def generate_wallet_backup_from_private_key( - private_key: str, +def _require_wallet_string(wallet: Dict[str, Any], key: str) -> str: + value = wallet.get(key) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"Wallet file must include non-empty {key}.") + return value.strip() + + +def _require_wallet_version(wallet: Dict[str, Any]) -> int: + try: + version = int(wallet.get("version")) + except (TypeError, ValueError): + raise ValueError("Wallet file must include numeric version.") from None + + if version not in {QORTIUM_WALLET_VERSION, QORTIUM_PRIVATE_KEY_WALLET_VERSION}: + raise ValueError(f"Unsupported Qortium Home wallet version: {version}.") + return version + + +def _strip_wrapping_quotes(value: str) -> str: + if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: + return value[1:-1] + return value + + +def normalize_wallet_file_path(path: str | Path) -> Path: + raw_path = str(path or "").strip() + if os.name != "nt": + try: + parts = shlex.split(raw_path) + except ValueError: + parts = [] + if len(parts) == 1: + raw_path = parts[0] + else: + raw_path = _strip_wrapping_quotes(raw_path) + else: + raw_path = _strip_wrapping_quotes(raw_path) + + return Path(raw_path).expanduser().resolve() + + +def decrypt_wallet_payload(wallet: Dict[str, Any], password: str) -> bytes: + wallet_password = str(password or "") + if not wallet_password: + raise ValueError("Wallet password is required.") + + _require_wallet_version(wallet) + try: + kdf_threads = int(wallet.get("kdfThreads")) + except (TypeError, ValueError): + raise ValueError("Wallet file must include numeric kdfThreads.") from None + if kdf_threads != KDF_THREADS: + raise ValueError(f"Unsupported wallet kdfThreads value: {kdf_threads}.") + + try: + encrypted_seed = b58decode(_require_wallet_string(wallet, "encryptedSeed")) + salt = b58decode(_require_wallet_string(wallet, "salt")) + iv = b58decode(_require_wallet_string(wallet, "iv")) + stored_mac = b58decode(_require_wallet_string(wallet, "mac")) + except ValueError as exc: + raise ValueError("Wallet file contains invalid Base58 data.") from exc + + if len(salt) != 32: + raise ValueError("Wallet salt must decode to exactly 32 bytes.") + if len(iv) != 16: + raise ValueError("Wallet IV must decode to exactly 16 bytes.") + if len(encrypted_seed) == 0 or len(encrypted_seed) % 16 != 0: + raise ValueError("Wallet encryptedSeed length is invalid.") + + key = qortal_hub_kdf(wallet_password) + encryption_key = key[:32] + mac_key = key[32:63] + computed_mac = hmac.new(mac_key, encrypted_seed, hashlib.sha512).digest() + if not hmac.compare_digest(computed_mac, stored_mac): + raise ValueError("Incorrect wallet password.") + + try: + decryptor = Cipher( + algorithms.AES(encryption_key), + modes.CBC(iv), + ).decryptor() + return decryptor.update(encrypted_seed) + decryptor.finalize() + except Exception as exc: + raise ValueError("Unable to unlock wallet.") from exc + + +def private_key_from_wallet(wallet: Dict[str, Any], password: str) -> str: + version = _require_wallet_version(wallet) + address = _require_wallet_string(wallet, "address0") + payload = decrypt_wallet_payload(wallet, password) + + if version == QORTIUM_PRIVATE_KEY_WALLET_VERSION: + if len(payload) != PRIVATE_KEY_BYTES: + raise ValueError("Version 3 wallet payload must be exactly 32 bytes.") + private_seed = payload + else: + if len(payload) != MASTER_SEED_BYTES: + raise ValueError("Version 2 wallet payload must be exactly 64 bytes.") + private_seed = derive_address_seed(payload, 0) + + derived_address = qortal_address_from_private_seed(private_seed) + if derived_address != address: + raise ValueError("Wallet password unlocked data, but address0 did not match.") + return b58encode(private_seed) + + +def private_key_from_wallet_file(path: str | Path, password: str) -> str: + wallet_path = normalize_wallet_file_path(path) + try: + wallet = json.loads(wallet_path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise ValueError(f"Wallet file not found: {wallet_path}") from exc + except json.JSONDecodeError as exc: + raise ValueError("Wallet file must be valid JSON.") from exc + + if not isinstance(wallet, dict): + raise ValueError("Wallet file must contain a JSON object.") + return private_key_from_wallet(wallet, password) + + +def _encrypt_wallet_payload( + payload: bytes, address: str, password: str, + version: int, *, salt: bytes | None = None, iv: bytes | None = None, ) -> Dict[str, Any]: - private_seed = decode_private_key_input(private_key) - wallet_address = str(address or "").strip() if not wallet_address: raise ValueError("Wallet address is empty.") - derived_address = qortal_address_from_private_seed(private_seed) - if derived_address != wallet_address: - raise ValueError( - "Private key does not match the configured wallet address." - ) wallet_password = str(password or "") if not wallet_password: @@ -78,6 +215,8 @@ def generate_wallet_backup_from_private_key( raise ValueError("Wallet backup salt must be exactly 32 bytes.") if len(iv_bytes) != 16: raise ValueError("Wallet backup IV must be exactly 16 bytes.") + if len(payload) == 0 or len(payload) % 16 != 0: + raise ValueError("Wallet backup payload length is invalid.") key = qortal_hub_kdf(wallet_password) encryption_key = key[:32] @@ -87,7 +226,7 @@ def generate_wallet_backup_from_private_key( algorithms.AES(encryption_key), modes.CBC(iv_bytes), ).encryptor() - encrypted_seed = encryptor.update(private_seed) + encryptor.finalize() + encrypted_seed = encryptor.update(payload) + encryptor.finalize() mac = hmac.new(mac_key, encrypted_seed, hashlib.sha512).digest() return { @@ -95,12 +234,76 @@ def generate_wallet_backup_from_private_key( "encryptedSeed": b58encode(encrypted_seed), "salt": b58encode(salt_bytes), "iv": b58encode(iv_bytes), - "version": QORTIUM_PRIVATE_KEY_WALLET_VERSION, + "version": version, "mac": b58encode(mac), "kdfThreads": KDF_THREADS, } +def generate_wallet_backup_from_master_seed( + master_seed: bytes, + password: str, + *, + salt: bytes | None = None, + iv: bytes | None = None, +) -> GeneratedWalletBackup: + master_seed_bytes = bytes(master_seed) + if len(master_seed_bytes) != MASTER_SEED_BYTES: + raise ValueError("Qortium Home master seed must be exactly 64 bytes.") + + private_seed = derive_address_seed(master_seed_bytes, 0) + address = qortal_address_from_private_seed(private_seed) + wallet = _encrypt_wallet_payload( + master_seed_bytes, + address, + password, + QORTIUM_WALLET_VERSION, + salt=salt, + iv=iv, + ) + return GeneratedWalletBackup( + address=address, + private_key=b58encode(private_seed), + wallet=wallet, + ) + + +def generate_new_wallet_backup(password: str) -> GeneratedWalletBackup: + return generate_wallet_backup_from_master_seed( + os.urandom(MASTER_SEED_BYTES), + password, + ) + + +def generate_wallet_backup_from_private_key( + private_key: str, + address: str, + password: str, + *, + salt: bytes | None = None, + iv: bytes | None = None, +) -> Dict[str, Any]: + private_seed = decode_private_key_input(private_key) + + wallet_address = str(address or "").strip() + if not wallet_address: + raise ValueError("Wallet address is empty.") + derived_address = qortal_address_from_private_seed(private_seed) + if derived_address != wallet_address: + raise ValueError( + "Private key does not match the configured wallet address." + ) + + return _encrypt_wallet_payload( + private_seed, + wallet_address, + password, + QORTIUM_PRIVATE_KEY_WALLET_VERSION, + salt=salt, + iv=iv, + ) + + def write_wallet_backup(path: Path, backup: Dict[str, Any]) -> Path: output_path = path.expanduser().resolve() output_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/tests/test_chat_commands.py b/tests/test_chat_commands.py new file mode 100644 index 0000000..d048af6 --- /dev/null +++ b/tests/test_chat_commands.py @@ -0,0 +1,390 @@ +import base64 +import io +import json +from contextlib import redirect_stdout +from pathlib import Path +from unittest import TestCase +from unittest.mock import MagicMock, patch + +from qortium_cli.chat_format import build_chat_message_text, build_reaction_message_text +from qortium_cli.models import AccountSettings, AppContext, ChatSettings, EndpointSettings +from qortium_cli.tools import ( + _editable_chat_threads, + _normalize_chat_message_input, + _replyable_chat_threads, + _replyable_chat_user_groups, + _run_chat_edit_command, + _run_chat_reaction_command, + _run_chat_reply_command, + _send_chat_message, + run_chat_room, +) + + +def make_context() -> AppContext: + return AppContext( + settings_dir=Path("."), + endpoint=EndpointSettings(base_url="http://127.0.0.1:24891", timeout_seconds=15), + account=AccountSettings( + name="tester", + account_address="QgV4s3xnzLhVBEJxcYui4u4q11yhUHsd9v", + public_key="public-key", + private_key="private-key", + api_key="api-key", + ), + chat=ChatSettings(), + debug=False, + ) + + +def base64_text(value: str) -> str: + return base64.b64encode(value.encode("utf-8")).decode("ascii") + + +def message(**overrides): + row = { + "data": base64_text("hello"), + "encoding": "BASE64", + "isEncrypted": False, + "isText": True, + "sender": "QgV4s3xnzLhVBEJxcYui4u4q11yhUHsd9v", + "signature": "sig-a", + "timestamp": 10, + "txGroupId": 0, + } + row.update(overrides) + return row + + +def reaction_message( + *, + sender: str, + timestamp: int, + chat_reference: str = "sig-target", + content: str = "\U0001f44d", + content_state: bool = True, +): + return message( + chatReference=chat_reference, + data=base64_text(build_reaction_message_text(content, content_state)), + sender=sender, + signature=f"reaction-{sender}-{timestamp}", + timestamp=timestamp, + ) + + +class ChatCommandTests(TestCase): + def test_build_chat_message_text_preserves_reply_target(self) -> None: + payload = build_chat_message_text("updated", "sig-parent") + + self.assertEqual(json.loads(payload), {"message": "updated", "repliedTo": "sig-parent"}) + self.assertEqual(build_chat_message_text("plain"), "plain") + + def test_build_reaction_message_text_builds_reaction_envelope(self) -> None: + payload = build_reaction_message_text("\U0001f44d", False) + + self.assertEqual( + json.loads(payload), + { + "message": "", + "type": "reaction", + "content": "\U0001f44d", + "contentState": False, + }, + ) + + def test_normalize_chat_message_input_escapes_leading_slash(self) -> None: + self.assertEqual(_normalize_chat_message_input("//help"), "/help") + self.assertEqual(_normalize_chat_message_input("/help"), "/help") + self.assertEqual(_normalize_chat_message_input("hello"), "hello") + + def test_run_chat_room_prints_help_hint_after_timeline(self) -> None: + ctx = make_context() + + with ( + patch("qortium_cli.tools.ensure_wallet_config_ready"), + patch("qortium_cli.tools.print_banner"), + patch("qortium_cli.tools._fetch_chat_timeline", return_value=[]), + patch("qortium_cli.tools._print_chat_timeline", side_effect=lambda messages: print("timeline\n")), + patch("qortium_cli.tools.prompt_str", return_value="/quit") as prompt, + ): + output = io.StringIO() + with redirect_stdout(output): + run_chat_room(ctx) + + text = output.getvalue() + self.assertIn("timeline\n\n/? for help\n", text) + self.assertNotIn("Use /help for commands", text) + prompt.assert_called_once_with("message > ", "") + + def test_editable_chat_threads_only_include_own_confirmed_text_messages(self) -> None: + ctx = make_context() + own = message(signature="sig-own", data=base64_text("own")) + own_edit = message( + chatReference="sig-own", + signature="sig-own-edit", + data=base64_text("own edited"), + ) + other = message( + sender="QotherAddress111111111111111111111111", + signature="sig-other", + data=base64_text("other"), + ) + encrypted = message( + signature="sig-encrypted", + data=base64_text("secret"), + isEncrypted=True, + ) + pending = message(signature="sig-pending", _unconfirmed=True) + + editable = _editable_chat_threads(ctx, [own, other, encrypted, pending, own_edit]) + + self.assertEqual(len(editable), 1) + self.assertEqual(editable[0].original["signature"], "sig-own") + self.assertEqual(editable[0].latest["signature"], "sig-own-edit") + + def test_replyable_chat_threads_include_confirmed_messages_with_signatures(self) -> None: + own = message(signature="sig-own", data=base64_text("own")) + other = message( + sender="QotherAddress111111111111111111111111", + senderName="Other", + signature="sig-other", + data=base64_text("other"), + ) + unsigned = message(signature="", data=base64_text("unsigned")) + pending = message(signature="sig-pending", _unconfirmed=True) + + replyable = _replyable_chat_threads([own, other, unsigned, pending]) + + self.assertEqual([thread.original["signature"] for thread in replyable], ["sig-own", "sig-other"]) + + def test_replyable_chat_user_groups_sort_by_latest_sender_message(self) -> None: + old_own = message( + signature="sig-own-old", + data=base64_text("old own"), + timestamp=10, + ) + recent_own = message( + signature="sig-own-recent", + data=base64_text("recent own"), + timestamp=30, + ) + other = message( + sender="QotherAddress111111111111111111111111", + senderName="Other", + signature="sig-other", + data=base64_text("other"), + timestamp=20, + ) + + groups = _replyable_chat_user_groups([old_own, other, recent_own]) + + self.assertEqual([label for _, label, _ in groups], [make_context().account.account_address, "Other"]) + self.assertEqual( + [thread.original["signature"] for thread in groups[0][2]], + ["sig-own-recent", "sig-own-old"], + ) + + def test_run_chat_reply_command_submits_reply_envelope(self) -> None: + ctx = make_context() + target = message( + sender="QotherAddress111111111111111111111111", + senderName="Other", + data=base64_text("target text"), + signature="sig-target", + ) + + with ( + patch("qortium_cli.tools.read_menu_choice", side_effect=["1", "1"]), + patch("qortium_cli.tools.prompt_str", return_value="reply body"), + patch("qortium_cli.tools._send_chat_message", return_value={"signature": "sig-reply"}) as send, + ): + with redirect_stdout(io.StringIO()): + result = _run_chat_reply_command(ctx, [target]) + + self.assertEqual(result, {"signature": "sig-reply"}) + send.assert_called_once() + _, message_text = send.call_args.args + self.assertEqual(send.call_args.kwargs, {}) + self.assertEqual(json.loads(message_text), {"message": "reply body", "repliedTo": "sig-target"}) + + def test_run_chat_reply_command_blank_input_cancels(self) -> None: + ctx = make_context() + target = message(signature="sig-target", data=base64_text("target text")) + + with ( + patch("qortium_cli.tools.read_menu_choice", side_effect=["1", "1"]), + patch("qortium_cli.tools.prompt_str", return_value=""), + patch("qortium_cli.tools._send_chat_message") as send, + ): + with redirect_stdout(io.StringIO()): + result = _run_chat_reply_command(ctx, [target]) + + self.assertIsNone(result) + send.assert_not_called() + + def test_run_chat_reaction_command_adds_new_reaction(self) -> None: + ctx = make_context() + target = message( + sender="QotherAddress111111111111111111111111", + senderName="Other", + data=base64_text("target text"), + signature="sig-target", + ) + + with ( + patch("qortium_cli.tools.read_menu_choice", side_effect=["1", "1", "2", "1"]), + patch("qortium_cli.tools._send_chat_message", return_value={"signature": "sig-reaction"}) as send, + ): + with redirect_stdout(io.StringIO()): + result = _run_chat_reaction_command(ctx, [target]) + + self.assertEqual(result, {"signature": "sig-reaction"}) + send.assert_called_once() + _, message_text = send.call_args.args + self.assertEqual(send.call_args.kwargs["chat_reference"], "sig-target") + self.assertEqual( + json.loads(message_text), + { + "message": "", + "type": "reaction", + "content": "\U0001f600", + "contentState": True, + }, + ) + + def test_run_chat_reaction_command_removes_existing_self_reaction(self) -> None: + ctx = make_context() + target = message( + sender="QotherAddress111111111111111111111111", + senderName="Other", + data=base64_text("target text"), + signature="sig-target", + ) + existing_reaction = reaction_message( + sender=ctx.account.account_address, + timestamp=20, + chat_reference="sig-target", + ) + + with ( + patch("qortium_cli.tools.read_menu_choice", side_effect=["1", "1", "2", "1"]), + patch("qortium_cli.tools._send_chat_message", return_value={"signature": "sig-reaction"}) as send, + ): + with redirect_stdout(io.StringIO()): + result = _run_chat_reaction_command(ctx, [target, existing_reaction]) + + self.assertEqual(result, {"signature": "sig-reaction"}) + _, message_text = send.call_args.args + self.assertEqual(send.call_args.kwargs["chat_reference"], "sig-target") + self.assertEqual(json.loads(message_text)["contentState"], False) + + def test_run_chat_reaction_command_can_add_when_self_reaction_exists(self) -> None: + ctx = make_context() + target = message( + sender="QotherAddress111111111111111111111111", + senderName="Other", + data=base64_text("target text"), + signature="sig-target", + ) + existing_reaction = reaction_message( + sender=ctx.account.account_address, + timestamp=20, + chat_reference="sig-target", + ) + + with ( + patch("qortium_cli.tools.read_menu_choice", side_effect=["1", "1", "1", "1", "1"]), + patch("qortium_cli.tools._send_chat_message", return_value={"signature": "sig-reaction"}) as send, + ): + with redirect_stdout(io.StringIO()): + result = _run_chat_reaction_command(ctx, [target, existing_reaction]) + + self.assertEqual(result, {"signature": "sig-reaction"}) + _, message_text = send.call_args.args + self.assertEqual(send.call_args.kwargs["chat_reference"], "sig-target") + self.assertEqual( + json.loads(message_text), + { + "message": "", + "type": "reaction", + "content": "\u2764\ufe0f", + "contentState": True, + }, + ) + + def test_run_chat_reaction_command_cancel_does_not_send(self) -> None: + ctx = make_context() + target = message(signature="sig-target", data=base64_text("target text")) + + with ( + patch("qortium_cli.tools.read_menu_choice", side_effect=["1", "1", "0"]), + patch("qortium_cli.tools._send_chat_message") as send, + ): + with redirect_stdout(io.StringIO()): + result = _run_chat_reaction_command(ctx, [target]) + + self.assertIsNone(result) + send.assert_not_called() + + def test_run_chat_edit_command_submits_reference_and_preserves_reply(self) -> None: + ctx = make_context() + original = message( + data=base64_text(json.dumps({"message": "old reply", "repliedTo": "sig-parent"})), + signature="sig-own", + ) + + with ( + patch("qortium_cli.tools.read_menu_choice", return_value="1"), + patch("qortium_cli.tools.prompt_str", return_value="updated reply"), + patch("qortium_cli.tools._send_chat_message", return_value={"signature": "sig-edit"}) as send, + ): + with redirect_stdout(io.StringIO()): + result = _run_chat_edit_command(ctx, [original]) + + self.assertEqual(result, {"signature": "sig-edit"}) + send.assert_called_once() + _, message_text = send.call_args.args + self.assertEqual(send.call_args.kwargs["chat_reference"], "sig-own") + self.assertEqual(json.loads(message_text), {"message": "updated reply", "repliedTo": "sig-parent"}) + + def test_run_chat_edit_command_blank_input_cancels(self) -> None: + ctx = make_context() + original = message(signature="sig-own", data=base64_text("old")) + + with ( + patch("qortium_cli.tools.read_menu_choice", return_value="1"), + patch("qortium_cli.tools.prompt_str", return_value=""), + patch("qortium_cli.tools._send_chat_message") as send, + ): + with redirect_stdout(io.StringIO()): + result = _run_chat_edit_command(ctx, [original]) + + self.assertIsNone(result) + send.assert_not_called() + + def test_send_chat_message_includes_chat_reference_when_present(self) -> None: + ctx = make_context() + session = MagicMock() + manager = MagicMock() + manager.__enter__.return_value = session + manager.__exit__.return_value = None + + with ( + patch("qortium_cli.tools.make_session", return_value=manager), + patch("qortium_cli.tools.to_base58_pubkey", return_value="sender-public-key"), + patch("qortium_cli.tools.b58encode", return_value="encoded-message") as b58encode, + patch("qortium_cli.tools.get_timestamp", return_value=123), + patch("qortium_cli.tools.build_chat", return_value="unsigned") as build_chat, + patch("qortium_cli.tools.compute_chat_nonce", return_value="unsigned-with-nonce"), + patch("qortium_cli.tools.sign_tx", return_value="signed"), + patch("qortium_cli.tools.process_tx", return_value={"signature": "sig-result"}), + ): + with redirect_stdout(io.StringIO()): + result = _send_chat_message(ctx, "edited", chat_reference="sig-own") + + self.assertEqual(result, {"signature": "sig-result"}) + b58encode.assert_called_once_with(b"edited") + payload = build_chat.call_args.args[1] + self.assertEqual(payload["chatReference"], "sig-own") + self.assertEqual(payload["data"], "encoded-message") diff --git a/tests/test_chat_format.py b/tests/test_chat_format.py new file mode 100644 index 0000000..9aea220 --- /dev/null +++ b/tests/test_chat_format.py @@ -0,0 +1,278 @@ +import base64 +import io +import json +from contextlib import redirect_stdout +from unittest import TestCase + +from qortium_cli.chat_format import ( + ChatReaction, + build_reaction_message_text, + build_message_reaction_index, + build_message_threads, + decode_chat_message, +) +from qortium_cli.tools import _print_chat_timeline + + +def base64_text(value: str) -> str: + return base64.b64encode(value.encode("utf-8")).decode("ascii") + + +def message(**overrides): + row = { + "data": base64_text("hello"), + "encoding": "BASE64", + "isEncrypted": False, + "isText": True, + "sender": "Qa", + "signature": "sig-a", + "timestamp": 10, + "txGroupId": 0, + } + row.update(overrides) + return row + + +def reaction_message( + *, + sender: str, + timestamp: int, + chat_reference: str = "sig-a", + content: str = "\U0001f44d", + content_state: bool = True, +): + return message( + chatReference=chat_reference, + data=base64_text( + json.dumps( + { + "message": "", + "type": "reaction", + "content": content, + "contentState": content_state, + } + ) + ), + sender=sender, + signature=f"reaction-{sender}-{timestamp}", + timestamp=timestamp, + ) + + +class ChatFormatTests(TestCase): + def test_decode_plain_base64_text(self) -> None: + decoded = decode_chat_message(message(data=base64_text("plain text"))) + + self.assertEqual(decoded.kind, "text") + self.assertEqual(decoded.body, "plain text") + self.assertIsNone(decoded.replied_to) + + def test_decode_reply_envelope(self) -> None: + decoded = decode_chat_message( + message( + data=base64_text( + json.dumps({"message": "reply body", "repliedTo": "sig-parent"}) + ) + ) + ) + + self.assertEqual(decoded.kind, "text") + self.assertEqual(decoded.body, "reply body") + self.assertEqual(decoded.replied_to, "sig-parent") + + def test_decode_nested_direct_reply_envelope(self) -> None: + nested = json.dumps({"message": "nested reply", "repliedTo": "sig-parent"}) + direct = json.dumps({"message": nested, "version": 2}) + + decoded = decode_chat_message(message(data=base64_text(direct))) + + self.assertEqual(decoded.kind, "text") + self.assertEqual(decoded.body, "nested reply") + self.assertEqual(decoded.replied_to, "sig-parent") + + def test_decode_reaction_envelope(self) -> None: + reaction = json.dumps( + { + "message": "", + "type": "reaction", + "content": "\U0001f44d", + "contentState": False, + } + ) + + decoded = decode_chat_message(message(data=base64_text(reaction))) + + self.assertEqual(decoded.kind, "reaction") + self.assertEqual(decoded.body, "") + self.assertEqual( + decoded.reaction, + ChatReaction(content="\U0001f44d", content_state=False), + ) + + def test_build_reaction_message_text_rejects_empty_reaction(self) -> None: + with self.assertRaises(ValueError): + build_reaction_message_text("", True) + + def test_decode_legacy_message_text_doc(self) -> None: + doc = { + "messageText": { + "type": "doc", + "content": [ + { + "type": "paragraph", + "content": [{"type": "text", "text": "rich text"}], + } + ], + } + } + + decoded = decode_chat_message(message(data=base64_text(json.dumps(doc)))) + + self.assertEqual(decoded.kind, "text") + self.assertEqual(decoded.body, "rich text") + + def test_decode_unreadable_encrypted_message_placeholder(self) -> None: + decoded = decode_chat_message( + message( + data=base64_text("secret"), + isEncrypted=True, + ) + ) + + self.assertEqual(decoded.kind, "encrypted") + self.assertEqual(decoded.body, "Encrypted message") + + def test_decode_decrypted_direct_wrapper(self) -> None: + decoded = decode_chat_message( + message( + data=base64_text(json.dumps({"message": "direct text", "version": 2})), + decryptionStatus="DECRYPTED", + isEncrypted=True, + ) + ) + + self.assertEqual(decoded.kind, "text") + self.assertEqual(decoded.body, "direct text") + + def test_build_message_threads_folds_same_sender_edits(self) -> None: + original = message(data=base64_text("original"), signature="sig-a", timestamp=10) + other = message(sender="Qb", signature="sig-b", timestamp=20) + first_edit = message( + chatReference="sig-a", + data=base64_text("edit one"), + signature="sig-edit-1", + timestamp=30, + ) + second_edit = message( + chatReference="sig-a", + data=base64_text("edit two"), + signature="sig-edit-2", + timestamp=40, + ) + + threads = build_message_threads([original, other, second_edit, first_edit]) + + self.assertEqual(len(threads), 2) + self.assertEqual(threads[0].original, original) + self.assertEqual(threads[0].latest, second_edit) + self.assertEqual(threads[0].revisions, (first_edit, second_edit)) + self.assertEqual(threads[1].latest, other) + + def test_build_message_threads_keeps_replies_and_skips_reactions(self) -> None: + original = message(data=base64_text("original"), signature="sig-a", timestamp=10) + reply = message( + chatReference="sig-a", + data=base64_text("reply"), + sender="Qb", + signature="sig-reply", + timestamp=20, + ) + reaction = reaction_message(sender="Qb", timestamp=30) + + threads = build_message_threads([original, reply, reaction]) + + self.assertEqual(len(threads), 2) + self.assertEqual(threads[0].latest, original) + self.assertEqual(threads[1].latest, reply) + + def test_build_message_reaction_index_uses_latest_sender_state(self) -> None: + reactions = build_message_reaction_index( + [ + reaction_message(sender="Qa", timestamp=20), + reaction_message(sender="Qb", timestamp=30), + reaction_message( + sender="Qc", + timestamp=40, + content="\u2764\ufe0f", + ), + reaction_message(sender="Qa", timestamp=50, content_state=False), + ], + self_address="Qb", + ) + + self.assertEqual(len(reactions["sig-a"]), 2) + self.assertEqual(reactions["sig-a"][0].content, "\U0001f44d") + self.assertEqual(reactions["sig-a"][0].count, 1) + self.assertTrue(reactions["sig-a"][0].reacted_by_self) + self.assertEqual([reactor.sender for reactor in reactions["sig-a"][0].reactors], ["Qb"]) + self.assertEqual(reactions["sig-a"][1].content, "\u2764\ufe0f") + self.assertEqual(reactions["sig-a"][1].count, 1) + + def test_print_chat_timeline_folds_edits_and_shows_reactions(self) -> None: + original = message( + data=base64_text("original body"), + senderName="Alice", + signature="sig-a", + timestamp=1700000000000, + ) + edit = message( + chatReference="sig-a", + data=base64_text("edited body"), + senderName="Alice", + signature="sig-edit", + timestamp=1700000060000, + ) + reply = message( + data=base64_text( + json.dumps({"message": "reply body", "repliedTo": "sig-a"}) + ), + sender="Qb", + senderName="Bob", + signature="sig-reply", + timestamp=1700000120000, + ) + reaction = reaction_message(sender="Qb", timestamp=1700000130000) + + output = io.StringIO() + with redirect_stdout(output): + _print_chat_timeline([original, edit, reply, reaction]) + + text = output.getvalue() + self.assertIn("Chat Timeline (2 messages)", text) + self.assertIn("[edited]", text) + self.assertIn("edited body", text) + self.assertIn("reply to Alice: edited body", text) + self.assertIn("Reactions: \U0001f44d 1", text) + self.assertIn("reply body", text) + self.assertNotIn("original body", text) + self.assertNotIn("sig-reaction", text) + + def test_print_chat_timeline_shows_unavailable_for_missing_reply(self) -> None: + reply = message( + data=base64_text( + json.dumps({"message": "orphan reply", "repliedTo": "missing-signature"}) + ), + sender="Qb", + senderName="Bob", + signature="sig-reply", + timestamp=1700000120000, + ) + + output = io.StringIO() + with redirect_stdout(output): + _print_chat_timeline([reply]) + + text = output.getvalue() + self.assertIn("reply to: unavailable", text) + self.assertIn("orphan reply", text) + self.assertNotIn("missing-signature", text) diff --git a/tests/test_core_detection.py b/tests/test_core_detection.py new file mode 100644 index 0000000..f12b6f1 --- /dev/null +++ b/tests/test_core_detection.py @@ -0,0 +1,128 @@ +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest import TestCase + +from qortium_cli.core_detection import detect_local_core_api_key + + +def write_fake_core_process( + proc_root: Path, + pid: int, + cwd: Path, + *, + api_key: str, + api_port: int, +) -> None: + cwd.mkdir(parents=True, exist_ok=True) + (cwd / "qortium.jar").write_text("placeholder", encoding="utf-8") + (cwd / "settings.json").write_text( + f'{{"apiPort": {api_port}, "apiKeyPath": "."}}\n', + encoding="utf-8", + ) + (cwd / "apikey.txt").write_text(api_key + "\n", encoding="utf-8") + + proc_dir = proc_root / str(pid) + proc_dir.mkdir(parents=True) + proc_dir.joinpath("cmdline").write_bytes(b"java\0-jar\0qortium.jar\0settings.json\0") + proc_dir.joinpath("cwd").symlink_to(cwd, target_is_directory=True) + + +class CoreDetectionTests(TestCase): + def test_detects_api_key_for_matching_local_core_port(self) -> None: + with TemporaryDirectory() as tmp: + root = Path(tmp) + proc_root = root / "proc" + write_fake_core_process( + proc_root, + 101, + root / "core", + api_key="detected-api-key", + api_port=24891, + ) + + result = detect_local_core_api_key("http://127.0.0.1:24891", proc_root=proc_root) + + self.assertIsNotNone(result) + self.assertEqual(result.api_key, "detected-api-key") + self.assertEqual(result.pid, 101) + self.assertEqual(result.api_key_path, (root / "core" / "apikey.txt").resolve()) + + def test_prefers_process_matching_selected_endpoint_port(self) -> None: + with TemporaryDirectory() as tmp: + root = Path(tmp) + proc_root = root / "proc" + write_fake_core_process( + proc_root, + 101, + root / "mainnet-core", + api_key="mainnet-api-key", + api_port=14891, + ) + write_fake_core_process( + proc_root, + 202, + root / "preview-core", + api_key="preview-api-key", + api_port=24891, + ) + + result = detect_local_core_api_key("http://localhost:24891", proc_root=proc_root) + + self.assertIsNotNone(result) + self.assertEqual(result.api_key, "preview-api-key") + self.assertEqual(result.pid, 202) + + def test_rejects_ambiguous_matching_processes(self) -> None: + with TemporaryDirectory() as tmp: + root = Path(tmp) + proc_root = root / "proc" + write_fake_core_process( + proc_root, + 101, + root / "preview-core-a", + api_key="preview-api-key-a", + api_port=24891, + ) + write_fake_core_process( + proc_root, + 202, + root / "preview-core-b", + api_key="preview-api-key-b", + api_port=24891, + ) + + result = detect_local_core_api_key("http://localhost:24891", proc_root=proc_root) + + self.assertIsNone(result) + + def test_ignores_known_non_matching_core_port(self) -> None: + with TemporaryDirectory() as tmp: + root = Path(tmp) + proc_root = root / "proc" + write_fake_core_process( + proc_root, + 101, + root / "mainnet-core", + api_key="mainnet-api-key", + api_port=14891, + ) + + result = detect_local_core_api_key("http://localhost:24891", proc_root=proc_root) + + self.assertIsNone(result) + + def test_ignores_non_local_endpoint(self) -> None: + with TemporaryDirectory() as tmp: + root = Path(tmp) + proc_root = root / "proc" + write_fake_core_process( + proc_root, + 101, + root / "core", + api_key="detected-api-key", + api_port=24891, + ) + + result = detect_local_core_api_key("http://node.example.com:24891", proc_root=proc_root) + + self.assertIsNone(result) diff --git a/tests/test_group_workflows.py b/tests/test_group_workflows.py index caa51a1..846dbbd 100644 --- a/tests/test_group_workflows.py +++ b/tests/test_group_workflows.py @@ -163,10 +163,10 @@ def test_permanent_invite_has_no_expiry(self) -> None: def test_main_menu_uses_requested_numbering(self) -> None: tools = build_tool_plugins() - self.assertEqual([tool.key for tool in tools], ["1", "2", "3", "4", "5", "6"]) + self.assertEqual([tool.key for tool in tools], ["1", "2", "3", "4", "5", "6", "8"]) self.assertEqual( [tool.label for tool in tools], - ["Node", "Chat", "Groups", "Register Name", "Wallet", "QDN Resources"], + ["Node", "Chat", "Groups", "Register Name", "Wallet", "QDN Resources", "Help/Info"], ) def test_accept_invite_submits_join_for_selected_group(self) -> None: diff --git a/tests/test_help_info.py b/tests/test_help_info.py new file mode 100644 index 0000000..ca4dd51 --- /dev/null +++ b/tests/test_help_info.py @@ -0,0 +1,48 @@ +import io +from contextlib import redirect_stdout +from pathlib import Path +from unittest import TestCase +from unittest.mock import patch + +from qortium_cli.constants import APP_TITLE +from qortium_cli.models import AccountSettings, AppContext, ChatSettings, EndpointSettings +from qortium_cli.tools import tool_help_info + + +def make_context() -> AppContext: + return AppContext( + settings_dir=Path("."), + endpoint=EndpointSettings(base_url="http://127.0.0.1:24891", timeout_seconds=15), + account=AccountSettings( + name="tester", + account_address="QgV4s3xnzLhVBEJxcYui4u4q11yhUHsd9v", + public_key="public-key", + private_key="private-key", + api_key="api-key", + ), + chat=ChatSettings(), + debug=False, + ) + + +class HelpInfoTests(TestCase): + def test_app_title_uses_v030(self) -> None: + self.assertEqual(APP_TITLE, "Qortium CLI 0.3.0") + + def test_help_info_shows_v030_whats_new_entry(self) -> None: + ctx = make_context() + + with ( + patch("qortium_cli.tools.print_banner"), + patch("qortium_cli.tools.read_menu_choice", side_effect=["1", "1", "0", "0"]), + patch("qortium_cli.tools.pause"), + ): + output = io.StringIO() + with redirect_stdout(output): + tool_help_info(ctx) + + text = output.getvalue() + self.assertIn("What's New?", text) + self.assertIn("v0.3.0", text) + self.assertIn("Chat commands added: /reply, /edit, /react, /help, and /?.", text) + self.assertIn("Reaction picker supports add/remove flows and emoji categories.", text) diff --git a/tests/test_qdn_resources.py b/tests/test_qdn_resources.py index e56c6c7..7c83e03 100644 --- a/tests/test_qdn_resources.py +++ b/tests/test_qdn_resources.py @@ -20,6 +20,9 @@ _submit_arbitrary_publish_transaction, browse_qdn_resources, select_qdn_resource, + tool_register_name, + tx_name_register, + tx_name_update, ) @@ -330,6 +333,136 @@ def test_build_app_publish_enforces_metadata_byte_limit(self) -> None: class QdnWorkflowTests(TestCase): + def test_register_name_menu_registers_directly_without_owned_names(self) -> None: + ctx = make_context() + session = MagicMock() + session_context = MagicMock() + session_context.__enter__.return_value = session + + with ( + patch("qortium_cli.tools.make_session", return_value=session_context), + patch("qortium_cli.tools.get_account_names", return_value=[]) as get_names, + patch("qortium_cli.tools.tx_name_register") as register_name, + patch("qortium_cli.tools.print_banner"), + patch("qortium_cli.tools.print_stat"), + patch("qortium_cli.tools.pause") as pause, + ): + tool_register_name(ctx) + + get_names.assert_called_once_with( + ctx, + ctx.account.account_address, + session, + limit=500, + ) + register_name.assert_called_once_with(ctx) + pause.assert_called_once() + + def test_register_name_menu_updates_existing_name(self) -> None: + ctx = make_context() + session = MagicMock() + session_context = MagicMock() + session_context.__enter__.return_value = session + owned_names = ["First Name", "Second Name"] + + with ( + patch("qortium_cli.tools.make_session", return_value=session_context), + patch("qortium_cli.tools.get_account_names", return_value=owned_names), + patch("qortium_cli.tools.read_menu_choice", return_value="1"), + patch("qortium_cli.tools.tx_name_update") as update_name, + patch("qortium_cli.tools.print_banner"), + patch("qortium_cli.tools.print_section"), + patch("qortium_cli.tools.print_stat"), + patch("qortium_cli.tools.print_option"), + patch("qortium_cli.tools.pause") as pause, + ): + tool_register_name(ctx) + + update_name.assert_called_once_with(ctx, owned_names) + pause.assert_called_once() + + def test_register_name_menu_can_start_new_name_when_owned_names_exist(self) -> None: + ctx = make_context() + session_context = MagicMock() + session_context.__enter__.return_value = MagicMock() + + with ( + patch("qortium_cli.tools.make_session", return_value=session_context), + patch("qortium_cli.tools.get_account_names", return_value=["My Name"]), + patch("qortium_cli.tools.read_menu_choice", return_value="2"), + patch("qortium_cli.tools.tx_name_register") as register_name, + patch("qortium_cli.tools.print_banner"), + patch("qortium_cli.tools.print_section"), + patch("qortium_cli.tools.print_stat"), + patch("qortium_cli.tools.print_option"), + patch("qortium_cli.tools.pause") as pause, + ): + tool_register_name(ctx) + + register_name.assert_called_once_with(ctx) + pause.assert_called_once() + + def test_name_register_can_cancel_with_blank_name(self) -> None: + ctx = make_context() + + with ( + patch("qortium_cli.tools.prompt_str", return_value=""), + patch("qortium_cli.tools._submit_builder_transaction") as submit_tx, + patch("qortium_cli.tools.warn"), + ): + tx_name_register(ctx) + + submit_tx.assert_not_called() + + def test_name_update_builds_update_name_transaction(self) -> None: + ctx = make_context() + + with ( + patch("qortium_cli.tools.read_menu_choice", return_value="2"), + patch( + "qortium_cli.tools.prompt_str", + side_effect=["Renamed Name", "{\"bio\":\"updated\"}"], + ), + patch("qortium_cli.tools.prompt_decimal", return_value=Decimal("0")), + patch("qortium_cli.tools.prompt_int", return_value=0), + patch( + "qortium_cli.tools.to_base58_pubkey", + return_value="owner-public-key", + ), + patch("qortium_cli.tools._submit_builder_transaction") as submit_tx, + patch("qortium_cli.tools.print_section"), + patch("qortium_cli.tools.print_option"), + ): + tx_name_update(ctx, ["First Name", "Second Name"]) + + submit_tx.assert_called_once_with( + ctx, + "/names/update", + "UPDATE_NAME", + { + "ownerPublicKey": "owner-public-key", + "name": "Second Name", + "newName": "Renamed Name", + "newData": "{\"bio\":\"updated\"}", + }, + "0.00000000", + 0, + ) + + def test_name_update_can_cancel_name_selection_with_zero(self) -> None: + ctx = make_context() + + with ( + patch("qortium_cli.tools.read_menu_choice", return_value="0"), + patch("qortium_cli.tools._submit_builder_transaction") as submit_tx, + patch("qortium_cli.tools.print_section"), + patch("qortium_cli.tools.print_option"), + patch("qortium_cli.tools.warn"), + ): + tx_name_update(ctx, ["My Name"]) + + submit_tx.assert_not_called() + def test_resource_tuple_normalizes_missing_identifier(self) -> None: self.assertEqual( _qdn_resource_tuple( diff --git a/tests/test_reconfigure.py b/tests/test_reconfigure.py index a8efe58..a33a42f 100644 --- a/tests/test_reconfigure.py +++ b/tests/test_reconfigure.py @@ -3,9 +3,22 @@ from unittest import TestCase from unittest.mock import patch +from qortium_cli.core_detection import LocalCoreApiKey from qortium_cli.models import AccountSettings, AppContext, ChatSettings, EndpointSettings -from qortium_cli.setup_wizard import configure_api_key, run_reconfigure_menu -from qortium_cli.storage import load_account_settings, write_config_file +from qortium_cli.setup_wizard import ( + configure_api_key, + configure_endpoint_url, + configure_wallet_identity, + run_initial_setup, + run_reconfigure_menu, +) +from qortium_cli.storage import ( + load_account_settings, + load_endpoint_settings, + write_config_file, + write_endpoint_file, +) +from qortium_cli.wallet_backup import GeneratedWalletBackup, QORTIUM_WALLET_VERSION def make_context(settings_dir: Path) -> AppContext: @@ -31,7 +44,10 @@ def test_configure_api_key_preserves_wallet_values(self) -> None: ctx = make_context(settings_dir) write_config_file(settings_dir, ctx.account) - with patch("qortium_cli.setup_wizard.prompt_secret", return_value="new-api-key"): + with ( + patch("qortium_cli.setup_wizard.detect_local_core_api_key", return_value=None), + patch("qortium_cli.setup_wizard.prompt_secret", return_value="new-api-key"), + ): configure_api_key(ctx) saved = load_account_settings(settings_dir) @@ -43,6 +59,83 @@ def test_configure_api_key_preserves_wallet_values(self) -> None: self.assertEqual(saved.account_address, "QgV4s3xnzLhVBEJxcYui4u4q11yhUHsd9v") self.assertEqual(saved.name, "old-name") + def test_configure_api_key_uses_detected_local_core_key_by_default(self) -> None: + with TemporaryDirectory() as tmp: + settings_dir = Path(tmp) + ctx = make_context(settings_dir) + write_config_file(settings_dir, ctx.account) + detected = LocalCoreApiKey( + api_key="detected-api-key", + api_key_path=settings_dir / "apikey.txt", + api_key_directory=settings_dir, + cwd=settings_dir, + jar_path=settings_dir / "qortium.jar", + pid=123, + settings_path=settings_dir / "settings.json", + ) + + with ( + patch("qortium_cli.setup_wizard.detect_local_core_api_key", return_value=detected), + patch("qortium_cli.setup_wizard.prompt_secret", return_value=""), + patch("qortium_cli.setup_wizard.ok"), + ): + configure_api_key(ctx) + + saved = load_account_settings(settings_dir) + + self.assertEqual(ctx.account.api_key, "detected-api-key") + self.assertEqual(saved.api_key, "detected-api-key") + self.assertEqual(saved.private_key, "old-private-key") + + def test_initial_setup_prefers_detected_api_key_over_existing_key(self) -> None: + with TemporaryDirectory() as tmp: + settings_dir = Path(tmp) + ctx = make_context(settings_dir) + detected = LocalCoreApiKey( + api_key="detected-api-key", + api_key_path=settings_dir / "apikey.txt", + api_key_directory=settings_dir, + cwd=settings_dir, + jar_path=settings_dir / "qortium.jar", + pid=123, + settings_path=settings_dir / "settings.json", + ) + next_account = AccountSettings( + name="detected-account", + account_address="Qdetected", + public_key="detected-public-key", + private_key="detected-private-key", + api_key="detected-api-key", + ) + + with ( + patch( + "qortium_cli.setup_wizard._prompt_endpoint_url_with_connection_check", + return_value="http://127.0.0.1:24891", + ), + patch("qortium_cli.setup_wizard.prompt_int", return_value=15), + patch("qortium_cli.setup_wizard.detect_local_core_api_key", return_value=detected), + patch("qortium_cli.setup_wizard.prompt_secret", return_value=""), + patch("qortium_cli.setup_wizard._prompt_private_key", return_value="detected-private-key"), + patch( + "qortium_cli.setup_wizard._account_from_private_key", + return_value=next_account, + ) as account_from_private_key, + patch("qortium_cli.setup_wizard.print_setup_banner"), + patch("qortium_cli.setup_wizard.pause"), + patch("qortium_cli.setup_wizard.ok"), + ): + run_initial_setup(ctx) + + saved = load_account_settings(settings_dir) + + account_from_private_key.assert_called_once_with( + ctx, + "detected-private-key", + "detected-api-key", + ) + self.assertEqual(saved.api_key, "detected-api-key") + def test_reconfigure_menu_can_update_only_api_key(self) -> None: with TemporaryDirectory() as tmp: settings_dir = Path(tmp) @@ -51,6 +144,7 @@ def test_reconfigure_menu_can_update_only_api_key(self) -> None: with ( patch("qortium_cli.setup_wizard.read_menu_choice", side_effect=["3", "0"]), + patch("qortium_cli.setup_wizard.detect_local_core_api_key", return_value=None), patch("qortium_cli.setup_wizard.prompt_secret", return_value="new-api-key"), patch("qortium_cli.setup_wizard.print_setup_banner"), patch("qortium_cli.setup_wizard.print_stat"), @@ -63,3 +157,199 @@ def test_reconfigure_menu_can_update_only_api_key(self) -> None: self.assertEqual(saved.api_key, "new-api-key") self.assertEqual(saved.private_key, "old-private-key") + + def test_reconfigure_menu_runs_initial_setup_then_returns(self) -> None: + with TemporaryDirectory() as tmp: + settings_dir = Path(tmp) + ctx = make_context(settings_dir) + + with ( + patch("qortium_cli.setup_wizard.read_menu_choice", return_value="9") as read_menu_choice, + patch("qortium_cli.setup_wizard.run_initial_setup") as run_initial_setup, + patch("qortium_cli.setup_wizard.print_setup_banner"), + patch("qortium_cli.setup_wizard.print_stat"), + patch("qortium_cli.setup_wizard.print_option") as print_option, + ): + run_reconfigure_menu(ctx) + + print_option.assert_any_call("9", "Run initial setup") + read_menu_choice.assert_called_once() + run_initial_setup.assert_called_once_with(ctx) + + def test_configure_wallet_identity_accepts_wallet_file(self) -> None: + with TemporaryDirectory() as tmp: + settings_dir = Path(tmp) + ctx = make_context(settings_dir) + wallet_path = settings_dir / "wallet.json" + next_account = AccountSettings( + name="wallet-account", + account_address="Qwallet", + public_key="wallet-public-key", + private_key="wallet-private-key", + api_key="old-api-key", + ) + + with ( + patch("qortium_cli.setup_wizard.read_menu_choice", return_value="3"), + patch("qortium_cli.setup_wizard.prompt_str", return_value=f"'{wallet_path}' "), + patch("qortium_cli.setup_wizard.prompt_secret", return_value="wallet-password"), + patch( + "qortium_cli.setup_wizard.private_key_from_wallet_file", + return_value="wallet-private-key", + ) as private_key_from_wallet_file, + patch( + "qortium_cli.setup_wizard._account_from_private_key", + return_value=next_account, + ) as account_from_private_key, + patch("qortium_cli.setup_wizard.print_option") as print_option, + patch("qortium_cli.setup_wizard.ok"), + ): + configure_wallet_identity(ctx) + + saved = load_account_settings(settings_dir) + + print_option.assert_any_call("3", "Use wallet file") + print_option.assert_any_call("4", "New wallet file") + private_key_from_wallet_file.assert_called_once_with(wallet_path.resolve(), "wallet-password") + account_from_private_key.assert_called_once_with(ctx, "wallet-private-key", "old-api-key") + self.assertEqual(ctx.account.private_key, "wallet-private-key") + self.assertEqual(saved.private_key, "wallet-private-key") + + def test_configure_wallet_identity_can_create_new_wallet_file(self) -> None: + with TemporaryDirectory() as tmp: + settings_dir = Path(tmp) + ctx = make_context(settings_dir) + output_path = settings_dir / "new-wallet.json" + generated_wallet = GeneratedWalletBackup( + address="Qnew", + private_key="new-private-key", + wallet={"address0": "Qnew", "version": QORTIUM_WALLET_VERSION}, + ) + next_account = AccountSettings( + name="new-account", + account_address="Qnew", + public_key="new-public-key", + private_key="new-private-key", + api_key="old-api-key", + ) + + with ( + patch("qortium_cli.setup_wizard.read_menu_choice", return_value="4"), + patch( + "qortium_cli.setup_wizard.prompt_secret", + side_effect=["wallet-password", "wallet-password"], + ), + patch( + "qortium_cli.setup_wizard.prompt_str", + side_effect=["My Wallet", str(output_path)], + ), + patch( + "qortium_cli.setup_wizard.generate_new_wallet_backup", + return_value=generated_wallet, + ) as generate_new_wallet, + patch( + "qortium_cli.setup_wizard.default_wallet_backup_path", + return_value=output_path, + ) as default_path, + patch( + "qortium_cli.setup_wizard.write_wallet_backup", + return_value=output_path, + ) as write_wallet_backup, + patch( + "qortium_cli.setup_wizard._account_from_private_key", + return_value=next_account, + ) as account_from_private_key, + patch("qortium_cli.setup_wizard.ok"), + ): + configure_wallet_identity(ctx) + + saved = load_account_settings(settings_dir) + + generate_new_wallet.assert_called_once_with("wallet-password") + default_path.assert_called_once_with("Qnew", wallet_name="My Wallet") + write_wallet_backup.assert_called_once_with( + output_path.resolve(), + generated_wallet.wallet, + ) + account_from_private_key.assert_called_once_with( + ctx, + "new-private-key", + "old-api-key", + ) + self.assertEqual(ctx.account.private_key, "new-private-key") + self.assertEqual(saved.private_key, "new-private-key") + + def test_configure_wallet_identity_does_not_create_wallet_on_password_mismatch(self) -> None: + with TemporaryDirectory() as tmp: + settings_dir = Path(tmp) + ctx = make_context(settings_dir) + + with ( + patch("qortium_cli.setup_wizard.read_menu_choice", return_value="4"), + patch( + "qortium_cli.setup_wizard.prompt_secret", + side_effect=["first-password", "different-password"], + ), + patch("qortium_cli.setup_wizard.generate_new_wallet_backup") as generate_new_wallet, + patch("qortium_cli.setup_wizard.write_wallet_backup") as write_wallet_backup, + ): + with self.assertRaisesRegex(RuntimeError, "Wallet passwords do not match"): + configure_wallet_identity(ctx) + + generate_new_wallet.assert_not_called() + write_wallet_backup.assert_not_called() + + def test_configure_endpoint_url_retries_after_failed_connection(self) -> None: + with TemporaryDirectory() as tmp: + settings_dir = Path(tmp) + ctx = make_context(settings_dir) + write_endpoint_file(settings_dir, ctx.endpoint) + + with ( + patch( + "qortium_cli.setup_wizard.prompt_str", + side_effect=["http://bad-node:24891", "http://good-node:24891"], + ), + patch( + "qortium_cli.setup_wizard.check_node_connection", + side_effect=[ + (False, "connection refused"), + (True, "Node API responded."), + ], + ) as check_node_connection, + patch("qortium_cli.setup_wizard.read_menu_choice", return_value="1"), + patch("qortium_cli.setup_wizard.print_option"), + patch("qortium_cli.setup_wizard.warn"), + patch("qortium_cli.setup_wizard.ok"), + ): + configure_endpoint_url(ctx) + + saved = load_endpoint_settings(settings_dir) + + self.assertEqual(ctx.endpoint.base_url, "http://good-node:24891") + self.assertEqual(saved.base_url, "http://good-node:24891") + self.assertEqual(check_node_connection.call_count, 2) + + def test_configure_endpoint_url_can_continue_after_failed_connection(self) -> None: + with TemporaryDirectory() as tmp: + settings_dir = Path(tmp) + ctx = make_context(settings_dir) + write_endpoint_file(settings_dir, ctx.endpoint) + + with ( + patch("qortium_cli.setup_wizard.prompt_str", return_value="http://bad-node:24891"), + patch( + "qortium_cli.setup_wizard.check_node_connection", + return_value=(False, "connection refused"), + ) as check_node_connection, + patch("qortium_cli.setup_wizard.read_menu_choice", return_value="2"), + patch("qortium_cli.setup_wizard.print_option"), + patch("qortium_cli.setup_wizard.warn"), + ): + configure_endpoint_url(ctx) + + saved = load_endpoint_settings(settings_dir) + + self.assertEqual(ctx.endpoint.base_url, "http://bad-node:24891") + self.assertEqual(saved.base_url, "http://bad-node:24891") + check_node_connection.assert_called_once_with("http://bad-node:24891", 15) diff --git a/tests/test_update_checker.py b/tests/test_update_checker.py new file mode 100644 index 0000000..229b532 --- /dev/null +++ b/tests/test_update_checker.py @@ -0,0 +1,118 @@ +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest import TestCase + +from qortium_cli.update_checker import ( + ReleaseInfo, + parse_version_tag, + record_update_check, + release_from_github_payload, + select_update_offers, + should_check_for_updates, +) + + +def release(tag_name: str, prerelease: bool = False) -> ReleaseInfo: + parsed = parse_version_tag(tag_name) + assert parsed is not None + return ReleaseInfo( + tag_name=tag_name, + version=parsed, + prerelease=prerelease, + html_url=f"https://example.test/{tag_name}", + ) + + +class UpdateCheckerTests(TestCase): + def test_stable_version_is_newer_than_same_base_prerelease(self) -> None: + prerelease = parse_version_tag("v0.4.0-preview.2") + stable = parse_version_tag("v0.4.0") + + self.assertIsNotNone(prerelease) + self.assertIsNotNone(stable) + self.assertLess(prerelease, stable) + + def test_select_update_offers_prerelease_first_when_it_is_ahead(self) -> None: + offers = select_update_offers( + "0.3.0", + [ + release("v0.3.1"), + release("v0.4.0-preview.1", prerelease=True), + ], + ) + + self.assertEqual( + [offer.tag_name for offer in offers], + ["v0.4.0-preview.1", "v0.3.1"], + ) + + def test_select_update_ignores_prerelease_when_stable_is_newer(self) -> None: + offers = select_update_offers( + "0.3.0", + [ + release("v0.4.0"), + release("v0.4.0-preview.2", prerelease=True), + ], + ) + + self.assertEqual([offer.tag_name for offer in offers], ["v0.4.0"]) + + def test_select_update_offers_stable_when_no_prerelease_is_ahead(self) -> None: + offers = select_update_offers( + "0.3.0", + [ + release("v0.3.1"), + release("v0.3.1-preview.1", prerelease=True), + ], + ) + + self.assertEqual([offer.tag_name for offer in offers], ["v0.3.1"]) + + def test_select_update_treats_prerelease_tag_as_prerelease(self) -> None: + offers = select_update_offers( + "0.3.0", + [ + release("v0.4.0"), + release("v0.4.0-preview.1"), + ], + ) + + self.assertEqual([offer.tag_name for offer in offers], ["v0.4.0"]) + + def test_select_update_can_offer_prerelease_when_no_stable_release_exists(self) -> None: + offers = select_update_offers( + "0.3.0", + [release("v0.3.1-preview.1", prerelease=True)], + ) + + self.assertEqual([offer.tag_name for offer in offers], ["v0.3.1-preview.1"]) + + def test_release_from_github_payload_skips_drafts_and_unknown_tags(self) -> None: + self.assertIsNone( + release_from_github_payload({"tag_name": "not-a-version", "draft": False}) + ) + self.assertIsNone( + release_from_github_payload({"tag_name": "v0.3.1", "draft": True}) + ) + + release_info = release_from_github_payload( + { + "tag_name": "v0.3.1", + "prerelease": False, + "html_url": "https://example.test/release", + } + ) + + self.assertIsNotNone(release_info) + self.assertEqual(release_info.tag_name, "v0.3.1") + + def test_should_check_for_updates_respects_daily_interval(self) -> None: + with TemporaryDirectory() as tmp: + settings_dir = Path(tmp) + + self.assertTrue(should_check_for_updates(settings_dir, now=1_000)) + record_update_check(settings_dir, "ok", now=1_000) + self.assertFalse(should_check_for_updates(settings_dir, now=1_000 + 60)) + self.assertTrue( + should_check_for_updates(settings_dir, now=1_000 + (24 * 60 * 60)) + ) diff --git a/tests/test_wallet_backup.py b/tests/test_wallet_backup.py index 675f4ca..f9e4138 100644 --- a/tests/test_wallet_backup.py +++ b/tests/test_wallet_backup.py @@ -12,10 +12,17 @@ from qortium_cli.models import AccountSettings, AppContext, ChatSettings, EndpointSettings from qortium_cli.tools import export_wallet_backup from qortium_cli.wallet_backup import ( + MASTER_SEED_BYTES, QORTIUM_PRIVATE_KEY_WALLET_VERSION, + QORTIUM_WALLET_VERSION, decode_private_key_input, + derive_address_seed, default_wallet_backup_path, + generate_new_wallet_backup, + generate_wallet_backup_from_master_seed, generate_wallet_backup_from_private_key, + normalize_wallet_file_path, + private_key_from_wallet, qortal_address_from_private_seed, write_wallet_backup, ) @@ -37,6 +44,27 @@ def make_context(private_key: str, address: str) -> AppContext: ) +def encrypted_wallet_payload(payload: bytes, address: str, password: str, version: int) -> dict: + key = qortal_hub_kdf(password) + iv = bytes(range(16)) + salt = bytes(range(32, 64)) + encryptor = Cipher( + algorithms.AES(key[:32]), + modes.CBC(iv), + ).encryptor() + encrypted_seed = encryptor.update(payload) + encryptor.finalize() + mac = hmac.new(key[32:63], encrypted_seed, hashlib.sha512).digest() + return { + "address0": address, + "encryptedSeed": b58encode(encrypted_seed), + "salt": b58encode(salt), + "iv": b58encode(iv), + "version": version, + "mac": b58encode(mac), + "kdfThreads": 16, + } + + class WalletBackupTests(TestCase): def test_qortium_home_crypto_matches_javascript_vector(self) -> None: payload = bytes(range(32)) @@ -98,12 +126,120 @@ def test_qortium_home_backup_decrypts_to_original_private_seed(self) -> None: ) self.assertEqual(backup["kdfThreads"], 16) + def test_master_seed_wallet_backup_decrypts_to_address_zero_key(self) -> None: + master_seed = bytes(range(MASTER_SEED_BYTES)) + private_seed = derive_address_seed(master_seed, 0) + address = qortal_address_from_private_seed(private_seed) + key = bytes(range(64)) + salt = bytes(range(32, 64)) + iv = bytes(range(16)) + + with patch("qortium_cli.wallet_backup.qortal_hub_kdf", return_value=key): + generated = generate_wallet_backup_from_master_seed( + master_seed, + "backup-password", + salt=salt, + iv=iv, + ) + restored_private_key = private_key_from_wallet( + generated.wallet, + "backup-password", + ) + + encrypted_seed = b58decode(generated.wallet["encryptedSeed"]) + decryptor = Cipher( + algorithms.AES(key[:32]), + modes.CBC(iv), + ).decryptor() + decrypted_seed = decryptor.update(encrypted_seed) + decryptor.finalize() + + self.assertEqual(generated.address, address) + self.assertEqual(generated.private_key, b58encode(private_seed)) + self.assertEqual(restored_private_key, generated.private_key) + self.assertEqual(decrypted_seed, master_seed) + self.assertEqual(generated.wallet["address0"], address) + self.assertEqual(generated.wallet["version"], QORTIUM_WALLET_VERSION) + + def test_generate_new_wallet_backup_uses_random_master_seed(self) -> None: + master_seed = bytes(range(MASTER_SEED_BYTES)) + + with patch( + "qortium_cli.wallet_backup.os.urandom", + side_effect=[master_seed, bytes(32), bytes(16)], + ) as urandom: + generated = generate_new_wallet_backup("backup-password") + + self.assertEqual(urandom.call_args_list[0].args, (MASTER_SEED_BYTES,)) + self.assertEqual( + generated.private_key, + b58encode(derive_address_seed(master_seed, 0)), + ) + self.assertEqual(generated.wallet["version"], QORTIUM_WALLET_VERSION) + + def test_master_seed_wallet_backup_rejects_wrong_seed_length(self) -> None: + with self.assertRaisesRegex(ValueError, "exactly 64 bytes"): + generate_wallet_backup_from_master_seed(b"short", "backup-password") + def test_qortium_home_64_byte_secret_key_uses_first_32_bytes(self) -> None: seed = bytes(range(32)) secret_key = seed + bytes(reversed(range(32))) self.assertEqual(decode_private_key_input(b58encode(secret_key)), seed) + def test_normalize_wallet_file_path_accepts_dragged_terminal_paths(self) -> None: + quoted_path = "'/home/user/Downloads/qortium-docs/wallet.json' " + escaped_path = "/home/user/Downloads/qortium-docs/My\\ Wallet.json" + + self.assertEqual( + normalize_wallet_file_path(quoted_path), + Path("/home/user/Downloads/qortium-docs/wallet.json"), + ) + self.assertEqual( + normalize_wallet_file_path(escaped_path), + Path("/home/user/Downloads/qortium-docs/My Wallet.json"), + ) + + def test_private_key_from_version_3_wallet_file_payload(self) -> None: + seed = bytes(range(32)) + address = qortal_address_from_private_seed(seed) + wallet = encrypted_wallet_payload( + seed, + address, + "wallet-password", + QORTIUM_PRIVATE_KEY_WALLET_VERSION, + ) + + self.assertEqual(private_key_from_wallet(wallet, "wallet-password"), b58encode(seed)) + + def test_private_key_from_version_2_wallet_file_payload(self) -> None: + master_seed = bytes(range(64)) + private_seed = derive_address_seed(master_seed, 0) + address = qortal_address_from_private_seed(private_seed) + wallet = encrypted_wallet_payload( + master_seed, + address, + "wallet-password", + QORTIUM_WALLET_VERSION, + ) + + self.assertEqual( + private_key_from_wallet(wallet, "wallet-password"), + b58encode(private_seed), + ) + + def test_private_key_from_wallet_rejects_wrong_password(self) -> None: + seed = bytes(range(32)) + address = qortal_address_from_private_seed(seed) + wallet = encrypted_wallet_payload( + seed, + address, + "wallet-password", + QORTIUM_PRIVATE_KEY_WALLET_VERSION, + ) + + with self.assertRaisesRegex(ValueError, "Incorrect wallet password"): + private_key_from_wallet(wallet, "wrong-password") + def test_backup_rejects_private_key_for_another_address(self) -> None: private_key = b58encode(bytes(range(32))) other_address = qortal_address_from_private_seed(bytes(range(1, 33)))