Hardening: use unsafe for ansible vars, ensure API use of JinjaTurtle uses safe XML parsing, avoid symlinks

src/jinjaturtle/cli.py

@@ -16,6 +16,7 @@ from .core import (
 
 from .multi import process_directory
 from .safety import TemplateSafetyError
+from .output_safety import OutputPathError, ensure_safe_directory, write_text_safely
 
 
 def _build_arg_parser() -> argparse.ArgumentParser:
@@ -72,6 +73,9 @@ def _main(argv: list[str] | None = None) -> int:
             f"jinjaturtle: refusing to generate unsafe template: {exc}", file=sys.stderr
         )
         return 2
+    except OutputPathError as exc:
+        print(f"jinjaturtle: refusing unsafe output path: {exc}", file=sys.stderr)
+        return 2
 
 
 def _run(argv: list[str] | None = None) -> int:
@@ -89,7 +93,7 @@ def _run(argv: list[str] | None = None) -> int:
 
         # Write defaults
         if args.defaults_output:
-            Path(args.defaults_output).write_text(defaults_yaml, encoding="utf-8")
+            write_text_safely(Path(args.defaults_output), defaults_yaml)
         else:
             print("# defaults/main.yml")
             print(defaults_yaml, end="")
@@ -100,12 +104,12 @@ def _run(argv: list[str] | None = None) -> int:
         if args.template_output:
             out_path = Path(args.template_output)
             if len(outputs) == 1 and not out_path.is_dir():
-                out_path.write_text(outputs[0].template, encoding="utf-8")
+                write_text_safely(out_path, outputs[0].template)
             else:
-                out_path.mkdir(parents=True, exist_ok=True)
+                ensure_safe_directory(out_path)
                 for o in outputs:
-                    (out_path / f"config.{o.fmt}.{template_ext}").write_text(
+                    write_text_safely(
-                        o.template, encoding="utf-8"
+                        out_path / f"config.{o.fmt}.{template_ext}", o.template
                     )
         else:
             for o in outputs:
@@ -144,13 +148,13 @@ def _run(argv: list[str] | None = None) -> int:
     )
 
     if args.defaults_output:
-        Path(args.defaults_output).write_text(ansible_yaml, encoding="utf-8")
+        write_text_safely(Path(args.defaults_output), ansible_yaml)
     else:
         print("# defaults/main.yml")
         print(ansible_yaml, end="")
 
     if args.template_output:
-        Path(args.template_output).write_text(template_str, encoding="utf-8")
+        write_text_safely(Path(args.template_output), template_str)
     else:
         print(f"# config.{j2.TEMPLATE_EXTENSION}")
         print(template_str, end="")

src/jinjaturtle/core.py

@@ -33,6 +33,22 @@ class QuotedString(str):
     pass
 
 
+class AnsibleUnsafeString(str):
+    """Marker type emitted with Ansible's !unsafe YAML tag.
+
+    Ansible recursively templates string values by default.  Source-derived
+    config values that contain Jinja delimiters must therefore be marked
+    unsafe in defaults/main.yml, otherwise a harvested value such as
+    ``{{ lookup('pipe', 'id') }}`` becomes executable on the Ansible
+    controller when the generated role is applied.
+    """
+
+    pass
+
+
+_JINJA_STARTS = ("{{", "{%", "{#")
+
+
 def _fallback_str_representer(dumper: yaml.SafeDumper, data: Any):
     """
     Fallback for objects the dumper doesn't know about.
@@ -52,7 +68,33 @@ def _quoted_str_representer(dumper: yaml.SafeDumper, data: QuotedString):
     return dumper.represent_scalar("tag:yaml.org,2002:str", str(data), style='"')
 
 
+def _ansible_unsafe_str_representer(dumper: yaml.SafeDumper, data: AnsibleUnsafeString):
+    return dumper.represent_scalar("!unsafe", str(data), style="'")
+
+
+def _needs_ansible_unsafe(value: str) -> bool:
+    return any(marker in value for marker in _JINJA_STARTS)
+
+
+def _mark_ansible_unsafe_values(obj: Any) -> Any:
+    """Recursively mark mapping/list values containing Jinja as !unsafe.
+
+    Mapping keys are intentionally left alone: they are variable names or YAML
+    structure, not Ansible-templated values.  Values nested in folder-mode item
+    lists, including source-derived ``id`` values, are protected.
+    """
+
+    if isinstance(obj, dict):
+        return {k: _mark_ansible_unsafe_values(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [_mark_ansible_unsafe_values(v) for v in obj]
+    if isinstance(obj, str) and _needs_ansible_unsafe(obj):
+        return AnsibleUnsafeString(obj)
+    return obj
+
+
 _TurtleDumper.add_representer(QuotedString, _quoted_str_representer)
+_TurtleDumper.add_representer(AnsibleUnsafeString, _ansible_unsafe_str_representer)
 # Use our fallback for any unknown object types
 _TurtleDumper.add_representer(None, _fallback_str_representer)
 
@@ -84,8 +126,9 @@ def dump_yaml(data: Any, *, sort_keys: bool = True) -> str:
 
     This is used by both the single-file and multi-file code paths.
     """
