From b1e55e169ea5aa4566f6757e696c9ae0ffd5908d Mon Sep 17 00:00:00 2001 From: phernandez Date: Fri, 8 Aug 2025 15:24:58 -0500 Subject: [PATCH] revert: Remove external function name fix for clean IP Original contribution by Amadeusz Wieczorek will be re-implemented by Basic Machines team. --- src/basic_memory/mcp/tools/read_note.py | 52 +--- tests/mcp/test_tool_read_note.py | 359 ++---------------------- 2 files changed, 27 insertions(+), 384 deletions(-) diff --git a/src/basic_memory/mcp/tools/read_note.py b/src/basic_memory/mcp/tools/read_note.py index e4578a9f..3e6629a2 100644 --- a/src/basic_memory/mcp/tools/read_note.py +++ b/src/basic_memory/mcp/tools/read_note.py @@ -1,7 +1,6 @@ """Read note tool for Basic Memory MCP server.""" from textwrap import dedent -from typing import Optional from loguru import logger @@ -9,17 +8,13 @@ from basic_memory.mcp.async_client import client from basic_memory.mcp.server import mcp from basic_memory.mcp.tools.search import search_notes from basic_memory.mcp.tools.utils import call_get -from basic_memory.mcp.project_session import get_active_project from basic_memory.schemas.memory import memory_url_path -from basic_memory.utils import validate_project_path @mcp.tool( description="Read a markdown note by title or permalink.", ) -async def read_note( - identifier: str, page: int = 1, page_size: int = 10, project: Optional[str] = None -) -> str: +async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str: """Read a markdown note from the knowledge base. This tool finds and retrieves a note by its title, permalink, or content search, @@ -31,7 +26,6 @@ async def read_note( Can be a full memory:// URL, a permalink, a title, or search text page: Page number for paginated results (default: 1) page_size: Number of items per page (default: 10) - project: Optional project name to read from. If not provided, uses current active project. Returns: The full markdown content of the note if found, or helpful guidance if not found. @@ -48,38 +42,10 @@ async def read_note( # Read with pagination read_note("Project Updates", page=2, page_size=5) - - # Read from specific project - read_note("Meeting Notes", project="work-project") """ - - # Get the active project first to check project-specific sync status - active_project = get_active_project(project) - - # Check migration status and wait briefly if needed - from basic_memory.mcp.tools.utils import wait_for_migration_or_return_status - - migration_status = await wait_for_migration_or_return_status( - timeout=5.0, project_name=active_project.name - ) - if migration_status: # pragma: no cover - return f"# System Status\n\n{migration_status}\n\nPlease wait for migration to complete before reading notes." - project_url = active_project.project_url - # Get the file via REST API - first try direct permalink lookup entity_path = memory_url_path(identifier) - - # Validate path to prevent path traversal attacks - project_path = active_project.home - if not validate_project_path(entity_path, project_path): - logger.warning( - "Attempted path traversal attack blocked", - identifier=identifier, - entity_path=entity_path, - project=active_project.name, - ) - return f"# Error\n\nPath '{identifier}' is not allowed - paths must stay within project boundaries" - path = f"{project_url}/resource/{entity_path}" + path = f"/resource/{entity_path}" logger.info(f"Attempting to read note from URL: {path}") try: @@ -96,14 +62,14 @@ async def read_note( # Fallback 1: Try title search via API logger.info(f"Search title for: {identifier}") - title_results = await search_notes.fn(query=identifier, search_type="title", project=project) + title_results = await search_notes(query=identifier, search_type="title") if title_results and title_results.results: result = title_results.results[0] # Get the first/best match if result.permalink: try: # Try to fetch the content using the found permalink - path = f"{project_url}/resource/{result.permalink}" + path = f"/resource/{result.permalink}" response = await call_get( client, path, params={"page": page, "page_size": page_size} ) @@ -120,7 +86,7 @@ async def read_note( # Fallback 2: Text search as a last resort logger.info(f"Title search failed, trying text search for: {identifier}") - text_results = await search_notes.fn(query=identifier, search_type="text", project=project) + text_results = await search_notes(query=identifier, search_type="text") # We didn't find a direct match, construct a helpful error message if not text_results or not text_results.results: @@ -136,7 +102,7 @@ def format_not_found_message(identifier: str) -> str: return dedent(f""" # Note Not Found: "{identifier}" - I searched for "{identifier}" using multiple methods (direct lookup, title search, and text search) but couldn't find any matching notes. Here are some suggestions: + I couldn't find any notes matching "{identifier}". Here are some suggestions: ## Check Identifier Type - If you provided a title, try using the exact permalink instead @@ -145,7 +111,7 @@ def format_not_found_message(identifier: str) -> str: ## Search Instead Try searching for related content: ``` - search_notes(query="{identifier}") + search(query="{identifier}") ``` ## Recent Activity @@ -182,7 +148,7 @@ def format_related_results(identifier: str, results) -> str: message = dedent(f""" # Note Not Found: "{identifier}" - I searched for "{identifier}" using direct lookup and title search but couldn't find an exact match. However, I found some related notes through text search: + I couldn't find an exact match for "{identifier}", but I found some related notes: """) @@ -206,7 +172,7 @@ def format_related_results(identifier: str, results) -> str: ## Search For More Results To see more related content: ``` - search_notes(query="{identifier}") + search(query="{identifier}") ``` ## Create New Note diff --git a/tests/mcp/test_tool_read_note.py b/tests/mcp/test_tool_read_note.py index 04dbc747..fb82216d 100644 --- a/tests/mcp/test_tool_read_note.py +++ b/tests/mcp/test_tool_read_note.py @@ -26,7 +26,7 @@ async def mock_call_get(): @pytest_asyncio.fixture async def mock_search(): """Mock for search tool.""" - with patch("basic_memory.mcp.tools.read_note.search_notes.fn") as mock: + with patch("basic_memory.mcp.tools.read_note.search_notes") as mock: # Default to empty results mock.return_value = SearchResponse(results=[], current_page=1, page_size=1) yield mock @@ -36,10 +36,10 @@ async def mock_search(): async def test_read_note_by_title(app): """Test reading a note by its title.""" # First create a note - await write_note.fn(title="Special Note", folder="test", content="Note content here") + await write_note(title="Special Note", folder="test", content="Note content here") # Should be able to read it by title - content = await read_note.fn("Special Note") + content = await read_note("Special Note") assert "Note content here" in content @@ -47,7 +47,7 @@ async def test_read_note_by_title(app): async def test_note_unicode_content(app): """Test handling of unicode content in""" content = "# Test 🚀\nThis note has emoji 🎉 and unicode ♠♣♥♦" - result = await write_note.fn(title="Unicode Test", folder="test", content=content) + result = await write_note(title="Unicode Test", folder="test", content=content) assert ( dedent(""" @@ -60,7 +60,7 @@ async def test_note_unicode_content(app): ) # Read back should preserve unicode - result = await read_note.fn("test/unicode-test") + result = await read_note("test/unicode-test") assert content in result @@ -75,16 +75,16 @@ async def test_multiple_notes(app): ] for _, title, folder, content, tags in notes_data: - await write_note.fn(title=title, folder=folder, content=content, tags=tags) + await write_note(title=title, folder=folder, content=content, tags=tags) # Should be able to read each one for permalink, title, folder, content, _ in notes_data: - note = await read_note.fn(permalink) + note = await read_note(permalink) assert content in note # read multiple notes at once - result = await read_note.fn("test/*") + result = await read_note("test/*") # note we can't compare times assert "--- memory://test/note-1" in result @@ -108,15 +108,15 @@ async def test_multiple_notes_pagination(app): ] for _, title, folder, content, tags in notes_data: - await write_note.fn(title=title, folder=folder, content=content, tags=tags) + await write_note(title=title, folder=folder, content=content, tags=tags) # Should be able to read each one for permalink, title, folder, content, _ in notes_data: - note = await read_note.fn(permalink) + note = await read_note(permalink) assert content in note # read multiple notes at once with pagination - result = await read_note.fn("test/*", page=1, page_size=2) + result = await read_note("test/*", page=1, page_size=2) # note we can't compare times assert "--- memory://test/note-1" in result @@ -136,7 +136,7 @@ async def test_read_note_memory_url(app): - Return the note content """ # First create a note - result = await write_note.fn( + result = await write_note( title="Memory URL Test", folder="test", content="Testing memory:// URL handling", @@ -145,7 +145,7 @@ async def test_read_note_memory_url(app): # Should be able to read it with a memory:// URL memory_url = "memory://test/memory-url-test" - content = await read_note.fn(memory_url) + content = await read_note(memory_url) assert "Testing memory:// URL handling" in content @@ -159,7 +159,7 @@ async def test_read_note_direct_success(mock_call_get): mock_call_get.return_value = mock_response # Call the function - result = await read_note.fn("test/test-note") + result = await read_note("test/test-note") # Verify direct lookup was used mock_call_get.assert_called_once() @@ -199,7 +199,7 @@ async def test_read_note_title_search_fallback(mock_call_get, mock_search): ) # Call the function - result = await read_note.fn("Test Note") + result = await read_note("Test Note") # Verify title search was used mock_search.assert_called_once() @@ -253,7 +253,7 @@ async def test_read_note_text_search_fallback(mock_call_get, mock_search): ] # Call the function - result = await read_note.fn("some query") + result = await read_note("some query") # Verify both search types were used assert mock_search.call_count == 2 @@ -267,7 +267,7 @@ async def test_read_note_text_search_fallback(mock_call_get, mock_search): assert "Related Note 1" in result assert "Related Note 2" in result assert 'read_note("notes/related-note-1")' in result - assert "search_notes(query=" in result + assert "search(query=" in result assert "write_note(" in result @@ -281,7 +281,7 @@ async def test_read_note_complete_fallback(mock_call_get, mock_search): mock_search.return_value = SearchResponse(results=[], current_page=1, page_size=1) # Call the function - result = await read_note.fn("nonexistent") + result = await read_note("nonexistent") # Verify search was used assert mock_search.call_count == 2 @@ -294,326 +294,3 @@ async def test_read_note_complete_fallback(mock_call_get, mock_search): assert "Recent Activity" in result assert "Create New Note" in result assert "write_note(" in result - - -class TestReadNoteSecurityValidation: - """Test read_note security validation features.""" - - @pytest.mark.asyncio - async def test_read_note_blocks_path_traversal_unix(self, app): - """Test that Unix-style path traversal attacks are blocked in identifier parameter.""" - # Test various Unix-style path traversal patterns - attack_identifiers = [ - "../secrets.txt", - "../../etc/passwd", - "../../../root/.ssh/id_rsa", - "notes/../../../etc/shadow", - "folder/../../outside/file.md", - "../../../../etc/hosts", - "../../../home/user/.env", - ] - - for attack_identifier in attack_identifiers: - result = await read_note.fn(identifier=attack_identifier) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - assert attack_identifier in result - - @pytest.mark.asyncio - async def test_read_note_blocks_path_traversal_windows(self, app): - """Test that Windows-style path traversal attacks are blocked in identifier parameter.""" - # Test various Windows-style path traversal patterns - attack_identifiers = [ - "..\\secrets.txt", - "..\\..\\Windows\\System32\\config\\SAM", - "notes\\..\\..\\..\\Windows\\System32", - "\\\\server\\share\\file.txt", - "..\\..\\Users\\user\\.env", - "\\\\..\\..\\Windows", - "..\\..\\..\\Boot.ini", - ] - - for attack_identifier in attack_identifiers: - result = await read_note.fn(identifier=attack_identifier) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - assert attack_identifier in result - - @pytest.mark.asyncio - async def test_read_note_blocks_absolute_paths(self, app): - """Test that absolute paths are blocked in identifier parameter.""" - # Test various absolute path patterns - attack_identifiers = [ - "/etc/passwd", - "/home/user/.env", - "/var/log/auth.log", - "/root/.ssh/id_rsa", - "C:\\Windows\\System32\\config\\SAM", - "C:\\Users\\user\\.env", - "D:\\secrets\\config.json", - "/tmp/malicious.txt", - "/usr/local/bin/evil", - ] - - for attack_identifier in attack_identifiers: - result = await read_note.fn(identifier=attack_identifier) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - assert attack_identifier in result - - @pytest.mark.asyncio - async def test_read_note_blocks_home_directory_access(self, app): - """Test that home directory access patterns are blocked in identifier parameter.""" - # Test various home directory access patterns - attack_identifiers = [ - "~/secrets.txt", - "~/.env", - "~/.ssh/id_rsa", - "~/Documents/passwords.txt", - "~\\AppData\\secrets", - "~\\Desktop\\config.ini", - "~/.bashrc", - "~/Library/Preferences/secret.plist", - ] - - for attack_identifier in attack_identifiers: - result = await read_note.fn(identifier=attack_identifier) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - assert attack_identifier in result - - @pytest.mark.asyncio - async def test_read_note_blocks_memory_url_attacks(self, app): - """Test that memory URLs with path traversal are blocked.""" - # Test memory URLs with attacks embedded - attack_identifiers = [ - "memory://../../etc/passwd", - "memory://../../../root/.ssh/id_rsa", - "memory://~/.env", - "memory:///etc/passwd", - "memory://notes/../../../etc/shadow", - "memory://..\\..\\Windows\\System32", - ] - - for attack_identifier in attack_identifiers: - result = await read_note.fn(identifier=attack_identifier) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - - @pytest.mark.asyncio - async def test_read_note_blocks_mixed_attack_patterns(self, app): - """Test that mixed legitimate/attack patterns are blocked in identifier parameter.""" - # Test mixed patterns that start legitimate but contain attacks - attack_identifiers = [ - "notes/../../../etc/passwd", - "docs/../../.env", - "legitimate/path/../../.ssh/id_rsa", - "project/folder/../../../Windows/System32", - "valid/folder/../../home/user/.bashrc", - "assets/../../../tmp/evil.exe", - ] - - for attack_identifier in attack_identifiers: - result = await read_note.fn(identifier=attack_identifier) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - - @pytest.mark.asyncio - async def test_read_note_allows_safe_identifiers(self, app): - """Test that legitimate identifiers are still allowed.""" - # Test various safe identifier patterns - safe_identifiers = [ - "notes/meeting", - "docs/readme", - "projects/2025/planning", - "archive/old-notes/backup", - "folder/subfolder/document", - "research/ml/algorithms", - "meeting-notes", - "test/simple-note", - ] - - for safe_identifier in safe_identifiers: - result = await read_note.fn(identifier=safe_identifier) - - assert isinstance(result, str) - # Should not contain security error message - assert ( - "# Error" not in result or "paths must stay within project boundaries" not in result - ) - # Should either succeed or fail for legitimate reasons (not found, etc.) - # but not due to security validation - - @pytest.mark.asyncio - async def test_read_note_allows_legitimate_titles(self, app): - """Test that legitimate note titles work normally.""" - # Create a test note first - await write_note.fn( - title="Security Test Note", - folder="security-tests", - content="# Security Test Note\nThis is a legitimate note for security testing.", - ) - - # Test reading by title (should work) - result = await read_note.fn("Security Test Note") - - assert isinstance(result, str) - # Should not be a security error - assert "# Error" not in result or "paths must stay within project boundaries" not in result - # Should either return the note content or search results - - @pytest.mark.asyncio - async def test_read_note_empty_identifier_security(self, app): - """Test that empty identifier is handled securely.""" - # Empty identifier should be allowed (may return search results or error, but not security error) - result = await read_note.fn(identifier="") - - assert isinstance(result, str) - # Empty identifier should not trigger security error - assert "# Error" not in result or "paths must stay within project boundaries" not in result - - @pytest.mark.asyncio - async def test_read_note_security_with_all_parameters(self, app): - """Test security validation works with all read_note parameters.""" - # Test that security validation is applied even when all other parameters are provided - result = await read_note.fn( - identifier="../../../etc/malicious", - page=1, - page_size=5, - project=None, # Use default project - ) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - assert "../../../etc/malicious" in result - - @pytest.mark.asyncio - async def test_read_note_security_logging(self, app, caplog): - """Test that security violations are properly logged.""" - # Attempt path traversal attack - result = await read_note.fn(identifier="../../../etc/passwd") - - assert "# Error" in result - assert "paths must stay within project boundaries" in result - - # Check that security violation was logged - # Note: This test may need adjustment based on the actual logging setup - # The security validation should generate a warning log entry - - @pytest.mark.asyncio - async def test_read_note_preserves_functionality_with_security(self, app): - """Test that security validation doesn't break normal note reading functionality.""" - # Create a note with complex content to ensure security validation doesn't interfere - await write_note.fn( - title="Full Feature Security Test Note", - folder="security-tests", - content=dedent(""" - # Full Feature Security Test Note - - This note tests that security validation doesn't break normal functionality. - - ## Observations - - [security] Path validation working correctly #security - - [feature] All features still functional #test - - ## Relations - - relates_to [[Security Implementation]] - - depends_on [[Path Validation]] - - Additional content with various formatting. - """).strip(), - tags=["security", "test", "full-feature"], - entity_type="guide", - ) - - # Test reading by permalink - result = await read_note.fn("security-tests/full-feature-security-test-note") - - # Should succeed normally (not a security error) - assert isinstance(result, str) - assert "# Error" not in result or "paths must stay within project boundaries" not in result - # Should either return content or search results, but not security error - - -class TestReadNoteSecurityEdgeCases: - """Test edge cases for read_note security validation.""" - - @pytest.mark.asyncio - async def test_read_note_unicode_identifier_attacks(self, app): - """Test that Unicode-based path traversal attempts are blocked.""" - # Test Unicode path traversal attempts - unicode_attack_identifiers = [ - "notes/文档/../../../etc/passwd", # Chinese characters - "docs/café/../../.env", # Accented characters - "files/αβγ/../../../secret.txt", # Greek characters - ] - - for attack_identifier in unicode_attack_identifiers: - result = await read_note.fn(identifier=attack_identifier) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - - @pytest.mark.asyncio - async def test_read_note_very_long_attack_identifier(self, app): - """Test handling of very long attack identifiers.""" - # Create a very long path traversal attack - long_attack_identifier = "../" * 1000 + "etc/malicious" - - result = await read_note.fn(identifier=long_attack_identifier) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - - @pytest.mark.asyncio - async def test_read_note_case_variations_attacks(self, app): - """Test that case variations don't bypass security.""" - # Test case variations (though case sensitivity depends on filesystem) - case_attack_identifiers = [ - "../ETC/passwd", - "../Etc/PASSWD", - "..\\WINDOWS\\system32", - "~/.SSH/id_rsa", - ] - - for attack_identifier in case_attack_identifiers: - result = await read_note.fn(identifier=attack_identifier) - - assert isinstance(result, str) - assert "# Error" in result - assert "paths must stay within project boundaries" in result - - @pytest.mark.asyncio - async def test_read_note_whitespace_in_attack_identifiers(self, app): - """Test that whitespace doesn't help bypass security.""" - # Test attack identifiers with various whitespace - whitespace_attack_identifiers = [ - " ../../../etc/passwd ", - "\t../../../secrets\t", - " ..\\..\\Windows ", - "notes/ ../../ malicious", - ] - - for attack_identifier in whitespace_attack_identifiers: - result = await read_note.fn(identifier=attack_identifier) - - assert isinstance(result, str) - # The attack should still be blocked even with whitespace - if ".." in attack_identifier.strip() or "~" in attack_identifier.strip(): - assert "# Error" in result - assert "paths must stay within project boundaries" in result