thePR0M3TH3AN/seedPass
1
0
Fork 0
mirror of https://github.com/PR0M3TH3AN/SeedPass.git synced 2025-09-08 07:18:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: ensure manifest IDs omit fingerprint

Browse Source
This commit is contained in:
thePR0M3TH3AN
2025-08-20 19:11:24 -04:00
parent e75e197270
commit 0315562d80
9 changed files with 75 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/nostr/backup_models.py
View File

@@ -25,3 +25,4 @@ class Manifest:
algo: str
chunks: List[ChunkMeta]
delta_since: Optional[int] = None
nonce: Optional[str] = None

5
src/nostr/client.py
View File

@@ -33,7 +33,7 @@ from .backup_models import (
)
from .connection import ConnectionHandler, DEFAULT_RELAYS
from .key_manager import KeyManager as SeedPassKeyManager
from .snapshot import MANIFEST_ID_PREFIX, SnapshotHandler, prepare_snapshot
from .snapshot import SnapshotHandler, prepare_snapshot
if TYPE_CHECKING: # pragma: no cover - imported for type hints
from seedpass.core.config_manager import ConfigManager
@@ -57,6 +57,7 @@ class NostrClient(ConnectionHandler, SnapshotHandler):
parent_seed: Optional[str] = None,
offline_mode: bool = False,
config_manager: Optional["ConfigManager"] = None,
key_index: bytes | None = None,
) -> None:
self.encryption_manager = encryption_manager
self.fingerprint = fingerprint
@@ -99,6 +100,7 @@ class NostrClient(ConnectionHandler, SnapshotHandler):
self.current_manifest: Manifest | None = None
self.current_manifest_id: str | None = None
self._delta_events: list[str] = []
self.key_index = key_index or b""
# Configure and initialize the nostr-sdk Client
signer = NostrSigner.keys(self.keys)
@@ -111,5 +113,4 @@ __all__ = [
"NostrClient",
"prepare_snapshot",
"DEFAULT_RELAYS",
"MANIFEST_ID_PREFIX",
]

64
src/nostr/snapshot.py
View File