+    safe_data = _mark_ansible_unsafe_values(data)
     return yaml.dump(
-        data,
+        safe_data,
         Dumper=_TurtleDumper,
         sort_keys=sort_keys,
         default_flow_style=False,

src/jinjaturtle/handlers/xml.py

@@ -3,7 +3,8 @@ from __future__ import annotations
 from collections import Counter, defaultdict
 from pathlib import Path
 from typing import Any
-import xml.etree.ElementTree as ET  # nosec
+import xml.etree.ElementTree as ET  # nosec B405 - safe trees only; parsing uses defusedxml
+import defusedxml.ElementTree as DET
 
 from .base import BaseHandler
 from .. import j2
@@ -20,11 +21,11 @@ class XmlHandler(BaseHandler):
 
     def parse(self, path: Path) -> ET.Element:
         text = path.read_text(encoding="utf-8")
-        parser = ET.XMLParser(
+        # Security must live in the handler, not only in the CLI entry point:
-            target=ET.TreeBuilder(insert_comments=False)
+        # callers may import JinjaTurtle as a library and invoke parse_config()
-        )  # nosec B314
+        # directly.  defusedxml rejects DTD/entity abuse and also discards
-        parser.feed(text)
+        # comments by default, matching the previous TreeBuilder behaviour.
-        root = parser.close()
+        root = DET.fromstring(text)
         return root
 
     def flatten(self, parsed: Any) -> list[tuple[tuple[str, ...], Any]]:

src/jinjaturtle/multi.py

@@ -22,7 +22,9 @@ Notes:
 
 from collections import Counter, defaultdict
 from copy import deepcopy
+import os
 import configparser
+import stat
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Iterable
@@ -44,8 +46,23 @@ SUPPORTED_SUFFIXES: dict[str, set[str]] = {
 }
 
 
+def _lstat(path: Path) -> os.stat_result:
+    return path.lstat()
+
+
 def is_supported_file(path: Path) -> bool:
-    if not path.is_file():
+    """Return True only for real regular files with supported suffixes.
+
+    pathlib.Path.is_file() follows symlinks.  Folder mode must not follow
+    attacker-controlled symlinks when run over an untrusted tree, especially if
+    an administrator accidentally runs the CLI as root.
+    """
+
+    try:
+        st = _lstat(path)
+    except FileNotFoundError:
+        return False
+    if not stat.S_ISREG(st.st_mode):
         return False
     suffix = path.suffix.lower()
     for exts in SUPPORTED_SUFFIXES.values():
@@ -55,11 +72,16 @@ def is_supported_file(path: Path) -> bool:
 
 
 def iter_supported_files(root: Path, recursive: bool) -> list[Path]:
-    if not root.exists():
+    try:
+        st = _lstat(root)
+    except FileNotFoundError:
         raise FileNotFoundError(str(root))
-    if root.is_file():
+
+    if stat.S_ISLNK(st.st_mode):
+        raise ValueError(f"refusing to follow symlink: {root}")
+    if stat.S_ISREG(st.st_mode):
         return [root] if is_supported_file(root) else []
-    if not root.is_dir():
+    if not stat.S_ISDIR(st.st_mode):
         return []
 
     it = root.rglob("*") if recursive else root.glob("*")

src/jinjaturtle/output_safety.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+"""Safer file-output helpers for the JinjaTurtle CLI.
+
+The CLI is often used by administrators.  A plain Path.write_text() follows a
+final-path symlink and can therefore be dangerous when a root-run invocation
+writes into an attacker-writable tree.  These helpers validate path components,
+write through a private temporary file in the target directory, and replace the
+final path atomically.  Existing final-path symlinks are refused rather than
+followed.
+"""
+
+import os
+from pathlib import Path
+import stat
+import tempfile
+
+
+class OutputPathError(OSError):
+    """Raised when a requested output path is unsafe."""
+
+
+def _absolute(path: Path) -> Path:
+    return path if path.is_absolute() else Path.cwd() / path
+
+
+def _check_existing_path_not_symlink(path: Path) -> None:
+    try:
+        st = path.lstat()
+    except FileNotFoundError:
+        return
+    if stat.S_ISLNK(st.st_mode):
+        raise OutputPathError(f"refusing to use symlink path: {path}")
+
+
+def _check_existing_output_file(path: Path) -> None:
+    try:
+        st = path.lstat()
+    except FileNotFoundError:
+        return
+    if stat.S_ISLNK(st.st_mode):
+        raise OutputPathError(f"refusing to write through symlink: {path}")
+    if not stat.S_ISREG(st.st_mode):
+        raise OutputPathError(f"refusing to replace non-regular file: {path}")
+
+
+def _check_parent_components(parent: Path) -> None:
+    """Require every existing parent component to be a real directory."""
+
+    parent = _absolute(parent)
+    parts = parent.parts
+    if not parts:
+        return
+
+    cur = Path(parts[0])
+    for part in parts[1:]:
+        cur = cur / part
+        try:
+            st = cur.lstat()
+        except FileNotFoundError as exc:
+            raise OutputPathError(f"output parent does not exist: {cur}") from exc
+        if stat.S_ISLNK(st.st_mode):
+            raise OutputPathError(f"refusing to use symlink parent: {cur}")
+        if not stat.S_ISDIR(st.st_mode):
+            raise OutputPathError(f"output parent is not a directory: {cur}")
+
+
+def ensure_safe_directory(path: Path) -> None:
+    """Create or validate a directory tree without accepting symlinks."""
+
+    path = _absolute(path)
+    parts = path.parts
+    if not parts:
+        return
+
+    cur = Path(parts[0])
+    for part in parts[1:]:
+        cur = cur / part
+        try:
+            st = cur.lstat()
+        except FileNotFoundError:
+            cur.mkdir(mode=0o700)
+            st = cur.lstat()
+        if stat.S_ISLNK(st.st_mode):
+            raise OutputPathError(f"refusing to use symlink directory: {cur}")
+        if not stat.S_ISDIR(st.st_mode):
+            raise OutputPathError(f"output path is not a directory: {cur}")
+
+
+def write_text_safely(path: Path, text: str, *, encoding: str = "utf-8") -> None:
+    """Write text without following a final-path symlink.
+
+    The target's parent must already exist and every parent component must be a
+    real directory.  The write is completed with os.replace(), which atomically
+    swaps the final directory entry and does not dereference a final symlink.
+    """
+
+    path = _absolute(path)
+    _check_parent_components(path.parent)
+    _check_existing_output_file(path)
+
+    fd = -1
+    tmp_name: str | None = None
+    try:
+        fd, tmp_name = tempfile.mkstemp(
+            prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent), text=True
+        )
+        with os.fdopen(fd, "w", encoding=encoding) as f:
+            fd = -1
+            f.write(text)
+            f.flush()
+            os.fsync(f.fileno())
+        os.chmod(tmp_name, 0o600)
+        _check_existing_output_file(path)
+        os.replace(tmp_name, path)
+        tmp_name = None
+    finally:
+        if fd >= 0:
+            os.close(fd)
+        if tmp_name is not None:
+            try:
+                os.unlink(tmp_name)
+            except FileNotFoundError:
+                pass

