# seedpass.core/manager.py """ Password Manager Module This module implements the PasswordManager class, which orchestrates various functionalities of the deterministic password manager, including encryption, entry management, password generation, backup, and checksum verification. It serves as the core interface for interacting with the password manager functionalities. """ import sys import json import logging import os import hashlib import hmac import base64 from typing import Optional, Literal, Any import shutil import time from datetime import datetime, timezone import builtins import threading import queue from dataclasses import dataclass import dataclasses from contextlib import contextmanager from functools import wraps from termcolor import colored from utils.color_scheme import color_text from utils.input_utils import timed_input from constants import MAX_RETRIES, RETRY_DELAY from .encryption import EncryptionManager from .entry_management import EntryManager from .password_generation import PasswordGenerator from .backup import BackupManager from .vault import Vault from .portable_backup import export_backup, import_backup, PortableMode from .errors import SeedPassError, DecryptionError from .totp import TotpManager from .entry_types import EntryType from .pubsub import bus from utils.key_derivation import ( derive_key_from_parent_seed, derive_key_from_password, derive_key_from_password_argon2, derive_index_key, EncryptionMode, KdfConfig, ) from utils.key_hierarchy import kd from utils.checksum import ( calculate_checksum, verify_checksum, json_checksum, initialize_checksum, update_checksum_file, ) from utils.password_prompt import ( prompt_for_password, prompt_existing_password, prompt_new_password, confirm_action, ) from utils import masked_input, prompt_seed_words from utils.memory_protection import InMemorySecret from utils.clipboard import copy_to_clipboard from utils.terminal_utils import ( clear_screen, pause, clear_and_print_profile_chain, clear_header_with_notification, ) from utils.fingerprint import generate_fingerprint from utils.atomic_write import atomic_write from constants import MIN_HEALTHY_RELAYS from .migrations import LATEST_VERSION from ..errors import VaultLockedError from constants import ( APP_DIR, PARENT_SEED_FILE, SCRIPT_CHECKSUM_FILE, MIN_PASSWORD_LENGTH, MAX_PASSWORD_LENGTH, DEFAULT_PASSWORD_LENGTH, INACTIVITY_TIMEOUT, DEFAULT_SEED_BACKUP_FILENAME, NOTIFICATION_DURATION, initialize_app, ) import asyncio import gzip import bcrypt from pathlib import Path from local_bip85.bip85 import BIP85, Bip85Error from bip_utils import Bip39SeedGenerator, Bip39MnemonicGenerator, Bip39Languages from mnemonic import Mnemonic from datetime import datetime from utils.fingerprint_manager import FingerprintManager from utils.logging_utils import pause_logging_for_ui # Import NostrClient from nostr.client import NostrClient from nostr.connection import DEFAULT_RELAYS from .config_manager import ConfigManager from .state_manager import StateManager from .stats_manager import StatsManager from .menu_handler import MenuHandler from .profile_service import ProfileService from .entry_service import EntryService from .display_service import DisplayService # Instantiate the logger logger = logging.getLogger(__name__) class AuditLogger: """Append-only logger producing tamper-evident JSON records.""" def __init__(self, key: bytes, path: Path | None = None) -> None: self.key = key self.path = path or (Path.home() / ".seedpass" / "audit.log") self.path.parent.mkdir(parents=True, exist_ok=True) self.last_sig = "0" * 64 if self.path.exists(): try: with self.path.open("r", encoding="utf-8") as fh: last = None for line in fh: if line.strip(): last = line if last is not None: self.last_sig = json.loads(last).get("sig", self.last_sig) except Exception: pass def log(self, event: str, details: dict[str, Any] | None = None) -> None: details = details or {} entry = { "timestamp": datetime.now(timezone.utc).isoformat(), "event": event, "details": details, } payload = json.dumps(entry, sort_keys=True, separators=(",", ":")) sig = hmac.new( self.key, f"{self.last_sig}{payload}".encode(), hashlib.sha256 ).hexdigest() entry["sig"] = sig with self.path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(entry, sort_keys=True) + "\n") self.last_sig = sig def calculate_profile_id(seed: str) -> str: """Return the fingerprint identifier for ``seed``.""" fp = generate_fingerprint(seed) return fp or "" def restore_backup_index(path: Path, fingerprint: str) -> None: """Restore the encrypted index for ``fingerprint`` from ``path``. This helper is intended for use before full :class:`PasswordManager` initialization. It simply copies the provided backup file into the profile's directory, replacing the existing index if present. A copy of the previous index is kept with a ``.bak`` suffix to allow manual recovery if needed. """ fingerprint_dir = APP_DIR / fingerprint fingerprint_dir.mkdir(parents=True, exist_ok=True) dest = fingerprint_dir / "seedpass_entries_db.json.enc" src = Path(path) # Ensure the source file can be read src.read_bytes() if dest.exists(): shutil.copy2(dest, dest.with_suffix(".bak")) shutil.copy2(src, dest) os.chmod(dest, 0o600) @dataclass class Notification: """Simple message container for UI notifications.""" message: str level: str = "INFO" def requires_unlocked(func): """Decorator to ensure the vault is unlocked before proceeding.""" @wraps(func) def wrapper(self, *args, **kwargs): if getattr(self, "is_locked", False): raise VaultLockedError("Vault is locked") return func(self, *args, **kwargs) return wrapper class AuthGuard: """Helper to enforce inactivity timeouts.""" def __init__( self, manager: "PasswordManager", time_fn: callable = time.time ) -> None: self.manager = manager self._time_fn = time_fn def check_timeout(self) -> None: """Lock the vault if the inactivity timeout has been exceeded.""" timeout = getattr(self.manager, "inactivity_timeout", 0) if self.manager.locked or timeout <= 0: return if self._time_fn() - self.manager.last_activity > timeout: self.manager.lock_vault() class PasswordManager: """ PasswordManager Class Manages the generation, encryption, and retrieval of deterministic passwords using a BIP-85 seed. It handles file encryption/decryption, password generation, entry management, backups, and checksum verification, ensuring the integrity and confidentiality of the stored password database. """ # Class-level fallbacks so attributes exist even if ``__init__`` is bypassed master_key: bytes | None = None KEY_STORAGE: bytes | None = None KEY_INDEX: bytes | None = None KEY_PW_DERIVE: bytes | None = None KEY_TOTP_DET: bytes | None = None deterministic_totp: bool = False def __init__( self, fingerprint: Optional[str] = None, *, password: Optional[str] = None ) -> None: """Initialize the PasswordManager. Parameters ---------- fingerprint: Optional seed profile fingerprint to select without prompting. """ initialize_app() self.ensure_script_checksum() self.encryption_mode: EncryptionMode = EncryptionMode.SEED_ONLY self.encryption_manager: Optional[EncryptionManager] = None self.entry_manager: Optional[EntryManager] = None self.password_generator: Optional[PasswordGenerator] = None self.backup_manager: Optional[BackupManager] = None self.vault: Optional[Vault] = None self.fingerprint_manager: Optional[FingerprintManager] = None self._parent_seed_secret: Optional[InMemorySecret] = None self.bip85: Optional[BIP85] = None self.nostr_client: Optional[NostrClient] = None self.config_manager: Optional[ConfigManager] = None self.state_manager: Optional[StateManager] = None self.stats_manager: StatsManager = StatsManager() self.notifications: queue.Queue[Notification] = queue.Queue() self.error_queue: queue.Queue[Exception] = queue.Queue() self._current_notification: Optional[Notification] = None self._notification_expiry: float = 0.0 self._bip85_cache: dict[tuple[int, int], bytes] = {} self.audit_logger: Optional[AuditLogger] = None # Derived key hierarchy self.master_key: bytes | None = None self.KEY_STORAGE: bytes | None = None self.KEY_INDEX: bytes | None = None self.KEY_PW_DERIVE: bytes | None = None self.KEY_TOTP_DET: bytes | None = None # Track changes to trigger periodic Nostr sync self.is_dirty: bool = False self.last_update: float = time.time() self.last_activity: float = time.time() self.locked: bool = False self.is_locked: bool = False self.inactivity_timeout: float = INACTIVITY_TIMEOUT self.secret_mode_enabled: bool = False self.deterministic_totp: bool = False self.clipboard_clear_delay: int = 45 self.offline_mode: bool = False self.profile_stack: list[tuple[str, Path, str]] = [] self.last_unlock_duration: float | None = None self.verbose_timing: bool = False self._suppress_entry_actions_menu: bool = False self.last_bip85_idx: int = 0 self.last_sync_ts: int = 0 self.nostr_account_idx: int = 0 self.auth_guard = AuthGuard(self) # Service composition self._menu_handler: MenuHandler | None = None self._profile_service: ProfileService | None = None self._entry_service: EntryService | None = None self._display_service: DisplayService | None = None # Initialize service instances self.menu_handler self.profile_service self.entry_service self.display_service # Initialize the fingerprint manager first self.initialize_fingerprint_manager() if fingerprint: # Load the specified profile without prompting self.select_fingerprint(fingerprint, password=password) else: # Ensure a parent seed is set up before accessing the fingerprint directory self.setup_parent_seed() # Set the current fingerprint directory after selection self.fingerprint_dir = ( self.fingerprint_manager.get_current_fingerprint_dir() ) @requires_unlocked def get_bip85_entropy( self, purpose: int, index: int, entropy_bytes: int = 32 ) -> bytes: """Return deterministic entropy via the cached BIP-85 function.""" if self.bip85 is None: raise RuntimeError("BIP-85 is not initialized") return self.bip85.derive_entropy( index=index, entropy_bytes=entropy_bytes, app_no=purpose ) @requires_unlocked def clear_bip85_cache(self) -> None: """Clear the internal BIP-85 cache.""" self._bip85_cache.clear() def derive_key_hierarchy(self, seed_bytes: bytes) -> None: """Populate sub-keys from ``seed_bytes`` using HKDF.""" master = kd(seed_bytes, b"seedpass:v1:master") self.master_key = master self.KEY_STORAGE = kd(master, b"seedpass:v1:storage") self.KEY_INDEX = kd(master, b"seedpass:v1:index") self.KEY_PW_DERIVE = kd(master, b"seedpass:v1:pw") self.KEY_TOTP_DET = kd(master, b"seedpass:v1:totp") def _get_kdf_mode(self) -> str: cfg_mgr = getattr(self, "config_manager", None) getter = getattr(cfg_mgr, "get_kdf_mode", None) if cfg_mgr else None if callable(getter): return getter() return "pbkdf2" def _get_kdf_iterations(self) -> int: floor = getattr(ConfigManager, "DEFAULT_PBKDF2_ITERATIONS", 200_000) cfg_mgr = getattr(self, "config_manager", None) getter = getattr(cfg_mgr, "get_kdf_iterations", None) if cfg_mgr else None if callable(getter): configured = int(getter()) if configured < floor: logger.warning( "Configured PBKDF2 iterations (%s) is below policy floor (%s); using floor.", configured, floor, ) return max(configured, floor) return floor def _get_argon2_time_cost(self) -> int: cfg_mgr = getattr(self, "config_manager", None) getter = getattr(cfg_mgr, "get_argon2_time_cost", None) if cfg_mgr else None if callable(getter): return getter() return 2 def _derive_seed_key( self, password: str, fingerprint: str, *, mode: str | None = None, iterations: int | None = None, ) -> bytes: """Derive the profile seed-encryption key from current KDF settings.""" chosen_mode = mode or self._get_kdf_mode() if chosen_mode == "argon2": salt = hashlib.sha256(fingerprint.encode()).digest()[:16] cfg = KdfConfig( params={ "time_cost": self._get_argon2_time_cost(), "memory_cost": 64 * 1024, "parallelism": 8, }, salt_b64=base64.b64encode(salt).decode(), ) return derive_key_from_password_argon2(password, cfg) if chosen_mode != "pbkdf2": logger.warning("Unknown kdf_mode '%s'; falling back to pbkdf2", chosen_mode) iter_count = ( iterations if iterations is not None else self._get_kdf_iterations() ) return derive_key_from_password(password, fingerprint, iterations=iter_count) def _build_seed_kdf_config( self, fingerprint: str, *, mode: str | None = None, iterations: int | None = None, ) -> KdfConfig: """Return the metadata block describing seed-key derivation policy.""" chosen_mode = mode or self._get_kdf_mode() salt = hashlib.sha256(fingerprint.encode()).digest()[:16] salt_b64 = base64.b64encode(salt).decode() if chosen_mode == "argon2": return KdfConfig( name="argon2id", params={ "time_cost": self._get_argon2_time_cost(), "memory_cost": 64 * 1024, "parallelism": 8, }, salt_b64=salt_b64, ) iter_count = ( int(iterations) if iterations is not None else self._get_kdf_iterations() ) return KdfConfig( name="pbkdf2", params={"iterations": iter_count}, salt_b64=salt_b64, ) def _encrypt_parent_seed_compat( self, seed_mgr: EncryptionManager, seed: str, fingerprint: str, *, mode: str | None = None, ) -> None: """Encrypt parent seed with metadata, falling back for legacy implementations.""" try: seed_mgr.encrypt_parent_seed( seed, kdf=self._build_seed_kdf_config(fingerprint, mode=mode), ) except TypeError: logger.warning( "EncryptionManager.encrypt_parent_seed does not accept kdf metadata; " "falling back to legacy call signature." ) seed_mgr.encrypt_parent_seed(seed) def ensure_key_hierarchy(self) -> None: """Ensure sub-keys are derived from the current parent seed.""" if ( self.KEY_STORAGE is None or self.KEY_INDEX is None or self.KEY_PW_DERIVE is None or self.KEY_TOTP_DET is None ) and getattr(self, "parent_seed", None): try: seed_bytes = Bip39SeedGenerator(self.parent_seed).Generate() except Exception: seed_bytes = hashlib.sha256(self.parent_seed.encode()).digest() self.derive_key_hierarchy(seed_bytes) def ensure_script_checksum(self) -> None: """Initialize or verify the checksum of the manager script.""" script_path = Path(__file__).resolve() if not SCRIPT_CHECKSUM_FILE.exists(): initialize_checksum(str(script_path), SCRIPT_CHECKSUM_FILE) return checksum = calculate_checksum(str(script_path)) if checksum and not verify_checksum(checksum, SCRIPT_CHECKSUM_FILE): logging.warning("Script checksum mismatch detected on startup") print( colored( "Warning: script checksum mismatch. " "Run 'Generate Script Checksum' in Settings if you've updated the app.", "yellow", ) ) @staticmethod def get_password_prompt() -> str: """Return the standard prompt for requesting a master password.""" return "Enter your master password: " @property def parent_seed(self) -> Optional[str]: """Return the decrypted parent seed if set.""" if self._parent_seed_secret is None: return None return self._parent_seed_secret.get_str() @parent_seed.setter def parent_seed(self, value: Optional[str]) -> None: if value is None: if self._parent_seed_secret: self._parent_seed_secret.wipe() self._parent_seed_secret = None else: self._parent_seed_secret = InMemorySecret(value.encode("utf-8")) @property def header_fingerprint(self) -> str | None: """Return the fingerprint chain for header display.""" if not getattr(self, "current_fingerprint", None): return None if not self.profile_stack: return self.current_fingerprint chain = [fp for fp, _path, _seed in self.profile_stack] + [ self.current_fingerprint ] header = chain[0] for fp in chain[1:]: header += f" > Managed Account > {fp}" return header @property def header_fingerprint_args(self) -> tuple[str | None, str | None, str | None]: """Return fingerprint parameters for header display.""" if not getattr(self, "current_fingerprint", None): return (None, None, None) if not self.profile_stack: return (self.current_fingerprint, None, None) parent_fp = self.profile_stack[-1][0] return (None, parent_fp, self.current_fingerprint) def update_activity(self) -> None: """Record activity and enforce inactivity timeout.""" guard = getattr(self, "auth_guard", None) if guard is None: guard = AuthGuard(self) self.auth_guard = guard guard.check_timeout() self.last_activity = time.time() def notify(self, message: str, level: str = "INFO") -> None: """Enqueue a notification and set it as the active message.""" note = Notification(message, level) if not hasattr(self, "notifications"): self.notifications = queue.Queue() self.notifications.put(note) self._current_notification = note self._notification_expiry = time.time() + NOTIFICATION_DURATION def get_current_notification(self) -> Optional[Notification]: """Return the active notification if it hasn't expired.""" if not self.notifications.empty(): latest = self.notifications.queue[-1] if latest is not self._current_notification: self._current_notification = latest self._notification_expiry = time.time() + NOTIFICATION_DURATION if ( self._current_notification is not None and time.time() < self._notification_expiry ): return self._current_notification return None def poll_background_errors(self) -> None: """Process exceptions raised by background threads.""" if not hasattr(self, "error_queue"): return while True: try: exc = self.error_queue.get_nowait() except queue.Empty: break logger.warning("Background task failed: %s", exc) self.notify(f"Background task failed: {exc}", level="WARNING") @property def menu_handler(self) -> MenuHandler: if getattr(self, "_menu_handler", None) is None: self._menu_handler = MenuHandler(self) return self._menu_handler @property def profile_service(self) -> ProfileService: if getattr(self, "_profile_service", None) is None: self._profile_service = ProfileService(self) return self._profile_service @property def entry_service(self) -> EntryService: if getattr(self, "is_locked", False): raise VaultLockedError("Vault is locked") if getattr(self, "_entry_service", None) is None: self._entry_service = EntryService(self) return self._entry_service @property def display_service(self) -> DisplayService: if getattr(self, "is_locked", False): raise VaultLockedError("Vault is locked") if getattr(self, "_display_service", None) is None: self._display_service = DisplayService(self) return self._display_service def lock_vault(self) -> None: """Clear sensitive information from memory.""" if self.entry_manager is not None: self.entry_manager.clear_cache() self.is_locked = True self.locked = True bus.publish("vault_locked") def unlock_vault(self, password: Optional[str] = None) -> float: """Unlock the vault using the provided ``password``. Parameters ---------- password: Master password for the active profile. Returns ------- float Duration of the unlock process in seconds. """ start = time.perf_counter() if not self.fingerprint_dir: raise ValueError("Fingerprint directory not set") if password is None: password = prompt_existing_password(self.get_password_prompt()) self.setup_encryption_manager(self.fingerprint_dir, password) self.initialize_bip85() self.initialize_managers() self.ensure_key_hierarchy() self.is_locked = False self.locked = False self.update_activity() if getattr(self, "audit_logger", None) is None and self.KEY_INDEX is not None: self.audit_logger = AuditLogger(self.KEY_INDEX) if ( getattr(self, "config_manager", None) and self.config_manager.get_quick_unlock() ): logger.info( "Quick unlock used by %s at %s", self.current_fingerprint or "unknown", datetime.now(timezone.utc).isoformat(), ) audit_logger = getattr(self, "audit_logger", None) if audit_logger: audit_logger.log( "quick_unlock", {"fingerprint": self.current_fingerprint} ) self.last_unlock_duration = time.perf_counter() - start if getattr(self, "verbose_timing", False): logger.info("Vault unlocked in %.2f seconds", self.last_unlock_duration) return self.last_unlock_duration def initialize_fingerprint_manager(self): """ Initializes the FingerprintManager. """ try: self.fingerprint_manager = FingerprintManager(APP_DIR) logger.debug("FingerprintManager initialized successfully.") except Exception as e: logger.error(f"Failed to initialize FingerprintManager: {e}", exc_info=True) print( colored(f"Error: Failed to initialize FingerprintManager: {e}", "red") ) raise SeedPassError(f"Failed to initialize FingerprintManager: {e}") from e def setup_parent_seed(self) -> None: """ Sets up the parent seed by determining if existing fingerprints are present or if a new one needs to be created. """ fingerprints = self.fingerprint_manager.list_fingerprints() if fingerprints: # There are existing fingerprints self.select_or_add_fingerprint() else: # No existing fingerprints, proceed to set up new seed self.handle_new_seed_setup() def select_or_add_fingerprint(self): """ Prompts the user to select an existing fingerprint or add a new one. """ try: fingerprints = self.fingerprint_manager.list_fingerprints() current = self.fingerprint_manager.current_fingerprint while True: # Auto-select when only one fingerprint exists if len(fingerprints) == 1 and self.select_fingerprint(fingerprints[0]): return print(colored("\nAvailable Seed Profiles:", "cyan")) for idx, fp in enumerate(fingerprints, start=1): label = ( self.fingerprint_manager.display_name(fp) if hasattr(self.fingerprint_manager, "display_name") else fp ) marker = " *" if fp == current else "" print(colored(f"{idx}. {label}{marker}", "cyan")) add_idx = len(fingerprints) + 1 recover_idx = len(fingerprints) + 2 print(colored(f"{add_idx}. Add a new seed profile", "cyan")) print( colored( f"{recover_idx}. Recover existing profile with blank local index", "cyan", ) ) print(colored("Q. Exit", "cyan")) choice = input("Select a seed profile by number: ").strip() if choice.lower() in {"q", "quit", "exit"}: raise SeedPassError("Operation cancelled by user") if not choice.isdigit() or not ( 1 <= int(choice) <= len(fingerprints) + 2 ): print(colored("Invalid selection. Please try again.", "red")) continue choice_num = int(choice) if choice_num == add_idx: # Add a new seed profile self.add_new_fingerprint() if getattr(self, "encryption_manager", None) is not None: return elif choice_num == recover_idx: if self.recover_profile_with_blank_index(): return else: # Select existing seed profile selected_fingerprint = fingerprints[choice_num - 1] if self.select_fingerprint(selected_fingerprint): return print( colored( "Profile unlock failed or cancelled. You can retry or use recovery.", "yellow", ) ) # refresh profile list if new profiles were added/removed fingerprints = self.fingerprint_manager.list_fingerprints() current = self.fingerprint_manager.current_fingerprint except Exception as e: logger.error(f"Error during seed profile selection: {e}", exc_info=True) print(colored(f"Error: Failed to select seed profile: {e}", "red")) raise SeedPassError(f"Failed to select seed profile: {e}") from e def add_new_fingerprint(self): """ Adds a new seed profile by prompting for encryption mode and generating it from a seed phrase. """ try: choice = input( "Choose an option:\n" "1. Paste in an existing seed in full\n" "2. Enter an existing seed one word at a time\n" "3. Generate a new seed\n" "Enter choice (1/2/3): " ).strip() if choice == "1": fingerprint = self.setup_existing_seed(method="paste") elif choice == "2": fingerprint = self.setup_existing_seed(method="words") elif choice == "3": fingerprint = self.generate_new_seed() else: print(colored("Invalid choice. Exiting.", "red")) raise SeedPassError("Invalid choice.") if not fingerprint: return None # Set current_fingerprint in FingerprintManager only self.fingerprint_manager.current_fingerprint = fingerprint print( colored( f"New seed profile '{fingerprint}' added and set as current.", "green", ) ) # Ensure managers are initialized for the newly created profile if getattr(self, "config_manager", None) is None: if getattr(self, "encryption_manager", None) is None: has_select = hasattr(self.fingerprint_manager, "select_fingerprint") if hasattr(self.fingerprint_manager, "list_fingerprints"): fingerprints = set(self.fingerprint_manager.list_fingerprints()) else: fingerprints = set() if fingerprint in fingerprints and has_select: self.select_fingerprint(fingerprint) elif has_select and not fingerprints: # If state is already staged for the new profile, initialize # directly; otherwise prefer selecting. if getattr(self, "fingerprint_dir", None) is not None: self.current_fingerprint = fingerprint self.initialize_managers() else: self.select_fingerprint(fingerprint) else: # Some flows can pre-populate current profile state before # persistence catches up; initialize managers from current state. self.current_fingerprint = fingerprint if getattr(self, "fingerprint_dir", None) is None: if hasattr(self.fingerprint_manager, "get_fingerprint_dir"): self.fingerprint_dir = ( self.fingerprint_manager.get_fingerprint_dir( fingerprint ) ) else: self.fingerprint_dir = None self.initialize_managers() else: self.initialize_managers() return fingerprint except Exception as e: logger.error(f"Error adding new seed profile: {e}", exc_info=True) print(colored(f"Error: Failed to add new seed profile: {e}", "red")) raise SeedPassError(f"Failed to add new seed profile: {e}") from e def select_fingerprint( self, fingerprint: str, *, password: Optional[str] = None ) -> bool: if self.fingerprint_manager.select_fingerprint(fingerprint): self.current_fingerprint = fingerprint # Add this line self.fingerprint_dir = ( self.fingerprint_manager.get_current_fingerprint_dir() ) if not self.fingerprint_dir: print( colored( f"Error: Seed profile directory for {fingerprint} not found.", "red", ) ) raise SeedPassError( f"Seed profile directory for {fingerprint} not found." ) # Setup the encryption manager and load parent seed if not self.setup_encryption_manager( self.fingerprint_dir, password, exit_on_fail=False, ): return False # Initialize BIP85 and other managers self.initialize_bip85() self.initialize_managers() print( colored( f"Seed profile {fingerprint} selected and managers initialized.", "green", ) ) return True else: print(colored(f"Error: Seed profile {fingerprint} not found.", "red")) raise SeedPassError(f"Seed profile {fingerprint} not found.") def recover_profile_with_blank_index(self) -> bool: """Reinitialize an existing profile with the same seed and blank local index.""" fingerprints = self.fingerprint_manager.list_fingerprints() if not fingerprints: print(colored("No existing profiles available for recovery.", "red")) return False print(colored("\nRecover Existing Profile", "cyan")) for idx, fp in enumerate(fingerprints, start=1): label = ( self.fingerprint_manager.display_name(fp) if hasattr(self.fingerprint_manager, "display_name") else fp ) print(colored(f"{idx}. {label}", "cyan")) choice = input("Select profile to recover by number: ").strip() if not choice.isdigit() or not (1 <= int(choice) <= len(fingerprints)): print(colored("Invalid selection.", "red")) return False fingerprint = fingerprints[int(choice) - 1] fingerprint_dir = self.fingerprint_manager.get_fingerprint_directory( fingerprint ) if not fingerprint_dir: print(colored("Profile directory not found.", "red")) return False print( colored( "This will overwrite the encrypted seed file and reset local entries to a blank index.", "yellow", ) ) if not confirm_action("Continue recovery for this profile? (Y/N): "): return False method = input( "Seed entry method:\n" "1. Paste in an existing seed in full\n" "2. Enter existing seed one word at a time\n" "Enter choice (1/2): " ).strip() if method == "1": parent_seed = masked_input("Enter your 12-word BIP-85 seed: ").strip() elif method == "2": parent_seed = prompt_seed_words() else: print(colored("Invalid choice.", "red")) return False if not self.validate_bip85_seed(parent_seed): print(colored("Error: Invalid BIP-85 seed phrase.", "red")) return False computed_fp = generate_fingerprint(parent_seed) if computed_fp != fingerprint: print( colored( "Error: Seed does not match selected profile fingerprint.", "red", ) ) return False try: password = prompt_for_password() self.current_fingerprint = fingerprint self.fingerprint_manager.current_fingerprint = fingerprint self.fingerprint_dir = fingerprint_dir self.parent_seed = parent_seed seed_bytes = Bip39SeedGenerator(parent_seed).Generate() self.derive_key_hierarchy(seed_bytes) index_key = base64.urlsafe_b64encode(self.KEY_STORAGE) seed_key = self._derive_seed_key(password, fingerprint) self.encryption_manager = EncryptionManager(index_key, fingerprint_dir) seed_mgr = EncryptionManager(seed_key, fingerprint_dir) self.vault = Vault(self.encryption_manager, fingerprint_dir) self.config_manager = ConfigManager( vault=self.vault, fingerprint_dir=fingerprint_dir, ) self._encrypt_parent_seed_compat( seed_mgr, parent_seed, fingerprint, mode=self._get_kdf_mode(), ) self.store_hashed_password(password) for stale_file in ( fingerprint_dir / "seedpass_entries_db.json.enc", fingerprint_dir / "seedpass_passwords_db.json.enc", fingerprint_dir / "seedpass_entries_db_checksum.txt", fingerprint_dir / "seedpass_passwords_db_checksum.txt", ): try: stale_file.unlink() except FileNotFoundError: pass self.initialize_bip85() self.initialize_managers() self.start_background_sync() print( colored( "Profile recovered successfully with a blank local index.", "green", ) ) return True except Exception as exc: logger.error("Profile recovery failed: %s", exc, exc_info=True) print(colored(f"Error: Profile recovery failed: {exc}", "red")) return False def setup_encryption_manager( self, fingerprint_dir: Path, password: Optional[str] = None, *, exit_on_fail: bool = True, ) -> bool: """Set up encryption for the current fingerprint and load the seed.""" attempts = 0 max_attempts = 5 while attempts < max_attempts: try: if password is None: password = prompt_existing_password("Enter your master password: ") mode = ( self.config_manager.get_kdf_mode() if getattr(self, "config_manager", None) else "pbkdf2" ) iterations = ( self.config_manager.get_kdf_iterations() if getattr(self, "config_manager", None) else getattr(ConfigManager, "DEFAULT_PBKDF2_ITERATIONS", 200_000) ) print("Deriving key...") salt_fp = fingerprint_dir.name iter_candidates: list[int] = [iterations] if mode != "argon2": iter_candidates.extend( getattr( ConfigManager, "LEGACY_PBKDF2_ITERATION_FALLBACKS", (50_000, 100_000), ) ) seed_mgr: EncryptionManager | None = None for iter_try in dict.fromkeys(iter_candidates): try: seed_key = self._derive_seed_key( password, salt_fp, mode=mode, iterations=iter_try if mode != "argon2" else None, ) seed_mgr = EncryptionManager(seed_key, fingerprint_dir) print("Decrypting seed...") self.parent_seed = seed_mgr.decrypt_parent_seed() if ( mode != "argon2" and iter_try != iterations and getattr(self, "config_manager", None) ): self.config_manager.set_kdf_iterations(iter_try) break except DecryptionError: seed_mgr = None if seed_mgr is None: msg = "Incorrect password or corrupt file" print(colored(msg, "red")) attempts += 1 password = None continue seed_bytes = Bip39SeedGenerator(self.parent_seed).Generate() self.derive_key_hierarchy(seed_bytes) key_b64 = base64.urlsafe_b64encode(self.KEY_STORAGE) self.encryption_manager = EncryptionManager(key_b64, fingerprint_dir) self.vault = Vault(self.encryption_manager, fingerprint_dir) self.config_manager = ConfigManager( vault=self.vault, fingerprint_dir=fingerprint_dir, ) self.fingerprint_dir = fingerprint_dir if not self.verify_password(password): print(colored("Invalid password. Please try again.", "red")) attempts += 1 password = None continue return True except KeyboardInterrupt: raise except Exception as e: logger.error(f"Failed to set up EncryptionManager: {e}", exc_info=True) print(colored(f"Error: Failed to set up encryption: {e}", "red")) if exit_on_fail: raise SeedPassError(f"Failed to set up encryption: {e}") from e return False if exit_on_fail: raise SeedPassError("Failed to set up encryption") return False def load_parent_seed( self, fingerprint_dir: Path, password: Optional[str] = None ) -> None: """Load and decrypt the parent seed using the password-only key.""" if self.parent_seed: return if password is None: password = prompt_existing_password("Enter your master password: ") try: mode = ( self.config_manager.get_kdf_mode() if getattr(self, "config_manager", None) else "pbkdf2" ) iterations = ( self.config_manager.get_kdf_iterations() if getattr(self, "config_manager", None) else getattr(ConfigManager, "DEFAULT_PBKDF2_ITERATIONS", 200_000) ) salt_fp = fingerprint_dir.name seed_key = self._derive_seed_key( password, salt_fp, mode=mode, iterations=iterations ) seed_mgr = EncryptionManager(seed_key, fingerprint_dir) self.parent_seed = seed_mgr.decrypt_parent_seed() seed_bytes = Bip39SeedGenerator(self.parent_seed).Generate() self.derive_key_hierarchy(seed_bytes) self.bip85 = BIP85(seed_bytes) except DecryptionError as e: logger.error(f"Failed to load parent seed: {e}", exc_info=True) print(colored("Incorrect password or corrupt file", "red")) raise SeedPassError("Incorrect password or corrupt file") from e except Exception as e: logger.error(f"Failed to load parent seed: {e}", exc_info=True) print(colored(f"Error: Failed to load parent seed: {e}", "red")) raise SeedPassError(f"Failed to load parent seed: {e}") from e @pause_logging_for_ui @requires_unlocked def handle_switch_fingerprint(self, *, password: Optional[str] = None) -> bool: return self.profile_service.handle_switch_fingerprint(password=password) @requires_unlocked def load_managed_account(self, index: int) -> None: """Load a managed account or seed profile derived from the current seed.""" if not self.entry_manager or not self.parent_seed: raise ValueError("Manager not initialized") entry = self.entry_manager.retrieve_entry(index) if not entry: raise ValueError(f"Entry #{index} not found") kind = str(entry.get("kind") or entry.get("type") or "").strip().lower() if kind == EntryType.MANAGED_ACCOUNT.value: seed = self.entry_manager.get_managed_account_seed(index, self.parent_seed) elif kind == EntryType.SEED.value: seed = self.entry_manager.get_seed_phrase(index, self.parent_seed) else: raise ValueError(f"Entry #{index} is not a loadable profile (kind: {kind})") managed_fp = generate_fingerprint(seed) account_dir = self.fingerprint_dir / "accounts" / managed_fp account_dir.mkdir(parents=True, exist_ok=True) self.profile_stack.append( (self.current_fingerprint, self.fingerprint_dir, self.parent_seed) ) self.current_fingerprint = managed_fp self.fingerprint_dir = account_dir self.parent_seed = seed seed_bytes = Bip39SeedGenerator(seed).Generate() self.derive_key_hierarchy(seed_bytes) key_b64 = base64.urlsafe_b64encode(self.KEY_STORAGE) self.encryption_manager = EncryptionManager(key_b64, account_dir) self.vault = Vault(self.encryption_manager, account_dir) self.initialize_bip85() self.initialize_managers() self.locked = False self.update_activity() self.start_background_sync() @requires_unlocked def exit_managed_account(self) -> None: """Return to the parent seed profile if one is on the stack.""" if not self.profile_stack: return fp, path, seed = self.profile_stack.pop() self.current_fingerprint = fp self.fingerprint_dir = path self.parent_seed = seed try: seed_bytes = Bip39SeedGenerator(seed).Generate() self.derive_key_hierarchy(seed_bytes) key_b64 = base64.urlsafe_b64encode(self.KEY_STORAGE) except Exception: key_b64 = derive_index_key(seed) self.encryption_manager = EncryptionManager(key_b64, path) self.vault = Vault(self.encryption_manager, path) self.initialize_bip85() self.initialize_managers() self.locked = False self.update_activity() self.start_background_sync() @pause_logging_for_ui def handle_existing_seed(self, *, password: Optional[str] = None) -> None: """ Handles the scenario where an existing parent seed file is found. Prompts the user for the master password to decrypt the seed. """ try: if password is None: password = prompt_existing_password("Enter your login password: ") # Initialize FingerprintManager if not already initialized if not self.fingerprint_manager: self.initialize_fingerprint_manager() # Prompt the user to select an existing seed profile fingerprints = self.fingerprint_manager.list_fingerprints() if not fingerprints: print( colored( "No seed profiles available. Please add a seed profile first.", "red", ) ) raise SeedPassError( "No seed profiles available. Please add a seed profile first." ) print(colored("Available Seed Profiles:", "cyan")) for idx, fp in enumerate(fingerprints, start=1): label = ( self.fingerprint_manager.display_name(fp) if hasattr(self.fingerprint_manager, "display_name") else fp ) print(colored(f"{idx}. {label}", "cyan")) choice = input("Select a seed profile by number: ").strip() if not choice.isdigit() or not (1 <= int(choice) <= len(fingerprints)): print(colored("Invalid selection. Exiting.", "red")) raise SeedPassError("Invalid selection.") selected_fingerprint = fingerprints[int(choice) - 1] self.current_fingerprint = selected_fingerprint fingerprint_dir = self.fingerprint_manager.get_fingerprint_directory( selected_fingerprint ) if not fingerprint_dir: print(colored("Error: Seed profile directory not found.", "red")) raise SeedPassError("Seed profile directory not found.") # Derive encryption key from password using selected fingerprint key = self._derive_seed_key(password, selected_fingerprint) seed_mgr = EncryptionManager(key, fingerprint_dir) self.vault = Vault(seed_mgr, fingerprint_dir) self.parent_seed = seed_mgr.decrypt_parent_seed() seed_bytes = Bip39SeedGenerator(self.parent_seed).Generate() self.derive_key_hierarchy(seed_bytes) key_b64 = base64.urlsafe_b64encode(self.KEY_STORAGE) self.encryption_manager = EncryptionManager(key_b64, fingerprint_dir) self.vault = Vault(self.encryption_manager, fingerprint_dir) logger.debug("Parent seed decrypted successfully.") # Validate the decrypted seed if not self.validate_bip85_seed(self.parent_seed): logging.error("Decrypted seed is invalid. Exiting.") print(colored("Error: Decrypted seed is invalid.", "red")) raise SeedPassError("Decrypted seed is invalid.") self.initialize_bip85() logging.debug("Parent seed decrypted and validated successfully.") except Exception as e: logging.error(f"Failed to decrypt parent seed: {e}", exc_info=True) print(colored(f"Error: Failed to decrypt parent seed: {e}", "red")) raise SeedPassError(f"Failed to decrypt parent seed: {e}") from e @pause_logging_for_ui def handle_new_seed_setup(self) -> None: """ Handles the setup process when no existing parent seed is found. Asks the user whether to enter an existing BIP-85 seed or generate a new one. """ self.notify("No existing seed found. Let's set up a new one!", level="WARNING") choice = input( "Choose an option:\n" "1. Paste in an existing seed in full\n" "2. Enter an existing seed one word at a time\n" "3. Generate a new seed\n" "4. Restore from Nostr\n" "5. Restore from local backup\n" "Enter choice (1/2/3/4/5): " ).strip() if choice == "1": self.setup_existing_seed(method="paste") elif choice == "2": self.setup_existing_seed(method="words") elif choice == "3": self.generate_new_seed() elif choice == "4": seed_phrase = masked_input("Enter your 12-word BIP-85 seed: ").strip() self.restore_from_nostr_with_guidance(seed_phrase) return elif choice == "5": backup_path = input("Enter backup file path: ").strip() if not getattr(self, "fingerprint_manager", None): self.initialize_fingerprint_manager() seed_phrase = masked_input("Enter your 12-word BIP-85 seed: ").strip() fp = self._finalize_existing_seed(seed_phrase) if fp: self.backup_manager.restore_from_backup(backup_path) return else: print(colored("Invalid choice. Exiting.", "red")) raise SeedPassError("Invalid choice.") # Some seed loading paths may not initialize managers; ensure they exist if getattr(self, "config_manager", None) is None: self.initialize_managers() def setup_existing_seed( self, method: Literal["paste", "words"] = "paste", *, seed: Optional[str] = None, password: Optional[str] = None, ) -> Optional[str]: """Prompt for an existing BIP-85 seed and set it up. Parameters ---------- method: ``"paste"`` to enter the entire phrase at once or ``"words"`` to be prompted one word at a time. Returns ------- Optional[str] The fingerprint if setup is successful, ``None`` otherwise. """ try: if seed is not None: parent_seed = seed elif method == "words": parent_seed = prompt_seed_words() else: parent_seed = masked_input("Enter your 12-word BIP-85 seed: ").strip() if not self.validate_bip85_seed(parent_seed): logging.error("Invalid BIP-85 seed phrase. Exiting.") print(colored("Error: Invalid BIP-85 seed phrase.", "red")) raise SeedPassError("Invalid BIP-85 seed phrase.") fingerprint = self._finalize_existing_seed(parent_seed, password=password) return fingerprint except KeyboardInterrupt: logging.info("Operation cancelled by user.") self.notify("Operation cancelled by user.", level="WARNING") raise SeedPassError("Operation cancelled by user.") def setup_existing_seed_word_by_word( self, *, seed: Optional[str] = None, password: Optional[str] = None ) -> Optional[str]: """Prompt for an existing seed one word at a time and set it up.""" return self.setup_existing_seed(method="words", seed=seed, password=password) def _finalize_existing_seed( self, parent_seed: str, *, password: Optional[str] = None ) -> Optional[str]: """Common logic for initializing an existing seed.""" if self.validate_bip85_seed(parent_seed): try: fingerprint = self.fingerprint_manager.add_fingerprint(parent_seed) except ValueError: existing_fp = generate_fingerprint(parent_seed) print(colored("Error: Seed profile already exists.", "red")) if confirm_action("Switch to existing profile? (Y/N): "): self.current_fingerprint = existing_fp self.fingerprint_manager.current_fingerprint = existing_fp self.fingerprint_dir = ( self.fingerprint_manager.get_fingerprint_directory(existing_fp) ) return existing_fp return None if not fingerprint: print( colored( "Error: Failed to generate seed profile for the provided seed.", "red", ) ) raise SeedPassError( "Failed to generate seed profile for the provided seed." ) fingerprint_dir = self.fingerprint_manager.get_fingerprint_directory( fingerprint ) if not fingerprint_dir: print( colored("Error: Failed to retrieve seed profile directory.", "red") ) raise SeedPassError("Failed to retrieve seed profile directory.") self.current_fingerprint = fingerprint self.fingerprint_manager.current_fingerprint = fingerprint self.fingerprint_dir = fingerprint_dir if not getattr(self, "manifest_id", None): self.manifest_id = None logging.info(f"Current seed profile set to {fingerprint}") try: if password is None: password = prompt_for_password() seed_bytes = Bip39SeedGenerator(parent_seed).Generate() self.derive_key_hierarchy(seed_bytes) index_key = base64.urlsafe_b64encode(self.KEY_STORAGE) seed_key = self._derive_seed_key(password, fingerprint) self.encryption_manager = EncryptionManager(index_key, fingerprint_dir) seed_mgr = EncryptionManager(seed_key, fingerprint_dir) self.vault = Vault(self.encryption_manager, fingerprint_dir) self.config_manager = ConfigManager( vault=self.vault, fingerprint_dir=fingerprint_dir, ) self._encrypt_parent_seed_compat(seed_mgr, parent_seed, fingerprint) logging.info("Parent seed encrypted and saved successfully.") self.store_hashed_password(password) logging.info("User password hashed and stored successfully.") self.parent_seed = parent_seed logger.debug("Parent seed initialized successfully.") self.initialize_bip85() self.initialize_managers() self.start_background_sync() return fingerprint except BaseException: self.fingerprint_manager.remove_fingerprint(fingerprint) raise else: logging.error("Invalid BIP-85 seed phrase. Exiting.") print(colored("Error: Invalid BIP-85 seed phrase.", "red")) raise SeedPassError("Invalid BIP-85 seed phrase.") @requires_unlocked def generate_new_seed(self) -> Optional[str]: """ Generates a new BIP-85 seed, displays it to the user, and prompts for confirmation before saving. Returns: Optional[str]: The fingerprint if generation is successful, None otherwise. """ new_seed = self.generate_bip85_seed() print(colored("Your new BIP-85 seed phrase is:", "green")) print(colored(new_seed, "yellow")) print(colored("Please write this down and keep it in a safe place!", "red")) if confirm_action("Do you want to use this generated seed? (Y/N): "): # Determine next account index if state manager is available next_idx = 0 if getattr(self, "state_manager", None) is not None: try: next_idx = self.state_manager.state.get("nostr_account_idx", 0) + 1 except Exception: next_idx = 0 # Add a new fingerprint using the generated seed try: fingerprint = self.fingerprint_manager.add_fingerprint(new_seed) except ValueError: print(colored("Error: Seed profile already exists.", "red")) if confirm_action("Switch to existing profile? (Y/N): "): existing_fp = generate_fingerprint(new_seed) self.current_fingerprint = existing_fp self.fingerprint_manager.current_fingerprint = existing_fp self.fingerprint_dir = ( self.fingerprint_manager.get_fingerprint_directory(existing_fp) ) return existing_fp return None if not fingerprint: print( colored( "Error: Failed to generate seed profile for the new seed.", "red", ) ) raise SeedPassError("Failed to generate seed profile for the new seed.") fingerprint_dir = self.fingerprint_manager.get_fingerprint_directory( fingerprint ) if not fingerprint_dir: print( colored("Error: Failed to retrieve seed profile directory.", "red") ) raise SeedPassError("Failed to retrieve seed profile directory.") # Persist the assigned account index for the new profile try: StateManager(fingerprint_dir).update_state(nostr_account_idx=next_idx) if getattr(self, "state_manager", None) is not None: self.state_manager.update_state(nostr_account_idx=next_idx) self.nostr_account_idx = next_idx except Exception: pass # Set the current fingerprint in both PasswordManager and FingerprintManager self.current_fingerprint = fingerprint self.fingerprint_manager.current_fingerprint = fingerprint logging.info(f"Current seed profile set to {fingerprint}") # Now, save and encrypt the seed with the fingerprint_dir try: self.save_and_encrypt_seed(new_seed, fingerprint_dir) self.start_background_sync() except BaseException: # Clean up partial profile on failure or interruption self.fingerprint_manager.remove_fingerprint(fingerprint) raise return fingerprint # Return the generated fingerprint else: self.notify("Seed generation cancelled. Exiting.", level="WARNING") raise SeedPassError("Seed generation cancelled.") def validate_bip85_seed(self, seed: str) -> bool: """ Validates the provided BIP-85 seed phrase. Parameters: seed (str): The seed phrase to validate. Returns: bool: True if valid, False otherwise. """ try: checker = Mnemonic("english") if checker.check(seed): return True logging.error("Invalid BIP-85 seed provided") return False except Exception as e: logging.error(f"Error validating BIP-85 seed: {e}") return False def generate_bip85_seed(self) -> str: """ Generates a new BIP-85 seed phrase. Returns: str: The generated 12-word mnemonic seed phrase. """ try: master_seed = os.urandom(32) # Generate a random 32-byte seed bip85 = BIP85(master_seed) mnemonic = bip85.derive_mnemonic(index=0, words_num=12) return mnemonic except Bip85Error as e: logging.error(f"Failed to generate BIP-85 seed: {e}", exc_info=True) print(colored(f"Error: Failed to generate BIP-85 seed: {e}", "red")) raise SeedPassError(f"Failed to generate BIP-85 seed: {e}") from e except Exception as e: logging.error(f"Failed to generate BIP-85 seed: {e}", exc_info=True) print(colored(f"Error: Failed to generate BIP-85 seed: {e}", "red")) raise SeedPassError(f"Failed to generate BIP-85 seed: {e}") from e @requires_unlocked def save_and_encrypt_seed( self, seed: str, fingerprint_dir: Path, *, password: Optional[str] = None ) -> None: """ Saves and encrypts the parent seed. Parameters: seed (str): The BIP-85 seed phrase to save and encrypt. fingerprint_dir (Path): The directory corresponding to the fingerprint. """ try: fingerprint = generate_fingerprint(seed) if ( hasattr(self, "fingerprint_manager") and self.fingerprint_manager is not None ): existing = self.fingerprint_manager.list_fingerprints() # Allow saving when creating a new profile that was already # registered via FingerprintManager. Only raise an error if # the fingerprint exists for a *different* directory. if fingerprint in existing and fingerprint_dir.name != fingerprint: print(colored("Error: Seed profile already exists.", "red")) raise ValueError("Fingerprint already exists") # Set self.fingerprint_dir self.fingerprint_dir = fingerprint_dir if password is None: password = prompt_for_password() seed_bytes = Bip39SeedGenerator(seed).Generate() self.derive_key_hierarchy(seed_bytes) index_key = base64.urlsafe_b64encode(self.KEY_STORAGE) seed_key = self._derive_seed_key(password, fingerprint_dir.name) self.encryption_manager = EncryptionManager(index_key, fingerprint_dir) seed_mgr = EncryptionManager(seed_key, fingerprint_dir) self.vault = Vault(self.encryption_manager, fingerprint_dir) # Ensure the config manager points to the new fingerprint before # storing the hashed password self.config_manager = ConfigManager( vault=self.vault, fingerprint_dir=fingerprint_dir, ) self.store_hashed_password(password) logging.info("User password hashed and stored successfully.") self._encrypt_parent_seed_compat( seed_mgr, seed, fingerprint_dir.name, mode=self._get_kdf_mode(), ) logging.info("Parent seed encrypted and saved successfully.") self.parent_seed = seed # Ensure this is a string logger.debug("Parent seed initialized successfully.") self.initialize_bip85() self.initialize_managers() except Exception as e: logging.error(f"Failed to encrypt and save parent seed: {e}", exc_info=True) print(colored(f"Error: Failed to encrypt and save parent seed: {e}", "red")) raise SeedPassError(f"Failed to encrypt and save parent seed: {e}") from e def initialize_bip85(self): """ Initializes the BIP-85 generator using the parent seed. """ try: seed_bytes = Bip39SeedGenerator(self.parent_seed).Generate() self.derive_key_hierarchy(seed_bytes) self.bip85 = BIP85(seed_bytes) self._bip85_cache = {} orig_derive = self.bip85.derive_entropy def cached_derive( index: int, entropy_bytes: int, app_no: int = 39 ) -> bytes: key = (app_no, index) if key not in self._bip85_cache: self._bip85_cache[key] = orig_derive( index=index, entropy_bytes=entropy_bytes, app_no=app_no ) return self._bip85_cache[key] self.bip85.derive_entropy = cached_derive logging.debug("BIP-85 initialized successfully.") # Reset cache directly so initialization is valid even while locked. self._bip85_cache.clear() except Exception as e: logging.error(f"Failed to initialize BIP-85: {e}", exc_info=True) print(colored(f"Error: Failed to initialize BIP-85: {e}", "red")) raise SeedPassError(f"Failed to initialize BIP-85: {e}") from e def initialize_managers(self) -> None: """ Initializes the EntryManager, PasswordGenerator, BackupManager, and NostrClient with the EncryptionManager and BIP-85 instance within the context of the selected fingerprint. """ try: # Ensure self.encryption_manager is already initialized if not self.encryption_manager: raise ValueError("EncryptionManager is not initialized.") # Derive sub-keys if needed self.ensure_key_hierarchy() self._initialize_config_and_state() migrated, last_migration_performed, index_exists = ( self._initialize_entry_manager() ) self._initialize_password_generator() self._initialize_nostr_client() self._handle_migration(migrated, last_migration_performed, index_exists) logger.debug("Managers re-initialized for the new fingerprint.") except Exception as e: logger.error(f"Failed to initialize managers: {e}", exc_info=True) print(colored(f"Error: Failed to initialize managers: {e}", "red")) raise SeedPassError(f"Failed to initialize managers: {e}") from e def _initialize_config_and_state(self) -> None: """Initialize configuration, state, and backup managers.""" self.config_manager = ConfigManager( vault=self.vault, fingerprint_dir=self.fingerprint_dir, ) self.state_manager = StateManager(self.fingerprint_dir) self.backup_manager = BackupManager( fingerprint_dir=self.fingerprint_dir, config_manager=self.config_manager, ) def _initialize_entry_manager(self) -> tuple[bool, bool, bool]: """ Initialize the entry manager and load the vault index. Returns: tuple: (migrated, last_migration_performed, index_exists) """ migrated = False last_migration_performed = False index_exists = ( self.vault.index_file.exists() or (self.vault.fingerprint_dir / "seedpass_passwords_db.json.enc").exists() ) try: _, migrated, last_migration_performed = self.vault.load_index( return_migration_flags=True ) except RuntimeError as exc: print(colored(str(exc), "red")) raise SeedPassError(str(exc)) self.entry_manager = EntryManager( vault=self.vault, backup_manager=self.backup_manager, ) return migrated, last_migration_performed, index_exists def _initialize_password_generator(self) -> None: """Initialize the password generator.""" pw_bip85 = BIP85(self.KEY_PW_DERIVE) self.password_generator = PasswordGenerator( encryption_manager=self.encryption_manager, parent_seed=self.KEY_PW_DERIVE, bip85=pw_bip85, policy=self.config_manager.get_password_policy(), ) def _initialize_nostr_client(self) -> None: """Load relay configuration and initialize NostrClient.""" config = self.config_manager.load_config() if getattr(self, "state_manager", None) is not None: state = self.state_manager.state relay_list = state.get("relays", list(DEFAULT_RELAYS)) self.last_bip85_idx = state.get("last_bip85_idx", 0) self.last_sync_ts = state.get("last_sync_ts", 0) self.manifest_id = state.get("manifest_id") self.delta_since = state.get("delta_since", 0) self.nostr_account_idx = state.get("nostr_account_idx", 0) else: relay_list = list(DEFAULT_RELAYS) self.last_bip85_idx = 0 self.last_sync_ts = 0 self.manifest_id = None self.delta_since = 0 self.nostr_account_idx = 0 self.offline_mode = bool(config.get("offline_mode", False)) self.inactivity_timeout = config.get("inactivity_timeout", INACTIVITY_TIMEOUT) self.secret_mode_enabled = bool(config.get("secret_mode_enabled", False)) self.clipboard_clear_delay = int(config.get("clipboard_clear_delay", 45)) self.verbose_timing = bool(config.get("verbose_timing", False)) if not self.offline_mode: print("Connecting to relays...") if not self.offline_mode and not bool( config.get("online_mode_notice_seen", False) ): self.notify( "Profile is ONLINE by default. Switch to offline mode in Settings if needed.", level="INFO", ) config["online_mode_notice_seen"] = True try: self.config_manager.save_config(config) except Exception as exc: logger.warning( "Failed to persist online mode onboarding notice state: %s", exc, ) self.nostr_client = NostrClient( encryption_manager=self.encryption_manager, fingerprint=self.current_fingerprint, relays=relay_list, offline_mode=self.offline_mode, config_manager=self.config_manager, parent_seed=getattr(self, "parent_seed", None), key_index=self.KEY_INDEX, account_index=self.nostr_account_idx, ) if getattr(self, "manifest_id", None) and hasattr( self.nostr_client, "_state_lock" ): from nostr.backup_models import Manifest with self.nostr_client._state_lock: self.nostr_client.current_manifest_id = self.manifest_id self.nostr_client.current_manifest = Manifest( ver=1, algo="gzip", chunks=[], delta_since=self.delta_since or None, ) def _handle_migration( self, migrated: bool, last_migration_performed: bool, index_exists: bool ) -> None: """Handle database migration notifications and sync.""" if migrated and index_exists: print(colored("Local database migration successful.", "green")) if self.encryption_manager is not None: self.encryption_manager.last_migration_performed = False if last_migration_performed and not self.offline_mode: if confirm_action( "Do you want to sync the migrated profile to Nostr now?" ): result = self.sync_vault() if result: print( colored( "Profile synchronized to Nostr successfully.", "green", ) ) else: print( colored( "Error: Failed to sync profile to Nostr.", "red", ) ) else: print( colored( "You can sync the migrated profile later from the main menu.", "yellow", ) ) async def sync_index_from_nostr_async(self) -> None: """Always fetch the latest vault data from Nostr and update the local index.""" if not getattr(self, "nostr_client", None): return start = time.perf_counter() try: if getattr(self, "current_fingerprint", None): self.nostr_client.fingerprint = self.current_fingerprint result = await self.nostr_client.fetch_latest_snapshot() if not result: if self.nostr_client.last_error: logger.warning( "Unable to fetch latest snapshot from Nostr relays %s: %s", self.nostr_client.relays, self.nostr_client.last_error, ) self.notify( f"Sync failed: {self.nostr_client.last_error}", level="WARNING", ) return manifest, chunks = result encrypted = gzip.decompress(b"".join(chunks)) current = self.vault.get_encrypted_index() updated = False migrated = False if current != encrypted: success, mig = self.vault.decrypt_and_save_index_from_nostr( encrypted, strict=False, merge=False, return_migration_flag=True ) if success: updated = True current = encrypted migrated = migrated or mig if manifest.delta_since: version = int(manifest.delta_since) deltas = await self.nostr_client.fetch_deltas_since(version) for delta in deltas: if current != delta: success, mig = self.vault.decrypt_and_save_index_from_nostr( delta, strict=False, merge=True, return_migration_flag=True, ) if success: updated = True current = delta migrated = migrated or mig if migrated: print(colored("Local database migration successful.", "green")) if self.encryption_manager is not None: self.encryption_manager.last_migration_performed = False if not getattr(self, "offline_mode", False): if confirm_action( "Do you want to sync the migrated profile to Nostr now?" ): try: result = await self.sync_vault_async() if result: print( colored( "Profile synchronized to Nostr successfully.", "green", ) ) else: print( colored( "Error: Failed to sync profile to Nostr.", "red", ) ) except Exception as sync_err: logger.error( "Failed to sync profile to Nostr after migration: %s", sync_err, exc_info=True, ) print( colored( "Error: Failed to sync profile to Nostr.", "red", ) ) else: print( colored( "You can sync the migrated profile later from the main menu.", "yellow", ) ) if updated: logger.info("Local database synchronized from Nostr.") self.notify("Index file updated from Nostr successfully.") except Exception as e: logger.warning( "Unable to sync index from Nostr relays %s: %s", self.nostr_client.relays, e, ) if self.nostr_client.last_error: logger.warning( "NostrClient last error: %s", self.nostr_client.last_error ) self.notify( f"Sync failed: {self.nostr_client.last_error or e}", level="WARNING", ) finally: if getattr(self, "verbose_timing", False): duration = time.perf_counter() - start logger.info("sync_index_from_nostr completed in %.2f seconds", duration) def sync_index_from_nostr(self) -> None: asyncio.run(self.sync_index_from_nostr_async()) def start_background_sync(self) -> None: """Launch a thread to synchronize the vault without blocking the UI.""" if getattr(self, "offline_mode", False): return if getattr(self, "nostr_client", None) is None: return if getattr(self, "_sync_task", None) and not getattr( self._sync_task, "done", True ): return async def _worker() -> None: try: await self.attempt_initial_sync_async() await self.sync_index_from_nostr_async() except Exception as exc: logger.warning(f"Background sync failed: {exc}") if hasattr(self, "error_queue") and getattr(self, "nostr_client", None): self.error_queue.put(exc) try: asyncio.get_running_loop() except RuntimeError: thread = threading.Thread( target=lambda: asyncio.run(_worker()), daemon=True ) thread.start() self._sync_task = thread else: self._sync_task = asyncio.create_task(_worker()) def cleanup(self) -> None: """Cancel any pending background sync task.""" task = getattr(self, "_sync_task", None) if isinstance(task, asyncio.Task) and not task.done(): task.cancel() elif isinstance(task, threading.Thread) and task.is_alive(): task.join(timeout=0.1) def start_background_relay_check(self) -> None: """Check relay health in a background thread.""" if ( hasattr(self, "_relay_thread") and self._relay_thread and self._relay_thread.is_alive() ): return def _worker() -> None: try: if getattr(self, "nostr_client", None) and hasattr( self.nostr_client, "check_relay_health" ): healthy = self.nostr_client.check_relay_health(MIN_HEALTHY_RELAYS) if healthy < MIN_HEALTHY_RELAYS: self.notify( f"Only {healthy} relay(s) responded with your latest event. " "Consider adding more relays via Settings.", level="WARNING", ) except Exception as exc: logger.warning(f"Relay health check failed: {exc}") if hasattr(self, "error_queue"): self.error_queue.put(exc) self._relay_thread = threading.Thread(target=_worker, daemon=True) self._relay_thread.start() def start_background_vault_sync(self, alt_summary: str | None = None) -> None: """Publish the vault to Nostr in a background thread.""" if getattr(self, "offline_mode", False): return def _worker() -> None: try: bus.publish("sync_started") result = asyncio.run(self.sync_vault_async(alt_summary=alt_summary)) bus.publish("sync_finished", result) except Exception as exc: logging.error(f"Background vault sync failed: {exc}", exc_info=True) if hasattr(self, "error_queue"): self.error_queue.put(exc) try: asyncio.get_running_loop() except RuntimeError: threading.Thread(target=_worker, daemon=True).start() else: async def _async_worker() -> None: bus.publish("sync_started") result = await self.sync_vault_async(alt_summary=alt_summary) bus.publish("sync_finished", result) asyncio.create_task(_async_worker()) async def attempt_initial_sync_async(self) -> bool: """Attempt to download the initial vault snapshot from Nostr. Returns ``True`` if the snapshot was successfully downloaded and the local index file was written. Returns ``False`` otherwise. The local index file is not created on failure. """ if not getattr(self, "nostr_client", None): return False index_file = self.fingerprint_dir / "seedpass_entries_db.json.enc" if index_file.exists(): return True have_data = False start = time.perf_counter() try: if getattr(self, "current_fingerprint", None): self.nostr_client.fingerprint = self.current_fingerprint result = await self.nostr_client.fetch_latest_snapshot() if result: manifest, chunks = result encrypted = gzip.decompress(b"".join(chunks)) migrated = False success, mig = self.vault.decrypt_and_save_index_from_nostr( encrypted, strict=False, merge=False, return_migration_flag=True, ) if success: have_data = True migrated = migrated or mig current = encrypted if manifest.delta_since: version = int(manifest.delta_since) deltas = await self.nostr_client.fetch_deltas_since(version) for delta in deltas: if current != delta: success, mig = ( self.vault.decrypt_and_save_index_from_nostr( delta, strict=False, merge=True, return_migration_flag=True, ) ) if success: current = delta migrated = migrated or mig if migrated and not getattr(self, "offline_mode", False): self.start_background_vault_sync() logger.info("Initialized local database from Nostr.") self.notify("Index file updated from Nostr successfully.") except Exception as e: # pragma: no cover - network errors logger.warning(f"Unable to sync index from Nostr: {e}") finally: if getattr(self, "verbose_timing", False): duration = time.perf_counter() - start logger.info("attempt_initial_sync completed in %.2f seconds", duration) return have_data def attempt_initial_sync(self) -> bool: return asyncio.run(self.attempt_initial_sync_async()) def sync_index_from_nostr_if_missing(self) -> None: """Retrieve the password database from Nostr if it doesn't exist locally. If no valid data is found or decryption fails, initialize a fresh local database and publish it to Nostr. """ asyncio.run(self.sync_index_from_nostr_if_missing_async()) async def sync_index_from_nostr_if_missing_async(self) -> None: if not getattr(self, "nostr_client", None): return success = await self.attempt_initial_sync_async() if not success: self.vault.save_index({"schema_version": LATEST_VERSION, "entries": {}}) try: await self.sync_vault_async() except Exception as exc: # pragma: no cover - best effort logger.warning(f"Unable to publish fresh database: {exc}") def check_nostr_backup_exists(self, seed_phrase: str) -> bool: """Return ``True`` if a snapshot exists on Nostr for ``seed_phrase``.""" if not self.nostr_client or getattr(self, "offline_mode", False): return False profile_id = calculate_profile_id(seed_phrase) previous = self.nostr_client.fingerprint try: if hasattr(self, "config_manager") and self.config_manager is not None: cfg = self.config_manager.load_config(require_pin=False) max_retries = int(cfg.get("nostr_max_retries", MAX_RETRIES)) delay = float(cfg.get("nostr_retry_delay", RETRY_DELAY)) else: max_retries = MAX_RETRIES delay = RETRY_DELAY for attempt in range(max_retries): self.nostr_client.fingerprint = profile_id result = asyncio.run(self.nostr_client.fetch_latest_snapshot()) if result is not None: return True if attempt < max_retries - 1: time.sleep(delay * (2**attempt)) return False finally: self.nostr_client.fingerprint = previous def restore_from_nostr_with_guidance(self, seed_phrase: str) -> None: """Restore a profile from Nostr, warning if no backup exists.""" self.notify("Searching for Nostr backup, this may take a moment...") have_backup = self.check_nostr_backup_exists(seed_phrase) if not have_backup: print(colored("No Nostr backup found for this seed profile.", "yellow")) if not confirm_action("Continue with an empty database? (Y/N): "): return fp = self._finalize_existing_seed(seed_phrase) if not fp: return if have_backup: self.notify("Backup found. Restoring vault...") success = self.attempt_initial_sync() if success: self.notify("Vault restored successfully from Nostr.", level="INFO") print(colored("Vault restored from Nostr.", "green")) else: self.notify( "Failed to download or decrypt vault from Nostr.", level="ERROR" ) print(colored("Failed to download vault from Nostr.", "red")) else: self.notify("Starting with a new, empty vault.", level="INFO") return def _clear_header_add_entry(self, subtitle: str) -> None: fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, f"Main Menu > Add Entry > {subtitle}", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) def _mark_dirty(self) -> None: """Mark the database as modified.""" self.is_dirty = True self.last_update = time.time() def _sync_and_pause(self, pause_after: bool = True) -> None: """Sync to Nostr and optionally pause.""" try: self.start_background_vault_sync() except Exception as nostr_error: # pragma: no cover - best effort logging.error( f"Failed to post updated index to Nostr: {nostr_error}", exc_info=True, ) if pause_after: pause() @contextmanager def _entry_op_context(self, operation_name: str): """Standardized context for add entry operations.""" try: yield except Exception as e: logging.error(f"Error during {operation_name} setup: {e}", exc_info=True) print(colored(f"Error: Failed to add {operation_name}: {e}", "red")) pause() def _terminal_document_editor( self, initial_content: str = "", *, title: str = "Document Editor" ) -> str | None: """Simple line-oriented terminal editor. Commands: - `p`: print buffer - `a`: append line - `i `: insert before line n - `e `: edit line n - `d `: delete line n - `w`: save and exit - `q`: cancel """ lines = initial_content.splitlines() print(colored(f"\n{title}", "cyan")) print(colored("Type 'h' for help.", "cyan")) def print_help() -> None: print(colored("Commands: p | a | i | e | d | w | q", "cyan")) def print_lines() -> None: if not lines: print(colored("[empty]", "yellow")) return for idx, line in enumerate(lines, start=1): print(colored(f"{idx:>4}: {line}", "cyan")) print_help() print_lines() while True: raw = input("(doc-edit)> ").rstrip() if not raw: continue cmd, *rest = raw.split(maxsplit=1) cmd = cmd.lower() if cmd == "h": print_help() continue if cmd == "p": print_lines() continue if cmd == "a": lines.append(input("append> ")) continue if cmd in {"i", "e", "d"}: if not rest or not rest[0].strip().isdigit(): print(colored("Line number required.", "red")) continue line_no = int(rest[0].strip()) if line_no < 1 or line_no > max(len(lines), 1): print(colored("Invalid line number.", "red")) continue if cmd == "i": lines.insert(line_no - 1, input("insert> ")) elif cmd == "e": if line_no > len(lines): print(colored("Line does not exist.", "red")) continue lines[line_no - 1] = input("edit> ") else: if line_no > len(lines): print(colored("Line does not exist.", "red")) continue lines.pop(line_no - 1) continue if cmd == "w": return "\n".join(lines) if cmd == "q": return None print(colored("Unknown command.", "red")) @pause_logging_for_ui def handle_add_password(self) -> None: self.entry_service.handle_add_password() @pause_logging_for_ui def handle_add_totp(self) -> None: """Add a TOTP entry either derived from the seed or imported.""" with self._entry_op_context("TOTP"): while True: self._clear_header_add_entry("2FA (TOTP)") print("\nAdd TOTP:") print("1. Make 2FA") print("2. Import 2FA (paste otpauth URI or secret)") choice = input("Select option or press Enter to go back: ").strip() if choice == "1": label = input("Label: ").strip() if not label: print(colored("Error: Label cannot be empty.", "red")) continue period = input("Period (default 30): ").strip() or "30" digits = input("Digits (default 6): ").strip() or "6" if not period.isdigit() or not digits.isdigit(): print( colored("Error: Period and digits must be numbers.", "red") ) continue notes = input("Notes (optional): ").strip() tags_input = input( "Enter tags (comma-separated, optional): " ).strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) entry_id = self.entry_manager.get_next_index() key = self.KEY_TOTP_DET if self.deterministic_totp else None totp_index = ( self.entry_manager.get_next_totp_index() if self.deterministic_totp else None ) uri = self.entry_manager.add_totp( label, key, index=totp_index, period=int(period), digits=int(digits), notes=notes, tags=tags, deterministic=self.deterministic_totp, ) if self.deterministic_totp: secret = TotpManager.derive_secret(key, totp_index or 0) color_cat = "deterministic" else: _lbl, secret, _, _ = TotpManager.parse_otpauth(uri) color_cat = "default" self._mark_dirty() print( colored( f"\n[+] TOTP entry added with ID {entry_id}.\n", "green" ) ) print(colored("Add this URI to your authenticator app:", "cyan")) print(colored(uri, "yellow")) TotpManager.print_qr_code(uri) print(color_text(f"Secret: {secret}\n", color_cat)) self._sync_and_pause() break elif choice == "2": raw = input("Paste otpauth URI or secret: ").strip() try: if raw.lower().startswith("otpauth://"): label, secret, period, digits = TotpManager.parse_otpauth( raw ) else: label = input("Label: ").strip() secret = raw.upper() period = int(input("Period (default 30): ").strip() or 30) digits = int(input("Digits (default 6): ").strip() or 6) notes = input("Notes (optional): ").strip() tags_input = input( "Enter tags (comma-separated, optional): " ).strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) entry_id = self.entry_manager.get_next_index() uri = self.entry_manager.add_totp( label, None, secret=secret, period=period, digits=digits, notes=notes, tags=tags, deterministic=False, ) self._mark_dirty() print( colored( f"\nImported \u2714 Codes for {label} are now stored in SeedPass at ID {entry_id}.", "green", ) ) TotpManager.print_qr_code(uri) self._sync_and_pause() break except ValueError as err: print(colored(f"Error: {err}", "red")) elif not choice: return else: print(colored("Invalid choice.", "red")) @pause_logging_for_ui def handle_add_ssh_key(self) -> None: """Add an SSH key pair entry and display the derived keys.""" with self._entry_op_context("SSH key"): self._clear_header_add_entry("SSH Key") label = input("Label (key): ").strip() if not label: print(colored("Error: Label cannot be empty.", "red")) return notes = input("Notes (optional): ").strip() tags_input = input("Enter tags (comma-separated, optional): ").strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) index = self.entry_manager.add_ssh_key( label, self.parent_seed, notes=notes, tags=tags ) priv_pem, pub_pem = self.entry_manager.get_ssh_key_pair( index, self.parent_seed ) self._mark_dirty() if not confirm_action( "WARNING: Displaying SSH keys reveals sensitive information. Continue? (Y/N): " ): self.notify("SSH key display cancelled.", level="WARNING") return print(colored(f"\n[+] SSH key entry added with ID {index}.\n", "green")) if notes: print(colored(f"Notes: {notes}", "cyan")) print(colored("Public Key:", "cyan")) print(color_text(pub_pem, "default")) print(colored("Private Key:", "cyan")) print(color_text(priv_pem, "deterministic")) self._sync_and_pause() @pause_logging_for_ui def handle_add_seed(self) -> None: """Add a derived BIP-39 seed phrase entry.""" with self._entry_op_context("seed phrase"): self._clear_header_add_entry("Seed Phrase") label = input("Label: ").strip() if not label: print(colored("Error: Label cannot be empty.", "red")) return words_input = input("Word count (12 or 24, default 24): ").strip() notes = input("Notes (optional): ").strip() tags_input = input("Enter tags (comma-separated, optional): ").strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) if words_input and words_input not in {"12", "24"}: print(colored("Invalid word count. Choose 12 or 24.", "red")) return words = int(words_input) if words_input else 24 index = self.entry_manager.add_seed( label, self.parent_seed, words_num=words, notes=notes, tags=tags ) phrase = self.entry_manager.get_seed_phrase(index, self.parent_seed) self._mark_dirty() if not confirm_action( "WARNING: Displaying the seed phrase reveals sensitive information. Continue? (Y/N): " ): self.notify("Seed phrase display cancelled.", level="WARNING") return print( colored( f"\n[+] Seed entry '{label}' added with ID {index}.\n", "green", ) ) print(colored(f"Index: {index}", "cyan")) print(colored(f"Label: {label}", "cyan")) if notes: print(colored(f"Notes: {notes}", "cyan")) print(colored("Seed Phrase:", "cyan")) print(color_text(phrase, "deterministic")) if confirm_action("Show Compact Seed QR? (Y/N): "): from .seedqr import encode_seedqr TotpManager.print_qr_code(encode_seedqr(phrase)) self._sync_and_pause() @pause_logging_for_ui def handle_add_pgp(self) -> None: """Add a PGP key entry and display the generated key.""" with self._entry_op_context("PGP key"): self._clear_header_add_entry("PGP Key") label = input("Label: ").strip() if not label: print(colored("Error: Label cannot be empty.", "red")) return key_type = ( input("Key type (ed25519 or rsa, default ed25519): ").strip().lower() or "ed25519" ) user_id = input("User ID (optional): ").strip() notes = input("Notes (optional): ").strip() tags_input = input("Enter tags (comma-separated, optional): ").strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) index = self.entry_manager.add_pgp_key( label, self.parent_seed, key_type=key_type, user_id=user_id, notes=notes, tags=tags, ) priv_key, pub_key, fingerprint = self.entry_manager.get_pgp_key( index, self.parent_seed ) self._mark_dirty() if not confirm_action( "WARNING: Displaying the PGP key reveals sensitive information. Continue? (Y/N): " ): self.notify("PGP key display cancelled.", level="WARNING") return print(colored(f"\n[+] PGP key entry added with ID {index}.\n", "green")) if user_id: print(colored(f"User ID: {user_id}", "cyan")) if notes: print(colored(f"Notes: {notes}", "cyan")) print(colored(f"Fingerprint: {fingerprint}", "cyan")) print(color_text(priv_key, "deterministic")) self._sync_and_pause() @pause_logging_for_ui def handle_add_nostr_key(self) -> None: """Add a Nostr key entry and display the derived keys.""" with self._entry_op_context("Nostr key"): self._clear_header_add_entry("Nostr Key Pair") label = input("Label: ").strip() if not label: print(colored("Error: Label cannot be empty.", "red")) return notes = input("Notes (optional): ").strip() tags_input = input("Enter tags (comma-separated, optional): ").strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) index = self.entry_manager.add_nostr_key( label, self.parent_seed, notes=notes, tags=tags ) npub, nsec = self.entry_manager.get_nostr_key_pair(index, self.parent_seed) self._mark_dirty() print(colored(f"\n[+] Nostr key entry added with ID {index}.\n", "green")) print(colored(f"npub: {npub}", "cyan")) if self.secret_mode_enabled: if copy_to_clipboard(nsec, self.clipboard_clear_delay): print( colored( f"[+] nsec copied to clipboard. Will clear in {self.clipboard_clear_delay} seconds.", "green", ) ) else: print(color_text(f"nsec: {nsec}", "deterministic")) if confirm_action("Show QR code for npub? (Y/N): "): TotpManager.print_qr_code(f"nostr:{npub}") if confirm_action( "WARNING: Displaying the nsec QR reveals your private key. Continue? (Y/N): " ): TotpManager.print_qr_code(nsec) self._sync_and_pause() @pause_logging_for_ui def handle_add_key_value(self) -> None: """Add a generic key/value entry.""" with self._entry_op_context("key/value"): self._clear_header_add_entry("Key/Value") label = input("Label: ").strip() if not label: print(colored("Error: Label cannot be empty.", "red")) return key_field = input("Key: ").strip() if not key_field: print(colored("Error: Key cannot be empty.", "red")) return value = input("Value: ").strip() notes = input("Notes (optional): ").strip() tags_input = input("Enter tags (comma-separated, optional): ").strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) custom_fields: list[dict[str, object]] = [] while True: add_field = input("Add custom field? (y/N): ").strip().lower() if add_field != "y": break field_label = input(" Field label: ").strip() field_value = input(" Field value: ").strip() hidden = input(" Hidden field? (y/N): ").strip().lower() == "y" custom_fields.append( { "label": field_label, "value": field_value, "is_hidden": hidden, } ) index = self.entry_manager.add_key_value( label, key_field, value, notes=notes, custom_fields=custom_fields, tags=tags, ) self._mark_dirty() print(colored(f"\n[+] Key/Value entry added with ID {index}.\n", "green")) if notes: print(colored(f"Notes: {notes}", "cyan")) if self.secret_mode_enabled: if copy_to_clipboard(value, self.clipboard_clear_delay): print( colored( f"[+] Value copied to clipboard. Will clear in {self.clipboard_clear_delay} seconds.", "green", ) ) else: print(color_text(f"Value: {value}", "deterministic")) self._sync_and_pause() @pause_logging_for_ui def handle_add_document(self) -> None: """Add a document entry with editable text content.""" with self._entry_op_context("document"): self._clear_header_add_entry("Document") print("1. New Document") print("2. Import Document File") mode = input("Select option or press Enter to cancel: ").strip() if not mode: return if mode not in {"1", "2"}: print(colored("Invalid choice.", "red")) return if mode == "2": file_path = input("Path to document file: ").strip() if not file_path: print(colored("Error: File path cannot be empty.", "red")) return label = input("Title override (optional): ").strip() or None notes = input("Notes (optional): ").strip() tags_input = input("Enter tags (comma-separated, optional): ").strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) index = self.entry_manager.import_document_file( file_path, label=label, notes=notes, tags=tags, ) else: label = input("Title: ").strip() if not label: print(colored("Error: Title cannot be empty.", "red")) return file_type = ( input("File type/extension (default txt): ").strip() or "txt" ) notes = input("Notes (optional): ").strip() tags_input = input("Enter tags (comma-separated, optional): ").strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) content = self._terminal_document_editor("", title=f"Editing: {label}") if content is None: print(colored("Document creation cancelled.", "yellow")) return index = self.entry_manager.add_document( label, content, file_type=file_type, notes=notes, tags=tags, ) self._mark_dirty() print(colored(f"\n[+] Document entry added with ID {index}.\n", "green")) self._sync_and_pause() @pause_logging_for_ui def handle_add_managed_account(self) -> None: """Add a managed account seed entry.""" with self._entry_op_context("managed account"): self._clear_header_add_entry("Managed Account") label = input("Label: ").strip() if not label: print(colored("Error: Label cannot be empty.", "red")) return notes = input("Notes (optional): ").strip() tags_input = input("Enter tags (comma-separated, optional): ").strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) index = self.entry_manager.add_managed_account( label, self.parent_seed, notes=notes, tags=tags ) self._mark_dirty() seed = self.entry_manager.get_managed_account_seed(index, self.parent_seed) print( colored( f"\n[+] Managed account '{label}' added with ID {index}.\n", "green", ) ) if confirm_action("Reveal seed now? (y/N): "): if self.secret_mode_enabled: if copy_to_clipboard(seed, self.clipboard_clear_delay): print( colored( f"[+] Seed copied to clipboard. Will clear in {self.clipboard_clear_delay} seconds.", "green", ) ) else: print(color_text(seed, "deterministic")) if confirm_action("Show Compact Seed QR? (Y/N): "): from .seedqr import encode_seedqr TotpManager.print_qr_code(encode_seedqr(seed)) self._sync_and_pause() def show_entry_details_by_index(self, index: int) -> None: """Display details for entry ``index`` and offer actions.""" try: entry = self.entry_manager.retrieve_entry(index) if not entry: return fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Entry Details", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) self.display_entry_details(index) self.display_sensitive_entry_info(entry, index) pause() self._entry_actions_menu(index, entry) except Exception as e: logging.error(f"Failed to display entry details: {e}", exc_info=True) print(colored(f"Error: Failed to display entry details: {e}", "red")) pause() def _prompt_toggle_archive(self, entry: dict, index: int) -> None: """Prompt the user to archive or restore ``entry`` based on its status.""" archived = entry.get("archived", entry.get("blacklisted", False)) prompt = ( "Restore this entry from archive? (y/N): " if archived else "Archive this entry? (y/N): " ) choice = input(prompt).strip().lower() if choice == "y": if archived: self.entry_manager.restore_entry(index) else: self.entry_manager.archive_entry(index) self.is_dirty = True self.last_update = time.time() def _entry_type_str(self, entry: dict) -> str: """Return the entry type as a lowercase string.""" entry_type = entry.get("type", entry.get("kind", EntryType.PASSWORD.value)) if isinstance(entry_type, EntryType): entry_type = entry_type.value return str(entry_type).lower() def _generate_password_for_entry( self, entry: dict, index: int, length: int | None = None ) -> str: """Generate a password for ``entry`` honoring any policy overrides.""" if length is None: length = int(entry.get("length", DEFAULT_PASSWORD_LENGTH)) overrides = entry.get("policy", {}) pg = self.password_generator if not hasattr(pg, "policy") or not isinstance(overrides, dict): return pg.generate_password(length, index) base_policy = pg.policy merged = dataclasses.replace( base_policy, **{k: overrides[k] for k in overrides if hasattr(base_policy, k)}, ) pg.policy = merged try: return pg.generate_password(length, index) finally: pg.policy = base_policy def _entry_actions_menu(self, index: int, entry: dict) -> None: """Provide actions for a retrieved entry.""" # Keep entry dict updated in closure scope current_entry = entry def refresh_entry(): nonlocal current_entry refreshed = self.entry_manager.retrieve_entry(index) if refreshed: current_entry = refreshed def toggle_archive(): archived = current_entry.get( "archived", current_entry.get("blacklisted", False) ) if archived: self.entry_manager.restore_entry(index) else: self.entry_manager.archive_entry(index) self.is_dirty = True self.last_update = time.time() refresh_entry() def add_note(): note = input("Enter note: ").strip() if note: notes = current_entry.get("notes", "") notes = f"{notes}\n{note}" if notes else note self.entry_manager.modify_entry(index, notes=notes) self.is_dirty = True self.last_update = time.time() refresh_entry() def add_custom_field(hidden: bool): label = input(" Field label: ").strip() if not label: print(colored("Field label cannot be empty.", "red")) else: value = input(" Field value: ").strip() custom_fields = current_entry.get("custom_fields", []) custom_fields.append( {"label": label, "value": value, "is_hidden": hidden} ) self.entry_manager.modify_entry(index, custom_fields=custom_fields) self.is_dirty = True self.last_update = time.time() refresh_entry() def edit_tags(): current_tags = current_entry.get("tags", []) print( colored( f"Current tags: {', '.join(current_tags) if current_tags else 'None'}", "cyan", ) ) tags_input = input( "Enter tags (comma-separated, leave blank to remove all tags): " ).strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else [] ) self.entry_manager.modify_entry(index, tags=tags) self.is_dirty = True self.last_update = time.time() refresh_entry() def edit_entry(): self._entry_edit_menu(index, current_entry) refresh_entry() def show_qr(): self._entry_qr_menu(index, current_entry) pause() refresh_entry() def export_document(): out = input( "Output path (file or directory, blank for current directory): " ).strip() try: dest = self.entry_manager.export_document_file( index, output_path=out or None ) print(colored(f"Document exported to: {dest}", "green")) except Exception as exc: print(colored(f"Export failed: {exc}", "red")) pause() refresh_entry() def get_options(): archived = current_entry.get( "archived", current_entry.get("blacklisted", False) ) entry_type = self._entry_type_str(current_entry) options = [] if archived: options.append(("u", "Unarchive", toggle_archive)) else: options.append(("a", "Archive", toggle_archive)) options.append(("n", "Add Note", add_note)) options.append(("c", "Add Custom Field", lambda: add_custom_field(False))) options.append(("h", "Add Hidden Field", lambda: add_custom_field(True))) options.append(("e", "Edit", edit_entry)) options.append(("t", "Edit Tags", edit_tags)) if entry_type in { EntryType.SEED.value, EntryType.MANAGED_ACCOUNT.value, EntryType.NOSTR.value, }: options.append(("q", "Show QR codes", show_qr)) if entry_type == EntryType.DOCUMENT.value: options.append(("x", "Export Document to File", export_document)) return options self.menu_handler.run_menu("Entry Actions", get_options) def _entry_edit_menu(self, index: int, entry: dict) -> None: """Sub-menu for editing common entry fields.""" # Keep entry dict updated in closure scope current_entry = entry def refresh_entry(): nonlocal current_entry refreshed = self.entry_manager.retrieve_entry(index) if refreshed: current_entry = refreshed def edit_label(): new_label = input("New label: ").strip() if new_label: self.entry_manager.modify_entry(index, label=new_label) self.is_dirty = True self.last_update = time.time() refresh_entry() def edit_key(): new_key = input("New key: ").strip() if new_key: self.entry_manager.modify_entry(index, key=new_key) self.is_dirty = True self.last_update = time.time() refresh_entry() def edit_value(): new_value = input("New value: ").strip() if new_value: self.entry_manager.modify_entry(index, value=new_value) self.is_dirty = True self.last_update = time.time() refresh_entry() def edit_document_content(): new_content = self._terminal_document_editor( current_entry.get("content", ""), title=f"Editing: {current_entry.get('label', 'Document')}", ) if new_content is None: print(colored("Edit cancelled.", "yellow")) return self.entry_manager.modify_entry(index, content=new_content) self.is_dirty = True self.last_update = time.time() refresh_entry() def edit_document_file_type(): current = current_entry.get("file_type", "txt") new_file_type = ( input(f"New file type/extension (current: {current}): ").strip() or current ) self.entry_manager.modify_entry(index, file_type=new_file_type) self.is_dirty = True self.last_update = time.time() refresh_entry() def edit_username(): new_username = input("New username: ").strip() self.entry_manager.modify_entry(index, username=new_username) self.is_dirty = True self.last_update = time.time() refresh_entry() def edit_url(): new_url = input("New URL: ").strip() self.entry_manager.modify_entry(index, url=new_url) self.is_dirty = True self.last_update = time.time() refresh_entry() def edit_period(): period_str = input("New period (seconds): ").strip() if period_str.isdigit(): self.entry_manager.modify_entry(index, period=int(period_str)) self.is_dirty = True self.last_update = time.time() refresh_entry() else: print(colored("Invalid period value.", "red")) def edit_digits(): digits_str = input("New digits: ").strip() if digits_str.isdigit(): self.entry_manager.modify_entry(index, digits=int(digits_str)) self.is_dirty = True self.last_update = time.time() refresh_entry() else: print(colored("Invalid digits value.", "red")) def get_options(): entry_type = self._entry_type_str(current_entry) options = [("l", "Edit Label", edit_label)] if entry_type == EntryType.KEY_VALUE.value: options.append(("k", "Edit Key", edit_key)) options.append(("v", "Edit Value", edit_value)) elif entry_type == EntryType.DOCUMENT.value: options.append(("c", "Edit Content", edit_document_content)) options.append(("f", "Edit File Type", edit_document_file_type)) if entry_type == EntryType.PASSWORD.value: options.append(("u", "Edit Username", edit_username)) options.append(("r", "Edit URL", edit_url)) elif entry_type == EntryType.TOTP.value: options.append(("p", "Edit Period", edit_period)) options.append(("d", "Edit Digits", edit_digits)) return options self.menu_handler.run_menu( "Edit Entry", get_options, prompt="Select option or press Enter to go back: ", ) def _entry_qr_menu(self, index: int, entry: dict) -> None: """Display QR codes for the given ``entry``.""" entry_type = self._entry_type_str(entry) try: if entry_type in {EntryType.SEED.value, EntryType.MANAGED_ACCOUNT.value}: if entry_type == EntryType.SEED.value: seed = self.entry_manager.get_seed_phrase(index, self.parent_seed) else: seed = self.entry_manager.get_managed_account_seed( index, self.parent_seed ) print(color_text(seed, "deterministic")) from .seedqr import encode_seedqr TotpManager.print_qr_code(encode_seedqr(seed)) pause() return if entry_type == EntryType.NOSTR.value: current_entry = entry def refresh_entry(): nonlocal current_entry refreshed = self.entry_manager.retrieve_entry(index) if refreshed: current_entry = refreshed def show_public_key(): npub, _ = self.entry_manager.get_nostr_key_pair( index, self.parent_seed ) print(colored(f"npub: {npub}", "cyan")) TotpManager.print_qr_code(f"nostr:{npub}") pause() refresh_entry() def show_private_key(): _, nsec = self.entry_manager.get_nostr_key_pair( index, self.parent_seed ) print(color_text(f"nsec: {nsec}", "deterministic")) TotpManager.print_qr_code(nsec) pause() refresh_entry() def get_options(): return [ ("p", "Public key", show_public_key), ("k", "Private key", show_private_key), ] self.menu_handler.run_menu( "QR Codes", get_options, prompt="Select option or press Enter to return: ", ) return self.notify("No QR codes available for this entry.", level="WARNING") except Exception as e: # pragma: no cover - best effort logging.error(f"Error displaying QR menu: {e}", exc_info=True) print(colored(f"Error: Failed to display QR codes: {e}", "red")) def display_sensitive_entry_info(self, entry: dict, index: int) -> None: """Display information for a sensitive entry. Parameters ---------- entry: dict Entry data retrieved from the vault. index: int Index of the entry being displayed. """ self.display_service.display_sensitive_entry_info(entry, index) @pause_logging_for_ui def handle_retrieve_entry(self) -> None: """Prompt for an index and display the corresponding entry.""" try: fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Retrieve Entry", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) index_input = input( "Enter the index number of the entry to retrieve: " ).strip() if not index_input.isdigit(): print(colored("Error: Index must be a number.", "red")) pause() return index = int(index_input) entry = self.entry_manager.retrieve_entry(index) if not entry: pause() return self.display_sensitive_entry_info(entry, index) pause() self._entry_actions_menu(index, entry) return except Exception as e: logging.error(f"Error during password retrieval: {e}", exc_info=True) print(colored(f"Error: Failed to retrieve password: {e}", "red")) pause() @pause_logging_for_ui def handle_modify_entry(self) -> None: """ Handles modifying an existing password entry by prompting the user for the index number and new details to update. """ try: fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Modify Entry", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) index_input = input( "Enter the index number of the entry to modify: " ).strip() if not index_input.isdigit(): print(colored("Error: Index must be a number.", "red")) return index = int(index_input) # Retrieve existing entry entry = self.entry_manager.retrieve_entry(index) if not entry: return entry_type = self._entry_type_str(entry) if entry_type == EntryType.TOTP.value: label = entry.get("label", "") period = int(entry.get("period", 30)) digits = int(entry.get("digits", 6)) blacklisted = entry.get("archived", entry.get("blacklisted", False)) notes = entry.get("notes", "") print( colored( f"Modifying 2FA entry '{label}' (Index: {index}):", "cyan", ) ) print(colored(f"Current Period: {period}s", "cyan")) print(colored(f"Current Digits: {digits}", "cyan")) print( colored( f"Current Archived Status: {'Archived' if blacklisted else 'Active'}", "cyan", ) ) new_label = ( input(f'Enter new label (leave blank to keep "{label}"): ').strip() or label ) period_input = input( f"Enter new period in seconds (current: {period}): " ).strip() new_period = period if period_input: if period_input.isdigit(): new_period = int(period_input) else: self.notify( "Invalid period value. Keeping current.", level="WARNING", ) digits_input = input( f"Enter new digit count (current: {digits}): " ).strip() new_digits = digits if digits_input: if digits_input.isdigit(): new_digits = int(digits_input) else: self.notify( "Invalid digits value. Keeping current.", level="WARNING", ) blacklist_input = ( input( f'Archive this 2FA code? (Y/N, current: {"Y" if blacklisted else "N"}): ' ) .strip() .lower() ) if blacklist_input == "": new_blacklisted = blacklisted elif blacklist_input == "y": new_blacklisted = True elif blacklist_input == "n": new_blacklisted = False else: self.notify( "Invalid input for archived status. Keeping the current status.", level="WARNING", ) new_blacklisted = blacklisted new_notes = ( input( f'Enter new notes (leave blank to keep "{notes or "N/A"}"): ' ).strip() or notes ) edit_fields = input("Edit custom fields? (y/N): ").strip().lower() custom_fields = None if edit_fields == "y": custom_fields = [] while True: label = input(" Field label (leave blank to finish): ").strip() if not label: break value = input(" Field value: ").strip() hidden = input(" Hidden field? (y/N): ").strip().lower() == "y" custom_fields.append( {"label": label, "value": value, "is_hidden": hidden} ) tags_input = input( "Enter tags (comma-separated, leave blank to keep current): " ).strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else None ) self.entry_manager.modify_entry( index, archived=new_blacklisted, notes=new_notes, label=new_label, period=new_period, digits=new_digits, custom_fields=custom_fields, tags=tags, ) elif entry_type == EntryType.SSH.value: label = entry.get("label", "") blacklisted = entry.get("archived", False) notes = entry.get("notes", "") print( colored( f"Modifying SSH key entry '{label}' (Index: {index}):", "cyan", ) ) print( colored( f"Current Archived Status: {'Archived' if blacklisted else 'Active'}", "cyan", ) ) new_label = ( input(f'Enter new label (leave blank to keep "{label}"): ').strip() or label ) blacklist_input = ( input( f'Archive this entry? (Y/N, current: {"Y" if blacklisted else "N"}): ' ) .strip() .lower() ) if blacklist_input == "": new_blacklisted = blacklisted elif blacklist_input == "y": new_blacklisted = True elif blacklist_input == "n": new_blacklisted = False else: self.notify( "Invalid input for archived status. Keeping the current status.", level="WARNING", ) new_blacklisted = blacklisted new_notes = ( input( f'Enter new notes (leave blank to keep "{notes or "N/A"}"): ' ).strip() or notes ) tags_input = input( "Enter tags (comma-separated, leave blank to keep current): " ).strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else None ) self.entry_manager.modify_entry( index, archived=new_blacklisted, notes=new_notes, label=new_label, tags=tags, ) elif entry_type == EntryType.DOCUMENT.value: label = entry.get("label", "") content = entry.get("content", "") file_type = entry.get("file_type", "txt") blacklisted = entry.get("archived", False) notes = entry.get("notes", "") print( colored( f"Modifying document '{label}' (Index: {index}):", "cyan", ) ) new_label = ( input(f'Enter new title (leave blank to keep "{label}"): ').strip() or label ) new_file_type = ( input( f'Enter file type/extension (leave blank to keep "{file_type}"): ' ).strip() or file_type ) if input("Edit document content now? (y/N): ").strip().lower() == "y": edited = self._terminal_document_editor( content, title=f"Editing: {new_label}" ) new_content = content if edited is None else edited else: new_content = content blacklist_input = ( input( f'Archive this entry? (Y/N, current: {"Y" if blacklisted else "N"}): ' ) .strip() .lower() ) if blacklist_input == "": new_blacklisted = blacklisted elif blacklist_input == "y": new_blacklisted = True elif blacklist_input == "n": new_blacklisted = False else: self.notify( "Invalid input for archived status. Keeping the current status.", level="WARNING", ) new_blacklisted = blacklisted new_notes = ( input( f'Enter new notes (leave blank to keep "{notes or "N/A"}"): ' ).strip() or notes ) tags_input = input( "Enter tags (comma-separated, leave blank to keep current): " ).strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else None ) self.entry_manager.modify_entry( index, archived=new_blacklisted, notes=new_notes, label=new_label, content=new_content, file_type=new_file_type, tags=tags, ) elif entry_type in ( EntryType.KEY_VALUE.value, EntryType.MANAGED_ACCOUNT.value, ): label = entry.get("label", "") value = entry.get("value", "") blacklisted = entry.get("archived", False) notes = entry.get("notes", "") entry_label = ( "key/value entry" if entry_type == EntryType.KEY_VALUE.value else "managed account" ) print( colored( f"Modifying {entry_label} '{label}' (Index: {index}):", "cyan", ) ) print( colored( f"Current Archived Status: {'Archived' if blacklisted else 'Active'}", "cyan", ) ) new_label = ( input(f'Enter new label (leave blank to keep "{label}"): ').strip() or label ) if entry_type == EntryType.KEY_VALUE.value: new_key = input( f'Enter new key (leave blank to keep "{entry.get("key", "")}"): ' ).strip() or entry.get("key", "") else: new_key = None new_value = ( input("Enter new value (leave blank to keep current): ").strip() or value ) blacklist_input = ( input( f'Archive this entry? (Y/N, current: {"Y" if blacklisted else "N"}): ' ) .strip() .lower() ) if blacklist_input == "": new_blacklisted = blacklisted elif blacklist_input == "y": new_blacklisted = True elif blacklist_input == "n": new_blacklisted = False else: self.notify( "Invalid input for archived status. Keeping the current status.", level="WARNING", ) new_blacklisted = blacklisted new_notes = ( input( f'Enter new notes (leave blank to keep "{notes or "N/A"}"): ' ).strip() or notes ) edit_fields = input("Edit custom fields? (y/N): ").strip().lower() custom_fields = None if edit_fields == "y": custom_fields = [] while True: f_label = input( " Field label (leave blank to finish): " ).strip() if not f_label: break f_value = input(" Field value: ").strip() hidden = input(" Hidden field? (y/N): ").strip().lower() == "y" custom_fields.append( {"label": f_label, "value": f_value, "is_hidden": hidden} ) tags_input = input( "Enter tags (comma-separated, leave blank to keep current): " ).strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else None ) modify_kwargs = { "archived": new_blacklisted, "notes": new_notes, "label": new_label, "value": new_value, "custom_fields": custom_fields, "tags": tags, } if entry_type == EntryType.KEY_VALUE.value: modify_kwargs["key"] = new_key self.entry_manager.modify_entry( index, **modify_kwargs, ) else: website_name = entry.get("label", entry.get("website")) username = entry.get("username") url = entry.get("url") blacklisted = entry.get("archived", entry.get("blacklisted")) notes = entry.get("notes", "") print( colored( f"Modifying entry for '{website_name}' (Index: {index}):", "cyan", ) ) print(colored(f"Current Label: {website_name}", "cyan")) print(colored(f"Current Username: {username or 'N/A'}", "cyan")) print(colored(f"Current URL: {url or 'N/A'}", "cyan")) print( colored( f"Current Archived Status: {'Archived' if blacklisted else 'Active'}", "cyan", ) ) new_label = ( input( f'Enter new label (leave blank to keep "{website_name}"): ' ).strip() or website_name ) new_username = ( input( f'Enter new username (leave blank to keep "{username or "N/A"}"): ' ).strip() or username ) new_url = ( input( f'Enter new URL (leave blank to keep "{url or "N/A"}"): ' ).strip() or url ) blacklist_input = ( input( f'Archive this password? (Y/N, current: {"Y" if blacklisted else "N"}): ' ) .strip() .lower() ) if blacklist_input == "": new_blacklisted = blacklisted elif blacklist_input == "y": new_blacklisted = True elif blacklist_input == "n": new_blacklisted = False else: self.notify( "Invalid input for archived status. Keeping the current status.", level="WARNING", ) new_blacklisted = blacklisted new_notes = ( input( f'Enter new notes (leave blank to keep "{notes or "N/A"}"): ' ).strip() or notes ) edit_fields = input("Edit custom fields? (y/N): ").strip().lower() custom_fields = None if edit_fields == "y": custom_fields = [] while True: label = input(" Field label (leave blank to finish): ").strip() if not label: break value = input(" Field value: ").strip() hidden = input(" Hidden field? (y/N): ").strip().lower() == "y" custom_fields.append( {"label": label, "value": value, "is_hidden": hidden} ) tags_input = input( "Enter tags (comma-separated, leave blank to keep current): " ).strip() tags = ( [t.strip() for t in tags_input.split(",") if t.strip()] if tags_input else None ) self.entry_manager.modify_entry( index, new_username, new_url, archived=new_blacklisted, notes=new_notes, label=new_label, custom_fields=custom_fields, tags=tags, ) # Mark database as dirty for background sync self.is_dirty = True self.last_update = time.time() print(colored(f"Entry updated successfully for index {index}.", "green")) # Push the updated index to Nostr so changes are backed up. try: self.start_background_vault_sync() logging.info( "Encrypted index posted to Nostr after entry modification." ) except Exception as nostr_error: logging.error( f"Failed to post updated index to Nostr: {nostr_error}", exc_info=True, ) pause() except Exception as e: logging.error(f"Error during modifying entry: {e}", exc_info=True) print(colored(f"Error: Failed to modify entry: {e}", "red")) @pause_logging_for_ui def handle_search_entries(self) -> None: """Prompt for a query, list matches and optionally show details.""" try: fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Search Entries", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) query = input("Enter search string: ").strip() if not query: self.notify("No search string provided.", level="WARNING") pause() return results = self.entry_manager.search_entries(query) if not results: self.notify("No matching entries found.", level="WARNING") pause() return while True: fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Search Entries", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) print(colored("\n[+] Search Results:\n", "green")) for idx, label, username, _url, _b, etype in results: display_label = label if username: display_label += f" ({username})" type_name = etype.value.replace("_", " ").title() print(colored(f"{idx}. {type_name} - {display_label}", "cyan")) idx_input = input( "Enter index to view details or press Enter to go back: " ).strip() if not idx_input: break if not idx_input.isdigit() or int(idx_input) not in [ r[0] for r in results ]: print(colored("Invalid index.", "red")) pause() continue self.show_entry_details_by_index(int(idx_input)) except Exception as e: logging.error(f"Failed to search entries: {e}", exc_info=True) print(colored(f"Error: Failed to search entries: {e}", "red")) pause() def display_entry_details(self, index: int) -> None: """Print detailed information for a single entry.""" entry = self.entry_manager.retrieve_entry(index) if not entry: return etype = self._entry_type_str(entry) print(color_text(f"Index: {index}", "index")) if etype == EntryType.TOTP.value: print(color_text(f" Label: {entry.get('label', '')}", "index")) print( color_text(f" Derivation Index: {entry.get('index', index)}", "index") ) print( color_text( f" Period: {entry.get('period', 30)}s Digits: {entry.get('digits', 6)}", "index", ) ) notes = entry.get("notes", "") if notes: print(color_text(f" Notes: {notes}", "index")) tags = entry.get("tags", []) if tags: print(color_text(f" Tags: {', '.join(tags)}", "index")) elif etype == EntryType.SEED.value: print(color_text(" Type: Seed Phrase", "index")) print(color_text(f" Label: {entry.get('label', '')}", "index")) words = entry.get("word_count", entry.get("words", 24)) print(color_text(f" Words: {words}", "index")) print( color_text( f" Derivation Index: {entry.get('index', index)}", "index", ) ) notes = entry.get("notes", "") if notes: print(color_text(f" Notes: {notes}", "index")) tags = entry.get("tags", []) if tags: print(color_text(f" Tags: {', '.join(tags)}", "index")) elif etype == EntryType.SSH.value: print(color_text(" Type: SSH Key", "index")) print(color_text(f" Label: {entry.get('label', '')}", "index")) print( color_text(f" Derivation Index: {entry.get('index', index)}", "index") ) pub_label = entry.get("public_key_label", "") if pub_label: print(color_text(f" Public Key Label: {pub_label}", "index")) ssh_fingerprint = entry.get("fingerprint", "") if ssh_fingerprint: print(color_text(f" Fingerprint: {ssh_fingerprint}", "index")) notes = entry.get("notes", "") if notes: print(color_text(f" Notes: {notes}", "index")) tags = entry.get("tags", []) if tags: print(color_text(f" Tags: {', '.join(tags)}", "index")) elif etype == EntryType.PGP.value: print(color_text(" Type: PGP Key", "index")) print(color_text(f" Label: {entry.get('label', '')}", "index")) print( color_text(f" Key Type: {entry.get('key_type', 'ed25519')}", "index") ) uid = entry.get("user_id", "") if uid: print(color_text(f" User ID: {uid}", "index")) print( color_text(f" Derivation Index: {entry.get('index', index)}", "index") ) try: _priv, _pub, pgp_fp = self.entry_manager.get_pgp_key(index, self.parent_seed) if pgp_fp: print(color_text(f" Fingerprint: {pgp_fp}", "index")) except Exception as pgp_err: # pragma: no cover - best effort logging logging.error( f"Failed to derive PGP fingerprint: {pgp_err}", exc_info=True ) notes = entry.get("notes", "") if notes: print(color_text(f" Notes: {notes}", "index")) tags = entry.get("tags", []) if tags: print(color_text(f" Tags: {', '.join(tags)}", "index")) elif etype == EntryType.NOSTR.value: print(color_text(" Type: Nostr Key", "index")) print(color_text(f" Label: {entry.get('label', '')}", "index")) print( color_text(f" Derivation Index: {entry.get('index', index)}", "index") ) notes = entry.get("notes", "") if notes: print(color_text(f" Notes: {notes}", "index")) tags = entry.get("tags", []) if tags: print(color_text(f" Tags: {', '.join(tags)}", "index")) elif etype == EntryType.KEY_VALUE.value: print(color_text(" Type: Key/Value", "index")) print(color_text(f" Label: {entry.get('label', '')}", "index")) print(color_text(f" Key: {entry.get('key', '')}", "index")) print(color_text(f" Value: {entry.get('value', '')}", "index")) notes = entry.get("notes", "") if notes: print(color_text(f" Notes: {notes}", "index")) tags = entry.get("tags", []) if tags: print(color_text(f" Tags: {', '.join(tags)}", "index")) blacklisted = entry.get("archived", entry.get("blacklisted", False)) print(color_text(f" Archived: {'Yes' if blacklisted else 'No'}", "index")) elif etype == EntryType.DOCUMENT.value: print(color_text(" Type: Document", "index")) print(color_text(f" Title: {entry.get('label', '')}", "index")) print( color_text( f" File Type: {entry.get('file_type', 'txt')}", "index", ) ) content = str(entry.get("content", "")) preview = content if len(content) <= 240 else f"{content[:240]}..." print(color_text(" Content Preview:", "index")) print(color_text(f"{preview}", "index")) notes = entry.get("notes", "") if notes: print(color_text(f" Notes: {notes}", "index")) tags = entry.get("tags", []) if tags: print(color_text(f" Tags: {', '.join(tags)}", "index")) if entry.get("date_added"): print(color_text(f" Date Added: {entry.get('date_added')}", "index")) if entry.get("date_modified"): print( color_text( f" Date Modified: {entry.get('date_modified')}", "index", ) ) blacklisted = entry.get("archived", entry.get("blacklisted", False)) print(color_text(f" Archived: {'Yes' if blacklisted else 'No'}", "index")) elif etype == EntryType.MANAGED_ACCOUNT.value: print(color_text(" Type: Managed Account", "index")) print(color_text(f" Label: {entry.get('label', '')}", "index")) words = entry.get("word_count", entry.get("words", 24)) print(color_text(f" Words: {words}", "index")) print( color_text(f" Derivation Index: {entry.get('index', index)}", "index") ) fingerprint = entry.get("fingerprint", "") if fingerprint: print(color_text(f" Fingerprint: {fingerprint}", "index")) notes = entry.get("notes", "") if notes: print(color_text(f" Notes: {notes}", "index")) tags = entry.get("tags", []) if tags: print(color_text(f" Tags: {', '.join(tags)}", "index")) blacklisted = entry.get("archived", entry.get("blacklisted", False)) print(color_text(f" Archived: {'Yes' if blacklisted else 'No'}", "index")) else: website = entry.get("label", entry.get("website", "")) username = entry.get("username", "") url = entry.get("url", "") blacklisted = entry.get("archived", entry.get("blacklisted", False)) notes = entry.get("notes", "") tags = entry.get("tags", []) print(color_text(f" Label: {website}", "index")) print(color_text(f" Username: {username or 'N/A'}", "index")) print(color_text(f" URL: {url or 'N/A'}", "index")) if notes: print(color_text(f" Notes: {notes}", "index")) if tags: print(color_text(f" Tags: {', '.join(tags)}", "index")) print( color_text( f" Archived: {'Yes' if blacklisted else 'No'}", "index", ) ) if entry.get("date_added"): print(color_text(f" Date Added: {entry.get('date_added')}", "index")) if entry.get("date_modified"): print(color_text(f" Date Modified: {entry.get('date_modified')}", "index")) print("-" * 40) @pause_logging_for_ui def handle_list_entries(self) -> None: self.menu_handler.handle_list_entries() def delete_entry(self) -> None: """Deletes an entry from the password index.""" try: index_input = input( "Enter the index number of the entry to delete: " ).strip() if not index_input.isdigit(): print(colored("Error: Index must be a number.", "red")) return index_to_delete = int(index_input) if not confirm_action( f"Are you sure you want to delete entry {index_to_delete}? (Y/N): " ): self.notify("Deletion cancelled.", level="WARNING") return self.entry_manager.delete_entry(index_to_delete) # Mark database as dirty for background sync self.is_dirty = True self.last_update = time.time() # Push updated index to Nostr after deletion try: self.start_background_vault_sync() logging.info("Encrypted index posted to Nostr after entry deletion.") except Exception as nostr_error: logging.error( f"Failed to post updated index to Nostr: {nostr_error}", exc_info=True, ) except Exception as e: logging.error(f"Error during entry deletion: {e}", exc_info=True) print(colored(f"Error: Failed to delete entry: {e}", "red")) @pause_logging_for_ui def handle_archive_entry(self) -> None: """Archive an entry without deleting it.""" try: index_input = input( "Enter the index number of the entry to archive: " ).strip() if not index_input.isdigit(): print(colored("Error: Index must be a number.", "red")) return index = int(index_input) self.entry_manager.archive_entry(index) self.is_dirty = True self.last_update = time.time() pause() except Exception as e: logging.error(f"Error archiving entry: {e}", exc_info=True) print(colored(f"Error: Failed to archive entry: {e}", "red")) @pause_logging_for_ui def handle_view_archived_entries(self) -> None: """Display archived entries and optionally view or restore them.""" try: def _load_archived_entries() -> ( list[tuple[int, str, str | None, str | None, bool]] ): entries = self.entry_manager.list_entries( include_archived=True, verbose=False ) return [e for e in entries if e[4]] archived = _load_archived_entries() if not archived: self.notify("No archived entries found.", level="WARNING") pause() return while True: archived = _load_archived_entries() if not archived: self.notify("No archived entries found.", level="WARNING") pause() return fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Archived Entries", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) print(colored("\n[+] Archived Entries:\n", "green")) for idx, label, _username, _url, _ in archived: print(colored(f"{idx}. {label}", "cyan")) idx_input = input( "Enter index to manage or press Enter to go back: " ).strip() if not idx_input: break if not idx_input.isdigit() or int(idx_input) not in [ e[0] for e in archived ]: print(colored("Invalid index.", "red")) continue entry_index = int(idx_input) while True: action = ( input( "Enter 'v' to view details, 'r' to restore, or press Enter to go back: " ) .strip() .lower() ) if action == "v": self.show_entry_details_by_index(entry_index) archived = _load_archived_entries() if entry_index not in [e[0] for e in archived]: break elif action == "r": self.entry_manager.restore_entry(entry_index) self.is_dirty = True self.last_update = time.time() pause() archived = _load_archived_entries() if not archived: print(colored("All entries restored.", "green")) pause() return break elif not action: break else: print(colored("Invalid choice.", "red")) except Exception as e: logging.error(f"Error viewing archived entries: {e}", exc_info=True) print(colored(f"Error: Failed to view archived entries: {e}", "red")) @pause_logging_for_ui def handle_display_totp_codes(self) -> None: self.menu_handler.handle_display_totp_codes() @pause_logging_for_ui def handle_verify_checksum(self) -> None: """ Handles verifying the script's checksum against the stored checksum to ensure integrity. """ try: fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Settings > Verify Script Checksum", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) current_checksum = calculate_checksum(__file__) try: verified = verify_checksum(current_checksum, SCRIPT_CHECKSUM_FILE) except FileNotFoundError: self.notify( "Checksum file missing. Run scripts/update_checksum.py or choose 'Generate Script Checksum' in Settings.", level="WARNING", ) logging.warning("Checksum file missing during verification.") return if verified: print(colored("Checksum verification passed.", "green")) logging.info("Checksum verification passed.") else: print( colored( "Checksum verification failed. The script may have been modified.", "red", ) ) logging.error("Checksum verification failed.") except Exception as e: logging.error(f"Error during checksum verification: {e}", exc_info=True) print(colored(f"Error: Failed to verify checksum: {e}", "red")) @pause_logging_for_ui def handle_update_script_checksum(self) -> None: """Generate a new checksum for the manager script.""" if not confirm_action("Generate new script checksum? (Y/N): "): self.notify("Operation cancelled.", level="WARNING") return try: fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Settings > Generate Script Checksum", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) script_path = Path(__file__).resolve() if update_checksum_file(str(script_path), str(SCRIPT_CHECKSUM_FILE)): print( colored( f"Checksum updated at '{SCRIPT_CHECKSUM_FILE}'.", "green", ) ) else: print(colored("Failed to update checksum.", "red")) except Exception as e: logging.error(f"Error updating checksum: {e}", exc_info=True) print(colored(f"Error: Failed to update checksum: {e}", "red")) def get_encrypted_data(self) -> Optional[bytes]: """ Retrieves the encrypted password index data. :return: The encrypted data as bytes, or None if retrieval fails. """ try: encrypted_data = self.vault.get_encrypted_index() if encrypted_data: logging.debug("Encrypted index data retrieved successfully.") return encrypted_data else: logging.error("Failed to retrieve encrypted index data.") print(colored("Error: Failed to retrieve encrypted index data.", "red")) return None except Exception as e: logging.error(f"Error retrieving encrypted data: {e}", exc_info=True) print(colored(f"Error: Failed to retrieve encrypted data: {e}", "red")) return None def decrypt_and_save_index_from_nostr(self, encrypted_data: bytes) -> None: """ Decrypts the encrypted data retrieved from Nostr and updates the local index. :param encrypted_data: The encrypted data retrieved from Nostr. """ try: self.vault.decrypt_and_save_index_from_nostr(encrypted_data, merge=True) logging.info("Index file updated from Nostr successfully.") except Exception as e: logging.error( f"Failed to decrypt and save data from Nostr: {e}", exc_info=True ) print( colored( f"Error: Failed to decrypt and save data from Nostr: {e}", "red" ) ) # Re-raise the exception to inform the calling function of the failure raise async def sync_vault_async( self, alt_summary: str | None = None ) -> dict[str, list[str] | str] | None: """Publish the current vault contents to Nostr and return event IDs.""" try: client = getattr(self, "nostr_client", None) if getattr(self, "offline_mode", False): if client is not None and hasattr(client, "last_error"): client.last_error = ( "Offline mode is enabled. Disable it in Settings to sync." ) return None encrypted = self.get_encrypted_data() if not encrypted: if client is not None and hasattr(client, "last_error"): client.last_error = "No encrypted index data available to sync." return None if client is not None and hasattr(client, "last_error"): client.last_error = None pub_snap = getattr(self.nostr_client, "publish_snapshot", None) manifest = None event_id = None if callable(pub_snap): if asyncio.iscoroutinefunction(pub_snap): manifest, event_id = await pub_snap(encrypted) else: manifest, event_id = pub_snap(encrypted) else: # Fallback for tests using simplified stubs event_id = await self.nostr_client.publish_json_to_nostr(encrypted) self.is_dirty = False if event_id is None: if ( client is not None and hasattr(client, "last_error") and not client.last_error ): client.last_error = ( "Failed to publish encrypted index to configured relays." ) return None chunk_ids: list[str] = [] if manifest is not None: chunk_ids = [c.event_id for c in manifest.chunks if c.event_id] delta_ids = self.nostr_client.get_delta_events() if manifest is not None and self.state_manager is not None: ts = manifest.delta_since or int(time.time()) self.state_manager.update_state( manifest_id=event_id, delta_since=ts, last_sync_ts=ts, ) self.last_sync_ts = ts return { "manifest_id": event_id, "chunk_ids": chunk_ids, "delta_ids": list(delta_ids), } except Exception as e: client = getattr(self, "nostr_client", None) if client is not None and hasattr(client, "last_error"): try: client.last_error = str(e) except Exception: pass logging.error(f"Failed to sync vault: {e}", exc_info=True) return None def sync_vault( self, alt_summary: str | None = None ) -> dict[str, list[str] | str] | None: return asyncio.run(self.sync_vault_async(alt_summary=alt_summary)) def backup_database(self) -> None: """ Creates a backup of the encrypted JSON index file. """ try: self.backup_manager.create_backup() print(colored("Backup created successfully.", "green")) except Exception as e: logging.error(f"Failed to create backup: {e}", exc_info=True) print(colored(f"Error: Failed to create backup: {e}", "red")) def restore_database(self) -> None: """ Restores the encrypted JSON index file from the latest backup. """ try: self.backup_manager.restore_latest_backup() print( colored( "Database restored from the latest backup successfully.", "green" ) ) except Exception as e: logging.error(f"Failed to restore backup: {e}", exc_info=True) print(colored(f"Error: Failed to restore backup: {e}", "red")) @pause_logging_for_ui def handle_export_database( self, dest: Path | None = None, *, encrypt: bool | None = None, ) -> Path | None: """Export the current database to a portable file. If ``encrypt`` is ``True`` (default) the payload is encrypted. When ``encrypt`` is ``False`` the export contains plaintext data. When ``encrypt`` is ``None`` the user is prompted interactively. """ try: fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Settings > Export database", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) if encrypt is False: logger.warning("Exporting database in plaintext mode") self.notify( "Warning: plaintext export selected; protect and delete the file quickly.", level="WARNING", ) if encrypt is None: if confirm_action("Export database without encryption? (Y/N): "): print( colored( "Warning: plaintext export contains all secrets in clear text.", "red", ) ) if not confirm_action( "Confirm plaintext export (high risk). Continue? (Y/N): " ): print(colored("Export cancelled.", "yellow")) return None encrypt = False else: encrypt = True path = export_backup( self.vault, self.backup_manager, dest, parent_seed=self.parent_seed, encrypt=encrypt, ) print(colored(f"Database exported to '{path}'.", "green")) audit_logger = getattr(self, "audit_logger", None) if audit_logger and path is not None: audit_logger.log( "backup_export", { "path": str(path), "encryption_mode": "encrypted" if encrypt else "none", }, ) return path except Exception as e: logging.error(f"Failed to export database: {e}", exc_info=True) print(colored(f"Error: Failed to export database: {e}", "red")) return None @pause_logging_for_ui def handle_import_database(self, src: Path) -> None: """Import a portable database file, replacing the current index.""" if not (src.name.endswith(".json.enc") or src.name.endswith(".json")): print( colored( "Error: Selected file must be a SeedPass database backup (.json or .json.enc).", "red", ) ) return # Determine encryption mode for post-processing mode = None try: raw = src.read_bytes() if src.suffix.endswith(".enc"): raw = self.vault.encryption_manager.decrypt_data(raw, context=str(src)) wrapper = json.loads(raw.decode("utf-8")) mode = wrapper.get("encryption_mode") except Exception: mode = None fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Settings > Import database", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) try: import_backup( self.vault, self.backup_manager, src, parent_seed=self.parent_seed, ) except DecryptionError: logging.error("Invalid backup token during import", exc_info=True) print( colored( "Error: Invalid backup. Please import a file created by SeedPass.", "red", ) ) return except FileNotFoundError: logging.error(f"Backup file not found: {src}", exc_info=True) print(colored(f"Error: File '{src}' not found.", "red")) return except Exception as e: logging.error(f"Failed to import database: {e}", exc_info=True) print( colored( f"Error: Failed to import database: {e}. Please verify the backup file.", "red", ) ) return if mode == PortableMode.NONE.value: try: password = prompt_new_password() seed_key = self._derive_seed_key(password, self.current_fingerprint) seed_mgr = EncryptionManager(seed_key, self.fingerprint_dir) self._encrypt_parent_seed_compat( seed_mgr, self.parent_seed, self.current_fingerprint ) self.store_hashed_password(password) except Exception as e: logging.error( f"Failed to set master password after import: {e}", exc_info=True ) print(colored(f"Error: Failed to set master password: {e}", "red")) return print(colored("Database imported successfully.", "green")) self.sync_vault() @pause_logging_for_ui def handle_export_totp_codes(self) -> Path | None: """Export all 2FA codes to a JSON file for other authenticator apps.""" try: self.ensure_key_hierarchy() fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Settings > Export 2FA codes", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) data = self.entry_manager.vault.load_index() entries = data.get("entries", {}) totp_entries = [] for entry in entries.values(): if self._entry_type_str(entry) == EntryType.TOTP.value: label = entry.get("label", "") period = int(entry.get("period", 30)) digits = int(entry.get("digits", 6)) if "secret" in entry: secret = entry["secret"] else: idx = int(entry.get("index", 0)) key = self.KEY_TOTP_DET or getattr(self, "parent_seed", None) secret = TotpManager.derive_secret(key, idx) uri = TotpManager.make_otpauth_uri(label, secret, period, digits) totp_entries.append( { "label": label, "secret": secret, "period": period, "digits": digits, "uri": uri, } ) if not totp_entries: self.notify("No 2FA codes to export.", level="WARNING") return None dest_str = input( "Enter destination file path (default: totp_export.json): " ).strip() dest = Path(dest_str) if dest_str else Path("totp_export.json") json_data = json.dumps({"entries": totp_entries}, indent=2) if confirm_action("Encrypt export with a password? (Y/N): "): password = prompt_new_password() iterations = self.config_manager.get_kdf_iterations() key = derive_key_from_password( password, self.current_fingerprint, iterations=iterations ) enc_mgr = EncryptionManager(key, dest.parent) data_bytes = enc_mgr.encrypt_data(json_data.encode("utf-8")) dest = dest.with_suffix(dest.suffix + ".enc") dest.write_bytes(data_bytes) else: dest.write_text(json_data) os.chmod(dest, 0o600) print(colored(f"2FA codes exported to '{dest}'.", "green")) return dest except Exception as e: logging.error(f"Failed to export TOTP codes: {e}", exc_info=True) print(colored(f"Error: Failed to export 2FA codes: {e}", "red")) return None @pause_logging_for_ui def handle_backup_reveal_parent_seed( self, file: Path | None = None, *, password: Optional[str] = None ) -> None: """Reveal the parent seed and optionally save an encrypted backup. Parameters ---------- file: Optional path where an encrypted backup should be written. When provided, the confirmation and filename prompts are skipped and the seed is saved directly to this location. """ try: fp, parent_fp, child_fp = self.header_fingerprint_args clear_header_with_notification( self, fp, "Main Menu > Settings > Backup Parent Seed", parent_fingerprint=parent_fp, child_fingerprint=child_fp, ) print(colored("\n=== Backup Parent Seed ===", "yellow")) self.notify( "Warning: Revealing your parent seed is a highly sensitive operation.", level="WARNING", ) self.notify( "Ensure you're in a secure, private environment and no one is watching your screen.", level="WARNING", ) # Verify user's identity with secure password verification if password is None: password = prompt_existing_password( "Enter your master password to continue: " ) if not self.verify_password(password): print(colored("Incorrect password. Operation aborted.", "red")) return # Double confirmation if not confirm_action( "Are you absolutely sure you want to reveal your parent seed? (Y/N): " ): self.notify("Operation cancelled by user.", level="WARNING") return # Reveal the parent seed print(colored("\n=== Your BIP-85 Parent Seed ===", "green")) print(color_text(self.parent_seed, "imported")) print( colored( "\nPlease write this down and store it securely. Do not share it with anyone.", "red", ) ) backup_path: Path | None = None if file is not None: backup_path = file save = True else: save = confirm_action( "Do you want to save this to an encrypted backup file? (Y/N): " ) if save: filename = input( f"Enter filename to save (default: {DEFAULT_SEED_BACKUP_FILENAME}): " ).strip() filename = filename if filename else DEFAULT_SEED_BACKUP_FILENAME backup_path = self.fingerprint_dir / filename if save and backup_path is not None: if not self.is_valid_filename(backup_path.name): print(colored("Invalid filename. Operation aborted.", "red")) return self.encryption_manager.encrypt_and_save_file( self.parent_seed.encode("utf-8"), backup_path ) print( colored( f"Encrypted seed backup saved to '{backup_path}'. Ensure this file is stored securely.", "green", ) ) audit_logger = getattr(self, "audit_logger", None) if audit_logger: details = { "backup_path": str(backup_path) if save and backup_path else None } audit_logger.log("seed_reveal", details) except Exception as e: logging.error(f"Error during parent seed backup/reveal: {e}", exc_info=True) print(colored(f"Error: Failed to backup/reveal parent seed: {e}", "red")) def verify_password(self, password: str) -> bool: """ Verifies the provided password against the stored hashed password. Parameters: password (str): The password to verify. Returns: bool: True if the password is correct, False otherwise. """ try: config = self.config_manager.load_config(require_pin=False) stored_hash = config.get("password_hash", "").encode() if not stored_hash: # Fallback to legacy file if hash not present in config legacy_file = self.fingerprint_dir / "hashed_password.enc" if legacy_file.exists(): with open(legacy_file, "rb") as f: stored_hash = f.read() self.config_manager.set_password_hash(stored_hash.decode()) else: logging.error("Hashed password not found.") print(colored("Error: Hashed password not found.", "red")) return False is_correct = bcrypt.checkpw(password.encode("utf-8"), stored_hash) if is_correct: logging.debug("Password verification successful.") else: logging.warning("Password verification failed.") return is_correct except DecryptionError as e: logging.error(f"Failed to decrypt config: {e}") print( colored( "Error: Could not decrypt configuration. The password may be incorrect or the file may be corrupted.", "red", ) ) return False except Exception as e: logging.error(f"Error verifying password: {e}", exc_info=True) print(colored(f"Error: Failed to verify password: {e}", "red")) return False def is_valid_filename(self, filename: str) -> bool: """ Validates the provided filename to prevent directory traversal and invalid characters. Parameters: filename (str): The filename to validate. Returns: bool: True if valid, False otherwise. """ # Basic validation: filename should not contain path separators or be empty invalid_chars = ["/", "\\", ".."] if any(char in filename for char in invalid_chars) or not filename: logging.warning(f"Invalid filename attempted: {filename}") return False return True def store_hashed_password(self, password: str) -> None: """ Hashes and stores the user's password securely using bcrypt. This should be called during the initial setup. """ try: hashed = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode() if self.config_manager: self.config_manager.set_password_hash(hashed) else: # Fallback to legacy file method if config_manager unavailable hashed_password_file = self.fingerprint_dir / "hashed_password.enc" atomic_write( hashed_password_file, lambda f: f.write(hashed.encode()), mode="wb", ) os.chmod(hashed_password_file, 0o600) logging.info("User password hashed and stored successfully.") except AttributeError: # If bcrypt.hashpw is not available, try using bcrypt directly salt = bcrypt.gensalt() hashed = bcrypt.hashpw(password.encode("utf-8"), salt).decode() if self.config_manager: self.config_manager.set_password_hash(hashed) else: hashed_password_file = self.fingerprint_dir / "hashed_password.enc" atomic_write( hashed_password_file, lambda f: f.write(hashed.encode()), mode="wb", ) os.chmod(hashed_password_file, 0o600) logging.info( "User password hashed and stored successfully (using alternative method)." ) except Exception as e: logging.error(f"Failed to store hashed password: {e}", exc_info=True) print(colored(f"Error: Failed to store hashed password: {e}", "red")) raise def change_password(self, old_password: str, new_password: str) -> None: """Change the master password used for encryption.""" try: self.ensure_key_hierarchy() if not self.verify_password(old_password): raise ValueError("Incorrect password") # Load data with existing encryption manager index_data = self.vault.load_index() config_data = self.config_manager.load_config(require_pin=False) # Create a new encryption manager with the new password new_key = base64.urlsafe_b64encode(self.KEY_STORAGE) seed_key = self._derive_seed_key(new_password, self.current_fingerprint) seed_mgr = EncryptionManager(seed_key, self.fingerprint_dir) new_enc_mgr = EncryptionManager(new_key, self.fingerprint_dir) self._encrypt_parent_seed_compat( seed_mgr, self.parent_seed, self.current_fingerprint ) self.vault.set_encryption_manager(new_enc_mgr) self.vault.save_index(index_data) self.config_manager.vault = self.vault self.config_manager.save_config(config_data) # Update hashed password and replace managers self.encryption_manager = new_enc_mgr self.password_generator.encryption_manager = new_enc_mgr self.store_hashed_password(new_password) if getattr(self, "state_manager", None) is not None: state = self.state_manager.state relay_list = state.get("relays", list(DEFAULT_RELAYS)) else: relay_list = list(DEFAULT_RELAYS) self.nostr_client = NostrClient( encryption_manager=self.encryption_manager, fingerprint=self.current_fingerprint, relays=relay_list, config_manager=self.config_manager, parent_seed=getattr(self, "parent_seed", None), key_index=self.KEY_INDEX, account_index=self.nostr_account_idx, ) if getattr(self, "manifest_id", None) and hasattr( self.nostr_client, "_state_lock" ): from nostr.backup_models import Manifest with self.nostr_client._state_lock: self.nostr_client.current_manifest_id = self.manifest_id self.nostr_client.current_manifest = Manifest( ver=1, algo="gzip", chunks=[], delta_since=self.delta_since or None, ) # Push a fresh backup to Nostr so the newly encrypted index is # stored remotely. Include a tag to mark the password change. try: summary = f"password-change-{int(time.time())}" self.sync_vault(alt_summary=summary) except Exception as nostr_error: logging.error( f"Failed to post updated index to Nostr after password change: {nostr_error}" ) except Exception as e: logging.error(f"Failed to change password: {e}", exc_info=True) raise def get_profile_stats(self) -> dict: """Return various statistics about the current seed profile.""" if not all([self.entry_manager, self.config_manager, self.backup_manager]): return {} stats: dict[str, object] = {} # Entry counts by type data = self.entry_manager.vault.load_index() entries = data.get("entries", {}) counts: dict[str, int] = {etype.value: 0 for etype in EntryType} for entry in entries.values(): etype = self._entry_type_str(entry) counts[etype] = counts.get(etype, 0) + 1 stats["entries"] = counts stats["total_entries"] = len(entries) # Schema version and database checksum status stats["schema_version"] = data.get("schema_version") json_content = json.dumps(data, indent=4) current_checksum = hashlib.sha256(json_content.encode("utf-8")).hexdigest() chk_path = self.entry_manager.checksum_file if chk_path.exists(): stored = chk_path.read_text().strip() stats["checksum_ok"] = stored == current_checksum else: stored = None stats["checksum_ok"] = False stats["checksum"] = stored # Script checksum status script_path = Path(__file__).resolve() try: script_checksum = calculate_checksum(str(script_path)) except Exception: script_checksum = None if SCRIPT_CHECKSUM_FILE.exists() and script_checksum: stored_script = SCRIPT_CHECKSUM_FILE.read_text().strip() stats["script_checksum_ok"] = stored_script == script_checksum else: stats["script_checksum_ok"] = False # Relay info cfg = self.config_manager.load_config(require_pin=False) relays = cfg.get("relays", []) stats["relays"] = relays stats["relay_count"] = len(relays) # Backup info backups = list( self.backup_manager.backup_dir.glob("entries_db_backup_*.json.enc") ) stats["backup_count"] = len(backups) stats["backup_dir"] = str(self.backup_manager.backup_dir) stats["additional_backup_path"] = ( self.config_manager.get_additional_backup_path() ) # Nostr sync info manifest = self.nostr_client.get_current_manifest() if manifest is not None: stats["chunk_count"] = len(manifest.chunks) stats["delta_since"] = manifest.delta_since stats["pending_deltas"] = len(self.nostr_client.get_delta_events()) else: stats["chunk_count"] = 0 stats["delta_since"] = None stats["pending_deltas"] = 0 return stats def display_stats(self) -> None: """Print a summary of :meth:`get_profile_stats` to the console.""" stats = self.get_profile_stats() if not stats: print(colored("No statistics available.", "red")) return print(color_text("\n=== Seed Profile Stats ===", "stats")) print(color_text(f"Total entries: {stats['total_entries']}", "stats")) for etype, count in stats["entries"].items(): print(color_text(f" {etype}: {count}", "stats")) print(color_text(f"Relays configured: {stats['relay_count']}", "stats")) print( color_text( f"Backups: {stats['backup_count']} (dir: {stats['backup_dir']})", "stats", ) ) if stats.get("additional_backup_path"): print( color_text( f"Additional backup: {stats['additional_backup_path']}", "stats" ) ) print(color_text(f"Schema version: {stats['schema_version']}", "stats")) print( color_text( f"Database checksum ok: {'yes' if stats['checksum_ok'] else 'no'}", "stats", ) ) print( color_text( f"Script checksum ok: {'yes' if stats['script_checksum_ok'] else 'no'}", "stats", ) ) print(color_text(f"Snapshot chunks: {stats['chunk_count']}", "stats")) print(color_text(f"Pending deltas: {stats['pending_deltas']}", "stats")) if stats.get("delta_since"): print( color_text(f"Latest delta timestamp: {stats['delta_since']}", "stats") )