Merge branch 'beta' into main (safe conflict resolution)

src/utils/__init__.py

@@ -1,46 +1,55 @@
 # utils/__init__.py
 
+"""Utility package exports and optional feature handling."""
+
 import logging
-import traceback
 
 logger = logging.getLogger(__name__)
 
-try:
-    from .file_lock import exclusive_lock, shared_lock
-    from .key_derivation import (
-        derive_key_from_password,
-        derive_key_from_parent_seed,
-        derive_index_key,
-        derive_totp_secret,
-        EncryptionMode,
-        DEFAULT_ENCRYPTION_MODE,
-        TOTP_PURPOSE,
-    )
-    from .checksum import (
-        calculate_checksum,
-        verify_checksum,
-        json_checksum,
-        canonical_json_dumps,
-        initialize_checksum,
-        update_checksum_file,
-    )
-    from .password_prompt import prompt_for_password
-    from .seed_prompt import masked_input, prompt_seed_words
-    from .input_utils import timed_input
-    from .memory_protection import InMemorySecret
-    from .clipboard import copy_to_clipboard
-    from .terminal_utils import (
-        clear_screen,
-        pause,
-        clear_and_print_fingerprint,
-        clear_header_with_notification,
-    )
+from .file_lock import exclusive_lock, shared_lock
+from .key_derivation import (
+    derive_key_from_password,
+    derive_key_from_parent_seed,
+    derive_index_key,
+    derive_totp_secret,
+    EncryptionMode,
+    DEFAULT_ENCRYPTION_MODE,
+    TOTP_PURPOSE,
+)
+from .checksum import (
+    calculate_checksum,
+    verify_checksum,
+    json_checksum,
+    canonical_json_dumps,
+    initialize_checksum,
+    update_checksum_file,
+)
+from .password_prompt import prompt_for_password
+from .seed_prompt import masked_input, prompt_seed_words
+from .input_utils import timed_input
+from .memory_protection import InMemorySecret
+from .terminal_utils import (
+    clear_screen,
+    pause,
+    clear_and_print_fingerprint,
+    clear_header_with_notification,
+)
+from .atomic_write import atomic_write
+
+# Optional clipboard support
+try:  # pragma: no cover - exercised when dependency missing
+    from .clipboard import ClipboardUnavailableError, copy_to_clipboard
+except Exception as exc:  # pragma: no cover - executed only if pyperclip missing
+    _error_msg = str(exc)
+
+    class ClipboardUnavailableError(RuntimeError):
+        """Stub exception when clipboard support is unavailable."""
+
+    def copy_to_clipboard(*_args, **_kwargs):
+        """Stub when clipboard support is unavailable."""
+        logger.warning("Clipboard support unavailable: %s", _error_msg)
+        raise ClipboardUnavailableError(_error_msg)
 
