"""Step definitions for pyyaml_security.feature. Verifies that PyYAML is pinned to a secure version (>=6.0.3) and that all YAML loading in the codebase uses safe_load to prevent arbitrary code execution (CVE-2017-18342 mitigation, issue #9055). """ from __future__ import annotations from importlib.metadata import version as pkg_version from typing import Any from behave import given, then, when from behave.runner import Context from cleveragents.actor.yaml_loader import load_yaml_text def _version_tuple(version_str: str) -> tuple[int, ...]: """Convert a version string like '6.0.3' to a comparable tuple.""" return tuple(int(part) for part in version_str.split(".")[:3]) # ── PyYAML version constraint ──────────────────────────────────────── @given("the PyYAML package is installed") def step_pyyaml_installed(context: Context) -> None: """Verify PyYAML is importable and its version is accessible.""" import yaml # noqa: F401 — presence check context.pyyaml_version = pkg_version("pyyaml") @when("I check the installed PyYAML version") def step_check_pyyaml_version(context: Context) -> None: """Record the installed PyYAML version tuple for assertion.""" context.pyyaml_version_tuple = _version_tuple(context.pyyaml_version) @then("the version should be at least 6.0.3") def step_assert_pyyaml_version(context: Context) -> None: """Assert that the installed PyYAML version is >= 6.0.3.""" minimum = (6, 0, 3) assert context.pyyaml_version_tuple >= minimum, ( f"PyYAML version {context.pyyaml_version} is below the required " f"minimum 6.0.3. Upgrade PyYAML to address CVE-2017-18342." ) # ── yaml_loader safe_load enforcement ─────────────────────────────── @when('I load the YAML text "{yaml_text}" using the secure yaml_loader') def step_load_yaml_text(context: Context, yaml_text: str) -> None: """Call load_yaml_text with the given YAML text and store the result. The YAML text is passed as a quoted string in the step, with ``\\n`` escape sequences representing literal newlines. """ normalised = yaml_text.replace("\\n", "\n") context.load_yaml_result: dict[str, Any] = load_yaml_text(normalised) @then('the secure yaml_loader result for key "{key}" should be "{value}"') def step_assert_yaml_result_key(context: Context, key: str, value: str) -> None: """Assert that the parsed YAML result contains the expected key/value pair.""" result: dict[str, Any] = context.load_yaml_result assert key in result, ( f"Expected key '{key}' in load_yaml_text result, but it was not found. " f"Actual keys: {list(result.keys())}" ) assert result[key] == value, ( f"Expected result['{key}'] == '{value}', but got '{result[key]}'." ) @when("I call load_yaml_text with unsafe YAML containing a Python object tag") def step_load_unsafe_yaml(context: Context) -> None: """Attempt to load YAML with a Python object constructor tag. PyYAML's safe_load rejects ``!!python/object`` tags, raising an error. This verifies that the yaml_loader does not use the unsafe ``yaml.load`` with the default Loader (which would execute arbitrary Python code). """ context.caught_error = None unsafe_yaml = "!!python/object/apply:os.system ['echo pwned']" try: load_yaml_text(unsafe_yaml) except Exception as exc: context.caught_error = exc @then("a pyyaml security ValueError should be raised") def step_assert_value_error_raised(context: Context) -> None: """Assert that an error was raised when loading unsafe YAML.""" assert context.caught_error is not None, ( "Expected an error when loading YAML with Python object tags, " "but none was raised. This may indicate unsafe YAML loading is in use." )