Fix validation of unsafe node source paths

concore_cli/commands/validate.py

@@ -8,6 +8,22 @@
 from ._source_dir import resolve_source_dir
 
 
+def _is_safe_source_path(filename):
+    """Match the generator's safe_relpath() checks for node source paths."""
+    if not filename:
+        return False
+    normalized = filename.replace("\\", "/")
+    if re.search(r"""[\x00-\x1F\x7F*?"<>|;&`$'()]""", normalized):
+        return False
+    if normalized.startswith("/") or normalized.startswith("~"):
+        return False
+    if re.match(r"^[A-Za-z]:", normalized):
+        return False
+    if ":" in normalized:
+        return False
+    if any(part in ("", "..") for part in normalized.split("/")):
+        return False
+    return True
 def _classify_message(message, bucket_name):
     if bucket_name == "info":
         if message.startswith("Found ") and "node(s)" in message:
@@ -52,6 +68,8 @@ def _classify_message(message, bucket_name):
         return {"error_type": "missing_node_filename"}
     if message.startswith("Node '") and message.endswith("has unusual file extension"):
         return {"error_type": "unusual_file_extension"}
+    if message.startswith("Unsafe source file path:"):
+        return {"error_type": "unsafe_source_path"}
     if message.startswith("Missing source file:"):
         return {"error_type": "missing_source_file"}
     if message.startswith("Node ") and message.endswith(" has no label"):
@@ -89,8 +107,10 @@ def _build_entries(bucket_name, messages, source_nodes):
         entry = {"message": message}
         entry.update(_classify_message(message, bucket_name))
 
-        if message.startswith("Missing source file:"):
-            filename = message.split(":", 1)[1].strip()
+        if message.startswith(
+            ("Missing source file:", "Unsafe source file path:")
+        ):
+            filename = message.split(": ", 1)[1].strip()
             node_id = source_nodes.get(filename)
             if node_id:
                 entry["node_id"] = node_id
@@ -256,6 +276,10 @@ def finalize():
                             source_nodes[filename] = node_id
                             if not filename:
                                 errors.append(f"Node '{label}' has no filename")
+                            elif not _is_safe_source_path(filename):
+                                errors.append(
+                                    f"Unsafe source file path: {filename}"
+                                )
                             elif not any(
                                 filename.endswith(ext)
                                 for ext in [".py", ".cpp", ".m", ".v", ".java"]

tests/test_cli.py

@@ -102,6 +102,34 @@ def test_validate_missing_node_file(self):
             self.assertNotEqual(result.exit_code, 0)
             self.assertIn("Missing source file", result.output)
 
+    def test_validate_rejects_source_path_outside_source_dir(self):
+        with self.runner.isolated_filesystem(temp_dir=self.temp_dir):
+            result = self.runner.invoke(cli, ["init", "test-project"])
+            self.assertEqual(result.exit_code, 0)
+            workflow_path = Path("test-project/workflow.graphml")
+            graphml = workflow_path.read_text(encoding="utf-8")
+            self.assertIn("<y:NodeLabel>N1:script.py</y:NodeLabel>", graphml)
+            graphml = graphml.replace("N1:script.py", "N1:../outside.py")
+            workflow_path.write_text(graphml, encoding="utf-8")
+            Path("test-project/outside.py").write_text(
+                "print('outside')", encoding="utf-8"
+            )
+            result = self.runner.invoke(
+                cli,
+                [
+                    "validate",
+                    "test-project/workflow.graphml",
+                    "--format",
+                    "json",
+                ],
+            )
+            self.assertNotEqual(result.exit_code, 0)
+            payload = json.loads(result.output)
+            self.assertFalse(payload["valid"])
+            self.assertEqual(
+                payload["errors"][0]["error_type"], "unsafe_source_path"
+            )
+            self.assertEqual(payload["errors"][0]["node_id"], "n1")
     def test_validate_json_output_for_valid_file(self):
         with self.runner.isolated_filesystem(temp_dir=self.temp_dir):
             result = self.runner.invoke(cli, ["init", "test-project"])