mirror of
https://github.com/PR0M3TH3AN/SeedPass.git
synced 2026-10-31 09:52:03 +00:00
Refactor initialize_managers in PasswordManager for better readability
Extracted logic from `initialize_managers` into helper methods: - `_initialize_config_and_state` - `_initialize_entry_manager` - `_initialize_password_generator` - `_initialize_nostr_client` - `_handle_migration` This improves code organization and maintainability without altering functionality. Verified with `test_vault_initialization.py`. Co-authored-by: PR0M3TH3AN <[email protected]>
This commit is contained in:
google-labs-jules[bot]andPR0M3TH3AN committed
1 parent
ece0dda09f
commit
24fa2e8e1a
1 file changed
+140
-120
+140
-120
@@ -1677,127 +1677,13 @@ class PasswordManager:
|
|||||||
# Derive sub-keys if needed
|
# Derive sub-keys if needed
|
||||||
self.ensure_key_hierarchy()
|
self.ensure_key_hierarchy()
|
||||||
|
|
||||||
# Reinitialize the managers with the updated EncryptionManager and current fingerprint context
|
self._initialize_config_and_state()
|
||||||
self.config_manager = ConfigManager(
|
migrated, last_migration_performed, index_exists = (
|
||||||
vault=self.vault,
|
self._initialize_entry_manager()
|
||||||
fingerprint_dir=self.fingerprint_dir,
|
|
||||||
)
|
)
|
||||||
self.state_manager = StateManager(self.fingerprint_dir)
|
self._initialize_password_generator()
|
||||||
self.backup_manager = BackupManager(
|
self._initialize_nostr_client()
|
||||||
fingerprint_dir=self.fingerprint_dir,
|
self._handle_migration(migrated, last_migration_performed, index_exists)
|
||||||
config_manager=self.config_manager,
|
|
||||||
)
|
|
||||||
|
|
||||||
migrated = False
|
|
||||||
last_migration_performed = False
|
|
||||||
index_exists = (
|
|
||||||
self.vault.index_file.exists()
|
|
||||||
or (
|
|
||||||
self.vault.fingerprint_dir / "seedpass_passwords_db.json.enc"
|
|
||||||
).exists()
|
|
||||||
)
|
|
||||||
try:
|
|
||||||
_, migrated, last_migration_performed = self.vault.load_index(
|
|
||||||
return_migration_flags=True
|
|
||||||
)
|
|
||||||
except RuntimeError as exc:
|
|
||||||
print(colored(str(exc), "red"))
|
|
||||||
raise SeedPassError(str(exc))
|
|
||||||
|
|
||||||
self.entry_manager = EntryManager(
|
|
||||||
vault=self.vault,
|
|
||||||
backup_manager=self.backup_manager,
|
|
||||||
)
|
|
||||||
|
|
||||||
pw_bip85 = BIP85(self.KEY_PW_DERIVE)
|
|
||||||
self.password_generator = PasswordGenerator(
|
|
||||||
encryption_manager=self.encryption_manager,
|
|
||||||
parent_seed=self.KEY_PW_DERIVE,
|
|
||||||
bip85=pw_bip85,
|
|
||||||
policy=self.config_manager.get_password_policy(),
|
|
||||||
)
|
|
||||||
|
|
||||||
# Load relay configuration and initialize NostrClient
|
|
||||||
config = self.config_manager.load_config()
|
|
||||||
if getattr(self, "state_manager", None) is not None:
|
|
||||||
state = self.state_manager.state
|
|
||||||
relay_list = state.get("relays", list(DEFAULT_RELAYS))
|
|
||||||
self.last_bip85_idx = state.get("last_bip85_idx", 0)
|
|
||||||
self.last_sync_ts = state.get("last_sync_ts", 0)
|
|
||||||
self.manifest_id = state.get("manifest_id")
|
|
||||||
self.delta_since = state.get("delta_since", 0)
|
|
||||||
self.nostr_account_idx = state.get("nostr_account_idx", 0)
|
|
||||||
else:
|
|
||||||
relay_list = list(DEFAULT_RELAYS)
|
|
||||||
self.last_bip85_idx = 0
|
|
||||||
self.last_sync_ts = 0
|
|
||||||
self.manifest_id = None
|
|
||||||
self.delta_since = 0
|
|
||||||
self.nostr_account_idx = 0
|
|
||||||
self.offline_mode = bool(config.get("offline_mode", True))
|
|
||||||
self.inactivity_timeout = config.get(
|
|
||||||
"inactivity_timeout", INACTIVITY_TIMEOUT
|
|
||||||
)
|
|
||||||
self.secret_mode_enabled = bool(config.get("secret_mode_enabled", False))
|
|
||||||
self.clipboard_clear_delay = int(config.get("clipboard_clear_delay", 45))
|
|
||||||
self.verbose_timing = bool(config.get("verbose_timing", False))
|
|
||||||
if not self.offline_mode:
|
|
||||||
print("Connecting to relays...")
|
|
||||||
self.nostr_client = NostrClient(
|
|
||||||
encryption_manager=self.encryption_manager,
|
|
||||||
fingerprint=self.current_fingerprint,
|
|
||||||
relays=relay_list,
|
|
||||||
offline_mode=self.offline_mode,
|
|
||||||
config_manager=self.config_manager,
|
|
||||||
parent_seed=getattr(self, "parent_seed", None),
|
|
||||||
key_index=self.KEY_INDEX,
|
|
||||||
account_index=self.nostr_account_idx,
|
|
||||||
)
|
|
||||||
|
|
||||||
if getattr(self, "manifest_id", None) and hasattr(
|
|
||||||
self.nostr_client, "_state_lock"
|
|
||||||
):
|
|
||||||
from nostr.backup_models import Manifest
|
|
||||||
|
|
||||||
with self.nostr_client._state_lock:
|
|
||||||
self.nostr_client.current_manifest_id = self.manifest_id
|
|
||||||
self.nostr_client.current_manifest = Manifest(
|
|
||||||
ver=1,
|
|
||||||
algo="gzip",
|
|
||||||
chunks=[],
|
|
||||||
delta_since=self.delta_since or None,
|
|
||||||
)
|
|
||||||
|
|
||||||
if migrated and index_exists:
|
|
||||||
print(colored("Local database migration successful.", "green"))
|
|
||||||
if self.encryption_manager is not None:
|
|
||||||
self.encryption_manager.last_migration_performed = False
|
|
||||||
if last_migration_performed and not self.offline_mode:
|
|
||||||
if confirm_action(
|
|
||||||
"Do you want to sync the migrated profile to Nostr now?"
|
|
||||||
):
|
|
||||||
result = self.sync_vault()
|
|
||||||
if result:
|
|
||||||
print(
|
|
||||||
colored(
|
|
||||||
"Profile synchronized to Nostr successfully.",
|
|
||||||
"green",
|
|
||||||
)
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
print(
|
|
||||||
colored(
|
|
||||||
"Error: Failed to sync profile to Nostr.",
|
|
||||||
"red",
|
|
||||||
)
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
print(
|
|
||||||
colored(
|
|
||||||
"You can sync the migrated profile later from the main menu.",
|
|
||||||
"yellow",
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
logger.debug("Managers re-initialized for the new fingerprint.")
|
logger.debug("Managers re-initialized for the new fingerprint.")
|
||||||
|
|
||||||
@@ -1806,6 +1692,140 @@ class PasswordManager:
|
|||||||
print(colored(f"Error: Failed to initialize managers: {e}", "red"))
|
print(colored(f"Error: Failed to initialize managers: {e}", "red"))
|
||||||
raise SeedPassError(f"Failed to initialize managers: {e}") from e
|
raise SeedPassError(f"Failed to initialize managers: {e}") from e
|
||||||
|
|
||||||
|
def _initialize_config_and_state(self) -> None:
|
||||||
|
"""Initialize configuration, state, and backup managers."""
|
||||||
|
self.config_manager = ConfigManager(
|
||||||
|
vault=self.vault,
|
||||||
|
fingerprint_dir=self.fingerprint_dir,
|
||||||
|
)
|
||||||
|
self.state_manager = StateManager(self.fingerprint_dir)
|
||||||
|
self.backup_manager = BackupManager(
|
||||||
|
fingerprint_dir=self.fingerprint_dir,
|
||||||
|
config_manager=self.config_manager,
|
||||||
|
)
|
||||||
|
|
||||||
|
def _initialize_entry_manager(self) -> tuple[bool, bool, bool]:
|
||||||
|
"""
|
||||||
|
Initialize the entry manager and load the vault index.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
tuple: (migrated, last_migration_performed, index_exists)
|
||||||
|
"""
|
||||||
|
migrated = False
|
||||||
|
last_migration_performed = False
|
||||||
|
index_exists = (
|
||||||
|
self.vault.index_file.exists()
|
||||||
|
or (self.vault.fingerprint_dir / "seedpass_passwords_db.json.enc").exists()
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
_, migrated, last_migration_performed = self.vault.load_index(
|
||||||
|
return_migration_flags=True
|
||||||
|
)
|
||||||
|
except RuntimeError as exc:
|
||||||
|
print(colored(str(exc), "red"))
|
||||||
|
raise SeedPassError(str(exc))
|
||||||
|
|
||||||
|
self.entry_manager = EntryManager(
|
||||||
|
vault=self.vault,
|
||||||
|
backup_manager=self.backup_manager,
|
||||||
|
)
|
||||||
|
return migrated, last_migration_performed, index_exists
|
||||||
|
|
||||||
|
def _initialize_password_generator(self) -> None:
|
||||||
|
"""Initialize the password generator."""
|
||||||
|
pw_bip85 = BIP85(self.KEY_PW_DERIVE)
|
||||||
|
self.password_generator = PasswordGenerator(
|
||||||
|
encryption_manager=self.encryption_manager,
|
||||||
|
parent_seed=self.KEY_PW_DERIVE,
|
||||||
|
bip85=pw_bip85,
|
||||||
|
policy=self.config_manager.get_password_policy(),
|
||||||
|
)
|
||||||
|
|
||||||
|
def _initialize_nostr_client(self) -> None:
|
||||||
|
"""Load relay configuration and initialize NostrClient."""
|
||||||
|
config = self.config_manager.load_config()
|
||||||
|
if getattr(self, "state_manager", None) is not None:
|
||||||
|
state = self.state_manager.state
|
||||||
|
relay_list = state.get("relays", list(DEFAULT_RELAYS))
|
||||||
|
self.last_bip85_idx = state.get("last_bip85_idx", 0)
|
||||||
|
self.last_sync_ts = state.get("last_sync_ts", 0)
|
||||||
|
self.manifest_id = state.get("manifest_id")
|
||||||
|
self.delta_since = state.get("delta_since", 0)
|
||||||
|
self.nostr_account_idx = state.get("nostr_account_idx", 0)
|
||||||
|
else:
|
||||||
|
relay_list = list(DEFAULT_RELAYS)
|
||||||
|
self.last_bip85_idx = 0
|
||||||
|
self.last_sync_ts = 0
|
||||||
|
self.manifest_id = None
|
||||||
|
self.delta_since = 0
|
||||||
|
self.nostr_account_idx = 0
|
||||||
|
self.offline_mode = bool(config.get("offline_mode", True))
|
||||||
|
self.inactivity_timeout = config.get("inactivity_timeout", INACTIVITY_TIMEOUT)
|
||||||
|
self.secret_mode_enabled = bool(config.get("secret_mode_enabled", False))
|
||||||
|
self.clipboard_clear_delay = int(config.get("clipboard_clear_delay", 45))
|
||||||
|
self.verbose_timing = bool(config.get("verbose_timing", False))
|
||||||
|
if not self.offline_mode:
|
||||||
|
print("Connecting to relays...")
|
||||||
|
self.nostr_client = NostrClient(
|
||||||
|
encryption_manager=self.encryption_manager,
|
||||||
|
fingerprint=self.current_fingerprint,
|
||||||
|
relays=relay_list,
|
||||||
|
offline_mode=self.offline_mode,
|
||||||
|
config_manager=self.config_manager,
|
||||||
|
parent_seed=getattr(self, "parent_seed", None),
|
||||||
|
key_index=self.KEY_INDEX,
|
||||||
|
account_index=self.nostr_account_idx,
|
||||||
|
)
|
||||||
|
|
||||||
|
if getattr(self, "manifest_id", None) and hasattr(
|
||||||
|
self.nostr_client, "_state_lock"
|
||||||
|
):
|
||||||
|
from nostr.backup_models import Manifest
|
||||||
|
|
||||||
|
with self.nostr_client._state_lock:
|
||||||
|
self.nostr_client.current_manifest_id = self.manifest_id
|
||||||
|
self.nostr_client.current_manifest = Manifest(
|
||||||
|
ver=1,
|
||||||
|
algo="gzip",
|
||||||
|
chunks=[],
|
||||||
|
delta_since=self.delta_since or None,
|
||||||
|
)
|
||||||
|
|
||||||
|
def _handle_migration(
|
||||||
|
self, migrated: bool, last_migration_performed: bool, index_exists: bool
|
||||||
|
) -> None:
|
||||||
|
"""Handle database migration notifications and sync."""
|
||||||
|
if migrated and index_exists:
|
||||||
|
print(colored("Local database migration successful.", "green"))
|
||||||
|
if self.encryption_manager is not None:
|
||||||
|
self.encryption_manager.last_migration_performed = False
|
||||||
|
if last_migration_performed and not self.offline_mode:
|
||||||
|
if confirm_action(
|
||||||
|
"Do you want to sync the migrated profile to Nostr now?"
|
||||||
|
):
|
||||||
|
result = self.sync_vault()
|
||||||
|
if result:
|
||||||
|
print(
|
||||||
|
colored(
|
||||||
|
"Profile synchronized to Nostr successfully.",
|
||||||
|
"green",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
print(
|
||||||
|
colored(
|
||||||
|
"Error: Failed to sync profile to Nostr.",
|
||||||
|
"red",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
print(
|
||||||
|
colored(
|
||||||
|
"You can sync the migrated profile later from the main menu.",
|
||||||
|
"yellow",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
async def sync_index_from_nostr_async(self) -> None:
|
async def sync_index_from_nostr_async(self) -> None:
|
||||||
"""Always fetch the latest vault data from Nostr and update the local index."""
|
"""Always fetch the latest vault data from Nostr and update the local index."""
|
||||||
if not getattr(self, "nostr_client", None):
|
if not getattr(self, "nostr_client", None):
|
||||||
|
|||||||
Reference in new issue
Block a user