From a4a3b1b6899abe4ba0e158dc84a13178ea8b8109 Mon Sep 17 00:00:00 2001 From: jope-bm Date: Mon, 28 Jul 2025 14:52:15 -0600 Subject: [PATCH] fix: handle missing 'name' key in memory JSON import (#241) Signed-off-by: Joe P Co-authored-by: Claude Co-authored-by: jope-bm --- .../cli/commands/import_memory_json.py | 3 +- .../importers/memory_json_importer.py | 26 +++-- src/basic_memory/mcp/tools/read_note.py | 2 +- src/basic_memory/schemas/importer.py | 1 + src/basic_memory/utils.py | 12 +-- tests/cli/test_import_memory_json.py | 40 ++++++++ tests/mcp/test_tool_move_note.py | 6 +- tests/mcp/test_tool_read_content.py | 96 +++++++++---------- tests/mcp/test_tool_read_note.py | 12 ++- tests/mcp/test_tool_write_note.py | 12 +-- tests/utils/test_validate_project_path.py | 83 +++++++++------- 11 files changed, 183 insertions(+), 110 deletions(-) diff --git a/src/basic_memory/cli/commands/import_memory_json.py b/src/basic_memory/cli/commands/import_memory_json.py index 0913511b..52cf7879 100644 --- a/src/basic_memory/cli/commands/import_memory_json.py +++ b/src/basic_memory/cli/commands/import_memory_json.py @@ -76,7 +76,8 @@ def memory_json( Panel( f"[green]Import complete![/green]\n\n" f"Created {result.entities} entities\n" - f"Added {result.relations} relations", + f"Added {result.relations} relations\n" + f"Skipped {result.skipped_entities} entities\n", expand=False, ) ) diff --git a/src/basic_memory/importers/memory_json_importer.py b/src/basic_memory/importers/memory_json_importer.py index e315f18a..c6d58acb 100644 --- a/src/basic_memory/importers/memory_json_importer.py +++ b/src/basic_memory/importers/memory_json_importer.py @@ -32,6 +32,7 @@ class MemoryJsonImporter(Importer[EntityImportResult]): # First pass - collect all relations by source entity entity_relations: Dict[str, List[Relation]] = {} entities: Dict[str, Dict[str, Any]] = {} + skipped_entities: int = 0 # Ensure the base path exists base_path = config.home # pragma: no cover @@ -42,7 +43,13 @@ class MemoryJsonImporter(Importer[EntityImportResult]): for line in source_data: data = line if data["type"] == "entity": - entities[data["name"]] = data + # Handle different possible name keys + entity_name = data.get("name") or data.get("entityName") or data.get("id") + if not entity_name: + logger.warning(f"Entity missing name field: {data}") + skipped_entities += 1 + continue + entities[entity_name] = data elif data["type"] == "relation": # Store relation with its source entity source = data.get("from") or data.get("from_id") @@ -58,25 +65,31 @@ class MemoryJsonImporter(Importer[EntityImportResult]): # Second pass - create and write entities entities_created = 0 for name, entity_data in entities.items(): + # Get entity type with fallback + entity_type = entity_data.get("entityType") or entity_data.get("type") or "entity" + # Ensure entity type directory exists - entity_type_dir = base_path / entity_data["entityType"] + entity_type_dir = base_path / entity_type entity_type_dir.mkdir(parents=True, exist_ok=True) + # Get observations with fallback to empty list + observations = entity_data.get("observations", []) + entity = EntityMarkdown( frontmatter=EntityFrontmatter( metadata={ - "type": entity_data["entityType"], + "type": entity_type, "title": name, - "permalink": f"{entity_data['entityType']}/{name}", + "permalink": f"{entity_type}/{name}", } ), content=f"# {name}\n", - observations=[Observation(content=obs) for obs in entity_data["observations"]], + observations=[Observation(content=obs) for obs in observations], relations=entity_relations.get(name, []), ) # Write entity file - file_path = base_path / f"{entity_data['entityType']}/{name}.md" + file_path = base_path / f"{entity_type}/{name}.md" await self.write_entity(entity, file_path) entities_created += 1 @@ -87,6 +100,7 @@ class MemoryJsonImporter(Importer[EntityImportResult]): success=True, entities=entities_created, relations=relations_count, + skipped_entities=skipped_entities, ) except Exception as e: # pragma: no cover diff --git a/src/basic_memory/mcp/tools/read_note.py b/src/basic_memory/mcp/tools/read_note.py index 6a2e7123..e4578a9f 100644 --- a/src/basic_memory/mcp/tools/read_note.py +++ b/src/basic_memory/mcp/tools/read_note.py @@ -68,7 +68,7 @@ async def read_note( # 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): diff --git a/src/basic_memory/schemas/importer.py b/src/basic_memory/schemas/importer.py index 14f505a4..aa976650 100644 --- a/src/basic_memory/schemas/importer.py +++ b/src/basic_memory/schemas/importer.py @@ -32,3 +32,4 @@ class EntityImportResult(ImportResult): entities: int = 0 relations: int = 0 + skipped_entities: int = 0 diff --git a/src/basic_memory/utils.py b/src/basic_memory/utils.py index 33ffa139..e0f1b136 100644 --- a/src/basic_memory/utils.py +++ b/src/basic_memory/utils.py @@ -220,23 +220,23 @@ def validate_project_path(path: str, project_path: Path) -> bool: # Allow empty strings as they resolve to the project root if not path: return True - + # Check for obvious path traversal patterns first if ".." in path or "~" in path: return False - + # Check for Windows-style path traversal (even on Unix systems) if "\\.." in path or path.startswith("\\"): return False - + # Block absolute paths (Unix-style starting with / or Windows-style with drive letters) if path.startswith("/") or (len(path) >= 2 and path[1] == ":"): return False - + # Block paths with control characters (but allow whitespace that will be stripped) - if path.strip() and any(ord(c) < 32 and c not in [' ', '\t'] for c in path): + if path.strip() and any(ord(c) < 32 and c not in [" ", "\t"] for c in path): return False - + try: resolved = (project_path / path).resolve() return resolved.is_relative_to(project_path.resolve()) diff --git a/tests/cli/test_import_memory_json.py b/tests/cli/test_import_memory_json.py index 08faf67e..3c336b10 100644 --- a/tests/cli/test_import_memory_json.py +++ b/tests/cli/test_import_memory_json.py @@ -113,3 +113,43 @@ def test_import_json_command_handle_old_format(tmp_path): result = runner.invoke(import_app, ["memory-json", str(json_file)]) assert result.exit_code == 0 assert "Import complete" in result.output + + +def test_import_json_command_missing_name_key(tmp_path): + """Test handling JSON with missing 'name' key using 'id' instead.""" + # Create JSON with id instead of name (common in Knowledge Graph Memory Server) + data_with_id = [ + { + "type": "entity", + "id": "test_entity_id", + "entityType": "test", + "observations": ["Test observation with id"], + }, + { + "type": "entity", + "entityName": "test_entity_2", + "entityType": "test", + "observations": ["Test observation with entityName"], + }, + { + "type": "entity", + "name": "test_entity_title", + "entityType": "test", + "observations": ["Test observation with name"], + }, + ] + + json_file = tmp_path / "missing_name.json" + with open(json_file, "w", encoding="utf-8") as f: + for item in data_with_id: + f.write(json.dumps(item) + "\n") + + # Set up test environment + monkeypatch = pytest.MonkeyPatch() + monkeypatch.setenv("HOME", str(tmp_path)) + + # Run import - should not fail even without 'name' key + result = runner.invoke(import_app, ["memory-json", str(json_file)]) + assert result.exit_code == 0 + assert "Import complete" in result.output + assert "Created 3 entities" in result.output diff --git a/tests/mcp/test_tool_move_note.py b/tests/mcp/test_tool_move_note.py index f124748b..d99a0422 100644 --- a/tests/mcp/test_tool_move_note.py +++ b/tests/mcp/test_tool_move_note.py @@ -650,7 +650,7 @@ class TestMoveNoteSecurityValidation: assert isinstance(result, str) # Should NOT contain security error message assert "Security Validation Error" not in result - + # If it fails, it should be for other reasons like "already exists" or API errors if "Move Failed" in result: assert "paths must stay within project boundaries" not in result @@ -672,7 +672,7 @@ class TestMoveNoteSecurityValidation: ) assert "# Move Failed - Security Validation Error" 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 @@ -710,7 +710,7 @@ class TestMoveNoteSecurityValidation: # Test current directory references (should be safe) safe_paths = [ "./notes/file.md", - "folder/./file.md", + "folder/./file.md", "./folder/subfolder/file.md", ] diff --git a/tests/mcp/test_tool_read_content.py b/tests/mcp/test_tool_read_content.py index 9c007a76..bb1a78af 100644 --- a/tests/mcp/test_tool_read_content.py +++ b/tests/mcp/test_tool_read_content.py @@ -139,18 +139,19 @@ class TestReadContentSecurityValidation: # Mock the API call to simulate a successful response with patch("basic_memory.mcp.tools.read_content.call_get") as mock_call_get: mock_response = MagicMock() - mock_response.headers = { - "content-type": "text/markdown", - "content-length": "100" - } + mock_response.headers = {"content-type": "text/markdown", "content-length": "100"} mock_response.text = f"# Content for {safe_path}\nThis is test content." mock_call_get.return_value = mock_response - + result = await read_content.fn(path=safe_path) # Should succeed (not a security error) assert isinstance(result, dict) - assert result["type"] != "error" or "paths must stay within project boundaries" not in result.get("error", "") + assert result[ + "type" + ] != "error" or "paths must stay within project boundaries" not in result.get( + "error", "" + ) @pytest.mark.asyncio async def test_read_content_memory_url_processing(self, client): @@ -178,7 +179,7 @@ class TestReadContentSecurityValidation: assert result["type"] == "error" assert "paths must stay within project boundaries" in result["error"] - + # 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 @@ -189,18 +190,19 @@ class TestReadContentSecurityValidation: # Mock the API call since empty path should be allowed (resolves to project root) with patch("basic_memory.mcp.tools.read_content.call_get") as mock_call_get: mock_response = MagicMock() - mock_response.headers = { - "content-type": "text/markdown", - "content-length": "50" - } + mock_response.headers = {"content-type": "text/markdown", "content-length": "50"} mock_response.text = "# Root content" mock_call_get.return_value = mock_response - + result = await read_content.fn(path="") assert isinstance(result, dict) # Empty path should not trigger security error (it's handled as project root) - assert result["type"] != "error" or "paths must stay within project boundaries" not in result.get("error", "") + assert result[ + "type" + ] != "error" or "paths must stay within project boundaries" not in result.get( + "error", "" + ) @pytest.mark.asyncio async def test_read_content_current_directory_references_security(self, client): @@ -208,7 +210,7 @@ class TestReadContentSecurityValidation: # Test current directory references (should be safe) safe_paths = [ "./notes/file.md", - "folder/./file.md", + "folder/./file.md", "./folder/subfolder/file.md", ] @@ -216,18 +218,19 @@ class TestReadContentSecurityValidation: # Mock the API call for these safe paths with patch("basic_memory.mcp.tools.read_content.call_get") as mock_call_get: mock_response = MagicMock() - mock_response.headers = { - "content-type": "text/markdown", - "content-length": "100" - } + mock_response.headers = {"content-type": "text/markdown", "content-length": "100"} mock_response.text = f"# Content for {safe_path}" mock_call_get.return_value = mock_response - + result = await read_content.fn(path=safe_path) assert isinstance(result, dict) # Should NOT contain security error message - assert result["type"] != "error" or "paths must stay within project boundaries" not in result.get("error", "") + assert result[ + "type" + ] != "error" or "paths must stay within project boundaries" not in result.get( + "error", "" + ) class TestReadContentFunctionality: @@ -246,13 +249,10 @@ class TestReadContentFunctionality: # Mock the API call to simulate reading the file with patch("basic_memory.mcp.tools.read_content.call_get") as mock_call_get: mock_response = MagicMock() - mock_response.headers = { - "content-type": "text/markdown", - "content-length": "100" - } + mock_response.headers = {"content-type": "text/markdown", "content-length": "100"} mock_response.text = "# Test Document\nThis is test content for reading." mock_call_get.return_value = mock_response - + result = await read_content.fn(path="docs/test-document.md") assert isinstance(result, dict) @@ -267,16 +267,16 @@ class TestReadContentFunctionality: # Mock the API call to simulate reading an image with patch("basic_memory.mcp.tools.read_content.call_get") as mock_call_get: # Create a simple fake image data - fake_image_data = b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\rIDATx\x9cc\x00\x01\x00\x00\x05\x00\x01\r\n-\xdb\x00\x00\x00\x00IEND\xaeB`\x82' - + fake_image_data = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\rIDATx\x9cc\x00\x01\x00\x00\x05\x00\x01\r\n-\xdb\x00\x00\x00\x00IEND\xaeB`\x82" + mock_response = MagicMock() mock_response.headers = { "content-type": "image/png", - "content-length": str(len(fake_image_data)) + "content-length": str(len(fake_image_data)), } mock_response.content = fake_image_data mock_call_get.return_value = mock_response - + # Mock PIL Image processing with patch("basic_memory.mcp.tools.read_content.PILImage") as mock_pil: mock_img = MagicMock() @@ -285,10 +285,10 @@ class TestReadContentFunctionality: mock_img.mode = "RGB" mock_img.getbands.return_value = ["R", "G", "B"] mock_pil.open.return_value = mock_img - + with patch("basic_memory.mcp.tools.read_content.optimize_image") as mock_optimize: mock_optimize.return_value = b"optimized_image_data" - + result = await read_content.fn(path="assets/safe-image.png") assert isinstance(result, dict) @@ -302,24 +302,22 @@ class TestReadContentFunctionality: """Test reading content with explicit project parameter.""" # Mock the API call and project configuration with patch("basic_memory.mcp.tools.read_content.call_get") as mock_call_get: - with patch("basic_memory.mcp.tools.read_content.get_active_project") as mock_get_project: + with patch( + "basic_memory.mcp.tools.read_content.get_active_project" + ) as mock_get_project: # Mock project configuration mock_project = MagicMock() mock_project.project_url = "http://test" mock_project.home = Path("/test/project") mock_get_project.return_value = mock_project - + mock_response = MagicMock() - mock_response.headers = { - "content-type": "text/plain", - "content-length": "50" - } + mock_response.headers = {"content-type": "text/plain", "content-length": "50"} mock_response.text = "Project-specific content" mock_call_get.return_value = mock_response - + result = await read_content.fn( - path="notes/project-file.txt", - project="specific-project" + path="notes/project-file.txt", project="specific-project" ) assert isinstance(result, dict) @@ -332,7 +330,7 @@ class TestReadContentFunctionality: # Mock API call to return 404 with patch("basic_memory.mcp.tools.read_content.call_get") as mock_call_get: mock_call_get.side_effect = Exception("File not found") - + # This should pass security validation but fail on API call try: result = await read_content.fn(path="docs/nonexistent-file.md") @@ -348,15 +346,15 @@ class TestReadContentFunctionality: # Mock the API call to simulate reading a binary file with patch("basic_memory.mcp.tools.read_content.call_get") as mock_call_get: binary_data = b"Binary file content with special bytes: \x00\x01\x02\x03" - + mock_response = MagicMock() mock_response.headers = { "content-type": "application/octet-stream", - "content-length": str(len(binary_data)) + "content-length": str(len(binary_data)), } mock_response.content = binary_data mock_call_get.return_value = mock_response - + result = await read_content.fn(path="files/safe-binary.bin") assert isinstance(result, dict) @@ -399,14 +397,14 @@ class TestReadContentEdgeCases: for attack_path in encoded_attacks: try: result = await read_content.fn(path=attack_path) - + # These may or may not be blocked depending on URL decoding, # but should not cause security issues assert isinstance(result, dict) - + # If not blocked by security validation, may fail at API level # which is also acceptable - + except Exception: # Exception due to API failure or other issues is acceptable # as long as no actual traversal occurs @@ -435,7 +433,7 @@ class TestReadContentEdgeCases: """Test handling of very long attack paths.""" # Create a very long path traversal attack long_attack = "../" * 1000 + "etc/passwd" - + result = await read_content.fn(path=long_attack) assert isinstance(result, dict) @@ -458,4 +456,4 @@ class TestReadContentEdgeCases: assert isinstance(result, dict) assert result["type"] == "error" - assert "paths must stay within project boundaries" in result["error"] \ No newline at end of file + assert "paths must stay within project boundaries" in result["error"] diff --git a/tests/mcp/test_tool_read_note.py b/tests/mcp/test_tool_read_note.py index 802db0a4..04dbc747 100644 --- a/tests/mcp/test_tool_read_note.py +++ b/tests/mcp/test_tool_read_note.py @@ -450,7 +450,9 @@ class TestReadNoteSecurityValidation: 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 + 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 @@ -466,7 +468,7 @@ class TestReadNoteSecurityValidation: # 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 @@ -506,7 +508,7 @@ class TestReadNoteSecurityValidation: 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 @@ -539,7 +541,7 @@ class TestReadNoteSecurityValidation: # 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 @@ -571,7 +573,7 @@ class TestReadNoteSecurityEdgeCases: """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) diff --git a/tests/mcp/test_tool_write_note.py b/tests/mcp/test_tool_write_note.py index dfe61409..96baa5c8 100644 --- a/tests/mcp/test_tool_write_note.py +++ b/tests/mcp/test_tool_write_note.py @@ -864,7 +864,7 @@ class TestWriteNoteSecurityValidation: # Test current directory references (should be safe) safe_folders = [ "./notes", - "folder/./subfolder", + "folder/./subfolder", "./folder/subfolder", ] @@ -912,7 +912,7 @@ class TestWriteNoteSecurityValidation: 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 @@ -950,16 +950,16 @@ class TestWriteNoteSecurityValidation: assert "# Created note" in result assert "file_path: security-tests/Full Feature Security Test.md" in result assert "permalink: security-tests/full-feature-security-test" in result - + # Should process observations and relations assert "## Observations" in result assert "## Relations" in result assert "## Tags" in result - + # Should show proper counts assert "security: 1" in result assert "feature: 1" in result - + class TestWriteNoteSecurityEdgeCases: """Test edge cases for write_note security validation.""" @@ -990,7 +990,7 @@ class TestWriteNoteSecurityEdgeCases: """Test handling of very long attack folder paths.""" # Create a very long path traversal attack long_attack_folder = "../" * 1000 + "etc/malicious" - + result = await write_note.fn( title="Long Attack Test", folder=long_attack_folder, diff --git a/tests/utils/test_validate_project_path.py b/tests/utils/test_validate_project_path.py index fe70ea16..f5e87a6b 100644 --- a/tests/utils/test_validate_project_path.py +++ b/tests/utils/test_validate_project_path.py @@ -30,7 +30,9 @@ class TestValidateProjectPathSafety: ] for path in safe_paths: - assert validate_project_path(path, project_path), f"Safe path '{path}' should be allowed" + assert validate_project_path(path, project_path), ( + f"Safe path '{path}' should be allowed" + ) def test_empty_and_current_directory(self, tmp_path): """Test handling of empty paths and current directory references.""" @@ -39,7 +41,7 @@ class TestValidateProjectPathSafety: # Current directory should be safe assert validate_project_path(".", project_path) - + # Files in current directory should be safe assert validate_project_path("./file.txt", project_path) @@ -55,7 +57,9 @@ class TestValidateProjectPathSafety: ] for path in nested_paths: - assert validate_project_path(path, project_path), f"Nested path '{path}' should be allowed" + assert validate_project_path(path, project_path), ( + f"Nested path '{path}' should be allowed" + ) class TestValidateProjectPathAttacks: @@ -71,7 +75,7 @@ class TestValidateProjectPathAttacks: "../../", "../../../", "../etc/passwd", - "../../etc/passwd", + "../../etc/passwd", "../../../etc/passwd", "../../../../etc/passwd", "../../.env", @@ -82,7 +86,9 @@ class TestValidateProjectPathAttacks: ] for path in attack_paths: - assert not validate_project_path(path, project_path), f"Attack path '{path}' should be blocked" + assert not validate_project_path(path, project_path), ( + f"Attack path '{path}' should be blocked" + ) def test_windows_path_traversal(self, tmp_path): """Test that Windows-style path traversal is blocked.""" @@ -102,7 +108,9 @@ class TestValidateProjectPathAttacks: ] for path in attack_paths: - assert not validate_project_path(path, project_path), f"Windows attack path '{path}' should be blocked" + assert not validate_project_path(path, project_path), ( + f"Windows attack path '{path}' should be blocked" + ) def test_mixed_traversal_patterns(self, tmp_path): """Test paths that mix legitimate content with traversal.""" @@ -119,7 +127,9 @@ class TestValidateProjectPathAttacks: ] for path in mixed_attacks: - assert not validate_project_path(path, project_path), f"Mixed attack path '{path}' should be blocked" + assert not validate_project_path(path, project_path), ( + f"Mixed attack path '{path}' should be blocked" + ) def test_home_directory_access(self, tmp_path): """Test that home directory access patterns are blocked.""" @@ -137,11 +147,13 @@ class TestValidateProjectPathAttacks: ] for path in home_attacks: - assert not validate_project_path(path, project_path), f"Home directory attack '{path}' should be blocked" + assert not validate_project_path(path, project_path), ( + f"Home directory attack '{path}' should be blocked" + ) def test_unc_and_network_paths(self, tmp_path): """Test that UNC and network paths are blocked.""" - project_path = tmp_path / "project" + project_path = tmp_path / "project" project_path.mkdir() network_attacks = [ @@ -152,7 +164,9 @@ class TestValidateProjectPathAttacks: ] for path in network_attacks: - assert not validate_project_path(path, project_path), f"Network path attack '{path}' should be blocked" + assert not validate_project_path(path, project_path), ( + f"Network path attack '{path}' should be blocked" + ) def test_absolute_paths(self, tmp_path): """Test that absolute paths are blocked (if they contain traversal).""" @@ -163,7 +177,7 @@ class TestValidateProjectPathAttacks: # but our function should catch traversal patterns first absolute_attacks = [ "/etc/passwd", - "/home/user/.env", + "/home/user/.env", "/var/log/auth.log", "/root/.ssh/id_rsa", "C:\\Windows\\System32\\config\\SAM", @@ -212,7 +226,7 @@ class TestValidateProjectPathEdgeCases: # Create a very long but legitimate path long_path = "/".join(["verylongdirectoryname" * 10 for _ in range(10)]) - + # Should handle long paths gracefully (either allow or reject based on filesystem limits) try: result = validate_project_path(long_path, project_path) @@ -225,7 +239,7 @@ class TestValidateProjectPathEdgeCases: def test_nonexistent_project_path(self): """Test behavior when project path doesn't exist.""" nonexistent_project = Path("/this/path/does/not/exist") - + # Should still be able to validate relative paths assert validate_project_path("notes/file.txt", nonexistent_project) assert not validate_project_path("../../../etc/passwd", nonexistent_project) @@ -263,34 +277,36 @@ class TestValidateProjectPathEdgeCases: # These should all be blocked regardless of case case_variations = [ "../file.txt", - "../FILE.TXT", + "../FILE.TXT", "~/file.txt", "~/FILE.TXT", ] for path in case_variations: - assert not validate_project_path(path, project_path), f"Case variation '{path}' should be blocked" + assert not validate_project_path(path, project_path), ( + f"Case variation '{path}' should be blocked" + ) def test_symbolic_link_behavior(self, tmp_path): """Test behavior with symbolic links (if supported by filesystem).""" project_path = tmp_path / "project" project_path.mkdir() - + # Create a directory outside the project outside_dir = tmp_path / "outside" outside_dir.mkdir() - + try: # Try to create a symlink inside the project pointing outside symlink_path = project_path / "symlink" symlink_path.symlink_to(outside_dir) - + # Paths through symlinks should be handled safely result = validate_project_path("symlink/file.txt", project_path) # The result can vary based on how pathlib handles symlinks, # but it shouldn't crash and should be a boolean assert isinstance(result, bool) - + except (OSError, NotImplementedError): # Symlinks might not be supported on this filesystem pytest.skip("Symbolic links not supported on this filesystem") @@ -325,24 +341,25 @@ class TestValidateProjectPathPerformance: # Test a mix of safe and dangerous paths test_paths = [] - + # Add safe paths for i in range(100): test_paths.append(f"folder{i}/file{i}.txt") - + # Add dangerous paths for i in range(100): test_paths.append(f"../../../etc/passwd{i}") import time + start_time = time.time() - + for path in test_paths: result = validate_project_path(path, project_path) assert isinstance(result, bool) - + end_time = time.time() - + # Should complete reasonably quickly (adjust threshold as needed) assert end_time - start_time < 1.0, "Path validation should be fast" @@ -354,21 +371,21 @@ class TestValidateProjectPathIntegration: """Test validation with actual files and directories.""" project_path = tmp_path / "project" project_path.mkdir() - + # Create some actual files and directories (project_path / "notes").mkdir() (project_path / "docs").mkdir() (project_path / "notes" / "meeting.md").write_text("# Meeting Notes") (project_path / "docs" / "readme.txt").write_text("README") - + # Test accessing existing files assert validate_project_path("notes/meeting.md", project_path) assert validate_project_path("docs/readme.txt", project_path) - + # Test accessing non-existent but safe paths assert validate_project_path("notes/new-file.md", project_path) assert validate_project_path("new-folder/file.txt", project_path) - + # Test that attacks are still blocked even with real filesystem assert not validate_project_path("../../../etc/passwd", project_path) assert not validate_project_path("notes/../../../etc/passwd", project_path) @@ -379,18 +396,18 @@ class TestValidateProjectPathIntegration: base_path = tmp_path / "workspace" project_path = base_path / "my-project" sibling_path = base_path / "other-project" - + base_path.mkdir() project_path.mkdir() sibling_path.mkdir() - + # Create a sensitive file in the sibling directory (sibling_path / "secrets.txt").write_text("secret data") - + # Try to access the sibling directory through traversal attack_path = "../other-project/secrets.txt" assert not validate_project_path(attack_path, project_path) - + # Verify that legitimate access within project works assert validate_project_path("my-file.txt", project_path) - assert validate_project_path("subdir/my-file.txt", project_path) \ No newline at end of file + assert validate_project_path("subdir/my-file.txt", project_path)