diff --git a/src/dfe_engine/schema/schema_loader.py b/src/dfe_engine/schema/schema_loader.py index e654843a..c0bdd742 100644 --- a/src/dfe_engine/schema/schema_loader.py +++ b/src/dfe_engine/schema/schema_loader.py @@ -260,11 +260,31 @@ def profile_file_stem(profile_name: str) -> str | None: return stem or None -def _profile_yaml_path(directory: Path, stem: str) -> Path: +def _contained_under(root: Path, candidate: Path, profile_name: str) -> Path: + """Return *candidate* with symlinks resolved, refusing it unless it stays under *root*. + + ``realpath`` plus a prefix check is the containment shape CodeQL's + path-injection query recognises, and resolving both sides also refuses a + symlink inside the tree that points out of it. + + Raises: + SchemaLoadError: The resolved candidate lies outside *root*. + """ + real_root = os.path.join(os.path.realpath(root), "") + real_candidate = os.path.realpath(candidate) + if not real_candidate.startswith(real_root): + raise SchemaLoadError(f"Profile {profile_name!r} resolves outside {root}") + return Path(real_candidate) + + +def _profile_yaml_path( + directory: Path, stem: str, profile_name: str, root: Path | None = None +) -> Path: + """``/.yaml``, contained under *root* (default *directory*).""" cleaned = _strip_yaml_suffix(stem).strip().strip("/") if not cleaned: raise SchemaLoadError("Profile name is empty") - return directory / f"{cleaned}.yaml" + return _contained_under(root or directory, directory / f"{cleaned}.yaml", profile_name) def _resolve_profile_yaml_path( @@ -277,7 +297,7 @@ def _resolve_profile_yaml_path( raise SchemaLoadError(f"Profile {profile_name!r} is empty") if profiles_dir is not None: - return _profile_yaml_path(Path(profiles_dir), stem) + return _profile_yaml_path(Path(profiles_dir), stem, profile_name) normalized = profile_name.replace("\\", "/").strip() # Traversal is already rejected in profile_file_stem; keep the same @@ -288,9 +308,9 @@ def candidate_under(root: Path) -> Path: if normalized.startswith(_COMMON_HEADER_PREFIX) or "/" in normalized: parts = [p for p in normalized.split("/") if p] if len(parts) == 1: - return _profile_yaml_path(root, parts[0]) - return _profile_yaml_path(root.joinpath(*parts[:-1]), parts[-1]) - return _profile_yaml_path(root / _COMMON_HEADER_SUBDIR, stem) + return _profile_yaml_path(root, parts[0], profile_name) + return _profile_yaml_path(root.joinpath(*parts[:-1]), parts[-1], profile_name, root) + return _profile_yaml_path(root / _COMMON_HEADER_SUBDIR, stem, profile_name, root) # Same as ``_resolve_subdir``: DFE_ENGINE_SCHEMAS_DIR may be a partial tree # (common-header only, no manifest.yaml). ``_resolve_schemas_root`` @@ -307,7 +327,7 @@ def candidate_under(root: Path) -> Path: if candidate.exists(): return candidate - return _profile_yaml_path(_resolve_profiles_dir(), stem) + return _profile_yaml_path(_resolve_profiles_dir(), stem, profile_name) def current_profile_version(profile_name: str) -> str: diff --git a/tests/unit/test_schema/test_schema_loader.py b/tests/unit/test_schema/test_schema_loader.py index a587dde3..8a98f596 100644 --- a/tests/unit/test_schema/test_schema_loader.py +++ b/tests/unit/test_schema/test_schema_loader.py @@ -148,6 +148,47 @@ def test_a_profile_ref_cannot_climb_out_of_the_profiles_directory(self, traversa with pytest.raises(SchemaLoadError, match="must not contain"): _resolve_profile_yaml_path(traversal) + @pytest.fixture + def schemas_tree(self, tmp_path, monkeypatch): + """A complete schemas tree under DFE_ENGINE_SCHEMAS_DIR, an outside file and a symlink to it.""" + root = tmp_path / "schemas" + header = root / "common-header" + header.mkdir(parents=True) + (root / "manifest.yaml").write_text("{}\n", encoding="utf-8") + (header / "minimal.yaml").write_text("columns: []\n", encoding="utf-8") + outside = tmp_path / "outside.yaml" + outside.write_text("columns: []\n", encoding="utf-8") + (header / "escape.yaml").symlink_to(outside) + monkeypatch.setenv("DFE_ENGINE_SCHEMAS_DIR", str(root)) + return root + + @pytest.mark.parametrize("ref", ["minimal", "minimal.yaml", "common-header/minimal"]) + def test_a_normal_profile_ref_resolves_inside_the_tree(self, schemas_tree, ref): + from dfe_engine.schema.schema_loader import _resolve_profile_yaml_path + + expected = (schemas_tree / "common-header" / "minimal.yaml").resolve() + assert _resolve_profile_yaml_path(ref) == expected + + @pytest.mark.parametrize("ref", ["escape", "common-header/escape"]) + @pytest.mark.parametrize("explicit_dir", [False, True]) + def test_a_symlink_out_of_the_tree_is_refused(self, schemas_tree, ref, explicit_dir): + from dfe_engine.schema.schema_loader import _resolve_profile_yaml_path + + profiles_dir = schemas_tree / "common-header" if explicit_dir else None + with pytest.raises(SchemaLoadError, match=r"'.*escape' resolves outside"): + _resolve_profile_yaml_path(ref, profiles_dir) + + @pytest.mark.parametrize("ref", ["/etc/passwd", "//etc/passwd", "\\etc\\passwd"]) + @pytest.mark.parametrize("explicit_dir", [False, True]) + def test_an_absolute_looking_ref_stays_inside_the_tree(self, schemas_tree, ref, explicit_dir): + """A leading separator is dropped, so the ref is rebuilt under the tree, never at /etc.""" + from dfe_engine.schema.schema_loader import _resolve_profile_yaml_path + + profiles_dir = schemas_tree / "common-header" if explicit_dir else None + resolved = _resolve_profile_yaml_path(ref, profiles_dir) + assert resolved.is_relative_to(schemas_tree.resolve()) + assert not resolved.exists() + def test_load_timeseries(self): columns = SchemaLoader.load_profile("timeseries") names = [c.name for c in columns]