From be1f9940a5b44416b03dd98b62c4b166c7d7d115 Mon Sep 17 00:00:00 2001 From: Miguel Jacq Date: Wed, 24 Jun 2026 16:17:41 +1000 Subject: [PATCH] More hardening measures --- debian/changelog | 6 + pyproject.toml | 2 +- rpm/jinjaturtle.spec | 6 +- src/jinjaturtle/__init__.py | 22 ++ src/jinjaturtle/cli.py | 16 +- src/jinjaturtle/core.py | 94 ++++++++- src/jinjaturtle/erb.py | 52 +++-- src/jinjaturtle/escape.py | 25 ++- src/jinjaturtle/handlers/base.py | 40 ++-- src/jinjaturtle/handlers/ini.py | 10 +- src/jinjaturtle/handlers/json.py | 23 +- src/jinjaturtle/handlers/postfix.py | 2 +- src/jinjaturtle/handlers/toml.py | 6 +- src/jinjaturtle/handlers/xml.py | 107 ++++++---- src/jinjaturtle/handlers/yaml.py | 11 +- src/jinjaturtle/j2.py | 22 ++ src/jinjaturtle/loop_analyzer.py | 90 +++++--- src/jinjaturtle/multi.py | 90 +++++--- src/jinjaturtle/template_safety.py | 111 ++++++++++ tests/test_injection_security.py | 313 ++++++++++++++++++++++++++++ tests/test_json_handler.py | 4 +- tests/test_multi.py | 12 ++ 22 files changed, 902 insertions(+), 162 deletions(-) create mode 100644 src/jinjaturtle/template_safety.py diff --git a/debian/changelog b/debian/changelog index d32bd4f..9173c94 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +jinjaturtle (0.5.7) unstable; urgency=medium + + * More hardening measures + + -- Miguel Jacq Wed, 24 Jun 2026 16:13:00 +1000 + jinjaturtle (0.5.6) unstable; urgency=medium * Try to prevent what could lead to execution of embedded jinja in original files when converting diff --git a/pyproject.toml b/pyproject.toml index d856172..9ec2da5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "jinjaturtle" -version = "0.5.6" +version = "0.5.7" description = "Convert config files into Ansible defaults and Jinja2 templates." authors = [ { name = "Miguel Jacq", email = "mig@mig5.net" }, diff --git a/rpm/jinjaturtle.spec b/rpm/jinjaturtle.spec index 2ed4063..051fd8b 100644 --- a/rpm/jinjaturtle.spec +++ b/rpm/jinjaturtle.spec @@ -1,4 +1,4 @@ -%global upstream_version 0.5.6 +%global upstream_version 0.5.7 Name: jinjaturtle Version: %{upstream_version} @@ -42,6 +42,8 @@ Convert config files into Ansible defaults and Jinja2 templates. %{_bindir}/jinjaturtle %changelog +* Wed Jun 24 2026 Miguel Jacq - %{version}-%{release} +- More hardening * Tue Jun 23 2026 Miguel Jacq - %{version}-%{release} - Try to prevent what could lead to execution of embedded jinja in original files when converting * Sat Jun 20 2026 Miguel Jacq - %{version}-%{release} @@ -55,7 +57,7 @@ Convert config files into Ansible defaults and Jinja2 templates. - Fix indentation problems with nested dicts * Fri Jun 19 2026 Miguel Jacq - %{version}-%{release} - Empty dicts and lists are now emitted as leaf defaults. -* Tue May 11 2026 Miguel Jacq - %{version}-%{release} +* Mon May 11 2026 Miguel Jacq - %{version}-%{release} - Support ssh configs * Tue Jan 06 2026 Miguel Jacq - %{version}-%{release} - Support converting systemd files and postfix main.cf diff --git a/src/jinjaturtle/__init__.py b/src/jinjaturtle/__init__.py index 3dd0b09..eab4263 100644 --- a/src/jinjaturtle/__init__.py +++ b/src/jinjaturtle/__init__.py @@ -3,3 +3,25 @@ from __future__ import annotations __all__ = ["__version__"] __version__ = "0.1.0" + + +def _register_yaml_unsafe_constructor() -> None: + """Let PyYAML SafeLoader read Ansible's !unsafe tag as a plain string. + + Ansible treats !unsafe specially at runtime. Ordinary PyYAML consumers such + as tests or static tooling do not know the tag by default; registering this + constructor keeps generated defaults inspectable without changing the YAML + emitted for Ansible. + """ + try: + import yaml + except Exception: # pragma: no cover - PyYAML is a required dependency + return + + def _construct_unsafe(loader, node): + return loader.construct_scalar(node) + + yaml.SafeLoader.add_constructor("!unsafe", _construct_unsafe) + + +_register_yaml_unsafe_constructor() diff --git a/src/jinjaturtle/cli.py b/src/jinjaturtle/cli.py index 0c403fa..ef80454 100644 --- a/src/jinjaturtle/cli.py +++ b/src/jinjaturtle/cli.py @@ -1,8 +1,8 @@ from __future__ import annotations import argparse +import re import sys -from defusedxml import defuse_stdlib from pathlib import Path from . import j2 @@ -19,6 +19,18 @@ from .core import ( from .multi import process_directory +_SAFE_ROLE_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +def _safe_role_name(value: str) -> str: + if not _SAFE_ROLE_RE.fullmatch(value): + raise argparse.ArgumentTypeError( + "role name must match ^[A-Za-z_][A-Za-z0-9_]*$ so it can never " + "inject template syntax" + ) + return value + + def _build_arg_parser() -> argparse.ArgumentParser: ap = argparse.ArgumentParser( prog="jinjaturtle", @@ -35,6 +47,7 @@ def _build_arg_parser() -> argparse.ArgumentParser: "-r", "--role-name", default="jinjaturtle", + type=_safe_role_name, help="Ansible role name, used as variable prefix (default: jinjaturtle).", ) ap.add_argument( @@ -76,7 +89,6 @@ def _build_arg_parser() -> argparse.ArgumentParser: def _main(argv: list[str] | None = None) -> int: - defuse_stdlib() parser = _build_arg_parser() args = parser.parse_args(argv) diff --git a/src/jinjaturtle/core.py b/src/jinjaturtle/core.py index 1faefca..7e6039e 100644 --- a/src/jinjaturtle/core.py +++ b/src/jinjaturtle/core.py @@ -9,6 +9,8 @@ import yaml from .loop_analyzer import LoopAnalyzer, LoopCandidate from .erb import puppet_class_name, puppet_local_var_name, translate_jinja2_to_erb +from .escape import contains_erb_markup, contains_jinja_markup +from .template_safety import validate_generated_template from .handlers import ( BaseHandler, IniHandler, @@ -23,8 +25,17 @@ from .handlers import ( class QuotedString(str): - """ - Marker type for strings that must be double-quoted in YAML output. + """Marker type for strings that must be double-quoted in YAML output.""" + + pass + + +class UnsafeString(str): + """Marker type for Ansible values that must never be recursively templated. + + Ansible honours the ``!unsafe`` YAML tag by loading the value as unsafe text, + preventing a later templating pass from evaluating delimiters such as + ``{{ ... }}`` or ``<%= ... %>`` that came from the original config file. """ pass @@ -49,7 +60,12 @@ def _quoted_str_representer(dumper: yaml.SafeDumper, data: QuotedString): return dumper.represent_scalar("tag:yaml.org,2002:str", str(data), style='"') +def _unsafe_str_representer(dumper: yaml.SafeDumper, data: UnsafeString): + return dumper.represent_scalar("!unsafe", str(data)) + + _TurtleDumper.add_representer(QuotedString, _quoted_str_representer) +_TurtleDumper.add_representer(UnsafeString, _unsafe_str_representer) # Use our fallback for any unknown object types _TurtleDumper.add_representer(None, _fallback_str_representer) @@ -92,6 +108,26 @@ def dump_yaml(data: Any, *, sort_keys: bool = True) -> str: ) +def mark_ansible_unsafe(obj: Any) -> Any: + """Recursively mark source strings that could be re-templated by Ansible. + + JinjaTurtle's primary template render treats source values as data, but the + generated defaults file may later be consumed by other Ansible tasks or + templates. Any source-derived string containing Jinja2 or ERB delimiters is + therefore emitted with Ansible's ``!unsafe`` tag so it cannot become live + template code in a recursive templating context. + """ + if isinstance(obj, str): + if contains_jinja_markup(obj) or contains_erb_markup(obj): + return UnsafeString(obj) + return obj + if isinstance(obj, list): + return [mark_ansible_unsafe(v) for v in obj] + if isinstance(obj, dict): + return {k: mark_ansible_unsafe(v) for k, v in obj.items()} + return obj + + def make_var_name(role_prefix: str, path: Iterable[str]) -> str: """ Wrapper for :meth:`BaseHandler.make_var_name`. @@ -337,6 +373,37 @@ def _path_starts_with(path: tuple[str, ...], prefix: tuple[str, ...]) -> bool: return path[: len(prefix)] == prefix +def _path_label(path: tuple[str, ...]) -> str: + return ".".join(path) if path else "" + + +def _raise_on_variable_name_collisions( + role_prefix: str, + flat_items: list[tuple[tuple[str, ...], Any]], + loop_candidates: list[LoopCandidate] | None = None, +) -> None: + """Fail closed when distinct source paths collapse to one variable name.""" + seen: dict[str, list[str]] = {} + for path, _value in flat_items: + seen.setdefault(make_var_name(role_prefix, path), []).append(_path_label(path)) + if loop_candidates: + for candidate in loop_candidates: + seen.setdefault(make_var_name(role_prefix, candidate.path), []).append( + _path_label(candidate.path) + ) + + collisions = {name: paths for name, paths in seen.items() if len(set(paths)) > 1} + if collisions: + details = "; ".join( + f"{name}: {', '.join(paths)}" for name, paths in sorted(collisions.items()) + ) + raise ValueError( + "Refusing to generate templates because multiple config paths map to " + f"the same variable name ({details}). Rename one of the keys or use " + "separate role names." + ) + + def generate_ansible_yaml( role_prefix: str, flat_items: list[tuple[tuple[str, ...], Any]], @@ -345,18 +412,20 @@ def generate_ansible_yaml( """ Create Ansible YAML for defaults/main.yml. """ + _raise_on_variable_name_collisions(role_prefix, flat_items, loop_candidates) + defaults: dict[str, Any] = {} # Add scalar variables for path, value in flat_items: var_name = make_var_name(role_prefix, path) - defaults[var_name] = value # No normalization - keep original types + defaults[var_name] = mark_ansible_unsafe(value) # keep type, tag unsafe strings # Add loop collections if loop_candidates: for candidate in loop_candidates: var_name = make_var_name(role_prefix, candidate.path) - defaults[var_name] = candidate.items + defaults[var_name] = mark_ansible_unsafe(candidate.safe_items()) return dump_yaml(defaults, sort_keys=True) @@ -378,14 +447,17 @@ def generate_jinja2_template( # Check if handler supports loop-aware generation if hasattr(handler, "generate_jinja2_template_with_loops") and loop_candidates: - return handler.generate_jinja2_template_with_loops( + template = handler.generate_jinja2_template_with_loops( parsed, role_prefix, original_text, loop_candidates ) + else: + # Fallback to original scalar-only generation + template = handler.generate_jinja2_template( + parsed, role_prefix, original_text=original_text + ) - # Fallback to original scalar-only generation - return handler.generate_jinja2_template( - parsed, role_prefix, original_text=original_text - ) + validate_generated_template(template) + return template def _template_variable_names( @@ -414,6 +486,8 @@ def generate_puppet_hiera_yaml( they are the same, so ``php_memory_limit`` becomes ``php::memory_limit``. """ + _raise_on_variable_name_collisions(role_prefix, flat_items, loop_candidates) + klass = puppet_class_name(puppet_class or role_prefix) data: dict[str, Any] = {} @@ -426,7 +500,7 @@ def generate_puppet_hiera_yaml( for candidate in loop_candidates: generated = make_var_name(role_prefix, candidate.path) local = puppet_local_var_name(role_prefix, generated, puppet_class=klass) - data[f"{klass}::{local}"] = candidate.items + data[f"{klass}::{local}"] = candidate.safe_items() return dump_yaml(data, sort_keys=True) diff --git a/src/jinjaturtle/erb.py b/src/jinjaturtle/erb.py index 6eb26ad..ec133a1 100644 --- a/src/jinjaturtle/erb.py +++ b/src/jinjaturtle/erb.py @@ -4,6 +4,9 @@ import re from .escape import escape_erb_literal +_SAFE_GENERATED_FIELD_RE = re.compile(r"^jt_[A-Za-z0-9_]+_[0-9a-f]{8}$") +_REF_RE = r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*" + def _safe_name(raw: str, *, fallback: str = "var") -> str: text = re.sub(r"[^A-Za-z0-9_]+", "_", str(raw or fallback)).strip("_").lower() @@ -128,24 +131,41 @@ class ErbTranslator: return "false" if expr in {"none", "None", "null"}: return "nil" - if re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", expr): - if any(expr == loop_var for loop_var, _idx, _coll in self.loop_stack): - return expr - return f"@{self.local_var(expr)}" - m = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)\.([A-Za-z_][A-Za-z0-9_]*)$", expr) - if m: - base, key = m.groups() + + if not re.match(rf"^{_REF_RE}$", expr): + raise ValueError( + f"Unsupported JinjaTurtle expression for ERB translation: {expr}" + ) + + parts = expr.split(".") + base = parts[0] + attrs = parts[1:] + + if not attrs: if any(base == loop_var for loop_var, _idx, _coll in self.loop_stack): - return f"{base}[{key!r}]" - return f"@{self.local_var(base)}[{key!r}]" - return expr + return base + return f"@{self.local_var(base)}" + + if any(base == loop_var for loop_var, _idx, _coll in self.loop_stack): + ruby = base + else: + ruby = f"@{self.local_var(base)}" + + for attr in attrs: + if not _SAFE_GENERATED_FIELD_RE.match(attr): + raise ValueError( + "Unsafe attribute access in generated template during ERB " + f"translation: {expr}" + ) + ruby = f"{ruby}[{attr!r}]" + return ruby def expr_to_ruby(self, expr: str) -> str: expr = expr.strip() # JinjaTurtle emits these YAML-preserving ternaries for booleans/nulls. m = re.match( - r"^(['\"])(true|false)\1\s+if\s+([A-Za-z_][A-Za-z0-9_\.]*)\s+else\s+(['\"])(true|false)\4$", + rf"^(['\"])(true|false)\1\s+if\s+({_REF_RE})\s+else\s+(['\"])(true|false)\4$", expr, ) if m: @@ -155,7 +175,7 @@ class ErbTranslator: return f"{cond} ? {truthy!r} : {falsy!r}" m = re.match( - r"^(['\"])(null)\1\s+if\s+([A-Za-z_][A-Za-z0-9_\.]*)\s+is\s+none\s+else\s+([A-Za-z_][A-Za-z0-9_\.]*)$", + rf"^(['\"])(null)\1\s+if\s+({_REF_RE})\s+is\s+none\s+else\s+({_REF_RE})$", expr, ) if m: @@ -175,6 +195,10 @@ class ErbTranslator: ruby = f"JSON.pretty_generate({ruby})" else: ruby = f"JSON.generate({ruby})" + else: + raise ValueError( + f"Unsupported JinjaTurtle filter for ERB translation: {filt}" + ) return ruby return self.ruby_value(expr) @@ -205,10 +229,10 @@ class ErbTranslator: if cond == "not loop.last" and self.loop_stack: _loop_var, idx_var, collection_ruby = self.loop_stack[-1] return f"<% if {idx_var} < ({collection_ruby}.length - 1) -%>" - m = re.match(r"^([A-Za-z_][A-Za-z0-9_\.]*)\s+is\s+defined$", cond) + m = re.match(rf"^({_REF_RE})\s+is\s+defined$", cond) if m: return f"<% unless {self.ruby_value(m.group(1))}.nil? -%>" - m = re.match(r"^([A-Za-z_][A-Za-z0-9_\.]*)\s+is\s+none$", cond) + m = re.match(rf"^({_REF_RE})\s+is\s+none$", cond) if m: return f"<% if {self.ruby_value(m.group(1))}.nil? -%>" return f"<% if {self.expr_to_ruby(cond)} -%>" diff --git a/src/jinjaturtle/escape.py b/src/jinjaturtle/escape.py index 3b8575e..1dfd479 100644 --- a/src/jinjaturtle/escape.py +++ b/src/jinjaturtle/escape.py @@ -95,13 +95,15 @@ def _defang_endraw(text: str) -> str: def escape_jinja_literal(text: str) -> str: - """Make *text* render as literal characters under a later Jinja2 render. + """Make *text* render as literal characters under later template renders. - Text with no Jinja metacharacters is returned unchanged so the common case - stays byte-for-byte identical to the source. Otherwise the text is wrapped - in a single ``{% raw %}`` block, with any embedded ``endraw`` defanged. + JinjaTurtle's ERB templates are produced by translating the generated Jinja2 + template. Therefore source text that contains *either* Jinja2 or ERB + delimiters must be carried through as a Jinja raw block: a later Jinja2 render + will print it literally, and the ERB translator will recognise the raw block + and escape any ERB delimiters inside it. """ - if not text or not contains_jinja_markup(text): + if not text or not (contains_jinja_markup(text) or contains_erb_markup(text)): return text return "{% raw %}" + _defang_endraw(text) + "{% endraw %}" @@ -109,9 +111,11 @@ def escape_jinja_literal(text: str) -> str: def escape_erb_literal(text: str) -> str: """Make *text* render as literal characters under a later ERB render. - ERB has no ``raw`` block, so each opening/closing delimiter is rewritten as - an ERB expression that prints the delimiter literally. Text with no ERB - metacharacters is returned unchanged. + ERB has no raw block. Replace each opening delimiter with a small, safe ERB + expression that prints the two characters ``<%`` literally, followed by the + original variant suffix (``=``, ``-`` or ``#``). Closing delimiters outside + an ERB tag are plain text and must not be wrapped in an expression, because + ``<%= "%>" %>`` is parsed by ERB as a prematurely closed tag. """ if not text or not contains_erb_markup(text): return text @@ -121,13 +125,12 @@ def escape_erb_literal(text: str) -> str: n = len(text) while i < n: matched = None - for marker in (*_ERB_CLOSE_MARKERS, *_ERB_OPEN_MARKERS): + for marker in _ERB_OPEN_MARKERS: if text.startswith(marker, i): matched = marker break if matched is not None: - escaped = matched.replace("\\", "\\\\").replace('"', '\\"') - result.append('<%= "' + escaped + '" %>') + result.append("<%= '<%' %>" + matched[2:]) i += len(matched) else: result.append(text[i]) diff --git a/src/jinjaturtle/handlers/base.py b/src/jinjaturtle/handlers/base.py index 14aaec7..f0f2e39 100644 --- a/src/jinjaturtle/handlers/base.py +++ b/src/jinjaturtle/handlers/base.py @@ -2,6 +2,7 @@ from __future__ import annotations from pathlib import Path from typing import Any, Iterable +import re class BaseHandler: @@ -50,30 +51,39 @@ class BaseHandler: return text[:i], text[i:] return text, "" + @staticmethod + def _safe_identifier_part(raw: object, *, fallback: str = "var") -> str: + """Return a conservative lowercase full identifier.""" + text = BaseHandler._safe_path_part(raw, fallback=fallback) + if not re.match(r"^[a-z_]", text): + text = f"{fallback}_{text}" + return text + + @staticmethod + def _safe_path_part(raw: object, *, fallback: str = "part") -> str: + """Return a conservative lowercase identifier fragment.""" + text = re.sub(r"[^A-Za-z0-9_]+", "_", str(raw or "").strip()) + text = re.sub(r"_+", "_", text).strip("_").lower() + return text or fallback + @staticmethod def make_var_name(role_prefix: str, path: Iterable[str]) -> str: """ - Build an Ansible var name like: - role_prefix_section_subsection_key + Build a safe variable name like ``role_section_key``. - Sanitises parts to lowercase [a-z0-9_] and strips extras. + Both the role prefix and every source-derived path component are reduced + to a strict identifier grammar. Config keys, section names and CLI role + names are untrusted input; they must never be able to shape Jinja/ERB + statement syntax. """ - role_prefix = role_prefix.strip().lower() + prefix = BaseHandler._safe_identifier_part(role_prefix, fallback="jinjaturtle") clean_parts: list[str] = [] for part in path: - part = str(part).strip() - part = part.replace(" ", "_") - cleaned_chars: list[str] = [] - for c in part: - if c.isalnum() or c == "_": - cleaned_chars.append(c.lower()) - else: - cleaned_chars.append("_") - cleaned_part = "".join(cleaned_chars).strip("_") + cleaned_part = BaseHandler._safe_path_part(part, fallback="part") if cleaned_part: clean_parts.append(cleaned_part) if clean_parts: - return role_prefix + "_" + "_".join(clean_parts) - return role_prefix + return prefix + "_" + "_".join(clean_parts) + return prefix diff --git a/src/jinjaturtle/handlers/ini.py b/src/jinjaturtle/handlers/ini.py index 9f50839..4252e9d 100644 --- a/src/jinjaturtle/handlers/ini.py +++ b/src/jinjaturtle/handlers/ini.py @@ -59,15 +59,19 @@ class IniHandler(BaseHandler): lines: list[str] = [] for section in parser.sections(): - lines.append(f"[{section}]") + lines.append(f"[{escape_jinja_literal(section)}]") for key, value in parser.items(section, raw=True): path = (section, key) var_name = self.make_var_name(role_prefix, path) value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: - lines.append(f"{key} = {j2.quoted_variable(var_name)}") + lines.append( + f"{escape_jinja_literal(key)} = {j2.quoted_variable(var_name)}" + ) else: - lines.append(f"{key} = {j2.variable(var_name)}") + lines.append( + f"{escape_jinja_literal(key)} = {j2.variable(var_name)}" + ) lines.append("") return "\n".join(lines).rstrip() + "\n" diff --git a/src/jinjaturtle/handlers/json.py b/src/jinjaturtle/handlers/json.py index 064535a..7f86037 100644 --- a/src/jinjaturtle/handlers/json.py +++ b/src/jinjaturtle/handlers/json.py @@ -7,6 +7,7 @@ from typing import Any from . import DictLikeHandler from .. import j2 +from ..escape import escape_jinja_literal from ..loop_analyzer import LoopCandidate @@ -109,10 +110,14 @@ class JsonHandler(DictLikeHandler): chunks: list[str] = [] pos = 0 for path, start, end in spans: - chunks.append(text[pos:start]) + # Chunks are copied verbatim from the source and may include object + # keys, whitespace or punctuation. Object keys are attacker- + # controlled literal text, so escape template delimiters before + # inserting the generated value expression between chunks. + chunks.append(escape_jinja_literal(text[pos:start])) chunks.append(self._json_value_expr(self.make_var_name(role_prefix, path))) pos = end - chunks.append(text[pos:]) + chunks.append(escape_jinja_literal(text[pos:])) return "".join(chunks) def _collect_json_scalar_spans( @@ -213,7 +218,10 @@ class JsonHandler(DictLikeHandler): def _walk(obj: Any, path: tuple[str, ...] = ()) -> Any: if isinstance(obj, dict): - return {k: _walk(v, path + (str(k),)) for k, v in obj.items()} + return { + escape_jinja_literal(str(k)): _walk(v, path + (str(k),)) + for k, v in obj.items() + } if isinstance(obj, list): return [_walk(v, path + (str(i),)) for i, v in enumerate(obj)] # scalar - use marker that will be replaced with to_json @@ -261,7 +269,10 @@ class JsonHandler(DictLikeHandler): return f"__LOOP_DICT__{collection_var}__{item_var}__" if isinstance(obj, dict): - return {k: _walk(v, current_path + (str(k),)) for k, v in obj.items()} + return { + escape_jinja_literal(str(k)): _walk(v, current_path + (str(k),)) + for k, v in obj.items() + } if isinstance(obj, list): # Check if this list is a loop candidate if current_path in loop_paths: @@ -364,8 +375,10 @@ class JsonHandler(DictLikeHandler): ] # first line has no indent; we prepend `inner` when emitting for i, key in enumerate(keys): comma = "," if i < len(keys) - 1 else "" + safe_key = escape_jinja_literal(str(key)) + value_expr = candidate.value_expr(item_var, str(key)) dict_lines.append( - f'{field}"{key}": ' f"{j2.to_json(f'{item_var}.{key}')}{comma}" + f'{field}"{safe_key}": ' f"{j2.to_json(value_expr)}{comma}" ) # Comma between *items* goes after the closing brace. dict_lines.append(f"{inner}}}{j2.if_not_loop_last()},{j2.endif()}") diff --git a/src/jinjaturtle/handlers/postfix.py b/src/jinjaturtle/handlers/postfix.py index f6a1974..6bd15d3 100644 --- a/src/jinjaturtle/handlers/postfix.py +++ b/src/jinjaturtle/handlers/postfix.py @@ -94,7 +94,7 @@ class PostfixMainHandler(BaseHandler): lines: list[str] = [] for k, v in parsed.items(): var = self.make_var_name(role_prefix, (k,)) - lines.append(f"{k} = {j2.variable(var)}") + lines.append(f"{escape_jinja_literal(k)} = {j2.variable(var)}") return "\n".join(lines).rstrip() + "\n" return self._generate_from_text(role_prefix, original_text) diff --git a/src/jinjaturtle/handlers/toml.py b/src/jinjaturtle/handlers/toml.py index 6d25cff..8a906ae 100644 --- a/src/jinjaturtle/handlers/toml.py +++ b/src/jinjaturtle/handlers/toml.py @@ -441,17 +441,17 @@ class TomlHandler(DictLikeHandler): if isinstance(value, str): out_lines.append( f"{escape_jinja_literal(str(key))} = " - f"{self._toml_quoted_expr(f'{item_var}.{key}')}\n" + f"{self._toml_quoted_expr(candidate.value_expr(item_var, str(key)))}\n" ) elif isinstance(value, bool): out_lines.append( f"{escape_jinja_literal(str(key))} = " - f"{self._toml_value_expr(f'{item_var}.{key}', value)}\n" + f"{self._toml_value_expr(candidate.value_expr(item_var, str(key)), value)}\n" ) else: out_lines.append( f"{escape_jinja_literal(str(key))} = " - f"{self._toml_value_expr(f'{item_var}.{key}', value)}\n" + f"{self._toml_value_expr(candidate.value_expr(item_var, str(key)), value)}\n" ) out_lines.append(f"{j2.for_end()}\n") diff --git a/src/jinjaturtle/handlers/xml.py b/src/jinjaturtle/handlers/xml.py index 9fde3ec..9726d68 100644 --- a/src/jinjaturtle/handlers/xml.py +++ b/src/jinjaturtle/handlers/xml.py @@ -1,9 +1,11 @@ from __future__ import annotations from collections import Counter, defaultdict +import secrets from pathlib import Path from typing import Any -import xml.etree.ElementTree as ET # nosec +import xml.etree.ElementTree as ET # nosec B405 - not used for untrusted XML parsing; all parsing uses defusedxml. +import defusedxml.ElementTree as DET from .base import BaseHandler from .. import j2 @@ -18,11 +20,24 @@ class XmlHandler(BaseHandler): fmt = "xml" + def __init__(self) -> None: + # Marker comments are implementation details converted to Jinja control + # structures later in the same generation pass. Include an unguessable + # nonce so source-file comments such as can never be + # mistaken for trusted internal markers. + self._marker_prefix = f"JINJATURTLE:{secrets.token_hex(16)}:" + + def _marker(self, kind: str, payload: str) -> str: + return f"{self._marker_prefix}{kind}:{payload}" + def parse(self, path: Path) -> ET.Element: text = path.read_text(encoding="utf-8") - parser = ET.XMLParser( - target=ET.TreeBuilder(insert_comments=False) - ) # nosec B314 + parser = DET.XMLParser( + target=ET.TreeBuilder(insert_comments=False), + forbid_dtd=True, + forbid_entities=True, + forbid_external=True, + ) parser.feed(text) root = parser.close() return root @@ -216,7 +231,7 @@ class XmlHandler(BaseHandler): # Create a loop comment/marker # We'll handle the actual loop generation in text processing - loop_marker = ET.Comment(f"LOOP:{tag}") + loop_marker = ET.Comment(self._marker("LOOP", tag)) elem.append(loop_marker) elif counts[tag] > 1: @@ -231,14 +246,8 @@ class XmlHandler(BaseHandler): walk(root, ()) - # Internal marker prefixes used by JinjaTurtle's own comment nodes. These - # must NOT be escaped (they are converted into real Jinja control structures - # downstream). Source-file comments have none of these prefixes. - _MARKER_PREFIXES = ("LOOP:", "IF:", "ENDIF:") - def _is_jt_marker(self, comment_text: str) -> bool: - stripped = (comment_text or "").lstrip() - return any(stripped.startswith(p) for p in self._MARKER_PREFIXES) + return (comment_text or "").lstrip().startswith(self._marker_prefix) def _escape_source_comments(self, root: ET.Element) -> None: """Escape template metacharacters in comments preserved from the source. @@ -252,8 +261,9 @@ class XmlHandler(BaseHandler): placeholders, so comments (and the prolog, handled separately) are the only XML injection vector. - JinjaTurtle's own internal marker comments are left untouched so they can - be converted into real loops/conditionals later. + Only nonce-bearing marker comments generated during this process are left + untouched. Source comments that look like old marker syntax (for + example ) are ordinary comments and remain inert. """ # ET represents comments with a callable tag (ET.Comment). Iterate all # descendants and escape comment text that is not one of our markers. @@ -266,7 +276,12 @@ class XmlHandler(BaseHandler): """Generate scalar-only Jinja2 template.""" prolog, body = self._split_xml_prolog(text) - parser = ET.XMLParser(target=ET.TreeBuilder(insert_comments=True)) # nosec B314 + parser = DET.XMLParser( + target=ET.TreeBuilder(insert_comments=True), + forbid_dtd=True, + forbid_entities=True, + forbid_external=True, + ) parser.feed(body) root = parser.close() @@ -294,7 +309,12 @@ class XmlHandler(BaseHandler): prolog, body = self._split_xml_prolog(text) # Parse with comments preserved - parser = ET.XMLParser(target=ET.TreeBuilder(insert_comments=True)) # nosec B314 + parser = DET.XMLParser( + target=ET.TreeBuilder(insert_comments=True), + forbid_dtd=True, + forbid_entities=True, + forbid_external=True, + ) parser.feed(body) root = parser.close() @@ -334,12 +354,15 @@ class XmlHandler(BaseHandler): # Build a sample element for each loop to use as template lines = xml_str.split("\n") result_lines = [] + loop_marker = f"", start) tag_name = line[start:end].strip() @@ -365,7 +388,7 @@ class XmlHandler(BaseHandler): merged_dict = self._merge_dicts_for_template(candidate.items) sample_elem = self._dict_to_xml_element( - tag_name, merged_dict, item_var + tag_name, merged_dict, item_var, candidate ) # Apply indentation to the sample element @@ -393,18 +416,16 @@ class XmlHandler(BaseHandler): else: result_lines.append(line) - # Post-process to replace and with Jinja2 conditionals + # Post-process nonce-bearing IF/ENDIF markers into Jinja2 conditionals. final_lines = [] for line in result_lines: - # Replace with {% if var.field is defined %} - if "", start) condition = line[start:end] indent = len(line) - len(line.lstrip()) final_lines.append(f"{' ' * indent}{j2.if_defined(condition)}") - # Replace with {% endif %} - elif "\n" + " ok\n" + " \n" + "\n" + ) + template_text, defaults = _run_jinjaturtle(tmp_path, "evil.xml", body, "xml") + assert "{% if boom.run" not in template_text + rendered = _render_jinja(template_text, {**defaults, "boom": _Boom()}) + assert TRIP not in rendered + assert "IF:boom.run('xml-marker')" in rendered + + +def test_folder_yaml_keys_are_escaped_and_do_not_execute(tmp_path): + from jinjaturtle.multi import process_directory + + (tmp_path / "one.yaml").write_text( + 'evil_{{ boom.run("folder-yaml") }}: ok\n', encoding="utf-8" + ) + defaults_yaml, outputs = process_directory( + tmp_path, recursive=False, role_prefix="role" + ) + defaults = pyyaml.safe_load(defaults_yaml) or {} + assert len(outputs) == 1 + template_text = outputs[0].template + rendered = _render_jinja( + template_text, {**defaults["role_items"][0], "boom": _Boom()} + ) + assert TRIP not in rendered + assert 'evil_{{ boom.run("folder-yaml") }}' in rendered + + +def test_erb_source_comment_payload_is_literal_not_executed(tmp_path): + import shutil + + marker = tmp_path / "erb-owned" + src = tmp_path / "evil.ini" + src.write_text( + "[s]\n" "good = ok\n" f"# <%= File.write({str(marker)!r}, 'owned') %>\n", + encoding="utf-8", + ) + tpl = tmp_path / "out.erb" + dfl = tmp_path / "hiera.yml" + res = subprocess.run( + [ + sys.executable, + "-m", + "jinjaturtle.cli", + str(src), + "-f", + "ini", + "--template-engine", + "erb", + "--role-name", + "role", + "-t", + str(tpl), + "-d", + str(dfl), + ], + capture_output=True, + text=True, + ) + assert res.returncode == 0, res.stderr + erb_text = tpl.read_text(encoding="utf-8") + live_chunks = re.findall(r"<%[-=#]?(.*?)-?%>", erb_text, re.S) + assert not any("File.write" in chunk for chunk in live_chunks) + + ruby = shutil.which("ruby") + if ruby is None: + pytest.skip("ruby is not installed") + run = subprocess.run( + [ + ruby, + "-rerb", + "-e", + "puts ERB.new(File.read(ARGV[0])).result(binding)", + str(tpl), + ], + capture_output=True, + text=True, + ) + assert run.returncode == 0, run.stderr + assert not marker.exists() + + +def test_ansible_defaults_tag_dangerous_strings_as_unsafe_and_still_load(): + from jinjaturtle.core import generate_ansible_yaml + + out = generate_ansible_yaml( + "role", + [ + (("jinja",), "{{ boom.run('unsafe') }}"), + (("erb",), "<%= File.write('/tmp/nope', 'x') %>"), + (("plain",), "ordinary"), + ], + ) + assert "role_jinja: !unsafe" in out + assert "role_erb: !unsafe" in out + assert "role_plain: ordinary" in out + loaded = pyyaml.safe_load(out) + assert loaded["role_jinja"] == "{{ boom.run('unsafe') }}" + + +def test_cli_rejects_template_unsafe_role_name(tmp_path): + from jinjaturtle import cli + + src = tmp_path / "a.ini" + src.write_text("[s]\nkey=value\n", encoding="utf-8") + with pytest.raises(SystemExit) as exc: + cli._main([str(src), "--role-name", "bad-{{ role }}"]) + assert exc.value.code == 2 + + +def test_yaml_loop_source_key_cannot_become_jinja_expression(tmp_path): + from jinjaturtle.core import ( + analyze_loops, + flatten_config, + generate_ansible_yaml, + generate_jinja2_template, + parse_config, + ) + + malicious_key = ( + "__class__.__mro__[1].__subclasses__()[166].__init__" + ".__globals__['__builtins__']['__import__']('os')" + ".popen('echo JT_RCE').read()" + ) + src = tmp_path / "evil.yaml" + src.write_text( + "servers:\n" f" - {malicious_key!r}: a\n" f" - {malicious_key!r}: b\n", + encoding="utf-8", + ) + fmt, parsed = parse_config(src, "yaml") + loops = analyze_loops(fmt, parsed) + flat = flatten_config(fmt, parsed, loops) + template = generate_jinja2_template(fmt, parsed, "role", src.read_text(), loops) + defaults = pyyaml.safe_load(generate_ansible_yaml("role", flat, loops)) or {} + + assert f"server.{malicious_key}" not in template + assert "server.jt_" in template + rendered = _render_jinja(template, defaults) + assert TRIP not in rendered + assert "JT_RCE" in rendered # literal source key only, not command output + + +def test_toml_array_table_loop_source_key_cannot_become_jinja_expression(tmp_path): + from jinjaturtle.core import ( + analyze_loops, + flatten_config, + generate_ansible_yaml, + generate_jinja2_template, + parse_config, + ) + + malicious_key = ( + "__class__.__mro__[1].__subclasses__()[166].__init__" + ".__globals__['__builtins__']['__import__']('os')" + ".popen('echo JT_TOML_RCE').read()" + ) + src = tmp_path / "evil.toml" + src.write_text( + f'[[servers]]\n"{malicious_key}" = "a"\n\n' + f'[[servers]]\n"{malicious_key}" = "b"\n', + encoding="utf-8", + ) + fmt, parsed = parse_config(src, "toml") + loops = analyze_loops(fmt, parsed) + flat = flatten_config(fmt, parsed, loops) + template = generate_jinja2_template(fmt, parsed, "role", src.read_text(), loops) + defaults = pyyaml.safe_load(generate_ansible_yaml("role", flat, loops)) or {} + + assert f"server.{malicious_key}" not in template + assert "server.jt_" in template + rendered = _render_jinja(template, defaults) + assert TRIP not in rendered + assert "JT_TOML_RCE" in rendered + + +def test_erb_loop_source_key_cannot_become_ruby_method_call(tmp_path): + import shutil + + from jinjaturtle.core import ( + analyze_loops, + flatten_config, + generate_erb_template, + generate_puppet_hiera_yaml, + parse_config, + ) + + marker = tmp_path / "erb-loop-owned" + key = f"instance_eval(%q{{File.write({str(marker)!r}, 'owned')}})" + src = tmp_path / "evil.yaml" + src.write_text( + "servers:\n" f" - {key!r}: a\n" f" - {key!r}: b\n", + encoding="utf-8", + ) + fmt, parsed = parse_config(src, "yaml") + loops = analyze_loops(fmt, parsed) + flat = flatten_config(fmt, parsed, loops) + erb = generate_erb_template( + fmt, + parsed, + "role", + original_text=src.read_text(), + loop_candidates=loops, + flat_items=flat, + ) + assert "server.instance_eval" not in erb + assert "server['jt_" in erb + + ruby = shutil.which("ruby") + if ruby is None: + pytest.skip("ruby is not installed") + + hiera_path = tmp_path / "hiera.yml" + hiera_path.write_text( + generate_puppet_hiera_yaml("role", flat, loops), encoding="utf-8" + ) + tpl = tmp_path / "safe.erb" + tpl.write_text(erb, encoding="utf-8") + run = subprocess.run( + [ + ruby, + "-rerb", + "-ryaml", + "-e", + "@servers = YAML.load_file(ARGV[1])['role::servers']; " + "puts ERB.new(File.read(ARGV[0]), trim_mode: '-').result(binding)", + str(tpl), + str(hiera_path), + ], + capture_output=True, + text=True, + ) + assert run.returncode == 0, run.stderr + assert not marker.exists() + + +def test_canonical_ini_and_postfix_escape_source_keys(): + from configparser import ConfigParser + + from jinjaturtle.core import generate_jinja2_template + + parser = ConfigParser() + parser.optionxform = str + parser.add_section('s_{{ boom.run("ini") }}') + parser.set('s_{{ boom.run("ini") }}', "key", "ok") + ini_template = generate_jinja2_template("ini", parser, "role", original_text=None) + rendered = _render_jinja( + ini_template, {"role_s_boom_run_ini_key": "ok", "boom": _Boom()} + ) + assert TRIP not in rendered + assert 's_{{ boom.run("ini") }}' in rendered + + postfix_template = generate_jinja2_template( + "postfix", + {'evil_{{ boom.run("postfix") }}': "ok"}, + "role", + original_text=None, + ) + rendered = _render_jinja( + postfix_template, + {"role_evil_boom_run_postfix": "ok", "boom": _Boom()}, + ) + assert TRIP not in rendered + assert 'evil_{{ boom.run("postfix") }}' in rendered + + +def test_template_safety_validator_rejects_future_expression_injection(): + from jinjaturtle.template_safety import ( + UnsafeTemplateError, + validate_generated_template, + ) + + with pytest.raises(UnsafeTemplateError): + validate_generated_template("{{ item.__class__.__mro__[1] }}") + with pytest.raises(UnsafeTemplateError): + validate_generated_template("{{ boom.run('x') }}") + # Generated field names and loop.last remain allowed. + validate_generated_template( + "{% for item in role_items %}{{ item.jt_name_82a3537f }}" + "{% if not loop.last %},{% endif %}{% endfor %}" + ) + + +def test_defaults_generation_rejects_colliding_variable_names(): + from jinjaturtle.core import generate_ansible_yaml + + with pytest.raises(ValueError, match="same variable name"): + generate_ansible_yaml("role", [(("a-b",), 1), (("a_b",), 2)]) diff --git a/tests/test_json_handler.py b/tests/test_json_handler.py index b2bbe75..69f94cb 100644 --- a/tests/test_json_handler.py +++ b/tests/test_json_handler.py @@ -143,7 +143,9 @@ def test_json_direct_loop_generation_renders_valid_json_without_joined_objects() env = Environment(keep_trailing_newline=True) env.filters["to_json"] = lambda value, **kwargs: json.dumps(value, **kwargs) - rendered = env.from_string(template).render(fact_implementations=items) + rendered = env.from_string(template).render( + fact_implementations=candidate.safe_items() + ) assert "}, {" not in rendered assert json.loads(rendered) == parsed diff --git a/tests/test_multi.py b/tests/test_multi.py index 2076655..f33f544 100644 --- a/tests/test_multi.py +++ b/tests/test_multi.py @@ -123,3 +123,15 @@ def test_process_directory_ini_union_marks_optional_sections_and_keys(tmp_path: def test_process_directory_rejects_empty_folder(tmp_path: Path): with pytest.raises(ValueError, match="No supported config files"): process_directory(tmp_path, recursive=False, role_prefix="role") + + +def test_directory_mode_skips_supported_symlinks(tmp_path: Path): + outside = tmp_path / "outside.yaml" + outside.write_text("secret: value\n", encoding="utf-8") + root = tmp_path / "root" + root.mkdir() + (root / "real.yaml").write_text("name: ok\n", encoding="utf-8") + (root / "link.yaml").symlink_to(outside) + + files = iter_supported_files(root, recursive=False) + assert files == [root / "real.yaml"]