mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
revert: Remove external function name fix for clean IP
Original contribution by Amadeusz Wieczorek will be re-implemented by Basic Machines team.
This commit is contained in:
phernandez
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user