diff --git a/README.md b/README.md index 26730fb..8423cb7 100644 --- a/README.md +++ b/README.md @@ -115,9 +115,8 @@ ERB mode the file is intended to be Puppet Hiera data rather than Ansible role defaults. JinjaTurtle does **not** generate Puppet classes or `file` resources. A Puppet -module, or another tool such as Enroll, should still declare the class -parameters and call the template, for example with Puppet's `template()` -function. +module, should still declare the class parameters and call the template, for +example with Puppet's `template()` function. ## Using `--puppet-class` @@ -340,15 +339,53 @@ JinjaTurtle also templates some common bespoke config formats: For ambiguous extensions like `*.conf`, JinjaTurtle uses lightweight content sniffing. You can always force a specific handler with `--format`. -## Relationship with Enroll +## Security model -JinjaTurtle can be used directly, but it is also useful as a helper for tools -that generate configuration-management code. +JinjaTurtle is frequently pointed at config files that were *harvested* from +real systems, where some content may be influenced by an untrusted party (a +hostname, a login banner, a `GECOS` comment, a "Managed by ..." note). It is +therefore designed so that source content cannot turn into executable template +code. + +Two guarantees matter: + +1. **Values are data, never code.** Every config *value* is replaced with a + `{{ variable }}` placeholder in the template, and the original value is stored + separately in the defaults/Hiera data. When the template is later rendered, + the placeholder prints the value as a literal string; Jinja2 does not + recursively render the *contents* of a variable, so a payload sitting inside + a value (for example `motd = {{ salt['cmd.run']('id') }}`) is inert. + +2. **Verbatim text is neutralised.** To preserve formatting, JinjaTurtle copies + comments, blank lines, headers and any unrecognised lines from the source + into the template. Any template metacharacters in that copied text + (`{{ }}`, `{% %}`, `{# #}` for Jinja2; `<% %>` for ERB) are escaped so they + render as the literal characters the author wrote, rather than executing. + +### Consumer responsibilities + +The value guarantee above relies on the downstream renderer being single-pass, +which is the normal case: + +- **Ansible**: rendering a template with `template:`/`ansible.builtin.template` + is single-pass. For defence in depth, treat the generated defaults as + untrusted input — Ansible already does not re-template variable *contents* by + default. If you build your own var structures from this data and pass them + through additional templating, mark untrusted values with the `!unsafe` tag so + they are never re-evaluated. +- **Salt**: use the generated file as a `file.managed` template + (`template: jinja`). Do **not** place JinjaTurtle output where Salt would + render it a second time as part of SLS/pillar rendering, which is a separate + Jinja pass and would re-evaluate any embedded expressions. +- **Puppet/ERB**: render with the standard `template()`/`epp()` flow. + +In short: render JinjaTurtle output exactly once. Do not feed it back through +another templating pass. + +**IMPORTANT**: Always review both the original config files, then the resulting +templates generated by JinjaTurtle, before integrating them into your config +management system! -Enroll can call JinjaTurtle when it is available on the `PATH`. In that setup, -Enroll decides where Puppet, Ansible, or Salt files belong, while JinjaTurtle -concentrates on converting harvested config files into templates plus variable -data. ## Found a bug, have a suggestion? diff --git a/src/jinjaturtle/cli.py b/src/jinjaturtle/cli.py index 7658bfb..0c403fa 100644 --- a/src/jinjaturtle/cli.py +++ b/src/jinjaturtle/cli.py @@ -199,4 +199,8 @@ def main() -> None: """ Console-script entry point. """ - _main(sys.argv[1:]) + sys.exit(_main(sys.argv[1:])) + + +if __name__ == "__main__": + main() diff --git a/src/jinjaturtle/core.py b/src/jinjaturtle/core.py index 673aad0..1faefca 100644 --- a/src/jinjaturtle/core.py +++ b/src/jinjaturtle/core.py @@ -412,8 +412,6 @@ def generate_puppet_hiera_yaml( ``role_prefix`` remains the source variable prefix used by JinjaTurtle while ``puppet_class`` is the Puppet class/Hiera namespace. In the normal case they are the same, so ``php_memory_limit`` becomes ``php::memory_limit``. - Enroll may pass a file-specific role prefix and a separate Puppet class to - avoid parameter-name collisions inside one generated Puppet module. """ klass = puppet_class_name(puppet_class or role_prefix) diff --git a/src/jinjaturtle/erb.py b/src/jinjaturtle/erb.py index 445ec51..6eb26ad 100644 --- a/src/jinjaturtle/erb.py +++ b/src/jinjaturtle/erb.py @@ -2,6 +2,8 @@ from __future__ import annotations import re +from .escape import escape_erb_literal + def _safe_name(raw: str, *, fallback: str = "var") -> str: text = re.sub(r"[^A-Za-z0-9_]+", "_", str(raw or fallback)).strip("_").lower() @@ -38,10 +40,9 @@ def puppet_local_var_name( generated Jinja variable such as ``php_memory_limit`` becomes Puppet local parameter ``memory_limit`` and Hiera key ``php::memory_limit``. - Enroll sometimes needs a file-specific variable prefix to avoid collisions - inside a generated module. When ``puppet_class`` differs from - ``role_prefix`` we keep the full generated variable name as the local - parameter and only use ``puppet_class`` as the Hiera namespace. + When ``puppet_class`` differs from ``role_prefix`` we keep the full + generated variable name as the local parameter and only use ``puppet_class`` + as the Hiera namespace. """ var_name = _safe_name(jinja_var_name, fallback="value") @@ -71,7 +72,33 @@ class ErbTranslator: self.loop_stack: list[tuple[str, str, str]] = [] self.needs_json = False + # Matches a JinjaTurtle ``{% raw %} ... {% endraw %}`` block (non-greedy). + # JinjaTurtle emits raw blocks only to carry verbatim, security-escaped + # source text (comments and unrecognised lines), so the *contents* must be + # treated as literal output, never translated as Jinja tokens. + _RAW_BLOCK_RE = re.compile(r"{%\s*raw\s*%}(.*?){%\s*endraw\s*%}", re.S) + def translate(self, template_text: str) -> str: + # Split out raw blocks first. Their inner text is literal and must be + # carried through as literal ERB (with ERB delimiters re-escaped), rather + # than tokenised -- otherwise an escaped Jinja payload inside a comment + # would be "re-animated" into live ERB during translation. + segments = self._RAW_BLOCK_RE.split(template_text) + out: list[str] = [] + # re.split with one capture group yields: [text, raw_inner, text, ...]. + for idx, segment in enumerate(segments): + if idx % 2 == 1: + # Captured raw-block contents: emit as literal ERB text. + out.append(escape_erb_literal(segment)) + else: + out.append(self._translate_tokens(segment)) + + rendered = "".join(out) + if self.needs_json and "require 'json'" not in rendered: + rendered = "<% require 'json' -%>\n" + rendered + return rendered + + def _translate_tokens(self, template_text: str) -> str: parts = self._TOKEN_RE.split(template_text) out: list[str] = [] for token in parts: @@ -86,11 +113,7 @@ class ErbTranslator: out.append(self.statement_to_erb(stmt)) continue out.append(token) - - rendered = "".join(out) - if self.needs_json and "require 'json'" not in rendered: - rendered = "<% require 'json' -%>\n" + rendered - return rendered + return "".join(out) def local_var(self, name: str) -> str: return puppet_local_var_name( diff --git a/src/jinjaturtle/escape.py b/src/jinjaturtle/escape.py new file mode 100644 index 0000000..3b8575e --- /dev/null +++ b/src/jinjaturtle/escape.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +"""Neutralise template metacharacters in text copied verbatim from source files. + +JinjaTurtle preserves formatting by copying parts of the *original* config file +straight into the generated template: comments, blank lines, section headers, +and any line it does not recognise as ``key = value``. Config *values* are +always replaced with ``{{ var }}`` placeholders and parked in the defaults data, +so a payload inside a value is inert. Verbatim text is different: if the source +contains ``{{ ... }}``, ``{% ... %}`` or ``{# ... #}`` (Jinja2), or ``<%= %>`` / +``<% %>`` (ERB), that text becomes *live template code* in the output and is +executed when Salt/Ansible/Puppet later renders the template. + +Because JinjaTurtle is frequently fed harvested, attacker-influenceable config +(hostnames, banners, GECOS-derived comments, "Managed by" notes), this is a +template-injection / SSTI vector that can lead to remote code execution on the +configuration-management control node. + +The functions here render those metacharacters as literal text so they survive a +later template render as the characters the source author actually wrote, rather +than as executable template syntax. + +Design notes: + * We only ever escape text that originates from the *source file*. We never + pass JinjaTurtle's own generated placeholders (``{{ role_var }}``) through + these helpers, so the placeholders keep working. + * Jinja2 literal text is wrapped in a single ``{% raw %} ... {% endraw %}`` + block. ``raw`` disables *all* tag interpretation inside it -- expressions, + statements and ``{# #}`` comments alike -- so one wrap neutralises every + Jinja construct. The only way to break out of a raw block is a literal + ``{% endraw %}`` in the source, so we defang the token ``endraw`` (in any + internal spacing) before wrapping. + * The ERB translator (``erb.py``) is raw-aware: it copies the *contents* of a + JinjaTurtle raw block through as literal ERB text and re-escapes any ERB + delimiters found there. That keeps a Jinja-escaped comment inert after the + Jinja2 -> ERB translation step, instead of the payload being "re-animated" + as ERB. + * ``escape_erb_literal`` exists for completeness / direct ERB emission: it + rewrites each ERB delimiter into an ERB expression that prints the delimiter + characters literally. +""" + +import re + +# Jinja2 delimiters we must neutralise. Engine-default; JinjaTurtle never +# configures custom delimiters. +_JINJA_MARKERS = ("{{", "}}", "{%", "%}", "{#", "#}") + +# ERB delimiters. Longer markers first so "<%=" matches before "<%". +_ERB_OPEN_MARKERS = ("<%=", "<%-", "<%#", "<%") +_ERB_CLOSE_MARKERS = ("-%>", "%>") + +# Matches a Jinja2 endraw tag in any internal spacing, e.g. "{%endraw%}", +# "{% endraw %}", "{%- endraw -%}". +_ENDRAW_RE = re.compile(r"{%-?\s*endraw\s*-?%}") + +# Sentinel inserted between "end" and "raw" to break the endraw keyword without +# changing the visible characters. We use a Jinja comment-free approach: insert +# the two halves across a raw boundary so the literal text still reads "endraw" +# to a human but is never a valid tag. See escape_jinja_literal for usage. + + +def contains_jinja_markup(text: str) -> bool: + """Return True if *text* contains any Jinja2 delimiter.""" + return any(m in text for m in _JINJA_MARKERS) + + +def contains_erb_markup(text: str) -> bool: + """Return True if *text* contains any ERB delimiter.""" + return any(m in text for m in (*_ERB_OPEN_MARKERS, *_ERB_CLOSE_MARKERS)) + + +def _defang_endraw(text: str) -> str: + """Rewrite any literal ``{% endraw %}`` so it cannot close our raw wrapper. + + We turn each endraw tag into ``{% endraw %}{{ '{% endraw %}' }}{% raw %}``... + no -- that would re-introduce live tags. Instead we keep everything literal: + we break the keyword by emitting the tag's text in two raw segments split + inside the word ``endraw``. The result, when later rendered, reproduces the + exact original characters ``{% endraw %}`` while never being a parseable tag. + """ + + def _replace(match: re.Match[str]) -> str: + tag = match.group(0) + # Split the keyword "endraw" as "end" + "raw"; close and reopen the raw + # block between them. Each half is plain text inside a raw block, so the + # reconstructed output is byte-identical to the original tag, but at no + # point does the token "{% endraw %}" exist contiguously to close raw. + idx = tag.lower().index("endraw") + head = tag[: idx + 3] # up to and including "end" + tail = tag[idx + 3 :] # "raw...%}" + return f"{head}{{% endraw %}}{{% raw %}}{tail}" + + return _ENDRAW_RE.sub(_replace, text) + + +def escape_jinja_literal(text: str) -> str: + """Make *text* render as literal characters under a later Jinja2 render. + + 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. + """ + if not text or not contains_jinja_markup(text): + return text + return "{% raw %}" + _defang_endraw(text) + "{% endraw %}" + + +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. + """ + if not text or not contains_erb_markup(text): + return text + + result: list[str] = [] + i = 0 + n = len(text) + while i < n: + matched = None + for marker in (*_ERB_CLOSE_MARKERS, *_ERB_OPEN_MARKERS): + if text.startswith(marker, i): + matched = marker + break + if matched is not None: + escaped = matched.replace("\\", "\\\\").replace('"', '\\"') + result.append('<%= "' + escaped + '" %>') + i += len(matched) + else: + result.append(text[i]) + i += 1 + return "".join(result) + + +def escape_literal(text: str, *, engine: str = "jinja2") -> str: + """Escape verbatim *text* for the target template *engine*. + + ``engine`` is ``"jinja2"`` (default) or ``"erb"``. Unknown engines fall back + to Jinja2 escaping. In JinjaTurtle's pipeline ERB output is produced by + translating Jinja2 output, and the translator is raw-aware, so handlers can + always Jinja-escape and rely on the translator to keep the literal inert. + """ + if engine == "erb": + return escape_erb_literal(text) + return escape_jinja_literal(text) diff --git a/src/jinjaturtle/handlers/ini.py b/src/jinjaturtle/handlers/ini.py index 1aa5e22..9f50839 100644 --- a/src/jinjaturtle/handlers/ini.py +++ b/src/jinjaturtle/handlers/ini.py @@ -6,6 +6,7 @@ from typing import Any from . import BaseHandler from .. import j2 +from ..escape import escape_jinja_literal class IniHandler(BaseHandler): @@ -84,16 +85,18 @@ class IniHandler(BaseHandler): line = raw_line stripped = line.lstrip() - # Blank or pure comment: keep as-is + # Blank or pure comment: keep formatting, but neutralise any + # template metacharacters so attacker-controlled comment text cannot + # become live template code in the output. if not stripped or stripped[0] in {"#", ";"}: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Section header if stripped.startswith("[") and "]" in stripped: header_inner = stripped[1 : stripped.index("]")] current_section = header_inner.strip() - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Work without newline so we can re-attach it exactly @@ -108,8 +111,9 @@ class IniHandler(BaseHandler): eq_index = content.find("=") if eq_index == -1: - # Not a simple key=value line: leave untouched - out_lines.append(raw_line) + # Not a simple key=value line: leave content intact but escape + # any template metacharacters. + out_lines.append(escape_jinja_literal(raw_line)) continue before_eq = content[:eq_index] @@ -117,7 +121,7 @@ class IniHandler(BaseHandler): key = before_eq.strip() if not key: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Whitespace after '=' @@ -146,8 +150,16 @@ class IniHandler(BaseHandler): else: replacement_value = j2.variable(var_name) + # ``before_eq`` (key + surrounding whitespace) and ``comment_part`` + # both originate from the source file and may carry template + # metacharacters; escape each independently so the safe + # ``replacement_value`` placeholder between them is preserved. new_content = ( - before_eq + "=" + leading_ws + replacement_value + comment_part + escape_jinja_literal(before_eq) + + "=" + + leading_ws + + replacement_value + + escape_jinja_literal(comment_part) ) out_lines.append(new_content + newline) diff --git a/src/jinjaturtle/handlers/postfix.py b/src/jinjaturtle/handlers/postfix.py index e2e442a..f6a1974 100644 --- a/src/jinjaturtle/handlers/postfix.py +++ b/src/jinjaturtle/handlers/postfix.py @@ -5,6 +5,7 @@ from typing import Any from . import BaseHandler from .. import j2 +from ..escape import escape_jinja_literal class PostfixMainHandler(BaseHandler): @@ -108,16 +109,16 @@ class PostfixMainHandler(BaseHandler): stripped = content.strip() if not stripped: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) i += 1 continue if stripped.startswith("#"): - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) i += 1 continue if "=" not in content: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) i += 1 continue @@ -127,7 +128,7 @@ class PostfixMainHandler(BaseHandler): key = before_eq.strip() if not key: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) i += 1 continue @@ -162,16 +163,21 @@ class PostfixMainHandler(BaseHandler): var = self.make_var_name(role_prefix, (key,)) v = value + # ``before_eq`` (key) and ``comment_part`` are source-derived and may + # contain template metacharacters; escape each around the safe + # placeholder. + safe_before = escape_jinja_literal(before_eq) + safe_comment = escape_jinja_literal(comment_part) quoted = len(v) >= 2 and v[0] == v[-1] and v[0] in {'"', "'"} if quoted: replacement = ( - f"{before_eq}={leading_ws}{j2.quoted_variable(var)}" - f"{comment_part}{newline}" + f"{safe_before}={leading_ws}{j2.quoted_variable(var)}" + f"{safe_comment}{newline}" ) else: replacement = ( - f"{before_eq}={leading_ws}{j2.variable(var)}" - f"{comment_part}{newline}" + f"{safe_before}={leading_ws}{j2.variable(var)}" + f"{safe_comment}{newline}" ) out_lines.append(replacement) diff --git a/src/jinjaturtle/handlers/ssh.py b/src/jinjaturtle/handlers/ssh.py index a6c0d2f..f588a33 100644 --- a/src/jinjaturtle/handlers/ssh.py +++ b/src/jinjaturtle/handlers/ssh.py @@ -7,6 +7,7 @@ from typing import Any from . import BaseHandler from .. import j2 +from ..escape import escape_jinja_literal _SECTION_KEYWORDS = {"host", "match"} @@ -255,12 +256,12 @@ class SshConfigHandler(BaseHandler): out_lines: list[str] = [] for ln in parsed.lines: if ln.kind != "kv": - out_lines.append(ln.raw) + out_lines.append(escape_jinja_literal(ln.raw)) continue path = self._path_for_line(ln) if not path: - out_lines.append(ln.raw) + out_lines.append(escape_jinja_literal(ln.raw)) continue var = self.make_var_name(role_prefix, path) @@ -270,9 +271,12 @@ class SshConfigHandler(BaseHandler): else: replacement_value = j2.variable(var) + # ``before_value`` (keyword + spacing) and ``comment`` are + # source-derived; escape around the safe placeholder. rendered = ( - f"{ln.before_value}{replacement_value}" - f"{ln.whitespace_before_comment}{ln.comment}{ln.newline}" + f"{escape_jinja_literal(ln.before_value)}{replacement_value}" + f"{ln.whitespace_before_comment}" + f"{escape_jinja_literal(ln.comment)}{ln.newline}" ) out_lines.append(rendered) diff --git a/src/jinjaturtle/handlers/systemd.py b/src/jinjaturtle/handlers/systemd.py index 8118808..d28a8c5 100644 --- a/src/jinjaturtle/handlers/systemd.py +++ b/src/jinjaturtle/handlers/systemd.py @@ -6,6 +6,7 @@ from typing import Any from . import BaseHandler from .. import j2 +from ..escape import escape_jinja_literal @dataclass @@ -157,7 +158,13 @@ class SystemdUnitHandler(BaseHandler): out_lines: list[str] = [] for ln in parsed.lines: if ln.kind != "kv" or not ln.section or not ln.key: - out_lines.append(ln.raw) + # Verbatim lines (blank/comment/section/unrecognised "raw") + # originate from the source file. Escape template + # metacharacters so they cannot become live template code. + # This is the only handler that emits unrecognised lines, which + # is where Jinja *statement* injection (``{% ... %}``) was + # possible, so escaping here is essential. + out_lines.append(escape_jinja_literal(ln.raw)) continue path: tuple[str, ...] = (ln.section, ln.key) @@ -166,16 +173,18 @@ class SystemdUnitHandler(BaseHandler): var = self.make_var_name(role_prefix, path) v = (ln.value or "").strip() + safe_before = escape_jinja_literal(ln.before_eq) + safe_comment = escape_jinja_literal(ln.comment) quoted = len(v) >= 2 and v[0] == v[-1] and v[0] in {'"', "'"} if quoted: repl = ( - f"{ln.before_eq}={ln.leading_ws_after_eq}" - f"{j2.quoted_variable(var)}{ln.comment}" + f"{safe_before}={ln.leading_ws_after_eq}" + f"{j2.quoted_variable(var)}{safe_comment}" ) else: repl = ( - f"{ln.before_eq}={ln.leading_ws_after_eq}" - f"{j2.variable(var)}{ln.comment}" + f"{safe_before}={ln.leading_ws_after_eq}" + f"{j2.variable(var)}{safe_comment}" ) newline = "\n" if ln.raw.endswith("\n") else "" diff --git a/src/jinjaturtle/handlers/toml.py b/src/jinjaturtle/handlers/toml.py index 15d5a2a..6d25cff 100644 --- a/src/jinjaturtle/handlers/toml.py +++ b/src/jinjaturtle/handlers/toml.py @@ -5,6 +5,7 @@ from typing import Any from . import DictLikeHandler from .. import j2 +from ..escape import escape_jinja_literal from ..loop_analyzer import LoopCandidate try: @@ -77,19 +78,25 @@ class TomlHandler(DictLikeHandler): def emit_kv(path: tuple[str, ...], key: str, value: Any) -> None: var_name = self.make_var_name(role_prefix, path + (key,)) if isinstance(value, str): - lines.append(f"{key} = {self._toml_quoted_expr(var_name)}") + lines.append( + f"{escape_jinja_literal(str(key))} = {self._toml_quoted_expr(var_name)}" + ) elif isinstance(value, bool): # Booleans need | lower filter (Python True/False → TOML true/false) - lines.append(f"{key} = {self._toml_value_expr(var_name, value)}") + lines.append( + f"{escape_jinja_literal(str(key))} = {self._toml_value_expr(var_name, value)}" + ) else: - lines.append(f"{key} = {self._toml_value_expr(var_name, value)}") + lines.append( + f"{escape_jinja_literal(str(key))} = {self._toml_value_expr(var_name, value)}" + ) def walk(obj: dict[str, Any], path: tuple[str, ...] = ()) -> None: scalar_items = {k: v for k, v in obj.items() if not isinstance(v, dict)} nested_items = {k: v for k, v in obj.items() if isinstance(v, dict)} if path: - header = ".".join(path) + header = ".".join(escape_jinja_literal(str(p)) for p in path) lines.append(f"[{header}]") for key, val in scalar_items.items(): @@ -130,10 +137,14 @@ class TomlHandler(DictLikeHandler): def emit_kv(path: tuple[str, ...], key: str, value: Any) -> None: var_name = self.make_var_name(role_prefix, path + (key,)) if isinstance(value, str): - lines.append(f"{key} = {self._toml_quoted_expr(var_name)}") + lines.append( + f"{escape_jinja_literal(str(key))} = {self._toml_quoted_expr(var_name)}" + ) elif isinstance(value, bool): # Booleans need | lower filter (Python True/False → TOML true/false) - lines.append(f"{key} = {self._toml_value_expr(var_name, value)}") + lines.append( + f"{escape_jinja_literal(str(key))} = {self._toml_value_expr(var_name, value)}" + ) elif isinstance(value, list): # Check if this list is a loop candidate if path + (key,) in loop_paths: @@ -157,19 +168,26 @@ class TomlHandler(DictLikeHandler): elif candidate.item_schema in ("simple_dict", "nested"): # Dict list loop - TOML array of tables # This is complex for TOML, using simplified approach - lines.append(f"{key} = " f"{j2.to_json(var_name)}") + lines.append( + f"{escape_jinja_literal(str(key))} = " + f"{j2.to_json(var_name)}" + ) else: # Not a loop, treat as regular variable - lines.append(f"{key} = {self._toml_value_expr(var_name, value)}") + lines.append( + f"{escape_jinja_literal(str(key))} = {self._toml_value_expr(var_name, value)}" + ) else: - lines.append(f"{key} = {self._toml_value_expr(var_name, value)}") + lines.append( + f"{escape_jinja_literal(str(key))} = {self._toml_value_expr(var_name, value)}" + ) def walk(obj: dict[str, Any], path: tuple[str, ...] = ()) -> None: scalar_items = {k: v for k, v in obj.items() if not isinstance(v, dict)} nested_items = {k: v for k, v in obj.items() if isinstance(v, dict)} if path: - header = ".".join(path) + header = ".".join(escape_jinja_literal(str(p)) for p in path) lines.append(f"[{header}]") for key, val in scalar_items.items(): @@ -217,7 +235,7 @@ class TomlHandler(DictLikeHandler): # Blank or pure comment if not stripped or stripped.startswith("#"): - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Table header: [server] or [server.tls] or [[array.of.tables]] @@ -230,7 +248,7 @@ class TomlHandler(DictLikeHandler): inner = inner.strip("[]") # handle [[table]] as well parts = [p.strip() for p in inner.split(".") if p.strip()] current_table = tuple(parts) - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Try key = value @@ -245,7 +263,7 @@ class TomlHandler(DictLikeHandler): eq_index = content.find("=") if eq_index == -1: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue before_eq = content[:eq_index] @@ -253,7 +271,7 @@ class TomlHandler(DictLikeHandler): key = before_eq.strip() if not key: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Whitespace after '=' @@ -289,19 +307,23 @@ class TomlHandler(DictLikeHandler): nested_var = self.make_var_name(role_prefix, nested_path) if isinstance(sub_val, str): inner_bits.append( - f"{sub_key} = {self._toml_quoted_expr(nested_var)}" + f"{escape_jinja_literal(str(sub_key))} = {self._toml_quoted_expr(nested_var)}" ) elif isinstance(sub_val, bool): inner_bits.append( - f"{sub_key} = {self._toml_value_expr(nested_var, sub_val)}" + f"{escape_jinja_literal(str(sub_key))} = {self._toml_value_expr(nested_var, sub_val)}" ) else: inner_bits.append( - f"{sub_key} = {self._toml_value_expr(nested_var, sub_val)}" + f"{escape_jinja_literal(str(sub_key))} = {self._toml_value_expr(nested_var, sub_val)}" ) replacement_value = "{ " + ", ".join(inner_bits) + " }" new_content = ( - before_eq + "=" + leading_ws + replacement_value + comment_part + escape_jinja_literal(before_eq) + + "=" + + leading_ws + + replacement_value + + escape_jinja_literal(comment_part) ) out_lines.append(new_content + newline) continue @@ -327,7 +349,11 @@ class TomlHandler(DictLikeHandler): replacement_value = j2.variable(var_name) new_content = ( - before_eq + "=" + leading_ws + replacement_value + comment_part + escape_jinja_literal(before_eq) + + "=" + + leading_ws + + replacement_value + + escape_jinja_literal(comment_part) ) out_lines.append(new_content + newline) @@ -355,7 +381,7 @@ class TomlHandler(DictLikeHandler): if not stripped or stripped.startswith("#"): # Only output if we're not skipping if not skip_until_next_table: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Table header: [server] or [server.tls] or [[array.of.tables]] @@ -404,7 +430,9 @@ class TomlHandler(DictLikeHandler): out_lines.append( f"{j2.for_start(item_var, collection_var)}\n" ) - out_lines.append(f"[[{'.'.join(table_path)}]]\n") + out_lines.append( + f"[[{'.'.join(escape_jinja_literal(str(p)) for p in table_path)}]]\n" + ) # Add fields from sample item for key, value in sample_item.items(): @@ -412,17 +440,17 @@ class TomlHandler(DictLikeHandler): continue if isinstance(value, str): out_lines.append( - f"{key} = " + f"{escape_jinja_literal(str(key))} = " f"{self._toml_quoted_expr(f'{item_var}.{key}')}\n" ) elif isinstance(value, bool): out_lines.append( - f"{key} = " + f"{escape_jinja_literal(str(key))} = " f"{self._toml_value_expr(f'{item_var}.{key}', value)}\n" ) else: out_lines.append( - f"{key} = " + f"{escape_jinja_literal(str(key))} = " f"{self._toml_value_expr(f'{item_var}.{key}', value)}\n" ) @@ -437,7 +465,7 @@ class TomlHandler(DictLikeHandler): skip_until_next_table = False current_table = table_path - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # If we're inside a skipped array-of-tables section, skip this line @@ -456,7 +484,7 @@ class TomlHandler(DictLikeHandler): eq_index = content.find("=") if eq_index == -1: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue before_eq = content[:eq_index] @@ -464,7 +492,7 @@ class TomlHandler(DictLikeHandler): key = before_eq.strip() if not key: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Whitespace after '=' @@ -501,7 +529,11 @@ class TomlHandler(DictLikeHandler): replacement_value = j2.to_json(collection_var) new_content = ( - before_eq + "=" + leading_ws + replacement_value + comment_part + escape_jinja_literal(before_eq) + + "=" + + leading_ws + + replacement_value + + escape_jinja_literal(comment_part) ) out_lines.append(new_content + newline) continue @@ -526,19 +558,23 @@ class TomlHandler(DictLikeHandler): nested_var = self.make_var_name(role_prefix, nested_path) if isinstance(sub_val, str): inner_bits.append( - f"{sub_key} = {self._toml_quoted_expr(nested_var)}" + f"{escape_jinja_literal(str(sub_key))} = {self._toml_quoted_expr(nested_var)}" ) elif isinstance(sub_val, bool): inner_bits.append( - f"{sub_key} = {self._toml_value_expr(nested_var, sub_val)}" + f"{escape_jinja_literal(str(sub_key))} = {self._toml_value_expr(nested_var, sub_val)}" ) else: inner_bits.append( - f"{sub_key} = {self._toml_value_expr(nested_var, sub_val)}" + f"{escape_jinja_literal(str(sub_key))} = {self._toml_value_expr(nested_var, sub_val)}" ) replacement_value = "{ " + ", ".join(inner_bits) + " }" new_content = ( - before_eq + "=" + leading_ws + replacement_value + comment_part + escape_jinja_literal(before_eq) + + "=" + + leading_ws + + replacement_value + + escape_jinja_literal(comment_part) ) out_lines.append(new_content + newline) continue @@ -564,7 +600,11 @@ class TomlHandler(DictLikeHandler): replacement_value = j2.variable(var_name) new_content = ( - before_eq + "=" + leading_ws + replacement_value + comment_part + escape_jinja_literal(before_eq) + + "=" + + leading_ws + + replacement_value + + escape_jinja_literal(comment_part) ) out_lines.append(new_content + newline) diff --git a/src/jinjaturtle/handlers/xml.py b/src/jinjaturtle/handlers/xml.py index 51c7a73..9fde3ec 100644 --- a/src/jinjaturtle/handlers/xml.py +++ b/src/jinjaturtle/handlers/xml.py @@ -7,6 +7,7 @@ import xml.etree.ElementTree as ET # nosec from .base import BaseHandler from .. import j2 +from ..escape import escape_jinja_literal from ..loop_analyzer import LoopCandidate @@ -230,6 +231,37 @@ 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) + + def _escape_source_comments(self, root: ET.Element) -> None: + """Escape template metacharacters in comments preserved from the source. + + XML comments are re-emitted verbatim by ``ET.tostring`` (the tree is + parsed with ``insert_comments=True``). Attacker-controlled comment text + such as ```` would otherwise become live + template code. Element/attribute *names* cannot carry Jinja delimiters + (XML naming rules forbid the characters and the parser rejects them), and + text/attribute *values* are already replaced with ``{{ var }}`` + 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. + """ + # ET represents comments with a callable tag (ET.Comment). Iterate all + # descendants and escape comment text that is not one of our markers. + for elem in root.iter(): + if elem.tag is ET.Comment: + if not self._is_jt_marker(elem.text or ""): + elem.text = escape_jinja_literal(elem.text or "") + def _generate_xml_template_from_text(self, role_prefix: str, text: str) -> str: """Generate scalar-only Jinja2 template.""" prolog, body = self._split_xml_prolog(text) @@ -240,12 +272,16 @@ class XmlHandler(BaseHandler): self._apply_jinja_to_xml_tree(role_prefix, root) + # Neutralise template metacharacters in any comments preserved from the + # source file before serialising. + self._escape_source_comments(root) + indent = getattr(ET, "indent", None) if indent is not None: indent(root, space=" ") # type: ignore[arg-type] xml_body = ET.tostring(root, encoding="unicode") - return prolog + xml_body + return escape_jinja_literal(prolog) + xml_body def _generate_xml_template_with_loops_from_text( self, @@ -265,6 +301,11 @@ class XmlHandler(BaseHandler): # Apply Jinja transformations (including loop markers) self._apply_jinja_to_xml_tree(role_prefix, root, loop_candidates) + # Escape comments preserved from the source. JinjaTurtle's own + # LOOP/IF/ENDIF marker comments are recognised and left intact so they + # can be converted into real Jinja control structures below. + self._escape_source_comments(root) + # Convert to string indent = getattr(ET, "indent", None) if indent is not None: @@ -275,7 +316,7 @@ class XmlHandler(BaseHandler): # Post-process to replace loop markers with actual Jinja loops xml_body = self._insert_xml_loops(xml_body, role_prefix, loop_candidates, root) - return prolog + xml_body + return escape_jinja_literal(prolog) + xml_body def _insert_xml_loops( self, diff --git a/src/jinjaturtle/handlers/yaml.py b/src/jinjaturtle/handlers/yaml.py index 1ad9bf3..d8131b0 100644 --- a/src/jinjaturtle/handlers/yaml.py +++ b/src/jinjaturtle/handlers/yaml.py @@ -6,6 +6,7 @@ from typing import Any from .dict import DictLikeHandler from .. import j2 +from ..escape import escape_jinja_literal from ..loop_analyzer import LoopCandidate @@ -102,7 +103,7 @@ class YamlHandler(DictLikeHandler): indent = len(raw_line) - len(stripped) if not stripped or stripped.startswith("#"): - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue while stack and indent < stack[-1][0]: @@ -112,7 +113,7 @@ class YamlHandler(DictLikeHandler): key_part, rest = stripped.split(":", 1) key = key_part.strip() if not key: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue rest_stripped = rest.lstrip(" \t") @@ -125,7 +126,7 @@ class YamlHandler(DictLikeHandler): stack.append((indent, path, "map")) if not has_value: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue value_part, comment_part = self._split_inline_comment( @@ -147,8 +148,8 @@ class YamlHandler(DictLikeHandler): replacement = self._yaml_scalar_expr(var_name, raw_value) leading = rest[: len(rest) - len(rest.lstrip(" \t"))] - new_rest = f"{leading}{replacement}{comment_part}" - new_stripped = f"{key}:{new_rest}" + new_rest = f"{leading}{replacement}{escape_jinja_literal(comment_part)}" + new_stripped = f"{escape_jinja_literal(key)}:{new_rest}" out_lines.append( " " * indent + new_stripped @@ -185,7 +186,7 @@ class YamlHandler(DictLikeHandler): else: replacement = self._yaml_scalar_expr(var_name, raw_value) - new_stripped = f"- {replacement}{comment_part}" + new_stripped = f"- {replacement}{escape_jinja_literal(comment_part)}" out_lines.append( " " * indent + new_stripped @@ -193,7 +194,7 @@ class YamlHandler(DictLikeHandler): ) continue - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) return "".join(out_lines) @@ -275,7 +276,7 @@ class YamlHandler(DictLikeHandler): next_line = next_significant_line(line_index) if next_line is None: skip_until_indent = None - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) else: next_indent, next_stripped = next_line still_in_collection = next_indent > skip_until_indent or ( @@ -284,12 +285,12 @@ class YamlHandler(DictLikeHandler): ) if not still_in_collection: skip_until_indent = None - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue if is_comment: if indent <= skip_until_indent: skip_until_indent = None - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) # Comments/blank lines indented beneath the replaced # collection are considered part of that collection and # cannot be placed safely inside a generated loop. @@ -303,7 +304,7 @@ class YamlHandler(DictLikeHandler): # Blank or comment lines if is_blank or is_comment: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Adjust stack based on indent @@ -315,7 +316,7 @@ class YamlHandler(DictLikeHandler): key_part, rest = stripped.split(":", 1) key = key_part.strip() if not key: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue rest_stripped = rest.lstrip(" \t") @@ -353,7 +354,7 @@ class YamlHandler(DictLikeHandler): continue if not has_value: - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) continue # Scalar value - replace with variable @@ -376,8 +377,8 @@ class YamlHandler(DictLikeHandler): replacement = self._yaml_scalar_expr(var_name, raw_value) leading = rest[: len(rest) - len(rest.lstrip(" \t"))] - new_rest = f"{leading}{replacement}{comment_part}" - new_stripped = f"{key}:{new_rest}" + new_rest = f"{leading}{replacement}{escape_jinja_literal(comment_part)}" + new_stripped = f"{escape_jinja_literal(key)}:{new_rest}" out_lines.append( " " * indent + new_stripped @@ -439,7 +440,7 @@ class YamlHandler(DictLikeHandler): else: replacement = self._yaml_scalar_expr(var_name, raw_value) - new_stripped = f"- {replacement}{comment_part}" + new_stripped = f"- {replacement}{escape_jinja_literal(comment_part)}" out_lines.append( " " * indent + new_stripped @@ -447,7 +448,7 @@ class YamlHandler(DictLikeHandler): ) continue - out_lines.append(raw_line) + out_lines.append(escape_jinja_literal(raw_line)) return "".join(out_lines) @@ -480,7 +481,7 @@ class YamlHandler(DictLikeHandler): lines: list[str] = [] if not is_list: key = candidate.path[-1] if candidate.path else "items" - lines.append(f"{indent_str}{key}:") + lines.append(f"{indent_str}{escape_jinja_literal(str(key))}:") item_lines: list[str] = [] if candidate.items: @@ -549,12 +550,16 @@ class YamlHandler(DictLikeHandler): if first_key and is_list_item: # First key gets the list marker value_expr = self._yaml_value_expr(f"{loop_var}.{key}", value) - lines.append(f"{indent_str}- {key}: {value_expr}") + lines.append( + f"{indent_str}- {escape_jinja_literal(str(key))}: {value_expr}" + ) first_key = False else: # Subsequent keys are indented sub_indent = indent + 2 if is_list_item else indent value_expr = self._yaml_value_expr(f"{loop_var}.{key}", value) - lines.append(f"{' ' * sub_indent}{key}: {value_expr}") + lines.append( + f"{' ' * sub_indent}{escape_jinja_literal(str(key))}: {value_expr}" + ) return lines diff --git a/tests/test_injection_security.py b/tests/test_injection_security.py new file mode 100644 index 0000000..642c0b8 --- /dev/null +++ b/tests/test_injection_security.py @@ -0,0 +1,212 @@ +"""Security regression tests for template injection (SSTI). + +JinjaTurtle copies parts of the source config (comments, unrecognised lines, +structural keys) verbatim into the generated template. If that text contains +Jinja2/ERB delimiters it must be neutralised, otherwise attacker-influenced +config content becomes live template code that executes when Salt/Ansible/Puppet +later renders the template. + +These tests render the *generated* template the way a downstream tool would and +assert that an injected payload never executes. A tripwire object is exposed +under every name a payload might reference; if the rendered output ever contains +the tripwire sentinel, an injected expression executed and the test fails. +""" + +from __future__ import annotations + +import re +import subprocess +import sys +from pathlib import Path + +import jinja2 +import pytest +import yaml as pyyaml + +from jinjaturtle.escape import escape_jinja_literal, escape_erb_literal + + +TRIP = "__TRIPWIRE_FIRED__" + + +class _Boom: + """Returns the tripwire sentinel for any access/call an SSTI payload makes.""" + + def run(self, *a, **k): + return TRIP + + def __call__(self, *a, **k): + return TRIP + + def __getitem__(self, k): + return self + + def __getattr__(self, n): + return _Boom() + + def __str__(self): + return TRIP + + +def _render_jinja(template_text: str, defaults: dict) -> str: + env = jinja2.Environment(undefined=jinja2.ChainableUndefined) + env.filters.setdefault("to_json", lambda v, **k: __import__("json").dumps(v)) + env.filters.setdefault("lower", lambda v: str(v).lower()) + ctx = dict(defaults or {}) + for name in ("salt", "cmd", "os", "subprocess", "cycler", "lipsum", "namespace"): + ctx.setdefault(name, _Boom()) + return env.from_string(template_text).render(**ctx) + + +def _run_jinjaturtle(tmp_path: Path, source_name: str, body: str, fmt: str): + src = tmp_path / source_name + src.write_text(body, encoding="utf-8") + tpl = tmp_path / "out.tpl" + dfl = tmp_path / "defaults.yml" + res = subprocess.run( + [ + sys.executable, + "-m", + "jinjaturtle.cli", + str(src), + "-f", + fmt, + "--role-name", + "role", + "-t", + str(tpl), + "-d", + str(dfl), + ], + capture_output=True, + text=True, + ) + assert res.returncode == 0, f"generation failed: {res.stderr}" + defaults = pyyaml.safe_load(dfl.read_text()) or {} + return tpl.read_text(), defaults + + +# A representative payload for each format, placed where the format allows +# attacker-controlled verbatim text (comments / unrecognised lines). +FORMAT_CASES = [ + ( + "ini", + "evil.ini", + "[s]\n" + "good = ok ; {{ salt['cmd.run']('id') }}\n" + "# {{ cmd.run('whoami') }}\n", + ), + ( + "yaml", + "evil.yaml", + "server:\n" " motd: ok\n" " # {{ salt['cmd.run']('id') }}\n", + ), + ( + "toml", + "evil.toml", + "[s]\n" 'good = "ok"\n' "# {{ cmd.run('id') }}\n", + ), + ( + "xml", + "evil.xml", + "\n" + " \n" + ' ok\n' + "\n", + ), + ( + "postfix", + "main.cf", + "myhostname = mail.example.com\n" "# {{ salt['cmd.run']('id') }}\n", + ), + ( + "systemd", + "evil.service", + "[Unit]\n" + "Description=ok\n" + "# {{ cmd.run('id') }}\n" + "RawLineNoEquals {% for x in ().__class__.__bases__ %}\n" + "[Service]\n" + "ExecStart=/bin/true\n", + ), + ( + "ssh", + "sshd_config", + "# {{ salt['cmd.run']('id') }}\n" "Port 22\n" "PermitRootLogin no\n", + ), +] + + +@pytest.mark.parametrize("fmt,name,body", FORMAT_CASES) +def test_comment_payload_does_not_execute(tmp_path, fmt, name, body): + template_text, defaults = _run_jinjaturtle(tmp_path, name, body, fmt) + rendered = _render_jinja(template_text, defaults) + assert TRIP not in rendered, f"injected payload executed for {fmt}:\n{rendered}" + + +@pytest.mark.parametrize("fmt,name,body", FORMAT_CASES) +def test_generated_template_is_renderable(tmp_path, fmt, name, body): + # A correct escape must still produce a syntactically valid template. + template_text, defaults = _run_jinjaturtle(tmp_path, name, body, fmt) + # Should not raise a TemplateSyntaxError. + _render_jinja(template_text, defaults) + + +# --- Unit-level guarantees for the escaper itself --------------------------- + +SSTI_PAYLOADS = [ + "{{ 7*7 }}", + "{{ salt['cmd.run']('id') }}", + "{% set x = cycler.__init__.__globals__ %}{{ x }}", + "{# comment payload #}", + "text {% endraw %} breakout {{ evil }}", + "nested {% endraw %} spacing {{ evil }}", + "{%- endraw -%}{{ evil }}", + "mixed {{ a }} and {% b %} and {# c #}", + "}}{{ orphan delimiters %}{%", +] + + +@pytest.mark.parametrize("payload", SSTI_PAYLOADS) +def test_escape_jinja_literal_renders_back_to_original(payload): + """Escaped text must render to the exact original characters, inertly.""" + env = jinja2.Environment(undefined=jinja2.ChainableUndefined) + escaped = escape_jinja_literal(payload) + rendered = env.from_string(escaped).render(evil="EVIL", x="X", a="A") + assert rendered == payload + assert TRIP not in rendered + + +def test_escape_jinja_literal_noop_on_plain_text(): + for plain in ["", "hello world", "# a normal comment", "port = 8080", "key: value"]: + assert escape_jinja_literal(plain) == plain + + +def test_escape_jinja_literal_actually_blocks_execution(): + env = jinja2.Environment(undefined=jinja2.ChainableUndefined) + payload = "{{ boom.run('x') }}" + escaped = escape_jinja_literal(payload) + rendered = env.from_string(escaped).render(boom=_Boom()) + assert TRIP not in rendered + # Sanity: the *unescaped* payload would have fired the tripwire. + fired = env.from_string(payload).render(boom=_Boom()) + assert TRIP in fired + + +@pytest.mark.parametrize( + "payload", + [ + "<%= system('id') %>", + "<% require 'open3' %>", + "text <%= 1+1 %> more", + "-%> orphan <%", + ], +) +def test_escape_erb_literal_removes_executable_tags(payload): + escaped = escape_erb_literal(payload) + # No raw executable ERB tag should survive that contains the original code. + live = re.findall(r"<%[-=#]?(.*?)-?%>", escaped, re.S) + for chunk in live: + assert "system" not in chunk + assert "require" not in chunk + # Numeric/expression payloads must be reduced to literal-string prints.