fix(context): reject Markdown import symlinks

docs/reference/context.md

@@ -626,7 +626,8 @@ files is idempotent. Memory-local `entities` and `relationships` are preserved a
 provenance but are not applied to `ContextGraph` by Markdown import. Use a dedicated
 export directory: matching files are overwritten, but unrelated or stale Markdown
 files are not deleted automatically. Export refuses to overwrite symbolic links and
-uses atomic file replacement. Timestamp offsets are preserved in Markdown and
+uses atomic file replacement; import also refuses symbolic-link files and directories.
+Timestamp offsets are preserved in Markdown and
 normalized to UTC only for comparisons, so aware and local-naive records can be
 queried together safely. Vector-store writes are deferred until the in-memory import
 commits; adapter synchronization remains best-effort and logs failures.

semantica/context/agent_memory.py

@@ -59,9 +59,11 @@ License: MIT
 """
 
 import copy
+import errno
 import hashlib
 import os
 import re
+import stat
 import tempfile
 from collections import deque
 from dataclasses import dataclass, field
@@ -1865,7 +1867,8 @@ class AgentMemory:
             if "\n" not in data and "\r" not in data:
                 candidate = Path(data)
                 try:
-                    candidate_exists = candidate.exists()
+                    candidate_is_symlink = candidate.is_symlink()
+                    candidate_exists = candidate_is_symlink or candidate.exists()
                 except OSError as exc:
                     error_message = (
                         "Failed to inspect possible Markdown import "
@@ -1907,29 +1910,64 @@ class AgentMemory:
         return memories
 
     def _read_markdown_path(self, path: Path) -> List[Tuple[str, str]]:
+        if path.is_symlink():
+            raise ValueError(f"Refusing to import Markdown symbolic link: {path}")
+
         if not path.exists():
             raise FileNotFoundError(f"Markdown import path does not exist: {path}")
 
         if path.is_dir():
-            file_paths = sorted(
-                (
-                    file_path
-                    for file_path in path.iterdir()
-                    if file_path.is_file()
-                    and file_path.suffix.lower() in self._MARKDOWN_EXTENSIONS
-                ),
-                key=lambda file_path: (file_path.name.casefold(), file_path.name),
-            )
+            file_paths = []
+            for file_path in path.iterdir():
+                if file_path.suffix.lower() not in self._MARKDOWN_EXTENSIONS:
+                    continue
+                if file_path.is_symlink():
+                    raise ValueError(
+                        f"Refusing to import Markdown symbolic link: {file_path}"
+                    )
+                if file_path.is_file():
+                    file_paths.append(file_path)
+            file_paths.sort(key=lambda item: (item.name.casefold(), item.name))
         elif path.is_file():
             file_paths = [path]
         else:
             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(file_path))
             for file_path in file_paths
         ]
 
+    @staticmethod
+    def _read_markdown_file(file_path: Path) -> str:
+        """Read a regular Markdown file without following a raced symlink."""
+        if file_path.is_symlink():
+            raise ValueError(f"Refusing to import Markdown symbolic link: {file_path}")
+
+        flags = os.O_RDONLY
+        nofollow_flag = getattr(os, "O_NOFOLLOW", 0)
+        flags |= nofollow_flag
+        try:
+            file_descriptor = os.open(file_path, flags)
+        except OSError as exc:
+            if nofollow_flag and exc.errno == errno.ELOOP:
+                raise ValueError(
+                    f"Refusing to import Markdown symbolic link: {file_path}"
+                ) from exc
+            raise
+
+        try:
+            if not stat.S_ISREG(os.fstat(file_descriptor).st_mode):
+                raise ValueError(
+                    f"Markdown import path is not a regular file: {file_path}"
+                )
+            with os.fdopen(file_descriptor, mode="r", encoding="utf-8") as source:
+                file_descriptor = -1
+                return source.read()
+        finally:
+            if file_descriptor >= 0:
+                os.close(file_descriptor)
+
     def _markdown_to_memory_dict(
         self, document: str, source: str = "markdown document"
     ) -> Dict[str, Any]:

tests/context/test_agent_memory_markdown.py

@@ -1,6 +1,8 @@
 import errno
+import os
 from copy import deepcopy
 from datetime import datetime, timedelta, timezone
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -698,6 +700,85 @@ def test_markdown_string_path_inspection_errors_are_actionable():
     assert exc_info.value.__cause__ is original_error
 
 
+@pytest.mark.parametrize("use_string_path", [False, True])
+def test_markdown_import_rejects_symlinked_file(tmp_path, use_string_path):
+    outside = tmp_path / "outside.md"
+    outside.write_text(
+        markdown_document(required_frontmatter(), "Do not import"),
+        encoding="utf-8",
+    )
+    source = tmp_path / "memory.md"
+    source.symlink_to(outside)
+    payload = str(source) if use_string_path else source
+    memory = AgentMemory()
+
+    with pytest.raises(ValueError, match="symbolic link"):
+        memory.import_data(payload, format="markdown")
+
+    assert memory.count() == 0
+
+
+@pytest.mark.parametrize("use_string_path", [False, True])
+def test_markdown_import_rejects_broken_symlink(tmp_path, use_string_path):
+    source = tmp_path / "missing-memory.md"
+    source.symlink_to(tmp_path / "missing-target.md")
+    payload = str(source) if use_string_path else source
+
+    with pytest.raises(ValueError, match="symbolic link"):
+        AgentMemory().import_data(payload, format="markdown")
+
+
+@pytest.mark.parametrize("use_string_path", [False, True])
+def test_markdown_import_rejects_symlinked_directory(tmp_path, use_string_path):
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "memory.md").write_text(
+        markdown_document(required_frontmatter(), "Do not import"),
+        encoding="utf-8",
+    )
+    source = tmp_path / "memory-export"
+    source.symlink_to(outside, target_is_directory=True)
+    payload = str(source) if use_string_path else source
+    memory = AgentMemory()
+
+    with pytest.raises(ValueError, match="symbolic link"):
+        memory.import_data(payload, format="markdown")
+
+    assert memory.count() == 0
+
+
+def test_markdown_import_rejects_symlinked_file_in_directory(tmp_path):
+    outside = tmp_path / "outside.md"
+    outside.write_text(
+        markdown_document(required_frontmatter(), "Do not import"),
+        encoding="utf-8",
+    )
+    source = tmp_path / "memory-export"
+    source.mkdir()
+    (source / "memory.md").symlink_to(outside)
+    memory = AgentMemory()
+
+    with pytest.raises(ValueError, match="symbolic link"):
+        memory.import_data(source, format="markdown")
+
+    assert memory.count() == 0
+
+
+@pytest.mark.skipif(not hasattr(os, "O_NOFOLLOW"), reason="requires O_NOFOLLOW")
+def test_markdown_import_does_not_follow_symlink_raced_before_open(tmp_path):
+    outside = tmp_path / "outside.md"
+    outside.write_text(
+        markdown_document(required_frontmatter(), "Do not import"),
+        encoding="utf-8",
+    )
+    source = tmp_path / "memory.md"
+    source.symlink_to(outside)
+
+    with patch.object(Path, "is_symlink", return_value=False):
+        with pytest.raises(ValueError, match="symbolic link"):
+            AgentMemory._read_markdown_file(source)
+
+
 def test_legacy_dict_import_behavior_is_unchanged():
     memory = AgentMemory()
     data = {