503 lines
22 KiB
Python
503 lines
22 KiB
Python
"""Encrypted local E2E index shard against a disposable HTTP provider.
|
|
|
|
E2E nodes are encrypted with a synthetic memory secret using the wrapper's
|
|
own envelope code. The real reindex, feed sync, search and shard persistence
|
|
code run; only the key cache is seeded (standing in for an unlocked vault).
|
|
"""
|
|
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import pathlib
|
|
import re
|
|
import stat
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import pytest
|
|
|
|
SCRIPT = pathlib.Path(__file__).resolve().parents[1] / "resources/scripts/napseer_mcp_server.py"
|
|
PROJECT = "project-1"
|
|
# Synthetic fixture strings; none may appear in any file in the state dir.
|
|
BODY_TERM = "zephyrquartz"
|
|
BODY = f"calibrate the {BODY_TERM} manifold before launch"
|
|
META_VALUE = "obsidianfalcon"
|
|
PLAINTEXT_MARKERS = (BODY_TERM.encode(), META_VALUE.encode(), b"manifold before launch", b"brandnewsecret")
|
|
|
|
|
|
def memory_secret(version=3, secret_bytes=bytes(range(32))):
|
|
return {
|
|
"project_id": PROJECT, "secret_kind": "memory", "version": version,
|
|
"wrapping_epoch": 1, "bundle_version": version, "data_key_epoch": version,
|
|
"secret_bytes": secret_bytes, "created_at": "2026-09-01T00:00:00Z", "last_rotated_at": "2026-09-01T00:00:00Z",
|
|
}
|
|
|
|
|
|
def cache_secret(mod, secret, active=True):
|
|
slot = (PROJECT, "memory", secret["version"], secret["wrapping_epoch"], secret["bundle_version"], secret["data_key_epoch"])
|
|
mod.MEMORY_SECRET_CACHE[slot] = secret
|
|
if active:
|
|
mod.ACTIVE_MEMORY_SECRET_VERSIONS[PROJECT] = slot
|
|
|
|
|
|
def lock(mod):
|
|
mod.e2e_shards_close(flush=True)
|
|
mod.MEMORY_SECRET_CACHE.clear()
|
|
mod.ACTIVE_MEMORY_SECRET_VERSIONS.clear()
|
|
mod.PROJECT_VAULT_PASSPHRASE = None
|
|
|
|
|
|
def unlock(mod, secret=None):
|
|
cache_secret(mod, secret or memory_secret())
|
|
|
|
|
|
def plain_node(name, minute, content=None, folder="/notes"):
|
|
stamp = f"2026-09-01T12:{minute:02d}:00Z"
|
|
return {
|
|
"id": name, "project_id": PROJECT, "full_path": f"{folder}/{name}", "folder_path": folder,
|
|
"name": name, "type": "note", "metadata": {}, "tags": [], "aliases": [], "links": [],
|
|
"content_text": content if content is not None else name, "updated_at": stamp, "archived_at": None,
|
|
}
|
|
|
|
|
|
def e2e_node(mod, name, minute, content, metadata=None, secret=None, archived=False, folder="/notes"):
|
|
secret = secret or memory_secret()
|
|
node = plain_node(name, minute, folder=folder)
|
|
context = {"full_path": node["full_path"]}
|
|
node.update({
|
|
"content_text": "", "metadata": {}, "encryption_state": "encrypted",
|
|
"project_secret_version": secret["version"],
|
|
"encrypted_content_envelope": mod.encrypt_memory_bytes(
|
|
PROJECT, secret, "node_content", content.encode(), context, "text/plain; charset=utf-8"),
|
|
"encrypted_metadata_envelope": mod.encrypt_memory_bytes(
|
|
PROJECT, secret, "node_metadata", mod.stable_json_bytes(metadata or {}), context,
|
|
"application/json; charset=utf-8") if metadata else None,
|
|
})
|
|
if archived:
|
|
node["archived_at"] = node["updated_at"]
|
|
return node
|
|
|
|
|
|
class Provider:
|
|
def __init__(self):
|
|
self.nodes = {}
|
|
self.events = []
|
|
self.requests = []
|
|
|
|
def put(self, node, kind):
|
|
self.nodes[node["id"]] = copy.deepcopy(node)
|
|
self.events.append({"type": kind, "node_id": node["id"], "full_path": node["full_path"],
|
|
"new_revision": node["updated_at"]})
|
|
|
|
def delete(self, node_id):
|
|
node = self.nodes.pop(node_id)
|
|
self.events.append({"type": "node_deleted", "node_id": node_id, "full_path": node["full_path"],
|
|
"new_revision": None})
|
|
|
|
|
|
@pytest.fixture
|
|
def env(tmp_path, monkeypatch):
|
|
provider = Provider()
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *_args):
|
|
pass
|
|
|
|
def reply(self, status, payload):
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(payload).encode())
|
|
|
|
def do_GET(self):
|
|
url = urlsplit(self.path)
|
|
query = parse_qs(url.query)
|
|
kind = "changes" if url.path.endswith("/changes") else "list" if url.path.endswith("/nodes") else "get"
|
|
provider.requests.append((kind, url.path))
|
|
if "/vault/secrets/" in url.path:
|
|
return self.reply(503, {"error": {"code": "test_failure", "message": "no vault in test"}})
|
|
if kind == "changes":
|
|
after = query.get("after", ["head"])[0]
|
|
start = len(provider.events) if after == "head" else int(re.fullmatch(r"c(\d+)", after).group(1))
|
|
limit = int(query.get("limit", ["100"])[0])
|
|
page = provider.events[start:start + limit]
|
|
items = [{**event, "cursor": f"c{start + i + 1}", "project_id": PROJECT, "changed_fields": [],
|
|
"created_at": "2026-09-01T12:00:00Z"} for i, event in enumerate(page)]
|
|
end = start + len(page)
|
|
return self.reply(200, {"items": items, "next_cursor": f"c{end}", "has_more": end < len(provider.events)})
|
|
if kind == "list":
|
|
include_archived = query.get("include_archived", ["false"])[0] in {"true", "True", "1"}
|
|
rows = [copy.deepcopy(n) for n in provider.nodes.values() if include_archived or not n["archived_at"]]
|
|
rows.sort(key=lambda n: (n["updated_at"], n["id"]), reverse=True)
|
|
return self.reply(200, {"items": rows, "next_cursor": None})
|
|
node_id = url.path.rsplit("/", 1)[-1]
|
|
if node_id not in provider.nodes:
|
|
return self.reply(404, {"error": {"code": "not_found", "message": "missing"}})
|
|
return self.reply(200, copy.deepcopy(provider.nodes[node_id]))
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
sqlite_tmp = tmp_path / "sqlite-tmp"
|
|
sqlite_tmp.mkdir()
|
|
monkeypatch.setenv("SQLITE_TMPDIR", str(sqlite_tmp))
|
|
monkeypatch.setenv("TMPDIR", str(sqlite_tmp))
|
|
for name in ("NAPSEER_VAULT_PASSPHRASE", "NAPSEER_MASTER_PASSPHRASE", "NAPSEER_CRYPTO_BACKEND"):
|
|
monkeypatch.delenv(name, raising=False)
|
|
monkeypatch.setenv("NAPSEER_TOOL_PROFILES", "all")
|
|
monkeypatch.syspath_prepend(str(SCRIPT.parent))
|
|
spec = importlib.util.spec_from_file_location("e2e_shard_test_worker", SCRIPT)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
mod.AUTH_DIR = tmp_path / "project-state"
|
|
mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
|
|
mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.lock"
|
|
mod.BASE_URL = f"http://127.0.0.1:{server.server_port}"
|
|
mod.TOKEN = "disposable-test-token"
|
|
mod.AUTH = {"account_id": "test-account"}
|
|
mod.DEFAULT_PROJECT_ID = PROJECT
|
|
mod.CONFIGURED_TOOL_PROFILES = {"all"}
|
|
mod.refresh_public_auth_state = lambda: None
|
|
mod.require_unlocked = lambda _name: None
|
|
mod.resolve_project_id = lambda args: args.get("project_id", PROJECT)
|
|
mod.project_encryption_state = lambda _project_id: "encrypted"
|
|
mod.MEMORY_SECRET_CACHE.clear()
|
|
mod.ACTIVE_MEMORY_SECRET_VERSIONS.clear()
|
|
unlock(mod)
|
|
try:
|
|
yield mod, provider, sqlite_tmp
|
|
finally:
|
|
mod.e2e_shards_close(flush=False)
|
|
mod.select_crypto_backend()
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=2)
|
|
|
|
|
|
def seed(mod, provider):
|
|
provider.put(e2e_node(mod, "secretnote", 1, BODY, metadata={"owner": META_VALUE}), "node_created")
|
|
provider.put(plain_node("plainnote", 2, "ordinary plain body text"), "node_created")
|
|
return mod.reindex_project({})
|
|
|
|
|
|
def local_search(mod, text):
|
|
return mod.search_local_index({"q": text, "limit": 10})
|
|
|
|
|
|
def hit_ids(result):
|
|
return [item["id"] for item in result["items"]]
|
|
|
|
|
|
def codes(result):
|
|
return [warning["code"] for warning in result.get("warnings") or []]
|
|
|
|
|
|
def shard_file(mod):
|
|
return mod.e2e_shard_path(PROJECT)
|
|
|
|
|
|
def assert_no_plaintext_on_disk(mod, *extra_dirs):
|
|
for directory in (mod.AUTH_DIR, *extra_dirs):
|
|
for path in directory.rglob("*"):
|
|
if path.is_file():
|
|
raw = path.read_bytes()
|
|
for marker in PLAINTEXT_MARKERS:
|
|
assert marker not in raw, f"{marker!r} leaked into {path.name}"
|
|
|
|
|
|
def test_e2e_body_term_is_findable_with_snippet_while_unlocked(env):
|
|
mod, provider, _ = env
|
|
result = seed(mod, provider)
|
|
assert result["e2e_index"]["status"] == "applied"
|
|
|
|
local = local_search(mod, BODY_TERM)
|
|
assert hit_ids(local) == ["secretnote"]
|
|
item = local["items"][0]
|
|
assert item["search_source"] == "local_e2e"
|
|
assert BODY_TERM in item["snippet"]
|
|
assert BODY_TERM in item["matched_terms"]
|
|
assert hit_ids(local_search(mod, META_VALUE)) == ["secretnote"]
|
|
# Non-E2E results are unchanged and not marked as shard hits.
|
|
plain = local_search(mod, "ordinary")
|
|
assert hit_ids(plain) == ["plainnote"] and "search_source" not in plain["items"][0]
|
|
|
|
discovered = mod.discover_memory({"q": BODY_TERM, "mode": "local"})
|
|
assert [item["node_id"] for item in discovered["items"]] == ["secretnote"]
|
|
assert discovered["items"][0]["search_source"] == "local_e2e"
|
|
assert "local_e2e" in discovered["sources"]
|
|
assert not discovered.get("warnings")
|
|
|
|
|
|
def test_e2e_shard_hits_filter_on_effective_type(env):
|
|
mod, provider, _ = env
|
|
provider.put(e2e_node(mod, "legacyplan", 1, BODY, folder="/plans"), "node_created")
|
|
provider.put(e2e_node(mod, "scratchnote", 2, BODY), "node_created")
|
|
assert mod.reindex_project({})["e2e_index"]["status"] == "applied"
|
|
|
|
# Bodies are only in the shard, so these hits come from the E2E path.
|
|
plans = mod.search_local_index({"q": BODY_TERM, "limit": 10, "type": "plan"})
|
|
assert hit_ids(plans) == ["legacyplan"]
|
|
assert plans["items"][0]["search_source"] == "local_e2e"
|
|
assert (plans["items"][0]["type"], plans["items"][0]["effective_type"]) == ("note", "plan")
|
|
notes = mod.search_local_index({"q": BODY_TERM, "limit": 10, "type": "note"})
|
|
assert hit_ids(notes) == ["scratchnote"]
|
|
assert notes["items"][0]["search_source"] == "local_e2e"
|
|
|
|
|
|
def test_unprovisioned_secret_yields_no_e2e_hits_and_an_explicit_warning(env):
|
|
mod, provider, _ = env
|
|
seed(mod, provider)
|
|
lock(mod)
|
|
|
|
local = local_search(mod, BODY_TERM)
|
|
assert hit_ids(local) == []
|
|
assert codes(local) == ["e2e_index_not_provisioned"]
|
|
warning = local["warnings"][0]
|
|
assert warning["memory_secret"] == "not_provisioned"
|
|
assert warning["next_action"] == "nap gateway vault setup"
|
|
response = mod.search_memory({"q": BODY_TERM, "mode": "local", "view": "summary"})
|
|
assert response["items"] == []
|
|
assert "e2e_index_not_provisioned" in codes(response)
|
|
# Identity fields of the E2E node remain searchable from the main index.
|
|
assert hit_ids(local_search(mod, "secretnote")) == ["secretnote"]
|
|
# An unprovisioned reindex never prompts and writes no shard.
|
|
shard_file(mod).unlink()
|
|
assert "e2e_index" in mod.reindex_project({})
|
|
assert not shard_file(mod).exists()
|
|
|
|
unlock(mod)
|
|
assert hit_ids(local_search(mod, BODY_TERM)) == [] # shard not rebuilt yet: reported, not guessed
|
|
assert "e2e_index_incomplete" in codes(local_search(mod, BODY_TERM))
|
|
synced = mod.sync_local_index({})
|
|
assert synced["e2e_index"]["status"] == "complete"
|
|
assert hit_ids(local_search(mod, BODY_TERM)) == ["secretnote"]
|
|
|
|
|
|
def test_shard_file_holds_no_plaintext_and_no_side_files_exist(env):
|
|
mod, provider, sqlite_tmp = env
|
|
seed(mod, provider)
|
|
assert hit_ids(local_search(mod, BODY_TERM)) == ["secretnote"]
|
|
shard = mod.e2e_shard_loaded(PROJECT)
|
|
with shard.lock:
|
|
assert shard.conn.execute("PRAGMA journal_mode").fetchone()[0] == "memory"
|
|
assert shard.conn.execute("PRAGMA temp_store").fetchone()[0] == 2
|
|
databases = shard.conn.execute("PRAGMA database_list").fetchall()
|
|
assert [(row[1], row[2]) for row in databases] == [("main", "")]
|
|
mod.e2e_shards_close(flush=True)
|
|
|
|
path = shard_file(mod)
|
|
assert path.exists() and stat.S_IMODE(path.stat().st_mode) == 0o600
|
|
assert path.read_bytes().startswith(mod.E2E_SHARD_MAGIC)
|
|
names = {entry.name for entry in mod.AUTH_DIR.iterdir()}
|
|
assert names <= {"index.sqlite", "index.sqlite-wal", "index.sqlite-shm", "index.lock", path.name}
|
|
assert not any(name.endswith(".tmp") or "journal" in name for name in names)
|
|
assert list(sqlite_tmp.iterdir()) == []
|
|
assert_no_plaintext_on_disk(mod, sqlite_tmp)
|
|
# The shard lives in the private state dir, never in the checkout.
|
|
assert path.parent == mod.AUTH_DIR
|
|
|
|
|
|
def test_tampered_shard_fails_closed_rebuilds_and_reports(env):
|
|
mod, provider, _ = env
|
|
seed(mod, provider)
|
|
mod.e2e_shards_close(flush=True)
|
|
path = shard_file(mod)
|
|
blob = bytearray(path.read_bytes())
|
|
blob[-20] ^= 0x01
|
|
path.write_bytes(bytes(blob))
|
|
|
|
response = mod.search_memory({"q": BODY_TERM, "mode": "local", "view": "summary"})
|
|
|
|
discarded = [w for w in response["warnings"] if w["code"] == "e2e_index_discarded"]
|
|
assert discarded and discarded[0]["reason"] == "tampered"
|
|
assert [item["node_id"] for item in response["items"]] == ["secretnote"] # rebuilt from source
|
|
mod.e2e_shards_close(flush=True)
|
|
rebuilt = path.read_bytes()
|
|
assert rebuilt != bytes(blob)
|
|
# The rebuilt file loads cleanly in a fresh process state.
|
|
local = local_search(mod, BODY_TERM)
|
|
assert hit_ids(local) == ["secretnote"] and not local.get("warnings")
|
|
|
|
|
|
def test_truncated_or_foreign_shard_file_is_discarded(env):
|
|
mod, provider, _ = env
|
|
seed(mod, provider)
|
|
mod.e2e_shards_close(flush=True)
|
|
shard_file(mod).write_bytes(b"not a shard")
|
|
result = mod.sync_local_index({})
|
|
assert [w["reason"] for w in result["e2e_index"]["warnings"]] == ["format"]
|
|
assert result["e2e_index"]["status"] == "complete"
|
|
assert hit_ids(local_search(mod, BODY_TERM)) == ["secretnote"]
|
|
|
|
|
|
def test_wrong_key_fails_closed_and_reports_undecryptable_nodes(env):
|
|
mod, provider, _ = env
|
|
seed(mod, provider)
|
|
lock(mod)
|
|
unlock(mod, memory_secret(secret_bytes=bytes(range(1, 33))))
|
|
|
|
response = mod.search_memory({"q": BODY_TERM, "mode": "local", "view": "summary"})
|
|
|
|
assert response["items"] == []
|
|
by_code = {w["code"]: w for w in response["warnings"]}
|
|
assert by_code["e2e_index_discarded"]["reason"] == "wrong_key"
|
|
assert by_code["e2e_index_undecryptable"]["nodes"] == 1
|
|
assert_no_plaintext_on_disk(mod)
|
|
|
|
|
|
def test_secret_version_change_discards_and_rebuilds_under_the_new_key(env):
|
|
mod, provider, _ = env
|
|
seed(mod, provider)
|
|
assert hit_ids(local_search(mod, BODY_TERM)) == ["secretnote"]
|
|
|
|
# Rotation while loaded: the in-memory shard is re-keyed by discarding it.
|
|
cache_secret(mod, memory_secret(version=4, secret_bytes=bytes(range(100, 132))))
|
|
response = mod.search_memory({"q": BODY_TERM, "mode": "local", "view": "summary"})
|
|
reasons = [w.get("reason") for w in response["warnings"] if w["code"] == "e2e_index_discarded"]
|
|
assert reasons == ["key_rotated"]
|
|
assert [item["node_id"] for item in response["items"]] == ["secretnote"] # v3 node, v4 shard key
|
|
mod.e2e_shards_close(flush=True)
|
|
header = shard_file(mod).read_bytes()
|
|
assert b'"secret_version":4' in header
|
|
|
|
# Rotation between processes: a v4 file opened with a v5 active secret.
|
|
cache_secret(mod, memory_secret(version=5, secret_bytes=bytes(range(50, 82))))
|
|
result = mod.sync_local_index({})
|
|
assert [w["reason"] for w in result["e2e_index"]["warnings"]] == ["key_rotated"]
|
|
assert hit_ids(local_search(mod, BODY_TERM)) == ["secretnote"]
|
|
|
|
|
|
def test_feed_update_archive_and_delete_update_the_shard(env):
|
|
mod, provider, _ = env
|
|
seed(mod, provider)
|
|
provider.put(e2e_node(mod, "othernote", 3, "second brandnewsecret body"), "node_created")
|
|
mod.sync_local_index_from_feed(PROJECT)
|
|
assert hit_ids(local_search(mod, "brandnewsecret")) == ["othernote"]
|
|
|
|
provider.put(e2e_node(mod, "secretnote", 10, "rewritten body mentions heliotrope"), "node_updated")
|
|
result = mod.sync_local_index_from_feed(PROJECT)
|
|
assert result["e2e_index"]["indexed"] == 1
|
|
assert hit_ids(local_search(mod, "heliotrope")) == ["secretnote"]
|
|
assert hit_ids(local_search(mod, BODY_TERM)) == []
|
|
|
|
provider.delete("othernote")
|
|
provider.put(e2e_node(mod, "secretnote", 11, "archived text", archived=True), "node_archived")
|
|
mod.sync_local_index_from_feed(PROJECT)
|
|
assert hit_ids(local_search(mod, "brandnewsecret")) == []
|
|
assert hit_ids(local_search(mod, "heliotrope")) == []
|
|
shard = mod.e2e_shard_loaded(PROJECT)
|
|
with shard.lock:
|
|
assert shard.conn.execute("SELECT COUNT(*) FROM e2e_nodes").fetchone()[0] == 0
|
|
# A reloaded shard reflects the removals as well.
|
|
mod.e2e_shards_close(flush=True)
|
|
assert hit_ids(local_search(mod, "brandnewsecret")) == []
|
|
|
|
|
|
def test_local_write_path_indexes_and_removes_shard_text(env):
|
|
mod, provider, _ = env
|
|
seed(mod, provider)
|
|
written = e2e_node(mod, "writtennote", 20, "locally written umbraquill text")
|
|
mod.index_node(written)
|
|
assert hit_ids(local_search(mod, "umbraquill")) == ["writtennote"]
|
|
mod.remove_indexed_node("writtennote")
|
|
assert hit_ids(local_search(mod, "umbraquill")) == []
|
|
|
|
|
|
def test_stale_shard_row_is_not_served_and_is_refetched(env):
|
|
mod, provider, _ = env
|
|
seed(mod, provider)
|
|
mod.e2e_shards_close(flush=True)
|
|
# Another process advanced the main index without this shard file.
|
|
updated = e2e_node(mod, "secretnote", 30, "fresh body with vermilionkite")
|
|
provider.put(updated, "node_updated")
|
|
with mod.index_connect() as conn:
|
|
mod.index_node(updated, connection=conn)
|
|
local = local_search(mod, BODY_TERM)
|
|
assert hit_ids(local) == [] and "e2e_index_incomplete" in codes(local)
|
|
mod.sync_local_index({})
|
|
assert hit_ids(local_search(mod, "vermilionkite")) == ["secretnote"]
|
|
|
|
|
|
@pytest.mark.parametrize("writer,reader", [("native", "pure_python"), ("pure_python", "native")])
|
|
def test_crypto_backends_read_each_others_shard(env, writer, reader):
|
|
mod, provider, _ = env
|
|
if not mod.native_crypto_available():
|
|
pytest.skip("cryptography is not installed")
|
|
mod.select_crypto_backend(writer)
|
|
seed(mod, provider)
|
|
mod.e2e_shards_close(flush=True)
|
|
mod.select_crypto_backend(reader)
|
|
local = local_search(mod, BODY_TERM)
|
|
assert hit_ids(local) == ["secretnote"] and not local.get("warnings")
|
|
|
|
|
|
def test_pure_backend_large_shard_loads_in_background_and_warns(env, monkeypatch):
|
|
mod, provider, _ = env
|
|
seed(mod, provider)
|
|
mod.e2e_shards_close(flush=True)
|
|
mod.select_crypto_backend("pure_python")
|
|
monkeypatch.setattr(mod, "E2E_SHARD_PURE_SYNC_LOAD_MAX_BYTES", 1)
|
|
|
|
first = local_search(mod, BODY_TERM)
|
|
assert hit_ids(first) == []
|
|
slow = [w for w in first["warnings"] if w["code"] == "e2e_index_slow_crypto"]
|
|
assert slow and slow[0]["state"] == "loading"
|
|
loader = mod.e2e_shard_get(PROJECT).loading
|
|
if loader is not None:
|
|
loader.join(timeout=30)
|
|
assert hit_ids(local_search(mod, BODY_TERM)) == ["secretnote"]
|
|
|
|
|
|
def test_pure_backend_does_not_persist_oversized_shard(env, monkeypatch):
|
|
mod, provider, _ = env
|
|
mod.select_crypto_backend("pure_python")
|
|
monkeypatch.setattr(mod, "E2E_SHARD_PURE_MAX_PERSIST_BYTES", 1)
|
|
result = seed(mod, provider)
|
|
assert result["e2e_index"]["persist"] == "skipped_slow_crypto"
|
|
assert not shard_file(mod).exists()
|
|
local = local_search(mod, BODY_TERM)
|
|
assert hit_ids(local) == ["secretnote"]
|
|
assert [w.get("state") for w in local["warnings"]] == ["not_persisted"]
|
|
|
|
|
|
def test_unsupported_sqlite_disables_e2e_search_with_reason(env, monkeypatch):
|
|
mod, provider, _ = env
|
|
monkeypatch.setattr(mod, "e2e_shard_supported", lambda: (False, "sqlite_serialize_unavailable"))
|
|
seed(mod, provider)
|
|
local = local_search(mod, BODY_TERM)
|
|
assert hit_ids(local) == []
|
|
assert codes(local) == ["e2e_index_unsupported"]
|
|
assert not shard_file(mod).exists()
|
|
|
|
|
|
def test_project_without_e2e_nodes_creates_no_shard_or_warnings(env):
|
|
mod, provider, _ = env
|
|
provider.put(plain_node("plainnote", 2, "ordinary plain body text"), "node_created")
|
|
result = mod.reindex_project({})
|
|
assert "e2e_index" not in result
|
|
local = local_search(mod, "ordinary")
|
|
assert hit_ids(local) == ["plainnote"] and "warnings" not in local
|
|
assert not shard_file(mod).exists()
|
|
assert mod.index_status({})["e2e_index"]["loaded"] is False
|
|
|
|
|
|
def test_shard_key_is_bound_to_project_and_secret_identity(env):
|
|
mod, _provider, _ = env
|
|
secret = memory_secret()
|
|
identity, key, key_check = mod.e2e_shard_key_material(PROJECT, secret)
|
|
other_project = mod.e2e_shard_key_material("project-2", secret)
|
|
other_version = mod.e2e_shard_key_material(PROJECT, {**secret, "data_key_epoch": 4})
|
|
assert len(key) == 32 and key not in {other_project[1], other_version[1]}
|
|
blob = mod.e2e_shard_encode(PROJECT, identity, key, key_check, b"synthetic")
|
|
assert mod.e2e_shard_decode(blob, PROJECT, identity, key, key_check) == b"synthetic"
|
|
assert blob != mod.e2e_shard_encode(PROJECT, identity, key, key_check, b"synthetic") # fresh nonce
|
|
with pytest.raises(mod.E2EShardFileInvalid) as info:
|
|
mod.e2e_shard_decode(blob, "project-2", identity, key, key_check)
|
|
assert info.value.reason == "project_mismatch"
|
|
# Rewriting the header (AAD) without the key is detected.
|
|
forged = blob.replace(b'"index_schema":2', b'"index_schema":2 ')
|
|
assert forged != blob
|
|
with pytest.raises(mod.E2EShardFileInvalid):
|
|
mod.e2e_shard_decode(forged, PROJECT, identity, key, key_check)
|