Merge pull request #613 from PR0M3TH3AN/codex/refactor-input-handling-in-manager.py

Refactor password change and unlock

src/seedpass/api.py

@@ -554,11 +554,13 @@ def backup_parent_seed(
 
 
 @app.post("/api/v1/change-password")
-def change_password(authorization: str | None = Header(None)) -> dict[str, str]:
+def change_password(
+    data: dict, authorization: str | None = Header(None)
+) -> dict[str, str]:
     """Change the master password for the active profile."""
     _check_token(authorization)
     assert _pm is not None
-    _pm.change_password()
+    _pm.change_password(data.get("old", ""), data.get("new", ""))
     return {"status": "ok"}
 
 

src/seedpass/cli.py

@@ -384,7 +384,27 @@ def vault_import(
 def vault_change_password(ctx: typer.Context) -> None:
     """Change the master password used for encryption."""
     pm = _get_pm(ctx)
-    pm.change_password()
+    old_pw = typer.prompt("Current password", hide_input=True)
+    new_pw = typer.prompt("New password", hide_input=True, confirmation_prompt=True)
+    try:
+        pm.change_password(old_pw, new_pw)
+    except Exception as exc:  # pragma: no cover - pass through errors
+        typer.echo(f"Error: {exc}")
+        raise typer.Exit(code=1)
+    typer.echo("Password updated")
+
+
+@vault_app.command("unlock")
+def vault_unlock(ctx: typer.Context) -> None:
+    """Unlock the vault for the active profile."""
+    pm = _get_pm(ctx)
+    password = typer.prompt("Master password", hide_input=True)
+    try:
+        duration = pm.unlock_vault(password)
+    except Exception as exc:  # pragma: no cover - pass through errors
+        typer.echo(f"Error: {exc}")
+        raise typer.Exit(code=1)
+    typer.echo(f"Unlocked in {duration:.2f}s")
 
 
 @vault_app.command("lock")

src/seedpass/core/manager.py

@@ -276,28 +276,31 @@ class PasswordManager:
         self.config_manager = None
         self.locked = True
 
-    def unlock_vault(self, password: Optional[str] = None) -> None:
-        """Unlock the vault using ``password`` without prompting if provided."""
+    def unlock_vault(self, password: str) -> float:
+        """Unlock the vault using the provided ``password``.
+
+        Parameters
+        ----------
+        password:
+            Master password for the active profile.
+
+        Returns
+        -------
+        float
+            Duration of the unlock process in seconds.
+        """
         start = time.perf_counter()
         if not self.fingerprint_dir:
             raise ValueError("Fingerprint directory not set")
-        if password is None:
-            self.setup_encryption_manager(self.fingerprint_dir)
-        else:
-            self.setup_encryption_manager(self.fingerprint_dir, password)
+        self.setup_encryption_manager(self.fingerprint_dir, password)
         self.initialize_bip85()
         self.initialize_managers()
         self.locked = False
         self.update_activity()
         self.last_unlock_duration = time.perf_counter() - start
-        print(
-            colored(
-                f"Vault unlocked in {self.last_unlock_duration:.2f} seconds",
-                "yellow",
-            )
-        )
         if getattr(self, "verbose_timing", False):
             logger.info("Vault unlocked in %.2f seconds", self.last_unlock_duration)
+        return self.last_unlock_duration
 
     def initialize_fingerprint_manager(self):
         """
@@ -3884,15 +3887,11 @@ class PasswordManager:
             print(colored(f"Error: Failed to store hashed password: {e}", "red"))
             raise
 
-    def change_password(self) -> None:
+    def change_password(self, old_password: str, new_password: str) -> None:
         """Change the master password used for encryption."""
         try:
-            current = prompt_existing_password("Enter your current master password: ")
-            if not self.verify_password(current):
-                print(colored("Incorrect password.", "red"))
-                return
-
-            new_password = prompt_for_password()
+            if not self.verify_password(old_password):
+                raise ValueError("Incorrect password")
 
             # Load data with existing encryption manager
             index_data = self.vault.load_index()
@@ -3927,8 +3926,6 @@ class PasswordManager:
                 parent_seed=getattr(self, "parent_seed", None),
             )
 
-            print(colored("Master password changed successfully.", "green"))
-
             # Push a fresh backup to Nostr so the newly encrypted index is
             # stored remotely. Include a tag to mark the password change.
             try:
@@ -3940,7 +3937,7 @@ class PasswordManager:
                 )
         except Exception as e:
             logging.error(f"Failed to change password: {e}", exc_info=True)
-            print(colored(f"Error: Failed to change password: {e}", "red"))
+            raise
 
     def get_profile_stats(self) -> dict:
         """Return various statistics about the current seed profile."""

src/tests/test_api.py

@@ -179,12 +179,16 @@ def test_change_password_route(client):
     cl, token = client
     called = {}
 
-    api._pm.change_password = lambda: called.setdefault("called", True)
+    api._pm.change_password = lambda o, n: called.setdefault("called", (o, n))
     headers = {"Authorization": f"Bearer {token}", "Origin": "http://example.com"}
-    res = cl.post("/api/v1/change-password", headers=headers)
+    res = cl.post(
+        "/api/v1/change-password",
+        headers=headers,
+        json={"old": "old", "new": "new"},
+    )
     assert res.status_code == 200
     assert res.json() == {"status": "ok"}
-    assert called.get("called") is True
+    assert called.get("called") == ("old", "new")
     assert res.headers.get("access-control-allow-origin") == "http://example.com"
 
 

src/tests/test_api_new_endpoints.py

@@ -291,8 +291,8 @@ def test_vault_lock_endpoint(client):
     assert res.json() == {"status": "locked"}
     assert called.get("locked") is True
     assert api._pm.locked is True
-    api._pm.unlock_vault = lambda: setattr(api._pm, "locked", False)
-    api._pm.unlock_vault()
+    api._pm.unlock_vault = lambda pw: setattr(api._pm, "locked", False)
+    api._pm.unlock_vault("pw")
     assert api._pm.locked is False
 
 

src/tests/test_cli_doc_examples.py

@@ -40,7 +40,7 @@ class DummyPM:
         self.handle_display_totp_codes = lambda: None
         self.handle_export_database = lambda path: None
         self.handle_import_database = lambda path: None
-        self.change_password = lambda: None
+        self.change_password = lambda *a, **kw: None
         self.lock_vault = lambda: None
         self.get_profile_stats = lambda: {"n": 1}
         self.handle_backup_reveal_parent_seed = lambda path=None: None

src/tests/test_password_change.py

@@ -36,14 +36,9 @@ def test_change_password_triggers_nostr_backup(monkeypatch):
         pm.store_hashed_password = lambda pw: None
         pm.verify_password = lambda pw: True
 
-        monkeypatch.setattr(
-            "seedpass.core.manager.prompt_existing_password", lambda *_: "old"
-        )
-        monkeypatch.setattr("seedpass.core.manager.prompt_for_password", lambda: "new")
-
         with patch("seedpass.core.manager.NostrClient") as MockClient:
             mock_instance = MockClient.return_value
             mock_instance.publish_snapshot = AsyncMock(return_value=(None, "abcd"))
             pm.nostr_client = mock_instance
-            pm.change_password()
+            pm.change_password("old", "new")
             mock_instance.publish_snapshot.assert_called_once()

src/tests/test_password_unlock_after_change.py

@@ -71,7 +71,7 @@ def test_password_change_and_unlock(monkeypatch):
             ),
         )
 
-        pm.change_password()
+        pm.change_password(old_pw, new_pw)
         pm.lock_vault()
 
         monkeypatch.setattr(
@@ -81,7 +81,7 @@ def test_password_change_and_unlock(monkeypatch):
         monkeypatch.setattr(PasswordManager, "initialize_managers", lambda self: None)
         monkeypatch.setattr(PasswordManager, "sync_index_from_nostr", lambda self: None)
 
-        pm.unlock_vault()
+        pm.unlock_vault(new_pw)
 
         assert pm.parent_seed == SEED
         assert pm.verify_password(new_pw)

src/tests/test_typer_cli.py

@@ -126,14 +126,14 @@ def test_vault_import_triggers_sync(monkeypatch, tmp_path):
 def test_vault_change_password(monkeypatch):
     called = {}
 
-    def change_pw():
-        called["called"] = True
+    def change_pw(old, new):
+        called["args"] = (old, new)
 
     pm = SimpleNamespace(change_password=change_pw, select_fingerprint=lambda fp: None)
     monkeypatch.setattr(cli, "PasswordManager", lambda: pm)
-    result = runner.invoke(app, ["vault", "change-password"])
+    result = runner.invoke(app, ["vault", "change-password"], input="old\nnew\nnew\n")
     assert result.exit_code == 0
-    assert called.get("called") is True
+    assert called.get("args") == ("old", "new")
 
 
 def test_vault_lock(monkeypatch):

src/tests/test_unlock_sync.py

@@ -22,7 +22,7 @@ def test_unlock_triggers_sync(monkeypatch, tmp_path):
 
     monkeypatch.setattr(PasswordManager, "sync_index_from_nostr", fake_sync)
 
-    pm.unlock_vault()
+    pm.unlock_vault("pw")
     pm.start_background_sync()
     time.sleep(0.05)
 

src/tests/test_verbose_timing.py

@@ -8,7 +8,7 @@ from helpers import dummy_nostr_client
 def test_unlock_vault_logs_time(monkeypatch, caplog, tmp_path):
     pm = PasswordManager.__new__(PasswordManager)
     pm.fingerprint_dir = tmp_path
-    pm.setup_encryption_manager = lambda path: None
+    pm.setup_encryption_manager = lambda path, pw=None: None
     pm.initialize_bip85 = lambda: None
     pm.initialize_managers = lambda: None
     pm.update_activity = lambda: None
@@ -16,7 +16,7 @@ def test_unlock_vault_logs_time(monkeypatch, caplog, tmp_path):
     caplog.set_level(logging.INFO, logger="seedpass.core.manager")
     times = iter([0.0, 1.0])
     monkeypatch.setattr("seedpass.core.manager.time.perf_counter", lambda: next(times))
-    pm.unlock_vault()
+    pm.unlock_vault("pw")
     assert "Vault unlocked in 1.00 seconds" in caplog.text