diff --git a/resources/scripts/napseer_mcp_server.py b/resources/scripts/napseer_mcp_server.py index e3b505d..1a2b2a6 100755 --- a/resources/scripts/napseer_mcp_server.py +++ b/resources/scripts/napseer_mcp_server.py @@ -12,6 +12,7 @@ Environment credentials: NAPSEER_API_KEY or NAPSEER_TOKEN; never printed. """ import json +import atexit import base64 import concurrent.futures import getpass @@ -1392,6 +1393,8 @@ def touch_gateway(): def clear_gateway_runtime_caches(): global PROJECT_VAULT_PASSPHRASE + # Persist (ciphertext) and drop decrypted E2E index text with the secrets. + e2e_shards_close(flush=True) PROJECT_VAULT_PASSPHRASE = None MEMORY_SECRET_CACHE.clear() ACTIVE_MEMORY_SECRET_VERSIONS.clear() @@ -1496,6 +1499,7 @@ def gateway_status(): "telemetry_enabled": telemetry_enabled(), "telemetry_config_path": str(TELEMETRY_PATH), "telemetry_anonymous_id_present": bool(telemetry_state().get("anonymous_install_id")), + **crypto_backend_status(), } @@ -3409,10 +3413,9 @@ def _aes_key_expand_256(key): return words -def _aes_encrypt_block_256(key, block): +def _aes_encrypt_block_words(words, block): if len(block) != 16: raise RuntimeError("AES block must be 16 bytes") - words = _aes_key_expand_256(key) state = bytearray(block) _aes_add_round_key(state, words, 0) for round_index in range(1, 14): @@ -3426,6 +3429,10 @@ def _aes_encrypt_block_256(key, block): return bytes(state) +def _aes_encrypt_block_256(key, block): + return _aes_encrypt_block_words(_aes_key_expand_256(key), block) + + def _gcm_gf_mul(left, right): z = 0 v = right @@ -3461,43 +3468,132 @@ def _gcm_inc32(counter): return prefix + value.to_bytes(4, "big") -def _aes_gcm_crypt(key, nonce, aad, data): +def _aes_gcm_crypt(key, nonce, aad, data, words=None): if len(nonce) != 12: raise RuntimeError("AES-GCM nonce must be 12 bytes") + if words is None: + words = _aes_key_expand_256(key) counter = nonce + b"\x00\x00\x00\x01" cursor = _gcm_inc32(counter) output = bytearray() for offset in range(0, len(data), 16): block = data[offset:offset + 16] - stream = _aes_encrypt_block_256(key, cursor) + stream = _aes_encrypt_block_words(words, cursor) output.extend(bytes(left ^ right for left, right in zip(block, stream))) cursor = _gcm_inc32(cursor) return bytes(output), counter -def aes_gcm_encrypt(key, nonce, aad, plaintext): - ciphertext, counter = _aes_gcm_crypt(key, nonce, aad, plaintext) - hash_subkey = _aes_encrypt_block_256(key, b"\x00" * 16) +def _aes_gcm_check_inputs(key, nonce): + if len(key) != 32: + raise RuntimeError("AES-256 key must be 32 bytes") + if len(nonce) != 12: + raise RuntimeError("AES-GCM nonce must be 12 bytes") + + +def _pure_aes_gcm_encrypt(key, nonce, aad, plaintext): + _aes_gcm_check_inputs(key, nonce) + words = _aes_key_expand_256(key) + ciphertext, counter = _aes_gcm_crypt(key, nonce, aad, plaintext, words) + hash_subkey = _aes_encrypt_block_words(words, b"\x00" * 16) auth = _gcm_ghash(hash_subkey, aad, ciphertext) - tag_mask = _aes_encrypt_block_256(key, counter) + tag_mask = _aes_encrypt_block_words(words, counter) tag = bytes(left ^ right for left, right in zip(auth, tag_mask)) return ciphertext + tag -def aes_gcm_decrypt(key, nonce, aad, packed): +def _pure_aes_gcm_decrypt(key, nonce, aad, packed): + _aes_gcm_check_inputs(key, nonce) if len(packed) < 16: raise RuntimeError("AES-GCM ciphertext is missing tag") + words = _aes_key_expand_256(key) ciphertext, tag = packed[:-16], packed[-16:] - hash_subkey = _aes_encrypt_block_256(key, b"\x00" * 16) + hash_subkey = _aes_encrypt_block_words(words, b"\x00" * 16) auth = _gcm_ghash(hash_subkey, aad, ciphertext) counter = nonce + b"\x00\x00\x00\x01" - expected = bytes(left ^ right for left, right in zip(auth, _aes_encrypt_block_256(key, counter))) + expected = bytes(left ^ right for left, right in zip(auth, _aes_encrypt_block_words(words, counter))) if not hmac.compare_digest(tag, expected): raise RuntimeError("AES-GCM authentication failed") - plaintext, _ = _aes_gcm_crypt(key, nonce, aad, ciphertext) + plaintext, _ = _aes_gcm_crypt(key, nonce, aad, ciphertext, words) return plaintext +def _native_aes_gcm_encrypt(key, nonce, aad, plaintext): + _aes_gcm_check_inputs(key, nonce) + return _NativeAESGCM(bytes(key)).encrypt(bytes(nonce), bytes(plaintext), bytes(aad)) + + +def _native_aes_gcm_decrypt(key, nonce, aad, packed): + _aes_gcm_check_inputs(key, nonce) + if len(packed) < 16: + raise RuntimeError("AES-GCM ciphertext is missing tag") + try: + return _NativeAESGCM(bytes(key)).decrypt(bytes(nonce), bytes(packed), bytes(aad)) + except _NativeInvalidTag: + # Same failure as the pure backend; no library detail is exposed. + raise RuntimeError("AES-GCM authentication failed") from None + + +# AES-256-GCM backend. The optional `cryptography` package (OpenSSL) is used +# when importable; otherwise the stdlib-only pure-Python implementation is +# used. Both produce and accept the same packed layout: ciphertext || 16-byte +# tag, 96-bit nonce, and raise RuntimeError("AES-GCM authentication failed") +# on a tag mismatch. NAPSEER_CRYPTO_BACKEND=pure_python forces the fallback. +CRYPTO_BACKEND_NATIVE = "native" +CRYPTO_BACKEND_PURE = "pure_python" +try: + from cryptography.hazmat.primitives.ciphers.aead import AESGCM as _NativeAESGCM + from cryptography.exceptions import InvalidTag as _NativeInvalidTag + _NATIVE_CRYPTO_ERROR = None +except Exception as _native_crypto_exc: # ImportError or a broken binding + _NativeAESGCM = None + _NativeInvalidTag = None + _NATIVE_CRYPTO_ERROR = _native_crypto_exc.__class__.__name__ + + +def native_crypto_available(): + return _NativeAESGCM is not None + + +def select_crypto_backend(name=None): + """Select the AES-GCM backend; returns the active backend name.""" + global CRYPTO_BACKEND + requested = str(name or os.environ.get("NAPSEER_CRYPTO_BACKEND") or "").strip().lower() + if requested == CRYPTO_BACKEND_PURE or not native_crypto_available(): + CRYPTO_BACKEND = CRYPTO_BACKEND_PURE + else: + CRYPTO_BACKEND = CRYPTO_BACKEND_NATIVE + return CRYPTO_BACKEND + + +CRYPTO_BACKEND = CRYPTO_BACKEND_PURE +select_crypto_backend() + + +def crypto_backend_status(): + status = {"crypto_backend": CRYPTO_BACKEND, "native_crypto_available": native_crypto_available()} + if CRYPTO_BACKEND == CRYPTO_BACKEND_PURE: + status["crypto_backend_note"] = ( + "Pure-Python AES-GCM is slow; install the optional `cryptography` package " + "(python3 -m pip install --user cryptography) for native speed." + if not native_crypto_available() + else "Pure-Python AES-GCM was selected by NAPSEER_CRYPTO_BACKEND." + ) + return status + + +def aes_gcm_encrypt(key, nonce, aad, plaintext): + if CRYPTO_BACKEND == CRYPTO_BACKEND_NATIVE: + return _native_aes_gcm_encrypt(key, nonce, aad, plaintext) + return _pure_aes_gcm_encrypt(key, nonce, aad, plaintext) + + +def aes_gcm_decrypt(key, nonce, aad, packed): + if CRYPTO_BACKEND == CRYPTO_BACKEND_NATIVE: + return _native_aes_gcm_decrypt(key, nonce, aad, packed) + return _pure_aes_gcm_decrypt(key, nonce, aad, packed) + + def relay_node_crypto(operation, key, aad, nonce, ciphertext=None, payload=None): if operation == "decrypt": return aes_gcm_decrypt(key, nonce, aad, ciphertext).decode("utf-8") @@ -4131,7 +4227,7 @@ def is_backend_at_rest_envelope(envelope): ) -def decrypt_node_for_return(node, project_id=None): +def decrypt_node_for_return(node, project_id=None, secret_resolver=None): if not isinstance(node, dict) or node.get("encryption_state") != "encrypted": return node content_envelope = node.get("encrypted_content_envelope") @@ -4146,7 +4242,7 @@ def decrypt_node_for_return(node, project_id=None): secret_version = node.get("project_secret_version") if not secret_version and isinstance(content_envelope, dict): secret_version = content_envelope.get("active_project_secret_version") - secret = memory_secret_for_read(project_id, secret_version) + secret = (secret_resolver or memory_secret_for_read)(project_id, secret_version) context = {"full_path": node.get("full_path")} if node.get("full_path") else {"node_id": node.get("id")} decrypted = dict(node) if content_envelope: @@ -7095,26 +7191,22 @@ def index_node(node, *, connection=None): conn = index_connect() try: with conn: - return index_node(node, connection=conn) + index_node(node, connection=conn) finally: conn.close() finally: release_index_file_lock(lock) + # Mirror the committed row into the encrypted E2E shard (never raises). + e2e_shard_after_index(node.get("project_id") or DEFAULT_PROJECT_ID or "", nodes=[node]) + return conn = connection tags = node.get("tags") or [] aliases = node.get("aliases") or [] links = node.get("links") or [] - encrypted = node.get("encryption_state") == "encrypted" - backend_at_rest_decrypted = bool(node.get("at_rest_key_version_id")) and not node.get("encrypted_content_envelope") - standard_content_for_index = (not encrypted) or backend_at_rest_decrypted + standard_content_for_index = not node_content_withheld_from_index(node) metadata = (node.get("metadata") or {}) if standard_content_for_index and not node.get("encrypted_metadata_envelope") else {} content_text = (node.get("content_text") or "") if standard_content_for_index else "" - title = ( - node.get("title") - or metadata.get("title") - or node.get("name") - or pathlib.PurePosixPath(str(node.get("full_path") or "")).name - ) + title, fts_values = local_index_fts_values(node, metadata=metadata, content_text=content_text) status = node_status(node) project_id = node.get("project_id") or DEFAULT_PROJECT_ID or "" delete_node_fts_row(conn, node["id"]) @@ -7200,30 +7292,58 @@ def index_node(node, *, connection=None): iso_now(), ), ) - fts_row = conn.execute( - f"INSERT INTO local_index_fts ({', '.join(LOCAL_INDEX_FTS_COLUMNS)}) " - f"VALUES ({', '.join('?' for _ in LOCAL_INDEX_FTS_COLUMNS)})", - ( - node["id"], - node.get("project_id") or DEFAULT_PROJECT_ID or "", - title or "", - node.get("full_path") or "", - node.get("name") or "", - node.get("folder_path") or "", - node.get("type") or "note", - " ".join(tags), - " ".join(aliases), - " ".join(flatten_index_values(links)), - " ".join(flatten_index_values(metadata)), - content_text, - ), - ) + fts_row = conn.execute(LOCAL_INDEX_FTS_INSERT.format(table="local_index_fts"), fts_values) conn.execute( "UPDATE local_index_nodes SET fts_rowid = ? WHERE node_id = ?", (fts_row.lastrowid, node["id"]), ) +LOCAL_INDEX_FTS_INSERT = ( + f"INSERT INTO {{table}} ({', '.join(LOCAL_INDEX_FTS_COLUMNS)}) " + f"VALUES ({', '.join('?' for _ in LOCAL_INDEX_FTS_COLUMNS)})" +) + + +def node_content_withheld_from_index(node): + """True for E2E nodes: encrypted without a backend at-rest decryption. + + Their body/metadata never enter the on-disk index; the encrypted E2E + shard holds that text instead. + """ + encrypted = node.get("encryption_state") == "encrypted" + backend_at_rest_decrypted = bool(node.get("at_rest_key_version_id")) and not node.get("encrypted_content_envelope") + return encrypted and not backend_at_rest_decrypted + + +def local_index_fts_values(node, *, metadata, content_text): + """(title, FTS row values in LOCAL_INDEX_FTS_COLUMNS order) for one node.""" + metadata = metadata if isinstance(metadata, dict) else {} + tags = node.get("tags") or [] + aliases = node.get("aliases") or [] + links = node.get("links") or [] + title = ( + node.get("title") + or metadata.get("title") + or node.get("name") + or pathlib.PurePosixPath(str(node.get("full_path") or "")).name + ) + return title, ( + node["id"], + node.get("project_id") or DEFAULT_PROJECT_ID or "", + title or "", + node.get("full_path") or "", + node.get("name") or "", + node.get("folder_path") or "", + node.get("type") or "note", + " ".join(str(tag) for tag in tags), + " ".join(str(alias) for alias in aliases), + " ".join(flatten_index_values(links)), + " ".join(flatten_index_values(metadata)), + content_text or "", + ) + + def node_revision_token(node): """The server revision of a node row (its RFC 3339 updated_at).""" value = node.get("revision") or node.get("updated_at") @@ -7280,6 +7400,7 @@ def remove_indexed_node(node_id): remove_indexed_node_rows(conn, node_id) finally: release_index_file_lock(lock) + e2e_shards_remove_nodes([node_id]) # Function words only. Domain words (plan, work, decide, done, ...) are @@ -7960,41 +8081,9 @@ def search_local_index(args): with index_connect() as conn: diagnostics = local_index_diagnostics(project_id, conn) variant = variants[0] - where = ["local_index_fts MATCH ?", "n.project_id = ?"] - parameters = [variant["query"], project_id] - if as_bool(args.get("archived_only"), False): - where.append("n.archived = 1") - elif not as_bool(args.get("include_archived"), False): - where.append("n.archived = 0") - if args.get("folder_path") not in (None, ""): - where.append("n.folder_path = ?") - parameters.append(str(args["folder_path"])) - if args.get("type") not in (None, ""): - where.append("n.node_type = ?") - parameters.append(str(args["type"])) - if args.get("tag") not in (None, ""): - where.append( - "EXISTS (SELECT 1 FROM local_index_node_tags t " - "WHERE t.project_id = n.project_id AND t.node_id = n.node_id AND t.tag = ?)" - ) - parameters.append(str(args["tag"])) - if args.get("updated_after"): - where.append("n.updated_at > ?") - parameters.append(args["updated_after"]) - if args.get("updated_before"): - where.append("n.updated_at < ?") - parameters.append(args["updated_before"]) - statuses = filters.get("status") or [] - if statuses: - where.append(f"n.status IN ({','.join('?' for _ in statuses)})") - parameters.extend(statuses) - excluded = list(filters.get("exclude_status") or []) - if filters.get("active_only"): - excluded = unique_preserve_order(excluded + sorted(ACTIVE_EXCLUDED_STATUSES)) - where.append("n.archived = 0") - if excluded: - where.append(f"COALESCE(n.status, '') NOT IN ({','.join('?' for _ in excluded)})") - parameters.extend(excluded) + filter_where, filter_parameters = local_index_filter_clauses(args, filters) + where = ["local_index_fts MATCH ?", "n.project_id = ?", *filter_where] + parameters = [variant["query"], project_id, *filter_parameters] parameters.append(candidate_limit) try: rows = conn.execute( @@ -8058,15 +8147,39 @@ def search_local_index(args): # One candidate per node even if an older worker left a stale FTS row. candidates = unique_nodes(candidates) attach_local_term_coverage(conn, candidates, query_analysis) + # E2E bodies live only in the encrypted in-memory shard. A shard + # hit replaces the identity-only main row of the same node. + try: + e2e = e2e_shard_search( + project_id, conn, variant["query"], query_analysis, + filter_where, filter_parameters, candidate_limit, + ) + except SafeToolError as exc: + if exc.code == "request_cancelled": + raise + e2e = e2e_search_failure(exc) + except Exception as exc: + # The identity-only answer stays usable; E2E bodies are reported unsearched. + e2e = e2e_search_failure(exc) + if e2e["candidates"]: + main_ranks = {item["id"]: item.get("rank") for item in candidates} + for item in e2e["candidates"]: + main_rank = main_ranks.get(item["id"]) + if isinstance(main_rank, (int, float)) and main_rank < item["rank"]: + item["rank"] = main_rank + e2e_ids = {item["id"] for item in e2e["candidates"]} + candidates = e2e["candidates"] + [item for item in candidates if item["id"] not in e2e_ids] items = rank_search_items( candidates, {**query_analysis, "requested_statuses": sorted(search_requested_statuses(args))}, limit=limit, ) - attach_local_snippets(conn, items, variant["query"]) + attach_local_snippets(conn, [item for item in items if not item.get("_e2e")], variant["query"]) + e2e_shard_attach_snippets(project_id, items, variant["query"]) for item in items: item.pop("_fts_rowid", None) item.pop("_content_withheld", None) + item.pop("_e2e", None) attempts = [ { "strategy": variant["strategy"], @@ -8080,6 +8193,7 @@ def search_local_index(args): diagnostics["returned"] = len(items) diagnostics["query_count"] = 1 diagnostics["candidate_limit"] = candidate_limit + diagnostics["e2e_index"] = e2e["diagnostics"] diagnostics["filters_applied"] = [ key for key in ( @@ -8096,12 +8210,55 @@ def search_local_index(args): ) if args.get(key) not in (None, "", False, []) ] - return { + result = { "items": items, "query": query_analysis, "search_strategies": attempts, "index": diagnostics, } + if e2e["warnings"]: + result["warnings"] = e2e["warnings"] + return result + + +def local_index_filter_clauses(args, filters): + """WHERE clauses (on local_index_nodes n) and parameters for search filters.""" + where = [] + parameters = [] + if as_bool(args.get("archived_only"), False): + where.append("n.archived = 1") + elif not as_bool(args.get("include_archived"), False): + where.append("n.archived = 0") + if args.get("folder_path") not in (None, ""): + where.append("n.folder_path = ?") + parameters.append(str(args["folder_path"])) + if args.get("type") not in (None, ""): + where.append("n.node_type = ?") + parameters.append(str(args["type"])) + if args.get("tag") not in (None, ""): + where.append( + "EXISTS (SELECT 1 FROM local_index_node_tags t " + "WHERE t.project_id = n.project_id AND t.node_id = n.node_id AND t.tag = ?)" + ) + parameters.append(str(args["tag"])) + if args.get("updated_after"): + where.append("n.updated_at > ?") + parameters.append(args["updated_after"]) + if args.get("updated_before"): + where.append("n.updated_at < ?") + parameters.append(args["updated_before"]) + statuses = filters.get("status") or [] + if statuses: + where.append(f"n.status IN ({','.join('?' for _ in statuses)})") + parameters.extend(statuses) + excluded = list(filters.get("exclude_status") or []) + if filters.get("active_only"): + excluded = unique_preserve_order(excluded + sorted(ACTIVE_EXCLUDED_STATUSES)) + where.append("n.archived = 0") + if excluded: + where.append(f"COALESCE(n.status, '') NOT IN ({','.join('?' for _ in excluded)})") + parameters.extend(excluded) + return where, parameters LOCAL_SNIPPET_TOKENS = 24 @@ -8242,16 +8399,37 @@ def reindex_project(args): conn.close() finally: release_index_file_lock(lock) - return { + # Rebuild the encrypted E2E shard from the same listing, outside the lock. + e2e_result = e2e_shard_after_index( + project_id, nodes=nodes, replace_all=True, wait=True, force_flush=True, + deadline=time.monotonic() + E2E_SHARD_SYNC_BUDGET_SECONDS, + ) + result = { "project_id": project_id, "indexed": len(nodes), "index_path": str(INDEX_PATH), "graph_complete": True, "feed": feed_status, } + if e2e_result.get("status") != "not_applicable": + result["e2e_index"] = e2e_result + return result def sync_local_index(args): + result = sync_local_index_main(args) + project_id = result.get("project_id") if isinstance(result, dict) else None + if project_id: + # Finish rebuilding the encrypted E2E shard (bounded) and report it. + e2e_result = e2e_shard_reconcile( + project_id, budget_seconds=E2E_SHARD_SYNC_BUDGET_SECONDS, wait=True, force_flush=True + ) + if e2e_result.get("status") != "not_applicable": + result["e2e_index"] = {**(result.get("e2e_index") or {}), **e2e_result} + return result + + +def sync_local_index_main(args): args = normalize_discovery_ranges(args or {}) project_id = resolve_project_id(args) limit = max(1, min(int(args.get("limit", 200)), 200)) @@ -8388,6 +8566,9 @@ def sync_local_index_by_timestamp(args, project_id, status_before, *, feed_statu continue index_node(node) indexed += 1 + shard = e2e_shard_loaded(project_id) + if shard is not None: + e2e_shard_flush(shard, force=True) return { "project_id": project_id, "status": "synced", @@ -8407,7 +8588,7 @@ def sync_local_index_by_timestamp(args, project_id, status_before, *, feed_statu def index_status(args): project_id = resolve_project_id(args) - return local_index_diagnostics(project_id) + return {**local_index_diagnostics(project_id), "e2e_index": e2e_shard_status(project_id)} CHANGE_FEED_NODE_REFRESH_TYPES = {"node_created", "node_updated", "node_restored", "node_changed"} @@ -8552,6 +8733,7 @@ def sync_local_index_from_feed(project_id, *, max_pages=FEED_SYNC_MAX_PAGES, raise SafeToolError("feed_not_initialized", "The local index has no change-feed cursor yet.") deadline = time.monotonic() + max(0.1, float(budget_seconds)) totals = {"events": 0, "indexed": 0, "removed": 0, "skipped_current": 0, "refetched": 0} + e2e_feed_results = [] pages = 0 has_more = False status = "synced" @@ -8640,12 +8822,21 @@ def sync_local_index_from_feed(project_id, *, max_pages=FEED_SYNC_MAX_PAGES, conn.close() finally: release_index_file_lock(lock) + if status != "superseded" and operations: + e2e_changes = e2e_shard_after_index( + project_id, + nodes=[value for action, value in operations if action == "index"], + removed_ids=[value for action, value in operations if action == "remove"], + deadline=deadline, + ) + if e2e_changes.get("status") not in (None, "not_applicable"): + e2e_feed_results.append(e2e_changes) pages += 1 cursor = next_cursor has_more = page.get("has_more") is True if not has_more: break - return { + result = { "project_id": project_id, "status": status, "mode": "feed", @@ -8655,6 +8846,16 @@ def sync_local_index_from_feed(project_id, *, max_pages=FEED_SYNC_MAX_PAGES, **totals, "index_path": str(INDEX_PATH), } + if e2e_feed_results: + shard = e2e_shard_loaded(project_id) + persist = e2e_shard_flush(shard, force=True) if shard is not None else None + result["e2e_index"] = { + "status": "partial" if any(item.get("status") != "applied" for item in e2e_feed_results) else "applied", + "indexed": sum(int(item.get("indexed") or 0) for item in e2e_feed_results), + "pending": sum(int(item.get("pending") or 0) for item in e2e_feed_results), + "persist": persist, + } + return result def auto_refresh_local_index(project_id): @@ -8809,9 +9010,1042 @@ def clear_local_index(args): set_local_index_feed_cursor(conn, project_id, None) finally: release_index_file_lock(lock) + e2e_shard_drop_if_unneeded(project_id) return {"project_id": project_id, "cleared": True} +# --------------------------------------------------------------------------- +# Encrypted local E2E index shard. +# +# E2E node bodies/metadata never enter the on-disk index (index_node withholds +# them). When the project memory secret is available, this process decrypts +# them into a separate SQLite database that lives only in memory (":memory:" +# + deserialize, journal and temp store in memory, nothing attached). It is +# persisted only as serialize() -> AES-256-GCM -> atomic 0600 write in the +# private per-worktree state dir. The main index stays authoritative for which +# E2E nodes exist and their revision; the shard is a cache reconciled against +# it, so a discarded or stale shard is rebuilt from the server. +# --------------------------------------------------------------------------- +E2E_SHARD_MAGIC = b"NAPE2EIX" +E2E_SHARD_FORMAT = 1 +E2E_SHARD_KIND = "napseer.local-e2e-index" +E2E_SHARD_SCHEMA_VERSION = 1 +E2E_SHARD_USER_VERSION = E2E_SHARD_SCHEMA_VERSION * 100 + INDEX_SCHEMA_VERSION +E2E_SHARD_HKDF_INFO = b"napseer.local-e2e-index.v1" +E2E_SHARD_MAX_HEADER_BYTES = 4096 +E2E_SHARD_SAVE_INTERVAL_SECONDS = 5.0 +# Pure-Python AES-GCM runs at ~150 KiB/s. Larger shards load on a background +# thread (searches report e2e_index_slow_crypto meanwhile) and are not +# persisted above the second bound; native crypto has no such limits. +E2E_SHARD_PURE_SYNC_LOAD_MAX_BYTES = 256 * 1024 +E2E_SHARD_PURE_MAX_PERSIST_BYTES = 512 * 1024 +E2E_SHARD_RECONCILE_INTERVAL_SECONDS = 10.0 +E2E_SHARD_RECONCILE_BUDGET_SECONDS = 2.0 +E2E_SHARD_SYNC_BUDGET_SECONDS = 60.0 +E2E_SHARD_SECRET_RETRY_SECONDS = 60.0 +E2E_SHARD_GET_THRESHOLD = 10 +E2E_SHARD_LIST_MAX_PAGES = 50 +E2E_SHARDS = {} +E2E_SHARDS_LOCK = threading.Lock() +E2E_RECONCILE_STATE = {} + +E2E_WARNING_MESSAGES = { + "e2e_index_locked": ( + "End-to-end encrypted node bodies were not searched: the project memory secret is locked on this " + "machine. Provide the project vault passphrase to the MCP server (NAPSEER_VAULT_PASSPHRASE) or run a " + "command that unlocks it, then search again." + ), + "e2e_index_unavailable": ( + "End-to-end encrypted node bodies were not searched: the project memory secret could not be loaded. " + "Retry when Napseer is reachable." + ), + "e2e_index_unsupported": ( + "End-to-end encrypted node bodies are not searchable locally: this Python's sqlite3 lacks " + "serialize/deserialize (Python 3.11+ is required)." + ), + "e2e_index_discarded": ( + "The encrypted local E2E index could not be used and was discarded; it is being rebuilt from Napseer. " + "E2E body results may be incomplete until the rebuild finishes." + ), + "e2e_index_incomplete": ( + "Some end-to-end encrypted nodes are not yet in the encrypted local E2E index; their bodies were not " + "searched. Run nap_index_sync to finish rebuilding it." + ), + "e2e_index_slow_crypto": ( + "The encrypted local E2E index uses the slow pure-Python AES-GCM backend; install the optional " + "`cryptography` package (python3 -m pip install --user cryptography) for native speed." + ), + "e2e_index_undecryptable": ( + "Some end-to-end encrypted nodes could not be decrypted with the available project memory secret; " + "their bodies were not searched." + ), +} + + +class E2EShardLocked(Exception): + """The memory secret needed for E2E text is not available without a prompt.""" + + +class E2EShardSecretUnavailable(Exception): + """The memory secret could not be loaded (network/server/bundle failure).""" + + +class E2EShardFileInvalid(Exception): + def __init__(self, reason): + super().__init__(reason) + self.reason = reason + + +def e2e_warning(code, **fields): + return {"code": code, "message": E2E_WARNING_MESSAGES[code], **fields} + + +def e2e_shard_supported(): + if not (hasattr(sqlite3.Connection, "serialize") and hasattr(sqlite3.Connection, "deserialize")): + return False, "sqlite_serialize_unavailable" + return True, None + + +def e2e_shard_path(project_id): + digest = hashlib.sha256(str(project_id).encode("utf-8")).hexdigest()[:32] + return AUTH_DIR / f"index-e2e-{digest}.enc" + + +def e2e_noninteractive_passphrase_available(): + return bool( + PROJECT_VAULT_PASSPHRASE + or os.environ.get("NAPSEER_VAULT_PASSPHRASE") + or os.environ.get("NAPSEER_MASTER_PASSPHRASE") + ) + + +def e2e_cached_memory_secret(project_id, version=None): + if version is None: + slot = ACTIVE_MEMORY_SECRET_VERSIONS.get(memory_secret_cache_key(project_id)) + else: + slot = find_memory_secret_cache_slot(project_id, version) + return MEMORY_SECRET_CACHE.get(slot) if slot else None + + +def e2e_memory_secret_for_read(project_id, version): + """memory_secret_for_read that never prompts on a TTY.""" + cached = e2e_cached_memory_secret(project_id, int(version or 0)) + if cached: + return cached + if not e2e_noninteractive_passphrase_available(): + raise E2EShardLocked() + try: + return memory_secret_for_read(project_id, version) + except Exception as exc: + raise E2EShardSecretUnavailable(safe_exception_text(exc)) from exc + + +def e2e_shard_key_material(project_id, secret): + """HKDF-SHA256 shard key bound to the project and memory-secret identity.""" + identity = ( + int(secret["version"]), + int(secret["wrapping_epoch"]), + int(secret["bundle_version"]), + int(secret["data_key_epoch"]), + ) + project = str(project_id).encode("utf-8") + salt = hashlib.sha256(E2E_SHARD_HKDF_INFO + b":salt:" + project).digest() + info = ( + E2E_SHARD_HKDF_INFO + + b":project:" + project + + (":memory:v%d:w%d:b%d:d%d" % identity).encode("ascii") + ) + key = hkdf_sha256(bytes(secret["secret_bytes"]), salt, info, 32) + key_check = hmac.new(key, E2E_SHARD_HKDF_INFO + b":key-check", hashlib.sha256).hexdigest()[:32] + return identity, key, key_check + + +def e2e_shard_header(project_id, identity, key_check): + return stable_json_bytes({ + "kind": E2E_SHARD_KIND, + "format": E2E_SHARD_FORMAT, + "aead": "AES-256-GCM", + "project_id": str(project_id), + "shard_schema": E2E_SHARD_SCHEMA_VERSION, + "index_schema": INDEX_SCHEMA_VERSION, + "secret_kind": MEMORY_SECRET_KIND, + "secret_version": identity[0], + "wrapping_epoch": identity[1], + "bundle_version": identity[2], + "data_key_epoch": identity[3], + "key_id": key_check, + }) + + +def e2e_shard_encode(project_id, identity, key, key_check, plaintext): + """magic | format u8 | header length u16 | header JSON | nonce | ciphertext||tag. + + The whole prefix (magic through header) is the AEAD associated data, so + the project, schema versions and secret identity are authenticated. A + fresh random 96-bit nonce is used for every save. + """ + header = e2e_shard_header(project_id, identity, key_check) + prefix = E2E_SHARD_MAGIC + struct.pack(">BH", E2E_SHARD_FORMAT, len(header)) + header + nonce = secrets.token_bytes(12) + return prefix + nonce + aes_gcm_encrypt(key, nonce, prefix, plaintext) + + +def e2e_shard_decode(blob, project_id, identity, key, key_check): + fixed = len(E2E_SHARD_MAGIC) + 3 + if len(blob) < fixed or blob[:len(E2E_SHARD_MAGIC)] != E2E_SHARD_MAGIC: + raise E2EShardFileInvalid("format") + file_format, header_length = struct.unpack(">BH", blob[len(E2E_SHARD_MAGIC):fixed]) + if (file_format != E2E_SHARD_FORMAT or header_length > E2E_SHARD_MAX_HEADER_BYTES + or len(blob) < fixed + header_length + 12 + 16): + raise E2EShardFileInvalid("format") + try: + header = json.loads(blob[fixed:fixed + header_length].decode("utf-8")) + except (UnicodeDecodeError, ValueError): + raise E2EShardFileInvalid("format") from None + if not isinstance(header, dict) or header.get("kind") != E2E_SHARD_KIND or header.get("format") != E2E_SHARD_FORMAT: + raise E2EShardFileInvalid("format") + if header.get("project_id") != str(project_id): + raise E2EShardFileInvalid("project_mismatch") + if header.get("shard_schema") != E2E_SHARD_SCHEMA_VERSION or header.get("index_schema") != INDEX_SCHEMA_VERSION: + raise E2EShardFileInvalid("schema_changed") + try: + stored = tuple( + int(header.get(name)) + for name in ("secret_version", "wrapping_epoch", "bundle_version", "data_key_epoch") + ) + except (TypeError, ValueError): + raise E2EShardFileInvalid("format") from None + if stored != tuple(identity): + raise E2EShardFileInvalid("key_rotated") + if not hmac.compare_digest(str(header.get("key_id") or ""), key_check): + raise E2EShardFileInvalid("wrong_key") + prefix = blob[:fixed + header_length] + nonce = blob[fixed + header_length:fixed + header_length + 12] + try: + return aes_gcm_decrypt(key, nonce, prefix, blob[fixed + header_length + 12:]) + except Exception: + raise E2EShardFileInvalid("tampered") from None + + +def e2e_open_memory_db(data=None): + """An in-memory shard database; nothing it holds is ever written to a file. + + journal_mode=MEMORY and temp_store=MEMORY keep rollback journals, sorts + and temporary b-trees in RAM. No database is ever attached. + """ + conn = sqlite3.connect(":memory:", check_same_thread=False) + try: + conn.row_factory = sqlite3.Row + if data is not None: + conn.deserialize(data) + conn.execute("PRAGMA temp_store = MEMORY") + conn.execute("PRAGMA journal_mode = MEMORY").fetchall() + if data is not None: + check = conn.execute("PRAGMA quick_check").fetchone() + version = conn.execute("PRAGMA user_version").fetchone()[0] + if not check or check[0] != "ok" or version != E2E_SHARD_USER_VERSION: + raise E2EShardFileInvalid("corrupt") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS e2e_nodes ( + node_id TEXT PRIMARY KEY, + revision TEXT, + state TEXT NOT NULL, + fts_rowid INTEGER + ) + """ + ) + conn.execute(LOCAL_INDEX_FTS_DDL.format(table="local_index_fts")) + conn.execute(f"PRAGMA user_version = {E2E_SHARD_USER_VERSION}") + conn.commit() + return conn + except E2EShardFileInvalid: + conn.close() + raise + except sqlite3.Error: + conn.close() + if data is None: + raise + raise E2EShardFileInvalid("corrupt") from None + except BaseException: + conn.close() + raise + + +def e2e_atomic_write(path, blob): + """tmp (0600, O_EXCL) + fsync + rename + directory fsync. Ciphertext only.""" + path.parent.mkdir(parents=True, exist_ok=True) + chmod_best_effort(path.parent, 0o700) + temporary = path.with_name(f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0) + try: + fd = os.open(str(temporary), flags, 0o600) + try: + view = memoryview(blob) + while view: + written = os.write(fd, view) + view = view[written:] + os.fsync(fd) + finally: + os.close(fd) + os.replace(temporary, path) + chmod_best_effort(path, 0o600) + try: + directory = os.open(str(path.parent), os.O_RDONLY) + except OSError: + directory = None + if directory is not None: + try: + os.fsync(directory) + except OSError: + pass + finally: + os.close(directory) + finally: + try: + temporary.unlink() + except FileNotFoundError: + pass + + +def e2e_revision_stale(shard_revision, index_revision): + """True when the shard row is older than the main index row.""" + if not index_revision: + return False + if not shard_revision: + return True + if same_node_revision(shard_revision, index_revision): + return False + left, right = parse_iso_datetime(shard_revision), parse_iso_datetime(index_revision) + if left is not None and right is not None: + return left < right + return True + + +def node_is_e2e_encrypted(node): + return isinstance(node, dict) and node_content_withheld_from_index(node) + + +def e2e_shard_entry(node, project_id): + """Decrypt one node into a shard entry, or None when the secret is unavailable. + + States: indexed (text searchable), no_e2e_content (nothing E2E-decryptable, + e.g. a backend at-rest envelope or a node that is no longer E2E) and + undecryptable (the envelope failed authentication with the secret). + """ + entry = {"node_id": node["id"], "revision": node_revision_token(node), "state": "no_e2e_content", "values": None} + if not node_is_e2e_encrypted(node): + return entry + content_envelope = node.get("encrypted_content_envelope") + metadata_envelope = node.get("encrypted_metadata_envelope") + if (is_backend_at_rest_envelope(content_envelope) or is_backend_at_rest_envelope(metadata_envelope) + or (not content_envelope and not metadata_envelope)): + return entry + try: + decrypted = decrypt_node_for_return(node, project_id=project_id, secret_resolver=e2e_memory_secret_for_read) + except (E2EShardLocked, E2EShardSecretUnavailable): + return None + except Exception: + entry["state"] = "undecryptable" + return entry + metadata = decrypted.get("metadata") if metadata_envelope else {} + content = decrypted.get("content_text") if content_envelope else "" + _title, values = local_index_fts_values( + node, metadata=metadata if isinstance(metadata, dict) else {}, content_text=str(content or "") + ) + entry["state"] = "indexed" + entry["values"] = values + return entry + + +class E2EShard: + """Per-process in-memory shard for one project; all access under .lock.""" + + def __init__(self, project_id): + self.project_id = project_id + self.path = e2e_shard_path(project_id) + self.lock = threading.RLock() + self.conn = None + self.identity = None + self.key = None + self.key_check = None + self.generation = 0 + self.saved_generation = 0 + self.last_save = 0.0 + self.last_saved_at = None + self.save_timer = None + self.loading = None + self.events = [] + self.last_discard = None + self.persist_skipped = False + self.last_error = None + self.secret_retry_at = 0.0 + self.secret_error = None + + # -- lifecycle ----------------------------------------------------- + def take_events(self): + events, self.events = self.events, [] + return events + + def close_memory(self): + if self.conn is not None: + try: + self.conn.close() + except sqlite3.Error: + pass + self.conn = None + self.key = None + self.key_check = None + self.identity = None + if self.save_timer is not None: + self.save_timer.cancel() + self.save_timer = None + + def discard(self, reason, *, report=True): + """Fail closed: drop the in-memory shard and delete the file.""" + self.close_memory() + try: + self.path.unlink() + except FileNotFoundError: + pass + except OSError as exc: + self.last_error = safe_exception_text(exc) + self.generation += 1 + self.saved_generation = self.generation + if report: + self.last_discard = {"reason": reason, "at": iso_now()} + self.events.append(e2e_warning("e2e_index_discarded", reason=reason)) + + def attach(self, conn, identity, key, key_check): + self.conn = conn + self.identity = identity + self.key = key + self.key_check = key_check + + def mark_dirty(self): + self.generation += 1 + + @property + def dirty(self): + return self.conn is not None and self.generation != self.saved_generation + + # -- rows ---------------------------------------------------------- + def remove(self, node_id): + row = self.conn.execute("SELECT fts_rowid FROM e2e_nodes WHERE node_id = ?", (node_id,)).fetchone() + if row is None: + return False + if row["fts_rowid"] is not None: + self.conn.execute("DELETE FROM local_index_fts WHERE rowid = ?", (row["fts_rowid"],)) + self.conn.execute("DELETE FROM e2e_nodes WHERE node_id = ?", (node_id,)) + return True + + def upsert(self, entry): + self.remove(entry["node_id"]) + fts_rowid = None + if entry["state"] == "indexed" and entry.get("values"): + fts_rowid = self.conn.execute( + LOCAL_INDEX_FTS_INSERT.format(table="local_index_fts"), entry["values"] + ).lastrowid + self.conn.execute( + "INSERT INTO e2e_nodes (node_id, revision, state, fts_rowid) VALUES (?, ?, ?, ?)", + (entry["node_id"], entry.get("revision"), entry["state"], fts_rowid), + ) + + def clear(self): + self.conn.execute("DELETE FROM local_index_fts") + self.conn.execute("DELETE FROM e2e_nodes") + + def rows(self): + return { + row["node_id"]: (row["revision"], row["state"]) + for row in self.conn.execute("SELECT node_id, revision, state FROM e2e_nodes") + } + + def reconcile_local(self, expected): + """Drop rows the main index no longer withholds; return missing/stale ids.""" + rows = self.rows() + removed = 0 + for node_id in set(rows) - set(expected): + self.remove(node_id) + removed += 1 + if removed: + self.conn.commit() + self.mark_dirty() + missing = [ + node_id for node_id, revision in expected.items() + if node_id not in rows or e2e_revision_stale(rows[node_id][0], revision) + ] + undecryptable = sum( + 1 for node_id, (revision, state) in rows.items() + if state == "undecryptable" and node_id in expected + ) + return missing, {"removed": removed, "undecryptable": undecryptable, "entries": len(rows) - removed} + + +def e2e_shard_get(project_id): + key = (str(AUTH_DIR), str(project_id)) + with E2E_SHARDS_LOCK: + shard = E2E_SHARDS.get(key) + if shard is None or shard.path != e2e_shard_path(project_id): + shard = E2E_SHARDS[key] = E2EShard(str(project_id)) + return shard + + +def e2e_shard_loaded(project_id): + with E2E_SHARDS_LOCK: + shard = E2E_SHARDS.get((str(AUTH_DIR), str(project_id))) + return shard if shard is not None and shard.conn is not None else None + + +def e2e_shard_active_secret(shard, allow_fetch): + """(secret, None) or (None, warning). Never prompts.""" + cached = e2e_cached_memory_secret(shard.project_id) + if cached: + return cached, None + if not e2e_noninteractive_passphrase_available(): + return None, e2e_warning("e2e_index_locked") + if not allow_fetch or time.monotonic() < shard.secret_retry_at: + return None, e2e_warning("e2e_index_unavailable", cause=shard.secret_error) + try: + return active_memory_secret(shard.project_id), None + except Exception as exc: + shard.secret_retry_at = time.monotonic() + E2E_SHARD_SECRET_RETRY_SECONDS + shard.secret_error = safe_exception_text(exc) + return None, e2e_warning("e2e_index_unavailable", cause=shard.secret_error) + + +def e2e_shard_read_file(shard, identity, key, key_check): + """Decrypt and deserialize the shard file; None when there is no file.""" + try: + blob = shard.path.read_bytes() + except FileNotFoundError: + return None + return e2e_open_memory_db(e2e_shard_decode(blob, shard.project_id, identity, key, key_check)) + + +def e2e_shard_background_load(shard, identity, key, key_check): + def run(): + try: + try: + conn = e2e_shard_read_file(shard, identity, key, key_check) + failure = None + except E2EShardFileInvalid as exc: + conn, failure = None, exc.reason + except Exception as exc: + conn, failure = None, None + shard.last_error = safe_exception_text(exc) + with shard.lock: + if shard.conn is None: + if failure: + shard.discard(failure) + if conn is None: + conn = e2e_open_memory_db() + shard.attach(conn, identity, key, key_check) + elif conn is not None: + conn.close() + finally: + with shard.lock: + shard.loading = None + + thread = threading.Thread(target=run, name="napseer-e2e-index-load", daemon=True) + shard.loading = thread + thread.start() + return thread + + +def e2e_shard_open(project_id, *, allow_fetch=True, wait=False): + """Return (shard with a ready connection, warnings) or (None, warnings). + + A missing, stale-keyed or invalid file yields an empty shard (after a + discard warning) that reconciliation refills. With the pure-Python + backend a large file loads on a background thread unless wait=True. + """ + supported, reason = e2e_shard_supported() + if not supported: + return None, [e2e_warning("e2e_index_unsupported", reason=reason)] + shard = e2e_shard_get(project_id) + secret, problem = e2e_shard_active_secret(shard, allow_fetch) + if secret is None: + with shard.lock: + if shard.conn is not None: + # Locked since load: stop serving plaintext; keep the file. + e2e_shard_flush(shard, force=True) + shard.close_memory() + return None, [problem] + identity, key, key_check = e2e_shard_key_material(project_id, secret) + with shard.lock: + warnings = [] + if shard.conn is not None and shard.identity != identity: + shard.discard("key_rotated") + if shard.conn is not None: + return shard, warnings + shard.take_events() + loader = shard.loading + if loader is not None: + if wait: + loader.join() + return e2e_shard_open(project_id, allow_fetch=allow_fetch, wait=wait) + return None, [e2e_warning("e2e_index_slow_crypto", state="loading", crypto_backend=CRYPTO_BACKEND)] + with shard.lock: + if shard.conn is not None or shard.loading is not None: + pass + else: + try: + size = shard.path.stat().st_size + except FileNotFoundError: + size = None + if size is None: + shard.attach(e2e_open_memory_db(), identity, key, key_check) + elif (CRYPTO_BACKEND == CRYPTO_BACKEND_PURE and size > E2E_SHARD_PURE_SYNC_LOAD_MAX_BYTES + and not wait): + e2e_shard_background_load(shard, identity, key, key_check) + return None, [e2e_warning("e2e_index_slow_crypto", state="loading", crypto_backend=CRYPTO_BACKEND)] + else: + try: + conn = e2e_shard_read_file(shard, identity, key, key_check) + except E2EShardFileInvalid as exc: + shard.discard(exc.reason) + conn = None + shard.attach(conn or e2e_open_memory_db(), identity, key, key_check) + if shard.conn is None: + return None, [e2e_warning("e2e_index_slow_crypto", state="loading", crypto_backend=CRYPTO_BACKEND)] + return shard, warnings + shard.take_events() + + +def e2e_shard_flush(shard, *, force=False): + """Persist the shard if dirty. Debounced unless force; never raises.""" + try: + with shard.lock: + if not shard.dirty: + return "clean" + if not force and time.monotonic() - shard.last_save < E2E_SHARD_SAVE_INTERVAL_SECONDS: + if shard.save_timer is None: + timer = threading.Timer(E2E_SHARD_SAVE_INTERVAL_SECONDS, e2e_shard_timer_flush, (shard,)) + timer.daemon = True + shard.save_timer = timer + timer.start() + return "deferred" + shard.conn.commit() + plaintext = shard.conn.serialize() + generation = shard.generation + identity, key, key_check = shard.identity, shard.key, shard.key_check + shard.last_save = time.monotonic() + if CRYPTO_BACKEND == CRYPTO_BACKEND_PURE and len(plaintext) > E2E_SHARD_PURE_MAX_PERSIST_BYTES: + shard.persist_skipped = True + return "skipped_slow_crypto" + blob = e2e_shard_encode(shard.project_id, identity, key, key_check, plaintext) + del plaintext + with shard.lock: + # Never resurrect a discarded/re-keyed shard or overwrite a newer save. + if shard.saved_generation >= generation or shard.identity != identity: + return "superseded" + e2e_atomic_write(shard.path, blob) + shard.saved_generation = generation + shard.persist_skipped = False + shard.last_saved_at = iso_now() + return "saved" + except Exception as exc: + shard.last_error = safe_exception_text(exc) + return "failed" + + +def e2e_shard_timer_flush(shard): + with shard.lock: + shard.save_timer = None + e2e_shard_flush(shard, force=True) + + +def e2e_shards_close(*, flush=True): + """Flush and drop every in-memory shard (clean shutdown or vault lock).""" + with E2E_SHARDS_LOCK: + shards = list(E2E_SHARDS.values()) + E2E_SHARDS.clear() + for shard in shards: + if flush: + e2e_shard_flush(shard, force=True) + with shard.lock: + shard.close_memory() + + +def e2e_shards_remove_nodes(node_ids): + """Apply removals to every loaded shard (unloaded shards reconcile on load).""" + with E2E_SHARDS_LOCK: + shards = [shard for shard in E2E_SHARDS.values() if shard.conn is not None] + for shard in shards: + try: + with shard.lock: + if shard.conn is None: + continue + changed = [node_id for node_id in node_ids if shard.remove(node_id)] + if changed: + shard.conn.commit() + shard.mark_dirty() + if changed: + e2e_shard_flush(shard) + except Exception as exc: + shard.last_error = safe_exception_text(exc) + + +def e2e_expected_rows(conn, project_id): + """Main-index E2E rows (content withheld) and their revisions.""" + return { + row["node_id"]: row["revision"] + for row in conn.execute( + "SELECT node_id, revision FROM local_index_nodes WHERE project_id = ? AND content_withheld = 1", + (project_id,), + ) + } + + +def e2e_shard_drop_if_unneeded(project_id): + """No E2E rows remain: remove the shard and its file without a warning.""" + shard = e2e_shard_get(project_id) + with shard.lock: + if shard.conn is not None or shard.path.exists(): + shard.discard("not_needed", report=False) + + +def e2e_shard_after_index(project_id, nodes=(), removed_ids=(), *, replace_all=False, deadline=None, + wait=False, force_flush=False): + """Mirror committed main-index changes into the E2E shard. Never raises. + + Decryption runs outside the index writer lock. Entries that cannot be + decrypted now (locked, secret unavailable, deadline) stay missing and are + rebuilt by reconciliation. + """ + try: + project_id = str(project_id or "") + nodes = [node for node in (nodes or ()) if isinstance(node, dict) and node.get("id")] + e2e_nodes = [node for node in nodes if node_is_e2e_encrypted(node)] + other_ids = [node["id"] for node in nodes if not node_is_e2e_encrypted(node)] + removed_ids = list(removed_ids or ()) + if not project_id or not e2e_shard_supported()[0]: + return {"status": "not_applicable"} + if not e2e_nodes: + if replace_all: + e2e_shard_drop_if_unneeded(project_id) + return {"status": "not_applicable"} + if e2e_shard_loaded(project_id) is not None: + e2e_shards_remove_nodes(removed_ids + other_ids) + return {"status": "not_applicable"} + shard, warnings = e2e_shard_open(project_id, allow_fetch=True, wait=wait) + if shard is None: + return {"status": "skipped", "pending": len(e2e_nodes), "warnings": warnings} + entries = [] + pending = 0 + for node in e2e_nodes: + if deadline is not None and time.monotonic() >= deadline: + pending += 1 + continue + entry = e2e_shard_entry(node, project_id) + if entry is None: + pending += 1 + continue + entries.append(entry) + with shard.lock: + if shard.conn is None: + return {"status": "skipped", "pending": len(e2e_nodes), "warnings": warnings} + if replace_all: + shard.clear() + for node_id in removed_ids + other_ids: + shard.remove(node_id) + for entry in entries: + shard.upsert(entry) + shard.conn.commit() + shard.mark_dirty() + persisted = e2e_shard_flush(shard, force=force_flush) + return { + "status": "applied" if not pending else "partial", + "indexed": sum(1 for entry in entries if entry["state"] == "indexed"), + "pending": pending, + "persist": persisted, + "warnings": warnings, + } + except Exception as exc: + return {"status": "failed", "error": safe_exception_text(exc)} + + +def e2e_fetch_nodes(project_id, node_ids, deadline): + """Fetch raw nodes for a rebuild. Returns (nodes, absent_ids, complete).""" + wanted = set(node_ids) + found = {} + absent = set() + + def remaining(): + return deadline - time.monotonic() + + if len(wanted) <= E2E_SHARD_GET_THRESHOLD: + for node_id in node_ids: + if remaining() <= 0: + return list(found.values()), absent, False + node = fetch_node_for_index(project_id, node_id, timeout_seconds=min(HTTP_TIMEOUT_SECONDS, max(remaining(), 0.5))) + if node is None: + absent.add(node_id) + else: + found[node_id] = node + return list(found.values()), absent, True + cursor = None + seen_cursors = set() + pages = 0 + while True: + if remaining() <= 0 or pages >= E2E_SHARD_LIST_MAX_PAGES: + return list(found.values()), absent, False + query = {"limit": 200, "view": "full", "include_archived": True} + if cursor: + query["cursor"] = cursor + page = request_json( + "GET", f"/v1/projects/{project_id}/nodes?{rest_query_string(query)}", + timeout_seconds=min(HTTP_TIMEOUT_SECONDS, max(remaining(), 0.5)), + ) + pages += 1 + for node in (page.get("items") or []) if isinstance(page, dict) else []: + if isinstance(node, dict) and node.get("id") in wanted and node.get("project_id") == project_id: + found[node["id"]] = node + cursor = page.get("next_cursor") if isinstance(page, dict) else None + if len(found) == len(wanted) or not cursor or cursor in seen_cursors: + break + seen_cursors.add(cursor) + absent = wanted - set(found) + return list(found.values()), absent, True + + +def e2e_shard_reconcile(project_id, *, budget_seconds, wait=False, force_flush=True): + """Bring the shard in line with the main index, fetching missing E2E nodes. + + Bounded by budget_seconds. Never raises; returns status and warnings. + """ + result = {"status": "not_applicable", "warnings": []} + try: + if not INDEX_PATH.exists(): + return result + with index_connect() as conn: + expected = e2e_expected_rows(conn, project_id) + if not expected: + e2e_shard_drop_if_unneeded(project_id) + return result + shard, warnings = e2e_shard_open(project_id, allow_fetch=True, wait=wait) + result["warnings"] = warnings + if shard is None: + result["status"] = "unavailable" + return result + with shard.lock: + if shard.conn is None: + result["status"] = "unavailable" + return result + missing, counts = shard.reconcile_local(expected) + result.update({"missing_before": len(missing), **counts}) + rebuilt = 0 + complete = True + if missing: + deadline = time.monotonic() + max(0.1, float(budget_seconds)) + nodes, absent, complete = e2e_fetch_nodes(project_id, missing, deadline) + entries = [] + for node in nodes: + if time.monotonic() >= deadline: + complete = False + break + entry = e2e_shard_entry(node, project_id) + if entry is None: + complete = False + continue + entries.append(entry) + for node_id in absent: + # Gone on the server; the feed removes the main row. Do not refetch meanwhile. + entries.append({"node_id": node_id, "revision": expected.get(node_id), "state": "no_e2e_content", "values": None}) + with shard.lock: + if shard.conn is not None: + for entry in entries: + shard.upsert(entry) + shard.conn.commit() + shard.mark_dirty() + rebuilt = sum(1 for entry in entries if entry["state"] == "indexed") + missing_after, counts = shard.reconcile_local(expected) + else: + missing_after = missing + result.update({"rebuilt": rebuilt, "remaining": len(missing_after), **counts}) + complete = complete and not missing_after + else: + result.update({"rebuilt": 0, "remaining": 0}) + result["persist"] = e2e_shard_flush(shard, force=force_flush) + result["status"] = "complete" if complete else "partial" + return result + except SafeToolError as exc: + if exc.code == "request_cancelled": + raise + result.update({"status": "failed", "error": safe_exception_text(exc)}) + return result + except Exception as exc: + result.update({"status": "failed", "error": safe_exception_text(exc)}) + return result + + +def e2e_shard_auto_reconcile(project_id): + """Throttled, bounded rebuild before a local search (search_memory).""" + key = (str(AUTH_DIR), str(project_id)) + now = time.monotonic() + with FEED_AUTO_SYNC_LOCK: + last = E2E_RECONCILE_STATE.get(key) + if last is not None and now - last < E2E_SHARD_RECONCILE_INTERVAL_SECONDS: + return None + E2E_RECONCILE_STATE[key] = now + return e2e_shard_reconcile( + project_id, budget_seconds=E2E_SHARD_RECONCILE_BUDGET_SECONDS, wait=False, force_flush=False + ) + + +def e2e_shard_search(project_id, conn, match_query, analysis, filter_where, filter_parameters, candidate_limit): + """Search the E2E shard; candidates are filtered through the main index rows. + + Returns {"candidates", "warnings", "diagnostics"}. Missing/stale shard + rows never produce hits; they are reported as e2e_index_incomplete. + """ + result = {"candidates": [], "warnings": [], "diagnostics": {"status": "not_applicable"}} + expected = e2e_expected_rows(conn, project_id) + if not expected: + return result + shard, warnings = e2e_shard_open(project_id, allow_fetch=True, wait=False) + result["warnings"].extend(warnings) + if shard is None: + result["diagnostics"] = {"status": "unavailable", "e2e_nodes": len(expected)} + return result + with shard.lock: + if shard.conn is None: + result["diagnostics"] = {"status": "unavailable", "e2e_nodes": len(expected)} + return result + missing, counts = shard.reconcile_local(expected) + try: + hits = shard.conn.execute( + f""" + SELECT local_index_fts.node_id AS node_id, local_index_fts.rowid AS fts_rowid, + s.revision AS revision, + bm25(local_index_fts, {LOCAL_INDEX_FTS_BM25}) AS rank + FROM local_index_fts + JOIN e2e_nodes s ON s.fts_rowid = local_index_fts.rowid + WHERE local_index_fts MATCH ? AND s.state = 'indexed' + ORDER BY rank ASC + LIMIT ? + """, + [match_query, candidate_limit], + ).fetchall() + except sqlite3.Error as exc: + hits = [] + result["diagnostics"]["error"] = safe_exception_payload(exc) + by_id = {row["node_id"]: row for row in hits} + candidates = [] + if by_id: + ids = list(by_id) + rows = conn.execute( + f""" + SELECT n.node_id, n.project_id, n.full_path, n.name, n.folder_path, + n.node_type, n.tags_json, n.updated_at, n.indexed_at, n.revision, + n.title, n.status, n.archived,{LOCAL_INDEX_EDGE_COUNT_COLUMNS}, + n.aliases_json + FROM local_index_nodes n + WHERE n.project_id = ? AND n.content_withheld = 1 + AND n.node_id IN ({','.join('?' for _ in ids)}) + {''.join(' AND ' + clause for clause in filter_where)} + """, + [project_id, *ids, *filter_parameters], + ).fetchall() + for row in rows: + hit = by_id[row["node_id"]] + if e2e_revision_stale(hit["revision"], row["revision"]): + continue + candidates.append({ + "id": row["node_id"], + "project_id": row["project_id"], + "full_path": row["full_path"], + "name": row["name"], + "title": row["title"], + "folder_path": row["folder_path"], + "type": row["node_type"], + "tags": json.loads(row["tags_json"] or "[]"), + "aliases": json.loads(row["aliases_json"] or "[]"), + "status": row["status"], + "archived": bool(row["archived"]), + "updated_at": row["updated_at"], + "indexed_at": row["indexed_at"], + "links_count": row["links_count"], + "backlinks_count": row["backlinks_count"], + "rank": hit["rank"], + "matched_strategy": "local_e2e_fts", + "search_source": "local_e2e", + "_fts_rowid": hit["fts_rowid"], + "_content_withheld": False, + "_e2e": True, + }) + with shard.lock: + if shard.conn is not None: + attach_local_term_coverage(shard.conn, candidates, analysis) + if missing: + result["warnings"].append(e2e_warning("e2e_index_incomplete", missing=len(missing), next_action="nap_index_sync")) + if counts.get("undecryptable"): + result["warnings"].append(e2e_warning("e2e_index_undecryptable", nodes=counts["undecryptable"])) + if shard.persist_skipped: + result["warnings"].append(e2e_warning("e2e_index_slow_crypto", state="not_persisted", crypto_backend=CRYPTO_BACKEND)) + result["candidates"] = candidates + result["diagnostics"] = { + "status": "used", + "e2e_nodes": len(expected), + "missing": len(missing), + "matched": len(hits), + "returned_candidates": len(candidates), + "crypto_backend": CRYPTO_BACKEND, + } + return result + + +def e2e_search_failure(exc): + cause = safe_exception_text(exc) + return { + "candidates": [], + "warnings": [e2e_warning("e2e_index_unavailable", cause=cause)], + "diagnostics": {"status": "failed", "error": cause}, + } + + +def e2e_shard_attach_snippets(project_id, items, match_query): + items = [item for item in items if item.get("_e2e")] + if not items: + return + shard = e2e_shard_loaded(project_id) + if shard is None: + return + with shard.lock: + if shard.conn is not None: + attach_local_snippets(shard.conn, items, match_query) + + +def e2e_shard_status(project_id): + supported, reason = e2e_shard_supported() + shard = e2e_shard_loaded(project_id) + path = e2e_shard_path(project_id) + status = { + "supported": supported, + "crypto_backend": CRYPTO_BACKEND, + "file_exists": path.exists(), + "loaded": shard is not None, + } + if reason: + status["unsupported_reason"] = reason + if shard is not None: + with shard.lock: + if shard.conn is not None: + status["entries"] = shard.conn.execute("SELECT COUNT(*) FROM e2e_nodes").fetchone()[0] + status.update({ + "dirty": shard.dirty, + "last_saved_at": shard.last_saved_at, + "last_discard": shard.last_discard, + "persist_skipped": shard.persist_skipped, + "last_error": shard.last_error, + }) + else: + status["locked"] = not (e2e_cached_memory_secret(project_id) or e2e_noninteractive_passphrase_available()) + return status + + +atexit.register(e2e_shards_close) + + def public_json(path): request = urllib.request.Request( f"{BASE_URL}{path}", @@ -14155,8 +15389,15 @@ def search_memory(args): "local_index": local_index_diagnostics(project_id), "server_queries": [], } + e2e_warnings = [] if mode in {"hybrid", "local"} and INDEX_PATH.exists(): + # Bounded, throttled rebuild of missing E2E shard rows (never raises + # except on cancellation); its warnings join the local answer. + e2e_refresh = e2e_shard_auto_reconcile(project_id) + if e2e_refresh and e2e_refresh.get("status") != "not_applicable": + diagnostics["e2e_refresh"] = {k: v for k, v in e2e_refresh.items() if k != "warnings"} + e2e_warnings.extend(e2e_refresh.get("warnings") or []) try: local_result = search_local_index({ **args, @@ -14167,12 +15408,15 @@ def search_memory(args): diagnostics["local_index"] = local_result.get("index", diagnostics["local_index"]) diagnostics["local_strategies"] = local_result.get("search_strategies", []) local = local_result.get("items", []) + e2e_warnings.extend(local_result.get("warnings") or []) for item in local: node = dict(item) - node["search_source"] = "local" + node["search_source"] = item.get("search_source") or "local" collected.append(node) - if local: + if any(item.get("search_source") != "local_e2e" for item in local): sources.append("local") + if any(item.get("search_source") == "local_e2e" for item in local): + sources.append("local_e2e") except Exception as exc: if mode == "local": raise @@ -14260,6 +15504,14 @@ def search_memory(args): response["freshness"] = local_read_freshness(feed_facts) if staleness is None and local_used and not server_needed: staleness = feed_staleness or local_answer_staleness(project_id, diagnostics=diagnostics.get("local_index")) + if e2e_warnings: + # Always visible (not only in verbose mode): E2E bodies were not, or + # not fully, searched and the caller must know why. + seen_codes = {item.get("code") for item in response.get("warnings") or [] if isinstance(item, dict)} + for warning in e2e_warnings: + if warning.get("code") not in seen_codes: + seen_codes.add(warning.get("code")) + response["warnings"] = [*(response.get("warnings") or []), warning] return apply_local_staleness(response, staleness) @@ -19615,6 +20867,7 @@ def doctor(args): "anchor_graph": local_anchor_integrity(), "contract": contract_profile({"include_tools": False}), "state_dir": state_dir_status_payload(), + "crypto": crypto_backend_status(), } profile_warnings = report["contract"].get("tool_profiles", {}).get("warnings") or [] if profile_warnings: diff --git a/tests/test_crypto_backend.py b/tests/test_crypto_backend.py new file mode 100644 index 0000000..727d1b5 --- /dev/null +++ b/tests/test_crypto_backend.py @@ -0,0 +1,152 @@ +"""AES-256-GCM backend selection: known answers, interop and tamper detection. + +Both the optional native backend (`cryptography`) and the stdlib-only +pure-Python backend run the same vectors through the public +aes_gcm_encrypt/aes_gcm_decrypt API used by node and bundle encryption. +""" + +import importlib.util +import pathlib + +import pytest + +SCRIPT = pathlib.Path(__file__).resolve().parents[1] / "resources/scripts/napseer_mcp_server.py" + +h = bytes.fromhex +GCM_KEY = h("feffe9928665731c6d6a8f9467308308feffe9928665731c6d6a8f9467308308") +GCM_IV = h("cafebabefacedbaddecaf888") +GCM_P = h( + "d9313225f88406e5a55909c5aff5269a86a7a9531534f7da2e4c303d8a318a72" + "1c3c0c95956809532fcf0e2449a6b525b16aedf5aa0de657ba637b391aafd255" +) +GCM_C = h( + "522dc1f099567d07f47f37a32a84427d643a8cdcbfe5c0c97598a2bd2555d1aa" + "8cb08e48590dbb3da7b08b1056828838c5f61e6393ba7a0abcc9f662898015ad" +) +# The GCM specification (McGrew & Viega) test cases 13-16, AES-256. +KNOWN_ANSWERS = [ + ("tc13", bytes(32), bytes(12), b"", b"", b"", h("530f8afbc74536b9a963b4f1c4cb738b")), + ("tc14", bytes(32), bytes(12), b"", bytes(16), h("cea7403d4d606b6e074ec5d3baf39d18"), + h("d0d1c8a799996bf0265b98b5d48ab919")), + ("tc15", GCM_KEY, GCM_IV, b"", GCM_P, GCM_C, h("b094dac5d93471bdec1a502270e3cc6c")), + ("tc16", GCM_KEY, GCM_IV, h("feedfacedeadbeeffeedfacedeadbeefabaddad2"), GCM_P[:60], GCM_C[:60], + h("76fc6ece0f4e1768cddf8853bb2d551b")), +] + + +@pytest.fixture(scope="module") +def mod(): + spec = importlib.util.spec_from_file_location("crypto_backend_test_worker", SCRIPT) + module = importlib.util.module_from_spec(spec) + import sys + sys.path.insert(0, str(SCRIPT.parent)) + spec.loader.exec_module(module) + yield module + module.select_crypto_backend() + + +def backends(mod): + names = [mod.CRYPTO_BACKEND_PURE] + if mod.native_crypto_available(): + names.append(mod.CRYPTO_BACKEND_NATIVE) + return names + + +def use(mod, name): + assert mod.select_crypto_backend(name) == name + return name + + +def test_native_backend_is_selected_when_cryptography_is_importable(mod, monkeypatch): + pytest.importorskip("cryptography") + monkeypatch.delenv("NAPSEER_CRYPTO_BACKEND", raising=False) + assert mod.select_crypto_backend() == mod.CRYPTO_BACKEND_NATIVE + monkeypatch.setenv("NAPSEER_CRYPTO_BACKEND", "pure_python") + assert mod.select_crypto_backend() == mod.CRYPTO_BACKEND_PURE + assert mod.crypto_backend_status()["crypto_backend"] == "pure_python" + + +def test_pure_backend_is_used_when_native_is_unavailable(mod, monkeypatch): + monkeypatch.setattr(mod, "_NativeAESGCM", None) + monkeypatch.delenv("NAPSEER_CRYPTO_BACKEND", raising=False) + try: + assert mod.select_crypto_backend() == mod.CRYPTO_BACKEND_PURE + status = mod.crypto_backend_status() + assert status["native_crypto_available"] is False + assert "cryptography" in status["crypto_backend_note"] + finally: + monkeypatch.undo() + mod.select_crypto_backend() + + +@pytest.mark.parametrize("case", KNOWN_ANSWERS, ids=[case[0] for case in KNOWN_ANSWERS]) +def test_known_answer_vectors_through_every_backend(mod, case): + _name, key, iv, aad, plaintext, ciphertext, tag = case + for backend in backends(mod): + use(mod, backend) + assert mod.aes_gcm_encrypt(key, iv, aad, plaintext) == ciphertext + tag, backend + assert mod.aes_gcm_decrypt(key, iv, aad, ciphertext + tag) == plaintext, backend + mod.select_crypto_backend() + + +def test_cross_backend_interop(mod): + pytest.importorskip("cryptography") + key = bytes(range(32)) + nonce = bytes(range(12)) + aad = b"napseer interop aad" + for size in (0, 1, 15, 16, 17, 1000): + plaintext = bytes((index * 7) & 0xFF for index in range(size)) + use(mod, mod.CRYPTO_BACKEND_PURE) + pure = mod.aes_gcm_encrypt(key, nonce, aad, plaintext) + use(mod, mod.CRYPTO_BACKEND_NATIVE) + native = mod.aes_gcm_encrypt(key, nonce, aad, plaintext) + assert pure == native + assert mod.aes_gcm_decrypt(key, nonce, aad, pure) == plaintext + use(mod, mod.CRYPTO_BACKEND_PURE) + assert mod.aes_gcm_decrypt(key, nonce, aad, native) == plaintext + mod.select_crypto_backend() + + +def test_tamper_detection_raises_the_same_error_in_every_backend(mod): + key = bytes(range(32)) + nonce = bytes(12) + for backend in backends(mod): + use(mod, backend) + packed = mod.aes_gcm_encrypt(key, nonce, b"aad", b"synthetic payload") + tampered_body = bytes([packed[0] ^ 1]) + packed[1:] + tampered_tag = packed[:-1] + bytes([packed[-1] ^ 0x80]) + wrong_key = bytes([key[0] ^ 1]) + key[1:] + for label, attempt in ( + ("body", lambda: mod.aes_gcm_decrypt(key, nonce, b"aad", tampered_body)), + ("tag", lambda: mod.aes_gcm_decrypt(key, nonce, b"aad", tampered_tag)), + ("aad", lambda: mod.aes_gcm_decrypt(key, nonce, b"other", packed)), + ("key", lambda: mod.aes_gcm_decrypt(wrong_key, nonce, b"aad", packed)), + ): + with pytest.raises(RuntimeError, match="^AES-GCM authentication failed$"): + attempt() + + with pytest.raises(RuntimeError, match="missing tag"): + mod.aes_gcm_decrypt(key, nonce, b"aad", b"short") + with pytest.raises(RuntimeError, match="nonce must be 12 bytes"): + mod.aes_gcm_encrypt(key, bytes(16), b"", b"x") + with pytest.raises(RuntimeError, match="key must be 32 bytes"): + mod.aes_gcm_encrypt(bytes(16), nonce, b"", b"x") + mod.select_crypto_backend() + + +def test_hkdf_sha256_matches_rfc5869_vectors(mod): + # RFC 5869 Appendix A, test cases 1 and 3. + okm = mod.hkdf_sha256(b"\x0b" * 22, bytes(range(13)), bytes(range(0xF0, 0xFA)), 42) + assert okm.hex() == ( + "3cb25f25faacd57a90434f64d0362f2a2d2d0a90cf1a5a4c5db02d56ecc4c5bf34007208d5b887185865" + ) + okm = mod.hkdf_sha256(b"\x0b" * 22, b"", b"", 42) + assert okm.hex() == ( + "8da4e775a563c18f715f802a063c5a31b8a11f5c5ee1879ec3454e5f3c738d2d9d201395faa4b61a96c8" + ) + + +def test_diagnostics_report_the_backend_without_key_material(mod): + status = mod.crypto_backend_status() + assert status["crypto_backend"] in {"native", "pure_python"} + assert set(status) <= {"crypto_backend", "native_crypto_available", "crypto_backend_note"} diff --git a/tests/test_mcp_e2e_index_shard.py b/tests/test_mcp_e2e_index_shard.py new file mode 100644 index 0000000..d6ed884 --- /dev/null +++ b/tests/test_mcp_e2e_index_shard.py @@ -0,0 +1,483 @@ +"""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): + stamp = f"2026-09-01T12:{minute:02d}:00Z" + return { + "id": name, "project_id": PROJECT, "full_path": f"/notes/{name}", "folder_path": "/notes", + "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): + secret = secret or memory_secret() + node = plain_node(name, minute) + 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_locked_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_locked"] + response = mod.search_memory({"q": BODY_TERM, "mode": "local", "view": "summary"}) + assert response["items"] == [] + assert "e2e_index_locked" in codes(response) + # Identity fields of the E2E node remain searchable from the main index. + assert hit_ids(local_search(mod, "secretnote")) == ["secretnote"] + # A locked 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) diff --git a/tests/test_mcp_memory_node_encryption.py b/tests/test_mcp_memory_node_encryption.py index 9fddeeb..04c6ff2 100644 --- a/tests/test_mcp_memory_node_encryption.py +++ b/tests/test_mcp_memory_node_encryption.py @@ -472,8 +472,22 @@ def run(): indexed["metadata"] = {"private": "secret metadata"} mod.index_node(indexed) assert mod.search_local_index({"q": "alpha", "limit": 5})["items"] - assert mod.search_local_index({"q": "needle", "limit": 5})["items"] == [] - assert mod.search_local_index({"q": "metadata", "limit": 5})["items"] == [] + # E2E body/metadata text never reaches the on-disk index; with the + # memory secret unlocked it is found only via the encrypted shard. + for term in ("needle", "metadata"): + hits = mod.search_local_index({"q": term, "limit": 5})["items"] + assert hits and all(item.get("search_source") == "local_e2e" for item in hits) + mod.e2e_shards_close(flush=True) + for path in state_dir.iterdir(): + if path.is_file(): + raw = path.read_bytes() + assert b"secret needle" not in raw and b"secret metadata" not in raw, path.name + mod.MEMORY_SECRET_CACHE.clear() + mod.ACTIVE_MEMORY_SECRET_VERSIONS.clear() + mod.PROJECT_VAULT_PASSPHRASE = None + locked = mod.search_local_index({"q": "needle", "limit": 5}) + assert locked["items"] == [] + assert [warning["code"] for warning in locked["warnings"]] == ["e2e_index_locked"] print("ok: local MCP memory node encryption smoke passed")