Refactor handle_add_* methods to remove duplication

- Introduced `_clear_header_add_entry`, `_mark_dirty`, `_sync_and_pause` helpers.
- Introduced `_entry_op_context` context manager for error handling.
- Applied these patterns to `handle_add_totp`, `handle_add_ssh_key`, `handle_add_seed`, `handle_add_pgp`, `handle_add_nostr_key`, `handle_add_key_value`, and `handle_add_managed_account`.
- Ensured `is_dirty` flag is set immediately after entry creation to prevent data loss if subsequent UI steps are cancelled.

Co-authored-by: PR0M3TH3AN <[email protected]>
This commit is contained in:
google-labs-jules[bot]andPR0M3TH3AN committed 2026-02-26 19:18:51 +00:00
1 parent ece0dda09f
commit 4ebc14e703
1 file changed
+70 -171
+70 -171
View File
@@ -25,6 +25,7 @@ import threading
import queue import queue
from dataclasses import dataclass from dataclasses import dataclass
import dataclasses import dataclasses
from contextlib import contextmanager
from functools import wraps from functools import wraps
from termcolor import colored from termcolor import colored
from utils.color_scheme import color_text from utils.color_scheme import color_text
@@ -2159,6 +2160,43 @@ class PasswordManager:
self.notify("Starting with a new, empty vault.", level="INFO") self.notify("Starting with a new, empty vault.", level="INFO")
return 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()
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_password(self) -> None: def handle_add_password(self) -> None:
self.entry_service.handle_add_password() self.entry_service.handle_add_password()
@@ -2166,16 +2204,9 @@ class PasswordManager:
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_totp(self) -> None: def handle_add_totp(self) -> None:
"""Add a TOTP entry either derived from the seed or imported.""" """Add a TOTP entry either derived from the seed or imported."""
try: with self._entry_op_context("TOTP"):
fp, parent_fp, child_fp = self.header_fingerprint_args
while True: while True:
clear_header_with_notification( self._clear_header_add_entry("2FA (TOTP)")
self,
fp,
"Main Menu > Add Entry > 2FA (TOTP)",
parent_fingerprint=parent_fp,
child_fingerprint=child_fp,
)
print("\nAdd TOTP:") print("\nAdd TOTP:")
print("1. Make 2FA") print("1. Make 2FA")
print("2. Import 2FA (paste otpauth URI or secret)") print("2. Import 2FA (paste otpauth URI or secret)")
@@ -2224,8 +2255,7 @@ class PasswordManager:
else: else:
_lbl, secret, _, _ = TotpManager.parse_otpauth(uri) _lbl, secret, _, _ = TotpManager.parse_otpauth(uri)
color_cat = "default" color_cat = "default"
self.is_dirty = True self._mark_dirty()
self.last_update = time.time()
print( print(
colored( colored(
f"\n[+] TOTP entry added with ID {entry_id}.\n", "green" f"\n[+] TOTP entry added with ID {entry_id}.\n", "green"
@@ -2235,14 +2265,7 @@ class PasswordManager:
print(colored(uri, "yellow")) print(colored(uri, "yellow"))
TotpManager.print_qr_code(uri) TotpManager.print_qr_code(uri)
print(color_text(f"Secret: {secret}\n", color_cat)) print(color_text(f"Secret: {secret}\n", color_cat))
try: self._sync_and_pause()
self.start_background_vault_sync()
except Exception as nostr_error:
logging.error(
f"Failed to post updated index to Nostr: {nostr_error}",
exc_info=True,
)
pause()
break break
elif choice == "2": elif choice == "2":
raw = input("Paste otpauth URI or secret: ").strip() raw = input("Paste otpauth URI or secret: ").strip()
@@ -2276,8 +2299,7 @@ class PasswordManager:
tags=tags, tags=tags,
deterministic=False, deterministic=False,
) )
self.is_dirty = True self._mark_dirty()
self.last_update = time.time()
print( print(
colored( colored(
f"\nImported \u2714 Codes for {label} are now stored in SeedPass at ID {entry_id}.", f"\nImported \u2714 Codes for {label} are now stored in SeedPass at ID {entry_id}.",
@@ -2285,14 +2307,7 @@ class PasswordManager:
) )
) )
TotpManager.print_qr_code(uri) TotpManager.print_qr_code(uri)
try: self._sync_and_pause()
self.start_background_vault_sync()
except Exception as nostr_error:
logging.error(
f"Failed to post updated index to Nostr: {nostr_error}",
exc_info=True,
)
pause()
break break
except ValueError as err: except ValueError as err:
print(colored(f"Error: {err}", "red")) print(colored(f"Error: {err}", "red"))
@@ -2300,23 +2315,12 @@ class PasswordManager:
return return
else: else:
print(colored("Invalid choice.", "red")) print(colored("Invalid choice.", "red"))
except Exception as e:
logging.error(f"Error during TOTP setup: {e}", exc_info=True)
print(colored(f"Error: Failed to add TOTP: {e}", "red"))
pause()
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_ssh_key(self) -> None: def handle_add_ssh_key(self) -> None:
"""Add an SSH key pair entry and display the derived keys.""" """Add an SSH key pair entry and display the derived keys."""
try: with self._entry_op_context("SSH key"):
fp, parent_fp, child_fp = self.header_fingerprint_args self._clear_header_add_entry("SSH Key")
clear_header_with_notification(
self,
fp,
"Main Menu > Add Entry > SSH Key",
parent_fingerprint=parent_fp,
child_fingerprint=child_fp,
)
label = input("Label (key): ").strip() label = input("Label (key): ").strip()
if not label: if not label:
print(colored("Error: Label cannot be empty.", "red")) print(colored("Error: Label cannot be empty.", "red"))
@@ -2334,8 +2338,7 @@ class PasswordManager:
priv_pem, pub_pem = self.entry_manager.get_ssh_key_pair( priv_pem, pub_pem = self.entry_manager.get_ssh_key_pair(
index, self.parent_seed index, self.parent_seed
) )
self.is_dirty = True self._mark_dirty()
self.last_update = time.time()
if not confirm_action( if not confirm_action(
"WARNING: Displaying SSH keys reveals sensitive information. Continue? (Y/N): " "WARNING: Displaying SSH keys reveals sensitive information. Continue? (Y/N): "
@@ -2350,31 +2353,13 @@ class PasswordManager:
print(color_text(pub_pem, "default")) print(color_text(pub_pem, "default"))
print(colored("Private Key:", "cyan")) print(colored("Private Key:", "cyan"))
print(color_text(priv_pem, "deterministic")) print(color_text(priv_pem, "deterministic"))
try: self._sync_and_pause()
self.start_background_vault_sync()
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 SSH key setup: {e}", exc_info=True)
print(colored(f"Error: Failed to add SSH key: {e}", "red"))
pause()
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_seed(self) -> None: def handle_add_seed(self) -> None:
"""Add a derived BIP-39 seed phrase entry.""" """Add a derived BIP-39 seed phrase entry."""
try: with self._entry_op_context("seed phrase"):
fp, parent_fp, child_fp = self.header_fingerprint_args self._clear_header_add_entry("Seed Phrase")
clear_header_with_notification(
self,
fp,
"Main Menu > Add Entry > Seed Phrase",
parent_fingerprint=parent_fp,
child_fingerprint=child_fp,
)
label = input("Label: ").strip() label = input("Label: ").strip()
if not label: if not label:
print(colored("Error: Label cannot be empty.", "red")) print(colored("Error: Label cannot be empty.", "red"))
@@ -2395,8 +2380,7 @@ class PasswordManager:
label, self.parent_seed, words_num=words, notes=notes, tags=tags label, self.parent_seed, words_num=words, notes=notes, tags=tags
) )
phrase = self.entry_manager.get_seed_phrase(index, self.parent_seed) phrase = self.entry_manager.get_seed_phrase(index, self.parent_seed)
self.is_dirty = True self._mark_dirty()
self.last_update = time.time()
if not confirm_action( if not confirm_action(
"WARNING: Displaying the seed phrase reveals sensitive information. Continue? (Y/N): " "WARNING: Displaying the seed phrase reveals sensitive information. Continue? (Y/N): "
@@ -2420,31 +2404,13 @@ class PasswordManager:
from .seedqr import encode_seedqr from .seedqr import encode_seedqr
TotpManager.print_qr_code(encode_seedqr(phrase)) TotpManager.print_qr_code(encode_seedqr(phrase))
try: self._sync_and_pause()
self.start_background_vault_sync()
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 seed phrase setup: {e}", exc_info=True)
print(colored(f"Error: Failed to add seed phrase: {e}", "red"))
pause()
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_pgp(self) -> None: def handle_add_pgp(self) -> None:
"""Add a PGP key entry and display the generated key.""" """Add a PGP key entry and display the generated key."""
try: with self._entry_op_context("PGP key"):
fp, parent_fp, child_fp = self.header_fingerprint_args self._clear_header_add_entry("PGP Key")
clear_header_with_notification(
self,
fp,
"Main Menu > Add Entry > PGP Key",
parent_fingerprint=parent_fp,
child_fingerprint=child_fp,
)
label = input("Label: ").strip() label = input("Label: ").strip()
if not label: if not label:
print(colored("Error: Label cannot be empty.", "red")) print(colored("Error: Label cannot be empty.", "red"))
@@ -2472,8 +2438,7 @@ class PasswordManager:
priv_key, fingerprint = self.entry_manager.get_pgp_key( priv_key, fingerprint = self.entry_manager.get_pgp_key(
index, self.parent_seed index, self.parent_seed
) )
self.is_dirty = True self._mark_dirty()
self.last_update = time.time()
if not confirm_action( if not confirm_action(
"WARNING: Displaying the PGP key reveals sensitive information. Continue? (Y/N): " "WARNING: Displaying the PGP key reveals sensitive information. Continue? (Y/N): "
@@ -2488,31 +2453,13 @@ class PasswordManager:
print(colored(f"Notes: {notes}", "cyan")) print(colored(f"Notes: {notes}", "cyan"))
print(colored(f"Fingerprint: {fingerprint}", "cyan")) print(colored(f"Fingerprint: {fingerprint}", "cyan"))
print(color_text(priv_key, "deterministic")) print(color_text(priv_key, "deterministic"))
try: self._sync_and_pause()
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,
)
pause()
except Exception as e:
logging.error(f"Error during PGP key setup: {e}", exc_info=True)
print(colored(f"Error: Failed to add PGP key: {e}", "red"))
pause()
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_nostr_key(self) -> None: def handle_add_nostr_key(self) -> None:
"""Add a Nostr key entry and display the derived keys.""" """Add a Nostr key entry and display the derived keys."""
try: with self._entry_op_context("Nostr key"):
fp, parent_fp, child_fp = self.header_fingerprint_args self._clear_header_add_entry("Nostr Key Pair")
clear_header_with_notification(
self,
fp,
"Main Menu > Add Entry > Nostr Key Pair",
parent_fingerprint=parent_fp,
child_fingerprint=child_fp,
)
label = input("Label: ").strip() label = input("Label: ").strip()
if not label: if not label:
print(colored("Error: Label cannot be empty.", "red")) print(colored("Error: Label cannot be empty.", "red"))
@@ -2528,8 +2475,8 @@ class PasswordManager:
label, self.parent_seed, notes=notes, tags=tags label, self.parent_seed, notes=notes, tags=tags
) )
npub, nsec = self.entry_manager.get_nostr_key_pair(index, self.parent_seed) npub, nsec = self.entry_manager.get_nostr_key_pair(index, self.parent_seed)
self.is_dirty = True self._mark_dirty()
self.last_update = time.time()
print(colored(f"\n[+] Nostr key entry added with ID {index}.\n", "green")) print(colored(f"\n[+] Nostr key entry added with ID {index}.\n", "green"))
print(colored(f"npub: {npub}", "cyan")) print(colored(f"npub: {npub}", "cyan"))
if self.secret_mode_enabled: if self.secret_mode_enabled:
@@ -2548,31 +2495,13 @@ class PasswordManager:
"WARNING: Displaying the nsec QR reveals your private key. Continue? (Y/N): " "WARNING: Displaying the nsec QR reveals your private key. Continue? (Y/N): "
): ):
TotpManager.print_qr_code(nsec) TotpManager.print_qr_code(nsec)
try: self._sync_and_pause()
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,
)
pause()
except Exception as e:
logging.error(f"Error during Nostr key setup: {e}", exc_info=True)
print(colored(f"Error: Failed to add Nostr key: {e}", "red"))
pause()
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_key_value(self) -> None: def handle_add_key_value(self) -> None:
"""Add a generic key/value entry.""" """Add a generic key/value entry."""
try: with self._entry_op_context("key/value"):
fp, parent_fp, child_fp = self.header_fingerprint_args self._clear_header_add_entry("Key/Value")
clear_header_with_notification(
self,
fp,
"Main Menu > Add Entry > Key/Value",
parent_fingerprint=parent_fp,
child_fingerprint=child_fp,
)
label = input("Label: ").strip() label = input("Label: ").strip()
if not label: if not label:
print(colored("Error: Label cannot be empty.", "red")) print(colored("Error: Label cannot be empty.", "red"))
@@ -2614,8 +2543,7 @@ class PasswordManager:
custom_fields=custom_fields, custom_fields=custom_fields,
tags=tags, tags=tags,
) )
self.is_dirty = True self._mark_dirty()
self.last_update = time.time()
print(colored(f"\n[+] Key/Value entry added with ID {index}.\n", "green")) print(colored(f"\n[+] Key/Value entry added with ID {index}.\n", "green"))
if notes: if notes:
@@ -2630,31 +2558,13 @@ class PasswordManager:
) )
else: else:
print(color_text(f"Value: {value}", "deterministic")) print(color_text(f"Value: {value}", "deterministic"))
try: self._sync_and_pause()
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,
)
pause()
except Exception as e:
logging.error(f"Error during key/value setup: {e}", exc_info=True)
print(colored(f"Error: Failed to add key/value entry: {e}", "red"))
pause()
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_managed_account(self) -> None: def handle_add_managed_account(self) -> None:
"""Add a managed account seed entry.""" """Add a managed account seed entry."""
try: with self._entry_op_context("managed account"):
fp, parent_fp, child_fp = self.header_fingerprint_args self._clear_header_add_entry("Managed Account")
clear_header_with_notification(
self,
fp,
"Main Menu > Add Entry > Managed Account",
parent_fingerprint=parent_fp,
child_fingerprint=child_fp,
)
label = input("Label: ").strip() label = input("Label: ").strip()
if not label: if not label:
print(colored("Error: Label cannot be empty.", "red")) print(colored("Error: Label cannot be empty.", "red"))
@@ -2669,9 +2579,9 @@ class PasswordManager:
index = self.entry_manager.add_managed_account( index = self.entry_manager.add_managed_account(
label, self.parent_seed, notes=notes, tags=tags label, self.parent_seed, notes=notes, tags=tags
) )
self._mark_dirty()
seed = self.entry_manager.get_managed_account_seed(index, self.parent_seed) seed = self.entry_manager.get_managed_account_seed(index, self.parent_seed)
self.is_dirty = True
self.last_update = time.time()
print( print(
colored( colored(
f"\n[+] Managed account '{label}' added with ID {index}.\n", f"\n[+] Managed account '{label}' added with ID {index}.\n",
@@ -2693,18 +2603,7 @@ class PasswordManager:
from .seedqr import encode_seedqr from .seedqr import encode_seedqr
TotpManager.print_qr_code(encode_seedqr(seed)) TotpManager.print_qr_code(encode_seedqr(seed))
try: self._sync_and_pause()
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,
)
pause()
except Exception as e:
logging.error(f"Error during managed account setup: {e}", exc_info=True)
print(colored(f"Error: Failed to add managed account: {e}", "red"))
pause()
def show_entry_details_by_index(self, index: int) -> None: def show_entry_details_by_index(self, index: int) -> None:
"""Display details for entry ``index`` and offer actions.""" """Display details for entry ``index`` and offer actions."""