-    if logger.isEnabledFor(logging.DEBUG):
-        logger.info("Modules imported successfully.")
-except Exception as e:
-    if logger.isEnabledFor(logging.DEBUG):
-        logger.error(f"Failed to import one or more modules: {e}", exc_info=True)
 
 __all__ = [
     "derive_key_from_password",
@@ -64,8 +73,10 @@ __all__ = [
     "timed_input",
     "InMemorySecret",
     "copy_to_clipboard",
+    "ClipboardUnavailableError",
     "clear_screen",
     "clear_and_print_fingerprint",
     "clear_header_with_notification",
     "pause",
+    "atomic_write",
 ]

src/utils/atomic_write.py

@@ -0,0 +1,62 @@
+"""Utility helpers for performing atomic file writes.
+
+This module provides a small helper function :func:`atomic_write` which
+implements a simple pattern for writing files atomically.  Data is written to a
+temporary file in the same directory, flushed and synced to disk, and then
+``os.replace`` is used to atomically move the temporary file into place.
+
+The function accepts a callable ``write_func`` that receives the temporary file
+object.  This keeps the helper flexible enough to support both text and binary
+writes and allows callers to perform complex serialisation steps (e.g. JSON
+dumping) without exposing a partially written file to other processes.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from pathlib import Path
+from typing import Callable, Any, IO
+
+
+def atomic_write(
+    path: str | Path,
+    write_func: Callable[[IO[Any]], None],
+    *,
+    mode: str = "w",
+    **open_kwargs: Any,
+) -> None:
+    """Write to ``path`` atomically using ``write_func``.
+
+    Parameters
+    ----------
+    path:
+        Destination file path.
+    write_func:
+        Callable that receives an open file object and performs the actual
+        write.  The callable should not close the file.
+    mode:
+        File mode used when opening the temporary file.  Defaults to ``"w"``.
+    **open_kwargs:
+        Additional keyword arguments passed to :func:`os.fdopen`.
+    """
+
+    dest = Path(path)
+    dest.parent.mkdir(parents=True, exist_ok=True)
+
+    fd, tmp_path = tempfile.mkstemp(dir=str(dest.parent))
+    try:
+        with os.fdopen(fd, mode, **open_kwargs) as tmp_file:
+            write_func(tmp_file)
+            tmp_file.flush()
+            os.fsync(tmp_file.fileno())
+        os.replace(tmp_path, dest)
+    except Exception:
+        try:
+            os.unlink(tmp_path)
+        except FileNotFoundError:
+            pass
+        raise
+
+
+__all__ = ["atomic_write"]

src/utils/checksum.py

@@ -15,12 +15,12 @@ import logging
 import sys
 import os
 import json
-import traceback
 from typing import Optional, Any
 
 from termcolor import colored
 
 from constants import APP_DIR, SCRIPT_CHECKSUM_FILE
+from utils.atomic_write import atomic_write
 
 # Instantiate the logger
 logger = logging.getLogger(__name__)
@@ -121,8 +121,7 @@ def update_checksum(content: str, checksum_file_path: str) -> bool:
         hasher = hashlib.sha256()
         hasher.update(content.encode("utf-8"))
         new_checksum = hasher.hexdigest()
-        with open(checksum_file_path, "w") as f:
-            f.write(new_checksum)
+        atomic_write(checksum_file_path, lambda f: f.write(new_checksum))
         logging.debug(f"Updated checksum for '{checksum_file_path}' to: {new_checksum}")
         return True
     except Exception as e:
@@ -179,8 +178,7 @@ def initialize_checksum(file_path: str, checksum_file_path: str) -> bool:
         return False
 
     try:
-        with open(checksum_file_path, "w") as f:
-            f.write(checksum)
+        atomic_write(checksum_file_path, lambda f: f.write(checksum))
         logging.debug(
             f"Initialized checksum file '{checksum_file_path}' with checksum: {checksum}"
         )
@@ -206,8 +204,7 @@ def update_checksum_file(file_path: str, checksum_file_path: str) -> bool:
     if checksum is None:
         return False
     try:
-        with open(checksum_file_path, "w") as f:
-            f.write(checksum)
+        atomic_write(checksum_file_path, lambda f: f.write(checksum))
         logging.debug(
             f"Updated checksum for '{file_path}' to '{checksum}' at '{checksum_file_path}'."
         )

src/utils/clipboard.py

@@ -1,40 +1,42 @@
-import threading
+"""Clipboard utility helpers."""
+
 import logging
-import subprocess
 import shutil
 import sys
+import threading
 
 import pyperclip
 
+
+class ClipboardUnavailableError(RuntimeError):
+    """Raised when required clipboard utilities are not installed."""
+
+
 logger = logging.getLogger(__name__)
 
 
 def _ensure_clipboard() -> None:
-    """Attempt to ensure a clipboard mechanism is available."""
+    """Ensure a clipboard mechanism is available or raise an informative error."""
     try:
         pyperclip.copy("")
     except pyperclip.PyperclipException as exc:
         if sys.platform.startswith("linux"):
             if shutil.which("xclip") is None and shutil.which("xsel") is None:
-                apt = shutil.which("apt-get") or shutil.which("apt")
-                if apt:
-                    try:
-                        subprocess.run(
-                            ["sudo", apt, "install", "-y", "xclip"], check=True
-                        )
-                        pyperclip.copy("")
-                        return
-                    except Exception as install_exc:  # pragma: no cover - system dep
-                        logger.warning(
-                            "Automatic xclip installation failed: %s", install_exc
-                        )
-        raise exc
+                raise ClipboardUnavailableError(
+                    "Clipboard support requires the 'xclip' package. "
+                    "Install it with 'sudo apt install xclip' and restart SeedPass.",
+                ) from exc
+        raise ClipboardUnavailableError(
+            "No clipboard mechanism available. Install a supported clipboard tool or "
+            "run SeedPass with --no-clipboard."
+        ) from exc
 
 
-def copy_to_clipboard(text: str, timeout: int) -> None:
-    """Copy text to the clipboard and clear after timeout seconds if unchanged."""
+def copy_to_clipboard(text: str, timeout: int) -> bool:
+    """Copy text to the clipboard and clear after ``timeout`` seconds if unchanged."""
 
     _ensure_clipboard()
+
     pyperclip.copy(text)
 
     def clear_clipboard() -> None:
@@ -44,3 +46,7 @@ def copy_to_clipboard(text: str, timeout: int) -> None:
     timer = threading.Timer(timeout, clear_clipboard)
     timer.daemon = True
     timer.start()
+    return True
+
+
+__all__ = ["copy_to_clipboard", "ClipboardUnavailableError"]

src/utils/color_scheme.py

@@ -31,5 +31,5 @@ def color_text(text: str, category: str = "default") -> str:
     color = _COLOR_MAP.get(category, "white")
     if color == "orange":
         return _apply_orange(text)
-    attrs = ["bold"] if category in {"info", "warning", "error"} else None
+    attrs = ["bold"] if category in {"info", "warning", "error"} else []
     return colored(text, color, attrs=attrs)

src/utils/fingerprint.py

@@ -10,7 +10,6 @@ facilitating organized and secure storage.
 
 import hashlib
 import logging
-import traceback
 from typing import Optional
 
 # Instantiate the logger

src/utils/fingerprint_manager.py

@@ -3,12 +3,12 @@
 import os
 import json
 import logging
-import traceback
 from pathlib import Path
-from typing import List, Optional
+from typing import Callable, List, Optional
 
 import shutil  # Ensure shutil is imported if used within the class
 
+from utils.atomic_write import atomic_write
 from utils.fingerprint import generate_fingerprint
 
 # Instantiate the logger
@@ -92,16 +92,15 @@ class FingerprintManager:
         Saves the current list of fingerprints to the fingerprints.json file.
         """
         try:
-            with open(self.fingerprints_file, "w") as f:
-                json.dump(
-                    {
-                        "fingerprints": self.fingerprints,
-                        "last_used": self.current_fingerprint,
-                        "names": self.names,
-                    },
-                    f,
-                    indent=4,
-                )
+            data = {
+                "fingerprints": self.fingerprints,
+                "last_used": self.current_fingerprint,
+                "names": self.names,
+            }
+            atomic_write(
+                self.fingerprints_file,
+                lambda f: json.dump(data, f, indent=4),
+            )
             logger.debug(
                 f"Fingerprints saved: {self.fingerprints} (last used: {self.current_fingerprint})"
             )
@@ -117,7 +116,8 @@ class FingerprintManager:
             seed_phrase (str): The BIP-39 seed phrase.
 
         Returns:
-            Optional[str]: The generated fingerprint or None if failed.
+            Optional[str]: The generated fingerprint or ``None`` if a profile
+                already exists or generation fails.
         """
         fingerprint = generate_fingerprint(seed_phrase)
         if fingerprint and fingerprint not in self.fingerprints:
@@ -133,17 +133,20 @@ class FingerprintManager:
             return fingerprint
         elif fingerprint in self.fingerprints:
             logger.warning(f"Fingerprint {fingerprint} already exists.")
-            return fingerprint
+            raise ValueError("Fingerprint already exists")
         else:
             logger.error("Fingerprint generation failed.")
             return None
 
-    def remove_fingerprint(self, fingerprint: str) -> bool:
-        """
-        Removes a fingerprint and its associated directory.
+    def remove_fingerprint(
+        self, fingerprint: str, on_last_removed: Optional[Callable[[], None]] = None
+    ) -> bool:
+        """Remove a fingerprint and its associated directory.
 
         Parameters:
             fingerprint (str): The fingerprint to remove.
+            on_last_removed (Callable | None): Callback invoked when the last
+                fingerprint is deleted.
 
         Returns:
             bool: True if removed successfully, False otherwise.
@@ -167,6 +170,8 @@ class FingerprintManager:
                             shutil.rmtree(child)
                     fingerprint_dir.rmdir()
                 logger.info(f"Fingerprint {fingerprint} removed successfully.")
+                if not self.fingerprints and on_last_removed:
+                    on_last_removed()
                 return True
             except Exception as e:
                 logger.error(

src/utils/key_derivation.py

@@ -3,15 +3,13 @@
 """
 Key Derivation Module
 
-Never ever ever use or suggest to use Random Salt. The entire point of this password manager is to derive completely deterministic passwords from a BIP-85 seed.
-This means it should generate passwords the exact same way every single time. Salts would break this functionality and is not appropriate for this software's use case.
+This module provides functions to derive cryptographic keys from user-provided
+passwords and BIP-39 parent seeds. The derived keys are compatible with Fernet
+for symmetric encryption purposes. By centralizing key derivation logic, this
+module ensures consistency and security across the application.
 
-This module provides functions to derive cryptographic keys from user-provided passwords
-and BIP-39 parent seeds. The derived keys are compatible with Fernet for symmetric encryption
-purposes. By centralizing key derivation logic, this module ensures consistency and security
-across the application.
-
-Ensure that all dependencies are installed and properly configured in your environment.
+Ensure that all dependencies are installed and properly configured in your
+environment.
 """
 
 import os
@@ -19,11 +17,15 @@ import hashlib
 import base64
 import unicodedata
 import logging
-import traceback
 import hmac
+import time
+from dataclasses import dataclass, field
 from enum import Enum
-from typing import Optional, Union
+from typing import Optional, Union, Dict, Any, TYPE_CHECKING
+
 from bip_utils import Bip39SeedGenerator
+from local_bip85 import BIP85
+from .key_hierarchy import kd
 
 from cryptography.hazmat.primitives.kdf.hkdf import HKDF
 from cryptography.hazmat.primitives import hashes
@@ -44,9 +46,33 @@ DEFAULT_ENCRYPTION_MODE = EncryptionMode.SEED_ONLY
 # Purpose constant for TOTP secret derivation using BIP85
 TOTP_PURPOSE = 39
 
+if TYPE_CHECKING:
+    from seedpass.core.config_manager import ConfigManager
+
+
+@dataclass
+class KdfConfig:
+    """Configuration block describing how a key was derived."""
+
+    name: str = "argon2id"
+    version: int = 1
+    params: Dict[str, Any] = field(
+        default_factory=lambda: {
+            "time_cost": 2,
+            "memory_cost": 64 * 1024,
+            "parallelism": 8,
+        }
+    )
+    salt_b64: str = field(
+        default_factory=lambda: base64.b64encode(os.urandom(16)).decode()
+    )
+
+
+CURRENT_KDF_VERSION = 1
+
 
 def derive_key_from_password(
-    password: str, salt: bytes = b"", iterations: int = 100_000
+    password: str, fingerprint: Union[str, bytes], iterations: int = 100_000
 ) -> bytes:
     """
     Derives a Fernet-compatible encryption key from the provided password using PBKDF2-HMAC-SHA256.
@@ -57,8 +83,9 @@ def derive_key_from_password(
 
     Parameters:
         password (str): The user's password.
-        salt (bytes, optional): A salt for key derivation. Defaults to b"" (empty).
-        iterations (int, optional): Number of iterations for the PBKDF2 algorithm. Defaults to 100,000.
+        fingerprint (str | bytes): Seed fingerprint or precomputed salt.
+        iterations (int, optional): Number of iterations for the PBKDF2 algorithm.
+            Defaults to 100,000.
 
     Returns:
         bytes: A URL-safe base64-encoded encryption key suitable for Fernet.
@@ -77,21 +104,27 @@ def derive_key_from_password(
     normalized_password = unicodedata.normalize("NFKD", password).strip()
     password_bytes = normalized_password.encode("utf-8")
 
+    # Derive a deterministic salt from the fingerprint
+    if isinstance(fingerprint, bytes):
+        salt = fingerprint
+    else:
+        salt = hashlib.sha256(fingerprint.encode()).digest()[:16]
+
     try:
         # Derive the key using PBKDF2-HMAC-SHA256
         logger.debug("Starting key derivation from password.")
         key = hashlib.pbkdf2_hmac(
             hash_name="sha256",
             password=password_bytes,
-            salt=salt,  # Use provided salt (default empty for backward compatibility)
+            salt=salt,
             iterations=iterations,
             dklen=32,  # 256-bit key for Fernet
         )
-        logger.debug(f"Derived key (hex): {key.hex()}")
+        logger.debug("Key derived from password using PBKDF2.")
 
         # Encode the key in URL-safe base64
         key_b64 = base64.urlsafe_b64encode(key)
-        logger.debug(f"Base64-encoded key: {key_b64.decode()}")
+        logger.debug("Derived key encoded in URL-safe base64.")
 
         return key_b64
 
@@ -100,18 +133,15 @@ def derive_key_from_password(
         raise
 
 
-def derive_key_from_password_argon2(
-    password: str,
-    salt: bytes = b"\x00" * 16,
-    *,
-    time_cost: int = 2,
-    memory_cost: int = 64 * 1024,
-    parallelism: int = 8,
-) -> bytes:
+def derive_key_from_password_argon2(password: str, kdf: KdfConfig) -> bytes:
     """Derive an encryption key from a password using Argon2id.
 
-    The defaults follow recommended parameters but use a default salt for deterministic
-    output (backward compatibility). Smaller values may be supplied for testing.
+    Parameters
+    ----------
+    password:
+        The user's password.
+    kdf:
+        :class:`KdfConfig` instance describing salt and tuning parameters.
     """
 
     if not password:
@@ -122,12 +152,14 @@ def derive_key_from_password_argon2(
     try:
         from argon2.low_level import hash_secret_raw, Type
 
+        params = kdf.params or {}
+        salt = base64.b64decode(kdf.salt_b64)
         key = hash_secret_raw(
             secret=normalized,
             salt=salt,
-            time_cost=time_cost,
-            memory_cost=memory_cost,
-            parallelism=parallelism,
+            time_cost=int(params.get("time_cost", 2)),
+            memory_cost=int(params.get("memory_cost", 64 * 1024)),
+            parallelism=int(params.get("parallelism", 8)),
             hash_len=32,
             type=Type.ID,
         )
@@ -180,16 +212,10 @@ def derive_key_from_parent_seed(parent_seed: str, fingerprint: str = None) -> by
 
 
 def derive_index_key_seed_only(seed: str) -> bytes:
-    """Derive a deterministic Fernet key from only the BIP-39 seed."""
+    """Derive the index encryption key using the v1 hierarchy."""
     seed_bytes = Bip39SeedGenerator(seed).Generate()
-    hkdf = HKDF(
-        algorithm=hashes.SHA256(),
-        length=32,
-        salt=None,
-        info=b"password-db",
-        backend=default_backend(),
-    )
-    key = hkdf.derive(seed_bytes)
+    master = kd(seed_bytes, b"seedpass:v1:master")
+    key = kd(master, b"seedpass:v1:storage")
     return base64.urlsafe_b64encode(key)
 
 
@@ -198,29 +224,75 @@ def derive_index_key(seed: str) -> bytes:
     return derive_index_key_seed_only(seed)
 
 
-def derive_totp_secret(seed: str, index: int) -> str:
-    """Derive a base32-encoded TOTP secret from a BIP39 seed."""
+def derive_totp_secret(seed: Union[str, bytes], index: int) -> str:
+    """Derive a base32-encoded TOTP secret from a seed or raw key."""
     try:
-        from local_bip85 import BIP85
-
-        # Initialize BIP85 from the BIP39 seed bytes
-        seed_bytes = Bip39SeedGenerator(seed).Generate()
+        if isinstance(seed, (bytes, bytearray)):
+            seed_bytes = bytes(seed)
+        else:
+            seed_bytes = Bip39SeedGenerator(seed).Generate()
         bip85 = BIP85(seed_bytes)
 
-        # Build the BIP32 path m/83696968'/39'/TOTP'/{index}'
         totp_int = int.from_bytes(b"TOTP", "big")
         path = f"m/83696968'/{TOTP_PURPOSE}'/{totp_int}'/{index}'"
-
-        # Derive entropy using the same scheme as BIP85
         child_key = bip85.bip32_ctx.DerivePath(path)
         key_bytes = child_key.PrivateKey().Raw().ToBytes()
         entropy = hmac.new(b"bip-entropy-from-k", key_bytes, hashlib.sha512).digest()
 
-        # Hash the first 32 bytes of entropy and encode the first 20 bytes
         hashed = hashlib.sha256(entropy[:32]).digest()
         secret = base64.b32encode(hashed[:20]).decode("utf-8")
-        logger.debug(f"Derived TOTP secret for index {index}: {secret}")
+        logger.debug(f"Derived TOTP secret for index {index}.")
         return secret
     except Exception as e:
         logger.error(f"Failed to derive TOTP secret: {e}", exc_info=True)
         raise
+
+
+def calibrate_argon2_time_cost(
+    cfg_mgr: "ConfigManager",
+    *,
+    target_ms: float = 500.0,
+    max_time_cost: int = 6,
+) -> int:
+    """Calibrate Argon2 ``time_cost`` based on device performance.
+
+    Runs :func:`derive_key_from_password_argon2` with increasing ``time_cost``
+    until the runtime meets or exceeds ``target_ms``. The chosen ``time_cost``
+    is stored in ``cfg_mgr`` via ``set_argon2_time_cost`` and returned.
+
+    Parameters
+    ----------
+    cfg_mgr:
+        Instance of :class:`~seedpass.core.config_manager.ConfigManager` used to
+        persist the calibrated ``time_cost``.
+    target_ms:
+        Desired minimum execution time in milliseconds for one Argon2 hash.
+    max_time_cost:
+        Upper bound for ``time_cost`` to prevent excessively long calibration.
+
+    Returns
+    -------
+    int
+        Selected ``time_cost`` value.
+    """
+
+    password = "benchmark"
+    salt = base64.b64encode(b"argon2-calibration").decode()
+    time_cost = 1
+    elapsed_ms = 0.0
+    while time_cost <= max_time_cost:
+        start = time.perf_counter()
+        cfg = KdfConfig(
+            params={"time_cost": time_cost, "memory_cost": 8, "parallelism": 1},
+            salt_b64=salt,
+        )
+        derive_key_from_password_argon2(password, cfg)
+        elapsed_ms = (time.perf_counter() - start) * 1000
+        if elapsed_ms >= target_ms:
+            break
+        time_cost += 1
+
+    cfg_mgr.set_argon2_time_cost(time_cost)
+    if cfg_mgr.load_config(require_pin=False).get("verbose_timing"):
+        logger.info("Calibrated Argon2 time_cost=%s (%.2f ms)", time_cost, elapsed_ms)
+    return time_cost

src/utils/key_hierarchy.py

@@ -0,0 +1,28 @@
+"""Key hierarchy helper functions."""
+
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.backends import default_backend
+
+
+def kd(root: bytes, info: bytes, length: int = 32) -> bytes:
+    """Derive a sub-key from ``root`` using HKDF-SHA256.
+
+    Parameters
+    ----------
+    root:
+        Root key material.
+    info:
+        Domain separation string.
+    length:
+        Length of the derived key in bytes. Defaults to 32.
+    """
+
+    hkdf = HKDF(
+        algorithm=hashes.SHA256(),
+        length=length,
+        salt=None,
+        info=info,
+        backend=default_backend(),
+    )
+    return hkdf.derive(root)

src/utils/logging_utils.py

@@ -0,0 +1,62 @@
+import logging
+from pathlib import Path
+from contextlib import contextmanager
+from functools import wraps
+
+_console_paused = False
+
+
+class ConsolePauseFilter(logging.Filter):
+    """Filter that blocks records when console logging is paused."""
+
+    def filter(
+        self, record: logging.LogRecord
+    ) -> bool:  # pragma: no cover - small utility
+        return not _console_paused
+
+
+class ChecksumWarningFilter(logging.Filter):
+    """Filter allowing only checksum warnings and errors to surface."""
+
+    def filter(
+        self, record: logging.LogRecord
+    ) -> bool:  # pragma: no cover - simple filter
+        if record.levelno >= logging.ERROR:
+            return True
+        return (
+            record.levelno == logging.WARNING
+            and Path(record.pathname).name == "checksum.py"
+        )
+
+
+def pause_console_logging() -> None:
+    """Temporarily pause logging to console handlers."""
+    global _console_paused
+    _console_paused = True
+
+
+def resume_console_logging() -> None:
+    """Resume logging to console handlers."""
+    global _console_paused
+    _console_paused = False
+
+
+@contextmanager
+def console_logging_paused() -> None:
+    """Context manager to pause console logging within a block."""
+    pause_console_logging()
+    try:
+        yield
+    finally:
+        resume_console_logging()
+
+
+def pause_logging_for_ui(func):
+    """Decorator to pause console logging while ``func`` executes."""
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        with console_logging_paused():
+            return func(*args, **kwargs)
+
+    return wrapper

src/utils/memory_protection.py

@@ -3,9 +3,16 @@ from __future__ import annotations
 import os
 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 
+# TODO: Replace this Python implementation with a Rust/WASM module for
+# critical cryptographic operations.
+
 
 class InMemorySecret:
-    """Store sensitive data encrypted in RAM using AES-GCM."""
+    """Store sensitive data encrypted in RAM using AES-GCM.
+
+    Zeroization is best-effort only; Python's memory management may retain
+    copies of the plaintext.
+    """
 
     def __init__(self, data: bytes) -> None:
         if not isinstance(data, (bytes, bytearray)):

src/utils/password_prompt.py

@@ -13,7 +13,9 @@ Ensure that all dependencies are installed and properly configured in your envir
 
 from utils.seed_prompt import masked_input
 import logging
+import os
 import sys
+import time
 import unicodedata
 
 from termcolor import colored
@@ -28,18 +30,55 @@ colorama_init()
 logger = logging.getLogger(__name__)
 
 
+DEFAULT_MAX_ATTEMPTS = 5
+
+
+def _env_password() -> str | None:
+    """Return a password supplied via environment for non-interactive use."""
+
+    return os.getenv("SEEDPASS_TEST_PASSWORD") or os.getenv("SEEDPASS_PASSWORD")
+
+
+def _get_max_attempts(override: int | None = None) -> int:
+    """Return the configured maximum number of prompt attempts."""
+
+    if override is not None:
+        return override
+    env = os.getenv("SEEDPASS_MAX_PROMPT_ATTEMPTS")
+    if env is not None:
+        try:
+            return int(env)
+        except ValueError:
+            pass
+    return DEFAULT_MAX_ATTEMPTS
+
+
+def _apply_backoff(attempts: int, max_attempts: int) -> None:
+    """Sleep using exponential backoff unless disabled."""
+
+    if max_attempts == 0:
+        return
+    delay = 2 ** (attempts - 1)
+    time.sleep(delay)
+
+
 class PasswordPromptError(Exception):
     """Exception raised for password prompt errors."""
 
     pass
 
 
-def prompt_new_password() -> str:
+def prompt_new_password(max_retries: int | None = None) -> str:
     """
     Prompts the user to enter and confirm a new password for encrypting the parent seed.
 
     This function ensures that the password meets the minimum length requirement and that the
-    password and confirmation match. It provides user-friendly messages and handles retries.
+    password and confirmation match. It provides user-friendly messages and handles retries with
+    an exponential backoff between attempts.
+
+    Parameters:
+        max_retries (int | None): Maximum number of attempts before aborting. ``0`` disables the
+            limit. Defaults to the ``SEEDPASS_MAX_PROMPT_ATTEMPTS`` environment variable or ``5``.
 
     Returns:
         str: The confirmed password entered by the user.
@@ -47,10 +86,17 @@ def prompt_new_password() -> str:
     Raises:
         PasswordPromptError: If the user fails to provide a valid password after multiple attempts.
     """
-    max_retries = 5
+    env_pw = _env_password()
+    if env_pw:
+        normalized = unicodedata.normalize("NFKD", env_pw)
+        if len(normalized) < MIN_PASSWORD_LENGTH:
+            raise PasswordPromptError("Environment password too short")
+        return normalized
+
+    max_retries = _get_max_attempts(max_retries)
     attempts = 0
 
-    while attempts < max_retries:
+    while max_retries == 0 or attempts < max_retries:
         try:
             password = masked_input("Enter a new password: ").strip()
             confirm_password = masked_input("Confirm your password: ").strip()
@@ -61,6 +107,7 @@ def prompt_new_password() -> str:
                 )
                 logging.warning("User attempted to enter an empty password.")
                 attempts += 1
+                _apply_backoff(attempts, max_retries)
                 continue
 
             if len(password) < MIN_PASSWORD_LENGTH:
@@ -74,6 +121,7 @@ def prompt_new_password() -> str:
                     f"User entered a password shorter than {MIN_PASSWORD_LENGTH} characters."
                 )
                 attempts += 1
+                _apply_backoff(attempts, max_retries)
                 continue
 
             if password != confirm_password:
@@ -82,6 +130,7 @@ def prompt_new_password() -> str:
                 )
                 logging.warning("User entered mismatching passwords.")
                 attempts += 1
+                _apply_backoff(attempts, max_retries)
                 continue
 
             # Normalize the password to NFKD form
@@ -99,6 +148,7 @@ def prompt_new_password() -> str:
             )
             print(colored(f"Error: {e}", "red"))
             attempts += 1
+            _apply_backoff(attempts, max_retries)
 
     print(colored("Maximum password attempts exceeded. Exiting.", "red"))
     logging.error("User failed to provide a valid password after multiple attempts.")
@@ -106,16 +156,19 @@ def prompt_new_password() -> str:
 
 
 def prompt_existing_password(
-    prompt_message: str = "Enter your password: ", max_retries: int = 5
+    prompt_message: str = "Enter your password: ", max_retries: int | None = None
 ) -> str:
     """
     Prompt the user for an existing password.
 
-    The user will be reprompted on empty input up to ``max_retries`` times.
+    The user will be reprompted on empty input up to ``max_retries`` times with
+    an exponential backoff between attempts.
 
     Parameters:
         prompt_message (str): Message displayed when prompting for the password.
-        max_retries (int): Number of attempts allowed before aborting.
+        max_retries (int | None): Number of attempts allowed before aborting. ``0``
+            disables the limit. Defaults to the ``SEEDPASS_MAX_PROMPT_ATTEMPTS``
+            environment variable or ``5``.
 
     Returns:
         str: The password provided by the user.
@@ -124,8 +177,13 @@ def prompt_existing_password(
         PasswordPromptError: If the user interrupts the operation or exceeds
             ``max_retries`` attempts.
     """
+    env_pw = _env_password()
+    if env_pw:
+        return unicodedata.normalize("NFKD", env_pw)
+
+    max_retries = _get_max_attempts(max_retries)
     attempts = 0
-    while attempts < max_retries:
+    while max_retries == 0 or attempts < max_retries:
         try:
             password = masked_input(prompt_message).strip()
 
@@ -135,6 +193,7 @@ def prompt_existing_password(
                 )
                 logging.warning("User attempted to enter an empty password.")
                 attempts += 1
+                _apply_backoff(attempts, max_retries)
                 continue
 
             normalized_password = unicodedata.normalize("NFKD", password)
@@ -152,6 +211,7 @@ def prompt_existing_password(
             )
             print(colored(f"Error: {e}", "red"))
             attempts += 1
+            _apply_backoff(attempts, max_retries)
 
     raise PasswordPromptError("Maximum password attempts exceeded")
 

src/utils/seed_prompt.py

@@ -1,5 +1,6 @@
 import os
 import sys
+import time
 
 try:
     import msvcrt  # type: ignore
@@ -14,6 +15,37 @@ except ImportError:  # pragma: no cover - POSIX only
     tty = None  # type: ignore
 
 from utils.terminal_utils import clear_screen
+from utils.logging_utils import pause_console_logging, resume_console_logging
+
+DEFAULT_MAX_ATTEMPTS = 5
+
+
+def _get_max_attempts(override: int | None = None) -> int:
+    """Return the configured maximum number of attempts.
+
+    ``override`` takes precedence, followed by the
+    ``SEEDPASS_MAX_PROMPT_ATTEMPTS`` environment variable. A value of ``0``
+    disables the limit entirely.
+    """
+
+    if override is not None:
+        return override
+    env = os.getenv("SEEDPASS_MAX_PROMPT_ATTEMPTS")
+    if env is not None:
+        try:
+            return int(env)
+        except ValueError:
+            pass
+    return DEFAULT_MAX_ATTEMPTS
+
+
+def _apply_backoff(attempts: int, max_attempts: int) -> None:
+    """Sleep using exponential backoff unless the limit is disabled."""
+
+    if max_attempts == 0:
+        return
+    delay = 2 ** (attempts - 1)
+    time.sleep(delay)
 
 
 def _masked_input_windows(prompt: str) -> str:
@@ -26,6 +58,8 @@ def _masked_input_windows(prompt: str) -> str:
     buffer: list[str] = []
     while True:
         ch = msvcrt.getwch()
+        if ch == "\x03":
+            raise KeyboardInterrupt
         if ch in ("\r", "\n"):
             sys.stdout.write("\n")
             return "".join(buffer)
@@ -53,6 +87,8 @@ def _masked_input_posix(prompt: str) -> str:
         tty.setraw(fd)
         while True:
             ch = sys.stdin.read(1)
+            if ch == "\x03":
+                raise KeyboardInterrupt
             if ch in ("\r", "\n"):
                 sys.stdout.write("\n")
                 return "".join(buffer)
@@ -70,12 +106,19 @@ def _masked_input_posix(prompt: str) -> str:
 
 def masked_input(prompt: str) -> str:
     """Return input from the user while masking typed characters."""
-    if sys.platform == "win32":
-        return _masked_input_windows(prompt)
-    return _masked_input_posix(prompt)
+    func = _masked_input_windows if sys.platform == "win32" else _masked_input_posix
+    pause_console_logging()
+    try:
+        return func(prompt)
+    except KeyboardInterrupt:
+        raise
+    except Exception:  # pragma: no cover - fallback when TTY operations fail
+        return input(prompt)
+    finally:
+        resume_console_logging()
 
 
-def prompt_seed_words(count: int = 12) -> str:
+def prompt_seed_words(count: int = 12, *, max_attempts: int | None = None) -> str:
     """Prompt the user for a BIP-39 seed phrase.
 
     The user is asked for each word one at a time. A numbered list is
@@ -88,6 +131,10 @@ def prompt_seed_words(count: int = 12) -> str:
     ----------
     count:
         Number of words to prompt for. Defaults to ``12``.
+    max_attempts:
+        Maximum number of invalid attempts before aborting. ``0`` disables the
+        limit. Defaults to the ``SEEDPASS_MAX_PROMPT_ATTEMPTS`` environment
+        variable or ``5`` if unset.
 
     Returns
     -------
@@ -105,6 +152,9 @@ def prompt_seed_words(count: int = 12) -> str:
     m = Mnemonic("english")
     words: list[str] = [""] * count
 
+    max_attempts = _get_max_attempts(max_attempts)
+    attempts = 0
+
     idx = 0
     while idx < count:
         clear_screen()
@@ -113,6 +163,10 @@ def prompt_seed_words(count: int = 12) -> str:
         entered = masked_input(f"Enter word number {idx+1}: ").strip().lower()
         if entered not in m.wordlist:
             print("Invalid word, try again.")
+            attempts += 1
+            if max_attempts != 0 and attempts >= max_attempts:
+                raise ValueError("Maximum seed prompt attempts exceeded")
+            _apply_backoff(attempts, max_attempts)
             continue
         words[idx] = entered
         idx += 1
@@ -141,9 +195,17 @@ def prompt_seed_words(count: int = 12) -> str:
                         words[i] = new_word
                         break
                     print("Invalid word, try again.")
+                    attempts += 1
+                    if max_attempts != 0 and attempts >= max_attempts:
+                        raise ValueError("Maximum seed prompt attempts exceeded")
+                    _apply_backoff(attempts, max_attempts)
                 # Ask for confirmation again with the new word
             else:
                 print("Please respond with 'Y' or 'N'.")
+                attempts += 1
+                if max_attempts != 0 and attempts >= max_attempts:
+                    raise ValueError("Maximum seed prompt attempts exceeded")
+                _apply_backoff(attempts, max_attempts)
                 continue
 
     phrase = " ".join(words)

src/utils/terminal_utils.py

@@ -1,7 +1,8 @@
 """Utility functions for terminal output."""
 
+import logging
 import sys
-
+import queue
 
 from termcolor import colored
 
@@ -17,8 +18,11 @@ def format_profile(fingerprint: str | None, pm=None) -> str | None:
             name = pm.fingerprint_manager.get_name(fingerprint)
             if name:
                 return f"{name} ({fingerprint})"
-        except Exception:
-            pass
+        except Exception as exc:  # pragma: no cover - unexpected errors
+            logging.error(
+                "Error retrieving name for fingerprint %s: %s", fingerprint, exc
+            )
+            raise
     return fingerprint
 
 
@@ -93,8 +97,11 @@ def clear_header_with_notification(
     if hasattr(pm, "get_current_notification"):
         try:
             note = pm.get_current_notification()
-        except Exception:
+        except (queue.Empty, AttributeError):
             note = None
+        except Exception as exc:  # pragma: no cover - unexpected errors
+            logging.error("Error getting current notification: %s", exc)
+            raise
 
     line = ""
     if note: