Merge pull request #55 from PR0M3TH3AN/codex/refactor-code-to-add-vault-class

Introduce Vault file layer
This commit is contained in:
thePR0M3TH3AN
2025-06-30 00:01:28 -04:00
committed by GitHub
11 changed files with 125 additions and 65 deletions

View File

@@ -19,4 +19,12 @@ except Exception as e:
logging.error(f"Failed to import ConfigManager module: {e}")
logging.error(traceback.format_exc())
__all__ = ["PasswordManager", "ConfigManager"]
try:
from .vault import Vault
logging.info("Vault module imported successfully.")
except Exception as e:
logging.error(f"Failed to import Vault module: {e}")
logging.error(traceback.format_exc())
__all__ = ["PasswordManager", "ConfigManager", "Vault"]

View File

@@ -10,7 +10,7 @@ import getpass
import bcrypt
from password_manager.encryption import EncryptionManager
from password_manager.vault import Vault
from nostr.client import DEFAULT_RELAYS as DEFAULT_NOSTR_RELAYS
logger = logging.getLogger(__name__)
@@ -21,8 +21,8 @@ class ConfigManager:
CONFIG_FILENAME = "seedpass_config.json.enc"
def __init__(self, encryption_manager: EncryptionManager, fingerprint_dir: Path):
self.encryption_manager = encryption_manager
def __init__(self, vault: Vault, fingerprint_dir: Path):
self.vault = vault
self.fingerprint_dir = fingerprint_dir
self.config_path = self.fingerprint_dir / self.CONFIG_FILENAME
@@ -39,7 +39,7 @@ class ConfigManager:
logger.info("Config file not found; returning defaults")
return {"relays": list(DEFAULT_NOSTR_RELAYS), "pin_hash": ""}
try:
data = self.encryption_manager.load_json_data(self.CONFIG_FILENAME)
data = self.vault.load_config()
if not isinstance(data, dict):
raise ValueError("Config data must be a dictionary")
# Ensure defaults for missing keys
@@ -61,7 +61,7 @@ class ConfigManager:
def save_config(self, config: dict) -> None:
"""Encrypt and save configuration."""
try:
self.encryption_manager.save_json_data(config, self.CONFIG_FILENAME)
self.vault.save_config(config)
except Exception as exc:
logger.error(f"Failed to save config: {exc}")
raise

View File

