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 #749 from PR0M3TH3AN/codex/update-migration-handling-in-vault-and-tests

Browse Source 
Track migrations and trigger sync
This commit is contained in:
thePR0M3TH3AN
2025-08-03 20:56:29 -04:00
committed by GitHub
parent aeee3b91d9 b5024d99de
commit b40a7416ab
4 changed files with 98 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/seedpass/core/encryption.py
View File

@@ -332,6 +332,9 @@ class EncryptionManager:
if relative_path is None:
relative_path = Path("seedpass_entries_db.json.enc")
is_legacy = not encrypted_data.startswith(b"V2:")
self.last_migration_performed = False
def _process(decrypted: bytes) -> dict:
if USE_ORJSON:
data = json_lib.loads(decrypted)
@@ -361,6 +364,7 @@ class EncryptionManager:
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"))
self.last_migration_performed = is_legacy
return True
except InvalidToken as e:
try:
@@ -382,6 +386,7 @@ class EncryptionManager:
"yellow",
)
)
self.last_migration_performed = True
return True
except Exception as e2:
if strict:

49
src/seedpass/core/manager.py
View File

@@ -1158,8 +1158,7 @@ class PasswordManager:
migrated = False
try:
self.vault.load_index()
migrated = getattr(self.vault, "migrated_from_legacy", False)
_, migrated = self.vault.load_index(return_migration_flag=True)
except RuntimeError as exc:
print(colored(str(exc), "red"))
sys.exit(1)
@@ -1258,23 +1257,28 @@ class PasswordManager:
updated = False
migrated = False
if current != encrypted:
if self.vault.decrypt_and_save_index_from_nostr(
encrypted, strict=False, merge=False
):
success, mig = self.vault.decrypt_and_save_index_from_nostr(
encrypted, strict=False, merge=False, return_migration_flag=True
)
if success:
updated = True
current = encrypted
migrated = migrated or self.vault.migrated_from_legacy
migrated = migrated or mig
if manifest.delta_since:
version = int(manifest.delta_since)
deltas = await self.nostr_client.fetch_deltas_since(version)
for delta in deltas:
if current != delta:
if self.vault.decrypt_and_save_index_from_nostr(
delta, strict=False, merge=True
):
success, mig = self.vault.decrypt_and_save_index_from_nostr(
delta,
strict=False,
merge=True,
return_migration_flag=True,
)
if success:
updated = True
current = delta
migrated = migrated or self.vault.migrated_from_legacy
migrated = migrated or mig
if migrated and not getattr(self, "offline_mode", False):
self.start_background_vault_sync()
if updated:
@@ -1410,25 +1414,32 @@ class PasswordManager:
manifest, chunks = result
encrypted = gzip.decompress(b"".join(chunks))
migrated = False
success = self.vault.decrypt_and_save_index_from_nostr(
encrypted, strict=False, merge=False
success, mig = self.vault.decrypt_and_save_index_from_nostr(
encrypted,
strict=False,
merge=False,
return_migration_flag=True,
)
if success:
have_data = True
migrated = migrated or self.vault.migrated_from_legacy
migrated = migrated or mig
current = encrypted
if manifest.delta_since:
version = int(manifest.delta_since)
deltas = await self.nostr_client.fetch_deltas_since(version)
for delta in deltas:
if current != delta:
if self.vault.decrypt_and_save_index_from_nostr(
delta, strict=False, merge=True
):
current = delta
migrated = (
migrated or self.vault.migrated_from_legacy
success, mig = (
self.vault.decrypt_and_save_index_from_nostr(
delta,
strict=False,
merge=True,
return_migration_flag=True,
)
)
if success:
current = delta
migrated = migrated or mig
if migrated and not getattr(self, "offline_mode", False):
self.start_background_vault_sync()
logger.info("Initialized local database from Nostr.")

37
src/seedpass/core/vault.py
View File

@@ -32,8 +32,8 @@ class Vault:
self.encryption_manager = manager
# ----- Password index helpers -----
def load_index(self) -> dict:
"""Return decrypted password index data as a dict, applying migrations.
def load_index(self, *, return_migration_flag: bool = False):
"""Return decrypted password index data, applying migrations.
If a legacy ``seedpass_passwords_db.json.enc`` file is detected, the
user is prompted to migrate it. A backup copy of the legacy file (and
@@ -87,7 +87,7 @@ class Vault:
)
data = self.encryption_manager.load_json_data(self.index_file)
self.migrated_from_legacy = self.migrated_from_legacy or getattr(
migration_performed = getattr(
self.encryption_manager, "last_migration_performed", False
)
from .migrations import apply_migrations, LATEST_VERSION
@@ -97,7 +97,13 @@ class Vault:
raise ValueError(
f"File schema version {version} is newer than supported {LATEST_VERSION}"
)
schema_migrated = version < LATEST_VERSION
data = apply_migrations(data)
self.migrated_from_legacy = (
self.migrated_from_legacy or migration_performed or schema_migrated
)
if return_migration_flag:
return data, self.migrated_from_legacy
return data
def save_index(self, data: dict) -> None:
@@ -109,15 +115,28 @@ class Vault:
return self.encryption_manager.get_encrypted_index()
def decrypt_and_save_index_from_nostr(
self, encrypted_data: bytes, *, strict: bool = True, merge: bool = False
) -> bool:
"""Decrypt Nostr payload and update the local index."""
self.migrated_from_legacy = not encrypted_data.startswith(b"V2:")
self,
encrypted_data: bytes,
*,
strict: bool = True,
merge: bool = False,
return_migration_flag: bool = False,
):
"""Decrypt Nostr payload and update the local index.
Returns ``True``/``False`` for success by default. When
``return_migration_flag`` is ``True`` a tuple ``(success, migrated)`` is
returned, where ``migrated`` indicates whether any legacy migration
occurred.
"""
result = self.encryption_manager.decrypt_and_save_index_from_nostr(
encrypted_data, strict=strict, merge=merge
)
if not result:
self.migrated_from_legacy = False
self.migrated_from_legacy = result and getattr(
self.encryption_manager, "last_migration_performed", False
)
if return_migration_flag:
return result, self.migrated_from_legacy
return result
# ----- Config helpers -----

35
src/tests/test_legacy_migration.py
View File

@@ -159,3 +159,38 @@ def test_legacy_nostr_payload_triggers_sync(monkeypatch, tmp_path: Path):
asyncio.run(pm.sync_index_from_nostr_async())
assert calls["sync"] == 1
assert pm.vault.load_index() == data
def test_legacy_index_reinit_triggers_sync_once(monkeypatch, tmp_path: Path):
vault, enc_mgr = create_vault(tmp_path, TEST_SEED, TEST_PASSWORD)
key = derive_index_key(TEST_SEED)
data = {"schema_version": 4, "entries": {}}
enc = Fernet(key).encrypt(json.dumps(data).encode())
legacy_file = tmp_path / "seedpass_passwords_db.json.enc"
legacy_file.write_bytes(enc)
monkeypatch.setattr("builtins.input", lambda *_a, **_k: "y")
pm = PasswordManager.__new__(PasswordManager)
pm.encryption_mode = EncryptionMode.SEED_ONLY
pm.encryption_manager = enc_mgr
pm.vault = Vault(enc_mgr, tmp_path)
pm.parent_seed = TEST_SEED
pm.fingerprint_dir = tmp_path
pm.current_fingerprint = tmp_path.name
pm.bip85 = SimpleNamespace()
monkeypatch.setattr(
"seedpass.core.manager.NostrClient", lambda *a, **k: SimpleNamespace()
)
calls = {"sync": 0}
pm.start_background_vault_sync = lambda *a, **k: calls.__setitem__(
"sync", calls["sync"] + 1
)
pm.initialize_managers()
pm.initialize_managers()
assert calls["sync"] == 1