tests/test_injection_security.py

@@ -29,6 +29,21 @@ from jinjaturtle.escape import (
 TRIP = "__TRIPWIRE_FIRED__"
 
 
+class _UnsafeAwareLoader(pyyaml.SafeLoader):
+    pass
+
+
+def _construct_unsafe(loader: _UnsafeAwareLoader, node: pyyaml.Node):
+    return loader.construct_scalar(node)
+
+
+_UnsafeAwareLoader.add_constructor("!unsafe", _construct_unsafe)
+
+
+def _safe_load_defaults(text: str):
+    return pyyaml.load(text, Loader=_UnsafeAwareLoader)
+
+
 class _Boom:
     """Returns the tripwire sentinel for any access/call an SSTI payload makes."""
 
@@ -82,7 +97,7 @@ def _run_jinjaturtle(tmp_path: Path, source_name: str, body: str, fmt: str):
         text=True,
     )
     assert res.returncode == 0, f"generation failed: {res.stderr}"
-    defaults = pyyaml.safe_load(dfl.read_text()) or {}
+    defaults = _safe_load_defaults(dfl.read_text()) or {}
     return tpl.read_text(), defaults
 
 

tests/test_security_hardening.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+from pathlib import Path
+import os
+
+import pytest
+import yaml
+from defusedxml.common import EntitiesForbidden
+
+from jinjaturtle import cli
+from jinjaturtle.core import generate_ansible_yaml, parse_config, flatten_config
+from jinjaturtle.multi import is_supported_file, iter_supported_files, process_directory
+from jinjaturtle.output_safety import OutputPathError, write_text_safely
+
+
+class UnsafeAwareLoader(yaml.SafeLoader):
+    pass
+
+
+def _unsafe(loader: UnsafeAwareLoader, node: yaml.Node):
+    return loader.construct_scalar(node)
+
+
+UnsafeAwareLoader.add_constructor("!unsafe", _unsafe)
+
+
+def test_jinja_values_are_emitted_as_ansible_unsafe(tmp_path: Path):
+    src = tmp_path / "app.ini"
+    src.write_text("[main]\ncmd = {{ lookup('pipe','id') }}\n", encoding="utf-8")
+
+    fmt, parsed = parse_config(src)
+    defaults_yaml = generate_ansible_yaml("role", flatten_config(fmt, parsed))
+
+    assert "role_main_cmd: !unsafe" in defaults_yaml
+    assert "{{ lookup(''pipe'',''id'') }}" in defaults_yaml
+    loaded = yaml.load(defaults_yaml, Loader=UnsafeAwareLoader)
+    assert loaded["role_main_cmd"] == "{{ lookup('pipe','id') }}"
+
+
+def test_folder_mode_marks_nested_jinja_values_and_ids_unsafe(tmp_path: Path):
+    src = tmp_path / "src"
+    src.mkdir()
+    # Filename ids are source-derived values too.
+    (src / "{{ bad }}.yaml").write_text(
+        "message: \"{{ lookup('pipe','id') }}\"\n", encoding="utf-8"
+    )
+
+    defaults_yaml, _outputs = process_directory(
+        src, recursive=False, role_prefix="role"
+    )
+
+    assert "id: !unsafe" in defaults_yaml
+    assert "role_message: !unsafe" in defaults_yaml
+    loaded = yaml.load(defaults_yaml, Loader=UnsafeAwareLoader)
+    assert loaded["role_items"][0]["id"] == "{{ bad }}.yaml"
+    assert loaded["role_items"][0]["role_message"] == "{{ lookup('pipe','id') }}"
+
+
+def test_xml_parser_rejects_entities_when_called_as_library(tmp_path: Path):
+    src = tmp_path / "bad.xml"
+    src.write_text(
+        "<!DOCTYPE root [<!ENTITY xxe SYSTEM 'file:///etc/passwd'>]><root>&xxe;</root>",
+        encoding="utf-8",
+    )
+
+    with pytest.raises(EntitiesForbidden):
+        parse_config(src, "xml")
+
+
+def test_folder_mode_does_not_follow_symlinked_files(tmp_path: Path):
+    real = tmp_path / "secret.ini"
+    real.write_text("[main]\nsecret=yes\n", encoding="utf-8")
+    root = tmp_path / "root"
+    root.mkdir()
+    link = root / "link.ini"
+    link.symlink_to(real)
+
+    assert not is_supported_file(link)
+    assert iter_supported_files(root, recursive=False) == []
+    assert iter_supported_files(root, recursive=True) == []
+
+
+@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks unavailable")
+def test_cli_refuses_to_write_through_final_symlink(tmp_path: Path):
+    target = tmp_path / "target.txt"
+    target.write_text("keep\n", encoding="utf-8")
+    link = tmp_path / "out.yml"
+    link.symlink_to(target)
+
+    with pytest.raises(OutputPathError):
+        write_text_safely(link, "replace\n")
+
+    assert target.read_text(encoding="utf-8") == "keep\n"
+    assert link.is_symlink()
+
+
+def test_cli_refuses_symlinked_output_parent(tmp_path: Path):
+    real_dir = tmp_path / "real"
+    real_dir.mkdir()
+    link_dir = tmp_path / "linkdir"
+    link_dir.symlink_to(real_dir, target_is_directory=True)
+
+    with pytest.raises(OutputPathError):
+        write_text_safely(link_dir / "out.yml", "data\n")
+
+    assert not (real_dir / "out.yml").exists()
+
+
+def test_cli_reports_unsafe_output_path_without_overwriting_symlink(tmp_path: Path):
+    cfg = tmp_path / "app.ini"
+    cfg.write_text("[main]\nname = ok\n", encoding="utf-8")
+    target = tmp_path / "target.yml"
+    target.write_text("keep\n", encoding="utf-8")
+    link = tmp_path / "defaults.yml"
+    link.symlink_to(target)
+
+    exit_code = cli._main([str(cfg), "--defaults-output", str(link)])
+
+    assert exit_code == 2
+    assert target.read_text(encoding="utf-8") == "keep\n"