mirror of
https://github.com/PR0M3TH3AN/SeedPass.git
synced 2025-09-08 15:28:44 +00:00
161 KiB
161 KiB
Okay, this is a significant refactoring effort! Let's break it down and implement the changes based on your plan and the provided code.
Phase 1: Create New Files and Basic Structures
1. Create password_manager/kinds.py:
# password_manager/kinds.py
import logging
from typing import Dict, Callable, List, Any
from termcolor import colored
# Forward declaration for type hinting if handlers need PasswordManager instance later
# class PasswordManager: pass
# from .encryption import EncryptionManager
logger = logging.getLogger(__name__)
# Placeholder handlers - will be imported properly later
def handle_generated_password(entry_data: Dict[str, Any], fingerprint: str, **kwargs):
logger.warning("Placeholder handler called for generated_password")
print(colored(f"Processing Generated Password (Placeholder): {entry_data.get('title')}", "grey"))
def handle_stored_password(entry_data: Dict[str, Any], fingerprint: str, **kwargs):
logger.warning("Placeholder handler called for stored_password")
print(colored(f"Processing Stored Password (Placeholder): {entry_data.get('title')}", "grey"))
def handle_note(entry_data: Dict[str, Any], fingerprint: str, **kwargs):
logger.warning("Placeholder handler called for note")
print(colored(f"Processing Note (Placeholder): Content length {len(entry_data.get('content', ''))}", "grey"))
# --- Actual KINDS Definition ---
# We'll import real handlers after creating them.
# Define the structure for kinds. Each kind maps to:
# - handler: The function to process/display the entry data.
# - description: User-friendly description for menus.
# - fields: List of expected keys within the 'data' part of an entry.
# - nostr_kind: The Nostr event kind used for this entry type.
# - identifier_tag: The Nostr tag ('d' tag) value prefix for this entry type.
KINDS: Dict[str, Dict[str, Any]] = {
"generated_password": {
"handler": handle_generated_password, # Placeholder
"description": "Generated Password (using BIP-85 index)",
"fields": ["title", "username", "email", "url", "length", "bip85_index"], # Note: password is not stored, bip85_index is key
"nostr_kind": 31111, # Example custom kind for SeedPass entries
"identifier_tag": "seedpass_gp_" # gp for generated password
},
"stored_password": {
"handler": handle_stored_password, # Placeholder
"description": "Stored Password / Credential",
"fields": ["title", "username", "password", "url", "notes"], # Password stored encrypted in 'data'
"nostr_kind": 31111,
"identifier_tag": "seedpass_sp_" # sp for stored password
},
"note": {
"handler": handle_note, # Placeholder
"description": "Secure Note",
"fields": ["title", "content", "tags"],
"nostr_kind": 31111,
"identifier_tag": "seedpass_note_"
},
# Add new kinds here in the future
}
def get_kind_details(kind_name: str) -> Optional[Dict[str, Any]]:
"""Safely retrieves details for a given kind."""
return KINDS.get(kind_name)
def get_all_kinds() -> List[str]:
"""Returns a list of all defined kind names."""
return list(KINDS.keys())
def get_nostr_kind(kind_name: str) -> Optional[int]:
"""Gets the Nostr event kind for a SeedPass kind."""
details = get_kind_details(kind_name)
return details.get("nostr_kind") if details else None
def get_identifier_tag_prefix(kind_name: str) -> Optional[str]:
"""Gets the 'd' tag prefix for a SeedPass kind."""
details = get_kind_details(kind_name)
return details.get("identifier_tag") if details else None
def get_required_fields(kind_name: str) -> List[str]:
"""Gets the list of required fields for a SeedPass kind."""
details = get_kind_details(kind_name)
return details.get("fields", []) if details else []
def get_kind_handler(kind_name: str) -> Optional[Callable]:
"""Gets the handler function for a SeedPass kind."""
details = get_kind_details(kind_name)
return details.get("handler") if details else None
2. Create password_manager/handlers/ directory and __init__.py:
mkdir -p password_manager/handlers
touch password_manager/handlers/__init__.py
3. Create Handler Files:
password_manager/handlers/generated_password_handler.py:# password_manager/handlers/generated_password_handler.py import logging from typing import Dict, Any from termcolor import colored # Avoid circular import - PasswordManager/EncryptionManager likely passed in kwargs # from ..manager import PasswordManager # from ..encryption import EncryptionManager # from ..password_generation import PasswordGenerator logger = logging.getLogger(__name__) def handle_generated_password(entry_data: Dict[str, Any], fingerprint: str, **kwargs): """Handles processing/displaying a generated password entry.""" # Expect PasswordGenerator instance in kwargs for actual generation password_generator = kwargs.get("password_generator") if not password_generator: logger.error("PasswordGenerator not provided to generated_password handler.") print(colored("Error: Cannot process generated password - internal setup issue.", "red")) return title = entry_data.get("title", "N/A") username = entry_data.get("username", "") email = entry_data.get("email", "") url = entry_data.get("url", "") length = entry_data.get("length") bip85_index = entry_data.get("bip85_index") if length is None or bip85_index is None: logger.error(f"Missing length or bip85_index for generated password entry: {title}") print(colored(f"Error: Incomplete data for generated password '{title}'.", "red")) return try: # Regenerate the password on the fly password = password_generator.generate_password(length=length, index=bip85_index) print(colored(f"--- Generated Password Entry ---", "cyan")) print(colored(f" Title: {title}", "cyan")) if username: print(colored(f" Username: {username}", "cyan")) if email: print(colored(f" Email: {email}", "cyan")) if url: print(colored(f" URL: {url}", "cyan")) print(colored(f" Length: {length}", "cyan")) print(colored(f" Index: {bip85_index}", "cyan")) print(colored(f" Password: {password}", "yellow")) # Display generated password print(colored(f"--------------------------------", "cyan")) except Exception as e: logger.error(f"Failed to generate password for entry {title}: {e}", exc_info=True) print(colored(f"Error generating password for '{title}': {e}", "red"))password_manager/handlers/stored_password_handler.py:# password_manager/handlers/stored_password_handler.py import logging from typing import Dict, Any from termcolor import colored # from ..encryption import EncryptionManager # Passed in kwargs logger = logging.getLogger(__name__) def handle_stored_password(entry_data: Dict[str, Any], fingerprint: str, **kwargs): """Handles processing/displaying a stored password entry.""" encryption_manager = kwargs.get("encryption_manager") if not encryption_manager: logger.error("EncryptionManager not provided to stored_password handler.") print(colored("Error: Cannot process stored password - internal setup issue.", "red")) return title = entry_data.get("title", "N/A") username = entry_data.get("username", "") encrypted_password_b64 = entry_data.get("password") # Expecting base64 encoded encrypted bytes url = entry_data.get("url", "") notes = entry_data.get("notes", "") if not encrypted_password_b64: logger.error(f"Missing encrypted password for stored password entry: {title}") print(colored(f"Error: Incomplete data for stored password '{title}'.", "red")) return try: # Decode from base64 then decrypt import base64 encrypted_password_bytes = base64.b64decode(encrypted_password_b64) password = encryption_manager.decrypt_data(encrypted_password_bytes).decode('utf-8') print(colored(f"--- Stored Password Entry ---", "cyan")) print(colored(f" Title: {title}", "cyan")) if username: print(colored(f" Username: {username}", "cyan")) if url: print(colored(f" URL: {url}", "cyan")) if notes: print(colored(f" Notes: {notes}", "cyan")) print(colored(f" Password: {password}", "yellow")) # Display decrypted password print(colored(f"-----------------------------", "cyan")) except Exception as e: logger.error(f"Failed to decrypt stored password for entry {title}: {e}", exc_info=True) print(colored(f"Error decrypting password for '{title}': {e}", "red"))password_manager/handlers/note_handler.py:# password_manager/handlers/note_handler.py import logging from typing import Dict, Any from termcolor import colored # from ..encryption import EncryptionManager # Passed in kwargs logger = logging.getLogger(__name__) def handle_note(entry_data: Dict[str, Any], fingerprint: str, **kwargs): """Handles processing/displaying a secure note entry.""" encryption_manager = kwargs.get("encryption_manager") if not encryption_manager: logger.error("EncryptionManager not provided to note handler.") print(colored("Error: Cannot process note - internal setup issue.", "red")) return title = entry_data.get("title", "N/A") encrypted_content_b64 = entry_data.get("content") # Expecting base64 encoded encrypted bytes tags = entry_data.get("tags", []) if not encrypted_content_b64: logger.error(f"Missing encrypted content for note entry: {title}") print(colored(f"Error: Incomplete data for note '{title}'.", "red")) return try: # Decode from base64 then decrypt import base64 encrypted_content_bytes = base64.b64decode(encrypted_content_b64) content = encryption_manager.decrypt_data(encrypted_content_bytes).decode('utf-8') print(colored(f"--- Secure Note Entry ---", "cyan")) print(colored(f" Title: {title}", "cyan")) if tags: print(colored(f" Tags: {', '.join(tags)}", "cyan")) print(colored(f" Content:\n{content}", "yellow")) print(colored(f"-------------------------", "cyan")) except Exception as e: logger.error(f"Failed to decrypt note content for entry {title}: {e}", exc_info=True) print(colored(f"Error decrypting note '{title}': {e}", "red"))- Update
password_manager/kinds.pyimports:# password_manager/kinds.py # ... (other imports) # --- Import Real Handlers --- from .handlers.generated_password_handler import handle_generated_password from .handlers.stored_password_handler import handle_stored_password from .handlers.note_handler import handle_note # Future handlers can be imported here # --- KINDS Definition --- (Use imported handlers now) KINDS: Dict[str, Dict[str, Any]] = { "generated_password": { "handler": handle_generated_password, # Use imported handler "description": "Generated Password (using BIP-85 index)", "fields": ["title", "username", "email", "url", "length", "bip85_index"], "nostr_kind": 31111, "identifier_tag": "seedpass_gp_" }, "stored_password": { "handler": handle_stored_password, # Use imported handler "description": "Stored Password / Credential", "fields": ["title", "username", "password", "url", "notes"], "nostr_kind": 31111, "identifier_tag": "seedpass_sp_" }, "note": { "handler": handle_note, # Use imported handler "description": "Secure Note", "fields": ["title", "content", "tags"], "nostr_kind": 31111, "identifier_tag": "seedpass_note_" }, # ... } # ... (rest of the helper functions)
4. Create password_manager/state_manager.py:
# password_manager/state_manager.py
import json
import logging
from pathlib import Path
from typing import Optional, Dict, Any
import fcntl
import os
import traceback
from utils.file_lock import lock_file # Use the existing file lock utility
logger = logging.getLogger(__name__)
class StateManager:
"""Manages persistent state for a fingerprint, like last index and sync time."""
STATE_FILENAME = "seedpass_state.json"
def __init__(self, fingerprint_dir: Path):
self.fingerprint_dir = fingerprint_dir
self.state_file_path = self.fingerprint_dir / self.STATE_FILENAME
self._state: Dict[str, Any] = self._load_state()
def _load_state(self) -> Dict[str, Any]:
"""Loads state from the JSON file, returns default if not found or invalid."""
default_state = {"last_generated_password_index": -1, "last_nostr_sync_time": 0}
if not self.state_file_path.exists():
logger.info(f"State file not found for {self.fingerprint_dir.name}. Initializing default state.")
return default_state
try:
with lock_file(self.state_file_path, fcntl.LOCK_SH):
with open(self.state_file_path, 'r') as f:
state = json.load(f)
# Ensure essential keys exist
for key, default_value in default_state.items():
if key not in state:
state[key] = default_value
logger.debug(f"State loaded for {self.fingerprint_dir.name}")
return state
except (json.JSONDecodeError, IOError, ValueError) as e:
logger.error(f"Failed to load or parse state file {self.state_file_path}: {e}. Using default state.", exc_info=True)
return default_state
except Exception as e:
logger.error(f"Unexpected error loading state file {self.state_file_path}: {e}. Using default state.", exc_info=True)
return default_state
def _save_state(self) -> bool:
"""Saves the current state to the JSON file."""
try:
with lock_file(self.state_file_path, fcntl.LOCK_EX):
with open(self.state_file_path, 'w') as f:
json.dump(self._state, f, indent=4)
os.chmod(self.state_file_path, 0o600) # Ensure permissions
logger.debug(f"State saved for {self.fingerprint_dir.name}")
return True
except IOError as e:
logger.error(f"Failed to save state file {self.state_file_path}: {e}", exc_info=True)
return False
except Exception as e:
logger.error(f"Unexpected error saving state file {self.state_file_path}: {e}", exc_info=True)
return False
def get_last_generated_password_index(self) -> int:
"""Gets the last used index for generated passwords."""
# Ensure the key exists, defaulting if necessary
if "last_generated_password_index" not in self._state:
self._state["last_generated_password_index"] = -1
return self._state.get("last_generated_password_index", -1)
def set_last_generated_password_index(self, index: int) -> bool:
"""Sets the last used index for generated passwords and saves state."""
if not isinstance(index, int) or index < -1:
logger.error(f"Invalid index provided to set_last_generated_password_index: {index}")
return False
self._state["last_generated_password_index"] = index
logger.info(f"Setting last generated password index to: {index}")
return self._save_state()
def get_next_generated_password_index(self) -> int:
"""Gets the next available index and increments the stored value."""
current_index = self.get_last_generated_password_index()
next_index = current_index + 1
if self.set_last_generated_password_index(next_index):
return next_index
else:
# Handle save failure - maybe raise an exception?
logger.critical("Failed to save state after incrementing index! Potential index reuse risk.")
raise RuntimeError("Failed to update state for next generated password index.")
def get_last_nostr_sync_time(self) -> int:
"""Gets the timestamp of the last successful Nostr sync."""
# Ensure the key exists, defaulting if necessary
if "last_nostr_sync_time" not in self._state:
self._state["last_nostr_sync_time"] = 0
return self._state.get("last_nostr_sync_time", 0)
def set_last_nostr_sync_time(self, timestamp: int) -> bool:
"""Sets the timestamp of the last successful Nostr sync and saves state."""
if not isinstance(timestamp, int) or timestamp < 0:
logger.error(f"Invalid timestamp provided to set_last_nostr_sync_time: {timestamp}")
return False
self._state["last_nostr_sync_time"] = timestamp
logger.info(f"Setting last Nostr sync time to: {timestamp}")
return self._save_state()
Phase 2: Refactor EntryManager and BackupManager
-
password_manager/entry_management.py(Refactored):# password_manager/entry_management.py import json import logging import hashlib import sys import os import shutil import time import traceback import fcntl from pathlib import Path from typing import Optional, Dict, Any, List from termcolor import colored from .encryption import EncryptionManager # Keep this from utils.file_lock import lock_file # Keep this logger = logging.getLogger(__name__) class EntryManager: """Manages storage and retrieval of individual encrypted entry files.""" ENTRY_FILENAME_TEMPLATE = "entry_{entry_num}.json.enc" ENTRY_CHECKSUM_FIELD = "checksum" # Field within the decrypted JSON metadata def __init__(self, encryption_manager: EncryptionManager, fingerprint_dir: Path): """ Initializes the EntryManager. :param encryption_manager: The encryption manager instance. :param fingerprint_dir: The directory corresponding to the fingerprint. """ self.encryption_manager = encryption_manager self.fingerprint_dir = fingerprint_dir self.entries_dir = self.fingerprint_dir / 'entries' # Ensure the entries directory exists self.entries_dir.mkdir(parents=True, exist_ok=True) logger.debug(f"EntryManager initialized for directory {self.entries_dir}") def _get_entry_path(self, entry_num: int) -> Path: """Constructs the file path for a given entry number.""" return self.entries_dir / self.ENTRY_FILENAME_TEMPLATE.format(entry_num=entry_num) def get_next_entry_num(self) -> int: """Determines the next available entry number based on existing files.""" try: existing_entries = list(self.entries_dir.glob('entry_*.json.enc')) if not existing_entries: return 0 entry_nums = [] for entry_path in existing_entries: try: # Extract number from filename like 'entry_123.json.enc' num_str = entry_path.stem.split('_')[1] entry_nums.append(int(num_str)) except (IndexError, ValueError): logger.warning(f"Could not parse entry number from filename: {entry_path.name}") return max(entry_nums) + 1 if entry_nums else 0 except Exception as e: logger.error(f"Error determining next entry number: {e}", exc_info=True) print(colored(f"Error determining next entry number: {e}", 'red')) # Returning 0 might be risky, perhaps raise or exit? raise RuntimeError("Could not determine the next entry number.") from e def calculate_checksum(self, data_dict: Dict[str, Any]) -> str: """Calculates SHA-256 checksum of the provided data dictionary.""" try: # Ensure consistent ordering for checksum calculation data_string = json.dumps(data_dict, sort_keys=True).encode('utf-8') return hashlib.sha256(data_string).hexdigest() except Exception as e: logger.error(f"Error calculating checksum: {e}", exc_info=True) raise ValueError("Could not calculate checksum for data.") from e def save_entry(self, entry_num: int, encrypted_entry_data: bytes) -> bool: """Saves the encrypted data for a specific entry number.""" entry_path = self._get_entry_path(entry_num) try: with lock_file(entry_path, fcntl.LOCK_EX): with open(entry_path, 'wb') as f: f.write(encrypted_entry_data) os.chmod(entry_path, 0o600) # Ensure permissions logger.info(f"Entry {entry_num} saved successfully to {entry_path}.") return True except IOError as e: logger.error(f"Failed to save entry {entry_num} to {entry_path}: {e}", exc_info=True) print(colored(f"Error: Failed to save entry {entry_num}: {e}", 'red')) return False except Exception as e: logger.error(f"Unexpected error saving entry {entry_num}: {e}", exc_info=True) return False def load_entry(self, entry_num: int) -> Optional[Dict[str, Any]]: """Loads, decrypts, and returns the entry data for a specific entry number.""" entry_path = self._get_entry_path(entry_num) if not entry_path.exists(): logger.warning(f"Entry file not found: {entry_path}") return None try: # Use EncryptionManager's decrypt_file which handles locking decrypted_data_bytes = self.encryption_manager.decrypt_file(entry_path.relative_to(self.fingerprint_dir)) entry_dict = json.loads(decrypted_data_bytes.decode('utf-8')) logger.debug(f"Entry {entry_num} loaded successfully.") return entry_dict except json.JSONDecodeError as e: logger.error(f"Failed to decode JSON for entry {entry_num} from {entry_path}: {e}", exc_info=True) print(colored(f"Error: Corrupted data found for entry {entry_num}.", 'red')) return None except Exception as e: # Includes InvalidToken from decrypt_file logger.error(f"Failed to load or decrypt entry {entry_num} from {entry_path}: {e}", exc_info=True) # Don't show raw error to user unless needed # print(colored(f"Error: Failed to load entry {entry_num}: {e}", 'red')) return None def get_entry_checksum(self, entry_num: int) -> Optional[str]: """Retrieves the stored checksum from within an entry's metadata.""" entry_data = self.load_entry(entry_num) if entry_data: checksum = entry_data.get("metadata", {}).get(self.ENTRY_CHECKSUM_FIELD) if checksum: return checksum else: logger.warning(f"Checksum not found in metadata for entry {entry_num}") return None def delete_entry_file(self, entry_num: int) -> bool: """Deletes the file associated with an entry number.""" entry_path = self._get_entry_path(entry_num) if not entry_path.exists(): logger.warning(f"Attempted to delete non-existent entry file: {entry_path}") return False # Or True, as the state is achieved? Decide consistency. try: with lock_file(entry_path, fcntl.LOCK_EX): # Lock before deleting entry_path.unlink() logger.info(f"Entry file {entry_path} deleted successfully.") return True except OSError as e: logger.error(f"Failed to delete entry file {entry_path}: {e}", exc_info=True) print(colored(f"Error: Failed to delete entry file {entry_num}: {e}", 'red')) return False except Exception as e: logger.error(f"Unexpected error deleting entry file {entry_num}: {e}", exc_info=True) return False def list_all_entry_nums(self) -> List[int]: """Lists all available entry numbers by scanning the directory.""" entry_nums = [] try: for entry_path in self.entries_dir.glob('entry_*.json.enc'): try: num_str = entry_path.stem.split('_')[1] entry_nums.append(int(num_str)) except (IndexError, ValueError): logger.warning(f"Could not parse entry number from filename: {entry_path.name}") return sorted(entry_nums) except Exception as e: logger.error(f"Error listing entry numbers: {e}", exc_info=True) return [] # --- Methods related to the old single index are removed --- # remove _load_index, _save_index, add_entry (old), retrieve_entry (old) etc. # remove update_checksum (old) # remove get_encrypted_index (old) -
password_manager/backup.py(Refactored):# password_manager/backup.py import logging import os import shutil import time import traceback from pathlib import Path import fcntl # Keep fcntl import if used in lock_file from typing import List, Optional from termcolor import colored from utils.file_lock import lock_file logger = logging.getLogger(__name__) class BackupManager: """Handles backups for individual entry files.""" BACKUP_FILENAME_TEMPLATE = 'entry_{entry_num}_backup_{timestamp}.json.enc' def __init__(self, fingerprint_dir: Path): """ Initializes the BackupManager. :param fingerprint_dir: The directory corresponding to the fingerprint. """ self.fingerprint_dir = fingerprint_dir self.entries_dir = self.fingerprint_dir / 'entries' self.backups_dir = self.fingerprint_dir / 'backups' self.backups_dir.mkdir(parents=True, exist_ok=True) logger.debug(f"BackupManager initialized for backup directory {self.backups_dir}") def _get_entry_path(self, entry_num: int) -> Path: """Constructs the original entry file path.""" return self.entries_dir / f'entry_{entry_num}.json.enc' def create_backup_for_entry(self, entry_num: int) -> Optional[Path]: """Creates a timestamped backup for a specific entry file.""" entry_file = self._get_entry_path(entry_num) if not entry_file.exists(): logger.warning(f"Entry {entry_num} file does not exist at {entry_file}. No backup created.") print(colored(f"Warning: Entry file {entry_num} does not exist. No backup created.", 'yellow')) return None try: timestamp = int(time.time()) backup_filename = self.BACKUP_FILENAME_TEMPLATE.format(entry_num=entry_num, timestamp=timestamp) backup_file_path = self.backups_dir / backup_filename # Lock the source file for reading during copy with lock_file(entry_file, fcntl.LOCK_SH): shutil.copy2(entry_file, backup_file_path) # copy2 preserves metadata logger.info(f"Backup created for entry {entry_num} at '{backup_file_path}'.") print(colored(f"Backup created successfully for entry {entry_num}.", 'green')) return backup_file_path except Exception as e: logger.error(f"Failed to create backup for entry {entry_num}: {e}", exc_info=True) print(colored(f"Error: Failed to create backup for entry {entry_num}: {e}", 'red')) return None def list_backups_for_entry(self, entry_num: int) -> List[Path]: """Lists available backup files for a specific entry, sorted by time (newest first).""" try: backup_pattern = f'entry_{entry_num}_backup_*.json.enc' backup_files = sorted( self.backups_dir.glob(backup_pattern), key=lambda x: x.stat().st_mtime, reverse=True ) return backup_files except Exception as e: logger.error(f"Failed to list backups for entry {entry_num}: {e}", exc_info=True) return [] def list_all_backups(self) -> List[Path]: """Lists all backup files, sorted by time (newest first).""" try: backup_files = sorted( self.backups_dir.glob('entry_*_backup_*.json.enc'), key=lambda x: x.stat().st_mtime, reverse=True ) return backup_files except Exception as e: logger.error(f"Failed to list all backups: {e}", exc_info=True) return [] def display_backups(self, entry_num: Optional[int] = None): """Prints available backups to the console.""" if entry_num is not None: backup_files = self.list_backups_for_entry(entry_num) print(colored(f"Available Backups for Entry {entry_num}:", 'cyan')) else: backup_files = self.list_all_backups() print(colored("Available Backups (All Entries):", 'cyan')) if not backup_files: logger.info("No backup files available.") print(colored("No backup files available.", 'yellow')) return for backup in backup_files: try: creation_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(backup.stat().st_mtime)) print(colored(f"- {backup.name} (Created on: {creation_time})", 'cyan')) except Exception as e: logger.warning(f"Could not get stat for backup file {backup.name}: {e}") print(colored(f"- {backup.name} (Error reading time)", "red")) def restore_entry_from_backup(self, entry_num: int, backup_filename: str) -> bool: """Restores an entry file from a specific backup file.""" entry_file = self._get_entry_path(entry_num) backup_file = self.backups_dir / backup_filename # Basic check to ensure the backup filename matches the entry number pattern if not backup_filename.startswith(f'entry_{entry_num}_backup_'): logger.error(f"Backup filename '{backup_filename}' does not match entry number {entry_num}.") print(colored("Error: Backup file name does not match the entry number.", 'red')) return False if not backup_file.exists(): logger.error(f"Backup file '{backup_file}' not found.") print(colored(f"Error: Backup file '{backup_filename}' not found.", 'red')) return False try: # Lock the destination file exclusively during restore with lock_file(entry_file, fcntl.LOCK_EX): shutil.copy2(backup_file, entry_file) # copy2 preserves metadata logger.info(f"Entry {entry_num} restored successfully from backup '{backup_filename}'.") print(colored(f"Restored entry {entry_num} from backup '{backup_filename}'.", 'green')) return True except Exception as e: logger.error(f"Failed to restore entry {entry_num} from backup '{backup_filename}': {e}", exc_info=True) print(colored(f"Error: Failed to restore entry {entry_num} from backup: {e}", 'red')) return False # --- Methods related to the old single index are removed --- # Remove restore_latest_backup (old), restore_backup_by_timestamp (old) etc.
Phase 3: Refactor PasswordManager
password_manager/manager.py(Major Refactoring):# password_manager/manager.py import sys import json import logging import getpass import os import base64 # Added import uuid # Added from datetime import datetime # Added from typing import Optional, Dict, Any, List import shutil from colorama import Fore, Style # Style Added from termcolor import colored from .encryption import EncryptionManager from .entry_management import EntryManager # Modified Import Path from .password_generation import PasswordGenerator from .backup import BackupManager # Modified Import Path from .state_manager import StateManager # Added from .kinds import KINDS, get_kind_details, get_all_kinds, get_required_fields, get_kind_handler # Added from utils.key_derivation import derive_key_from_password from utils.checksum import calculate_checksum as calculate_script_checksum, verify_checksum as verify_script_checksum # Renamed for clarity from utils.password_prompt import prompt_for_password, prompt_existing_password, confirm_action from constants import ( APP_DIR, PARENT_SEED_FILE as OLD_PARENT_SEED_FILENAME, # Rename old constant if needed SCRIPT_CHECKSUM_FILE, MIN_PASSWORD_LENGTH, MAX_PASSWORD_LENGTH, DEFAULT_PASSWORD_LENGTH, DEFAULT_SEED_BACKUP_FILENAME ) import traceback import bcrypt from pathlib import Path from local_bip85.bip85 import BIP85 from bip_utils import Bip39SeedGenerator from utils.fingerprint_manager import FingerprintManager from nostr.client import NostrClient logger = logging.getLogger(__name__) # --- Define constants for new structure --- ENTRIES_DIR_NAME = "entries" BACKUPS_DIR_NAME = "backups" PARENT_SEED_FILENAME = "parent_seed.enc" HASHED_PASSWORD_FILENAME = "hashed_password.enc" OLD_INDEX_FILENAME = 'seedpass_passwords_db.json.enc' # For migration check class PasswordManager: """ Manages password entries, encryption, Nostr sync, and user interaction using individual entry files and 'kinds'. """ def __init__(self): 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.fingerprint_manager: Optional[FingerprintManager] = None self.state_manager: Optional[StateManager] = None # Added self.parent_seed: Optional[str] = None self.bip85: Optional[BIP85] = None self.nostr_client: Optional[NostrClient] = None self.current_fingerprint: Optional[str] = None # Added for clarity self.fingerprint_dir: Optional[Path] = None # Added for clarity self.entries_dir: Optional[Path] = None # Added self.backups_dir: Optional[Path] = None # Added try: self.initialize_fingerprint_manager() self.setup_parent_seed() # This now includes selecting/adding fingerprint and initializing managers # Perform data migration check *after* managers are initialized for the selected fingerprint if self.fingerprint_dir: # Ensure fingerprint_dir is set self.migrate_data_if_needed() # Initial synchronization with Nostr after setup/migration if self.nostr_client: self.synchronize_with_nostr() # Optional: run sync on startup except Exception as e: logger.critical(f"Critical error during PasswordManager initialization: {e}", exc_info=True) print(colored(f"FATAL ERROR during startup: {e}. Check logs.", "red", attrs=["bold"])) sys.exit(1) 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')) sys.exit(1) def setup_parent_seed(self) -> None: """Guides user through selecting or adding a fingerprint and initializes components.""" fingerprints = self.fingerprint_manager.list_fingerprints() if fingerprints: self.select_or_add_fingerprint() else: print(colored("No existing SeedPass profiles (fingerprints) found.", 'yellow')) self.handle_new_seed_setup() # Ensure initialization happened after selection/creation if not self.current_fingerprint or not self.fingerprint_dir or not self.encryption_manager: logger.critical("Fingerprint selection or initialization failed.") print(colored("Error: Could not set up a valid SeedPass profile.", 'red')) sys.exit(1) def select_or_add_fingerprint(self): """Prompts user to select existing fingerprint or add a new one.""" try: print(colored("\nAvailable SeedPass Profiles (Fingerprints):", 'cyan')) fingerprints = self.fingerprint_manager.list_fingerprints() for idx, fp in enumerate(fingerprints, start=1): print(colored(f"{idx}. {fp}", 'cyan')) print(colored(f"{len(fingerprints) + 1}. Add a new profile (generate or import seed)", 'cyan')) print(colored(f"{len(fingerprints) + 2}. Exit", 'cyan')) while True: choice_str = input("Select a profile by number or choose an action: ").strip() if not choice_str.isdigit(): print(colored("Invalid input. Please enter a number.", 'red')) continue choice = int(choice_str) if 1 <= choice <= len(fingerprints): selected_fingerprint = fingerprints[choice - 1] self.select_fingerprint(selected_fingerprint) break # Exit loop on valid selection elif choice == len(fingerprints) + 1: # Add a new fingerprint new_fingerprint = self.add_new_fingerprint() if new_fingerprint: self.select_fingerprint(new_fingerprint) # Select the newly added one else: print(colored("Failed to add new profile. Exiting.", "red")) sys.exit(1) break # Exit loop elif choice == len(fingerprints) + 2: print(colored("Exiting.", "yellow")) sys.exit(0) else: print(colored("Invalid selection.", 'red')) except Exception as e: logger.error(f"Error during fingerprint selection: {e}", exc_info=True) print(colored(f"Error: Failed to select profile: {e}", 'red')) sys.exit(1) def add_new_fingerprint(self) -> Optional[str]: """Guides user to add a new fingerprint/profile. Returns the new fingerprint or None.""" try: print(colored("\n--- Add New SeedPass Profile ---", "yellow")) choice = input("Do you want to (1) Enter an existing 12-word seed or (2) Generate a new 12-word seed? (1/2): ").strip() new_fingerprint = None if choice == '1': new_fingerprint = self.setup_existing_seed() elif choice == '2': new_fingerprint = self.generate_new_seed() else: print(colored("Invalid choice.", 'red')) return None # Indicate failure if new_fingerprint: # Don't automatically select here, let select_or_add_fingerprint handle it print(colored(f"New profile with fingerprint '{new_fingerprint}' created.", 'green')) return new_fingerprint else: return None # Indicate failure except Exception as e: logger.error(f"Error adding new fingerprint: {e}", exc_info=True) print(colored(f"Error: Failed to add new profile: {e}", 'red')) return None def select_fingerprint(self, fingerprint: str) -> bool: """Sets the selected fingerprint as active and initializes all managers.""" if self.fingerprint_manager.select_fingerprint(fingerprint): self.current_fingerprint = fingerprint self.fingerprint_dir = self.fingerprint_manager.get_current_fingerprint_dir() if not self.fingerprint_dir: print(colored(f"Error: Fingerprint directory for {fingerprint} not found.", 'red')) return False # Indicate failure # Setup encryption requires password for the selected fingerprint password = prompt_existing_password(f"Enter master password for profile '{fingerprint}': ") if not self.setup_encryption_manager(self.fingerprint_dir, password): # setup_encryption_manager now handles verify_password internally print(colored("Password verification failed. Cannot switch profile.", "red")) # Reset state if needed self.current_fingerprint = None self.fingerprint_dir = None self.encryption_manager = None return False # Indicate failure # Define entry/backup dirs based on selected fingerprint self.entries_dir = self.fingerprint_dir / ENTRIES_DIR_NAME self.backups_dir = self.fingerprint_dir / BACKUPS_DIR_NAME self.entries_dir.mkdir(parents=True, exist_ok=True) # Ensure they exist self.backups_dir.mkdir(parents=True, exist_ok=True) # Load parent seed (requires encryption manager) if not self.load_parent_seed(self.fingerprint_dir): # Reset state self.current_fingerprint = None self.fingerprint_dir = None self.encryption_manager = None return False # Indicate failure # Initialize BIP85 (requires parent seed) if not self.initialize_bip85(): return False # Indicate failure # Initialize other managers (requires encryption_manager, dirs, bip85 etc.) if not self.initialize_managers(): return False # Indicate failure print(colored(f"Profile '{fingerprint}' selected and ready.", 'green')) return True else: print(colored(f"Error: Profile (fingerprint) '{fingerprint}' not found.", 'red')) return False def setup_encryption_manager(self, fingerprint_dir: Path, password: str) -> bool: """Sets up EncryptionManager and verifies password. Returns True on success.""" try: key = derive_key_from_password(password) self.encryption_manager = EncryptionManager(key, fingerprint_dir) logger.debug(f"EncryptionManager set up for {fingerprint_dir.name}.") # Verify password against stored hash if not self.verify_password(password): self.encryption_manager = None # Clear invalid manager return False # Indicate failure return True # Success 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')) self.encryption_manager = None return False def load_parent_seed(self, fingerprint_dir: Path) -> bool: """Loads and decrypts parent seed. Returns True on success.""" if not self.encryption_manager: logger.error("Cannot load parent seed: EncryptionManager not initialized.") return False try: self.parent_seed = self.encryption_manager.decrypt_parent_seed() logger.debug(f"Parent seed loaded for profile {self.current_fingerprint}.") return True except Exception as e: # Decrypt_parent_seed already logs and prints errors logger.error(f"Failed to load parent seed for {self.current_fingerprint}: {e}", exc_info=False) # Avoid redundant stack trace print(colored(f"Error: Could not load the parent seed for this profile.", 'red')) self.parent_seed = None return False def initialize_bip85(self) -> bool: """Initializes BIP85 generator. Returns True on success.""" if not self.parent_seed: logger.error("Cannot initialize BIP85: Parent seed not loaded.") return False try: seed_bytes = Bip39SeedGenerator(self.parent_seed).Generate() self.bip85 = BIP85(seed_bytes) logger.debug("BIP-85 initialized successfully.") return True except Exception as e: logger.error(f"Failed to initialize BIP-85: {e}", exc_info=True) print(colored(f"Error: Failed to initialize BIP-85: {e}", 'red')) self.bip85 = None return False def initialize_managers(self) -> bool: """Initializes EntryManager, PasswordGenerator, BackupManager, StateManager, NostrClient.""" # Check prerequisites if not all([self.encryption_manager, self.fingerprint_dir, self.entries_dir, self.backups_dir, self.parent_seed, self.bip85, self.current_fingerprint]): logger.error("Cannot initialize managers: Prerequisites missing.") return False try: # Initialize State Manager first self.state_manager = StateManager(self.fingerprint_dir) self.entry_manager = EntryManager( encryption_manager=self.encryption_manager, fingerprint_dir=self.fingerprint_dir # entries_dir passed via fingerprint_dir in its init ) self.password_generator = PasswordGenerator( encryption_manager=self.encryption_manager, # Needed for derive_seed_from_mnemonic parent_seed=self.parent_seed, bip85=self.bip85 ) self.backup_manager = BackupManager( fingerprint_dir=self.fingerprint_dir # backup_dir passed via fingerprint_dir in its init ) # Initialize NostrClient (ensure NostrClient init is updated) self.nostr_client = NostrClient( encryption_manager=self.encryption_manager, fingerprint=self.current_fingerprint, # Pass PasswordManager instance for callbacks if needed by EventHandler # password_manager_ref=self ) logger.debug(f"All managers initialized for profile {self.current_fingerprint}.") return True 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')) # Clean up partially initialized managers? self.state_manager = None self.entry_manager = None self.password_generator = None self.backup_manager = None self.nostr_client = None return False # --- Seed Setup Handlers (Modified) --- def handle_new_seed_setup(self) -> None: """Handles setup when no profiles exist.""" print(colored("Welcome to SeedPass! Let's create your first profile.", 'yellow')) new_fingerprint = self.add_new_fingerprint() # This handles generate/import choice if new_fingerprint: self.select_fingerprint(new_fingerprint) # Select and initialize else: print(colored("Failed to create initial profile. Exiting.", "red")) sys.exit(1) def setup_existing_seed(self) -> Optional[str]: """Handles importing an existing seed phrase.""" try: parent_seed = getpass.getpass(prompt='Enter your 12-word BIP-39 seed phrase: ').strip() if not self.validate_bip85_seed(parent_seed): print(colored("Error: Invalid 12-word seed phrase format.", 'red')) return None fingerprint = self.fingerprint_manager.add_fingerprint(parent_seed) if not fingerprint: print(colored("Error: Failed to add profile for the provided seed (maybe it already exists?).", 'red')) # FingerprintManager logs specific error return None # Could be duplicate or generation failure fingerprint_dir = self.fingerprint_manager.get_fingerprint_directory(fingerprint) if not fingerprint_dir: print(colored("Error: Failed to create profile directory.", 'red')) # Attempt cleanup? self.fingerprint_manager.remove_fingerprint(fingerprint) return None print(colored(f"Profile '{fingerprint}' created. Now set its master password.", 'green')) # Need to save the seed and password hash *for this new fingerprint* # Temporarily set context to save correctly temp_fp_dir = self.fingerprint_dir # Save old context if any self.fingerprint_dir = fingerprint_dir if not self.save_seed_and_password(parent_seed, fingerprint_dir): print(colored("Error saving seed or password. Rolling back profile creation.", "red")) self.fingerprint_manager.remove_fingerprint(fingerprint) # Cleanup self.fingerprint_dir = temp_fp_dir # Restore context return None self.fingerprint_dir = temp_fp_dir # Restore context return fingerprint except KeyboardInterrupt: print(colored("\nOperation cancelled by user.", 'yellow')) return None except Exception as e: logger.error(f"Error setting up existing seed: {e}", exc_info=True) print(colored(f"Error importing seed: {e}", 'red')) return None def generate_new_seed(self) -> Optional[str]: """Handles generating a new seed phrase.""" try: new_seed = self.generate_bip85_seed() print(colored("\n=== Your New 12-Word Master Seed Phrase ===", 'yellow', attrs=['bold'])) print(colored(new_seed, 'cyan')) print(colored("=============================================", 'yellow', attrs=['bold'])) print(colored("WRITE THIS DOWN NOW!", 'red', attrs=['blink'])) print(colored("Store it securely offline. Losing this means losing all derived passwords.", 'red')) print(colored("Do not store it digitally unless you understand the risks.", 'red')) if not confirm_action("\nHave you securely written down this seed phrase? (Y/N): "): print(colored("Seed generation cancelled. Please run again when ready.", 'yellow')) return None fingerprint = self.fingerprint_manager.add_fingerprint(new_seed) if not fingerprint: print(colored("Error: Failed to add profile for the new seed.", 'red')) return None fingerprint_dir = self.fingerprint_manager.get_fingerprint_directory(fingerprint) if not fingerprint_dir: print(colored("Error: Failed to create profile directory.", 'red')) self.fingerprint_manager.remove_fingerprint(fingerprint) return None print(colored(f"\nProfile '{fingerprint}' created. Now set its master password.", 'green')) # Temporarily set context to save correctly temp_fp_dir = self.fingerprint_dir # Save old context if any self.fingerprint_dir = fingerprint_dir if not self.save_seed_and_password(new_seed, fingerprint_dir): print(colored("Error saving seed or password. Rolling back profile creation.", "red")) self.fingerprint_manager.remove_fingerprint(fingerprint) # Cleanup self.fingerprint_dir = temp_fp_dir # Restore context return None self.fingerprint_dir = temp_fp_dir # Restore context return fingerprint except KeyboardInterrupt: print(colored("\nOperation cancelled by user.", 'yellow')) return None except Exception as e: logger.error(f"Error generating new seed: {e}", exc_info=True) print(colored(f"Error generating seed: {e}", 'red')) return None def save_seed_and_password(self, seed: str, fingerprint_dir: Path) -> bool: """Internal helper to prompt for password, save hash, and save encrypted seed.""" try: password = prompt_for_password() # Prompts for new + confirm # Derive key and setup temporary encryption manager for saving key = derive_key_from_password(password) temp_enc_mgr = EncryptionManager(key, fingerprint_dir) # Store hashed password within the target fingerprint dir if not self._store_hashed_password(password, fingerprint_dir): raise RuntimeError("Failed to store hashed password.") # Encrypt and save parent seed within the target fingerprint dir temp_enc_mgr.encrypt_parent_seed(seed) # encrypt_parent_seed handles saving to file logger.info(f"Seed and password hash saved successfully for profile {fingerprint_dir.name}.") return True except Exception as e: logger.error(f"Failed to encrypt/save seed or password hash for {fingerprint_dir.name}: {e}", exc_info=True) # Cleanup potentially created hash file? Difficult to do atomically here. return False # --- Core Entry Operations (NEW) --- def add_entry(self, kind: str, entry_data: Dict[str, Any]) -> Optional[int]: """ Adds a new entry of the specified kind, saves locally, and posts to Nostr. :param kind: The type of entry (must exist in KINDS). :param entry_data: The data payload for the entry. :return: The assigned entry number if successful, None otherwise. """ if not all([self.entry_manager, self.encryption_manager, self.state_manager, self.nostr_client, self.current_fingerprint]): logger.error("Cannot add entry: PasswordManager not fully initialized.") print(colored("Error: System not ready. Please restart.", "red")) return None kind_details = get_kind_details(kind) if not kind_details: logger.error(f"Attempted to add entry with unknown kind: {kind}") print(colored(f"Error: Unknown entry type '{kind}'.", "red")) return None # Add necessary metadata entry_num = self.entry_manager.get_next_entry_num() timestamp = datetime.utcnow().isoformat() + 'Z' checksum = self.entry_manager.calculate_checksum(entry_data) # Checksum of the *data* part # Handle bip85 index for generated passwords bip85_index = None if kind == "generated_password": # Check if bip85_index was passed in entry_data (e.g. during migration) if "bip85_index" not in entry_data: bip85_index = self.state_manager.get_next_generated_password_index() entry_data["bip85_index"] = bip85_index # Add it to the data part # Recalculate checksum if index was added checksum = self.entry_manager.calculate_checksum(entry_data) else: bip85_index = entry_data["bip85_index"] # Ensure state manager is updated if migrating an index higher than current max last_known_index = self.state_manager.get_last_generated_password_index() if bip85_index > last_known_index: self.state_manager.set_last_generated_password_index(bip85_index) # Encrypt sensitive fields within entry_data before creating the full entry JSON # Example: encrypt 'password' for stored_password, 'content' for note if kind == "stored_password" and "password" in entry_data: try: pwd_bytes = entry_data["password"].encode('utf-8') encrypted_pwd_bytes = self.encryption_manager.encrypt_data(pwd_bytes) entry_data["password"] = base64.b64encode(encrypted_pwd_bytes).decode('utf-8') # Store as base64 string checksum = self.entry_manager.calculate_checksum(entry_data) # Recalculate checksum except Exception as enc_err: logger.error(f"Failed to encrypt password for stored_password entry {entry_num}: {enc_err}", exc_info=True) print(colored("Error encrypting password data.", "red")) return None elif kind == "note" and "content" in entry_data: try: content_bytes = entry_data["content"].encode('utf-8') encrypted_content_bytes = self.encryption_manager.encrypt_data(content_bytes) entry_data["content"] = base64.b64encode(encrypted_content_bytes).decode('utf-8') # Store as base64 string checksum = self.entry_manager.calculate_checksum(entry_data) # Recalculate checksum except Exception as enc_err: logger.error(f"Failed to encrypt content for note entry {entry_num}: {enc_err}", exc_info=True) print(colored("Error encrypting note data.", "red")) return None # Construct the full entry structure (to be encrypted) full_entry = { "entry_num": entry_num, "fingerprint": self.current_fingerprint, "kind": kind, "data": entry_data, # Contains potentially pre-encrypted fields "timestamp": timestamp, # UTC timestamp of creation/last update "metadata": { "created_at": timestamp, # Keep original creation time separate? maybe not needed. "updated_at": timestamp, "checksum": checksum # Checksum of the 'data' part } } # Add bip85_index to top level for generated_password for easier access if needed # This is somewhat redundant but might be useful for retrieval/display logic. if kind == "generated_password": full_entry["bip85_index"] = bip85_index try: # Encrypt the entire entry structure entry_json = json.dumps(full_entry).encode('utf-8') encrypted_entry_data = self.encryption_manager.encrypt_data(entry_json) # Save the encrypted entry locally if not self.entry_manager.save_entry(entry_num, encrypted_entry_data): # EntryManager logs the error print(colored(f"Error: Failed to save entry {entry_num} locally.", 'red')) # Potential rollback needed? Difficult state. return None # Create a backup of the newly saved entry self.backup_manager.create_backup_for_entry(entry_num) # Post the encrypted entry to Nostr # Use a unique identifier ('d' tag) for replaceable events identifier = f"{kind_details['identifier_tag']}{entry_num}" nostr_kind_int = kind_details['nostr_kind'] self.nostr_client.publish_entry( encrypted_entry_data=encrypted_entry_data, # Already encrypted full entry nostr_kind=nostr_kind_int, d_tag=identifier ) logger.info(f"Entry {entry_num} (Kind: {kind}, ID: {identifier}) added locally and posted to Nostr.") print(colored(f"Entry {entry_num} added successfully.", 'green')) return entry_num except Exception as e: logger.error(f"Failed during final steps of adding entry {entry_num}: {e}", exc_info=True) print(colored(f"Error: Failed to complete adding entry {entry_num}: {e}", 'red')) # Attempt to clean up the saved file if posting failed? # self.entry_manager.delete_entry_file(entry_num) # Risky if Nostr post *did* succeed partially return None def modify_entry(self, entry_num: int, updated_data_fields: Dict[str, Any]) -> bool: """ Modifies an existing entry, saves locally, and posts update to Nostr. :param entry_num: The number of the entry to modify. :param updated_data_fields: Dictionary containing only the fields to update within the 'data' part. :return: True if successful, False otherwise. """ if not all([self.entry_manager, self.encryption_manager, self.nostr_client]): logger.error("Cannot modify entry: PasswordManager not fully initialized.") return False # Load existing entry existing_entry = self.entry_manager.load_entry(entry_num) if not existing_entry: print(colored(f"Error: Entry {entry_num} not found.", 'red')) return False kind = existing_entry.get("kind") kind_details = get_kind_details(kind) if not kind_details: logger.error(f"Cannot modify entry {entry_num}: Unknown kind '{kind}' found in loaded data.") print(colored(f"Error: Cannot modify entry {entry_num} due to corrupted kind.", 'red')) return False # Create backup before modifying self.backup_manager.create_backup_for_entry(entry_num) # Update the 'data' part original_data = existing_entry.get("data", {}) # Decrypt sensitive fields *before* updating if necessary # Example: Decrypt 'password' for stored_password, 'content' for note if kind == "stored_password" and "password" in original_data: try: pwd_b64 = original_data["password"] pwd_bytes = self.encryption_manager.decrypt_data(base64.b64decode(pwd_b64)) original_data["password"] = pwd_bytes.decode('utf-8') # Temporarily store decrypted for update logic except Exception as dec_err: logger.error(f"Failed to decrypt password for modification in entry {entry_num}: {dec_err}", exc_info=True) print(colored("Error preparing password field for modification.", "red")) return False elif kind == "note" and "content" in original_data: try: content_b64 = original_data["content"] content_bytes = self.encryption_manager.decrypt_data(base64.b64decode(content_b64)) original_data["content"] = content_bytes.decode('utf-8') # Temporarily store decrypted except Exception as dec_err: logger.error(f"Failed to decrypt content for modification in entry {entry_num}: {dec_err}", exc_info=True) print(colored("Error preparing note content for modification.", "red")) return False # Apply the updates from updated_data_fields original_data.update(updated_data_fields) # Re-encrypt sensitive fields *after* updating if kind == "stored_password" and "password" in original_data: try: pwd_bytes = original_data["password"].encode('utf-8') encrypted_pwd_bytes = self.encryption_manager.encrypt_data(pwd_bytes) original_data["password"] = base64.b64encode(encrypted_pwd_bytes).decode('utf-8') # Store as base64 string again except Exception as enc_err: logger.error(f"Failed to re-encrypt password for stored_password entry {entry_num}: {enc_err}", exc_info=True) print(colored("Error encrypting updated password data.", "red")) return False elif kind == "note" and "content" in original_data: try: content_bytes = original_data["content"].encode('utf-8') encrypted_content_bytes = self.encryption_manager.encrypt_data(content_bytes) original_data["content"] = base64.b64encode(encrypted_content_bytes).decode('utf-8') # Store as base64 string again except Exception as enc_err: logger.error(f"Failed to re-encrypt content for note entry {entry_num}: {enc_err}", exc_info=True) print(colored("Error encrypting updated note data.", "red")) return False # Update timestamp and recalculate checksum new_timestamp = datetime.utcnow().isoformat() + 'Z' new_checksum = self.entry_manager.calculate_checksum(original_data) # Update the full entry structure existing_entry["data"] = original_data # Put potentially re-encrypted data back existing_entry["timestamp"] = new_timestamp if "metadata" not in existing_entry: existing_entry["metadata"] = {} existing_entry["metadata"]["updated_at"] = new_timestamp existing_entry["metadata"]["checksum"] = new_checksum try: # Encrypt the updated full entry entry_json = json.dumps(existing_entry).encode('utf-8') encrypted_entry_data = self.encryption_manager.encrypt_data(entry_json) # Save locally if not self.entry_manager.save_entry(entry_num, encrypted_entry_data): print(colored(f"Error: Failed to save updated entry {entry_num} locally.", 'red')) return False # Post update to Nostr (as a replaceable event) identifier = f"{kind_details['identifier_tag']}{entry_num}" nostr_kind_int = kind_details['nostr_kind'] self.nostr_client.publish_entry( encrypted_entry_data=encrypted_entry_data, nostr_kind=nostr_kind_int, d_tag=identifier ) logger.info(f"Entry {entry_num} modified locally and update posted to Nostr.") print(colored(f"Entry {entry_num} updated successfully.", 'green')) return True except Exception as e: logger.error(f"Failed during final steps of modifying entry {entry_num}: {e}", exc_info=True) print(colored(f"Error: Failed to complete modifying entry {entry_num}: {e}", 'red')) # Consider attempting to restore the backup? return False def delete_entry(self, entry_num: int) -> bool: """Deletes an entry locally and posts a deletion marker to Nostr.""" if not all([self.entry_manager, self.nostr_client]): logger.error("Cannot delete entry: PasswordManager not fully initialized.") return False # Load entry to get kind details for Nostr deletion marker entry_data = self.entry_manager.load_entry(entry_num) if not entry_data: print(colored(f"Warning: Entry {entry_num} not found locally. Cannot delete.", 'yellow')) # Maybe still try to post deletion to Nostr? # For now, assume local file must exist. return False kind = entry_data.get("kind") kind_details = get_kind_details(kind) if not kind_details: logger.warning(f"Cannot determine kind for entry {entry_num} during deletion.") # Proceed with file deletion, but maybe skip Nostr? else: # Create backup before deleting self.backup_manager.create_backup_for_entry(entry_num) # Delete local file first if not self.entry_manager.delete_entry_file(entry_num): print(colored(f"Error: Failed to delete local file for entry {entry_num}.", 'red')) # Don't post deletion to Nostr if local delete failed return False # Post deletion marker to Nostr (e.g., Kind 5 event referencing the replaceable event) if kind_details: identifier = f"{kind_details['identifier_tag']}{entry_num}" nostr_kind_to_delete = kind_details['nostr_kind'] # We need the event ID of the event we want to delete if using Kind 5 # Fetching the event ID first might be complex/slow. # Alternative: Publish an empty content replaceable event? Easier. # Let's publish an empty content update for the replaceable event. # Note: Relays might prune empty events faster. Kind 5 is more explicit. # Decision: Publish empty content replaceable event for simplicity now. try: # Create a dummy entry structure with empty data for checksum empty_data_checksum = self.entry_manager.calculate_checksum({}) tombstone_entry = { "entry_num": entry_num, "fingerprint": self.current_fingerprint, "kind": kind, "data": {}, # Empty data "timestamp": datetime.utcnow().isoformat() + 'Z', "metadata": { "deleted": True, # Add deletion flag "updated_at": datetime.utcnow().isoformat() + 'Z', "checksum": empty_data_checksum } } entry_json = json.dumps(tombstone_entry).encode('utf-8') encrypted_tombstone_data = self.encryption_manager.encrypt_data(entry_json) self.nostr_client.publish_entry( encrypted_entry_data=encrypted_tombstone_data, nostr_kind=nostr_kind_to_delete, d_tag=identifier, is_deletion=True # Add flag for logging/handling in client ) logger.info(f"Deletion marker for entry {entry_num} (ID: {identifier}) posted to Nostr.") except Exception as e: logger.error(f"Failed to post deletion marker to Nostr for entry {entry_num}: {e}", exc_info=True) # Local file is already deleted. Log inconsistency. print(colored(f"Warning: Local entry {entry_num} deleted, but failed to post deletion to Nostr.", 'yellow')) # Still return True as local deletion succeeded? Or False due to incomplete operation? # Let's return True as the primary goal (local deletion) was met. print(colored(f"Entry {entry_num} deleted successfully.", 'green')) return True def list_all_entries(self) -> List[Dict[str, Any]]: """Loads all local entries and returns them as a list of dictionaries.""" if not self.entry_manager: return [] all_entries = [] entry_nums = self.entry_manager.list_all_entry_nums() for num in entry_nums: entry = self.entry_manager.load_entry(num) if entry: all_entries.append(entry) return all_entries def process_entry(self, entry: Dict[str, Any]): """ Processes an individual entry based on its kind using the registered handler. :param entry: The entry data dictionary (decrypted). """ if not self.encryption_manager or not self.password_generator: logger.error("Cannot process entry: Required managers not initialized.") return try: kind = entry.get('kind') data = entry.get('data', {}) fingerprint = entry.get('fingerprint') entry_num = entry.get('entry_num', 'N/A') handler = get_kind_handler(kind) if handler: # Pass necessary components to the handler via kwargs handler_kwargs = { "encryption_manager": self.encryption_manager, "password_generator": self.password_generator, # Add other managers if handlers need them } handler(data, fingerprint, **handler_kwargs) logger.debug(f"Processed entry {entry_num} of kind '{kind}'.") else: logger.warning(f"No handler found for kind '{kind}'. Skipping processing for entry {entry_num}.") print(colored(f"Warning: Cannot process entry {entry_num} - unknown type '{kind}'.", "yellow")) except Exception as e: logger.error(f"Failed to process entry {entry.get('entry_num', 'N/A')}: {e}", exc_info=True) print(colored(f"Error processing entry {entry.get('entry_num', 'N/A')}: {e}", 'red')) def synchronize_with_nostr(self): """Fetches entries from Nostr and updates local storage.""" if not self.nostr_client or not self.entry_manager or not self.encryption_manager or not self.state_manager: logger.error("Cannot synchronize: Required managers not initialized.") print(colored("Error: Cannot synchronize with Nostr - system not ready.", "red")) return print(colored("Synchronizing with Nostr... Please wait.", "yellow")) try: last_sync_time = self.state_manager.get_last_nostr_sync_time() # Fetch events since last sync # Modify fetch_all_entries_async in NostrClient to accept a 'since' timestamp # Use a reasonable limit initially, might need pagination for huge histories nostr_events = self.nostr_client.fetch_all_entries_sync(since=last_sync_time, limit=500) # Sync version if nostr_events is None: # Indicates an error during fetch print(colored("Synchronization failed: Could not retrieve data from Nostr.", "red")) return if not nostr_events: print(colored("No new entries found on Nostr since last sync.", "green")) # Still update sync time? Yes, confirms we checked. self.state_manager.set_last_nostr_sync_time(int(time.time())) return newest_event_time = last_sync_time processed_count = 0 updated_count = 0 new_count = 0 deleted_count = 0 error_count = 0 # Process newest events first for event in sorted(nostr_events, key=lambda e: e.created_at, reverse=True): if event.created_at > newest_event_time: newest_event_time = event.created_at try: encrypted_content_b64 = event.content encrypted_content_bytes = base64.b64decode(encrypted_content_b64) decrypted_content_bytes = self.encryption_manager.decrypt_data(encrypted_content_bytes) entry = json.loads(decrypted_content_bytes.decode('utf-8')) entry_num = entry.get('entry_num') remote_checksum = entry.get('metadata', {}).get('checksum') is_deleted = entry.get('metadata', {}).get('deleted', False) # Check deletion flag if entry_num is None or remote_checksum is None: logger.warning(f"Skipping invalid Nostr event (ID: {event.id}): Missing entry_num or checksum.") error_count += 1 continue local_entry_path = self.entry_manager._get_entry_path(entry_num) # Use internal helper if is_deleted: # Handle deletion marker if local_entry_path.exists(): print(colored(f"Processing deletion for entry {entry_num}...", "magenta")) # Optional: backup before deleting based on sync? Risky. # self.backup_manager.create_backup_for_entry(entry_num) if self.entry_manager.delete_entry_file(entry_num): deleted_count += 1 else: error_count += 1 # Failed local delete else: logger.debug(f"Received deletion marker for already deleted/non-existent entry {entry_num}.") continue # Don't process further if deleted # Compare with local version if local_entry_path.exists(): local_checksum = self.entry_manager.get_entry_checksum(entry_num) if local_checksum is None: # Error reading local checksum logger.warning(f"Could not read local checksum for entry {entry_num}. Skipping update check.") error_count += 1 continue if local_checksum != remote_checksum: # Remote is newer or different, update local print(colored(f"Updating entry {entry_num} from Nostr...", "yellow")) if self.entry_manager.save_entry(entry_num, encrypted_content_bytes): updated_count += 1 # Optional: process updated entry immediately? # self.process_entry(entry) else: error_count += 1 # Failed local save else: # Checksums match, no update needed logger.debug(f"Entry {entry_num} is already up-to-date.") else: # Entry exists on Nostr but not locally, save it print(colored(f"Downloading new entry {entry_num} from Nostr...", "green")) if self.entry_manager.save_entry(entry_num, encrypted_content_bytes): new_count += 1 # Optional: process new entry immediately? # self.process_entry(entry) else: error_count += 1 # Failed local save processed_count +=1 except (base64.binascii.Error, json.JSONDecodeError) as decode_err: logger.error(f"Failed to decode/decrypt Nostr event content (ID: {event.id}): {decode_err}") error_count += 1 except InvalidToken: # From decryption logger.error(f"Decryption failed for Nostr event content (ID: {event.id}). Invalid key or corrupt data?") error_count += 1 except Exception as proc_err: logger.error(f"Unexpected error processing Nostr event (ID: {event.id}): {proc_err}", exc_info=True) error_count += 1 # Update last sync time to the timestamp of the newest processed event # Add a small buffer (1 sec) to avoid missing events published exactly at sync time? if newest_event_time > last_sync_time: self.state_manager.set_last_nostr_sync_time(newest_event_time + 1) print(colored(f"Synchronization complete. New: {new_count}, Updated: {updated_count}, Deleted: {deleted_count}, Errors: {error_count}", "blue")) except Exception as e: logger.error(f"Failed to synchronize with Nostr: {e}", exc_info=True) print(colored(f"Error: Failed to synchronize with Nostr: {e}", 'red')) def migrate_data_if_needed(self): """Checks for the old index file and performs migration if found.""" if not self.fingerprint_dir: return # Should not happen if called correctly old_index_path = self.fingerprint_dir / OLD_INDEX_FILENAME if not old_index_path.exists(): logger.info("Old index file not found. Migration not required.") return print(colored(f"Old index file found for profile {self.current_fingerprint}. Migrating to new format...", "yellow")) # Backup the old index file before migration try: timestamp = int(time.time()) backup_old_index_path = self.backups_dir / f"{OLD_INDEX_FILENAME}.backup_{timestamp}" shutil.copy2(old_index_path, backup_old_index_path) logger.info(f"Backed up old index file to {backup_old_index_path}") except Exception as backup_err: logger.error(f"Failed to backup old index file before migration: {backup_err}", exc_info=True) print(colored("Error: Could not back up old data file. Migration aborted.", "red")) return try: # Load old data (uses EncryptionManager correctly) old_data = self.encryption_manager.load_json_data(old_index_path.relative_to(self.fingerprint_dir)) old_passwords = old_data.get('passwords', {}) if not old_passwords: print(colored("Old index file is empty or invalid. No entries to migrate.", "yellow")) # Optionally delete the empty/invalid old file? # old_index_path.unlink() return migrated_count = 0 error_count = 0 print(colored(f"Found {len(old_passwords)} entries in old format. Starting migration...", "cyan")) # Iterate through old entries and use add_entry logic # Note: old index was string, new entry_num is int for old_idx_str, old_entry_data in old_passwords.items(): try: old_idx = int(old_idx_str) # Map old fields to new 'generated_password' kind structure new_entry_data = { "title": old_entry_data.get('website', f"Migrated Entry {old_idx}"), "username": old_entry_data.get('username', ''), "email": "", # Old format didn't have email "url": old_entry_data.get('url', ''), "length": old_entry_data.get('length'), "bip85_index": old_idx # Use the old index as the bip85_index # Blacklisted status? Decide how to handle. Maybe add to notes? } # Validate required fields for generated_password if new_entry_data["length"] is None: logger.warning(f"Skipping migration for old index {old_idx}: Missing 'length'. Data: {old_entry_data}") error_count += 1 continue # Use the add_entry method which handles saving and posting to nostr result_entry_num = self.add_entry(kind="generated_password", entry_data=new_entry_data) if result_entry_num is not None: migrated_count += 1 print(f" Migrated old index {old_idx} -> new entry {result_entry_num}") else: error_count += 1 print(colored(f" Failed to migrate old index {old_idx}", "red")) # Should we stop migration on first error? Or continue? Let's continue. except ValueError: logger.warning(f"Skipping migration for invalid old index key: {old_idx_str}") error_count += 1 continue except Exception as migrate_entry_err: logger.error(f"Error migrating old index {old_idx_str}: {migrate_entry_err}", exc_info=True) error_count += 1 print(colored(f" Error migrating old index {old_idx_str}", "red")) print(colored(f"Migration finished. Migrated: {migrated_count}, Errors: {error_count}", "blue")) if error_count == 0: # Optionally delete the old index file after successful migration if confirm_action("Migration successful. Delete the old index file? (Y/N): "): try: with lock_file(old_index_path, fcntl.LOCK_EX): old_index_path.unlink() print(colored("Old index file deleted.", "green")) except Exception as del_err: logger.error(f"Failed to delete old index file {old_index_path}: {del_err}", exc_info=True) print(colored("Error: Failed to delete old index file.", "red")) else: print(colored("Migration completed with errors. Please review logs.", "yellow")) print(colored("The old index file has NOT been deleted.", "yellow")) except Exception as e: logger.error(f"Critical error during data migration: {e}", exc_info=True) print(colored(f"Error: Failed to migrate data: {e}. Old data remains.", 'red')) # --- Utility Methods (Password Hashing, Seed Validation, etc.) --- def validate_bip85_seed(self, seed: str) -> bool: """Validates the provided BIP-39 seed phrase (12 words).""" try: words = seed.split() if len(words) == 12: # Basic check # Add bip_utils validation? Bip39MnemonicValidator(seed).IsValid() - needs wordlist return True return False except Exception: return False def generate_bip85_seed(self) -> str: """Generates a new 12-word BIP-39 seed phrase.""" try: # Generate entropy suitable for a 12-word mnemonic (128 bits / 16 bytes) entropy = os.urandom(16) mnemonic = Bip39MnemonicGenerator(Bip39Languages.ENGLISH).FromEntropy(entropy) return mnemonic.ToStr() except Exception as e: logger.error(f"Failed to generate BIP-39 seed: {e}", exc_info=True) print(colored(f"Error: Failed to generate seed: {e}", 'red')) sys.exit(1) def verify_password(self, password: str) -> bool: """Verifies provided password against the stored hash for the current fingerprint.""" if not self.fingerprint_dir: logger.error("Cannot verify password, fingerprint directory not set.") return False hashed_password_file = self.fingerprint_dir / HASHED_PASSWORD_FILENAME if not hashed_password_file.exists(): logger.error(f"Hashed password file not found: {hashed_password_file}") print(colored("Error: Password hash file missing for this profile.", 'red')) return False try: with lock_file(hashed_password_file, fcntl.LOCK_SH): with open(hashed_password_file, 'rb') as f: stored_hash = f.read() # Normalize entered password before checking normalized_password = unicodedata.normalize('NFKD', password).strip() is_correct = bcrypt.checkpw(normalized_password.encode('utf-8'), stored_hash) if is_correct: logger.debug("Password verification successful.") else: logger.warning("Password verification failed.") return is_correct except ValueError as e: # Handle potential bcrypt errors like "invalid salt" logger.error(f"Error during password check (likely invalid hash file): {e}") print(colored("Error: Problem verifying password - hash file might be corrupt.", 'red')) return False except Exception as e: logger.error(f"Error verifying password: {e}", exc_info=True) print(colored(f"Error: Failed to verify password: {e}", 'red')) return False def _store_hashed_password(self, password: str, fingerprint_dir: Path) -> bool: """Hashes and stores password for a specific fingerprint directory.""" hashed_password_file = fingerprint_dir / HASHED_PASSWORD_FILENAME try: # Normalize password before hashing normalized_password = unicodedata.normalize('NFKD', password).strip() hashed = bcrypt.hashpw(normalized_password.encode('utf-8'), bcrypt.gensalt()) with lock_file(hashed_password_file, fcntl.LOCK_EX): with open(hashed_password_file, 'wb') as f: f.write(hashed) os.chmod(hashed_password_file, 0o600) logger.info(f"Password hash stored for profile {fingerprint_dir.name}.") return True except Exception as e: logger.error(f"Failed to store hashed password for {fingerprint_dir.name}: {e}", exc_info=True) print(colored(f"Error: Failed to store password hash: {e}", 'red')) return False # --- CLI Handler Methods (Adapting old ones) --- def handle_add_entry_cli(self) -> None: """Handles the CLI interaction for adding a new entry.""" print(colored("\n--- Add New Entry ---", "yellow")) available_kinds = get_all_kinds() print("Available entry types:") for i, kind_name in enumerate(available_kinds): details = get_kind_details(kind_name) print(f" {i+1}. {kind_name} ({details['description']})") while True: try: choice_str = input("Select entry type number: ").strip() choice = int(choice_str) - 1 if 0 <= choice < len(available_kinds): selected_kind = available_kinds[choice] break else: print(colored("Invalid selection.", "red")) except ValueError: print(colored("Invalid input. Please enter a number.", "red")) print(colored(f"\nAdding new '{selected_kind}' entry...", "cyan")) entry_data = {} required_fields = get_required_fields(selected_kind) # Special handling for generated_password length/index (not prompted here) if selected_kind == "generated_password": try: entry_data["title"] = input("Enter Title/Website Name: ").strip() if not entry_data["title"]: print(colored("Title cannot be empty.", "red")) return entry_data["username"] = input("Enter Username (optional): ").strip() entry_data["email"] = input("Enter Email (optional): ").strip() entry_data["url"] = input("Enter URL (optional): ").strip() length_input = input(f'Enter desired password length (default {DEFAULT_PASSWORD_LENGTH}): ').strip() length = DEFAULT_PASSWORD_LENGTH if length_input: length = int(length_input) # Add validation if not (MIN_PASSWORD_LENGTH <= length <= MAX_PASSWORD_LENGTH): print(colored(f"Error: Password length must be between {MIN_PASSWORD_LENGTH} and {MAX_PASSWORD_LENGTH}.", 'red')) return entry_data["length"] = length # bip85_index is added automatically by add_entry method except ValueError: print(colored("Invalid length input.", "red")) return # Generic prompt for other kinds else: for field in required_fields: # Skip password field for stored_password - handle specially if selected_kind == "stored_password" and field == "password": entry_data[field] = getpass.getpass(f"Enter Password for '{entry_data.get('title', 'entry')}': ").strip() # Add confirmation? if not entry_data[field]: print(colored("Password cannot be empty.", "red")) return continue # Skip content field for note - handle specially? Maybe allow multiline? if selected_kind == "note" and field == "content": print(f"Enter {field.capitalize()} (end with 'EOF' on a new line):") lines = [] while True: line = input() if line == "EOF": break lines.append(line) entry_data[field] = "\n".join(lines) continue # Standard prompt prompt_text = f"Enter {field.replace('_', ' ').capitalize()}" if field == "tags" and selected_kind == "note": prompt_text += " (comma-separated)" user_input = input(f"{prompt_text}: ").strip() if field == "tags" and selected_kind == "note": entry_data[field] = [tag.strip() for tag in user_input.split(',') if tag.strip()] else: # Add validation based on field type if needed later entry_data[field] = user_input # Add the entry using the main logic self.add_entry(selected_kind, entry_data) def handle_retrieve_entry_cli(self) -> None: """Handles the CLI interaction for retrieving/displaying an entry.""" print(colored("\n--- Retrieve Entry ---", "yellow")) all_entries = self.list_all_entries() if not all_entries: print(colored("No entries found locally.", "yellow")) return print("Available Entries:") # Sort by entry_num for consistent display for entry in sorted(all_entries, key=lambda x: x.get("entry_num", -1)): num = entry.get("entry_num", "N/A") kind = entry.get("kind", "Unknown") title = entry.get("data", {}).get("title", "No Title") timestamp = entry.get("timestamp", "No Date") print(f" {Style.BRIGHT}{num}{Style.RESET_ALL}. {title} ({kind}) - Last Updated: {timestamp}") while True: try: choice_str = input("Enter entry number to display: ").strip() entry_num_to_display = int(choice_str) # Find the selected entry selected_entry = next((e for e in all_entries if e.get("entry_num") == entry_num_to_display), None) if selected_entry: print(colored(f"\nDisplaying Entry {entry_num_to_display}:", "blue")) self.process_entry(selected_entry) # Use the handler logic break else: print(colored("Invalid entry number.", "red")) except ValueError: print(colored("Invalid input. Please enter a number.", "red")) except KeyboardInterrupt: print(colored("\nCancelled.", "yellow")) break def handle_modify_entry_cli(self) -> None: """Handles the CLI interaction for modifying an entry.""" print(colored("\n--- Modify Entry ---", "yellow")) all_entries = self.list_all_entries() if not all_entries: print(colored("No entries found locally to modify.", "yellow")) return print("Available Entries:") for entry in sorted(all_entries, key=lambda x: x.get("entry_num", -1)): num = entry.get("entry_num", "N/A") kind = entry.get("kind", "Unknown") title = entry.get("data", {}).get("title", "No Title") print(f" {Style.BRIGHT}{num}{Style.RESET_ALL}. {title} ({kind})") while True: try: choice_str = input("Enter entry number to modify: ").strip() entry_num_to_modify = int(choice_str) existing_entry = next((e for e in all_entries if e.get("entry_num") == entry_num_to_modify), None) if existing_entry: break else: print(colored("Invalid entry number.", "red")) except ValueError: print(colored("Invalid input. Please enter a number.", "red")) except KeyboardInterrupt: print(colored("\nCancelled.", "yellow")) return # Exit modify handler kind = existing_entry.get("kind") current_data = existing_entry.get("data", {}) print(colored(f"\nModifying Entry {entry_num_to_modify} (Kind: {kind}, Title: {current_data.get('title', 'N/A')})", "cyan")) # Decrypt sensitive fields for display/editing if needed display_data = current_data.copy() # Work on a copy for display/prompting if kind == "stored_password" and "password" in display_data: try: pwd_b64 = display_data["password"] pwd_bytes = self.encryption_manager.decrypt_data(base64.b64decode(pwd_b64)) display_data["password"] = pwd_bytes.decode('utf-8') except Exception: display_data["password"] = "*** Error Decrypting ***" elif kind == "note" and "content" in display_data: try: content_b64 = display_data["content"] content_bytes = self.encryption_manager.decrypt_data(base64.b64decode(content_b64)) display_data["content"] = content_bytes.decode('utf-8') except Exception: display_data["content"] = "*** Error Decrypting ***" updated_data_fields = {} fields_to_modify = get_required_fields(kind) # Cannot modify bip85_index or length for generated_password if kind == "generated_password": fields_to_modify = [f for f in fields_to_modify if f not in ["length", "bip85_index"]] for field in fields_to_modify: current_value = display_data.get(field, "") # Handle special display/prompt for password/content if field == "password" and kind == "stored_password": print(f"Current Password: {'*' * len(current_value) if current_value else 'Not Set'}") new_value = getpass.getpass(f"Enter new Password (leave blank to keep current): ").strip() elif field == "content" and kind == "note": print(f"Current Content:\n---\n{current_value}\n---") print(f"Enter new {field.capitalize()} (leave blank to keep, end with 'EOF' on a new line):") lines = [] while True: line = input() if line == "EOF": break lines.append(line) new_value = "\n".join(lines) if lines else "" # Empty string if no input elif field == "tags" and kind == "note": print(f"Current Tags: {', '.join(current_value) if current_value else 'None'}") new_value = input(f"Enter new Tags (comma-separated, leave blank to keep): ").strip() else: print(f"Current {field.replace('_',' ').capitalize()}: {current_value}") new_value = input(f"Enter new {field.replace('_',' ').capitalize()} (leave blank to keep): ").strip() if new_value: # Only add to update dict if user provided input if field == "tags" and kind == "note": updated_data_fields[field] = [tag.strip() for tag in new_value.split(',') if tag.strip()] else: updated_data_fields[field] = new_value if not updated_data_fields: print(colored("No changes entered.", "yellow")) return # Confirm changes before applying print("\nChanges to be applied:") for field, value in updated_data_fields.items(): print(f" {field}: {value[:50] + '...' if len(value)>50 else value}") # Truncate long values if confirm_action("Proceed with these modifications? (Y/N): "): self.modify_entry(entry_num_to_modify, updated_data_fields) else: print(colored("Modification cancelled.", "yellow")) def handle_delete_entry_cli(self) -> None: """Handles the CLI interaction for deleting an entry.""" print(colored("\n--- Delete Entry ---", "yellow")) all_entries = self.list_all_entries() if not all_entries: print(colored("No entries found locally to delete.", "yellow")) return print("Available Entries:") for entry in sorted(all_entries, key=lambda x: x.get("entry_num", -1)): num = entry.get("entry_num", "N/A") kind = entry.get("kind", "Unknown") title = entry.get("data", {}).get("title", "No Title") print(f" {Style.BRIGHT}{num}{Style.RESET_ALL}. {title} ({kind})") while True: try: choice_str = input("Enter entry number to DELETE: ").strip() entry_num_to_delete = int(choice_str) # Verify entry exists before confirming existing_entry = next((e for e in all_entries if e.get("entry_num") == entry_num_to_delete), None) if existing_entry: title_to_delete = existing_entry.get("data", {}).get("title", "No Title") break else: print(colored("Invalid entry number.", "red")) except ValueError: print(colored("Invalid input. Please enter a number.", "red")) except KeyboardInterrupt: print(colored("\nCancelled.", "yellow")) return if confirm_action(colored(f"Are you SURE you want to delete entry {entry_num_to_delete} ('{title_to_delete}')?\nThis is IRREVERSIBLE locally and will post a deletion marker to Nostr. (Y/N): ", "red", attrs=["bold"])): self.delete_entry(entry_num_to_delete) else: print(colored("Deletion cancelled.", "yellow")) def handle_backup_entry_cli(self) -> None: """Handles CLI for backing up a specific entry.""" print(colored("\n--- Backup Entry ---", "yellow")) all_entries = self.list_all_entries() if not all_entries: print(colored("No entries found locally to back up.", "yellow")) return print("Available Entries:") for entry in sorted(all_entries, key=lambda x: x.get("entry_num", -1)): num = entry.get("entry_num", "N/A") kind = entry.get("kind", "Unknown") title = entry.get("data", {}).get("title", "No Title") print(f" {Style.BRIGHT}{num}{Style.RESET_ALL}. {title} ({kind})") while True: try: choice_str = input("Enter entry number to backup: ").strip() entry_num_to_backup = int(choice_str) if any(e.get("entry_num") == entry_num_to_backup for e in all_entries): self.backup_manager.create_backup_for_entry(entry_num_to_backup) break else: print(colored("Invalid entry number.", "red")) except ValueError: print(colored("Invalid input. Please enter a number.", "red")) except KeyboardInterrupt: print(colored("\nCancelled.", "yellow")) break def handle_restore_entry_cli(self) -> None: """Handles CLI for restoring an entry from backup.""" print(colored("\n--- Restore Entry from Backup ---", "yellow")) all_entries = self.list_all_entries() if not all_entries: print(colored("No entries exist. Cannot restore.", "yellow")) # Or maybe allow restoring to create? For now, require existing entry number. # If allowing restore-to-create, need to list all backups first. return print("Select entry number to restore:") for entry in sorted(all_entries, key=lambda x: x.get("entry_num", -1)): num = entry.get("entry_num", "N/A") kind = entry.get("kind", "Unknown") title = entry.get("data", {}).get("title", "No Title") print(f" {Style.BRIGHT}{num}{Style.RESET_ALL}. {title} ({kind})") entry_num_to_restore = None while entry_num_to_restore is None: try: choice_str = input("Enter entry number: ").strip() num = int(choice_str) if any(e.get("entry_num") == num for e in all_entries): entry_num_to_restore = num else: print(colored("Invalid entry number.", "red")) except ValueError: print(colored("Invalid input. Please enter a number.", "red")) except KeyboardInterrupt: print(colored("\nCancelled.", "yellow")) return # List backups for the selected entry backups = self.backup_manager.list_backups_for_entry(entry_num_to_restore) if not backups: print(colored(f"No backups found for entry {entry_num_to_restore}.", "yellow")) return print(colored(f"\nAvailable Backups for Entry {entry_num_to_restore}:", "cyan")) for i, backup_path in enumerate(backups): try: creation_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(backup_path.stat().st_mtime)) print(colored(f" {i+1}. {backup_path.name} ({creation_time})", "cyan")) except Exception: print(colored(f" {i+1}. {backup_path.name} (Error reading time)", "red")) while True: try: choice_str = input("Select backup number to restore: ").strip() choice = int(choice_str) - 1 if 0 <= choice < len(backups): selected_backup_path = backups[choice] if confirm_action(f"Restore entry {entry_num_to_restore} from {selected_backup_path.name}? This will overwrite the current entry. (Y/N): "): self.backup_manager.restore_entry_from_backup(entry_num_to_restore, selected_backup_path.name) # Ask user if they want to post the restored version to Nostr? if confirm_action("Do you want to post this restored version to Nostr (overwriting any newer version there)? (Y/N):"): restored_entry = self.entry_manager.load_entry(entry_num_to_restore) if restored_entry: kind_details = get_kind_details(restored_entry.get("kind")) if kind_details: entry_json = json.dumps(restored_entry).encode('utf-8') encrypted_entry_data = self.encryption_manager.encrypt_data(entry_json) identifier = f"{kind_details['identifier_tag']}{entry_num_to_restore}" nostr_kind_int = kind_details['nostr_kind'] self.nostr_client.publish_entry( encrypted_entry_data=encrypted_entry_data, nostr_kind=nostr_kind_int, d_tag=identifier ) print(colored("Restored entry posted to Nostr.", "green")) else: print(colored("Could not post to Nostr: Unknown kind.", "red")) else: print(colored("Could not post to Nostr: Failed to reload restored entry.", "red")) else: print(colored("Restore cancelled.", "yellow")) break # Exit loop else: print(colored("Invalid selection.", "red")) except ValueError: print(colored("Invalid input. Please enter a number.", "red")) except KeyboardInterrupt: print(colored("\nCancelled.", "yellow")) break # Exit loop def handle_verify_checksum(self) -> None: """Verifies main script checksum.""" # This remains unchanged as it checks the script file itself try: # Assuming __main__.__file__ gives the path to main.py when run script_path = os.path.abspath(sys.modules['__main__'].__file__) current_checksum = calculate_script_checksum(script_path) if verify_script_checksum(current_checksum, str(SCRIPT_CHECKSUM_FILE)): # Convert Path to str print(colored("Script checksum verification passed.", 'green')) logging.info("Script checksum verification passed.") else: print(colored("Checksum verification failed. The main script may have been modified.", 'red')) logging.error("Script checksum verification failed.") except Exception as e: logging.error(f"Error during script checksum verification: {e}", exc_info=True) print(colored(f"Error: Failed to verify script checksum: {e}", 'red')) def handle_backup_reveal_parent_seed(self) -> None: """Handles backup/reveal of the parent seed (remains largely unchanged).""" if not self.parent_seed or not self.fingerprint_dir or not self.encryption_manager: print(colored("Error: Profile not fully loaded.", "red")) return try: print(colored("\n=== Backup/Reveal Parent Seed ===", 'yellow')) print(colored("Warning: Revealing your parent seed is a highly sensitive operation.", 'red')) print(colored("Ensure you're in a secure, private environment.", 'red')) password = prompt_existing_password("Enter your master password to continue: ") if not self.verify_password(password): print(colored("Incorrect password. Operation aborted.", 'red')) return if not confirm_action("Are you absolutely SURE you want to reveal your parent seed? (Y/N): "): print(colored("Operation cancelled by user.", 'yellow')) return print(colored("\n=== Your 12-Word BIP-39 Parent Seed ===", 'green', attrs=['bold'])) print(colored(self.parent_seed, 'yellow')) print(colored("\nWRITE THIS DOWN if you haven't. Store it securely offline.", 'red')) if confirm_action("Do you want to save this seed to a separate encrypted backup file? (Y/N): "): default_name = f"seedpass_seed_{self.current_fingerprint}_backup.enc" filename = input(f"Enter filename (default: {default_name}): ").strip() or default_name # Basic filename validation (avoids path traversal) if '/' in filename or '\\' in filename or '..' in filename: print(colored("Invalid filename.", "red")) return backup_path = self.fingerprint_dir / filename # Use encrypt_and_save_file which handles locking etc. self.encryption_manager.encrypt_and_save_file(self.parent_seed.encode('utf-8'), backup_path.relative_to(self.fingerprint_dir)) print(colored(f"Encrypted seed backup saved to '{backup_path}'. Keep this file safe!", 'green')) except Exception as e: logger.error(f"Error during parent seed backup/reveal: {e}", exc_info=True) print(colored(f"Error: Failed during seed backup/reveal: {e}", 'red')) # --- Fingerprint Management Handlers (No change needed here) --- def handle_switch_fingerprint(self) -> bool: """Handles switching active profile.""" print(colored("\n--- Switch SeedPass Profile ---", "yellow")) # Get current selection before listing current_fp = self.current_fingerprint fingerprints = self.fingerprint_manager.list_fingerprints() if not fingerprints or len(fingerprints) <= 1: print(colored("No other profiles available to switch to.", "yellow")) return False print("Available Profiles:") available_to_switch = [] display_idx = 1 for fp in fingerprints: if fp != current_fp: print(colored(f"{display_idx}. {fp}", 'cyan')) available_to_switch.append(fp) display_idx += 1 else: print(colored(f" {fp} (Current)", "grey")) if not available_to_switch: print(colored("No other profiles available to switch to.", "yellow")) return False while True: choice_str = input("Select profile number to switch to (or 'c' to cancel): ").strip().lower() if choice_str == 'c': print(colored("Switch cancelled.", "yellow")) return False if not choice_str.isdigit(): print(colored("Invalid input.", "red")) continue choice = int(choice_str) if 1 <= choice <= len(available_to_switch): selected_fingerprint = available_to_switch[choice - 1] # select_fingerprint handles password prompt and manager re-init return self.select_fingerprint(selected_fingerprint) else: print(colored("Invalid selection.", 'red')) # Other fingerprint handlers (add_new_fingerprint_cli, remove_fingerprint_cli, list_fingerprints_cli) # would call the underlying FingerprintManager methods, similar to the existing structure in main.py, # but should be methods within PasswordManager for better encapsulation. def handle_add_new_fingerprint_cli(self): return self.add_new_fingerprint() # Calls the internal method def handle_remove_fingerprint_cli(self): print(colored("\n--- Remove SeedPass Profile ---", "yellow", attrs=['bold'])) print(colored("WARNING: This will delete the profile's fingerprint, encrypted seed,", attrs=['bold']), colored("all associated entries, and backups locally.", "red", attrs=['bold'])) print(colored("This action is IRREVERSIBLE.", "red", attrs=['bold'])) fingerprints = self.fingerprint_manager.list_fingerprints() if not fingerprints: print(colored("No profiles available to remove.", 'yellow')) return print("Available Profiles:") current_fp = self.current_fingerprint removable_fps = [] display_idx = 1 for fp in fingerprints: is_current = "(Current)" if fp == current_fp else "" print(colored(f"{display_idx}. {fp} {is_current}", 'cyan' if fp != current_fp else 'grey')) removable_fps.append(fp) display_idx += 1 while True: choice_str = input("Enter profile number to remove (or 'c' to cancel): ").strip().lower() if choice_str == 'c': print(colored("Removal cancelled.", "yellow")) return if not choice_str.isdigit(): print(colored("Invalid input.", "red")) continue choice = int(choice_str) if 1 <= choice <= len(removable_fps): selected_fingerprint = removable_fps[choice - 1] if selected_fingerprint == self.current_fingerprint: print(colored("Cannot remove the currently active profile. Switch profiles first.", "red")) return if confirm_action(colored(f"REALLY remove profile '{selected_fingerprint}' and all its data? (Y/N): ", "red")): if self.fingerprint_manager.remove_fingerprint(selected_fingerprint): print(colored(f"Profile {selected_fingerprint} removed successfully.", 'green')) else: print(colored("Failed to remove profile.", 'red')) else: print(colored("Removal cancelled.", 'yellow')) return # Exit after attempt or cancel else: print(colored("Invalid selection.", 'red')) def handle_list_fingerprints_cli(self): print(colored("\n--- SeedPass Profiles (Fingerprints) ---", "yellow")) fingerprints = self.fingerprint_manager.list_fingerprints() if not fingerprints: print(colored("No profiles configured.", 'yellow')) return current_fp = self.current_fingerprint for fp in fingerprints: is_current = colored("(Current)", "green") if fp == current_fp else "" print(colored(f"- {fp} {is_current}", 'cyan')) # --- Old/Removed Methods --- # Remove handle_generate_password, handle_retrieve_password, handle_modify_entry (old index versions) # Remove get_encrypted_data, decrypt_and_save_index_from_nostr (old index versions) # Remove backup_database, restore_database (old index versions)
Phase 4: Refactor NostrClient
nostr/client.py(Refactored):# nostr/client.py import os import sys import logging import traceback import json import time import base64 import hashlib import asyncio import concurrent.futures from typing import List, Optional, Callable, Dict, Any from pathlib import Path from monstr.client.client import ClientPool, Client from monstr.encrypt import Keys # Keep Keys # Remove NIP4Encrypt unless needed for direct DMs (not needed for current backup plan) # from monstr.encrypt import NIP4Encrypt from monstr.event.event import Event from monstr.event.event_handlers import StoreEventHandler # Useful for collecting events from monstr.util import util_funcs # For relay set conversion import threading import uuid import fcntl # Import necessary components from SeedPass structure from password_manager.encryption import EncryptionManager # Used in init from .key_manager import KeyManager # EventHandler is now different - handles processing entries # from .event_handler import EventHandler # Remove old event handler import from constants import APP_DIR # Keep if needed, but paths managed by PasswordManager now from utils.file_lock import lock_file # Keep if needed logger = logging.getLogger(__name__) # Set the logging level specific to this module if desired # logger.setLevel(logging.DEBUG) # Example: More verbose Nostr logs DEFAULT_RELAYS = [ "wss://relay.snort.social", "wss://nostr.oxtr.dev", "wss://relay.primal.net", "wss://relay.damus.io", "wss://nostr.wine" ] # Define the Nostr Kind for SeedPass entries SEEDPASS_NOSTR_KIND = 31111 # Replaceable event kind for entries class NostrClient: """ Handles interactions with the Nostr network for SeedPass entries. Uses replaceable events (Kind 31111) with 'd' tags for synchronization. """ def __init__(self, encryption_manager: EncryptionManager, fingerprint: str, relays: Optional[List[str]] = None): """ Initializes the NostrClient. :param encryption_manager: Instance for decrypting the parent seed. :param fingerprint: The active fingerprint for deriving Nostr keys. :param relays: Optional list of relay URLs. """ self.encryption_manager = encryption_manager self.fingerprint = fingerprint # Derive keys *immediately* upon init try: self.key_manager = KeyManager( self.encryption_manager.decrypt_parent_seed(), # Decrypt seed here self.fingerprint ) except Exception as key_err: logger.critical(f"Failed to derive Nostr keys for fingerprint {fingerprint}: {key_err}", exc_info=True) print(colored(f"Error: Could not initialize Nostr identity for profile {fingerprint}.", "red")) raise RuntimeError("Nostr key generation failed") from key_err # Use default or provided relays self.relays = relays if relays else DEFAULT_RELAYS # Convert relay list to set for ClientPool if needed by monstr version relay_set = util_funcs.str_filter_to_set(self.relays) if not relay_set: logger.warning("No valid relays configured for NostrClient.") relay_set = {"wss://relay.damus.io"} # Fallback? Or raise error? self.client_pool = ClientPool(list(relay_set)) # ClientPool might expect list self.subscriptions: Dict[str, Any] = {} # Track subscriptions # For async operations from sync methods self.loop = asyncio.new_event_loop() self.loop_thread = threading.Thread(target=self._run_event_loop, daemon=True) self.loop_thread.start() # Wait for initial connection self.wait_for_connection() logger.info(f"NostrClient initialized for fingerprint {fingerprint} (PubKey: {self.key_manager.get_public_key_hex()[:10]}...).") # Shutdown flag self.is_shutting_down = False def _run_event_loop(self): """Runs the asyncio event loop in a separate thread.""" asyncio.set_event_loop(self.loop) try: self.loop.run_forever() finally: # Clean up loop resources before thread exits tasks = asyncio.all_tasks(loop=self.loop) for task in tasks: task.cancel() # Run loop briefly to allow tasks to finish cancelling self.loop.run_until_complete(asyncio.sleep(0.1)) self.loop.close() logger.info("NostrClient event loop closed.") def wait_for_connection(self, timeout=10): """Waits for the client pool to connect to at least one relay.""" start_time = time.time() while not self.client_pool.connected: if time.time() - start_time > timeout: logger.warning(f"NostrClient connection timeout after {timeout}s.") print(colored("Warning: Could not connect to Nostr relays within timeout.", "yellow")) # Decide if this is fatal or not. Maybe allow offline operation? # For now, let it proceed but log warning. break time.sleep(0.2) if self.client_pool.connected: logger.debug("NostrClient connected to relays.") async def publish_entry_async(self, encrypted_entry_data: bytes, nostr_kind: int, d_tag: str, is_deletion: bool = False): """ Asynchronously publishes an entry as a replaceable event. :param encrypted_entry_data: The fully encrypted entry JSON as bytes. :param nostr_kind: The Nostr event kind (e.g., 31111). :param d_tag: The unique identifier for the 'd' tag (e.g., "seedpass_gp_123"). :param is_deletion: If True, content might be empty/special marker (though we encrypt empty dict currently). """ try: content_b64 = base64.b64encode(encrypted_entry_data).decode('utf-8') # Create replaceable event event = Event( kind=nostr_kind, content=content_b64, pub_key=self.key_manager.get_public_key_hex(), tags=[ ["d", d_tag], ["t", "seedpass"] # General tag for SeedPass entries # Add ["k", str(nostr_kind)] ? Maybe redundant. ] ) # created_at will be set automatically by monstr on sign if not present event.sign(self.key_manager.get_private_key_hex()) logger.debug(f"Prepared Nostr Event (Kind: {nostr_kind}, d: {d_tag}, ID: {event.id})") # Publish using the client pool self.client_pool.publish(event) logger.info(f"Published entry {'(Deletion Marker)' if is_deletion else ''} to Nostr (Kind: {nostr_kind}, d: {d_tag}, EventID: {event.id})") except Exception as e: logger.error(f"Failed to publish Nostr event (Kind: {nostr_kind}, d: {d_tag}): {e}", exc_info=True) # Should this raise or just log? Logging for now. print(colored(f"Error: Failed to post entry update to Nostr: {e}", "red")) def publish_entry(self, encrypted_entry_data: bytes, nostr_kind: int, d_tag: str, is_deletion: bool = False): """Synchronous wrapper to publish an entry.""" if not self.loop.is_running(): logger.error("Cannot publish entry: Event loop is not running.") return future = asyncio.run_coroutine_threadsafe( self.publish_entry_async(encrypted_entry_data, nostr_kind, d_tag, is_deletion), self.loop ) try: future.result(timeout=10) # Wait for publish to be sent except concurrent.futures.TimeoutError: logger.warning(f"Timeout waiting for Nostr publish confirmation (Kind: {nostr_kind}, d: {d_tag}). Event might still be sent.") print(colored("Warning: Timeout posting to Nostr. Update might be delayed.", "yellow")) except Exception as e: logger.error(f"Error submitting publish task to event loop: {e}", exc_info=True) async def fetch_all_entries_async(self, since: Optional[int] = None, limit: int = 500) -> Optional[List[Event]]: """ Asynchronously fetches all SeedPass entries (Kind 31111) from Nostr. :param since: Optional Unix timestamp to fetch events newer than this. :param limit: Max number of events per relay query (relays might override). :return: A list of Event objects, or None if a critical error occurs. """ if not self.client_pool.connected: logger.warning("Cannot fetch entries: Nostr client not connected.") # Return empty list instead of None to indicate no *new* entries found due to connection issue return [] results = [] err_flag = asyncio.Event() # To signal errors from handler # Using StoreEventHandler to collect events store = StoreEventHandler() def on_error_handler(the_client: Client, sub_id: str, data: Any): logger.error(f"Error received on subscription {sub_id} from {the_client.url}: {data}") err_flag.set() # Signal that an error occurred # Filter for the specific replaceable kind authored by the user filters = [{ "authors": [self.key_manager.get_public_key_hex()], "kinds": [SEEDPASS_NOSTR_KIND], "#t": ["seedpass"], # Filter by general tag "limit": limit }] if since is not None and isinstance(since, int) and since >= 0: filters[0]["since"] = since # Add time filter if provided sub_id = None try: sub_id = f"seedpass_fetch_{uuid.uuid4()}" logger.debug(f"Subscribing to fetch entries with filter: {filters}, sub_id: {sub_id}") # Subscribe using the store handler and error handler self.client_pool.subscribe( handlers=[store, on_error_handler], # Pass list of handlers filters=filters, sub_id=sub_id, eose_func=lambda client, sub_id, events: logger.debug(f"Received EOSE for {sub_id} from {client.url}") ) self.subscriptions[sub_id] = filters # Store subscription info # Wait for EOSE from relays or a timeout/error # Timeout needs to be long enough for relays to respond fetch_timeout = 15.0 try: await asyncio.wait_for( self.client_pool.eose_matching(sub_id=sub_id), # Wait for EOSE events timeout=fetch_timeout ) logger.info(f"Received EOSE from relays for fetch subscription {sub_id}.") except asyncio.TimeoutError: logger.warning(f"Timeout waiting for EOSE on fetch subscription {sub_id} after {fetch_timeout}s.") # Continue with whatever events were received # Check if any error occurred during subscription if err_flag.is_set(): logger.error(f"Error occurred during Nostr fetch subscription {sub_id}.") # Depending on severity, maybe return None or partial results? # For now, return None to indicate failure. return None # Unsubscribe after fetching self.client_pool.unsubscribe(sub_id) if sub_id in self.subscriptions: del self.subscriptions[sub_id] logger.debug(f"Unsubscribed from fetch subscription {sub_id}.") # Get collected events from the store # Need to filter results by sub_id if store is reused, or use a fresh store each time. # Assuming store collects globally, filter results by pubkey/kind again for safety. # Actually, StoreEventHandler stores by event ID. Need a way to get all events received for the sub. # Let's refine this - maybe collect in a simple list within this function? # --- Alternative Collection --- collected_events = [] eose_received = asyncio.Event() def event_collector(the_client: Client, r_sub_id: str, evt: Event): if r_sub_id == sub_id: # Basic validation if evt.pub_key == self.key_manager.get_public_key_hex() and evt.kind == SEEDPASS_NOSTR_KIND: collected_events.append(evt) else: logger.warning(f"Received unexpected event during fetch: {evt.id} from {the_client.url}") def eose_marker(the_client: Client, r_sub_id: str, evts: List): if r_sub_id == sub_id: logger.debug(f"Received EOSE for {sub_id} from {the_client.url}") # We need to know when *enough* relays have sent EOSE. # This simple approach just sets a flag. `client_pool.eose_matching` is better. # For simplicity here, let's just use a timeout after subscribing. # --- Revert to simpler timeout-based fetch --- # This is less reliable than waiting for EOSE but simpler to implement without deeper monstr changes. collected_events_dict: Dict[str, Event] = {} # Use dict to store latest per d_tag def event_collector_simple(the_client: Client, r_sub_id: str, evt: Event): if r_sub_id == sub_id: if evt.pub_key == self.key_manager.get_public_key_hex() and evt.kind == SEEDPASS_NOSTR_KIND: d_tag_val = evt.get_tags("d") if d_tag_val: # Ensure 'd' tag exists d_tag = d_tag_val[0] # Get first 'd' tag # Store only the latest event for each 'd' tag if d_tag not in collected_events_dict or evt.created_at > collected_events_dict[d_tag].created_at: collected_events_dict[d_tag] = evt else: logger.warning(f"Received unexpected event during fetch: {evt.id} from {the_client.url}") sub_id = f"seedpass_fetch_{uuid.uuid4()}" self.client_pool.subscribe( handlers=event_collector_simple, filters=filters, sub_id=sub_id ) self.subscriptions[sub_id] = filters logger.debug(f"Subscribed to fetch entries with filter: {filters}, sub_id: {sub_id}") # Wait for a fixed time to allow events to arrive await asyncio.sleep(5.0) # Adjust this wait time as needed self.client_pool.unsubscribe(sub_id) if sub_id in self.subscriptions: del self.subscriptions[sub_id] logger.debug(f"Unsubscribed from fetch subscription {sub_id}. Collected {len(collected_events_dict)} unique entries.") return list(collected_events_dict.values()) # Return the latest event for each d_tag except Exception as e: logger.error(f"Failed during Nostr fetch: {e}", exc_info=True) # Clean up subscription if needed if sub_id and sub_id in self.subscriptions: try: self.client_pool.unsubscribe(sub_id) del self.subscriptions[sub_id] except Exception as unsub_err: logger.error(f"Error unsubscribing during fetch error handling: {unsub_err}") return None # Indicate failure def fetch_all_entries_sync(self, since: Optional[int] = None, limit: int = 500) -> Optional[List[Event]]: """Synchronous wrapper to fetch all entries.""" if not self.loop.is_running(): logger.error("Cannot fetch entries: Event loop is not running.") return None future = asyncio.run_coroutine_threadsafe( self.fetch_all_entries_async(since=since, limit=limit), self.loop ) try: return future.result(timeout=20) # Longer timeout for fetching except concurrent.futures.TimeoutError: logger.error("Timeout occurred while fetching entries from Nostr.") print(colored("Error: Timeout occurred while fetching entries from Nostr.", "red")) return None except Exception as e: logger.error(f"Error submitting fetch task to event loop: {e}", exc_info=True) return None def close_client_pool(self): """Gracefully shuts down the Nostr client pool and event loop.""" if self.is_shutting_down: logger.debug("Shutdown already in progress.") return self.is_shutting_down = True logger.info("Initiating NostrClient shutdown...") # Schedule the async close in the running loop if self.loop.is_running(): future = asyncio.run_coroutine_threadsafe(self._close_pool_async(), self.loop) try: future.result(timeout=10) # Wait for async close to finish except (concurrent.futures.TimeoutError, Exception) as e: logger.warning(f"Error or timeout during async pool close: {e}. Proceeding with loop stop.") # Stop the loop from the thread that owns it if self.loop.is_running(): self.loop.call_soon_threadsafe(self.loop.stop) else: logger.warning("NostrClient event loop was not running during shutdown.") # Wait for the thread to finish if self.loop_thread.is_alive(): self.loop_thread.join(timeout=5) if self.loop_thread.is_alive(): logger.warning("NostrClient event loop thread did not exit cleanly.") logger.info("NostrClient shutdown complete.") self.is_shutting_down = False async def _close_pool_async(self): """Async part of the shutdown sequence.""" try: logger.debug("Closing Nostr subscriptions...") sub_ids = list(self.subscriptions.keys()) for sub_id in sub_ids: try: self.client_pool.unsubscribe(sub_id) if sub_id in self.subscriptions: del self.subscriptions[sub_id] logger.debug(f"Unsubscribed from {sub_id}") except Exception as e: logger.warning(f"Error unsubscribing from {sub_id}: {e}") logger.debug("Closing Nostr client connections...") # Use await self.client_pool.disconnect() if available and preferred by monstr version # Otherwise, manually close underlying clients if accessible if hasattr(self.client_pool, '_clients'): # Accessing protected member, check monstr docs tasks = [self._safe_close_connection(c) for c in self.client_pool._clients.values()] await asyncio.gather(*tasks, return_exceptions=True) elif hasattr(self.client_pool, 'clients'): # Public attribute? tasks = [self._safe_close_connection(c) for c in self.client_pool.clients] await asyncio.gather(*tasks, return_exceptions=True) else: logger.warning("Cannot access client pool clients for explicit closure.") logger.debug("Async pool closure steps finished.") except Exception as e: logger.error(f"Error during async Nostr pool closure: {e}", exc_info=True) async def _safe_close_connection(self, client: Client): """Safely attempts to close a single client connection.""" # Older monstr versions might not have close_connection or disconnect close_method = getattr(client, 'disconnect', getattr(client, 'close_connection', None)) if close_method and asyncio.iscoroutinefunction(close_method): try: await asyncio.wait_for(close_method(), timeout=3) logger.debug(f"Closed connection to {client.url}") except asyncio.TimeoutError: logger.warning(f"Timeout closing connection to {client.url}") except Exception as e: logger.warning(f"Error closing connection to {client.url}: {e}") elif close_method: # Non-async close? Less likely for websockets. try: close_method() logger.debug(f"Closed connection to {client.url} (sync)") except Exception as e: logger.warning(f"Error closing connection to {client.url} (sync): {e}") else: logger.warning(f"No suitable close method found for client connected to {client.url}") # --- Remove Old Methods --- # remove publish_event, subscribe, retrieve_json_from_nostr_async, retrieve_json_from_nostr # remove do_post_async, subscribe_feed_async, publish_and_subscribe_async, publish_and_subscribe # remove decrypt_and_save_index_from_nostr, save_json_data, update_checksum, decrypt_data_from_file # remove publish_json_to_nostr, retrieve_json_from_nostr_sync, decrypt_and_save_index_from_nostr_public
Phase 5: Refactor EventHandler
-
nostr/event_handler.py(Simplified/Removed): The event handling logic is now tightly coupled with synchronization inPasswordManager. A separateEventHandlerclass primarily for logging received events (as it was before) might still be useful for debugging, but it won't be directly involved in processing SeedPass entries anymore. Theevent_collector_simplefunction insideNostrClient.fetch_all_entries_asyncnow handles the basic reception. The actual processing happens inPasswordManager.synchronize_with_nostr. Decision: We can remove the oldEventHandlerclass or keep it purely for debug logging if needed, but it's not essential for the new flow. Let's comment it out for now.# nostr/event_handler.py # import time # import logging # import traceback # from monstr.event.event import Event # logger = logging.getLogger(__name__) # class EventHandler: # """ # Handles incoming Nostr events (Primarily for Debug Logging now). # Actual entry processing is done within PasswordManager.synchronize_with_nostr. # """ # def __init__(self): # pass # No password manager reference needed now # def handle_new_event(self, the_client, sub_id, evt: Event): # """Processes incoming events by logging their details.""" # # This might be attached to a general subscription for debugging # try: # created_at_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(evt.created_at)) # logger.debug( # f"[Debug Event Handler] Received Event:" # f" SubID: {sub_id}" # f" | Relay: {the_client.url}" # f" | Kind: {evt.kind}" # f" | ID: {evt.id}" # f" | Created: {created_at_str}" # f" | Content Preview: {evt.content[:50]}..." # ) # except Exception as e: # logger.error(f"Error in debug event handler: {e}", exc_info=True)
Phase 6: Refactor main.py
main.py(Major Changes to Menu and Handlers):# main.py import os import sys import logging import signal from colorama import init as colorama_init, Style from termcolor import colored import traceback # Import PasswordManager - NostrClient is now initialized within it from password_manager.manager import PasswordManager # from nostr.client import NostrClient # No longer needed here colorama_init(autoreset=True) # Autoreset colors # --- Logging Configuration (Keep as is) --- def configure_logging(): # ... (keep existing logging setup) ... pass # Keep existing code # --- Confirmation Helper (Keep as is) --- def confirm_action(prompt: str) -> bool: # ... (keep existing confirmation logic) ... pass # Keep existing code # --- New CLI Interaction Logic --- def display_main_menu(password_manager: PasswordManager): """Displays the main interactive menu.""" print(colored("\n--- SeedPass Main Menu ---", "blue", attrs=["bold"])) print(f"Active Profile: {colored(password_manager.current_fingerprint, 'green')}\n") menu_options = { "1": ("Add New Entry", password_manager.handle_add_entry_cli), "2": ("List / Retrieve Entries", password_manager.handle_retrieve_entry_cli), "3": ("Modify Entry", password_manager.handle_modify_entry_cli), "4": ("Delete Entry", password_manager.handle_delete_entry_cli), "5": ("Synchronize with Nostr", password_manager.synchronize_with_nostr), # Changed "6": ("Display Nostr Public Key (npub)", handle_display_npub), # Needs adapting "7": ("Manage Backups", handle_backup_menu), # New Sub-menu "8": ("Manage Profiles (Seeds)", handle_profile_menu), # New Sub-menu "9": ("Verify Script Checksum", password_manager.handle_verify_checksum), "10": ("Exit", None) # Handled in loop } for key, (text, _) in menu_options.items(): print(f" {Style.BRIGHT}{key}{Style.RESET_ALL}. {text}") return menu_options def handle_display_npub(password_manager: PasswordManager): """Displays the Nostr public key (npub).""" # Assumes nostr_client and key_manager are initialized if password_manager.nostr_client and password_manager.nostr_client.key_manager: try: npub = password_manager.nostr_client.key_manager.get_npub() print(colored(f"\nYour Nostr Public Key (npub) for profile '{password_manager.current_fingerprint}':", 'cyan')) print(colored(npub, 'yellow')) print(colored("Share this key for others to send you encrypted messages (if supported).", 'cyan')) except Exception as e: logger.error(f"Failed to get npub: {e}", exc_info=True) print(colored(f"Error displaying npub: {e}", "red")) else: print(colored("Nostr client not initialized for this profile.", "red")) def handle_backup_menu(password_manager: PasswordManager): """Handles the backup management sub-menu.""" if not password_manager.backup_manager: print(colored("Backup manager not initialized.", "red")) return backup_menu_options = { "1": ("Backup Specific Entry", password_manager.handle_backup_entry_cli), "2": ("Restore Specific Entry", password_manager.handle_restore_entry_cli), "3": ("List Backups for Entry", lambda: password_manager.backup_manager.display_backups( entry_num=int(input("Enter entry number to list backups for: ")) # Add error handling )), "4": ("List All Backups", lambda: password_manager.backup_manager.display_backups()), "5": ("Return to Main Menu", None) } while True: print(colored("\n--- Backup Management ---", "blue")) for key, (text, _) in backup_menu_options.items(): print(f" {key}. {text}") choice = input("Enter your choice: ").strip() if choice == '5': break # Return to main menu selected_option = backup_menu_options.get(choice) if selected_option and selected_option[1]: try: selected_option[1]() # Call the handler function except ValueError: print(colored("Invalid numeric input.", "red")) except Exception as e: logger.error(f"Error in backup menu option {choice}: {e}", exc_info=True) print(colored(f"An error occurred: {e}", "red")) elif selected_option: # Option exists but no function (like return) pass else: print(colored("Invalid choice.", "red")) def handle_profile_menu(password_manager: PasswordManager): """Handles the profile (fingerprint) management sub-menu.""" if not password_manager.fingerprint_manager: print(colored("Profile manager not initialized.", "red")) return profile_menu_options = { "1": ("Switch Active Profile", password_manager.handle_switch_fingerprint), # Assumes this returns bool "2": ("Add New Profile", password_manager.handle_add_new_fingerprint_cli), "3": ("Remove Profile", password_manager.handle_remove_fingerprint_cli), "4": ("List All Profiles", password_manager.handle_list_fingerprints_cli), "5": ("Backup/Reveal Current Profile Seed", password_manager.handle_backup_reveal_parent_seed), # Moved here "6": ("Return to Main Menu", None) } while True: print(colored("\n--- Profile Management ---", "blue")) for key, (text, _) in profile_menu_options.items(): print(f" {key}. {text}") choice = input("Enter your choice: ").strip() if choice == '6': break # Return to main menu selected_option = profile_menu_options.get(choice) if selected_option and selected_option[1]: try: result = selected_option[1]() # Call the handler function # Handle specific results if needed (e.g., switch profile might fail) if selected_option[0] == "Switch Active Profile" and result: print(colored("Profile switched successfully. Returning to main menu.", "green")) break # Exit sub-menu after successful switch except Exception as e: logger.error(f"Error in profile menu option {choice}: {e}", exc_info=True) print(colored(f"An error occurred: {e}", "red")) elif selected_option: pass else: print(colored("Invalid choice.", "red")) # --- Main Execution Logic --- if __name__ == '__main__': configure_logging() logger = logging.getLogger(__name__) logger.info("--- Starting SeedPass ---") password_manager: Optional[PasswordManager] = None # Define before try block try: # Initialization is now more complex, handled inside PasswordManager __init__ password_manager = PasswordManager() logger.info("PasswordManager initialization complete.") except SystemExit: logger.warning("SystemExit during initialization.") # Don't print error message again if sys.exit was called intentionally sys.exit(1) # Ensure exit code reflects failure except Exception as e: # Catch any other unexpected init errors logger.critical(f"Unhandled exception during PasswordManager initialization: {e}", exc_info=True) print(colored(f"FATAL ERROR during startup: {e}. Check logs.", "red", attrs=["bold"])) # Ensure cleanup if partially initialized? Difficult here. if password_manager and password_manager.nostr_client: password_manager.nostr_client.close_client_pool() sys.exit(1) # Register signal handlers for graceful shutdown def signal_handler(sig, frame): print(colored("\nReceived shutdown signal. Exiting gracefully...", 'yellow')) logging.info(f"Received shutdown signal: {sig}. Initiating graceful shutdown.") if password_manager and password_manager.nostr_client: try: password_manager.nostr_client.close_client_pool() logging.info("NostrClient closed successfully.") except Exception as e: logging.error(f"Error closing NostrClient during shutdown: {e}", exc_info=True) print(colored(f"Error during Nostr shutdown: {e}", 'red')) logging.info("--- SeedPass Shutting Down ---") sys.exit(0) signal.signal(signal.SIGINT, signal_handler) # Handle Ctrl+C signal.signal(signal.SIGTERM, signal_handler) # Handle termination signals # --- Main Application Loop --- try: while True: menu = display_main_menu(password_manager) choice = input(colored('Enter your choice: ', "magenta")).strip() if choice == '10': # Exit option break # Exit loop # Execute chosen action selected_option = menu.get(choice) if selected_option and selected_option[1]: try: # Call the appropriate handler function (now mostly methods of PasswordManager) # Pass password_manager instance only if the handler is a standalone function (like handle_display_npub) if selected_option[0] == "Display Nostr Public Key (npub)": handle_display_npub(password_manager) elif selected_option[0] == "Manage Backups": handle_backup_menu(password_manager) elif selected_option[0] == "Manage Profiles (Seeds)": handle_profile_menu(password_manager) else: selected_option[1]() # Call method on password_manager instance except Exception as menu_err: logger.error(f"Error during menu action '{selected_option[0]}': {menu_err}", exc_info=True) print(colored(f"An error occurred: {menu_err}", "red")) elif selected_option: # Option exists but no function (should not happen with this menu structure) pass else: print(colored("Invalid choice. Please select a valid option.", 'red')) except KeyboardInterrupt: logger.info("Program terminated by user (Ctrl+C in main loop).") print(colored("\nExiting...", 'yellow')) # Signal handler should have been called, but call cleanup just in case if password_manager and password_manager.nostr_client: password_manager.nostr_client.close_client_pool() sys.exit(0) except Exception as main_loop_err: logger.critical(f"An unexpected error occurred in the main loop: {main_loop_err}", exc_info=True) print(colored(f"FATAL ERROR: An unexpected error occurred: {main_loop_err}", 'red', attrs=["bold"])) # Attempt cleanup if password_manager and password_manager.nostr_client: password_manager.nostr_client.close_client_pool() sys.exit(1) finally: # Ensure cleanup runs on normal exit too logger.info("Exiting main loop.") if password_manager and password_manager.nostr_client: password_manager.nostr_client.close_client_pool() logging.info("--- SeedPass Finished ---") print(colored("Exiting SeedPass.", 'green')) sys.exit(0)
Phase 7: Remove Obsolete Files
- Delete
nostr/encryption_manager.py.
Summary of Key Changes:
kinds.py: Central definition for entry types.handlers/: Specific logic for processing eachkind.state_manager.py: Tracks last generated password index and sync time.EntryManager: Now manages individual entry files (saving, loading, deleting, checksumming data within the entry). No longer holds the index logic.BackupManager: Adapted to back up/restore individual entry files.PasswordManager: Orchestrates the new flow. Contains methods foradd_entry,modify_entry,delete_entry,synchronize_with_nostr,process_entry, data migration, and CLI handlers. Reads/writes state viaStateManager.NostrClient: Publishes/fetches individual replaceable events (Kind 31111 suggested) usingdtags. Fetching retrieves latest versions. UsesStoreEventHandler(or similar) for collecting results.publish_json_to_nostrandretrieve_json_from_nostr_syncremoved/replaced.main.py: Updated menu structure, calling newPasswordManagermethods and sub-menu handlers.
Next Steps:
- Review Thoroughly: Carefully compare the refactored code against the original and the plan.
- Implement Tests: Write unit tests for
StateManager, newEntryManagermethods,kinds.pyhelpers, and the handlers. Write integration tests for theadd -> sync -> retrieveflow. - Refine
NostrClientFetching: Thefetch_all_entries_asyncimplementation using a simple timeout is basic. Improving it to reliably use EOSE detection across multiple relays (client_pool.eose_matching) would be more robust. - Error Handling: Add more specific error handling, especially around file I/O, network issues, and migration edge cases.
- User Experience: Test the CLI flow extensively. Ensure prompts are clear and error messages are helpful. Consider how users will select entries for modification/deletion (the current implementation lists all and asks for a number). Search functionality might be needed later.