mirror of
https://github.com/PR0M3TH3AN/SeedPass.git
synced 2025-09-08 07:18:47 +00:00
22
.github/workflows/python-ci.yml
vendored
Normal file
22
.github/workflows/python-ci.yml
vendored
Normal file
@@ -0,0 +1,22 @@
|
||||
name: CI
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ "main" ]
|
||||
pull_request:
|
||||
branches: [ "main" ]
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- uses: actions/setup-python@v4
|
||||
with:
|
||||
python-version: '3.11'
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install -r src/requirements.txt
|
||||
- name: Test with pytest
|
||||
run: pytest -q src/tests
|
2
.gitignore
vendored
2
.gitignore
vendored
@@ -24,4 +24,4 @@ Thumbs.db
|
||||
|
||||
# Python env
|
||||
.env
|
||||
*.env
|
||||
*.env
|
||||
|
40
AGENTS.md
Normal file
40
AGENTS.md
Normal file
@@ -0,0 +1,40 @@
|
||||
# Repository Guidelines
|
||||
|
||||
This project is written in **Python**. Follow these instructions when working with the code base.
|
||||
|
||||
## Running Tests
|
||||
|
||||
1. Set up a virtual environment and install dependencies:
|
||||
|
||||
```bash
|
||||
python3 -m venv venv
|
||||
source venv/bin/activate
|
||||
pip install -r src/requirements.txt
|
||||
```
|
||||
|
||||
2. Run the test suite using **pytest**:
|
||||
|
||||
```bash
|
||||
pytest
|
||||
```
|
||||
|
||||
Currently the test folder is located in `src/tests/`. New tests should be placed there so `pytest` can discover them automatically.
|
||||
|
||||
## Style Guidelines
|
||||
|
||||
- Adhere to **PEP 8** conventions (4‑space indentation, descriptive names, docstrings).
|
||||
- Use [**black**](https://black.readthedocs.io/) to format Python files before committing:
|
||||
|
||||
```bash
|
||||
black .
|
||||
```
|
||||
|
||||
- Optionally run **flake8** or another linter to catch style issues.
|
||||
|
||||
## Security Practices
|
||||
|
||||
- Never commit seed phrases, passwords, private keys, or other sensitive data.
|
||||
- Use environment variables or local configuration files (ignored by Git) for secrets.
|
||||
- Review code for potential information leaks (e.g., verbose logging) before submitting.
|
||||
|
||||
Following these practices helps keep the code base consistent and secure.
|
52
README.md
52
README.md
@@ -58,7 +58,7 @@ git clone https://github.com/PR0M3TH3AN/SeedPass.git
|
||||
Navigate to the project directory:
|
||||
|
||||
```bash
|
||||
cd SeedPass/src
|
||||
cd SeedPass
|
||||
```
|
||||
|
||||
### 2. Create a Virtual Environment
|
||||
@@ -93,7 +93,7 @@ Install the required Python packages and build dependencies using `pip`:
|
||||
|
||||
```bash
|
||||
pip install --upgrade pip
|
||||
pip install -r requirements.txt
|
||||
pip install -r src/requirements.txt
|
||||
```
|
||||
|
||||
## Usage
|
||||
@@ -101,16 +101,16 @@ pip install -r requirements.txt
|
||||
After successfully installing the dependencies, you can run SeedPass using the following command:
|
||||
|
||||
```bash
|
||||
python main.py
|
||||
python src/main.py
|
||||
```
|
||||
|
||||
### Running the Application
|
||||
|
||||
1. **Start the Application:**
|
||||
|
||||
```bash
|
||||
python main.py
|
||||
```
|
||||
```bash
|
||||
python src/main.py
|
||||
```
|
||||
|
||||
2. **Follow the Prompts:**
|
||||
|
||||
@@ -134,9 +134,10 @@ python main.py
|
||||
10. Add a New Seed Profile
|
||||
11. Remove an Existing Seed Profile
|
||||
12. List All Seed Profiles
|
||||
13. Exit
|
||||
13. Settings
|
||||
14. Exit
|
||||
|
||||
Enter your choice (1-13):
|
||||
Enter your choice (1-14):
|
||||
```
|
||||
|
||||
### Managing Multiple Seeds
|
||||
@@ -159,11 +160,41 @@ SeedPass allows you to manage multiple seed profiles (previously referred to as
|
||||
|
||||
**Note:** The term "seed profile" is used to represent different sets of seeds you can manage within SeedPass. This provides an intuitive way to handle multiple identities or sets of passwords.
|
||||
|
||||
### Configuration File and Settings
|
||||
|
||||
SeedPass keeps per-profile settings in an encrypted file named `seedpass_config.json.enc` inside each profile directory under `~/.seedpass/`. This file stores your chosen Nostr relays and the optional settings PIN. New profiles start with the following default relays:
|
||||
|
||||
```
|
||||
wss://relay.snort.social
|
||||
wss://nostr.oxtr.dev
|
||||
wss://relay.primal.net
|
||||
```
|
||||
|
||||
You can manage the relay list or change the PIN through the **Settings** menu:
|
||||
|
||||
1. From the main menu, choose option `13` (**Settings**).
|
||||
2. Select `1` to view your current relays.
|
||||
3. Choose `2` to add a new relay URL.
|
||||
4. Select `3` to remove a relay by number.
|
||||
5. Choose `4` to reset to the default relay list.
|
||||
6. Select `5` to change the settings PIN.
|
||||
7. Choose `6` to return to the main menu.
|
||||
|
||||
## Running Tests
|
||||
|
||||
SeedPass includes a small suite of unit tests. After activating your virtual environment and installing dependencies, run the tests with **pytest**:
|
||||
|
||||
```bash
|
||||
pip install -r src/requirements.txt
|
||||
pytest
|
||||
```
|
||||
|
||||
## Security Considerations
|
||||
|
||||
**Important:** The password you use to encrypt your parent seed is also required to decrypt the seed index data retrieved from Nostr. **It is imperative to remember this password** and be sure to use it with the same seed, as losing it means you won't be able to access your stored index. Secure your 12-word seed **and** your master password.
|
||||
|
||||
- **Backup Your Data:** Regularly back up your encrypted data and checksum files to prevent data loss.
|
||||
- **Backup the Settings PIN:** Your settings PIN is stored in the encrypted configuration file. Keep a copy of this file or remember the PIN, as losing it will require deleting the file and reconfiguring your relays.
|
||||
- **Protect Your Passwords:** Do not share your master password or seed phrases with anyone and ensure they are strong and unique.
|
||||
- **Checksum Verification:** Always verify the script's checksum to ensure its integrity and protect against unauthorized modifications.
|
||||
- **Potential Bugs and Limitations:** Be aware that the software may contain bugs and lacks certain features. The maximum size of the password index before encountering issues with Nostr backups is unknown. Additionally, the security of memory management and logs has not been thoroughly evaluated and may pose risks of leaking sensitive information.
|
||||
@@ -228,8 +259,9 @@ The SeedPass roadmap outlines a structured development plan divided into distinc
|
||||
- **Implementation Steps:**
|
||||
- Create a `config.yaml` or `config.json` file in the SeedPass data directory.
|
||||
- Define a structure to store user configurations, starting with a list of Nostr relay URLs.
|
||||
- Allow users to add, remove, and manage an unlimited number of Nostr relays through the CLI or configuration file.
|
||||
- Ensure the configuration file is securely stored and encrypted if necessary.
|
||||
- Allow users to add, remove, and manage an unlimited number of Nostr relays through the CLI or configuration file.
|
||||
- Ensure the configuration file is securely stored and encrypted if necessary.
|
||||
- The Nostr client loads its relay list from this encrypted file. New accounts start with the default relays until you update the settings.
|
||||
|
||||
2. **Individual JSON File Management**
|
||||
- **Separate Entry Files:**
|
||||
|
@@ -418,6 +418,13 @@ seedpass show-pubkey
|
||||
|
||||
**Description:**
|
||||
Allows users to specify custom Nostr relays for publishing their encrypted backup index, providing flexibility and control over data distribution.
|
||||
Relay URLs are stored in an encrypted configuration file located in `~/.seedpass/<fingerprint>/seedpass_config.json.enc` and loaded each time the Nostr client starts. New accounts use the following default relays until changed:
|
||||
|
||||
```
|
||||
wss://relay.snort.social
|
||||
wss://nostr.oxtr.dev
|
||||
wss://relay.primal.net
|
||||
```
|
||||
|
||||
**Usage Example:**
|
||||
```bash
|
||||
|
@@ -1,138 +1,78 @@
|
||||
Okay, acknowledging the strict requirement that **exported data must remain encrypted and ultimately depend on the master seed/password for decryption**, here is a prioritized feature list to-do:
|
||||
---
|
||||
|
||||
# SeedPass Feature Back‑Log (v2)
|
||||
|
||||
> **Encryption invariant** Everything at rest **and** in export remains cipher‑text that ultimately derives from the **profile master‑password + parent seed**. No unencrypted payload leaves the vault.
|
||||
>
|
||||
> **Surface rule** UI layers (CLI, GUI, future mobile) may *display* decrypted data **after** user unlock, but must never write plaintext to disk or network.
|
||||
|
||||
---
|
||||
|
||||
## SeedPass Feature To-Do List
|
||||
## Track vocabulary
|
||||
|
||||
**Key Constraint:** All data storage and export mechanisms must ensure data remains encrypted. Access to usable, decrypted information must always require the user's Master Password for the specific profile (which in turn decrypts the Parent Seed or the necessary keys derived from it). *Plaintext export for migration to other tools is explicitly excluded by this constraint.*
|
||||
| Label | Meaning |
|
||||
| ------------ | ------------------------------------------------------------------------------ |
|
||||
| **Core API** | `seedpass.api` – headless services consumed by CLI / GUI |
|
||||
| **Profile** | A fingerprint‑scoped vault: parent‑seed + hashed pw + entries |
|
||||
| **Entry** | One encrypted JSON blob on disk *and* one replaceable Nostr event (kind 31111) |
|
||||
| **GUI MVP** | Desktop app built with PySide 6 announced in the v2 roadmap |
|
||||
|
||||
---
|
||||
|
||||
### **Phase 1: High Priority (Core Usability & Control)**
|
||||
## Phase A • Core‑level enhancements (blockers for GUI)
|
||||
|
||||
1. **Search Functionality (Encrypted Search)**
|
||||
* **Goal:** Allow users to quickly find specific entries without manually listing all of them.
|
||||
* **Key Implementation Steps:**
|
||||
* Add a "Search Entries" option to the main CLI menu.
|
||||
* Implement search logic in `PasswordManager`:
|
||||
* Iterate through all local entry files (`entry_manager.list_all_entry_nums()` -> `entry_manager.load_entry()`).
|
||||
* For each entry, decrypt *only the necessary searchable fields* defined per `kind` (e.g., 'title', 'username', 'url', 'tags'). **Do NOT decrypt passwords/secrets for searching.**
|
||||
* Perform case-insensitive substring matching on the decrypted searchable fields against the user's query.
|
||||
* Display a list of matching entries (e.g., `EntryNum: Title (Kind)`).
|
||||
* Allow the user to select a search result to view its full details (triggering the appropriate handler which *will* decrypt sensitive data for display only).
|
||||
* **Encryption Consideration:** Only non-secret metadata fields are decrypted *during the search process*. Sensitive data remains encrypted until explicitly requested for display via the entry handler.
|
||||
* **Priority:** High
|
||||
| Prio | Feature | Notes |
|
||||
| ------ | ---------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| 🔥 | **Encrypted Search API** | • `VaultService.search(query:str, *, kinds=None) -> List[EntryMeta]` <br>• Decrypt *only* whitelisted meta‑fields per `kind` (title, username, url, tags) for in‑memory matching. |
|
||||
| 🔥 | **Rich Listing / Sort / Filter** | • `list_entries(sort_by="updated", kind="note")` <br>• Sorting by `title` must decrypt that field on‑the‑fly. |
|
||||
| 🔥 | **Custom Relay Set (per profile)** | • `StateManager.state["relays"]: List[str]` <br>• CRUD CLI commands & GUI dialog. <br>• `NostrClient` reads from state at instantiation. |
|
||||
| ⚡ | **Session Lock & Idle Timeout** | • Config `SESSION_TIMEOUT` (default 15 min). <br>• `AuthGuard` clears in‑memory keys & seeds. <br>• CLI `seedpass lock` + GUI menu “Lock vault”. |
|
||||
|
||||
2. **Custom Nostr Relays (Per Profile)**
|
||||
* **Goal:** Allow users to specify which Nostr relays to use for synchronization for each profile, enhancing reliability and privacy.
|
||||
* **Key Implementation Steps:**
|
||||
* Modify `StateManager` to load/save a `relays: List[str]` field in `seedpass_state.json`. Default to `constants.DEFAULT_RELAYS` if not present.
|
||||
* Add options to the "Manage Profiles" sub-menu:
|
||||
* `View Relays`: Display current relays for the active profile.
|
||||
* `Add Relay`: Prompt user for a relay URL and add it to the list.
|
||||
* `Remove Relay`: Display current relays with numbers, prompt user to select one for removal.
|
||||
* `Set Default Relays`: Reset the list to `constants.DEFAULT_RELAYS`.
|
||||
* Update `NostrClient.__init__` to read the relay list from `StateManager` for the current profile.
|
||||
* Ensure `StateManager._save_state()` is called after modifications.
|
||||
* **Encryption Consideration:** Relay list itself is not sensitive and stored in plaintext within the profile's state file.
|
||||
* **Priority:** High
|
||||
|
||||
3. **Entry Listing with Sorting & Filtering**
|
||||
* **Goal:** Provide more organized ways to view local entries beyond just retrieving a single one by number.
|
||||
* **Key Implementation Steps:**
|
||||
* Enhance the "List / Retrieve Entries" option or create a dedicated "List Entries" option.
|
||||
* Load all local entries (`password_manager.list_all_entries()`).
|
||||
* Add sub-prompts or flags for:
|
||||
* **Sorting:** By Entry Number (default), Title (requires decrypting 'title'), Kind, Last Updated Timestamp.
|
||||
* **Filtering:** By Kind (`--kind note`).
|
||||
* Implement the sorting logic (decrypting 'title' in memory only for sorting purposes).
|
||||
* Implement the filtering logic.
|
||||
* Display the formatted, sorted, and/or filtered list.
|
||||
* **Encryption Consideration:** Only the 'title' field needs temporary in-memory decryption for sorting by title. All other data remains encrypted until an entry is selected for full display.
|
||||
* **Priority:** High
|
||||
**Exit‑criteria** : All functions green in CI, consumed by both CLI (Typer) *and* a minimal Qt test harness.
|
||||
|
||||
---
|
||||
|
||||
### **Phase 2: Medium Priority (Data Management - Securely)**
|
||||
## Phase B • Data Portability (encrypted only)
|
||||
|
||||
4. **Secure Data Export (Profile Backup)**
|
||||
* **Goal:** Allow users to create a single, encrypted backup file containing *all* entries for a specific profile, suitable for transferring or archiving *within the SeedPass ecosystem*.
|
||||
* **Key Implementation Steps:**
|
||||
* Add an "Export Profile Data" option (requires password confirmation).
|
||||
* Prompt for an output filename (e.g., `seedpass_profile_<fingerprint>_export.json.enc`).
|
||||
* Load all local entries for the current profile.
|
||||
* Construct a JSON object containing a list of all *un-decrypted* (as loaded from disk) entry data structures. Include metadata like export date and profile fingerprint.
|
||||
* Convert this JSON object to bytes.
|
||||
* **Crucially:** Encrypt this *entire byte stream* using the profile's `EncryptionManager` (i.e., using the key derived from the master password).
|
||||
* Save the resulting encrypted blob to the user-specified file.
|
||||
* **Encryption Consideration:** The entire export is a single encrypted blob. It requires the *exact same* SeedPass profile (same seed + master password) to decrypt and import it later. It is **not** interoperable with other tools. This adheres to the "no plaintext export" rule.
|
||||
* **Priority:** Medium
|
||||
|
||||
5. **Secure Data Import (Profile Restore/Merge)**
|
||||
* **Goal:** Allow users to import entries from a previously created secure export file.
|
||||
* **Key Implementation Steps:**
|
||||
* Add an "Import Profile Data" option (requires password confirmation).
|
||||
* Prompt for the path to the encrypted export file (`.json.enc`).
|
||||
* Use the current profile's `EncryptionManager` to decrypt the entire file blob.
|
||||
* Parse the decrypted JSON to get the list of exported entries.
|
||||
* **Crucially:** Verify the fingerprint inside the imported data matches the current profile's fingerprint. Abort if mismatched.
|
||||
* Iterate through the imported entries:
|
||||
* For each imported entry, check if an entry with the same `entry_num` already exists locally.
|
||||
* **Conflict Strategy:** Decide how to handle conflicts (e.g., skip import, overwrite local if import is newer based on timestamp, prompt user). Prompting is safest but less automated. Start with "skip if exists" or "overwrite if newer".
|
||||
* If importing (either new or overwriting):
|
||||
* Validate the `kind` and structure.
|
||||
* Save the encrypted entry data (as provided in the import file) locally using `entry_manager.save_entry()`.
|
||||
* Optionally, post the imported/updated entry to Nostr.
|
||||
* **Encryption Consideration:** Import only works if the current profile's master password can decrypt the export file. Fingerprint matching prevents accidental cross-profile imports. Data remains encrypted until processed by `save_entry`.
|
||||
* **Priority:** Medium
|
||||
| Prio | Feature | Notes | |
|
||||
| ------ | ------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------- |
|
||||
| ⭐ | **Encrypted Profile Export** | • CLI `seedpass export --out myprofile.enc` <br>• Serialise *encrypted* entry files → single JSON wrapper → `EncryptionManager.encrypt_data()` <br>• Always require active profile unlock. | |
|
||||
| ⭐ | **Encrypted Profile Import / Merge** | • CLI \`seedpass import myprofile.enc \[--strategy skip | overwrite-newer]` <br>• Verify fingerprint match before ingest. <br>• Conflict policy pluggable; default `skip\`. |
|
||||
|
||||
---
|
||||
|
||||
### **Phase 3: Lower Priority (Convenience & Advanced)**
|
||||
## Phase C • Advanced secrets & sync
|
||||
|
||||
6. **Session Lock / Auto-Timeout**
|
||||
* **Goal:** Enhance security by requiring password re-entry after inactivity or manual locking.
|
||||
* **Key Implementation Steps:**
|
||||
* Track `last_activity_time` within `PasswordManager`. Update it on each successful user action.
|
||||
* Add a configurable `SESSION_TIMEOUT` constant (e.g., 900 seconds for 15 mins).
|
||||
* Before executing sensitive operations (anything requiring decryption/generation), check if `time.time() - last_activity_time > SESSION_TIMEOUT`.
|
||||
* If timed out, clear sensitive in-memory data (`encryption_manager.key = None`, `encryption_manager.fernet = None`, `parent_seed = None`, `bip85 = None`) and prompt for the master password again using `prompt_existing_password`. Re-initialize the necessary components upon success.
|
||||
* Add a "Lock Session" menu option that immediately clears sensitive data and forces password re-entry on the next action.
|
||||
* **Encryption Consideration:** Focuses on clearing in-memory keys/seeds, not on-disk encryption which remains unchanged.
|
||||
* **Priority:** Low
|
||||
|
||||
7. **Additional `Kind` Types (e.g., TOTP)**
|
||||
* **Goal:** Extend SeedPass to securely manage other types of secrets like Time-based One-Time Passwords.
|
||||
* **Key Implementation Steps:**
|
||||
* Define `kind = "totp_secret"` in `kinds.py` with fields like `title`, `issuer`, `username`, `secret_key`.
|
||||
* Ensure `secret_key` is encrypted/base64'd within the `data` payload like other sensitive fields (`stored_password`, `note` content).
|
||||
* Create `handlers/totp_secret_handler.py`.
|
||||
* The handler should decrypt the `secret_key`.
|
||||
* **Decision:** Should it *display* the secret key, or *generate* the current code? Generating is more useful but adds a dependency (`pyotp`) and time sensitivity.
|
||||
* If generating codes: Add `pyotp` to `requirements.txt`. The handler uses `pyotp.TOTP(decrypted_secret_key).now()`. Display the code along with other metadata.
|
||||
* **Encryption Consideration:** The TOTP secret key itself is stored encrypted. Generating the code requires decrypting it in memory temporarily within the handler.
|
||||
* **Priority:** Low
|
||||
|
||||
8. **Enhanced Sync Conflict Resolution (Manual Prompt)**
|
||||
* **Goal:** Provide user control when a sync detects that an entry was modified both locally and on Nostr since the last sync.
|
||||
* **Key Implementation Steps:**
|
||||
* In `PasswordManager.synchronize_with_nostr`: When `local_entry_path.exists()` and `local_checksum != remote_checksum`:
|
||||
* Load the local entry's full data and timestamp.
|
||||
* Compare the `updated_at` timestamp from the local entry's metadata with the `created_at` timestamp of the Nostr event.
|
||||
* If timestamps differ significantly *and* checksums differ, flag as a conflict.
|
||||
* Prompt the user: "Conflict detected for Entry X ('Title'). Keep Local version (updated Y) or Remote version (updated Z)? (L/R/Skip)".
|
||||
* Based on user input, either save the remote version (as currently done), skip the update for this entry, or do nothing (keep local).
|
||||
* **Encryption Consideration:** Requires decrypting local entry metadata (`updated_at`) and comparing with Nostr event metadata (`created_at`). Sensitive data decryption only happens if the user chooses to view details or keep a specific version.
|
||||
* **Priority:** Low
|
||||
| Prio | Feature | Notes |
|
||||
| ------ | ---------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| ◇ | **TOTP entry kind** | • `kind="totp_secret"` fields: title, issuer, username, secret\_key <br>• `secret_key` encrypted; handler uses `pyotp` to show current code. |
|
||||
| ◇ | **Manual Conflict Resolver** | • When `checksum` mismatch *and* both sides newer than last sync → prompt user (CLI) or modal (GUI). |
|
||||
|
||||
---
|
||||
|
||||
### **Phase 4: Future / Major Effort**
|
||||
## Phase D • Desktop GUI MVP (Qt 6)
|
||||
|
||||
9. **GUI / TUI Implementation**
|
||||
* **Goal:** Provide a more user-friendly interface than the current CLI menu system.
|
||||
* **Key Implementation Steps:** Requires selecting a framework (`curses`, `Tkinter`, `PyQt`, etc.) and redesigning the entire user interaction flow. Major undertaking.
|
||||
* **Encryption Consideration:** No change to the core encryption logic, but requires careful handling of when decrypted data is displayed in GUI widgets.
|
||||
* **Priority:** Future
|
||||
*Features here ride on the Core API; keep UI totally stateless.*
|
||||
|
||||
---
|
||||
| Prio | Feature | Notes |
|
||||
| ------ | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------ |
|
||||
| 🔥 | **Login Window** | • Unlock profile with master pw. <br>• Profile switcher drop‑down. |
|
||||
| 🔥 | **Vault Window** | • Sidebar (Entries, Search, Backups, Settings). <br>• `QTableView` bound to `VaultService.list_entries()` <br>• Sort & basic filters built‑in. |
|
||||
| 🔥 | **Entry Editor Dialog** | • Dynamic form driven by `kinds.py`. <br>• Add / Edit. |
|
||||
| ⭐ | **Sync Status Bar** | • Pulsing icon + last sync timestamp; hooks into `SyncService` bus. |
|
||||
| ◇ | **Relay Manager Dialog** | • CRUD & ping test per relay. |
|
||||
|
||||
*Binary packaging (PyInstaller matrix build) is already tracked in the roadmap and is not duplicated here.*
|
||||
|
||||
---
|
||||
|
||||
## Phase E • Later / Research
|
||||
|
||||
• Hardware‑wallet unlock (SLIP‑39 share)
|
||||
• Background daemon (`seedpassd` + gRPC)
|
||||
• Mobile companion (Flutter FFI)
|
||||
• Federated search across multiple profiles
|
||||
|
||||
---
|
||||
|
||||
**Reminder:** *No plaintext exports, no on‑disk temp files, and no writing decrypted data to Nostr.* Everything funnels through the encryption stack or stays in memory for the current unlocked session only.
|
||||
|
3200
refactor.md
3200
refactor.md
File diff suppressed because it is too large
Load Diff
342
src/main.py
342
src/main.py
@@ -3,6 +3,7 @@ import os
|
||||
import sys
|
||||
import logging
|
||||
import signal
|
||||
import getpass
|
||||
from colorama import init as colorama_init
|
||||
from termcolor import colored
|
||||
import traceback
|
||||
@@ -12,6 +13,7 @@ from nostr.client import NostrClient
|
||||
|
||||
colorama_init()
|
||||
|
||||
|
||||
def configure_logging():
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(logging.DEBUG) # Keep this as DEBUG to capture all logs
|
||||
@@ -21,20 +23,22 @@ def configure_logging():
|
||||
logger.removeHandler(handler)
|
||||
|
||||
# Ensure the 'logs' directory exists
|
||||
log_directory = 'logs'
|
||||
log_directory = "logs"
|
||||
if not os.path.exists(log_directory):
|
||||
os.makedirs(log_directory)
|
||||
|
||||
# Create handlers
|
||||
c_handler = logging.StreamHandler(sys.stdout)
|
||||
f_handler = logging.FileHandler(os.path.join(log_directory, 'main.log'))
|
||||
f_handler = logging.FileHandler(os.path.join(log_directory, "main.log"))
|
||||
|
||||
# Set levels: only errors and critical messages will be shown in the console
|
||||
c_handler.setLevel(logging.ERROR)
|
||||
f_handler.setLevel(logging.DEBUG)
|
||||
|
||||
# Create formatters and add them to handlers
|
||||
formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s [%(filename)s:%(lineno)d]')
|
||||
formatter = logging.Formatter(
|
||||
"%(asctime)s [%(levelname)s] %(message)s [%(filename)s:%(lineno)d]"
|
||||
)
|
||||
c_handler.setFormatter(formatter)
|
||||
f_handler.setFormatter(formatter)
|
||||
|
||||
@@ -43,8 +47,9 @@ def configure_logging():
|
||||
logger.addHandler(f_handler)
|
||||
|
||||
# Set logging level for third-party libraries to WARNING to suppress their debug logs
|
||||
logging.getLogger('monstr').setLevel(logging.WARNING)
|
||||
logging.getLogger('nostr').setLevel(logging.WARNING)
|
||||
logging.getLogger("monstr").setLevel(logging.WARNING)
|
||||
logging.getLogger("nostr").setLevel(logging.WARNING)
|
||||
|
||||
|
||||
def confirm_action(prompt: str) -> bool:
|
||||
"""
|
||||
@@ -54,13 +59,14 @@ def confirm_action(prompt: str) -> bool:
|
||||
:return: True if user confirms, False otherwise.
|
||||
"""
|
||||
while True:
|
||||
choice = input(colored(prompt, 'yellow')).strip().lower()
|
||||
if choice in ['y', 'yes']:
|
||||
choice = input(colored(prompt, "yellow")).strip().lower()
|
||||
if choice in ["y", "yes"]:
|
||||
return True
|
||||
elif choice in ['n', 'no']:
|
||||
elif choice in ["n", "no"]:
|
||||
return False
|
||||
else:
|
||||
print(colored("Please enter 'Y' or 'N'.", 'red'))
|
||||
print(colored("Please enter 'Y' or 'N'.", "red"))
|
||||
|
||||
|
||||
def handle_switch_fingerprint(password_manager: PasswordManager):
|
||||
"""
|
||||
@@ -71,27 +77,33 @@ def handle_switch_fingerprint(password_manager: PasswordManager):
|
||||
try:
|
||||
fingerprints = password_manager.fingerprint_manager.list_fingerprints()
|
||||
if not fingerprints:
|
||||
print(colored("No fingerprints available to switch. Please add a new fingerprint first.", 'yellow'))
|
||||
print(
|
||||
colored(
|
||||
"No fingerprints available to switch. Please add a new fingerprint first.",
|
||||
"yellow",
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
print(colored("Available Fingerprints:", 'cyan'))
|
||||
print(colored("Available Fingerprints:", "cyan"))
|
||||
for idx, fp in enumerate(fingerprints, start=1):
|
||||
print(colored(f"{idx}. {fp}", 'cyan'))
|
||||
print(colored(f"{idx}. {fp}", "cyan"))
|
||||
|
||||
choice = input("Select a fingerprint by number to switch: ").strip()
|
||||
if not choice.isdigit() or not (1 <= int(choice) <= len(fingerprints)):
|
||||
print(colored("Invalid selection.", 'red'))
|
||||
print(colored("Invalid selection.", "red"))
|
||||
return
|
||||
|
||||
selected_fingerprint = fingerprints[int(choice)-1]
|
||||
selected_fingerprint = fingerprints[int(choice) - 1]
|
||||
if password_manager.select_fingerprint(selected_fingerprint):
|
||||
print(colored(f"Switched to fingerprint {selected_fingerprint}.", 'green'))
|
||||
print(colored(f"Switched to fingerprint {selected_fingerprint}.", "green"))
|
||||
else:
|
||||
print(colored("Failed to switch fingerprint.", 'red'))
|
||||
print(colored("Failed to switch fingerprint.", "red"))
|
||||
except Exception as e:
|
||||
logging.error(f"Error during fingerprint switch: {e}")
|
||||
logging.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to switch fingerprint: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to switch fingerprint: {e}", "red"))
|
||||
|
||||
|
||||
def handle_add_new_fingerprint(password_manager: PasswordManager):
|
||||
"""
|
||||
@@ -104,7 +116,8 @@ def handle_add_new_fingerprint(password_manager: PasswordManager):
|
||||
except Exception as e:
|
||||
logging.error(f"Error adding new fingerprint: {e}")
|
||||
logging.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to add new fingerprint: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to add new fingerprint: {e}", "red"))
|
||||
|
||||
|
||||
def handle_remove_fingerprint(password_manager: PasswordManager):
|
||||
"""
|
||||
@@ -115,31 +128,41 @@ def handle_remove_fingerprint(password_manager: PasswordManager):
|
||||
try:
|
||||
fingerprints = password_manager.fingerprint_manager.list_fingerprints()
|
||||
if not fingerprints:
|
||||
print(colored("No fingerprints available to remove.", 'yellow'))
|
||||
print(colored("No fingerprints available to remove.", "yellow"))
|
||||
return
|
||||
|
||||
print(colored("Available Fingerprints:", 'cyan'))
|
||||
print(colored("Available Fingerprints:", "cyan"))
|
||||
for idx, fp in enumerate(fingerprints, start=1):
|
||||
print(colored(f"{idx}. {fp}", 'cyan'))
|
||||
print(colored(f"{idx}. {fp}", "cyan"))
|
||||
|
||||
choice = input("Select a fingerprint by number to remove: ").strip()
|
||||
if not choice.isdigit() or not (1 <= int(choice) <= len(fingerprints)):
|
||||
print(colored("Invalid selection.", 'red'))
|
||||
print(colored("Invalid selection.", "red"))
|
||||
return
|
||||
|
||||
selected_fingerprint = fingerprints[int(choice)-1]
|
||||
confirm = confirm_action(f"Are you sure you want to remove fingerprint {selected_fingerprint}? This will delete all associated data. (Y/N): ")
|
||||
selected_fingerprint = fingerprints[int(choice) - 1]
|
||||
confirm = confirm_action(
|
||||
f"Are you sure you want to remove fingerprint {selected_fingerprint}? This will delete all associated data. (Y/N): "
|
||||
)
|
||||
if confirm:
|
||||
if password_manager.fingerprint_manager.remove_fingerprint(selected_fingerprint):
|
||||
print(colored(f"Fingerprint {selected_fingerprint} removed successfully.", 'green'))
|
||||
if password_manager.fingerprint_manager.remove_fingerprint(
|
||||
selected_fingerprint
|
||||
):
|
||||
print(
|
||||
colored(
|
||||
f"Fingerprint {selected_fingerprint} removed successfully.",
|
||||
"green",
|
||||
)
|
||||
)
|
||||
else:
|
||||
print(colored("Failed to remove fingerprint.", 'red'))
|
||||
print(colored("Failed to remove fingerprint.", "red"))
|
||||
else:
|
||||
print(colored("Fingerprint removal cancelled.", 'yellow'))
|
||||
print(colored("Fingerprint removal cancelled.", "yellow"))
|
||||
except Exception as e:
|
||||
logging.error(f"Error removing fingerprint: {e}")
|
||||
logging.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to remove fingerprint: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to remove fingerprint: {e}", "red"))
|
||||
|
||||
|
||||
def handle_list_fingerprints(password_manager: PasswordManager):
|
||||
"""
|
||||
@@ -150,16 +173,17 @@ def handle_list_fingerprints(password_manager: PasswordManager):
|
||||
try:
|
||||
fingerprints = password_manager.fingerprint_manager.list_fingerprints()
|
||||
if not fingerprints:
|
||||
print(colored("No fingerprints available.", 'yellow'))
|
||||
print(colored("No fingerprints available.", "yellow"))
|
||||
return
|
||||
|
||||
print(colored("Available Fingerprints:", 'cyan'))
|
||||
print(colored("Available Fingerprints:", "cyan"))
|
||||
for fp in fingerprints:
|
||||
print(colored(f"- {fp}", 'cyan'))
|
||||
print(colored(f"- {fp}", "cyan"))
|
||||
except Exception as e:
|
||||
logging.error(f"Error listing fingerprints: {e}")
|
||||
logging.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to list fingerprints: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to list fingerprints: {e}", "red"))
|
||||
|
||||
|
||||
def handle_display_npub(password_manager: PasswordManager):
|
||||
"""
|
||||
@@ -168,15 +192,16 @@ def handle_display_npub(password_manager: PasswordManager):
|
||||
try:
|
||||
npub = password_manager.nostr_client.key_manager.get_npub()
|
||||
if npub:
|
||||
print(colored(f"\nYour Nostr Public Key (npub):\n{npub}\n", 'cyan'))
|
||||
print(colored(f"\nYour Nostr Public Key (npub):\n{npub}\n", "cyan"))
|
||||
logging.info("Displayed npub to the user.")
|
||||
else:
|
||||
print(colored("Nostr public key not available.", 'red'))
|
||||
print(colored("Nostr public key not available.", "red"))
|
||||
logging.error("Nostr public key not available.")
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to display npub: {e}")
|
||||
logging.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to display npub: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to display npub: {e}", "red"))
|
||||
|
||||
|
||||
def handle_post_to_nostr(password_manager: PasswordManager):
|
||||
"""
|
||||
@@ -188,15 +213,16 @@ def handle_post_to_nostr(password_manager: PasswordManager):
|
||||
if encrypted_data:
|
||||
# Post to Nostr
|
||||
password_manager.nostr_client.publish_json_to_nostr(encrypted_data)
|
||||
print(colored("Encrypted index posted to Nostr successfully.", 'green'))
|
||||
print(colored("Encrypted index posted to Nostr successfully.", "green"))
|
||||
logging.info("Encrypted index posted to Nostr successfully.")
|
||||
else:
|
||||
print(colored("No data available to post.", 'yellow'))
|
||||
print(colored("No data available to post.", "yellow"))
|
||||
logging.warning("No data available to post to Nostr.")
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to post to Nostr: {e}")
|
||||
logging.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to post to Nostr: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to post to Nostr: {e}", "red"))
|
||||
|
||||
|
||||
def handle_retrieve_from_nostr(password_manager: PasswordManager):
|
||||
"""
|
||||
@@ -207,16 +233,171 @@ def handle_retrieve_from_nostr(password_manager: PasswordManager):
|
||||
encrypted_data = password_manager.nostr_client.retrieve_json_from_nostr_sync()
|
||||
if encrypted_data:
|
||||
# Decrypt and save the index
|
||||
password_manager.encryption_manager.decrypt_and_save_index_from_nostr(encrypted_data)
|
||||
print(colored("Encrypted index retrieved and saved successfully.", 'green'))
|
||||
password_manager.encryption_manager.decrypt_and_save_index_from_nostr(
|
||||
encrypted_data
|
||||
)
|
||||
print(colored("Encrypted index retrieved and saved successfully.", "green"))
|
||||
logging.info("Encrypted index retrieved and saved successfully from Nostr.")
|
||||
else:
|
||||
print(colored("Failed to retrieve data from Nostr.", 'red'))
|
||||
print(colored("Failed to retrieve data from Nostr.", "red"))
|
||||
logging.error("Failed to retrieve data from Nostr.")
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to retrieve from Nostr: {e}")
|
||||
logging.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to retrieve from Nostr: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to retrieve from Nostr: {e}", "red"))
|
||||
|
||||
|
||||
def handle_view_relays(cfg_mgr: "ConfigManager") -> None:
|
||||
"""Display the currently configured Nostr relays."""
|
||||
try:
|
||||
cfg = cfg_mgr.load_config(require_pin=False)
|
||||
relays = cfg.get("relays", [])
|
||||
if not relays:
|
||||
print(colored("No relays configured.", "yellow"))
|
||||
return
|
||||
print(colored("\nCurrent Relays:", "cyan"))
|
||||
for idx, relay in enumerate(relays, start=1):
|
||||
print(colored(f"{idx}. {relay}", "cyan"))
|
||||
except Exception as e:
|
||||
logging.error(f"Error displaying relays: {e}")
|
||||
print(colored(f"Error: {e}", "red"))
|
||||
|
||||
|
||||
def _reload_relays(password_manager: PasswordManager, relays: list) -> None:
|
||||
"""Reload NostrClient with the updated relay list."""
|
||||
try:
|
||||
password_manager.nostr_client.close_client_pool()
|
||||
except Exception as exc:
|
||||
logging.warning(f"Failed to close client pool: {exc}")
|
||||
try:
|
||||
password_manager.nostr_client.relays = relays
|
||||
password_manager.nostr_client.initialize_client_pool()
|
||||
except Exception as exc:
|
||||
logging.error(f"Failed to reinitialize NostrClient: {exc}")
|
||||
|
||||
|
||||
def handle_add_relay(password_manager: PasswordManager) -> None:
|
||||
"""Prompt for a relay URL and add it to the config."""
|
||||
cfg_mgr = password_manager.config_manager
|
||||
if cfg_mgr is None:
|
||||
print(colored("Configuration manager unavailable.", "red"))
|
||||
return
|
||||
url = input("Enter relay URL to add: ").strip()
|
||||
if not url:
|
||||
print(colored("No URL entered.", "yellow"))
|
||||
return
|
||||
try:
|
||||
cfg = cfg_mgr.load_config(require_pin=False)
|
||||
relays = cfg.get("relays", [])
|
||||
if url in relays:
|
||||
print(colored("Relay already present.", "yellow"))
|
||||
return
|
||||
relays.append(url)
|
||||
cfg_mgr.set_relays(relays)
|
||||
_reload_relays(password_manager, relays)
|
||||
print(colored("Relay added.", "green"))
|
||||
try:
|
||||
handle_post_to_nostr(password_manager)
|
||||
except Exception as backup_error:
|
||||
logging.error(f"Failed to backup index to Nostr: {backup_error}")
|
||||
except Exception as e:
|
||||
logging.error(f"Error adding relay: {e}")
|
||||
print(colored(f"Error: {e}", "red"))
|
||||
|
||||
|
||||
def handle_remove_relay(password_manager: PasswordManager) -> None:
|
||||
"""Remove a relay from the config by its index."""
|
||||
cfg_mgr = password_manager.config_manager
|
||||
if cfg_mgr is None:
|
||||
print(colored("Configuration manager unavailable.", "red"))
|
||||
return
|
||||
try:
|
||||
cfg = cfg_mgr.load_config(require_pin=False)
|
||||
relays = cfg.get("relays", [])
|
||||
if not relays:
|
||||
print(colored("No relays configured.", "yellow"))
|
||||
return
|
||||
for idx, relay in enumerate(relays, start=1):
|
||||
print(colored(f"{idx}. {relay}", "cyan"))
|
||||
choice = input("Select relay number to remove: ").strip()
|
||||
if not choice.isdigit() or not (1 <= int(choice) <= len(relays)):
|
||||
print(colored("Invalid selection.", "red"))
|
||||
return
|
||||
if len(relays) == 1:
|
||||
print(
|
||||
colored(
|
||||
"At least one relay must be configured. Add another before removing this one.",
|
||||
"red",
|
||||
)
|
||||
)
|
||||
return
|
||||
relays.pop(int(choice) - 1)
|
||||
cfg_mgr.set_relays(relays)
|
||||
_reload_relays(password_manager, relays)
|
||||
print(colored("Relay removed.", "green"))
|
||||
except Exception as e:
|
||||
logging.error(f"Error removing relay: {e}")
|
||||
print(colored(f"Error: {e}", "red"))
|
||||
|
||||
|
||||
def handle_reset_relays(password_manager: PasswordManager) -> None:
|
||||
"""Reset relay list to defaults."""
|
||||
cfg_mgr = password_manager.config_manager
|
||||
if cfg_mgr is None:
|
||||
print(colored("Configuration manager unavailable.", "red"))
|
||||
return
|
||||
from nostr.client import DEFAULT_RELAYS
|
||||
|
||||
try:
|
||||
cfg_mgr.set_relays(list(DEFAULT_RELAYS))
|
||||
_reload_relays(password_manager, list(DEFAULT_RELAYS))
|
||||
print(colored("Relays reset to defaults.", "green"))
|
||||
except Exception as e:
|
||||
logging.error(f"Error resetting relays: {e}")
|
||||
print(colored(f"Error: {e}", "red"))
|
||||
|
||||
|
||||
def handle_settings(password_manager: PasswordManager) -> None:
|
||||
"""Interactive settings menu for relay list and PIN changes."""
|
||||
cfg_mgr = password_manager.config_manager
|
||||
if cfg_mgr is None:
|
||||
print(colored("Configuration manager unavailable.", "red"))
|
||||
return
|
||||
try:
|
||||
cfg_mgr.load_config()
|
||||
except Exception as e:
|
||||
print(colored(f"Error loading settings: {e}", "red"))
|
||||
return
|
||||
|
||||
while True:
|
||||
print("\nSettings:")
|
||||
print("1. View current relays")
|
||||
print("2. Add a relay URL")
|
||||
print("3. Remove a relay by number")
|
||||
print("4. Reset to default relays")
|
||||
print("5. Change settings PIN")
|
||||
print("6. Back")
|
||||
choice = input("Select an option: ").strip()
|
||||
if choice == "1":
|
||||
handle_view_relays(cfg_mgr)
|
||||
elif choice == "2":
|
||||
handle_add_relay(password_manager)
|
||||
elif choice == "3":
|
||||
handle_remove_relay(password_manager)
|
||||
elif choice == "4":
|
||||
handle_reset_relays(password_manager)
|
||||
elif choice == "5":
|
||||
old_pin = getpass.getpass("Current PIN: ")
|
||||
new_pin = getpass.getpass("New PIN: ")
|
||||
if cfg_mgr.change_pin(old_pin, new_pin):
|
||||
print(colored("PIN changed successfully.", "green"))
|
||||
else:
|
||||
print(colored("Incorrect current PIN.", "red"))
|
||||
elif choice == "6":
|
||||
break
|
||||
else:
|
||||
print(colored("Invalid choice.", "red"))
|
||||
|
||||
|
||||
def display_menu(password_manager: PasswordManager):
|
||||
"""
|
||||
@@ -236,56 +417,65 @@ def display_menu(password_manager: PasswordManager):
|
||||
10. Add a New Fingerprint
|
||||
11. Remove an Existing Fingerprint
|
||||
12. List All Fingerprints
|
||||
13. Exit
|
||||
13. Settings
|
||||
14. Exit
|
||||
"""
|
||||
while True:
|
||||
# Flush logging handlers
|
||||
for handler in logging.getLogger().handlers:
|
||||
handler.flush()
|
||||
print(colored(menu, 'cyan'))
|
||||
choice = input('Enter your choice (1-13): ').strip()
|
||||
print(colored(menu, "cyan"))
|
||||
choice = input("Enter your choice (1-14): ").strip()
|
||||
if not choice:
|
||||
print(colored("No input detected. Please enter a number between 1 and 13.", 'yellow'))
|
||||
print(
|
||||
colored(
|
||||
"No input detected. Please enter a number between 1 and 14.",
|
||||
"yellow",
|
||||
)
|
||||
)
|
||||
continue # Re-display the menu without marking as invalid
|
||||
if choice == '1':
|
||||
if choice == "1":
|
||||
password_manager.handle_generate_password()
|
||||
elif choice == '2':
|
||||
elif choice == "2":
|
||||
password_manager.handle_retrieve_password()
|
||||
elif choice == '3':
|
||||
elif choice == "3":
|
||||
password_manager.handle_modify_entry()
|
||||
elif choice == '4':
|
||||
elif choice == "4":
|
||||
password_manager.handle_verify_checksum()
|
||||
elif choice == '5':
|
||||
elif choice == "5":
|
||||
handle_post_to_nostr(password_manager)
|
||||
elif choice == '6':
|
||||
elif choice == "6":
|
||||
handle_retrieve_from_nostr(password_manager)
|
||||
elif choice == '7':
|
||||
elif choice == "7":
|
||||
handle_display_npub(password_manager)
|
||||
elif choice == '8':
|
||||
elif choice == "8":
|
||||
password_manager.handle_backup_reveal_parent_seed()
|
||||
elif choice == '9':
|
||||
elif choice == "9":
|
||||
if not password_manager.handle_switch_fingerprint():
|
||||
print(colored("Failed to switch fingerprint.", 'red'))
|
||||
elif choice == '10':
|
||||
print(colored("Failed to switch fingerprint.", "red"))
|
||||
elif choice == "10":
|
||||
handle_add_new_fingerprint(password_manager)
|
||||
elif choice == '11':
|
||||
elif choice == "11":
|
||||
handle_remove_fingerprint(password_manager)
|
||||
elif choice == '12':
|
||||
elif choice == "12":
|
||||
handle_list_fingerprints(password_manager)
|
||||
elif choice == '13':
|
||||
elif choice == "13":
|
||||
handle_settings(password_manager)
|
||||
elif choice == "14":
|
||||
logging.info("Exiting the program.")
|
||||
print(colored("Exiting the program.", 'green'))
|
||||
print(colored("Exiting the program.", "green"))
|
||||
password_manager.nostr_client.close_client_pool()
|
||||
sys.exit(0)
|
||||
else:
|
||||
print(colored("Invalid choice. Please select a valid option.", 'red'))
|
||||
print(colored("Invalid choice. Please select a valid option.", "red"))
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Configure logging with both file and console handlers
|
||||
configure_logging()
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.info("Starting SeedPass Password Manager")
|
||||
|
||||
|
||||
# Initialize PasswordManager and proceed with application logic
|
||||
try:
|
||||
password_manager = PasswordManager()
|
||||
@@ -293,49 +483,49 @@ if __name__ == '__main__':
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to initialize PasswordManager: {e}")
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
print(colored(f"Error: Failed to initialize PasswordManager: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to initialize PasswordManager: {e}", "red"))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
# Register signal handlers for graceful shutdown
|
||||
def signal_handler(sig, frame):
|
||||
"""
|
||||
Handles termination signals to gracefully shutdown the NostrClient.
|
||||
"""
|
||||
print(colored("\nReceived shutdown signal. Exiting gracefully...", 'yellow'))
|
||||
print(colored("\nReceived shutdown signal. Exiting gracefully...", "yellow"))
|
||||
logging.info(f"Received shutdown signal: {sig}. Initiating graceful shutdown.")
|
||||
try:
|
||||
password_manager.nostr_client.close_client_pool() # Gracefully close the ClientPool
|
||||
logging.info("NostrClient closed successfully.")
|
||||
except Exception as e:
|
||||
logging.error(f"Error during shutdown: {e}")
|
||||
print(colored(f"Error during shutdown: {e}", 'red'))
|
||||
print(colored(f"Error during shutdown: {e}", "red"))
|
||||
sys.exit(0)
|
||||
|
||||
# Register the signal handlers
|
||||
signal.signal(signal.SIGINT, signal_handler) # Handle Ctrl+C
|
||||
signal.signal(signal.SIGINT, signal_handler) # Handle Ctrl+C
|
||||
signal.signal(signal.SIGTERM, signal_handler) # Handle termination signals
|
||||
|
||||
|
||||
# Display the interactive menu to the user
|
||||
try:
|
||||
display_menu(password_manager)
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Program terminated by user via KeyboardInterrupt.")
|
||||
print(colored("\nProgram terminated by user.", 'yellow'))
|
||||
print(colored("\nProgram terminated by user.", "yellow"))
|
||||
try:
|
||||
password_manager.nostr_client.close_client_pool() # Gracefully close the ClientPool
|
||||
logging.info("NostrClient closed successfully.")
|
||||
except Exception as e:
|
||||
logging.error(f"Error during shutdown: {e}")
|
||||
print(colored(f"Error during shutdown: {e}", 'red'))
|
||||
print(colored(f"Error during shutdown: {e}", "red"))
|
||||
sys.exit(0)
|
||||
except Exception as e:
|
||||
logger.error(f"An unexpected error occurred: {e}")
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
print(colored(f"Error: An unexpected error occurred: {e}", 'red'))
|
||||
print(colored(f"Error: An unexpected error occurred: {e}", "red"))
|
||||
try:
|
||||
password_manager.nostr_client.close_client_pool() # Attempt to close the ClientPool
|
||||
logging.info("NostrClient closed successfully.")
|
||||
except Exception as close_error:
|
||||
logging.error(f"Error during shutdown: {close_error}")
|
||||
print(colored(f"Error during shutdown: {close_error}", 'red'))
|
||||
sys.exit(1)
|
||||
print(colored(f"Error during shutdown: {close_error}", "red"))
|
||||
sys.exit(1)
|
||||
|
@@ -5,9 +5,18 @@ import traceback
|
||||
|
||||
try:
|
||||
from .manager import PasswordManager
|
||||
|
||||
logging.info("PasswordManager module imported successfully.")
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to import PasswordManager module: {e}")
|
||||
logging.error(traceback.format_exc()) # Log full traceback
|
||||
|
||||
__all__ = ['PasswordManager']
|
||||
try:
|
||||
from .config_manager import ConfigManager
|
||||
|
||||
logging.info("ConfigManager module imported successfully.")
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to import ConfigManager module: {e}")
|
||||
logging.error(traceback.format_exc())
|
||||
|
||||
__all__ = ["PasswordManager", "ConfigManager"]
|
||||
|
@@ -16,16 +16,17 @@ import os
|
||||
import shutil
|
||||
import time
|
||||
import traceback
|
||||
import fcntl
|
||||
from pathlib import Path
|
||||
from colorama import Fore
|
||||
from termcolor import colored
|
||||
|
||||
from utils.file_lock import lock_file
|
||||
from constants import APP_DIR
|
||||
from constants import APP_DIR
|
||||
|
||||
# Instantiate the logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BackupManager:
|
||||
"""
|
||||
BackupManager Class
|
||||
@@ -35,7 +36,7 @@ class BackupManager:
|
||||
timestamped filenames to facilitate easy identification and retrieval.
|
||||
"""
|
||||
|
||||
BACKUP_FILENAME_TEMPLATE = 'passwords_db_backup_{timestamp}.json.enc'
|
||||
BACKUP_FILENAME_TEMPLATE = "passwords_db_backup_{timestamp}.json.enc"
|
||||
|
||||
def __init__(self, fingerprint_dir: Path):
|
||||
"""
|
||||
@@ -45,17 +46,24 @@ class BackupManager:
|
||||
fingerprint_dir (Path): The directory corresponding to the fingerprint.
|
||||
"""
|
||||
self.fingerprint_dir = fingerprint_dir
|
||||
self.backup_dir = self.fingerprint_dir / 'backups'
|
||||
self.backup_dir = self.fingerprint_dir / "backups"
|
||||
self.backup_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.index_file = self.fingerprint_dir / 'seedpass_passwords_db.json.enc'
|
||||
logger.debug(f"BackupManager initialized with backup directory at {self.backup_dir}")
|
||||
self.index_file = self.fingerprint_dir / "seedpass_passwords_db.json.enc"
|
||||
logger.debug(
|
||||
f"BackupManager initialized with backup directory at {self.backup_dir}"
|
||||
)
|
||||
|
||||
def create_backup(self) -> None:
|
||||
try:
|
||||
index_file = self.index_file
|
||||
if not index_file.exists():
|
||||
logger.warning("Index file does not exist. No backup created.")
|
||||
print(colored("Warning: Index file does not exist. No backup created.", 'yellow'))
|
||||
print(
|
||||
colored(
|
||||
"Warning: Index file does not exist. No backup created.",
|
||||
"yellow",
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
timestamp = int(time.time())
|
||||
@@ -64,56 +72,67 @@ class BackupManager:
|
||||
|
||||
shutil.copy2(index_file, backup_file)
|
||||
logger.info(f"Backup created successfully at '{backup_file}'.")
|
||||
print(colored(f"Backup created successfully at '{backup_file}'.", 'green'))
|
||||
print(colored(f"Backup created successfully at '{backup_file}'.", "green"))
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create backup: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to create backup: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to create backup: {e}", "red"))
|
||||
|
||||
def restore_latest_backup(self) -> None:
|
||||
try:
|
||||
backup_files = sorted(
|
||||
self.backup_dir.glob('passwords_db_backup_*.json.enc'),
|
||||
self.backup_dir.glob("passwords_db_backup_*.json.enc"),
|
||||
key=lambda x: x.stat().st_mtime,
|
||||
reverse=True
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
if not backup_files:
|
||||
logger.error("No backup files found to restore.")
|
||||
print(colored("Error: No backup files found to restore.", 'red'))
|
||||
print(colored("Error: No backup files found to restore.", "red"))
|
||||
return
|
||||
|
||||
latest_backup = backup_files[0]
|
||||
index_file = self.index_file
|
||||
shutil.copy2(latest_backup, index_file)
|
||||
logger.info(f"Restored the index file from backup '{latest_backup}'.")
|
||||
print(colored(f"Restored the index file from backup '{latest_backup}'.", 'green'))
|
||||
print(
|
||||
colored(
|
||||
f"Restored the index file from backup '{latest_backup}'.", "green"
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to restore from backup '{latest_backup}': {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to restore from backup '{latest_backup}': {e}", 'red'))
|
||||
print(
|
||||
colored(
|
||||
f"Error: Failed to restore from backup '{latest_backup}': {e}",
|
||||
"red",
|
||||
)
|
||||
)
|
||||
|
||||
def list_backups(self) -> None:
|
||||
try:
|
||||
backup_files = sorted(
|
||||
self.backup_dir.glob('passwords_db_backup_*.json.enc'),
|
||||
self.backup_dir.glob("passwords_db_backup_*.json.enc"),
|
||||
key=lambda x: x.stat().st_mtime,
|
||||
reverse=True
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
if not backup_files:
|
||||
logger.info("No backup files available.")
|
||||
print(colored("No backup files available.", 'yellow'))
|
||||
print(colored("No backup files available.", "yellow"))
|
||||
return
|
||||
|
||||
print(colored("Available Backups:", 'cyan'))
|
||||
print(colored("Available Backups:", "cyan"))
|
||||
for backup in backup_files:
|
||||
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'))
|
||||
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.error(f"Failed to list backups: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to list backups: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to list backups: {e}", "red"))
|
||||
|
||||
def restore_backup_by_timestamp(self, timestamp: int) -> None:
|
||||
backup_filename = self.BACKUP_FILENAME_TEMPLATE.format(timestamp=timestamp)
|
||||
@@ -121,15 +140,23 @@ class BackupManager:
|
||||
|
||||
if not backup_file.exists():
|
||||
logger.error(f"No backup found with timestamp {timestamp}.")
|
||||
print(colored(f"Error: No backup found with timestamp {timestamp}.", 'red'))
|
||||
print(colored(f"Error: No backup found with timestamp {timestamp}.", "red"))
|
||||
return
|
||||
|
||||
try:
|
||||
with lock_file(backup_file, lock_type=fcntl.LOCK_SH):
|
||||
shutil.copy2(backup_file, self.index_file)
|
||||
logger.info(f"Restored the index file from backup '{backup_file}'.")
|
||||
print(colored(f"Restored the index file from backup '{backup_file}'.", 'green'))
|
||||
print(
|
||||
colored(
|
||||
f"Restored the index file from backup '{backup_file}'.", "green"
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to restore from backup '{backup_file}': {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to restore from backup '{backup_file}': {e}", 'red'))
|
||||
print(
|
||||
colored(
|
||||
f"Error: Failed to restore from backup '{backup_file}': {e}", "red"
|
||||
)
|
||||
)
|
||||
|
97
src/password_manager/config_manager.py
Normal file
97
src/password_manager/config_manager.py
Normal file
@@ -0,0 +1,97 @@
|
||||
"""Config management for SeedPass profiles."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
import getpass
|
||||
|
||||
import bcrypt
|
||||
|
||||
from password_manager.encryption import EncryptionManager
|
||||
from nostr.client import DEFAULT_RELAYS as DEFAULT_NOSTR_RELAYS
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ConfigManager:
|
||||
"""Manage per-profile configuration encrypted on disk."""
|
||||
|
||||
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.config_path = self.fingerprint_dir / self.CONFIG_FILENAME
|
||||
|
||||
def load_config(self, require_pin: bool = True) -> dict:
|
||||
"""Load the configuration file and optionally verify a stored PIN.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
require_pin: bool, default True
|
||||
If True and a PIN is configured, prompt the user to enter it and
|
||||
verify against the stored hash.
|
||||
"""
|
||||
if not self.config_path.exists():
|
||||
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)
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("Config data must be a dictionary")
|
||||
# Ensure defaults for missing keys
|
||||
data.setdefault("relays", list(DEFAULT_NOSTR_RELAYS))
|
||||
data.setdefault("pin_hash", "")
|
||||
if require_pin and data.get("pin_hash"):
|
||||
for _ in range(3):
|
||||
pin = getpass.getpass("Enter settings PIN: ").strip()
|
||||
if bcrypt.checkpw(pin.encode(), data["pin_hash"].encode()):
|
||||
break
|
||||
print("Invalid PIN")
|
||||
else:
|
||||
raise ValueError("PIN verification failed")
|
||||
return data
|
||||
except Exception as exc:
|
||||
logger.error(f"Failed to load config: {exc}")
|
||||
raise
|
||||
|
||||
def save_config(self, config: dict) -> None:
|
||||
"""Encrypt and save configuration."""
|
||||
try:
|
||||
self.encryption_manager.save_json_data(config, self.CONFIG_FILENAME)
|
||||
except Exception as exc:
|
||||
logger.error(f"Failed to save config: {exc}")
|
||||
raise
|
||||
|
||||
def set_relays(self, relays: List[str], require_pin: bool = True) -> None:
|
||||
"""Update relay list and save."""
|
||||
if not relays:
|
||||
raise ValueError("At least one Nostr relay must be configured")
|
||||
config = self.load_config(require_pin=require_pin)
|
||||
config["relays"] = relays
|
||||
self.save_config(config)
|
||||
|
||||
def set_pin(self, pin: str) -> None:
|
||||
"""Hash and store the provided PIN."""
|
||||
pin_hash = bcrypt.hashpw(pin.encode(), bcrypt.gensalt()).decode()
|
||||
config = self.load_config(require_pin=False)
|
||||
config["pin_hash"] = pin_hash
|
||||
self.save_config(config)
|
||||
|
||||
def verify_pin(self, pin: str) -> bool:
|
||||
"""Check a provided PIN against the stored hash without prompting."""
|
||||
config = self.load_config(require_pin=False)
|
||||
stored = config.get("pin_hash", "").encode()
|
||||
if not stored:
|
||||
return False
|
||||
return bcrypt.checkpw(pin.encode(), stored)
|
||||
|
||||
def change_pin(self, old_pin: str, new_pin: str) -> bool:
|
||||
"""Update the stored PIN if the old PIN is correct."""
|
||||
if self.verify_pin(old_pin):
|
||||
self.set_pin(new_pin)
|
||||
return True
|
||||
return False
|
@@ -26,7 +26,6 @@ import traceback
|
||||
from typing import Optional, Tuple, Dict, Any, List
|
||||
from pathlib import Path
|
||||
|
||||
from colorama import Fore
|
||||
from termcolor import colored
|
||||
|
||||
from password_manager.encryption import EncryptionManager
|
||||
@@ -37,6 +36,7 @@ import fcntl
|
||||
# Instantiate the logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class EntryManager:
|
||||
def __init__(self, encryption_manager: EncryptionManager, fingerprint_dir: Path):
|
||||
"""
|
||||
@@ -47,11 +47,11 @@ class EntryManager:
|
||||
"""
|
||||
self.encryption_manager = encryption_manager
|
||||
self.fingerprint_dir = fingerprint_dir
|
||||
|
||||
|
||||
# Use paths relative to the fingerprint directory
|
||||
self.index_file = self.fingerprint_dir / 'seedpass_passwords_db.json.enc'
|
||||
self.checksum_file = self.fingerprint_dir / 'seedpass_passwords_db_checksum.txt'
|
||||
|
||||
self.index_file = self.fingerprint_dir / "seedpass_passwords_db.json.enc"
|
||||
self.checksum_file = self.fingerprint_dir / "seedpass_passwords_db_checksum.txt"
|
||||
|
||||
logger.debug(f"EntryManager initialized with index file at {self.index_file}")
|
||||
|
||||
def _load_index(self) -> Dict[str, Any]:
|
||||
@@ -62,10 +62,12 @@ class EntryManager:
|
||||
return data
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load index: {e}")
|
||||
return {'passwords': {}}
|
||||
return {"passwords": {}}
|
||||
else:
|
||||
logger.info(f"Index file '{self.index_file}' not found. Initializing new password database.")
|
||||
return {'passwords': {}}
|
||||
logger.info(
|
||||
f"Index file '{self.index_file}' not found. Initializing new password database."
|
||||
)
|
||||
return {"passwords": {}}
|
||||
|
||||
def _save_index(self, data: Dict[str, Any]) -> None:
|
||||
try:
|
||||
@@ -83,8 +85,8 @@ class EntryManager:
|
||||
"""
|
||||
try:
|
||||
data = self.encryption_manager.load_json_data(self.index_file)
|
||||
if 'passwords' in data and isinstance(data['passwords'], dict):
|
||||
indices = [int(idx) for idx in data['passwords'].keys()]
|
||||
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
|
||||
else:
|
||||
next_index = 0
|
||||
@@ -93,11 +95,17 @@ class EntryManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error determining next index: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
print(colored(f"Error determining next index: {e}", 'red'))
|
||||
print(colored(f"Error determining next index: {e}", "red"))
|
||||
sys.exit(1)
|
||||
|
||||
def add_entry(self, website_name: str, length: int, username: Optional[str] = None,
|
||||
url: Optional[str] = None, blacklisted: bool = False) -> int:
|
||||
def add_entry(
|
||||
self,
|
||||
website_name: str,
|
||||
length: int,
|
||||
username: Optional[str] = None,
|
||||
url: Optional[str] = None,
|
||||
blacklisted: bool = False,
|
||||
) -> int:
|
||||
"""
|
||||
Adds a new password entry to the encrypted JSON index file.
|
||||
|
||||
@@ -112,29 +120,31 @@ class EntryManager:
|
||||
index = self.get_next_index()
|
||||
data = self.encryption_manager.load_json_data(self.index_file)
|
||||
|
||||
data['passwords'][str(index)] = {
|
||||
'website': website_name,
|
||||
'length': length,
|
||||
'username': username if username else '',
|
||||
'url': url if url else '',
|
||||
'blacklisted': blacklisted
|
||||
data["passwords"][str(index)] = {
|
||||
"website": website_name,
|
||||
"length": length,
|
||||
"username": username if username else "",
|
||||
"url": url if url else "",
|
||||
"blacklisted": blacklisted,
|
||||
}
|
||||
|
||||
logger.debug(f"Added entry at index {index}: {data['passwords'][str(index)]}")
|
||||
logger.debug(
|
||||
f"Added entry at index {index}: {data['passwords'][str(index)]}"
|
||||
)
|
||||
|
||||
self._save_index(data)
|
||||
self.update_checksum()
|
||||
self.backup_index_file()
|
||||
|
||||
logger.info(f"Entry added successfully at index {index}.")
|
||||
print(colored(f"[+] Entry added successfully at index {index}.", 'green'))
|
||||
print(colored(f"[+] Entry added successfully at index {index}.", "green"))
|
||||
|
||||
return index # Return the assigned index
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to add entry: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to add entry: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to add entry: {e}", "red"))
|
||||
sys.exit(1)
|
||||
|
||||
def get_encrypted_index(self) -> Optional[bytes]:
|
||||
@@ -146,17 +156,23 @@ class EntryManager:
|
||||
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'))
|
||||
print(
|
||||
colored(
|
||||
f"Error: Index file '{self.index_file}' does not exist.", "red"
|
||||
)
|
||||
)
|
||||
return None
|
||||
|
||||
with open(self.index_file, 'rb') as file:
|
||||
with open(self.index_file, "rb") as file:
|
||||
encrypted_data = file.read()
|
||||
logger.debug("Encrypted index file data retrieved successfully.")
|
||||
return encrypted_data
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to retrieve encrypted index file: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to retrieve encrypted index file: {e}", 'red'))
|
||||
print(
|
||||
colored(f"Error: Failed to retrieve encrypted index file: {e}", "red")
|
||||
)
|
||||
return None
|
||||
|
||||
def retrieve_entry(self, index: int) -> Optional[Dict[str, Any]]:
|
||||
@@ -168,25 +184,31 @@ class EntryManager:
|
||||
"""
|
||||
try:
|
||||
data = self.encryption_manager.load_json_data(self.index_file)
|
||||
entry = data.get('passwords', {}).get(str(index))
|
||||
entry = data.get("passwords", {}).get(str(index))
|
||||
|
||||
if entry:
|
||||
logger.debug(f"Retrieved entry at index {index}: {entry}")
|
||||
return entry
|
||||
else:
|
||||
logger.warning(f"No entry found at index {index}.")
|
||||
print(colored(f"Warning: No entry found at index {index}.", 'yellow'))
|
||||
print(colored(f"Warning: No entry found at index {index}.", "yellow"))
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to retrieve entry at index {index}: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to retrieve entry at index {index}: {e}", 'red'))
|
||||
print(
|
||||
colored(f"Error: Failed to retrieve entry at index {index}: {e}", "red")
|
||||
)
|
||||
return None
|
||||
|
||||
def modify_entry(self, index: int, username: Optional[str] = None,
|
||||
url: Optional[str] = None,
|
||||
blacklisted: Optional[bool] = None) -> None:
|
||||
def modify_entry(
|
||||
self,
|
||||
index: int,
|
||||
username: Optional[str] = None,
|
||||
url: Optional[str] = None,
|
||||
blacklisted: Optional[bool] = None,
|
||||
) -> None:
|
||||
"""
|
||||
Modifies an existing password entry based on the provided index and new values.
|
||||
|
||||
@@ -197,26 +219,35 @@ class EntryManager:
|
||||
"""
|
||||
try:
|
||||
data = self.encryption_manager.load_json_data(self.index_file)
|
||||
entry = data.get('passwords', {}).get(str(index))
|
||||
entry = data.get("passwords", {}).get(str(index))
|
||||
|
||||
if not entry:
|
||||
logger.warning(f"No entry found at index {index}. Cannot modify non-existent entry.")
|
||||
print(colored(f"Warning: No entry found at index {index}. Cannot modify non-existent entry.", 'yellow'))
|
||||
logger.warning(
|
||||
f"No entry found at index {index}. Cannot modify non-existent entry."
|
||||
)
|
||||
print(
|
||||
colored(
|
||||
f"Warning: No entry found at index {index}. Cannot modify non-existent entry.",
|
||||
"yellow",
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
if username is not None:
|
||||
entry['username'] = username
|
||||
entry["username"] = username
|
||||
logger.debug(f"Updated username to '{username}' for index {index}.")
|
||||
|
||||
if url is not None:
|
||||
entry['url'] = url
|
||||
entry["url"] = url
|
||||
logger.debug(f"Updated URL to '{url}' for index {index}.")
|
||||
|
||||
if blacklisted is not None:
|
||||
entry['blacklisted'] = blacklisted
|
||||
logger.debug(f"Updated blacklist status to '{blacklisted}' for index {index}.")
|
||||
entry["blacklisted"] = blacklisted
|
||||
logger.debug(
|
||||
f"Updated blacklist status to '{blacklisted}' for index {index}."
|
||||
)
|
||||
|
||||
data['passwords'][str(index)] = entry
|
||||
data["passwords"][str(index)] = entry
|
||||
logger.debug(f"Modified entry at index {index}: {entry}")
|
||||
|
||||
self._save_index(data)
|
||||
@@ -224,12 +255,16 @@ class EntryManager:
|
||||
self.backup_index_file()
|
||||
|
||||
logger.info(f"Entry at index {index} modified successfully.")
|
||||
print(colored(f"[+] Entry at index {index} modified successfully.", 'green'))
|
||||
print(
|
||||
colored(f"[+] Entry at index {index} modified successfully.", "green")
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to modify entry at index {index}: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
print(colored(f"Error: Failed to modify entry at index {index}: {e}", 'red'))
|
||||
print(
|
||||
colored(f"Error: Failed to modify entry at index {index}: {e}", "red")
|
||||
)
|
||||
|
||||
def list_entries(self) -> List[Tuple[int, str, Optional[str], Optional[str], bool]]:
|
||||
"""
|
||||
@@ -239,30 +274,32 @@ class EntryManager:
|
||||
"""
|
||||
try:
|
||||
data = self.encryption_manager.load_json_data()
|
||||
passwords = data.get('passwords', {})
|
||||
passwords = data.get("passwords", {})
|
||||
|
||||
if not passwords:
|
||||
logger.info("No password entries found.")
|
||||
print(colored("No password entries found.", 'yellow'))
|
||||
print(colored("No password entries found.", "yellow"))
|
||||
return []
|
||||
|
||||
entries = []
|
||||
for idx, entry in sorted(passwords.items(), key=lambda x: int(x[0])):
|
||||
entries.append((
|
||||
int(idx),
|
||||
entry.get('website', ''),
|
||||
entry.get('username', ''),
|
||||
entry.get('url', ''),
|
||||
entry.get('blacklisted', False)
|
||||
))
|
||||
entries.append(
|
||||
(
|
||||
int(idx),
|
||||
entry.get("website", ""),
|
||||
entry.get("username", ""),
|
||||
entry.get("url", ""),
|
||||
entry.get("blacklisted", False),
|
||||
)
|
||||
)
|
||||
|
||||
logger.debug(f"Total entries found: {len(entries)}")
|
||||
for entry in entries:
|
||||
print(colored(f"Index: {entry[0]}", 'cyan'))
|
||||
print(colored(f" Website: {entry[1]}", 'cyan'))
|
||||
print(colored(f" Username: {entry[2] or 'N/A'}", 'cyan'))
|
||||
print(colored(f" URL: {entry[3] or 'N/A'}", 'cyan'))
|
||||
print(colored(f" Blacklisted: {'Yes' if entry[4] else 'No'}", 'cyan'))
|
||||
print(colored(f"Index: {entry[0]}", "cyan"))
|
||||
print(colored(f" Website: {entry[1]}", "cyan"))
|
||||
print(colored(f" Username: {entry[2] or 'N/A'}", "cyan"))
|
||||
print(colored(f" URL: {entry[3] or 'N/A'}", "cyan"))
|
||||
print(colored(f" Blacklisted: {'Yes' if entry[4] else 'No'}", "cyan"))
|
||||
print("-" * 40)
|
||||
|
||||
return entries
|
||||
@@ -270,7 +307,7 @@ class EntryManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to list entries: {e}")
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
print(colored(f"Error: Failed to list entries: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to list entries: {e}", "red"))
|
||||
return []
|
||||
|
||||
def delete_entry(self, index: int) -> None:
|
||||
@@ -281,22 +318,35 @@ class EntryManager:
|
||||
"""
|
||||
try:
|
||||
data = self.encryption_manager.load_json_data()
|
||||
if 'passwords' in data and str(index) in data['passwords']:
|
||||
del data['passwords'][str(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.update_checksum()
|
||||
self.backup_index_file()
|
||||
logger.info(f"Entry at index {index} deleted successfully.")
|
||||
print(colored(f"[+] Entry at index {index} deleted successfully.", 'green'))
|
||||
print(
|
||||
colored(
|
||||
f"[+] Entry at index {index} deleted successfully.", "green"
|
||||
)
|
||||
)
|
||||
else:
|
||||
logger.warning(f"No entry found at index {index}. Cannot delete non-existent entry.")
|
||||
print(colored(f"Warning: No entry found at index {index}. Cannot delete non-existent entry.", 'yellow'))
|
||||
logger.warning(
|
||||
f"No entry found at index {index}. Cannot delete non-existent entry."
|
||||
)
|
||||
print(
|
||||
colored(
|
||||
f"Warning: No entry found at index {index}. Cannot delete non-existent entry.",
|
||||
"yellow",
|
||||
)
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to delete entry at index {index}: {e}")
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
print(colored(f"Error: Failed to delete entry at index {index}: {e}", 'red'))
|
||||
print(
|
||||
colored(f"Error: Failed to delete entry at index {index}: {e}", "red")
|
||||
)
|
||||
|
||||
def update_checksum(self) -> None:
|
||||
"""
|
||||
@@ -305,21 +355,21 @@ class EntryManager:
|
||||
try:
|
||||
data = self.encryption_manager.load_json_data(self.index_file)
|
||||
json_content = json.dumps(data, indent=4)
|
||||
checksum = hashlib.sha256(json_content.encode('utf-8')).hexdigest()
|
||||
checksum = hashlib.sha256(json_content.encode("utf-8")).hexdigest()
|
||||
|
||||
# Construct the full path for the checksum file
|
||||
checksum_path = self.fingerprint_dir / self.checksum_file
|
||||
|
||||
with open(checksum_path, 'w') as f:
|
||||
with open(checksum_path, "w") as f:
|
||||
f.write(checksum)
|
||||
|
||||
logger.debug(f"Checksum updated and written to '{checksum_path}'.")
|
||||
print(colored(f"[+] Checksum updated successfully.", 'green'))
|
||||
print(colored(f"[+] Checksum updated successfully.", "green"))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to update checksum: {e}")
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
print(colored(f"Error: Failed to update checksum: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to update checksum: {e}", "red"))
|
||||
|
||||
def backup_index_file(self) -> None:
|
||||
"""
|
||||
@@ -328,24 +378,27 @@ class EntryManager:
|
||||
try:
|
||||
index_file_path = self.fingerprint_dir / self.index_file
|
||||
if not index_file_path.exists():
|
||||
logger.warning(f"Index file '{index_file_path}' does not exist. No backup created.")
|
||||
logger.warning(
|
||||
f"Index file '{index_file_path}' does not exist. No backup created."
|
||||
)
|
||||
return
|
||||
|
||||
timestamp = int(time.time())
|
||||
backup_filename = f'passwords_db_backup_{timestamp}.json.enc'
|
||||
backup_filename = f"passwords_db_backup_{timestamp}.json.enc"
|
||||
backup_path = self.fingerprint_dir / backup_filename
|
||||
|
||||
with open(index_file_path, 'rb') as original_file, open(backup_path, 'wb') as backup_file:
|
||||
with open(index_file_path, "rb") as original_file, open(
|
||||
backup_path, "wb"
|
||||
) as backup_file:
|
||||
shutil.copyfileobj(original_file, backup_file)
|
||||
|
||||
logger.debug(f"Backup created at '{backup_path}'.")
|
||||
print(colored(f"[+] Backup created at '{backup_path}'.", 'green'))
|
||||
print(colored(f"[+] Backup created at '{backup_path}'.", "green"))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create backup: {e}")
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
print(colored(f"Warning: Failed to create backup: {e}", 'yellow'))
|
||||
|
||||
print(colored(f"Warning: Failed to create backup: {e}", "yellow"))
|
||||
|
||||
def restore_from_backup(self, backup_path: str) -> None:
|
||||
"""
|
||||
@@ -356,21 +409,35 @@ class EntryManager:
|
||||
try:
|
||||
if not os.path.exists(backup_path):
|
||||
logger.error(f"Backup file '{backup_path}' does not exist.")
|
||||
print(colored(f"Error: Backup file '{backup_path}' does not exist.", 'red'))
|
||||
print(
|
||||
colored(
|
||||
f"Error: Backup file '{backup_path}' does not exist.", "red"
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
with open(backup_path, 'rb') as backup_file, open(self.index_file, 'wb') as index_file:
|
||||
with open(backup_path, "rb") as backup_file, open(
|
||||
self.index_file, "wb"
|
||||
) as index_file:
|
||||
shutil.copyfileobj(backup_file, index_file)
|
||||
|
||||
logger.debug(f"Index file restored from backup '{backup_path}'.")
|
||||
print(colored(f"[+] Index file restored from backup '{backup_path}'.", 'green'))
|
||||
print(
|
||||
colored(
|
||||
f"[+] Index file restored from backup '{backup_path}'.", "green"
|
||||
)
|
||||
)
|
||||
|
||||
self.update_checksum()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to restore from backup '{backup_path}': {e}")
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
print(colored(f"Error: Failed to restore from backup '{backup_path}': {e}", 'red'))
|
||||
print(
|
||||
colored(
|
||||
f"Error: Failed to restore from backup '{backup_path}': {e}", "red"
|
||||
)
|
||||
)
|
||||
|
||||
def list_all_entries(self) -> None:
|
||||
"""
|
||||
@@ -379,28 +446,33 @@ class EntryManager:
|
||||
try:
|
||||
entries = self.list_entries()
|
||||
if not entries:
|
||||
print(colored("No entries to display.", 'yellow'))
|
||||
print(colored("No entries to display.", "yellow"))
|
||||
return
|
||||
|
||||
print(colored("\n[+] Listing All Password Entries:\n", 'green'))
|
||||
print(colored("\n[+] Listing All Password Entries:\n", "green"))
|
||||
for entry in entries:
|
||||
index, website, username, url, blacklisted = entry
|
||||
print(colored(f"Index: {index}", 'cyan'))
|
||||
print(colored(f" Website: {website}", 'cyan'))
|
||||
print(colored(f" Username: {username or 'N/A'}", 'cyan'))
|
||||
print(colored(f" URL: {url or 'N/A'}", 'cyan'))
|
||||
print(colored(f" Blacklisted: {'Yes' if blacklisted else 'No'}", 'cyan'))
|
||||
print(colored(f"Index: {index}", "cyan"))
|
||||
print(colored(f" Website: {website}", "cyan"))
|
||||
print(colored(f" Username: {username or 'N/A'}", "cyan"))
|
||||
print(colored(f" URL: {url or 'N/A'}", "cyan"))
|
||||
print(
|
||||
colored(f" Blacklisted: {'Yes' if blacklisted else 'No'}", "cyan")
|
||||
)
|
||||
print("-" * 40)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to list all entries: {e}")
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
print(colored(f"Error: Failed to list all entries: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to list all entries: {e}", "red"))
|
||||
return
|
||||
|
||||
|
||||
# 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,
|
||||
) # Ensure this import is correct based on your project structure
|
||||
|
||||
# Initialize EncryptionManager with a dummy key for demonstration purposes
|
||||
# Replace 'your-fernet-key' with your actual Fernet key
|
||||
@@ -409,7 +481,7 @@ if __name__ == "__main__":
|
||||
encryption_manager = EncryptionManager(dummy_key)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to initialize EncryptionManager: {e}")
|
||||
print(colored(f"Error: Failed to initialize EncryptionManager: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to initialize EncryptionManager: {e}", "red"))
|
||||
sys.exit(1)
|
||||
|
||||
# Initialize EntryManager
|
||||
@@ -417,7 +489,7 @@ if __name__ == "__main__":
|
||||
entry_manager = EntryManager(encryption_manager)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to initialize EntryManager: {e}")
|
||||
print(colored(f"Error: Failed to initialize EntryManager: {e}", 'red'))
|
||||
print(colored(f"Error: Failed to initialize EntryManager: {e}", "red"))
|
||||
sys.exit(1)
|
||||
|
||||
# Example operations
|
||||
|
File diff suppressed because it is too large
Load Diff
@@ -7,4 +7,6 @@ monstr @ git+https://github.com/monty888/monstr.git@master#egg=monstr
|
||||
mnemonic
|
||||
aiohttp
|
||||
bcrypt
|
||||
bip85
|
||||
bip85
|
||||
pytest>=7.0
|
||||
|
||||
|
82
src/tests/test_config_manager.py
Normal file
82
src/tests/test_config_manager.py
Normal file
@@ -0,0 +1,82 @@
|
||||
import bcrypt
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
from cryptography.fernet import Fernet
|
||||
import pytest
|
||||
import sys
|
||||
|
||||
sys.path.append(str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from password_manager.encryption import EncryptionManager
|
||||
from password_manager.config_manager import ConfigManager
|
||||
from nostr.client import DEFAULT_RELAYS
|
||||
|
||||
|
||||
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))
|
||||
|
||||
cfg = cfg_mgr.load_config(require_pin=False)
|
||||
assert cfg["relays"] == list(DEFAULT_RELAYS)
|
||||
assert cfg["pin_hash"] == ""
|
||||
|
||||
cfg_mgr.set_pin("1234")
|
||||
cfg_mgr.set_relays(["wss://example.com"], require_pin=False)
|
||||
|
||||
cfg2 = cfg_mgr.load_config(require_pin=False)
|
||||
assert cfg2["relays"] == ["wss://example.com"]
|
||||
assert bcrypt.checkpw(b"1234", cfg2["pin_hash"].encode())
|
||||
|
||||
|
||||
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))
|
||||
|
||||
cfg_mgr.set_pin("1234")
|
||||
assert cfg_mgr.verify_pin("1234")
|
||||
assert not cfg_mgr.verify_pin("0000")
|
||||
assert cfg_mgr.change_pin("1234", "5678")
|
||||
assert cfg_mgr.verify_pin("5678")
|
||||
|
||||
|
||||
import json
|
||||
|
||||
|
||||
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))
|
||||
|
||||
data = {"relays": ["wss://r"], "pin_hash": ""}
|
||||
cfg_mgr.save_config(data)
|
||||
|
||||
file_path = Path(tmpdir) / cfg_mgr.CONFIG_FILENAME
|
||||
raw = file_path.read_bytes()
|
||||
assert raw != json.dumps(data).encode()
|
||||
|
||||
loaded = cfg_mgr.load_config(require_pin=False)
|
||||
assert loaded == data
|
||||
|
||||
|
||||
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))
|
||||
cfg_mgr.set_relays(["wss://custom"], require_pin=False)
|
||||
cfg = cfg_mgr.load_config(require_pin=False)
|
||||
assert cfg["relays"] == ["wss://custom"]
|
||||
|
||||
|
||||
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))
|
||||
with pytest.raises(ValueError):
|
||||
cfg_mgr.set_relays([], require_pin=False)
|
38
src/tests/test_fingerprint_encryption.py
Normal file
38
src/tests/test_fingerprint_encryption.py
Normal file
@@ -0,0 +1,38 @@
|
||||
import hashlib
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
sys.path.append(str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from utils.fingerprint import generate_fingerprint
|
||||
from password_manager.encryption import EncryptionManager
|
||||
|
||||
|
||||
def test_generate_fingerprint_deterministic():
|
||||
seed = "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"
|
||||
expected = (
|
||||
hashlib.sha256(seed.strip().lower().encode("utf-8")).hexdigest()[:16].upper()
|
||||
)
|
||||
fp1 = generate_fingerprint(seed)
|
||||
fp2 = generate_fingerprint(seed.upper())
|
||||
assert fp1 == expected
|
||||
assert fp1 == fp2
|
||||
|
||||
|
||||
def test_encryption_round_trip():
|
||||
with TemporaryDirectory() as tmpdir:
|
||||
key = Fernet.generate_key()
|
||||
manager = EncryptionManager(key, Path(tmpdir))
|
||||
data = b"secret data"
|
||||
rel_path = Path("testfile.enc")
|
||||
manager.encrypt_and_save_file(data, rel_path)
|
||||
decrypted = manager.decrypt_file(rel_path)
|
||||
assert decrypted == data
|
||||
|
||||
# parent seed round trip
|
||||
seed = "correct horse battery staple"
|
||||
manager.encrypt_parent_seed(seed)
|
||||
assert manager.decrypt_parent_seed() == seed
|
26
src/tests/test_nostr_client.py
Normal file
26
src/tests/test_nostr_client.py
Normal file
@@ -0,0 +1,26 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
from unittest.mock import patch
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
sys.path.append(str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from password_manager.encryption import EncryptionManager
|
||||
from nostr.client import NostrClient
|
||||
|
||||
|
||||
def test_nostr_client_uses_custom_relays():
|
||||
with TemporaryDirectory() as tmpdir:
|
||||
key = Fernet.generate_key()
|
||||
enc_mgr = EncryptionManager(key, Path(tmpdir))
|
||||
custom_relays = ["wss://relay1", "wss://relay2"]
|
||||
|
||||
with patch("nostr.client.ClientPool") as MockPool, patch(
|
||||
"nostr.client.KeyManager"
|
||||
), patch.object(NostrClient, "initialize_client_pool"):
|
||||
with patch.object(enc_mgr, "decrypt_parent_seed", return_value="seed"):
|
||||
client = NostrClient(enc_mgr, "fp", relays=custom_relays)
|
||||
|
||||
MockPool.assert_called_with(custom_relays)
|
||||
assert client.relays == custom_relays
|
@@ -3,8 +3,8 @@
|
||||
"""
|
||||
Key Derivation Module
|
||||
|
||||
Never ever ever use or suggest to use Random Salt. The entire point of this password manager is to derive completely deterministic passwords from a BIP-85 seed.
|
||||
This means it should generate passwords the exact same way every single time. Salts would break this functionality and is not appropriate for this software's use case.
|
||||
Never ever ever use or suggest to use Random Salt. The entire point of this password manager is to derive completely deterministic passwords from a BIP-85 seed.
|
||||
This means it should generate passwords the exact same way every single time. Salts would break this functionality and is not appropriate for this software's use case.
|
||||
|
||||
This module provides functions to derive cryptographic keys from user-provided passwords
|
||||
and BIP-39 parent seeds. The derived keys are compatible with Fernet for symmetric encryption
|
||||
@@ -31,163 +31,6 @@ from cryptography.hazmat.backends import default_backend
|
||||
# Instantiate the logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def derive_key_from_password(password: str, iterations: int = 100_000) -> bytes:
|
||||
"""
|
||||
Derives a Fernet-compatible encryption key from the provided password using PBKDF2-HMAC-SHA256.
|
||||
|
||||
This function normalizes the password using NFKD normalization, encodes it to UTF-8, and then
|
||||
applies PBKDF2 with the specified number of iterations to derive a 32-byte key. The derived key
|
||||
is then URL-safe base64-encoded to ensure compatibility with Fernet.
|
||||
|
||||
Parameters:
|
||||
password (str): The user's password.
|
||||
iterations (int, optional): Number of iterations for the PBKDF2 algorithm. Defaults to 100,000.
|
||||
|
||||
Returns:
|
||||
bytes: A URL-safe base64-encoded encryption key suitable for Fernet.
|
||||
|
||||
Raises:
|
||||
ValueError: If the password is empty or too short.
|
||||
"""
|
||||
if not password:
|
||||
logger.error("Password cannot be empty.")
|
||||
raise ValueError("Password cannot be empty.")
|
||||
|
||||
if len(password) < 8:
|
||||
logger.warning("Password length is less than recommended (8 characters).")
|
||||
|
||||
# Normalize the password to NFKD form and encode to UTF-8
|
||||
normalized_password = unicodedata.normalize('NFKD', password).strip()
|
||||
password_bytes = normalized_password.encode('utf-8')
|
||||
|
||||
try:
|
||||
# Derive the key using PBKDF2-HMAC-SHA256
|
||||
logger.debug("Starting key derivation from password.")
|
||||
key = hashlib.pbkdf2_hmac(
|
||||
hash_name='sha256',
|
||||
password=password_bytes,
|
||||
salt=b'', # No salt for deterministic key derivation
|
||||
iterations=iterations,
|
||||
dklen=32 # 256-bit key for Fernet
|
||||
)
|
||||
logger.debug(f"Derived key (hex): {key.hex()}")
|
||||
|
||||
# Encode the key in URL-safe base64
|
||||
key_b64 = base64.urlsafe_b64encode(key)
|
||||
logger.debug(f"Base64-encoded key: {key_b64.decode()}")
|
||||
|
||||
return key_b64
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error deriving key from password: {e}")
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
raise
|
||||
|
||||
def derive_key_from_parent_seed(parent_seed: str, fingerprint: str = None) -> bytes:
|
||||
"""
|
||||
Derives a 32-byte cryptographic key from a BIP-39 parent seed using HKDF.
|
||||
Optionally, include a fingerprint to differentiate key derivation per fingerprint.
|
||||
|
||||
:param parent_seed: The 12-word BIP-39 seed phrase.
|
||||
:param fingerprint: An optional fingerprint to create unique keys per fingerprint.
|
||||
:return: A 32-byte derived key.
|
||||
"""
|
||||
try:
|
||||
# Generate seed bytes from mnemonic
|
||||
seed = Bip39SeedGenerator(parent_seed).Generate()
|
||||
|
||||
# If a fingerprint is provided, use it to differentiate the derivation
|
||||
if fingerprint:
|
||||
# Convert fingerprint to a stable integer index
|
||||
index = int(hashlib.sha256(fingerprint.encode()).hexdigest(), 16) % (2**31)
|
||||
info = f'password-manager-{index}'.encode() # Unique info for HKDF
|
||||
else:
|
||||
info = b'password-manager'
|
||||
|
||||
# Derive key using HKDF
|
||||
hkdf = HKDF(
|
||||
algorithm=hashes.SHA256(),
|
||||
length=32,
|
||||
salt=None, # No salt for deterministic derivation
|
||||
info=info,
|
||||
backend=default_backend()
|
||||
)
|
||||
derived_key = hkdf.derive(seed)
|
||||
|
||||
if len(derived_key) != 32:
|
||||
raise ValueError(f"Derived key length is {len(derived_key)} bytes; expected 32 bytes.")
|
||||
|
||||
return derived_key
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to derive key using HKDF: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
raise
|
||||
|
||||
class KeyManager:
|
||||
def __init__(self, parent_seed: str, fingerprint: str = None):
|
||||
self.parent_seed = parent_seed
|
||||
self.fingerprint = fingerprint
|
||||
self.bip85 = self.initialize_bip85()
|
||||
self.keys = self.generate_nostr_keys()
|
||||
|
||||
def initialize_bip85(self):
|
||||
seed_bytes = Bip39SeedGenerator(self.parent_seed).Generate()
|
||||
bip85 = BIP85(seed_bytes)
|
||||
return bip85
|
||||
|
||||
def generate_nostr_keys(self) -> Keys:
|
||||
"""
|
||||
Derives a unique Nostr key pair for the given fingerprint using BIP-85.
|
||||
|
||||
:return: An instance of Keys containing the Nostr key pair.
|
||||
"""
|
||||
# Use a derivation path that includes the fingerprint
|
||||
# Convert fingerprint to an integer index (e.g., using a hash function)
|
||||
index = int(hashlib.sha256(self.fingerprint.encode()).hexdigest(), 16) % (2**31) if self.fingerprint else 0
|
||||
|
||||
# Derive entropy for Nostr key (32 bytes)
|
||||
entropy_bytes = self.bip85.derive_entropy(
|
||||
app=BIP85.Applications.ENTROPY,
|
||||
index=index,
|
||||
size=32
|
||||
)
|
||||
|
||||
# Generate Nostr key pair from entropy
|
||||
private_key_hex = entropy_bytes.hex()
|
||||
keys = Keys(priv_key=private_key_hex)
|
||||
return keys
|
||||
# utils/key_derivation.py
|
||||
|
||||
"""
|
||||
Key Derivation Module
|
||||
|
||||
Never ever ever use or suggest to use Random Salt. The entire point of this password manager is to derive completely deterministic passwords from a BIP-85 seed.
|
||||
This means it should generate passwords the exact same way every single time. Salts would break this functionality and is not appropriate for this software's use case.
|
||||
|
||||
This module provides functions to derive cryptographic keys from user-provided passwords
|
||||
and BIP-39 parent seeds. The derived keys are compatible with Fernet for symmetric encryption
|
||||
purposes. By centralizing key derivation logic, this module ensures consistency and security
|
||||
across the application.
|
||||
|
||||
Ensure that all dependencies are installed and properly configured in your environment.
|
||||
"""
|
||||
|
||||
import os
|
||||
import hashlib
|
||||
import base64
|
||||
import unicodedata
|
||||
import logging
|
||||
import traceback
|
||||
from typing import Union
|
||||
from bip_utils import Bip39SeedGenerator
|
||||
from local_bip85.bip85 import BIP85
|
||||
from monstr.encrypt import Keys
|
||||
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
|
||||
# Instantiate the logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def derive_key_from_password(password: str, iterations: int = 100_000) -> bytes:
|
||||
"""
|
||||
@@ -213,20 +56,20 @@ def derive_key_from_password(password: str, iterations: int = 100_000) -> bytes:
|
||||
|
||||
if len(password) < 8:
|
||||
logger.warning("Password length is less than recommended (8 characters).")
|
||||
|
||||
|
||||
# Normalize the password to NFKD form and encode to UTF-8
|
||||
normalized_password = unicodedata.normalize('NFKD', password).strip()
|
||||
password_bytes = normalized_password.encode('utf-8')
|
||||
normalized_password = unicodedata.normalize("NFKD", password).strip()
|
||||
password_bytes = normalized_password.encode("utf-8")
|
||||
|
||||
try:
|
||||
# Derive the key using PBKDF2-HMAC-SHA256
|
||||
logger.debug("Starting key derivation from password.")
|
||||
key = hashlib.pbkdf2_hmac(
|
||||
hash_name='sha256',
|
||||
hash_name="sha256",
|
||||
password=password_bytes,
|
||||
salt=b'', # No salt for deterministic key derivation
|
||||
salt=b"", # No salt for deterministic key derivation
|
||||
iterations=iterations,
|
||||
dklen=32 # 256-bit key for Fernet
|
||||
dklen=32, # 256-bit key for Fernet
|
||||
)
|
||||
logger.debug(f"Derived key (hex): {key.hex()}")
|
||||
|
||||
@@ -241,6 +84,7 @@ def derive_key_from_password(password: str, iterations: int = 100_000) -> bytes:
|
||||
logger.error(traceback.format_exc()) # Log full traceback
|
||||
raise
|
||||
|
||||
|
||||
def derive_key_from_parent_seed(parent_seed: str, fingerprint: str = None) -> bytes:
|
||||
"""
|
||||
Derives a 32-byte cryptographic key from a BIP-39 parent seed using HKDF.
|
||||
@@ -253,34 +97,37 @@ def derive_key_from_parent_seed(parent_seed: str, fingerprint: str = None) -> by
|
||||
try:
|
||||
# Generate seed bytes from mnemonic
|
||||
seed = Bip39SeedGenerator(parent_seed).Generate()
|
||||
|
||||
|
||||
# If a fingerprint is provided, use it to differentiate the derivation
|
||||
if fingerprint:
|
||||
# Convert fingerprint to a stable integer index
|
||||
index = int(hashlib.sha256(fingerprint.encode()).hexdigest(), 16) % (2**31)
|
||||
info = f'password-manager-{index}'.encode() # Unique info for HKDF
|
||||
info = f"password-manager-{index}".encode() # Unique info for HKDF
|
||||
else:
|
||||
info = b'password-manager'
|
||||
|
||||
info = b"password-manager"
|
||||
|
||||
# Derive key using HKDF
|
||||
hkdf = HKDF(
|
||||
algorithm=hashes.SHA256(),
|
||||
length=32,
|
||||
salt=None, # No salt for deterministic derivation
|
||||
info=info,
|
||||
backend=default_backend()
|
||||
backend=default_backend(),
|
||||
)
|
||||
derived_key = hkdf.derive(seed)
|
||||
|
||||
|
||||
if len(derived_key) != 32:
|
||||
raise ValueError(f"Derived key length is {len(derived_key)} bytes; expected 32 bytes.")
|
||||
|
||||
raise ValueError(
|
||||
f"Derived key length is {len(derived_key)} bytes; expected 32 bytes."
|
||||
)
|
||||
|
||||
return derived_key
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to derive key using HKDF: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
raise
|
||||
|
||||
|
||||
class KeyManager:
|
||||
def __init__(self, parent_seed: str, fingerprint: str = None):
|
||||
self.parent_seed = parent_seed
|
||||
@@ -301,13 +148,15 @@ class KeyManager:
|
||||
"""
|
||||
# Use a derivation path that includes the fingerprint
|
||||
# Convert fingerprint to an integer index (e.g., using a hash function)
|
||||
index = int(hashlib.sha256(self.fingerprint.encode()).hexdigest(), 16) % (2**31) if self.fingerprint else 0
|
||||
index = (
|
||||
int(hashlib.sha256(self.fingerprint.encode()).hexdigest(), 16) % (2**31)
|
||||
if self.fingerprint
|
||||
else 0
|
||||
)
|
||||
|
||||
# Derive entropy for Nostr key (32 bytes)
|
||||
entropy_bytes = self.bip85.derive_entropy(
|
||||
app=BIP85.Applications.ENTROPY,
|
||||
index=index,
|
||||
size=32
|
||||
app=BIP85.Applications.ENTROPY, index=index, size=32
|
||||
)
|
||||
|
||||
# Generate Nostr key pair from entropy
|
||||
|
Reference in New Issue
Block a user