fix: add namespace validation and fix TOCTOU in elasticsearch memory …

[mkmeral]

Description

This PR adds namespace validation and fixes a TOCTOU (Time-of-Check to Time-of-Use) vulnerability in the Elasticsearch memory tool — the same class of vulnerability that was fixed for MongoDB in PR #321.

The _delete_memory function previously performed deletion using document ID only with no server-side namespace constraint. Combined with the lack of namespace input validation and a client-side-only namespace check in _get_memory, this allowed potential cross-tenant memory access and deletion.

Changes

Namespace validation (_validate_namespace)

• Rejects non-string types (blocks dict-based injection payloads like {"$ne": ""})
• Enforces ^[A-Za-z0-9_-]{1,64}$ — same pattern as the MongoDB fix
• Validated early in the elasticsearch_memory() entry point before any operations

_get_memory — server-side namespace enforcement

• Previously: es_client.get(id=memory_id) then client-side if source.get("namespace") != namespace
• Now: es_client.search() with a bool query requiring both memory_id AND namespace terms
• Namespace is enforced by Elasticsearch server-side, not by Python code after retrieval

_delete_memory — atomic delete with namespace constraint

• Previously: _get_memory() check → es_client.delete(id=memory_id) (TOCTOU race window, no namespace in delete)
• Now: single es_client.delete_by_query() with both memory_id AND namespace in the query
• Eliminates the race condition entirely — verify and delete happen atomically

Related Issues

• Mirrors the fix from PR fix: fix mongodb namespace secruity issue #321 (MongoDB namespace injection)

Type of Change

Bug fix

Testing

• All 31 tests pass (27 existing + 4 new)
• New tests cover: injection prevention with dict payloads, strict character validation, valid namespace acceptance, and atomic delete namespace enforcement
• I ran hatch run test tests/test_elasticsearch_memory.py -v

Also ran integ tests with the script below

"""Integration test against a real local Elasticsearch instance.

Validates:
1. Basic CRUD operations work with namespace
2. Namespace validation rejects bad input
3. _get_memory enforces namespace server-side (no cross-tenant read)
4. _delete_memory enforces namespace atomically (no cross-tenant delete / no TOCTOU)
"""

import sys
import time

from elasticsearch import Elasticsearch

# We import the internal functions directly to test them without needing Bedrock
from src.strands_tools.elasticsearch_memory import (
    ElasticsearchValidationError,
    _delete_memory,
    _ensure_index_exists,
    _get_memory,
    _list_memories,
    _validate_namespace,
)

ES_URL = "http://localhost:9200"
INDEX = f"test_security_{int(time.time())}"

es = Elasticsearch(hosts=[ES_URL])
assert es.ping(), "Cannot connect to Elasticsearch"

print(f"Connected to ES. Using index: {INDEX}")

# --- 1. Namespace validation ---
print("\n=== Test 1: Namespace validation ===")

assert _validate_namespace(None) == "default"
assert _validate_namespace("user_123") == "user_123"
assert _validate_namespace("tenant-abc") == "tenant-abc"

for bad in [{"$ne": ""}, {"$gt": ""}, 123, [], "", "   ", "a" * 65, "user.name", "user@x", "a/b"]:
    try:
        _validate_namespace(bad)
        print(f"  FAIL: should have rejected {bad!r}")
        sys.exit(1)
    except ElasticsearchValidationError:
        pass

print("  PASS: all invalid namespaces rejected")

# --- 2. Setup: create index and insert docs in two namespaces ---
print("\n=== Test 2: Setup index and insert docs ===")

_ensure_index_exists(es, INDEX, es_url=ES_URL)

# Insert doc in tenant_a
es.index(index=INDEX, id="mem_a1", body={
    "memory_id": "mem_a1",
    "content": "Secret data for tenant A",
    "namespace": "tenant_a",
    "timestamp": "2025-01-01T00:00:00Z",
    "metadata": {},
})

# Insert doc in tenant_b
es.index(index=INDEX, id="mem_b1", body={
    "memory_id": "mem_b1",
    "content": "Secret data for tenant B",
    "namespace": "tenant_b",
    "timestamp": "2025-01-01T00:00:00Z",
    "metadata": {},
})

# Refresh so docs are searchable
es.indices.refresh(index=INDEX)
print("  PASS: docs inserted in tenant_a and tenant_b")

# --- 3. _get_memory enforces namespace server-side ---
print("\n=== Test 3: _get_memory namespace enforcement ===")

# tenant_a can get their own doc
result = _get_memory(es, INDEX, "tenant_a", "mem_a1")
assert result["memory_id"] == "mem_a1"
assert result["namespace"] == "tenant_a"
print("  PASS: tenant_a can read mem_a1")

# tenant_a CANNOT get tenant_b's doc
try:
    _get_memory(es, INDEX, "tenant_a", "mem_b1")
    print("  FAIL: tenant_a should NOT be able to read mem_b1")
    sys.exit(1)
except Exception as e:
    assert "not found in namespace tenant_a" in str(e)
    print("  PASS: tenant_a blocked from reading mem_b1")

# --- 4. _list_memories only returns own namespace ---
print("\n=== Test 4: _list_memories namespace isolation ===")

list_a = _list_memories(es, INDEX, "tenant_a", 100)
list_b = _list_memories(es, INDEX, "tenant_b", 100)

assert list_a["total"] == 1
assert list_a["memories"][0]["memory_id"] == "mem_a1"
assert list_b["total"] == 1
assert list_b["memories"][0]["memory_id"] == "mem_b1"
print("  PASS: each tenant only sees their own memories")

# --- 5. _delete_memory enforces namespace atomically (TOCTOU fix) ---
print("\n=== Test 5: _delete_memory namespace enforcement (TOCTOU fix) ===")

# tenant_a tries to delete tenant_b's doc — should fail
try:
    _delete_memory(es, INDEX, "tenant_a", "mem_b1")
    print("  FAIL: tenant_a should NOT be able to delete mem_b1")
    sys.exit(1)
except Exception as e:
    assert "not found in namespace tenant_a" in str(e)
    print("  PASS: tenant_a blocked from deleting mem_b1")

# Verify mem_b1 still exists
es.indices.refresh(index=INDEX)
result = _get_memory(es, INDEX, "tenant_b", "mem_b1")
assert result["memory_id"] == "mem_b1"
print("  PASS: mem_b1 still exists after failed cross-tenant delete")

# tenant_b can delete their own doc
result = _delete_memory(es, INDEX, "tenant_b", "mem_b1")
assert result["result"] == "deleted"
print("  PASS: tenant_b successfully deleted mem_b1")

# Verify it's gone
es.indices.refresh(index=INDEX)
try:
    _get_memory(es, INDEX, "tenant_b", "mem_b1")
    print("  FAIL: mem_b1 should be gone")
    sys.exit(1)
except Exception:
    print("  PASS: mem_b1 confirmed deleted")

# --- Cleanup ---
es.indices.delete(index=INDEX)
print(f"\nCleaned up index {INDEX}")
print("\n✅ ALL INTEGRATION TESTS PASSED")

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature
• My changes generate no new warnings
• Any dependent changes have been merged and published

[mkmeral]

/strands review