@@ -28,7 +28,7 @@ from pathlib import Path
from termcolor import colored
from password_manager.encryption import EncryptionManager
from password_manager.vault import Vault
from utils.file_lock import exclusive_lock
@@ -37,14 +37,14 @@ logger = logging.getLogger(__name__)
class EntryManager:
def __init__(self, encryption_manager: EncryptionManager, fingerprint_dir: Path):
def __init__(self, vault: Vault, fingerprint_dir: Path):
"""
Initializes the EntryManager with the EncryptionManager and fingerprint directory.
:param encryption_manager: The encryption manager instance.
:param vault: The Vault instance for file access.
:param fingerprint_dir: The directory corresponding to the fingerprint.
"""
self.encryption_manager = encryption_manager
self.vault = vault
self.fingerprint_dir = fingerprint_dir
# Use paths relative to the fingerprint directory
@@ -56,7 +56,7 @@ class EntryManager:
def _load_index(self) -> Dict[str, Any]:
if self.index_file.exists():
try:
data = self.encryption_manager.load_json_data(self.index_file)
data = self.vault.load_index()
logger.debug("Index loaded successfully.")
return data
except Exception as e:
@@ -70,7 +70,7 @@ class EntryManager:
def _save_index(self, data: Dict[str, Any]) -> None:
try:
self.encryption_manager.save_json_data(data, self.index_file)
self.vault.save_index(data)
logger.debug("Index saved successfully.")
except Exception as e:
logger.error(f"Failed to save index: {e}")
@@ -83,7 +83,7 @@ class EntryManager:
:return: The next index number as an integer.
"""
try:
data = self.encryption_manager.load_json_data(self.index_file)
data = self.vault.load_index()
if "passwords" in data and isinstance(data["passwords"], dict):
indices = [int(idx) for idx in data["passwords"].keys()]
next_index = max(indices) + 1 if indices else 0
@@ -117,7 +117,7 @@ class EntryManager:
"""
try:
index = self.get_next_index()
data = self.encryption_manager.load_json_data(self.index_file)
data = self.vault.load_index()
data["passwords"][str(index)] = {
"website": website_name,
@@ -153,19 +153,7 @@ class EntryManager:
:return: The encrypted data as bytes, or None if retrieval fails.
"""
try:
if not self.index_file.exists():
logger.error(f"Index file '{self.index_file}' does not exist.")
print(
colored(
f"Error: Index file '{self.index_file}' does not exist.", "red"
)
)
return None
with open(self.index_file, "rb") as file:
encrypted_data = file.read()
logger.debug("Encrypted index file data retrieved successfully.")
return encrypted_data
return self.vault.get_encrypted_index()
except Exception as e:
logger.error(f"Failed to retrieve encrypted index file: {e}")
logger.error(traceback.format_exc())
@@ -182,7 +170,7 @@ class EntryManager:
:return: A dictionary containing the entry details or None if not found.
"""
try:
data = self.encryption_manager.load_json_data(self.index_file)
data = self.vault.load_index()
entry = data.get("passwords", {}).get(str(index))
if entry:
@@ -217,7 +205,7 @@ class EntryManager:
:param blacklisted: (Optional) The new blacklist status.
"""
try:
data = self.encryption_manager.load_json_data(self.index_file)
data = self.vault.load_index()
entry = data.get("passwords", {}).get(str(index))
if not entry:
@@ -272,7 +260,7 @@ class EntryManager:
:return: A list of tuples containing entry details: (index, website, username, url, blacklisted)
"""
try:
data = self.encryption_manager.load_json_data()
data = self.vault.load_index()
passwords = data.get("passwords", {})
if not passwords:
@@ -316,11 +304,11 @@ class EntryManager:
:param index: The index number of the password entry to delete.
"""
try:
data = self.encryption_manager.load_json_data()
data = self.vault.load_index()
if "passwords" in data and str(index) in data["passwords"]:
del data["passwords"][str(index)]
logger.debug(f"Deleted entry at index {index}.")
self.encryption_manager.save_json_data(data)
self.vault.save_index(data)
self.update_checksum()
self.backup_index_file()
logger.info(f"Entry at index {index} deleted successfully.")
@@ -352,7 +340,7 @@ class EntryManager:
Updates the checksum file for the password database to ensure data integrity.
"""
try:
data = self.encryption_manager.load_json_data(self.index_file)
data = self.vault.load_index()
json_content = json.dumps(data, indent=4)
checksum = hashlib.sha256(json_content.encode("utf-8")).hexdigest()
@@ -470,15 +458,15 @@ class EntryManager:
# Example usage (this part should be removed or commented out when integrating into the larger application)
if __name__ == "__main__":
from password_manager.encryption import (
EncryptionManager,
) # Ensure this import is correct based on your project structure
from password_manager.encryption import EncryptionManager
from password_manager.vault import Vault
# Initialize EncryptionManager with a dummy key for demonstration purposes
# Replace 'your-fernet-key' with your actual Fernet key
try:
dummy_key = Fernet.generate_key()
encryption_manager = EncryptionManager(dummy_key)
encryption_manager = EncryptionManager(dummy_key, Path("."))
vault = Vault(encryption_manager, Path("."))
except Exception as e:
logger.error(f"Failed to initialize EncryptionManager: {e}")
print(colored(f"Error: Failed to initialize EncryptionManager: {e}", "red"))
@@ -486,7 +474,7 @@ if __name__ == "__main__":
# Initialize EntryManager
try:
entry_manager = EntryManager(encryption_manager)
entry_manager = EntryManager(vault, Path("."))
except Exception as e:
logger.error(f"Failed to initialize EntryManager: {e}")
print(colored(f"Error: Failed to initialize EntryManager: {e}", "red"))

View File

@@ -22,6 +22,7 @@ from password_manager.encryption import EncryptionManager
from password_manager.entry_management import EntryManager
from password_manager.password_generation import PasswordGenerator
from password_manager.backup import BackupManager
from password_manager.vault import Vault
from utils.key_derivation import derive_key_from_parent_seed, derive_key_from_password
from utils.checksum import calculate_checksum, verify_checksum
from utils.password_prompt import (
@@ -75,6 +76,7 @@ class PasswordManager:
self.entry_manager: Optional[EntryManager] = None
self.password_generator: Optional[PasswordGenerator] = None
self.backup_manager: Optional[BackupManager] = None
self.vault: Optional[Vault] = None
self.fingerprint_manager: Optional[FingerprintManager] = None
self.parent_seed: Optional[str] = None
self.bip85: Optional[BIP85] = None
@@ -228,6 +230,7 @@ class PasswordManager:
# Derive key from password
key = derive_key_from_password(password)
self.encryption_manager = EncryptionManager(key, fingerprint_dir)
self.vault = Vault(self.encryption_manager, fingerprint_dir)
logger.debug(
"EncryptionManager set up successfully for selected fingerprint."
)
@@ -386,6 +389,7 @@ class PasswordManager:
# Initialize EncryptionManager with key and fingerprint_dir
self.encryption_manager = EncryptionManager(key, fingerprint_dir)
self.vault = Vault(self.encryption_manager, fingerprint_dir)
self.parent_seed = self.encryption_manager.decrypt_parent_seed()
# Log the type and content of parent_seed
@@ -469,6 +473,7 @@ class PasswordManager:
password = prompt_for_password()
key = derive_key_from_password(password)
self.encryption_manager = EncryptionManager(key, fingerprint_dir)
self.vault = Vault(self.encryption_manager, fingerprint_dir)
# Encrypt and save the parent seed
self.encryption_manager.encrypt_parent_seed(parent_seed)
@@ -650,7 +655,7 @@ class PasswordManager:
# Reinitialize the managers with the updated EncryptionManager and current fingerprint context
self.entry_manager = EntryManager(
encryption_manager=self.encryption_manager,
vault=self.vault,
fingerprint_dir=self.fingerprint_dir,
)
@@ -664,7 +669,7 @@ class PasswordManager:
# Load relay configuration and initialize NostrClient
self.config_manager = ConfigManager(
encryption_manager=self.encryption_manager,
vault=self.vault,
fingerprint_dir=self.fingerprint_dir,
)
config = self.config_manager.load_config()
@@ -692,7 +697,7 @@ class PasswordManager:
try:
encrypted = self.nostr_client.retrieve_json_from_nostr_sync()
if encrypted:
self.encryption_manager.decrypt_and_save_index_from_nostr(encrypted)
self.vault.decrypt_and_save_index_from_nostr(encrypted)
logger.info("Initialized local database from Nostr.")
except Exception as e:
logger.warning(f"Unable to sync index from Nostr: {e}")
@@ -990,7 +995,7 @@ class PasswordManager:
:return: The encrypted data as bytes, or None if retrieval fails.
"""
try:
encrypted_data = self.entry_manager.get_encrypted_index()
encrypted_data = self.vault.get_encrypted_index()
if encrypted_data:
logging.debug("Encrypted index data retrieved successfully.")
return encrypted_data
@@ -1011,14 +1016,7 @@ class PasswordManager:
:param encrypted_data: The encrypted data retrieved from Nostr.
"""
try:
# Decrypt the data using EncryptionManager's decrypt_data method
decrypted_data = self.encryption_manager.decrypt_data(encrypted_data)
# Save the decrypted data to the index file
index_file_path = self.fingerprint_dir / "seedpass_passwords_db.json.enc"
with open(index_file_path, "wb") as f:
f.write(decrypted_data)
self.vault.decrypt_and_save_index_from_nostr(encrypted_data)
logging.info("Index file updated from Nostr successfully.")
print(colored("Index file updated from Nostr successfully.", "green"))
except Exception as e:
@@ -1223,7 +1221,7 @@ class PasswordManager:
new_password = prompt_for_password()
# Load data with existing encryption manager
index_data = self.entry_manager.encryption_manager.load_json_data()
index_data = self.vault.load_index()
config_data = self.config_manager.load_config(require_pin=False)
# Create a new encryption manager with the new password
@@ -1232,13 +1230,13 @@ class PasswordManager:
# Re-encrypt sensitive files using the new manager
new_enc_mgr.encrypt_parent_seed(self.parent_seed)
new_enc_mgr.save_json_data(index_data)
self.config_manager.encryption_manager = new_enc_mgr
self.vault.set_encryption_manager(new_enc_mgr)
self.vault.save_index(index_data)
self.config_manager.vault = self.vault
self.config_manager.save_config(config_data)
# Update hashed password and replace managers
self.encryption_manager = new_enc_mgr
self.entry_manager.encryption_manager = new_enc_mgr
self.password_generator.encryption_manager = new_enc_mgr
self.store_hashed_password(new_password)

View File

@@ -0,0 +1,49 @@
"""Vault utilities for reading and writing encrypted files."""
from pathlib import Path
from typing import Optional
from .encryption import EncryptionManager
class Vault:
"""Simple wrapper around :class:`EncryptionManager` for vault storage."""
INDEX_FILENAME = "seedpass_passwords_db.json.enc"
CONFIG_FILENAME = "seedpass_config.json.enc"
def __init__(self, encryption_manager: EncryptionManager, fingerprint_dir: Path):
self.encryption_manager = encryption_manager
self.fingerprint_dir = fingerprint_dir
self.index_file = self.fingerprint_dir / self.INDEX_FILENAME
self.config_file = self.fingerprint_dir / self.CONFIG_FILENAME
def set_encryption_manager(self, manager: EncryptionManager) -> None:
"""Replace the internal encryption manager."""
self.encryption_manager = manager
# ----- Password index helpers -----
def load_index(self) -> dict:
"""Return decrypted password index data as a dict."""
return self.encryption_manager.load_json_data(self.index_file)
def save_index(self, data: dict) -> None:
"""Encrypt and write password index."""
self.encryption_manager.save_json_data(data, self.index_file)
def get_encrypted_index(self) -> Optional[bytes]:
"""Return the encrypted index bytes if present."""
return self.encryption_manager.get_encrypted_index()
def decrypt_and_save_index_from_nostr(self, encrypted_data: bytes) -> None:
"""Decrypt Nostr payload and overwrite the local index."""
self.encryption_manager.decrypt_and_save_index_from_nostr(encrypted_data)
# ----- Config helpers -----
def load_config(self) -> dict:
"""Load decrypted configuration."""
return self.encryption_manager.load_json_data(self.config_file)
def save_config(self, config: dict) -> None:
"""Encrypt and persist configuration."""
self.encryption_manager.save_json_data(config, self.config_file)

View File

@@ -9,6 +9,7 @@ sys.path.append(str(Path(__file__).resolve().parents[1]))
from password_manager.encryption import EncryptionManager
from password_manager.config_manager import ConfigManager
from password_manager.vault import Vault
from nostr.client import DEFAULT_RELAYS
@@ -16,7 +17,8 @@ def test_config_defaults_and_round_trip():
with TemporaryDirectory() as tmpdir:
key = Fernet.generate_key()
enc_mgr = EncryptionManager(key, Path(tmpdir))
cfg_mgr = ConfigManager(enc_mgr, Path(tmpdir))
vault = Vault(enc_mgr, Path(tmpdir))
cfg_mgr = ConfigManager(vault, Path(tmpdir))
cfg = cfg_mgr.load_config(require_pin=False)
assert cfg["relays"] == list(DEFAULT_RELAYS)
@@ -34,7 +36,8 @@ def test_pin_verification_and_change():
with TemporaryDirectory() as tmpdir:
key = Fernet.generate_key()
enc_mgr = EncryptionManager(key, Path(tmpdir))
cfg_mgr = ConfigManager(enc_mgr, Path(tmpdir))
vault = Vault(enc_mgr, Path(tmpdir))
cfg_mgr = ConfigManager(vault, Path(tmpdir))
cfg_mgr.set_pin("1234")
assert cfg_mgr.verify_pin("1234")
@@ -50,7 +53,8 @@ def test_config_file_encrypted_after_save():
with TemporaryDirectory() as tmpdir:
key = Fernet.generate_key()
enc_mgr = EncryptionManager(key, Path(tmpdir))
cfg_mgr = ConfigManager(enc_mgr, Path(tmpdir))
vault = Vault(enc_mgr, Path(tmpdir))
cfg_mgr = ConfigManager(vault, Path(tmpdir))
data = {"relays": ["wss://r"], "pin_hash": ""}
cfg_mgr.save_config(data)
@@ -67,7 +71,8 @@ def test_set_relays_persists_changes():
with TemporaryDirectory() as tmpdir:
key = Fernet.generate_key()
enc_mgr = EncryptionManager(key, Path(tmpdir))
cfg_mgr = ConfigManager(enc_mgr, Path(tmpdir))
vault = Vault(enc_mgr, Path(tmpdir))
cfg_mgr = ConfigManager(vault, Path(tmpdir))
cfg_mgr.set_relays(["wss://custom"], require_pin=False)
cfg = cfg_mgr.load_config(require_pin=False)
assert cfg["relays"] == ["wss://custom"]
@@ -77,6 +82,7 @@ def test_set_relays_requires_at_least_one():
with TemporaryDirectory() as tmpdir:
key = Fernet.generate_key()
enc_mgr = EncryptionManager(key, Path(tmpdir))
cfg_mgr = ConfigManager(enc_mgr, Path(tmpdir))
vault = Vault(enc_mgr, Path(tmpdir))
cfg_mgr = ConfigManager(vault, Path(tmpdir))
with pytest.raises(ValueError):
cfg_mgr.set_relays([], require_pin=False)

View File

@@ -11,6 +11,7 @@ sys.path.append(str(Path(__file__).resolve().parents[1]))
from password_manager.encryption import EncryptionManager
from password_manager.entry_management import EntryManager
from password_manager.config_manager import ConfigManager
from password_manager.vault import Vault
from password_manager.manager import PasswordManager
@@ -18,13 +19,15 @@ def test_change_password_does_not_trigger_nostr_backup(monkeypatch):
with TemporaryDirectory() as tmpdir:
fp = Path(tmpdir)
enc_mgr = EncryptionManager(Fernet.generate_key(), fp)
entry_mgr = EntryManager(enc_mgr, fp)
cfg_mgr = ConfigManager(enc_mgr, fp)
vault = Vault(enc_mgr, fp)
entry_mgr = EntryManager(vault, fp)
cfg_mgr = ConfigManager(vault, fp)
pm = PasswordManager.__new__(PasswordManager)
pm.encryption_manager = enc_mgr
pm.entry_manager = entry_mgr
pm.config_manager = cfg_mgr
pm.vault = vault
pm.password_generator = SimpleNamespace(encryption_manager=enc_mgr)
pm.fingerprint_dir = fp
pm.current_fingerprint = "fp"

View File

@@ -13,6 +13,7 @@ import main
from nostr.client import DEFAULT_RELAYS
from password_manager.encryption import EncryptionManager
from password_manager.config_manager import ConfigManager
from password_manager.vault import Vault
from utils.fingerprint_manager import FingerprintManager
@@ -26,7 +27,8 @@ def setup_pm(tmp_path, monkeypatch):
fp_dir = constants.APP_DIR / "fp"
fp_dir.mkdir(parents=True)
enc_mgr = EncryptionManager(Fernet.generate_key(), fp_dir)
cfg_mgr = ConfigManager(enc_mgr, fp_dir)
vault = Vault(enc_mgr, fp_dir)
cfg_mgr = ConfigManager(vault, fp_dir)
fp_mgr = FingerprintManager(constants.APP_DIR)
nostr_stub = SimpleNamespace(

View File

@@ -7,13 +7,15 @@ sys.path.append(str(Path(__file__).resolve().parents[1]))
from password_manager.encryption import EncryptionManager
from password_manager.entry_management import EntryManager
from password_manager.vault import Vault
def test_list_entries_empty():
with TemporaryDirectory() as tmpdir:
key = Fernet.generate_key()
enc_mgr = EncryptionManager(key, Path(tmpdir))
entry_mgr = EntryManager(enc_mgr, Path(tmpdir))
vault = Vault(enc_mgr, Path(tmpdir))
entry_mgr = EntryManager(vault, Path(tmpdir))
entries = entry_mgr.list_entries()
assert entries == []

View File

@@ -7,13 +7,15 @@ sys.path.append(str(Path(__file__).resolve().parents[1]))
from password_manager.encryption import EncryptionManager
from password_manager.entry_management import EntryManager
from password_manager.vault import Vault
def test_add_and_retrieve_entry():
with TemporaryDirectory() as tmpdir:
key = Fernet.generate_key()
enc_mgr = EncryptionManager(key, Path(tmpdir))
entry_mgr = EntryManager(enc_mgr, Path(tmpdir))
vault = Vault(enc_mgr, Path(tmpdir))
entry_mgr = EntryManager(vault, Path(tmpdir))
index = entry_mgr.add_entry("example.com", 12, "user")
entry = entry_mgr.retrieve_entry(index)

View File

@@ -8,6 +8,7 @@ sys.path.append(str(Path(__file__).resolve().parents[1]))
from password_manager.encryption import EncryptionManager
from password_manager.entry_management import EntryManager
from password_manager.vault import Vault
from nostr.client import NostrClient
@@ -16,7 +17,8 @@ def test_backup_and_publish_to_nostr():
tmp_path = Path(tmpdir)
key = Fernet.generate_key()
enc_mgr = EncryptionManager(key, tmp_path)
entry_mgr = EntryManager(enc_mgr, tmp_path)
vault = Vault(enc_mgr, tmp_path)
entry_mgr = EntryManager(vault, tmp_path)
# create an index by adding an entry
entry_mgr.add_entry("example.com", 12)