security(context): harden Markdown import against TOCTOU symlink races

CHANGELOG.md

@@ -52,6 +52,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Markdown import hardened against TOCTOU symlink races during file reads** (#932, closes #856) by @lakshanmuruganandam, with fixes by @Sameer6305
+  - `AgentMemory._read_markdown_path` read files via `Path.read_text()` after a `Path.is_symlink()` pre-check, leaving a time-of-check/time-of-use window: a path validated as a regular file could be swapped for a symlink before the actual read, causing the importer to follow the link and read an unintended target
+  - Reads now go through a new `_read_markdown_file_content()` helper: the path is opened via low-level `os.open()` with `os.O_NOFOLLOW` on platforms that support it (POSIX), so a symlink substituted after validation fails atomically with `ELOOP` instead of being followed; the resulting file descriptor is then verified with `os.fstat()`/`stat.S_ISREG()` to reject non-regular files (FIFOs, devices) even after a successful open
+  - Directory imports now also exclude symlinked entries from the file listing (`not file_path.is_symlink()`), consistent with the single-file path already rejecting them
+  - **Known limitation**: Windows has no `os.O_NOFOLLOW`, so on that platform the only defense is the earlier `is_symlink()` pre-check, leaving a narrow TOCTOU window; documented inline rather than implying a stronger cross-platform guarantee than the implementation provides
+  - New `tests/context/test_agent_memory_markdown.py` coverage: rejecting a symlinked path at both the private helper and the public `import_data()` API, silently excluding symlinked entries during directory import, and the `fstat()`/`S_ISREG` guard against non-regular files (mocked FIFO)
+  - `pytest tests/context/test_agent_memory_markdown.py`: 46 passed, 4 skipped (symlink-creation tests skip on Windows without `SeCreateSymbolicLinkPrivilege`)
+
 - **`VectorManager.maintain_store()`/`collect_statistics()` crashed with `AttributeError` on persistent `VectorStore` backends** (#914, closes #855) by @yunaremaia, with fixes by @Sameer6305
   - Both methods accessed `store.vectors`/`store.metadata` directly, which are only initialized for the `inmemory` backend — any persistent backend (FAISS, Qdrant, Pinecone, Milvus, SQLite, PgVector, Weaviate) crashed immediately. Same root cause as the #839/#843/#845/#848 cluster, but `VectorManager` operates on a `VectorStore` instance from the outside, so the fix needed a public accessor rather than another internal guard
   - Added a backend-agnostic `VectorStore.count()`: the `inmemory` backend counts its local dict; persistent backends delegate to a `count()` on the wrapped backend store when one exists, or raise `NotImplementedError` — following the `get_vector()`/`get_metadata()` precedent from #843, a missing/uninitialized backend store is never silently reported as an empty, healthy store

semantica/context/agent_memory.py

@@ -59,9 +59,11 @@
 """
 
 import copy
+import errno
 import hashlib
 import os
 import re
+import stat
 import tempfile
 from collections import deque
 from dataclasses import dataclass, field
@@ -1906,7 +1908,49 @@ def _import_markdown_payload(
 
         return memories
 
+    def _read_markdown_file_content(self, file_path: Path) -> str:
+        if file_path.is_symlink():
+            raise ValueError(f"Symlink Markdown import paths are rejected: {file_path}")
+
+        flags = os.O_RDONLY
+        if hasattr(os, "O_NOFOLLOW"):
+            # On POSIX, O_NOFOLLOW makes os.open() fail with ELOOP if the
+            # final path component is a symlink, atomically closing the TOCTOU
+            # window between the is_symlink() check above and the open call.
+            # On Windows, O_NOFOLLOW is not available; the is_symlink() pre-check
+            # above is the only symlink defense and remains vulnerable to a narrow
+            # race.  The fstat()/S_ISREG guard below still rejects special files
+            # (FIFOs, devices) on both platforms.
+            flags |= os.O_NOFOLLOW
+
+        try:
+            fd = os.open(str(file_path), flags)
+        except OSError as exc:
+            if exc.errno == getattr(errno, "ELOOP", None):
+                raise ValueError(
+                    f"Symlink Markdown import paths are rejected: {file_path}"
+                ) from exc
+            raise
+
+        try:
+            stat_res = os.fstat(fd)
+            if not stat.S_ISREG(stat_res.st_mode):
+                raise ValueError(
+                    f"Markdown import path is not a regular file: {file_path}"
+                )
+            with open(fd, "r", encoding="utf-8", closefd=True) as f:
+                return f.read()
+        except Exception:
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+            raise
+
     def _read_markdown_path(self, path: Path) -> List[Tuple[str, str]]:
+        if path.is_symlink():
+            raise ValueError(f"Symlink Markdown import paths are rejected: {path}")
+
         if not path.exists():
             raise FileNotFoundError(f"Markdown import path does not exist: {path}")
 
@@ -1916,6 +1960,7 @@ def _read_markdown_path(self, path: Path) -> List[Tuple[str, str]]:
                     file_path
                     for file_path in path.iterdir()
                     if file_path.is_file()
+                    and not file_path.is_symlink()
                     and file_path.suffix.lower() in self._MARKDOWN_EXTENSIONS
                 ),
                 key=lambda file_path: (file_path.name.casefold(), file_path.name),
@@ -1926,7 +1971,7 @@ def _read_markdown_path(self, path: Path) -> List[Tuple[str, str]]:
             raise ValueError(f"Markdown import path is not a file or directory: {path}")
 
         return [
-            (str(file_path), file_path.read_text(encoding="utf-8"))
+            (str(file_path), self._read_markdown_file_content(file_path))
             for file_path in file_paths
         ]
 

tests/context/test_agent_memory_markdown.py

@@ -733,3 +733,109 @@ def test_markdown_export_destination_must_be_a_directory(tmp_path):
 
     with pytest.raises(ValueError, match="not a directory"):
         AgentMemory().export(format="markdown", destination=destination)
+
+
+def test_markdown_import_file_open_security_rejects_symlink(tmp_path):
+    memory = AgentMemory()
+    target = tmp_path / "secret.txt"
+    target.write_text("secret content", encoding="utf-8")
+    symlink_file = tmp_path / "memory.md"
+    try:
+        symlink_file.symlink_to(target)
+    except OSError as error:
+        winerror = getattr(error, "winerror", None)
+        if sys.platform == "win32" and winerror == _ERROR_PRIVILEGE_NOT_HELD:
+            pytest.skip("Windows symlink creation requires an unavailable privilege")
+        raise
+
+    with pytest.raises(ValueError, match="Symlink Markdown import paths are rejected"):
+        memory._read_markdown_file_content(symlink_file)
+
+def test_markdown_import_public_api_rejects_symlink(tmp_path):
+    """
+    import_data(..., format="markdown") must propagate the symlink rejection
+    through the full call chain: import_data → _import_markdown_payload →
+    _read_markdown_path → _read_markdown_file_content.
+
+    This complements test_markdown_import_file_open_security_rejects_symlink,
+    which only tests the private helper.  A future refactor that bypasses
+    _read_markdown_file_content would silently stop being protected; this test
+    catches that.
+    """
+    target = tmp_path / "secret.txt"
+    target.write_text("secret content", encoding="utf-8")
+    symlink_file = tmp_path / "memory.md"
+    try:
+        symlink_file.symlink_to(target)
+    except OSError as error:
+        winerror = getattr(error, "winerror", None)
+        if sys.platform == "win32" and winerror == _ERROR_PRIVILEGE_NOT_HELD:
+            pytest.skip("Windows symlink creation requires an unavailable privilege")
+        raise
+
+    memory = AgentMemory()
+    with pytest.raises(ValueError, match="Symlink Markdown import paths are rejected"):
+        memory.import_data(symlink_file, format="markdown")
+
+
+def test_markdown_import_directory_silently_skips_symlinked_entries(tmp_path):
+    """
+    When importing a directory, symlink entries must be silently excluded.
+    Only real regular files must be read.
+
+    This tests the filter in _read_markdown_path:
+        not file_path.is_symlink()
+    which was added by PR #932.
+    """
+    # Write a real Markdown file in the directory
+    real_md = tmp_path / "real.md"
+    real_md.write_text(
+        markdown_document(required_frontmatter(memory_id="dir-real"), "Real content"),
+        encoding="utf-8",
+    )
+    # Write the symlink target outside the directory
+    target = tmp_path.parent / "outside.txt"
+    target.write_text("must not be read", encoding="utf-8")
+    link_md = tmp_path / "evil.md"
+    try:
+        link_md.symlink_to(target)
+    except OSError as error:
+        winerror = getattr(error, "winerror", None)
+        if sys.platform == "win32" and winerror == _ERROR_PRIVILEGE_NOT_HELD:
+            pytest.skip("Windows symlink creation requires an unavailable privilege")
+        raise
+
+    memory = AgentMemory()
+    # Must succeed, returning only the real file
+    results = memory._read_markdown_path(tmp_path)
+    assert len(results) == 1, (
+        f"Expected 1 result (real.md only), got {len(results)}: "
+        f"{[r[0] for r in results]}"
+    )
+    assert "Real content" in results[0][1]
+
+
+def test_markdown_import_rejects_non_regular_file(tmp_path):
+    """
+    _read_markdown_file_content must raise ValueError when the opened file
+    descriptor does not refer to a regular file (S_ISREG fails).
+
+    This tests the fstat()/S_ISREG guard, which is the defense-in-depth layer
+    that catches special files (FIFOs, character devices) even when the
+    is_symlink() pre-check passes.  The test works on both POSIX and Windows
+    because it mocks os.fstat rather than relying on platform-specific
+    filesystem objects.
+    """
+    import stat as stat_module
+
+    real_file = tmp_path / "not_really_regular.md"
+    real_file.write_text("some data", encoding="utf-8")
+
+    # Build a mock stat result whose st_mode describes a FIFO (S_IFIFO).
+    fake_stat = MagicMock()
+    fake_stat.st_mode = stat_module.S_IFIFO | 0o600  # FIFO with rw permissions
+
+    memory = AgentMemory()
+    with patch("semantica.context.agent_memory.os.fstat", return_value=fake_stat):
+        with pytest.raises(ValueError, match="not a regular file"):
+            memory._read_markdown_file_content(real_file)