@@ -2,8 +2,10 @@ import asyncio
import base64
import gzip
import hashlib
import hmac
import json
import logging
import os
import time
from datetime import timedelta
from typing import Tuple
@@ -23,9 +25,6 @@ from .backup_models import (
logger = logging.getLogger("nostr.client")
logger.setLevel(logging.WARNING)
# Identifier prefix for replaceable manifest events
MANIFEST_ID_PREFIX = "seedpass-manifest-"
def prepare_snapshot(
encrypted_bytes: bytes, limit: int
@@ -47,6 +46,19 @@ def prepare_snapshot(
return manifest, chunks
def new_manifest_id(key_index: bytes) -> tuple[str, bytes]:
"""Return a new manifest identifier and nonce.
The identifier is computed as HMAC-SHA256 of ``b"manifest|" + nonce``
using ``key_index`` as the HMAC key. The nonce is returned so it can be
embedded inside the manifest itself.
"""
nonce = os.urandom(16)
digest = hmac.new(key_index, b"manifest|" + nonce, hashlib.sha256).hexdigest()
return digest, nonce
class SnapshotHandler:
"""Mixin providing chunk and manifest handling."""
@@ -84,34 +96,43 @@ class SnapshotHandler:
except Exception:
meta.event_id = None
if (
self.current_manifest_id
and self.current_manifest
and getattr(self.current_manifest, "nonce", None)
):
manifest_id = self.current_manifest_id
manifest.nonce = self.current_manifest.nonce
else:
manifest_id, nonce = new_manifest_id(self.key_index)
manifest.nonce = base64.b64encode(nonce).decode("utf-8")
manifest_json = json.dumps(
{
"ver": manifest.ver,
"algo": manifest.algo,
"chunks": [meta.__dict__ for meta in manifest.chunks],
"delta_since": manifest.delta_since,
"nonce": manifest.nonce,
}
)
manifest_identifier = (
self.current_manifest_id or f"{MANIFEST_ID_PREFIX}{self.fingerprint}"
)
manifest_event = (
nostr_client.EventBuilder(nostr_client.Kind(KIND_MANIFEST), manifest_json)
.tags([nostr_client.Tag.identifier(manifest_identifier)])
.tags([nostr_client.Tag.identifier(manifest_id)])
.build(self.keys.public_key())
.sign_with_keys(self.keys)
)
await self.client.send_event(manifest_event)
with self._state_lock:
self.current_manifest = manifest
self.current_manifest_id = manifest_identifier
self.current_manifest_id = manifest_id
self.current_manifest.delta_since = int(time.time())
self._delta_events = []
if getattr(self, "verbose_timing", False):
duration = time.perf_counter() - start
logger.info("publish_snapshot completed in %.2f seconds", duration)
return manifest, manifest_identifier
return manifest, manifest_id
async def _fetch_chunks_with_retry(
self, manifest_event
@@ -129,6 +150,7 @@ class SnapshotHandler:
if data.get("delta_since") is not None
else None
),
nonce=data.get("nonce"),
)
except Exception:
return None
@@ -204,14 +226,11 @@ class SnapshotHandler:
pubkey = self.keys.public_key()
timeout = timedelta(seconds=10)
ident = f"{MANIFEST_ID_PREFIX}{self.fingerprint}"
f = (
nostr_client.Filter()
.author(pubkey)
.kind(nostr_client.Kind(KIND_MANIFEST))
.identifier(ident)
.limit(1)
)
ident = self.current_manifest_id
f = nostr_client.Filter().author(pubkey).kind(nostr_client.Kind(KIND_MANIFEST))
if ident:
f = f.identifier(ident)
f = f.limit(1)
try:
events = (await self.client.fetch_events(f, timeout)).to_vec()
except Exception as e: # pragma: no cover - network errors
@@ -223,13 +242,11 @@ class SnapshotHandler:
)
return None
if not events:
ident = MANIFEST_ID_PREFIX.rstrip("-")
if not events and ident:
f = (
nostr_client.Filter()
.author(pubkey)
.kind(nostr_client.Kind(KIND_MANIFEST))
.identifier(ident)
.limit(1)
)
try:
@@ -245,8 +262,6 @@ class SnapshotHandler:
if not events:
return None
logger.info("Fetched manifest using identifier %s", ident)
for manifest_event in events:
try:
result = await self._fetch_chunks_with_retry(manifest_event)
@@ -300,7 +315,9 @@ class SnapshotHandler:
return
await self._connect_async()
pubkey = self.keys.public_key()
ident = self.current_manifest_id or f"{MANIFEST_ID_PREFIX}{self.fingerprint}"
ident = self.current_manifest_id
if ident is None:
return
f = (
nostr_client.Filter()
.author(pubkey)
@@ -358,6 +375,7 @@ class SnapshotHandler:
meta.__dict__ for meta in self.current_manifest.chunks
],
"delta_since": self.current_manifest.delta_since,
"nonce": self.current_manifest.nonce,
}
)
manifest_event = (

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

@@ -106,7 +106,6 @@ from utils.fingerprint_manager import FingerprintManager
# Import NostrClient
from nostr.client import NostrClient
from nostr.connection import DEFAULT_RELAYS
from nostr.snapshot import MANIFEST_ID_PREFIX
from .config_manager import ConfigManager
from .state_manager import StateManager
from .stats_manager import StatsManager
@@ -1089,7 +1088,7 @@ class PasswordManager:
self.fingerprint_manager.current_fingerprint = fingerprint
self.fingerprint_dir = fingerprint_dir
if not getattr(self, "manifest_id", None):
self.manifest_id = f"{MANIFEST_ID_PREFIX}{fingerprint}"
self.manifest_id = None
logging.info(f"Current seed profile set to {fingerprint}")
try:
@@ -1431,6 +1430,7 @@ class PasswordManager:
offline_mode=self.offline_mode,
config_manager=self.config_manager,
parent_seed=getattr(self, "parent_seed", None),
key_index=self.KEY_INDEX,
)
if getattr(self, "manifest_id", None) and hasattr(
@@ -4487,6 +4487,7 @@ class PasswordManager:
relays=relay_list,
config_manager=self.config_manager,
parent_seed=getattr(self, "parent_seed", None),
key_index=self.KEY_INDEX,
)
if getattr(self, "manifest_id", None) and hasattr(

4
src/seedpass/core/profile_service.py
View File

@@ -6,7 +6,6 @@ from typing import Optional, TYPE_CHECKING
from termcolor import colored
import seedpass.core.manager as manager_module
from nostr.snapshot import MANIFEST_ID_PREFIX
from utils.password_prompt import prompt_existing_password
@@ -44,7 +43,7 @@ class ProfileService:
pm.fingerprint_manager.current_fingerprint = selected_fingerprint
pm.current_fingerprint = selected_fingerprint
if not getattr(pm, "manifest_id", None):
pm.manifest_id = f"{MANIFEST_ID_PREFIX}{selected_fingerprint}"
pm.manifest_id = None
pm.fingerprint_dir = pm.fingerprint_manager.get_current_fingerprint_dir()
if not pm.fingerprint_dir:
@@ -77,6 +76,7 @@ class ProfileService:
fingerprint=pm.current_fingerprint,
config_manager=getattr(pm, "config_manager", None),
parent_seed=getattr(pm, "parent_seed", None),
key_index=pm.KEY_INDEX,
)
if getattr(pm, "manifest_id", None) and hasattr(
pm.nostr_client, "_state_lock"

18
src/tests/test_manifest_id_privacy.py Normal file
View File

@@ -0,0 +1,18 @@
import asyncio
from helpers import dummy_nostr_client
def test_published_events_no_fingerprint(dummy_nostr_client):
client, relay = dummy_nostr_client
asyncio.run(client.publish_snapshot(b"secret"))
fingerprint = "fp"
events = list(relay.manifests) + list(relay.chunks.values())
seen = set()
for ev in events:
if id(ev) in seen:
continue
seen.add(id(ev))
assert fingerprint not in ev.id
for tag in getattr(ev, "tags", []):
assert fingerprint not in tag

9
src/tests/test_nostr_legacy_key_fallback.py
View File

@@ -5,7 +5,6 @@ import json
from helpers import DummyEvent, DummyFilter, dummy_nostr_client
from nostr.backup_models import KIND_MANIFEST, KIND_SNAPSHOT_CHUNK
from nostr.client import MANIFEST_ID_PREFIX
from nostr_sdk import Keys
@@ -55,9 +54,7 @@ def test_fetch_snapshot_legacy_key_fallback(dummy_nostr_client, monkeypatch):
],
}
)
manifest_event = DummyEvent(
KIND_MANIFEST, manifest_json, tags=[f"{MANIFEST_ID_PREFIX}fp"]
)
manifest_event = DummyEvent(KIND_MANIFEST, manifest_json, tags=["legacy"])
chunk_event = DummyEvent(
KIND_SNAPSHOT_CHUNK,
base64.b64encode(chunk_bytes).decode("utf-8"),
@@ -69,9 +66,9 @@ def test_fetch_snapshot_legacy_key_fallback(dummy_nostr_client, monkeypatch):
async def fake_fetch_events(f, _timeout):
call["count"] += 1
call["authors"].append(getattr(f, "author_pk", None))
if call["count"] <= 2:
if call["count"] == 1:
return type("R", (), {"to_vec": lambda self: []})()
elif call["count"] == 3:
elif call["count"] == 2:
return type("R", (), {"to_vec": lambda self: [manifest_event]})()
else:
return type("R", (), {"to_vec": lambda self: [chunk_event]})()

49
src/tests/test_nostr_legacy_manifest_id.py
View File

@@ -1,49 +0,0 @@
import asyncio
from helpers import TEST_SEED, dummy_nostr_client
from nostr.backup_models import KIND_MANIFEST
from nostr.client import MANIFEST_ID_PREFIX, NostrClient
def test_fetch_latest_snapshot_legacy_identifier(dummy_nostr_client, monkeypatch):
client, relay = dummy_nostr_client
data = b"legacy"
asyncio.run(client.publish_snapshot(data))
relay.manifests[-1].tags = [MANIFEST_ID_PREFIX.rstrip("-")]
relay.filters.clear()
orig_fetch = relay.fetch_events
async def fetch_events(self, f, timeout):
identifier = f.ids[0] if getattr(f, "ids", None) else None
kind = getattr(f, "kind_val", None)
if kind == KIND_MANIFEST:
events = [m for m in self.manifests if identifier in m.tags]
self.filters.append(f)
class Res:
def __init__(self, evs):
self._evs = evs
def to_vec(self):
return self._evs
return Res(events)
return await orig_fetch(f, timeout)
monkeypatch.setattr(
relay, "fetch_events", fetch_events.__get__(relay, relay.__class__)
)
enc_mgr = client.encryption_manager
monkeypatch.setattr(
enc_mgr, "decrypt_parent_seed", lambda: TEST_SEED, raising=False
)
monkeypatch.setattr("nostr.client.KeyManager", type(client.key_manager))
client2 = NostrClient(enc_mgr, "fp")
relay.filters.clear()
result = asyncio.run(client2.fetch_latest_snapshot())
assert result is not None
ids = [f.ids[0] for f in relay.filters]
assert ids[0] == f"{MANIFEST_ID_PREFIX}fp"
assert MANIFEST_ID_PREFIX.rstrip("-") in ids

6
src/tests/test_publish_json_result.py
View File

@@ -82,9 +82,11 @@ def test_publish_snapshot_success():
with patch.object(
client.client, "send_event", side_effect=fake_send
) as mock_send:
manifest, event_id = asyncio.run(client.publish_snapshot(b"data"))
with patch("nostr.snapshot.new_manifest_id", return_value=("id", b"nonce")):
manifest, event_id = asyncio.run(client.publish_snapshot(b"data"))
assert isinstance(manifest, Manifest)
assert event_id == "seedpass-manifest-fp"
assert event_id == "id"
assert manifest.nonce == base64.b64encode(b"nonce").decode("utf-8")
assert mock_send.await_count >= 1