thePR0M3TH3AN/seedPass
1
0
Fork 0
mirror of https://github.com/PR0M3TH3AN/SeedPass.git synced 2025-09-08 07:18:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #513 from PR0M3TH3AN/codex/find-conditional-network-calls-in-vault-and-tui

Browse Source 
Avoid blocking startup with network calls
This commit is contained in:
thePR0M3TH3AN
2025-07-13 18:35:57 -04:00
committed by GitHub
parent 00332fb145 0e7c3e8a84
commit 9d69044895
2 changed files with 75 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
src/password_manager/encryption.py
View File

@@ -8,7 +8,9 @@ try:
JSONDecodeError = orjson.JSONDecodeError
USE_ORJSON = True
except Exception: # pragma: no cover - fallback for environments without orjson
except (
Exception
): # pragma: no cover - fallback for environments without orjson
import json as json_lib
from json import JSONDecodeError
@@ -58,7 +60,9 @@ class EncryptionManager:
self.key = base64.urlsafe_b64decode(self.key_b64)
self.cipher = AESGCM(self.key)
logger.debug(f"EncryptionManager initialized for {self.fingerprint_dir}")
logger.debug(
f"EncryptionManager initialized for {self.fingerprint_dir}"
)
except Exception as e:
logger.error(
f"Failed to initialize ciphers with provided encryption key: {e}",
@@ -91,12 +95,16 @@ class EncryptionManager:
ciphertext = encrypted_data[15:]
return self.cipher.decrypt(nonce, ciphertext, None)
except InvalidTag as e:
logger.error("AES-GCM decryption failed: Invalid authentication tag.")
logger.error(
"AES-GCM decryption failed: Invalid authentication tag."
)
raise InvalidToken("AES-GCM decryption failed.") from e
# If it's not V2, it must be the legacy Fernet format
else:
logger.warning("Data is in legacy Fernet format. Attempting migration.")
logger.warning(
"Data is in legacy Fernet format. Attempting migration."
)
try:
return self.fernet.decrypt(encrypted_data)
except InvalidToken as e:
@@ -118,7 +126,9 @@ class EncryptionManager:
fh.truncate()
fh.write(encrypted_data)
os.chmod(self.parent_seed_file, 0o600)
logger.info(f"Parent seed encrypted and saved to '{self.parent_seed_file}'.")
logger.info(
f"Parent seed encrypted and saved to '{self.parent_seed_file}'."
)
def decrypt_parent_seed(self) -> str:
"""Decrypts and returns the parent seed, handling migration."""
@@ -130,7 +140,9 @@ class EncryptionManager:
decrypted_data = self.decrypt_data(encrypted_data)
if is_legacy:
logger.info("Parent seed was in legacy format. Re-encrypting to V2 format.")
logger.info(
"Parent seed was in legacy format. Re-encrypting to V2 format."
)
self.encrypt_parent_seed(decrypted_data.decode("utf-8").strip())
return decrypted_data.decode("utf-8").strip()
@@ -143,6 +155,8 @@ class EncryptionManager:
fh.seek(0)
fh.truncate()
fh.write(encrypted_data)
fh.flush()
os.fsync(fh.fileno())
os.chmod(file_path, 0o600)
def decrypt_file(self, relative_path: Path) -> bytes:
@@ -152,13 +166,17 @@ class EncryptionManager:
encrypted_data = fh.read()
return self.decrypt_data(encrypted_data)
def save_json_data(self, data: dict, relative_path: Optional[Path] = None) -> None:
def save_json_data(
self, data: dict, relative_path: Optional[Path] = None
) -> None:
if relative_path is None:
relative_path = Path("seedpass_entries_db.json.enc")
if USE_ORJSON:
json_data = json_lib.dumps(data)
else:
json_data = json_lib.dumps(data, separators=(",", ":")).encode("utf-8")
json_data = json_lib.dumps(data, separators=(",", ":")).encode(
"utf-8"
)
self.encrypt_and_save_file(json_data, relative_path)
logger.debug(f"JSON data encrypted and saved to '{relative_path}'.")
@@ -189,7 +207,9 @@ class EncryptionManager:
# If it was a legacy file, re-save it in the new format now
if is_legacy:
logger.info(f"Migrating and re-saving legacy vault file: {file_path}")
logger.info(
f"Migrating and re-saving legacy vault file: {file_path}"
)
self.save_json_data(data, relative_path)
self.update_checksum(relative_path)
@@ -224,17 +244,25 @@ class EncryptionManager:
data = json_lib.loads(decrypted_data)
else:
data = json_lib.loads(decrypted_data.decode("utf-8"))
self.save_json_data(data, relative_path) # This always saves in V2 format
self.save_json_data(
data, relative_path
) # This always saves in V2 format
self.update_checksum(relative_path)
logger.info("Index file from Nostr was processed and saved successfully.")
print(colored("Index file updated from Nostr successfully.", "green"))
logger.info(
"Index file from Nostr was processed and saved successfully."
)
print(
colored("Index file updated from Nostr successfully.", "green")
)
except Exception as e:
logger.error(
f"Failed to decrypt and save data from Nostr: {e}", exc_info=True
f"Failed to decrypt and save data from Nostr: {e}",
exc_info=True,
)
print(
colored(
f"Error: Failed to decrypt and save data from Nostr: {e}", "red"
f"Error: Failed to decrypt and save data from Nostr: {e}",
"red",
)
)
raise
@@ -258,10 +286,13 @@ class EncryptionManager:
fh.seek(0)
fh.truncate()
fh.write(checksum.encode("utf-8"))
fh.flush()
os.fsync(fh.fileno())
os.chmod(checksum_file, 0o600)
except Exception as e:
logger.error(
f"Failed to update checksum for '{relative_path}': {e}", exc_info=True
f"Failed to update checksum for '{relative_path}': {e}",
exc_info=True,
)
raise
@@ -272,17 +303,24 @@ class EncryptionManager:
if len(words) != 12:
logger.error("Seed phrase does not contain exactly 12 words.")
print(
colored("Error: Seed phrase must contain exactly 12 words.", "red")
colored(
"Error: Seed phrase must contain exactly 12 words.",
"red",
)
)
return False
logger.debug("Seed phrase validated successfully.")
return True
except Exception as e:
logging.error(f"Error validating seed phrase: {e}", exc_info=True)
print(colored(f"Error: Failed to validate seed phrase: {e}", "red"))
print(
colored(f"Error: Failed to validate seed phrase: {e}", "red")
)
return False
def derive_seed_from_mnemonic(self, mnemonic: str, passphrase: str = "") -> bytes:
def derive_seed_from_mnemonic(
self, mnemonic: str, passphrase: str = ""
) -> bytes:
try:
if not isinstance(mnemonic, str):
if isinstance(mnemonic, list):
@@ -290,13 +328,21 @@ class EncryptionManager:
else:
mnemonic = str(mnemonic)
if not isinstance(mnemonic, str):
raise TypeError("Mnemonic must be a string after conversion")
raise TypeError(
"Mnemonic must be a string after conversion"
)
from bip_utils import Bip39SeedGenerator
seed = Bip39SeedGenerator(mnemonic).Generate(passphrase)
logger.debug("Seed derived successfully from mnemonic.")
return seed
except Exception as e:
logger.error(f"Failed to derive seed from mnemonic: {e}", exc_info=True)
print(colored(f"Error: Failed to derive seed from mnemonic: {e}", "red"))
logger.error(
f"Failed to derive seed from mnemonic: {e}", exc_info=True
)
print(
colored(
f"Error: Failed to derive seed from mnemonic: {e}", "red"
)
)
raise

31
src/password_manager/manager.py
View File

@@ -587,7 +587,7 @@ class PasswordManager:
self.initialize_managers()
self.locked = False
self.update_activity()
self.sync_index_from_nostr_if_missing()
self.start_background_sync()
def exit_managed_account(self) -> None:
"""Return to the parent seed profile if one is on the stack."""
@@ -934,7 +934,6 @@ class PasswordManager:
self.initialize_bip85()
self.initialize_managers()
self.start_background_sync()
except Exception as e:
logging.error(f"Failed to encrypt and save parent seed: {e}", exc_info=True)
print(colored(f"Error: Failed to encrypt and save parent seed: {e}", "red"))
@@ -1053,7 +1052,10 @@ class PasswordManager:
def _worker() -> None:
try:
self.sync_index_from_nostr()
if hasattr(self, "nostr_client") and hasattr(self, "vault"):
self.sync_index_from_nostr_if_missing()
if hasattr(self, "sync_index_from_nostr"):
self.sync_index_from_nostr()
except Exception as exc:
logger.warning(f"Background sync failed: {exc}")
@@ -3785,29 +3787,12 @@ class PasswordManager:
# Nostr sync info
manifest = getattr(self.nostr_client, "current_manifest", None)
if manifest is None:
try:
result = asyncio.run(self.nostr_client.fetch_latest_snapshot())
if result:
manifest, _ = result
except Exception:
manifest = None
if manifest is not None:
stats["chunk_count"] = len(manifest.chunks)
stats["delta_since"] = manifest.delta_since
delta_count = 0
if manifest.delta_since:
try:
version = int(manifest.delta_since)
except ValueError:
version = 0
try:
deltas = asyncio.run(self.nostr_client.fetch_deltas_since(version))
delta_count = len(deltas)
except Exception:
delta_count = 0
stats["pending_deltas"] = delta_count
stats["pending_deltas"] = len(
getattr(self.nostr_client, "_delta_events", [])
)
else:
stats["chunk_count"] = 0
stats["delta_since"] = None