fix(engine): contain a relative sub-workflow path before it is stat'ed

dw/workflow.py

@@ -75,6 +75,7 @@
 )
 from .workflow_sources import (
     builtin_root,
+    catalog_root,
     resolve_sub_workflow,
     SubWorkflowNotFound,
 )
@@ -188,16 +189,10 @@ def catalog_root_dir(file_spec):
     workflow_dir of its own (an unconfined CLI run) - the same "last
     'workflows' segment" rule workflow_output_subfolder uses for output
     naming, but returning the directory itself rather than what sits under
-    it.
+    it. It is `catalog_root` asked for a file rather than a directory, so
+    the resolver (dw/workflow_sources.py) confines to exactly this root.
     """
-    directory = os.path.dirname(os.path.abspath(file_spec))
-    parts = os.path.normpath(directory).split(os.sep)
-    try:
-        index = len(parts) - 1 - parts[::-1].index("workflows")
-    except ValueError:
-        return directory
-
-    return os.sep.join(parts[: index + 1])
+    return catalog_root(os.path.dirname(os.path.abspath(file_spec)))
 
 
 def pipeline_cache_key(pipeline_definition):

dw/workflow_sources.py

@@ -39,6 +39,25 @@ def builtin_root():
     return os.path.join(os.path.dirname(os.path.abspath(__file__)), "workflows")
 
 
+def catalog_root(directory):
+    """The nearest ancestor of `directory` literally named 'workflows', else
+    the directory itself.
+
+    The root a run with no workflow_dir of its own confines a relative
+    sub-workflow reference to, so a template under templates/ can still climb
+    to a sibling models/ without leaving the catalog. `catalog_root_dir`
+    (dw/workflow.py) is this rule asked for a file rather than a directory.
+    """
+    directory = os.path.normpath(os.path.abspath(directory))
+    parts = directory.split(os.sep)
+    try:
+        index = len(parts) - 1 - parts[::-1].index("workflows")
+    except ValueError:
+        return directory
+
+    return os.sep.join(parts[: index + 1])
+
+
 class WorkflowSource:
     """One root on the search path."""
 
@@ -223,7 +242,10 @@ def resolve_sub_workflow(path, base_dir, confine_to):
     An absolute path is taken as written and confined to whichever root
     holds it, so the sandbox still refuses one that belongs to no source.
 
-    Raises SubWorkflowNotFound, naming every candidate it looked at.
+    A relative path that climbs out of the root it will be handed back with is
+    a PathTraversalError rather than a SubWorkflowNotFound - a refusal, not a
+    name that was absent. Otherwise raises SubWorkflowNotFound, naming every
+    candidate it looked at.
     """
     roots = []
     if confine_to:
@@ -245,8 +267,26 @@ def resolve_sub_workflow(path, base_dir, confine_to):
         return candidate, confine_to
 
     if base_dir:
+        # The root this candidate would be handed back with: the caller's
+        # confinement when it named one, else the catalog root the run itself
+        # would confine to. Containment is checked before the stat, so a name
+        # that climbs out of the catalog is never even looked at - which is
+        # what the callers' own validate_workflow_path caught a step too late,
+        # leaving this `os.path.isfile` the dw/path-injection query's only
+        # unguarded sink. A climb out is refused rather than reported as
+        # absent, the distinction the search path below keeps: the
+        # PathTraversalError propagates to the caller
+        root = confine_to or catalog_root(base_dir)
         for name in _candidate_names(path):
-            candidate = os.path.normpath(os.path.join(base_dir, name))
+            # normpath first: validate_path refuses a '..' outright, so a
+            # climb that stays inside the root has to be collapsed before it
+            # is judged. Its return value is what gets stat'ed - and being
+            # the validator's own, it is contained by construction
+            candidate = validate_path(
+                os.path.normpath(os.path.join(base_dir, name)),
+                root,
+                allow_create=True,
+            )
             tried.append(candidate)
             if os.path.isfile(candidate):
                 return candidate, confine_to

tests/test_realize.py

@@ -305,3 +305,40 @@ def test_a_child_outside_the_confinement_reads_as_none(self, tmp_path):
             read_sub_workflow("../outside/child.json", str(confined), str(confined))
             is None
         )
