Make TUI v3 the default and add index0 foundations

This commit is contained in:
thePR0M3TH3AN committed 2026-03-05 21:52:39 -05:00
1 parent 0fa8aebc1b
commit e8a2b81e7b
17 files changed
+3005 -76

No files matched your search

+2 -1
View File
@@ -1,5 +1,5 @@
from dataclasses import dataclass
from typing import List, Optional
from typing import Any, List, Optional
# Event kind constants used for SeedPass backups
KIND_MANIFEST = 30070
@@ -26,3 +26,4 @@ class Manifest:
chunks: List[ChunkMeta]
delta_since: Optional[int] = None
nonce: Optional[str] = None
index0: Optional[dict[str, Any]] = None
+5
View File
@@ -114,6 +114,7 @@ class SnapshotHandler:
"chunks": [meta.__dict__ for meta in manifest.chunks],
"delta_since": manifest.delta_since,
"nonce": manifest.nonce,
"index0": manifest.index0,
}
)
@@ -151,6 +152,9 @@ class SnapshotHandler:
else None
),
nonce=data.get("nonce"),
index0=(
data.get("index0") if isinstance(data.get("index0"), dict) else None
),
)
except Exception:
return None
@@ -376,6 +380,7 @@ class SnapshotHandler:
],
"delta_since": self.current_manifest.delta_since,
"nonce": self.current_manifest.nonce,
"index0": self.current_manifest.index0,
}
)
manifest_event = (
+45 -42
View File
@@ -14,7 +14,7 @@ from .common import _get_entry_service, _get_services
from seedpass.core.errors import SeedPassError
from constants import GUI_BACKEND_CONFIG
from seedpass.tui_v2.app import check_tui2_runtime, launch_tui2
from seedpass.tui_v3 import launch_tui3
from seedpass.tui_v3 import check_tui3_runtime, launch_tui3
app = typer.Typer(
help=(
@@ -49,11 +49,11 @@ deterministic_totp_option = typer.Option(
legacy_tui_option = typer.Option(
False,
"--legacy-tui",
help="Use legacy interactive TUI (temporary fallback during TUI v2 cutover)",
help="Use legacy interactive TUI explicitly",
)
tui_version_option = typer.Option(
"v2",
"v3",
"--tui",
help="Select TUI version: legacy, v2, v3",
)
@@ -154,6 +154,36 @@ def _launch_legacy_tui(*, fingerprint: Optional[str]) -> typer.Exit:
return typer.Exit(legacy_main(fingerprint=fingerprint))
def _launch_tui3_or_exit(
*,
fingerprint: Optional[str],
services: Optional[dict[str, object]] = None,
) -> None:
kwargs: dict[str, object] = {"fingerprint": fingerprint}
if services is not None:
kwargs.update(
{
"entry_service_factory": lambda: services["entry"],
"profile_service_factory": lambda: services["profile"],
"config_service_factory": lambda: services["config"],
"nostr_service_factory": lambda: services["nostr"],
"sync_service_factory": lambda: services["sync"],
"utility_service_factory": lambda: services["utility"],
"vault_service_factory": lambda: services["vault"],
"semantic_service_factory": lambda: services["semantic"],
}
)
launched = launch_tui3(**kwargs)
if launched:
return
typer.echo(
"TUI v3 runtime unavailable. Run `seedpass tui3 --check` for diagnostics, "
"`seedpass tui2` to launch TUI v2, or `seedpass legacy` for the legacy interface.",
err=True,
)
raise typer.Exit(1)
@app.callback(invoke_without_command=True)
def main(
ctx: typer.Context,
@@ -188,27 +218,17 @@ def main(
if legacy_tui or tui_version == "legacy":
raise _launch_legacy_tui(fingerprint=fingerprint)
if tui_version == "v3":
_launch_tui3_or_exit(fingerprint=fingerprint)
return
services, preflight_error = _prime_tui2_service(ctx)
if preflight_error is not None:
typer.echo(
f"TUI {tui_version} preflight failed; falling back to legacy TUI. "
f"Details: {preflight_error}"
f"TUI {tui_version} preflight failed: {preflight_error}",
err=True,
)
raise _launch_legacy_tui(fingerprint=fingerprint)
if tui_version == "v3":
launch_tui3(
fingerprint=fingerprint,
entry_service_factory=lambda: services["entry"],
profile_service_factory=lambda: services["profile"],
config_service_factory=lambda: services["config"],
nostr_service_factory=lambda: services["nostr"],
sync_service_factory=lambda: services["sync"],
utility_service_factory=lambda: services["utility"],
vault_service_factory=lambda: services["vault"],
semantic_service_factory=lambda: services["semantic"],
)
return
raise typer.Exit(1)
launched = launch_tui2(
fingerprint=fingerprint,
@@ -225,10 +245,10 @@ def main(
return
typer.echo(
f"TUI {tui_version} runtime unavailable; falling back to legacy TUI. "
"Run `seedpass tui2 --check` for diagnostics."
"TUI v2 runtime unavailable. Run `seedpass tui2 --check` for diagnostics, "
"or use `seedpass legacy` for the legacy interface."
)
raise _launch_legacy_tui(fingerprint=fingerprint)
raise typer.Exit(1)
@app.command("lock")
@@ -310,30 +330,13 @@ def tui3(
help="Check whether TUI v3 runtime dependencies are available",
),
) -> None:
"""Launch the experimental TUI v3 scratch architecture."""
"""Launch the current TUI v3 interface."""
if check:
from seedpass.tui_v3 import check_tui3_runtime
typer.echo(json.dumps(check_tui3_runtime(), indent=2, sort_keys=True))
return
fingerprint = (ctx.obj or {}).get("fingerprint")
services, preflight_error = _prime_tui2_service(ctx)
if preflight_error is not None:
typer.echo(f"TUI v3 preflight failed: {preflight_error}", err=True)
raise typer.Exit(1)
launch_tui3(
fingerprint=fingerprint,
entry_service_factory=lambda: services["entry"],
profile_service_factory=lambda: services["profile"],
config_service_factory=lambda: services["config"],
nostr_service_factory=lambda: services["nostr"],
sync_service_factory=lambda: services["sync"],
utility_service_factory=lambda: services["utility"],
vault_service_factory=lambda: services["vault"],
semantic_service_factory=lambda: services["semantic"],
)
_launch_tui3_or_exit(fingerprint=fingerprint)
@app.command("legacy")
+5 -3
View File
@@ -14,11 +14,12 @@ def build_capabilities() -> dict[str, Any]:
"product": "SeedPass",
"interfaces": {
"cli": {
"default_interface": "tui3",
"root_commands": [
"lock",
"gui",
"tui2",
"tui3",
"tui2",
"legacy",
"capabilities",
"entry",
@@ -266,8 +267,9 @@ def build_capabilities() -> dict[str, Any]:
"Run `seedpass --help` for top-level commands.",
"Run `seedpass <group> --help` for detailed options.",
"Use `seedpass capabilities --format json` for machine-readable discovery.",
"Run `seedpass tui2 --check` before adopting Textual-based TUI v2.",
"Use `seedpass --legacy-tui` to force the legacy interface during cutover.",
"Run `seedpass` or `seedpass tui3` to launch the current TUI v3.",
"Run `seedpass tui2 --check` before using the maintained-but-non-default TUI v2.",
"Use `seedpass legacy` or `seedpass --legacy-tui` for the legacy interface.",
"In TUI v2, use `?` for help, `j` to jump to entry id, `1/2/3` to focus panes, `Ctrl+P` for palette, `p/n` for paging, and `x` to retry.",
"For graph workflows, start with `seedpass entry link-add --help` and `seedpass entry links --help`.",
"For semantic KB retrieval, run `seedpass semantic --help` and check `seedpass semantic status`.",
+2
View File
@@ -29,6 +29,7 @@ from mnemonic import Mnemonic
from utils.password_prompt import prompt_existing_password
from utils.key_derivation import KdfConfig, CURRENT_KDF_VERSION
from .errors import DecryptionError
from .index0 import ensure_index0_payload
from .sync_conflict import merge_index_payloads
# Instantiate the logger
@@ -479,6 +480,7 @@ class EncryptionManager:
data = json_lib.loads(decrypted)
else:
data = json_lib.loads(decrypted.decode("utf-8"))
data = ensure_index0_payload(data)
existing_file = self.resolve_relative_path(relative_path)
if merge and existing_file.exists():
current = self.load_json_data(relative_path)
+437
View File
@@ -0,0 +1,437 @@
from __future__ import annotations
import hashlib
import json
from typing import Any
INDEX0_SCHEMA_VERSION = 1
LOCAL_ONLY_VIEW_TYPES = {"hot_nodes", "conversation_index", "semantic_neighbors"}
def _canonical(value: Any) -> str:
return json.dumps(value, sort_keys=True, separators=(",", ":"))
def _sha256_hex(value: Any) -> str:
if isinstance(value, str):
payload = value.encode("utf-8")
else:
payload = _canonical(value).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def _normalize_string(value: Any) -> str:
return str(value).strip()
def _safe_int(raw: Any, *, default: int = 0) -> int:
try:
return int(raw)
except Exception:
return int(default)
def _normalize_tags(raw_tags: Any) -> list[str]:
if not isinstance(raw_tags, list):
return []
out = sorted({_normalize_string(tag) for tag in raw_tags if _normalize_string(tag)})
return out
def _normalize_links(raw_links: Any) -> list[dict[str, Any]]:
if not isinstance(raw_links, list):
return []
out: dict[str, dict[str, Any]] = {}
for item in raw_links:
if not isinstance(item, dict):
continue
target_id = _normalize_string(item.get("target_id", ""))
relation = _normalize_string(item.get("relation", ""))
note = _normalize_string(item.get("note", ""))
if not target_id or not relation:
continue
normalized = {"target_id": target_id, "relation": relation}
if note:
normalized["note"] = note
out[_canonical(normalized)] = normalized
return [out[key] for key in sorted(out.keys())]
def _normalize_mapping(raw: Any) -> dict[str, Any]:
return dict(raw) if isinstance(raw, dict) else {}
def normalize_index0(raw: Any) -> dict[str, Any]:
data = dict(raw) if isinstance(raw, dict) else {}
normalized = {
"schema_version": max(
INDEX0_SCHEMA_VERSION,
_safe_int(data.get("schema_version", INDEX0_SCHEMA_VERSION), default=1),
),
"events": {},
"checkpoints": {},
"canonical_views": {},
"view_manifest": {},
"heads": {},
"stats": {},
}
for event_id, event in _normalize_mapping(data.get("events")).items():
normalized_event = normalize_index0_event(event)
if normalized_event is None:
continue
normalized["events"][str(event_id)] = normalized_event
for checkpoint_id, checkpoint in _normalize_mapping(
data.get("checkpoints")
).items():
normalized_checkpoint = normalize_index0_checkpoint(checkpoint)
if normalized_checkpoint is None:
continue
normalized["checkpoints"][str(checkpoint_id)] = normalized_checkpoint
view_manifest = normalize_view_manifest(data.get("view_manifest"))
normalized["view_manifest"] = view_manifest
for view_id, view in _normalize_mapping(data.get("canonical_views")).items():
normalized_view = normalize_canonical_view(view, view_manifest=view_manifest)
if normalized_view is None:
continue
normalized["canonical_views"][str(view_id)] = normalized_view
for writer_id, head in _normalize_mapping(data.get("heads")).items():
normalized_head = normalize_head(head)
if normalized_head is None:
continue
normalized["heads"][str(writer_id)] = normalized_head
normalized["stats"] = recompute_index0_stats(normalized)
return normalized
def ensure_index0_payload(data: dict[str, Any]) -> dict[str, Any]:
out = dict(data) if isinstance(data, dict) else {}
system = dict(out.get("_system")) if isinstance(out.get("_system"), dict) else {}
system["index0"] = normalize_index0(system.get("index0"))
out["_system"] = system
return out
def normalize_view_manifest(raw: Any) -> dict[str, Any]:
data = dict(raw) if isinstance(raw, dict) else {}
canonical_view_types = sorted(
{
_normalize_string(item)
for item in data.get("canonical_view_types", [])
if _normalize_string(item)
}
)
local_only_view_types = sorted(
{
_normalize_string(item)
for item in data.get("local_only_view_types", [])
if _normalize_string(item)
}
)
builder_versions = {}
raw_versions = data.get("builder_versions", {})
if isinstance(raw_versions, dict):
for key, value in raw_versions.items():
key_str = _normalize_string(key)
if not key_str:
continue
builder_versions[key_str] = _safe_int(value, default=1)
return {
"version": max(1, _safe_int(data.get("version", 1), default=1)),
"canonical_view_types": canonical_view_types,
"local_only_view_types": local_only_view_types,
"builder_versions": dict(sorted(builder_versions.items())),
}
def compute_event_hash(event: dict[str, Any]) -> str:
body = dict(event)
body.pop("integrity_hash", None)
return _sha256_hex(body)
def compute_event_id(event: dict[str, Any]) -> str:
return f"e:{compute_event_hash(event)}"
def compute_head_hash(event: dict[str, Any]) -> str:
marker = {
"event_id": _normalize_string(event.get("event_id", "")),
"integrity_hash": _normalize_string(event.get("integrity_hash", "")),
"prev_hash": _normalize_string(event.get("prev_hash", "")),
"writer_id": _normalize_string(event.get("writer_id", "")),
}
return _sha256_hex(marker)
def compute_checkpoint_hash(checkpoint: dict[str, Any]) -> str:
body = dict(checkpoint)
body.pop("summary_hash", None)
return _sha256_hex(body)
def compute_view_hash(view: dict[str, Any]) -> str:
body = dict(view)
body.pop("view_hash", None)
return _sha256_hex(body)
def normalize_index0_event(raw: Any) -> dict[str, Any] | None:
if not isinstance(raw, dict):
return None
modified_ts = _safe_int(raw.get("modified_ts", 0), default=0)
if modified_ts <= 0:
return None
event = {
"event_type": _normalize_string(raw.get("event_type", "")),
"subject_type": _normalize_string(raw.get("subject_type", "")),
"subject_id": _normalize_string(raw.get("subject_id", "")),
"subject_kind": _normalize_string(raw.get("subject_kind", "")),
"scope_path": _normalize_string(raw.get("scope_path", "")),
"actor_type": _normalize_string(raw.get("actor_type", "")),
"actor_id": _normalize_string(raw.get("actor_id", "")),
"writer_id": _normalize_string(raw.get("writer_id", "")),
"modified_ts": modified_ts,
"prev_hash": _normalize_string(raw.get("prev_hash", "")),
"classification": _normalize_string(raw.get("classification", "internal"))
or "internal",
"partition": _normalize_string(raw.get("partition", "standard")) or "standard",
"payload_ref": _normalize_mapping(raw.get("payload_ref")),
"links": _normalize_links(raw.get("links")),
"tags": _normalize_tags(raw.get("tags")),
"visibility": _normalize_string(raw.get("visibility", "private")) or "private",
}
for key in ("policy_ref", "source", "source_event_id", "summary"):
value = _normalize_string(raw.get(key, ""))
if value:
event[key] = value
required = (
event["event_type"],
event["subject_type"],
event["subject_id"],
event["writer_id"],
)
if not all(required):
return None
event["integrity_hash"] = _normalize_string(raw.get("integrity_hash", ""))
expected_hash = compute_event_hash(event)
if not event["integrity_hash"]:
event["integrity_hash"] = expected_hash
event["event_id"] = (
_normalize_string(raw.get("event_id", "")) or f"e:{expected_hash}"
)
return event
def normalize_index0_checkpoint(raw: Any) -> dict[str, Any] | None:
if not isinstance(raw, dict):
return None
checkpoint_id = _normalize_string(raw.get("checkpoint_id", ""))
writer_id = _normalize_string(raw.get("writer_id", ""))
window_type = _normalize_string(raw.get("window_type", ""))
window_key = _normalize_string(raw.get("window_key", ""))
modified_ts = _safe_int(raw.get("modified_ts", 0), default=0)
if (
not checkpoint_id
or not writer_id
or not window_type
or not window_key
or modified_ts <= 0
):
return None
rollup_raw = _normalize_mapping(raw.get("rollup"))
events_by_type = _normalize_mapping(rollup_raw.get("events_by_type"))
subjects_by_kind = _normalize_mapping(rollup_raw.get("subjects_by_kind"))
subjects = _normalize_tags(rollup_raw.get("subjects"))
checkpoint = {
"checkpoint_id": checkpoint_id,
"window_type": window_type,
"window_key": window_key,
"writer_id": writer_id,
"window_start_ts": _safe_int(raw.get("window_start_ts", 0), default=0),
"window_end_ts": _safe_int(raw.get("window_end_ts", 0), default=0),
"event_count": max(0, _safe_int(raw.get("event_count", 0), default=0)),
"head_hash": _normalize_string(raw.get("head_hash", "")),
"rollup": {
"events_by_type": {
_normalize_string(k): _safe_int(v, default=0)
for k, v in sorted(
events_by_type.items(), key=lambda item: str(item[0])
)
if _normalize_string(k)
},
"subjects_by_kind": {
_normalize_string(k): _safe_int(v, default=0)
for k, v in sorted(
subjects_by_kind.items(), key=lambda item: str(item[0])
)
if _normalize_string(k)
},
"subjects": subjects,
},
"modified_ts": modified_ts,
}
checkpoint["summary_hash"] = _normalize_string(
raw.get("summary_hash", "")
) or compute_checkpoint_hash(checkpoint)
return checkpoint
def normalize_canonical_view(
raw: Any, *, view_manifest: dict[str, Any] | None = None
) -> dict[str, Any] | None:
if not isinstance(raw, dict):
return None
view_type = _normalize_string(raw.get("view_type", ""))
modified_ts = _safe_int(raw.get("modified_ts", 0), default=0)
if not view_type or modified_ts <= 0:
return None
if view_type in LOCAL_ONLY_VIEW_TYPES:
return None
if view_manifest:
local_only = set(view_manifest.get("local_only_view_types", []))
if view_type in local_only:
return None
view = {
"view_id": _normalize_string(raw.get("view_id", "")),
"view_type": view_type,
"scope_path": _normalize_string(raw.get("scope_path", "")),
"source_checkpoint_ids": _normalize_tags(raw.get("source_checkpoint_ids")),
"source_event_ids": _normalize_tags(raw.get("source_event_ids")),
"data": _normalize_mapping(raw.get("data")),
"modified_ts": modified_ts,
}
if not view["view_id"]:
return None
view["view_hash"] = _normalize_string(
raw.get("view_hash", "")
) or compute_view_hash(view)
return view
def normalize_head(raw: Any) -> dict[str, Any] | None:
if not isinstance(raw, dict):
return None
event_id = _normalize_string(raw.get("event_id", ""))
head_hash = _normalize_string(raw.get("head_hash", ""))
modified_ts = _safe_int(raw.get("modified_ts", 0), default=0)
if not event_id or not head_hash or modified_ts <= 0:
return None
return {"event_id": event_id, "head_hash": head_hash, "modified_ts": modified_ts}
def _prefer_hashed_record(
current: dict[str, Any], incoming: dict[str, Any], *, hash_field: str
) -> bool:
cur_ts = _safe_int(current.get("modified_ts", 0), default=0)
inc_ts = _safe_int(incoming.get("modified_ts", 0), default=0)
if inc_ts != cur_ts:
return inc_ts > cur_ts
return _normalize_string(incoming.get(hash_field, "")) > _normalize_string(
current.get(hash_field, "")
)
def merge_system_index0(current: Any, incoming: Any) -> dict[str, Any]:
cur = normalize_index0(current)
inc = normalize_index0(incoming)
events = dict(cur["events"])
for event_id, event in inc["events"].items():
existing = events.get(event_id)
if existing is None:
events[event_id] = event
continue
if _canonical(existing) == _canonical(event):
continue
if _sha256_hex(event) > _sha256_hex(existing):
events[event_id] = event
checkpoints = dict(cur["checkpoints"])
for checkpoint_id, checkpoint in inc["checkpoints"].items():
existing = checkpoints.get(checkpoint_id)
if existing is None:
checkpoints[checkpoint_id] = checkpoint
continue
if _prefer_hashed_record(existing, checkpoint, hash_field="summary_hash"):
checkpoints[checkpoint_id] = checkpoint
view_manifest = normalize_view_manifest(
{
"version": max(
_safe_int(cur["view_manifest"].get("version", 1), default=1),
_safe_int(inc["view_manifest"].get("version", 1), default=1),
),
"canonical_view_types": sorted(
set(cur["view_manifest"].get("canonical_view_types", []))
| set(inc["view_manifest"].get("canonical_view_types", []))
),
"local_only_view_types": sorted(
set(cur["view_manifest"].get("local_only_view_types", []))
| set(inc["view_manifest"].get("local_only_view_types", []))
),
"builder_versions": {
**cur["view_manifest"].get("builder_versions", {}),
**inc["view_manifest"].get("builder_versions", {}),
},
}
)
views = dict(cur["canonical_views"])
for view_id, view in inc["canonical_views"].items():
normalized_view = normalize_canonical_view(view, view_manifest=view_manifest)
if normalized_view is None:
continue
existing = views.get(view_id)
if existing is None:
views[view_id] = normalized_view
continue
if _prefer_hashed_record(existing, normalized_view, hash_field="view_hash"):
views[view_id] = normalized_view
heads = dict(cur["heads"])
for writer_id, head in inc["heads"].items():
existing = heads.get(writer_id)
if existing is None or _prefer_hashed_record(
existing, head, hash_field="head_hash"
):
heads[writer_id] = head
merged = {
"schema_version": max(cur["schema_version"], inc["schema_version"]),
"events": dict(sorted(events.items())),
"checkpoints": dict(sorted(checkpoints.items())),
"canonical_views": dict(sorted(views.items())),
"view_manifest": view_manifest,
"heads": dict(sorted(heads.items())),
"stats": {},
}
merged["stats"] = recompute_index0_stats(merged)
return merged
def recompute_index0_stats(index0: dict[str, Any]) -> dict[str, Any]:
events = _normalize_mapping(index0.get("events"))
checkpoints = _normalize_mapping(index0.get("checkpoints"))
heads = _normalize_mapping(index0.get("heads"))
stats = {
"event_count": len(events),
"checkpoint_count": len(checkpoints),
"writer_count": len(heads),
"last_compaction_ts": 0,
"last_validation_ts": 0,
}
checkpoint_ts = [
_safe_int(value.get("modified_ts", 0), default=0)
for value in checkpoints.values()
if isinstance(value, dict)
]
head_ts = [
_safe_int(value.get("modified_ts", 0), default=0)
for value in heads.values()
if isinstance(value, dict)
]
if checkpoint_ts:
stats["last_compaction_ts"] = max(checkpoint_ts)
if head_ts:
stats["last_validation_ts"] = max(head_ts)
return stats
+134 -2
View File
@@ -245,7 +245,11 @@ class PasswordManager:
deterministic_totp: bool = False
def __init__(
self, fingerprint: Optional[str] = None, *, password: Optional[str] = None
self,
fingerprint: Optional[str] = None,
*,
password: Optional[str] = None,
bootstrap_only: bool = False,
) -> None:
"""Initialize the PasswordManager.
@@ -321,13 +325,16 @@ class PasswordManager:
if fingerprint:
# Load the specified profile without prompting
self.select_fingerprint(fingerprint, password=password)
else:
elif not bootstrap_only:
# Ensure a parent seed is set up before accessing the fingerprint directory
self.setup_parent_seed()
# Set the current fingerprint directory after selection
self.fingerprint_dir = (
self.fingerprint_manager.get_current_fingerprint_dir()
)
else:
self.current_fingerprint = None
self.fingerprint_dir = None
@requires_unlocked
def get_bip85_entropy(
@@ -1017,6 +1024,77 @@ class PasswordManager:
print(colored(f"Error: Profile recovery failed: {exc}", "red"))
return False
def recover_profile_with_blank_index_data(
self,
*,
fingerprint: str,
parent_seed: str,
password: str,
) -> bool:
"""Recover ``fingerprint`` with ``parent_seed`` and a blank local index."""
fingerprints = self.fingerprint_manager.list_fingerprints()
if fingerprint not in fingerprints:
raise SeedPassError("Selected profile does not exist.")
fingerprint_dir = self.fingerprint_manager.get_fingerprint_directory(
fingerprint
)
if not fingerprint_dir:
raise SeedPassError("Profile directory not found.")
if not self.validate_bip85_seed(parent_seed):
raise SeedPassError("Invalid BIP-85 seed phrase.")
computed_fp = generate_fingerprint(parent_seed)
if computed_fp != fingerprint:
raise SeedPassError("Seed does not match selected profile fingerprint.")
try:
self.current_fingerprint = fingerprint
self.fingerprint_manager.current_fingerprint = fingerprint
self.fingerprint_dir = fingerprint_dir
self.parent_seed = parent_seed
seed_bytes = Bip39SeedGenerator(parent_seed).Generate()
self.derive_key_hierarchy(seed_bytes)
index_key = base64.urlsafe_b64encode(self.KEY_STORAGE)
seed_key = self._derive_seed_key(password, fingerprint)
self.encryption_manager = EncryptionManager(index_key, fingerprint_dir)
seed_mgr = EncryptionManager(seed_key, fingerprint_dir)
self.vault = Vault(self.encryption_manager, fingerprint_dir)
self.config_manager = ConfigManager(
vault=self.vault,
fingerprint_dir=fingerprint_dir,
)
self._encrypt_parent_seed_compat(
seed_mgr,
parent_seed,
fingerprint,
mode=self._get_kdf_mode(),
)
self.store_hashed_password(password)
for stale_file in (
fingerprint_dir / "seedpass_entries_db.json.enc",
fingerprint_dir / "seedpass_passwords_db.json.enc",
fingerprint_dir / "seedpass_entries_db_checksum.txt",
fingerprint_dir / "seedpass_passwords_db_checksum.txt",
):
try:
stale_file.unlink()
except FileNotFoundError:
pass
self.initialize_bip85()
self.initialize_managers()
self.start_background_sync()
return True
except Exception as exc:
logger.error("Profile recovery failed: %s", exc, exc_info=True)
raise SeedPassError(f"Profile recovery failed: {exc}") from exc
def setup_encryption_manager(
self,
fingerprint_dir: Path,
@@ -1605,6 +1683,19 @@ class PasswordManager:
print(colored(f"Error: Failed to generate BIP-85 seed: {e}", "red"))
raise SeedPassError(f"Failed to generate BIP-85 seed: {e}") from e
def create_profile_from_generated_seed(
self,
*,
password: str,
seed: Optional[str] = None,
) -> tuple[str, str]:
"""Create a new profile from a generated seed and return ``(fingerprint, seed)``."""
new_seed = seed or self.generate_bip85_seed()
fingerprint = self._finalize_existing_seed(new_seed, password=password)
if not fingerprint:
raise SeedPassError("Failed to create profile from generated seed.")
return fingerprint, new_seed
@requires_unlocked
def save_and_encrypt_seed(
self, seed: str, fingerprint_dir: Path, *, password: Optional[str] = None
@@ -2234,6 +2325,47 @@ class PasswordManager:
self.notify("Starting with a new, empty vault.", level="INFO")
return
def restore_from_nostr_with_guidance_data(
self,
*,
seed_phrase: str,
password: str,
continue_without_backup: bool = False,
) -> tuple[str, bool]:
"""Restore a profile from Nostr without interactive prompts.
Returns ``(fingerprint, have_backup)``.
"""
have_backup = self.check_nostr_backup_exists(seed_phrase)
if not have_backup and not continue_without_backup:
raise SeedPassError("No Nostr backup found for this seed profile.")
fingerprint = self._finalize_existing_seed(seed_phrase, password=password)
if not fingerprint:
raise SeedPassError("Failed to initialize profile from provided seed.")
if have_backup:
success = self.attempt_initial_sync()
if not success:
raise SeedPassError("Failed to download or decrypt vault from Nostr.")
return fingerprint, have_backup
def restore_from_local_backup_data(
self,
*,
seed_phrase: str,
password: str,
backup_path: str,
) -> str:
"""Restore a profile from a local encrypted backup without prompts."""
fingerprint = self._finalize_existing_seed(seed_phrase, password=password)
if not fingerprint:
raise SeedPassError("Failed to initialize profile from provided seed.")
if not self.backup_manager:
raise SeedPassError("Backup manager is not initialized.")
self.backup_manager.restore_from_backup(backup_path)
return fingerprint
def _clear_header_add_entry(self, subtitle: str) -> None:
fp, parent_fp, child_fp = self.header_fingerprint_args
clear_header_with_notification(
+14 -1
View File
@@ -4,6 +4,8 @@ import hashlib
import json
from typing import Any
from .index0 import ensure_index0_payload, merge_system_index0
TOMBSTONE_RETENTION_CAP = 2048
@@ -190,7 +192,8 @@ def merge_index_payloads(
current: dict[str, Any], incoming: dict[str, Any], *, source_tag: str = ""
) -> dict[str, Any]:
"""Merge two index payloads using deterministic conflict resolution."""
out = dict(current) if isinstance(current, dict) else {}
out = ensure_index0_payload(dict(current) if isinstance(current, dict) else {})
incoming = ensure_index0_payload(incoming if isinstance(incoming, dict) else {})
cur_entries = out.get("entries", {})
if not isinstance(cur_entries, dict):
cur_entries = {}
@@ -318,4 +321,14 @@ def merge_index_payloads(
}
)
out["_sync_meta"] = meta
out_system = out.get("_system", {})
inc_system = incoming.get("_system", {})
if not isinstance(out_system, dict):
out_system = {}
if not isinstance(inc_system, dict):
inc_system = {}
out_system["index0"] = merge_system_index0(
out_system.get("index0"), inc_system.get("index0")
)
out["_system"] = out_system
return out
+3
View File
@@ -14,6 +14,7 @@ from .encryption import (
USE_ORJSON,
json_lib,
)
from .index0 import ensure_index0_payload
from utils.key_derivation import KdfConfig, CURRENT_KDF_VERSION
from utils.password_prompt import prompt_existing_password
@@ -232,12 +233,14 @@ class Vault:
self.migrated_from_legacy = (
legacy_detected or migration_performed or schema_migrated
)
data = ensure_index0_payload(data)
if return_migration_flags:
return data, self.migrated_from_legacy, migration_performed
return data
def save_index(self, data: dict) -> None:
"""Encrypt and write password index."""
data = ensure_index0_payload(data)
self.encryption_manager.save_json_data(
data, self.index_file, kdf=self._hkdf_kdf()
)
+674 -7
View File
@@ -2,12 +2,14 @@ from __future__ import annotations
import time
from typing import Any, Callable
from constants import APP_DIR
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.reactive import reactive
from textual.screen import Screen
from textual.widgets import Header, Footer, Static
from textual.widgets import Button, Input, Label, Static
from textual.containers import Horizontal, Vertical
from utils.fingerprint_manager import FingerprintManager
from .widgets.header import RibbonHeader
from .widgets.sidebar import SidebarContainer
@@ -171,6 +173,507 @@ class MainScreen(Screen):
yield ActionBar(id="action-bar")
class StartupScreen(Screen):
"""Profile-selection and unlock screen for TUI v3 startup."""
BINDINGS = [
Binding("enter", "submit", "Unlock", show=False),
Binding("escape", "app.quit", "Quit", show=False),
]
def __init__(
self,
*,
selected_fingerprint: str | None = None,
prompt: str | None = None,
) -> None:
super().__init__()
self.selected_fingerprint = selected_fingerprint
self.prompt = prompt or "Select a seed profile and unlock to continue."
def compose(self) -> ComposeResult:
with Vertical(id="startup-shell"):
yield Label("SeedPass TUI v3", id="startup-title")
yield Static(self.prompt, id="startup-prompt")
yield Static("", id="startup-profiles")
yield Input(placeholder="Profile number", id="startup-profile-choice")
yield Input(
placeholder="Master password",
password=True,
id="startup-password",
)
with Horizontal(id="startup-actions"):
yield Button("Unlock", id="startup-unlock", variant="primary")
yield Button("Refresh", id="startup-refresh")
yield Button("Add New", id="startup-add")
yield Button("Recover", id="startup-recover")
yield Static("", id="startup-status")
def on_mount(self) -> None:
self.refresh_profiles()
def refresh_profiles(self) -> None:
profiles = self.app._list_boot_profiles()
lines = ["Available Seed Profiles:"]
if profiles:
for idx, profile in enumerate(profiles, start=1):
marker = (
" *" if profile["fingerprint"] == self.selected_fingerprint else ""
)
lines.append(f"{idx}. {profile['label']}{marker}")
else:
lines.append("No existing seed profiles found.")
lines.append(f"{len(profiles) + 1}. Add a new seed profile")
lines.append(
f"{len(profiles) + 2}. Recover existing profile with blank local index"
)
lines.append("Q. Exit")
self.query_one("#startup-profiles", Static).update("\n".join(lines))
profile_input = self.query_one("#startup-profile-choice", Input)
if self.selected_fingerprint:
for idx, profile in enumerate(profiles, start=1):
if profile["fingerprint"] == self.selected_fingerprint:
profile_input.value = str(idx)
break
elif len(profiles) == 1 and not profile_input.value:
profile_input.value = "1"
def action_submit(self) -> None:
self._submit_unlock()
def on_input_submitted(self, _event: Input.Submitted) -> None:
self._submit_unlock()
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
if button_id == "startup-unlock":
self._submit_unlock()
return
if button_id == "startup-refresh":
self.refresh_profiles()
self._set_status("Profile list refreshed.")
return
if button_id == "startup-add":
self.app.push_screen(CreateProfileScreen())
return
if button_id == "startup-recover":
self.app.push_screen(RecoverProfileScreen())
return
def _set_status(self, message: str) -> None:
self.query_one("#startup-status", Static).update(message)
def _submit_unlock(self) -> None:
profiles = self.app._list_boot_profiles()
choice = self.query_one("#startup-profile-choice", Input).value.strip()
password = self.query_one("#startup-password", Input).value
if choice.lower() in {"q", "quit", "exit"}:
self.app.exit()
return
if not choice.isdigit():
self._set_status("Enter the profile number from the list above.")
return
selected = int(choice)
add_idx = len(profiles) + 1
recover_idx = len(profiles) + 2
if selected == add_idx:
self.app.push_screen(CreateProfileScreen())
return
if selected == recover_idx:
self.app.push_screen(RecoverProfileScreen())
return
if not (1 <= selected <= len(profiles)):
self._set_status("Invalid selection.")
return
if not password:
self._set_status("Enter the master password for the selected profile.")
return
fingerprint = profiles[selected - 1]["fingerprint"]
self.app._bootstrap_profile_session(fingerprint, password)
DEFAULT_CSS = """
StartupScreen {
align: center middle;
background: #999999;
}
#startup-shell {
width: 90;
max-width: 90;
border: heavy black;
background: #000000;
color: #ffffff;
padding: 1 2;
}
#startup-title {
text-style: bold;
margin-bottom: 1;
}
#startup-prompt {
color: #cccccc;
margin-bottom: 1;
}
#startup-profiles {
border: solid #ffffff;
padding: 1;
margin-bottom: 1;
min-height: 10;
}
#startup-profile-choice, #startup-password {
margin-bottom: 1;
}
#startup-actions {
height: auto;
margin-bottom: 1;
}
#startup-actions Button {
margin-right: 1;
}
#startup-status {
min-height: 2;
color: #daf2e5;
}
"""
class CreateProfileScreen(Screen):
"""Create a new profile or import an existing seed inside TUI v3."""
def compose(self) -> ComposeResult:
with Vertical(id="startup-shell"):
yield Label("Create Seed Profile", id="startup-title")
yield Static(
"Modes: existing, words, generate, nostr, backup. This screen now covers the main legacy onboarding branches inside v3.",
id="startup-prompt",
)
yield Input(
placeholder="Mode: existing | words | generate | nostr | backup",
id="create-mode",
)
yield Input(
placeholder="Seed phrase (paste full phrase; for 'words' you can type the 12 words space-separated)",
id="create-seed",
)
yield Input(
placeholder="Backup path (required for backup mode)",
id="create-backup-path",
)
yield Input(
placeholder="Continue without remote backup? yes | no (used for nostr mode)",
id="create-nostr-empty-ok",
)
yield Input(
placeholder="Master password",
password=True,
id="create-password",
)
with Horizontal(id="startup-actions"):
yield Button("Word Entry", id="create-words")
yield Button("Generate Seed", id="create-generate")
yield Button("Create Profile", id="create-submit", variant="primary")
yield Button("Back", id="create-back")
yield Static("", id="create-status")
def on_mount(self) -> None:
self.query_one("#create-mode", Input).value = "existing"
self.query_one("#create-nostr-empty-ok", Input).value = "no"
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
if button_id == "create-back":
self.app.pop_screen()
return
if button_id == "create-words":
self.app.push_screen(SeedWordsScreen(on_done=self._apply_seed_words))
return
if button_id == "create-generate":
self._generate_seed()
return
if button_id == "create-submit":
self._submit()
def on_input_submitted(self, _event: Input.Submitted) -> None:
self._submit()
def _set_status(self, message: str) -> None:
self.query_one("#create-status", Static).update(message)
def _apply_seed_words(self, seed_phrase: str) -> None:
self.query_one("#create-mode", Input).value = "words"
self.query_one("#create-seed", Input).value = seed_phrase
self._set_status("Seed phrase captured from word-by-word entry.")
def _generate_seed(self) -> None:
try:
seed = self.app._generate_bootstrap_seed()
self.query_one("#create-mode", Input).value = "generate"
self.query_one("#create-seed", Input).value = seed
self._set_status(
"Generated new seed. Review it carefully and store it before creating the profile."
)
except Exception as e:
self._set_status(f"Seed generation failed: {e}")
def _submit(self) -> None:
mode = self.query_one("#create-mode", Input).value.strip().lower() or "existing"
seed = self.query_one("#create-seed", Input).value.strip()
backup_path = self.query_one("#create-backup-path", Input).value.strip()
continue_without_backup = self.query_one(
"#create-nostr-empty-ok", Input
).value.strip().lower() in {"y", "yes", "true", "1"}
password = self.query_one("#create-password", Input).value
if not password:
self._set_status("Enter a master password.")
return
try:
if mode == "generate":
if not seed:
seed = self.app._generate_bootstrap_seed()
fingerprint = self.app._create_generated_profile(
password=password, seed=seed
)
elif mode == "nostr":
if not seed:
self._set_status("Enter the seed phrase for Nostr restore.")
return
fingerprint = self.app._restore_from_nostr_profile(
seed=seed,
password=password,
continue_without_backup=continue_without_backup,
)
elif mode == "backup":
if not seed:
self._set_status("Enter the seed phrase for backup restore.")
return
if not backup_path:
self._set_status("Enter the backup path for backup mode.")
return
fingerprint = self.app._restore_from_backup_profile(
seed=seed,
password=password,
backup_path=backup_path,
)
else:
if not seed:
self._set_status("Enter the seed phrase to import.")
return
fingerprint = self.app._create_existing_profile(
seed=seed,
password=password,
)
self.app._bootstrap_profile_session(fingerprint, password)
except Exception as e:
self._set_status(f"Profile creation failed: {e}")
class SeedWordsScreen(Screen):
"""Guided word-by-word seed entry for onboarding parity."""
def __init__(self, *, on_done: Callable[[str], None]) -> None:
super().__init__()
self._on_done = on_done
from mnemonic import Mnemonic
self._mnemonic = Mnemonic("english")
def compose(self) -> ComposeResult:
with Vertical(id="startup-shell"):
yield Label("Enter Seed Words", id="startup-title")
yield Static(
"Enter the 12-word seed phrase one word per field, following the legacy flow.",
id="startup-prompt",
)
yield Static("Progress: 0/12 words entered", id="seed-words-progress")
for idx in range(12):
yield Input(placeholder=f"Word {idx + 1}", id=f"seed-word-{idx + 1}")
with Horizontal(id="startup-actions"):
yield Button("Use Phrase", id="seed-words-submit", variant="primary")
yield Button("Back", id="seed-words-back")
yield Static("", id="seed-words-status")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "seed-words-back":
self.app.pop_screen()
return
if event.button.id == "seed-words-submit":
self._submit()
def on_input_submitted(self, _event: Input.Submitted) -> None:
self._submit()
def on_input_changed(self, event: Input.Changed) -> None:
input_id = event.input.id or ""
if not input_id.startswith("seed-word-"):
return
self._refresh_progress()
value = event.value.strip().lower()
if not value:
self.query_one("#seed-words-status", Static).update(
f"{input_id.replace('seed-word-', 'Word ')} is empty."
)
return
if value not in self._mnemonic.wordlist:
self.query_one("#seed-words-status", Static).update(
f"{input_id.replace('seed-word-', 'Word ')} is not in the BIP-39 wordlist."
)
return
self.query_one("#seed-words-status", Static).update(
f"{input_id.replace('seed-word-', 'Word ')} accepted."
)
def _refresh_progress(self) -> None:
count = 0
invalid = 0
for idx in range(12):
value = self.query_one(f"#seed-word-{idx + 1}", Input).value.strip().lower()
if not value:
continue
count += 1
if value not in self._mnemonic.wordlist:
invalid += 1
suffix = f" | invalid: {invalid}" if invalid else ""
self.query_one("#seed-words-progress", Static).update(
f"Progress: {count}/12 words entered{suffix}"
)
def _submit(self) -> None:
words: list[str] = []
for idx in range(12):
value = self.query_one(f"#seed-word-{idx + 1}", Input).value.strip().lower()
if not value:
self.query_one("#seed-words-status", Static).update(
f"Word {idx + 1} is required."
)
return
if value not in self._mnemonic.wordlist:
self.query_one("#seed-words-status", Static).update(
f"Word {idx + 1} is not in the BIP-39 wordlist."
)
return
words.append(value)
phrase = " ".join(words)
if not self._mnemonic.check(phrase):
self.query_one("#seed-words-status", Static).update(
"The full 12-word phrase failed BIP-39 validation."
)
return
self.app.push_screen(
SeedWordsReviewScreen(
phrase=phrase,
on_confirm=self._confirm_phrase,
)
)
def _confirm_phrase(self, phrase: str) -> None:
self._on_done(phrase)
self.app.pop_screen()
self.app.pop_screen()
class SeedWordsReviewScreen(Screen):
"""Confirmation screen for the assembled seed phrase."""
def __init__(self, *, phrase: str, on_confirm: Callable[[str], None]) -> None:
super().__init__()
self._phrase = phrase
self._on_confirm = on_confirm
def compose(self) -> ComposeResult:
with Vertical(id="startup-shell"):
yield Label("Review Seed Phrase", id="startup-title")
yield Static(
"Confirm this phrase before it is applied to the create flow.",
id="startup-prompt",
)
yield Static(self._phrase, id="seed-review-phrase")
with Horizontal(id="startup-actions"):
yield Button(
"Confirm Phrase", id="seed-review-confirm", variant="primary"
)
yield Button("Edit Words", id="seed-review-back")
yield Static("", id="seed-review-status")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "seed-review-back":
self.app.pop_screen()
return
if event.button.id == "seed-review-confirm":
self._on_confirm(self._phrase)
class RecoverProfileScreen(Screen):
"""Recover an existing profile with a blank local index inside TUI v3."""
def compose(self) -> ComposeResult:
with Vertical(id="startup-shell"):
yield Label("Recover Existing Profile", id="startup-title")
yield Static(
"This matches the legacy recovery path: select an existing profile, provide the matching seed, and reset the local index.",
id="startup-prompt",
)
yield Static("", id="recover-profiles")
yield Input(placeholder="Profile number", id="recover-choice")
yield Input(placeholder="Seed phrase", id="recover-seed")
yield Input(
placeholder="Master password",
password=True,
id="recover-password",
)
with Horizontal(id="startup-actions"):
yield Button("Recover Profile", id="recover-submit", variant="primary")
yield Button("Back", id="recover-back")
yield Static("", id="recover-status")
def on_mount(self) -> None:
self.refresh_profiles()
def refresh_profiles(self) -> None:
profiles = self.app._list_boot_profiles()
lines = ["Available Seed Profiles:"]
for idx, profile in enumerate(profiles, start=1):
lines.append(f"{idx}. {profile['label']}")
self.query_one("#recover-profiles", Static).update("\n".join(lines))
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
if button_id == "recover-back":
self.app.pop_screen()
return
if button_id == "recover-submit":
self._submit()
def on_input_submitted(self, _event: Input.Submitted) -> None:
self._submit()
def _set_status(self, message: str) -> None:
self.query_one("#recover-status", Static).update(message)
def _submit(self) -> None:
profiles = self.app._list_boot_profiles()
choice = self.query_one("#recover-choice", Input).value.strip()
seed = self.query_one("#recover-seed", Input).value.strip()
password = self.query_one("#recover-password", Input).value
if not choice.isdigit():
self._set_status("Enter the profile number to recover.")
return
idx = int(choice)
if not (1 <= idx <= len(profiles)):
self._set_status("Invalid profile selection.")
return
if not seed or not password:
self._set_status("Seed phrase and master password are required.")
return
fingerprint = profiles[idx - 1]["fingerprint"]
try:
self.app._recover_profile(
fingerprint=fingerprint,
seed=seed,
password=password,
)
self.app._bootstrap_profile_session(fingerprint, password)
except Exception as e:
self._set_status(f"Recovery failed: {e}")
class SeedPassTuiV3(App[None]):
"""
SeedPass TUI v3 - Rebuilt from scratch for modularity and mockup fidelity.
@@ -256,21 +759,176 @@ class SeedPassTuiV3(App[None]):
# Initialized services
self.services: dict[str, Any] = {}
self._initial_fingerprint = fingerprint
self._main_screen_initialized = False
def on_mount(self) -> None:
"""Initialize services and push the main screen."""
"""Initialize provided services or start in profile-unlock mode."""
self.processor = CommandProcessor(self)
self._initialize_factory_services()
if self.services:
self._enter_main_workspace(initial=True)
else:
self.present_startup_screen(
selected_fingerprint=self._initial_fingerprint,
prompt="Select a seed profile and unlock to enter SeedPass.",
)
# Global UI Heartbeat (for 2FA ticking etc)
self.set_interval(1.0, self.action_refresh_ui_quiet)
def _initialize_factory_services(self) -> None:
for name, factory in self.factories.items():
if factory:
try:
self.services[name] = factory()
service = factory()
except Exception as e:
self.log(f"Failed to init service {name}: {e}")
continue
if service is not None:
self.services[name] = service
self.active_fingerprint = self._initial_fingerprint or ""
self.push_screen(MainScreen())
# Global UI Heartbeat (for 2FA ticking etc)
self.set_interval(1.0, self.action_refresh_ui_quiet)
def _list_boot_profiles(self) -> list[dict[str, str]]:
try:
manager = FingerprintManager(APP_DIR)
return [
{"fingerprint": fp, "label": manager.display_name(fp)}
for fp in manager.list_fingerprints()
]
except Exception as e:
self.log(f"Failed to list startup profiles: {e}")
return []
def present_startup_screen(
self,
*,
selected_fingerprint: str | None = None,
prompt: str | None = None,
) -> None:
self.push_screen(
StartupScreen(
selected_fingerprint=selected_fingerprint,
prompt=prompt,
)
)
def _bootstrap_profile_session(self, fingerprint: str, password: str) -> None:
try:
from seedpass.core.api import (
ConfigService,
EntryService,
NostrService,
ProfileService,
SemanticIndexService,
SyncService,
UtilityService,
VaultService,
)
from seedpass.core.manager import PasswordManager
pm = PasswordManager(fingerprint=fingerprint, password=password)
self.services = {
"entry": EntryService(pm),
"profile": ProfileService(pm),
"config": ConfigService(pm),
"nostr": NostrService(pm),
"sync": SyncService(pm),
"utility": UtilityService(pm),
"vault": VaultService(pm),
"semantic": SemanticIndexService(pm),
}
self.active_fingerprint = pm.current_fingerprint or fingerprint
self.session_locked = False
self._enter_main_workspace(initial=not self._main_screen_initialized)
self.notify(f"Unlocked profile {self.active_fingerprint[:12]}")
except Exception as e:
try:
if isinstance(self.screen, StartupScreen):
self.screen._set_status(f"Unlock failed: {e}")
except Exception:
pass
self.notify(f"Unlock failed: {e}", severity="error")
def _make_bootstrap_manager(self):
from seedpass.core.manager import PasswordManager
return PasswordManager(bootstrap_only=True)
def _generate_bootstrap_seed(self) -> str:
manager = self._make_bootstrap_manager()
return manager.generate_bip85_seed()
def _create_existing_profile(self, *, seed: str, password: str) -> str:
manager = self._make_bootstrap_manager()
fingerprint = manager.setup_existing_seed(seed=seed, password=password)
if not fingerprint:
raise RuntimeError("Profile creation did not return a fingerprint.")
return fingerprint
def _create_generated_profile(
self, *, password: str, seed: str | None = None
) -> str:
manager = self._make_bootstrap_manager()
fingerprint, _generated_seed = manager.create_profile_from_generated_seed(
password=password,
seed=seed,
)
return fingerprint
def _recover_profile(self, *, fingerprint: str, seed: str, password: str) -> None:
manager = self._make_bootstrap_manager()
manager.recover_profile_with_blank_index_data(
fingerprint=fingerprint,
parent_seed=seed,
password=password,
)
def _restore_from_nostr_profile(
self,
*,
seed: str,
password: str,
continue_without_backup: bool,
) -> str:
manager = self._make_bootstrap_manager()
fingerprint, _have_backup = manager.restore_from_nostr_with_guidance_data(
seed_phrase=seed,
password=password,
continue_without_backup=continue_without_backup,
)
return fingerprint
def _restore_from_backup_profile(
self,
*,
seed: str,
password: str,
backup_path: str,
) -> str:
manager = self._make_bootstrap_manager()
return manager.restore_from_local_backup_data(
seed_phrase=seed,
password=password,
backup_path=backup_path,
)
def _resolve_active_fingerprint(self) -> str:
try:
manager = self.services["vault"]._manager
current = getattr(manager, "current_fingerprint", None)
if current:
return str(current)
except Exception:
pass
return self._initial_fingerprint or self.active_fingerprint or ""
def _enter_main_workspace(self, *, initial: bool) -> None:
self.active_fingerprint = self._resolve_active_fingerprint()
if initial or not self._main_screen_initialized:
self.push_screen(MainScreen())
self._main_screen_initialized = True
return
if isinstance(self.screen, StartupScreen):
self.pop_screen()
self.action_refresh()
def on_command_palette_command_executed(
self, message: CommandPalette.CommandExecuted
@@ -338,6 +996,8 @@ class SeedPassTuiV3(App[None]):
def action_refresh(self) -> None:
"""Force a global UI refresh."""
if not self._main_screen_initialized:
return
self.screen.query_one("#profile-tree")._refresh_tree()
self.screen.query_one("#entry-data-table")._refresh_data()
self.notify("UI Refreshed")
@@ -421,6 +1081,10 @@ class SeedPassTuiV3(App[None]):
locker()
self.session_locked = True
self.notify("Vault locked")
self.present_startup_screen(
selected_fingerprint=self.active_fingerprint or None,
prompt="Vault locked. Re-enter your master password to continue.",
)
except Exception as e:
self.notify(f"Lock failed: {e}", severity="error")
@@ -442,6 +1106,9 @@ class SeedPassTuiV3(App[None]):
except Exception:
unlocker(password)
self.session_locked = False
if isinstance(self.screen, StartupScreen) and self._main_screen_initialized:
self.pop_screen()
self.action_refresh()
self.notify("Vault unlocked")
except Exception as e:
self.notify(f"Unlock failed: {e}", severity="error")
+7 -2
View File
@@ -70,8 +70,13 @@ class ProfileTree(Tree):
self.app.notify(f"Opened Entry #{entry_id} from profile tree")
return
self.app.active_fingerprint = value
self.app.notify(f"Switched to profile: {value[:8]}")
if value == self.app.active_fingerprint:
self.app.notify(f"Profile already active: {value[:8]}")
return
self.app.present_startup_screen(
selected_fingerprint=value,
prompt=f"Unlock {value[:12]} to switch profiles.",
)
class SidebarContainer(Static):
+130
View File
@@ -0,0 +1,130 @@
from pathlib import Path
from tempfile import TemporaryDirectory
from helpers import create_vault
from nostr.backup_models import ChunkMeta, Manifest
from seedpass.core.index0 import (
compute_event_hash,
compute_event_id,
ensure_index0_payload,
merge_system_index0,
normalize_index0,
)
def test_ensure_index0_payload_initializes_reserved_namespace():
payload = ensure_index0_payload({"schema_version": 4, "entries": {}})
assert payload["_system"]["index0"]["schema_version"] == 1
assert payload["_system"]["index0"]["events"] == {}
assert payload["_system"]["index0"]["canonical_views"] == {}
assert payload["_system"]["index0"]["stats"]["event_count"] == 0
def test_merge_system_index0_is_order_independent_for_events_and_heads():
base_event = {
"event_type": "entry_modified",
"subject_type": "entry",
"subject_id": "1",
"subject_kind": "password",
"scope_path": "seed/root",
"actor_type": "user",
"actor_id": "fp1",
"writer_id": "writer:profile:fp1",
"modified_ts": 100,
"payload_ref": {"entry_id": "1"},
"links": [],
"tags": ["b", "a"],
"visibility": "private",
"classification": "internal",
"partition": "standard",
}
event_a = dict(base_event)
event_a["integrity_hash"] = compute_event_hash(event_a)
event_a["event_id"] = compute_event_id(event_a)
event_b = dict(base_event)
event_b["modified_ts"] = 200
event_b["prev_hash"] = "prev"
event_b["integrity_hash"] = compute_event_hash(event_b)
event_b["event_id"] = compute_event_id(event_b)
current = normalize_index0(
{
"events": {event_a["event_id"]: event_a},
"heads": {
"writer:profile:fp1": {
"event_id": event_a["event_id"],
"head_hash": "aaa",
"modified_ts": 100,
}
},
}
)
incoming = normalize_index0(
{
"events": {event_b["event_id"]: event_b},
"heads": {
"writer:profile:fp1": {
"event_id": event_b["event_id"],
"head_hash": "bbb",
"modified_ts": 200,
}
},
}
)
ab = merge_system_index0(current, incoming)
ba = merge_system_index0(incoming, current)
assert ab == ba
assert ab["stats"]["event_count"] == 2
assert ab["heads"]["writer:profile:fp1"]["head_hash"] == "bbb"
def test_vault_save_and_load_index_normalizes_index0_namespace():
with TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
vault, _ = create_vault(base)
vault.save_index({"schema_version": 4, "entries": {}})
loaded = vault.load_index()
assert "_system" in loaded
assert "index0" in loaded["_system"]
assert loaded["_system"]["index0"]["stats"]["writer_count"] == 0
def test_manifest_supports_optional_index0_metadata_round_trip():
manifest = Manifest(
ver=1,
algo="gzip",
chunks=[ChunkMeta(id="c1", size=10, hash="abc", event_id="evt1")],
delta_since=123,
nonce="nonce",
index0={
"schema_version": 1,
"checkpoint_ids": ["cp:day:2026-03-05:writer:profile:fp1"],
"checkpoint_hashes": {"cp:day:2026-03-05:writer:profile:fp1": "abc123"},
"stream_heads": {"writer:profile:fp1": "head123"},
},
)
payload = {
"ver": manifest.ver,
"algo": manifest.algo,
"chunks": [meta.__dict__ for meta in manifest.chunks],
"delta_since": manifest.delta_since,
"nonce": manifest.nonce,
"index0": manifest.index0,
}
parsed = Manifest(
ver=payload["ver"],
algo=payload["algo"],
chunks=[ChunkMeta(**chunk) for chunk in payload["chunks"]],
delta_since=payload["delta_since"],
nonce=payload["nonce"],
index0=payload["index0"],
)
assert parsed.index0 is not None
assert parsed.index0["stream_heads"]["writer:profile:fp1"] == "head123"
+252 -1
View File
@@ -5,9 +5,17 @@ from types import SimpleNamespace
import pytest
pytest.importorskip("textual")
from textual.widgets import DataTable
from textual.widgets import DataTable, Input
from seedpass.tui_v3 import launch_tui3
from seedpass.tui_v3.app import (
CreateProfileScreen,
RecoverProfileScreen,
SeedPassTuiV3,
SeedWordsScreen,
SeedWordsReviewScreen,
StartupScreen,
)
class V3EntryService:
@@ -126,6 +134,7 @@ async def test_tui3_palette_lock_unlock_and_session_status() -> None:
await pilot.pause()
assert app.session_locked is True
assert vault.locked is True
assert isinstance(app.screen, StartupScreen)
app.processor.execute("unlock hunter2")
await pilot.pause()
assert app.session_locked is False
@@ -169,3 +178,245 @@ async def test_tui3_grid_focus_refresh_enables_selection_after_runtime_add() ->
table._refresh_data()
await pilot.pause()
assert table.row_count >= 4
@pytest.mark.anyio
async def test_tui3_starts_with_unlock_screen_when_services_are_not_preloaded() -> None:
app = SeedPassTuiV3()
app._list_boot_profiles = lambda: [
{"fingerprint": "EFBE51E70ED1B53A", "label": "Primary (EFBE51E70ED1B53A)"}
]
captured: dict[str, str] = {}
def _bootstrap(fingerprint: str, password: str) -> None:
captured["fingerprint"] = fingerprint
captured["password"] = password
app._bootstrap_profile_session = _bootstrap
async with app.run_test() as pilot:
await pilot.pause()
assert isinstance(app.screen, StartupScreen)
assert app.screen.query_one("#startup-profile-choice", Input).value == "1"
app.screen.query_one("#startup-password", Input).value = "hunter2"
await pilot.press("enter")
await pilot.pause()
assert captured == {
"fingerprint": "EFBE51E70ED1B53A",
"password": "hunter2",
}
@pytest.mark.anyio
async def test_tui3_create_profile_screen_uses_in_app_bootstrap_helpers() -> None:
app = SeedPassTuiV3()
app._list_boot_profiles = lambda: []
created: dict[str, str] = {}
booted: dict[str, str] = {}
def _create_existing_profile(*, seed: str, password: str) -> str:
created["seed"] = seed
created["password"] = password
return "NEWFP123"
def _bootstrap(fingerprint: str, password: str) -> None:
booted["fingerprint"] = fingerprint
booted["password"] = password
app._create_existing_profile = _create_existing_profile
app._bootstrap_profile_session = _bootstrap
async with app.run_test() as pilot:
await pilot.pause()
app.push_screen(CreateProfileScreen())
await pilot.pause()
screen = app.screen
screen.query_one("#create-mode", Input).value = "existing"
screen.query_one("#create-seed", Input).value = (
"abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"
)
screen.query_one("#create-password", Input).value = "hunter2"
await pilot.press("enter")
await pilot.pause()
assert created["password"] == "hunter2"
assert booted == {"fingerprint": "NEWFP123", "password": "hunter2"}
@pytest.mark.anyio
async def test_tui3_recover_profile_screen_uses_in_app_bootstrap_helpers() -> None:
app = SeedPassTuiV3()
app._list_boot_profiles = lambda: [
{"fingerprint": "EFBE51E70ED1B53A", "label": "Primary (EFBE51E70ED1B53A)"}
]
recovered: dict[str, str] = {}
booted: dict[str, str] = {}
def _recover_profile(*, fingerprint: str, seed: str, password: str) -> None:
recovered["fingerprint"] = fingerprint
recovered["seed"] = seed
recovered["password"] = password
def _bootstrap(fingerprint: str, password: str) -> None:
booted["fingerprint"] = fingerprint
booted["password"] = password
app._recover_profile = _recover_profile
app._bootstrap_profile_session = _bootstrap
async with app.run_test() as pilot:
await pilot.pause()
app.push_screen(RecoverProfileScreen())
await pilot.pause()
screen = app.screen
screen.query_one("#recover-choice", Input).value = "1"
screen.query_one("#recover-seed", Input).value = (
"abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"
)
screen.query_one("#recover-password", Input).value = "hunter2"
await pilot.press("enter")
await pilot.pause()
assert recovered["fingerprint"] == "EFBE51E70ED1B53A"
assert booted == {
"fingerprint": "EFBE51E70ED1B53A",
"password": "hunter2",
}
@pytest.mark.anyio
async def test_tui3_create_profile_screen_supports_nostr_restore_mode() -> None:
app = SeedPassTuiV3()
app._list_boot_profiles = lambda: []
restored: dict[str, object] = {}
booted: dict[str, str] = {}
def _restore_from_nostr_profile(
*, seed: str, password: str, continue_without_backup: bool
) -> str:
restored["seed"] = seed
restored["password"] = password
restored["continue_without_backup"] = continue_without_backup
return "NOSTRFP1"
def _bootstrap(fingerprint: str, password: str) -> None:
booted["fingerprint"] = fingerprint
booted["password"] = password
app._restore_from_nostr_profile = _restore_from_nostr_profile
app._bootstrap_profile_session = _bootstrap
async with app.run_test() as pilot:
await pilot.pause()
app.push_screen(CreateProfileScreen())
await pilot.pause()
screen = app.screen
screen.query_one("#create-mode", Input).value = "nostr"
screen.query_one("#create-seed", Input).value = (
"abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"
)
screen.query_one("#create-nostr-empty-ok", Input).value = "yes"
screen.query_one("#create-password", Input).value = "hunter2"
await pilot.press("enter")
await pilot.pause()
assert restored["continue_without_backup"] is True
assert booted == {"fingerprint": "NOSTRFP1", "password": "hunter2"}
@pytest.mark.anyio
async def test_tui3_create_profile_screen_supports_backup_restore_mode() -> None:
app = SeedPassTuiV3()
app._list_boot_profiles = lambda: []
restored: dict[str, str] = {}
booted: dict[str, str] = {}
def _restore_from_backup_profile(
*, seed: str, password: str, backup_path: str
) -> str:
restored["seed"] = seed
restored["password"] = password
restored["backup_path"] = backup_path
return "BACKUPFP1"
def _bootstrap(fingerprint: str, password: str) -> None:
booted["fingerprint"] = fingerprint
booted["password"] = password
app._restore_from_backup_profile = _restore_from_backup_profile
app._bootstrap_profile_session = _bootstrap
async with app.run_test() as pilot:
await pilot.pause()
app.push_screen(CreateProfileScreen())
await pilot.pause()
screen = app.screen
screen.query_one("#create-mode", Input).value = "backup"
screen.query_one("#create-seed", Input).value = (
"abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"
)
screen.query_one("#create-backup-path", Input).value = "/tmp/profile-backup.enc"
screen.query_one("#create-password", Input).value = "hunter2"
await pilot.press("enter")
await pilot.pause()
assert restored["backup_path"] == "/tmp/profile-backup.enc"
assert booted == {"fingerprint": "BACKUPFP1", "password": "hunter2"}
@pytest.mark.anyio
async def test_tui3_word_by_word_seed_entry_feeds_create_profile_screen() -> None:
app = SeedPassTuiV3()
app._list_boot_profiles = lambda: []
async with app.run_test() as pilot:
await pilot.pause()
app.push_screen(CreateProfileScreen())
await pilot.pause()
create_screen = app.screen
assert isinstance(create_screen, CreateProfileScreen)
create_screen.query_one("#create-words").press()
await pilot.pause()
assert isinstance(app.screen, SeedWordsScreen)
for idx, word in enumerate(
[
"abandon",
"abandon",
"abandon",
"abandon",
"abandon",
"abandon",
"abandon",
"abandon",
"abandon",
"abandon",
"abandon",
"about",
],
start=1,
):
app.screen.query_one(f"#seed-word-{idx}", Input).value = word
await pilot.press("enter")
await pilot.pause()
assert isinstance(app.screen, SeedWordsReviewScreen)
app.screen.query_one("#seed-review-confirm").press()
await pilot.pause()
assert isinstance(app.screen, CreateProfileScreen)
assert app.screen.query_one("#create-mode", Input).value == "words"
assert (
app.screen.query_one("#create-seed", Input).value
== "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"
)
@pytest.mark.anyio
async def test_tui3_word_entry_flags_invalid_bip39_word() -> None:
app = SeedPassTuiV3()
app._list_boot_profiles = lambda: []
async with app.run_test() as pilot:
await pilot.pause()
app.push_screen(SeedWordsScreen(on_done=lambda _phrase: None))
await pilot.pause()
app.screen.query_one("#seed-word-1", Input).value = "notaword"
await pilot.pause()
status = str(app.screen.query_one("#seed-words-status").render())
progress = str(app.screen.query_one("#seed-words-progress").render())
assert "not in the BIP-39 wordlist" in status
assert "invalid: 1" in progress
+33 -17
View File
@@ -684,6 +684,7 @@ def test_capabilities_json():
payload = json.loads(result.stdout)
assert payload["schema_version"] == 1
assert "cli" in payload["interfaces"]
assert payload["interfaces"]["cli"]["default_interface"] == "tui3"
assert "agent" in payload["interfaces"]["cli"]["root_commands"]
assert "tui3" in payload["interfaces"]["cli"]["root_commands"]
assert "legacy" in payload["interfaces"]["cli"]["root_commands"]
@@ -1011,8 +1012,23 @@ def test_update_checksum_command(monkeypatch):
assert called.get("called") is True
def test_root_tui2_forward_fingerprint(monkeypatch):
"""Ensure --fingerprint is forwarded when launching default TUI v2."""
def test_root_tui3_forward_fingerprint(monkeypatch):
"""Ensure --fingerprint is forwarded when launching default TUI v3."""
called = {}
def fake_launch(**kwargs):
called["fp"] = kwargs.get("fingerprint")
called["factory"] = kwargs.get("entry_service_factory")
return True
monkeypatch.setattr(cli, "launch_tui3", fake_launch)
result = runner.invoke(app, ["--fingerprint", "abc"])
assert result.exit_code == 0
assert called.get("fp") == "abc"
assert called.get("factory") is None
def test_root_explicit_tui2_forward_fingerprint(monkeypatch):
called = {}
def fake_launch(**kwargs):
@@ -1022,7 +1038,7 @@ def test_root_tui2_forward_fingerprint(monkeypatch):
monkeypatch.setattr(cli, "_get_entry_service", lambda _ctx: object())
monkeypatch.setattr(cli, "launch_tui2", fake_launch)
result = runner.invoke(app, ["--fingerprint", "abc"])
result = runner.invoke(app, ["--tui", "v2", "--fingerprint", "abc"])
assert result.exit_code == 0
assert called.get("fp") == "abc"
assert callable(called.get("factory"))
@@ -1095,8 +1111,8 @@ def test_tui2_check(monkeypatch):
def test_tui3_check(monkeypatch):
monkeypatch.setattr(
cli,
"check_tui2_runtime",
lambda: {"status": "ok", "backend": "textual", "textual_available": True},
"check_tui3_runtime",
lambda: {"status": "ok", "backend": "textual-v3", "textual_available": True},
)
result = runner.invoke(app, ["tui3", "--check"])
assert result.exit_code == 0
@@ -1153,24 +1169,24 @@ def test_tui2_fallback_legacy_strips_subcommand_argv(monkeypatch):
assert called.get("argv") == []
def test_root_fallback_legacy_when_tui2_unavailable(monkeypatch):
called = {}
monkeypatch.setattr(cli, "_get_entry_service", lambda _ctx: object())
monkeypatch.setattr(cli, "launch_tui2", lambda **_: False)
def test_root_tui3_unavailable_returns_error(monkeypatch):
monkeypatch.setattr(cli, "launch_tui3", lambda **_: False)
result = runner.invoke(app, ["--fingerprint", "abc"])
assert result.exit_code == 1
assert "TUI v3 runtime unavailable" in result.stderr
assert "seedpass tui2" in result.stderr
assert "seedpass legacy" in result.stderr
def fake_main(*, fingerprint=None):
called["fp"] = fingerprint
return 0
def test_root_tui3_does_not_preflight_entry_service(monkeypatch):
monkeypatch.setattr(
cli,
"importlib",
SimpleNamespace(import_module=lambda _: SimpleNamespace(main=fake_main)),
"_get_entry_service",
lambda _ctx: (_ for _ in ()).throw(RuntimeError("should not run")),
)
result = runner.invoke(app, ["--fingerprint", "abc"])
monkeypatch.setattr(cli, "launch_tui3", lambda **_: True)
result = runner.invoke(app, [])
assert result.exit_code == 0
assert called.get("fp") == "abc"
assert "falling back to legacy TUI" in result.stdout
def test_tui2_preflight_failure_without_fallback(monkeypatch):