+
+    def test_a_child_climbing_out_of_the_catalog_reads_as_none_unconfined(
+        self, tmp_path
+    ):
+        """An unconfined run (no workflow_dir - a bare CLI run) still confines
+        a relative reference to the catalog root, so the read refuses the same
+        climb the run does rather than reaching outside it - the read the
+        unguarded stat used to allow."""
+        from dw.realize import read_sub_workflow
+
+        catalog = tmp_path / "workflows"
+        (catalog / "templates").mkdir(parents=True)
+        (tmp_path / "Outside.json").write_text(json.dumps({"id": "c", "steps": []}))
+
+        assert (
+            read_sub_workflow("../../Outside.json", str(catalog / "templates"), None)
+            is None
+        )
+
+    def test_a_child_climbing_to_a_sibling_catalog_folder_still_reads(self, tmp_path):
+        """The confinement is the catalog root, not the referencing file's own
+        directory - '../models/x.json' from templates/ is the form every
+        template uses, and stays readable."""
+        from dw.realize import read_sub_workflow
+
+        catalog = tmp_path / "workflows"
+        (catalog / "templates").mkdir(parents=True)
+        (catalog / "models").mkdir()
+        (catalog / "models" / "Child.json").write_text(
+            json.dumps({"id": "c", "steps": []})
+        )
+
+        raw = read_sub_workflow(
+            "../models/Child.json", str(catalog / "templates"), None
+        )
+
+        assert json.loads(raw) == {"id": "c", "steps": []}

tests/test_workflow_sources.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+from dw.security import SecurityError
 from dw.workflow_sources import (
     BUILTIN_ORIGIN,
     EXAMPLES_ORIGIN,
@@ -14,6 +15,7 @@
     find_workflow,
     listing,
     resolve_in_source,
+    resolve_sub_workflow,
     source_for_path,
     workflow_names,
     workflow_sources,
@@ -109,6 +111,57 @@ def test_a_path_knows_which_source_it_belongs_to(self, roots, tmp_path):
         assert source_for_path(sources, str(tmp_path / "elsewhere.json")) is None
 
 
+class TestSubWorkflowResolution:
+    """A composed step's relative path is confined to the root it is handed
+    back with, so a name that climbs out of the catalog is never resolved -
+    and never stat'ed, which the dw/path-injection query flagged here. The
+    '../models/x.json' form a template uses to reach a sibling catalog folder
+    still resolves."""
+
+    @pytest.fixture
+    def catalog(self, tmp_path):
+        """workflows/templates/ beside workflows/models/, and a decoy outside
+        the catalog that a climbing name reaches."""
+        root = tmp_path / "workflows"
+        (root / "templates").mkdir(parents=True)
+        (root / "models").mkdir()
+        (root / "models" / "Child.json").write_text(json.dumps({"id": "child"}))
+        outside = tmp_path / "Outside.json"
+        outside.write_text(json.dumps({"id": "outside"}))
+        return root, outside
+
+    def test_a_climb_inside_the_confinement_resolves(self, catalog):
+        root, _outside = catalog
+        candidate, confine_to = resolve_sub_workflow(
+            "../models/Child.json", str(root / "templates"), str(root)
+        )
+        assert os.path.basename(candidate) == "Child.json"
+        assert confine_to == str(root)
+
+    def test_a_climb_out_of_the_confinement_is_refused_not_resolved(self, catalog):
+        root, outside = catalog
+        assert os.path.isfile(outside), "the decoy has to exist to be reachable"
+        with pytest.raises(SecurityError):
+            resolve_sub_workflow(
+                "../../Outside.json", str(root / "templates"), str(root)
+            )
+
+    def test_an_unconfined_caller_still_confines_to_the_catalog(self, catalog):
+        """No confine_to (a bare CLI run) confines to the catalog root the
+        run itself would use - the nearest ancestor named 'workflows' - so
+        the climb out is refused rather than stat'ed."""
+        root, _outside = catalog
+        with pytest.raises(SecurityError):
+            resolve_sub_workflow("../../Outside.json", str(root / "templates"), None)
+
+    def test_an_unconfined_caller_may_climb_to_a_sibling_catalog_folder(self, catalog):
+        root, _outside = catalog
+        candidate, _confine_to = resolve_sub_workflow(
+            "../models/Child.json", str(root / "templates"), None
+        )
+        assert os.path.basename(candidate) == "Child.json"
+
+
 class TestNames:
     def test_names_are_relative_and_slash_separated(self, roots):
         _workspace, examples = roots