From f63cb92b35389d604ce078db1e5317c2a2d2a539 Mon Sep 17 00:00:00 2001 From: Miguel Jacq Date: Wed, 24 Jun 2026 20:29:20 +1000 Subject: [PATCH 1/4] Byebye --- README.md | 178 +++++----------------- debian/changelog | 1 + rpm/jinjaturtle.spec | 1 + src/jinjaturtle/cli.py | 123 ++++++---------- src/jinjaturtle/core.py | 243 +++++++++++++++++++------------ src/jinjaturtle/erb.py | 243 ------------------------------- src/jinjaturtle/escape.py | 114 +++------------ src/jinjaturtle/handlers/base.py | 20 ++- src/jinjaturtle/handlers/json.py | 74 +++++----- src/jinjaturtle/handlers/xml.py | 62 ++++---- src/jinjaturtle/multi.py | 27 +++- src/jinjaturtle/safety.py | 51 +------ tests/test_cli.py | 54 ------- tests/test_deep_nesting_dos.py | 122 ++++++++++++++++ tests/test_injection_security.py | 38 +---- tests/test_output_safety_gate.py | 15 -- tests/test_release_blockers.py | 98 +++++++++++++ tests/test_yaml_handler.py | 15 +- 18 files changed, 593 insertions(+), 886 deletions(-) delete mode 100644 src/jinjaturtle/erb.py create mode 100644 tests/test_deep_nesting_dos.py create mode 100644 tests/test_release_blockers.py diff --git a/README.md b/README.md index 2905cfe..bbcda4e 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,11 @@ # JinjaTurtle +## ABANDONED + +Project is abandoned. To many potential security issues. Please uninstall it. Sorry for wasting your time. + +____ +
JinjaTurtle logo
@@ -7,15 +13,12 @@ JinjaTurtle is a command-line tool that helps turn existing native configuration files into reusable configuration-management templates. -By default it generates: +It generates: - a **Jinja2** template; and - an **Ansible defaults YAML** file containing the variables used by that template. -It can also generate **ERB** templates and **Puppet Hiera-style YAML** data for -Puppet workflows. - JinjaTurtle does not try to replace configuration-management tools. Its job is to speed up the boring first pass: take a real config file, discover the values inside it, replace those values with variables, and write the corresponding @@ -26,8 +29,6 @@ variable data beside the template. JinjaTurtle examines a source config file and keeps the original structure as much as possible. -For the default Jinja2/Ansible mode: - 1. The config file is parsed. 2. Variable names are generated from the config keys and paths. 3. Those variable names are prefixed with `--role-name`, which should usually @@ -37,16 +38,6 @@ For the default Jinja2/Ansible mode: 5. An Ansible defaults YAML file is generated with those variables and the original values. -For ERB/Puppet mode: - -1. The same parse/flatten/loop analysis is used. -2. An ERB template is generated with Puppet-style instance variables such as - `<%= @memory_limit %>`. -3. The variables file is written as Puppet Hiera-style data, such as - `php::memory_limit: 256M`. -4. If `--puppet-class` is supplied, that class name is used as the Hiera - namespace while `--role-name` remains the local variable prefix. - By default, the generated variable data and template are printed to stdout. Use `--defaults-output` and `--template-output` to write them to files. @@ -80,78 +71,6 @@ and defaults data like: php_memory_limit: 256M ``` -## ERB / Puppet example - -Use `--template-engine erb` when you want Puppet ERB output: - -```shell -jinjaturtle php.ini \ - --role-name php \ - --template-engine erb \ - --defaults-output data/common.yaml \ - --template-output templates/php.ini.erb -``` - -Given the same source value: - -```ini -memory_limit = 256M -``` - -JinjaTurtle will produce an ERB template value like: - -```erb -memory_limit = <%= @memory_limit %> -``` - -and Hiera-style data like: - -```yaml -php::memory_limit: 256M -``` - -The `--defaults-output` option name is retained for CLI compatibility, but in -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, should still declare the class parameters and call the template, for -example with Puppet's `template()` function. - -## Using `--puppet-class` - -Most direct usage can simply use the same value for the role name and Puppet -class name: - -```shell -jinjaturtle php.ini --role-name php --template-engine erb -``` - -This creates Hiera keys such as: - -```yaml -php::memory_limit: 256M -``` - -and local ERB variables such as: - -```erb -<%= @memory_limit %> -``` - -For generated systems, it can be useful to make `--role-name` more specific -while keeping the Hiera keys under the real Puppet class. For example: - -```shell -jinjaturtle php.ini \ - --role-name php_etc_php_ini \ - --puppet-class php \ - --template-engine erb -``` - -In that case the variable prefix can stay file-specific, while the Hiera data -is still written under `php::...`. - ## What sort of config files can it handle? JinjaTurtle supports common structured and semi-structured config formats: @@ -176,15 +95,15 @@ homogeneous enough. If it is not confident, it falls back to flattened scalar variables. Some very complex files will still need manual cleanup. The goal is to speed up -conversion into Jinja2 or ERB templates, not to guarantee a perfect final module -without review. +conversion into Jinja2 templates, not to guarantee a perfect final role without +review. ## JSON, quoting, and type preservation JinjaTurtle tries to preserve rendered config types. -For JSON, it uses JSON-aware expressions rather than plain string substitution. -This avoids generating invalid JSON such as: +For JSON, it uses JSON-aware Jinja2 expressions rather than plain string +substitution. This avoids generating invalid JSON such as: ```json {"enabled": True} @@ -196,18 +115,6 @@ when the correct rendered JSON should be: {"enabled": true} ``` -In Jinja2 mode this uses Ansible-style JSON filters. In ERB mode it emits Ruby -JSON generation where required, for example: - -```erb -<% require 'json' -%> -{ - "enabled": <%= JSON.generate(@enabled) %> -} -``` - -That is expected for JSON ERB templates. - ## Can I convert multiple files at once? Yes. Pass a directory instead of a single file and JinjaTurtle will convert the @@ -287,8 +194,6 @@ poetry install usage: jinjaturtle [-h] [-r ROLE_NAME] [--recursive] [-f {ini,json,toml,yaml,xml,postfix,systemd,ssh}] [-d DEFAULTS_OUTPUT] [-t TEMPLATE_OUTPUT] - [--template-engine {jinja2,erb}] - [--puppet-class PUPPET_CLASS] config Convert a config file into an Ansible defaults file and Jinja2 template. @@ -301,25 +206,18 @@ positional arguments: options: -h, --help show this help message and exit -r, --role-name ROLE_NAME - Role name / variable prefix. In Jinja2 mode this is - usually the Ansible role name. In ERB mode it is used - as the local variable prefix. Defaults to jinjaturtle. + Ansible role name, used as variable prefix. Defaults + to jinjaturtle. --recursive When CONFIG is a folder, recurse into subfolders. -f, --format {ini,json,toml,yaml,xml,postfix,systemd,ssh} Force config format instead of auto-detecting from filename. -d, --defaults-output DEFAULTS_OUTPUT - Path to write the generated variable YAML. If omitted, - it is printed to stdout. + Path to write defaults/main.yml. If omitted, defaults + YAML is printed to stdout. -t, --template-output TEMPLATE_OUTPUT - Path to write the generated config template. If omitted, - it is printed to stdout. - --template-engine {jinja2,erb} - Template syntax to generate. Defaults to jinja2. Use - erb for Puppet templates. - --puppet-class PUPPET_CLASS - Puppet class / Hiera namespace to use with - --template-engine erb. Defaults to --role-name. + Path to write the generated config template. If + omitted, template is printed to stdout. ``` ## Additional supported formats @@ -345,43 +243,37 @@ 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. + `{{ variable }}` placeholder in the template, and the original value is + stored separately in the defaults data. String values in generated defaults + are tagged with Ansible's `!unsafe` tag by default, so a payload sitting + inside a value (for example `motd = {{ lookup('pipe', 'id') }}`) remains + inert even if downstream playbooks accidentally place that value in another + templating context. 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. + into the template. Any Jinja2 metacharacters in that copied text (`{{ }}`, + `{% %}`, `{# #}`) 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: +Treat generated defaults as untrusted input. JinjaTurtle emits extracted +string values with Ansible's `!unsafe` tag, but you should preserve that tag if +you merge the generated data into larger defaults files or transform it with +other tooling. -- **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 strip `!unsafe` tags or +feed the data back through another templating pass. -In short: render JinjaTurtle output exactly once. Do not feed it back through -another templating pass. +Directory mode deliberately ignores symlinks and will not traverse outside the +input directory. This prevents a malicious folder tree from smuggling in +configuration files through links to unrelated filesystem locations. **IMPORTANT**: Always review both the original config files, then the resulting templates generated by JinjaTurtle, before integrating them into your config management system! - ## Found a bug, have a suggestion? You can e-mail me; see `pyproject.toml` for details. You can also contact me on diff --git a/debian/changelog b/debian/changelog index 9173c94..af99a92 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,5 +1,6 @@ jinjaturtle (0.5.7) unstable; urgency=medium + * Remove ERB/Puppet output support and keep Jinja2-only generation. * More hardening measures -- Miguel Jacq Wed, 24 Jun 2026 16:13:00 +1000 diff --git a/rpm/jinjaturtle.spec b/rpm/jinjaturtle.spec index 051fd8b..16ebd66 100644 --- a/rpm/jinjaturtle.spec +++ b/rpm/jinjaturtle.spec @@ -43,6 +43,7 @@ Convert config files into Ansible defaults and Jinja2 templates. %changelog * Wed Jun 24 2026 Miguel Jacq - %{version}-%{release} +- Remove ERB/Puppet output support and keep Jinja2-only generation. - 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 diff --git a/src/jinjaturtle/cli.py b/src/jinjaturtle/cli.py index 6345fae..057abdc 100644 --- a/src/jinjaturtle/cli.py +++ b/src/jinjaturtle/cli.py @@ -1,6 +1,7 @@ from __future__ import annotations import argparse +import os import sys from defusedxml import defuse_stdlib from pathlib import Path @@ -12,12 +13,29 @@ from .core import ( flatten_config, generate_ansible_yaml, generate_jinja2_template, - generate_puppet_hiera_yaml, - generate_erb_template, ) from .multi import process_directory -from .safety import TemplateSafetyError, verify_erb_template_safe +from .safety import TemplateSafetyError + + +def _write_text_no_symlink(path: Path, text: str) -> None: + """Write a user-requested output file without following a final symlink.""" + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + fd = os.open(path, flags, 0o600) + except OSError as exc: + raise OSError(f"refusing or unable to write output file {path}: {exc}") from exc + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(text) + + +def _ensure_output_dir(path: Path) -> None: + if path.exists() and path.is_symlink(): + raise OSError(f"refusing to use symlink as output directory: {path}") + path.mkdir(parents=True, exist_ok=True) def _build_arg_parser() -> argparse.ArgumentParser: @@ -59,20 +77,6 @@ def _build_arg_parser() -> argparse.ArgumentParser: "--template-output", help="Path to write the generated config template. If omitted, template is printed to stdout.", ) - ap.add_argument( - "--template-engine", - choices=[j2.NAME, "erb"], - default=j2.NAME, - help="Template syntax to generate (default: jinja2). Use erb for Puppet templates.", - ) - ap.add_argument( - "--puppet-class", - help=( - "Puppet class/Hiera namespace to use with --template-engine erb. " - "Defaults to --role-name. This lets tools use a file-specific " - "variable prefix while writing Hiera keys under the real Puppet class." - ), - ) return ap @@ -88,6 +92,9 @@ def _main(argv: list[str] | None = None) -> int: f"jinjaturtle: refusing to generate unsafe template: {exc}", file=sys.stderr ) return 2 + except (OSError, TypeError, ValueError) as exc: + print(f"jinjaturtle: error: {exc}", file=sys.stderr) + return 1 def _run(argv: list[str] | None = None) -> int: @@ -105,51 +112,36 @@ 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_no_symlink(Path(args.defaults_output), defaults_yaml) else: print("# defaults/main.yml") print(defaults_yaml, end="") - # Optionally translate folder-mode templates to ERB. Folder mode keeps - # the existing data shape; single-file mode below is the preferred - # Puppet path because it can produce class-parameter Hiera keys. - if args.template_engine == "erb": - from .erb import translate_jinja2_to_erb - - for o in outputs: - o.template = translate_jinja2_to_erb( - o.template, - role_prefix=args.role_name, - puppet_class=args.puppet_class or args.role_name, - ) - verify_erb_template_safe(o.template) - - template_ext = "erb" if args.template_engine == "erb" else j2.TEMPLATE_EXTENSION - # Write templates 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_no_symlink(out_path, outputs[0].template) else: - out_path.mkdir(parents=True, exist_ok=True) + _ensure_output_dir(out_path) for o in outputs: - (out_path / f"config.{o.fmt}.{template_ext}").write_text( - o.template, encoding="utf-8" + _write_text_no_symlink( + out_path / f"config.{o.fmt}.{j2.TEMPLATE_EXTENSION}", + o.template, ) else: for o in outputs: name = ( - f"config.{template_ext}" + f"config.{j2.TEMPLATE_EXTENSION}" if len(outputs) == 1 - else f"config.{o.fmt}.{template_ext}" + else f"config.{o.fmt}.{j2.TEMPLATE_EXTENSION}" ) print(f"# {name}") print(o.template, end="") return 0 - # Single-file mode (existing behaviour) + # Single-file mode config_text = config_path.read_text(encoding="utf-8") # Parse the config @@ -161,51 +153,28 @@ def _run(argv: list[str] | None = None) -> int: # Flatten config (excluding loop paths if loops are detected) flat_items = flatten_config(fmt, parsed, loop_candidates) - if args.template_engine == "erb": - ansible_yaml = generate_puppet_hiera_yaml( - args.role_name, - flat_items, - loop_candidates, - puppet_class=args.puppet_class or args.role_name, - ) - template_str = generate_erb_template( - fmt, - parsed, - args.role_name, - original_text=config_text, - loop_candidates=loop_candidates, - flat_items=flat_items, - puppet_class=args.puppet_class or args.role_name, - ) - else: - # Generate defaults YAML (with loop collections if detected) - ansible_yaml = generate_ansible_yaml( - args.role_name, flat_items, loop_candidates - ) + # Generate defaults YAML (with loop collections if detected) + ansible_yaml = generate_ansible_yaml(args.role_name, flat_items, loop_candidates) - # Generate template (with loops if detected) - template_str = generate_jinja2_template( - fmt, - parsed, - args.role_name, - original_text=config_text, - loop_candidates=loop_candidates, - ) + # Generate template (with loops if detected) + template_str = generate_jinja2_template( + fmt, + parsed, + args.role_name, + original_text=config_text, + loop_candidates=loop_candidates, + ) if args.defaults_output: - Path(args.defaults_output).write_text(ansible_yaml, encoding="utf-8") + _write_text_no_symlink(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_no_symlink(Path(args.template_output), template_str) else: - print( - "# config.erb" - if args.template_engine == "erb" - else f"# config.{j2.TEMPLATE_EXTENSION}" - ) + print(f"# config.{j2.TEMPLATE_EXTENSION}") print(template_str, end="") return 0 diff --git a/src/jinjaturtle/core.py b/src/jinjaturtle/core.py index 4d35260..8f7d9cf 100644 --- a/src/jinjaturtle/core.py +++ b/src/jinjaturtle/core.py @@ -7,12 +7,39 @@ import datetime import re import yaml + +class AnsibleUnsafeString(str): + """Marker for strings emitted with Ansible's !unsafe YAML tag.""" + + pass + + +class VariableNameCollisionError(ValueError): + """Raised when distinct config paths collapse to the same variable name.""" + + pass + + +class ConfigTooDeeplyNestedError(ValueError): + """Raised when a parsed config nests deeper than ``MAX_CONFIG_DEPTH``. + + Pathologically nested input (thousands of levels) would otherwise drive the + recursive walkers in this module past Python's stack limit and raise an + uncaught ``RecursionError``. We bound the depth here and fail with a clear, + catchable error instead. The limit is well above any plausible real config + while staying comfortably under the interpreter's default recursion ceiling. + """ + + pass + + +# Maximum container-nesting depth accepted from a parsed config. Real-world +# configuration files are nowhere near this deep; the cap exists purely to turn +# a hostile/degenerate input into a clean error rather than a stack overflow. +MAX_CONFIG_DEPTH = 200 + from .loop_analyzer import LoopAnalyzer, LoopCandidate -from .erb import puppet_class_name, puppet_local_var_name, translate_jinja2_to_erb -from .safety import ( - verify_erb_template_safe, - verify_jinja2_template_safe, -) +from .safety import verify_jinja2_template_safe from .handlers import ( BaseHandler, IniHandler, @@ -53,7 +80,38 @@ 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)) + + +def _ansible_unsafe_str_constructor( + loader: yaml.SafeLoader, node: yaml.nodes.ScalarNode +) -> str: + return loader.construct_scalar(node) + + +def _mark_ansible_unsafe_values(data: Any) -> Any: + """Recursively tag string *values* as Ansible !unsafe. + + Mapping keys remain ordinary strings because they are variable names in the + generated defaults document, not attacker-controlled config values. + """ + if isinstance(data, AnsibleUnsafeString): + return data + if isinstance(data, str): + return AnsibleUnsafeString(data) + if isinstance(data, dict): + return {k: _mark_ansible_unsafe_values(v) for k, v in data.items()} + if isinstance(data, list): + return [_mark_ansible_unsafe_values(v) for v in data] + if isinstance(data, tuple): + return tuple(_mark_ansible_unsafe_values(v) for v in data) + return data + + _TurtleDumper.add_representer(QuotedString, _quoted_str_representer) +_TurtleDumper.add_representer(AnsibleUnsafeString, _ansible_unsafe_str_representer) +yaml.SafeLoader.add_constructor("!unsafe", _ansible_unsafe_str_constructor) # Use our fallback for any unknown object types _TurtleDumper.add_representer(None, _fallback_str_representer) @@ -83,10 +141,12 @@ _HANDLERS["ssh"] = _SSH_HANDLER def dump_yaml(data: Any, *, sort_keys: bool = True) -> str: """Dump YAML using JinjaTurtle's dumper settings. - This is used by both the single-file and multi-file code paths. + String values extracted from input configuration are emitted as Ansible + ``!unsafe`` values so a downstream playbook cannot accidentally re-template + malicious Jinja syntax stored in defaults. Mapping keys are left untouched. """ return yaml.dump( - data, + _mark_ansible_unsafe_values(data), Dumper=_TurtleDumper, sort_keys=sort_keys, default_flow_style=False, @@ -263,6 +323,31 @@ def detect_format(path: Path, explicit: str | None = None) -> str: return "ini" +def _check_xml_depth(elem: Any, _depth: int = 0) -> None: + """Enforce :data:`MAX_CONFIG_DEPTH` on an ElementTree element. + + XML parses to an ``Element`` rather than dict/list containers, so it is not + covered by the depth bound in :func:`_stringify_timestamps`. Without this + check a pathologically nested document would drive the XML flattener and the + loop analyzer's ``_analyze_xml`` / ``_xml_elem_to_dict`` walkers into an + uncaught ``RecursionError``. We walk the tree iteratively (an explicit + stack, so the check itself cannot overflow) and fail loudly at the limit. + """ + # (element, depth) pairs; iterative traversal avoids its own recursion. + stack = [(elem, 0)] + while stack: + node, depth = stack.pop() + if depth > MAX_CONFIG_DEPTH: + raise ConfigTooDeeplyNestedError( + f"XML config nests deeper than the supported maximum of " + f"{MAX_CONFIG_DEPTH} levels; refusing to process it" + ) + for child in list(node): + # Skip comment/PI nodes whose tag is not a str. + if isinstance(getattr(child, "tag", None), str): + stack.append((child, depth + 1)) + + def parse_config(path: Path, fmt: str | None = None) -> tuple[str, Any]: """ Parse config file into a Python object. @@ -271,7 +356,22 @@ def parse_config(path: Path, fmt: str | None = None) -> tuple[str, Any]: handler = _HANDLERS.get(fmt) if handler is None: raise ValueError(f"Unsupported config format: {fmt}") - parsed = handler.parse(path) + try: + parsed = handler.parse(path) + except RecursionError as exc: + # Some underlying parsers (e.g. PyYAML, json) recurse while *building* + # the object, so they can exhaust the stack before our own depth guards + # below ever run. Convert that into the same clean, catchable error. + raise ConfigTooDeeplyNestedError( + f"config in {path} is nested too deeply for the parser to handle; " + "refusing to process it" + ) from exc + # Bound nesting depth before any recursive walker runs, turning degenerate + # deeply-nested input into a clean, catchable error instead of a stack + # overflow. dict/list configs are bounded inside _stringify_timestamps; + # XML (an Element tree) is bounded explicitly here. + if hasattr(parsed, "tag"): # ElementTree Element + _check_xml_depth(parsed) # Make sure datetime objects are treated as strings (TOML, YAML) parsed = _stringify_timestamps(parsed) @@ -341,6 +441,31 @@ def _path_starts_with(path: tuple[str, ...], prefix: tuple[str, ...]) -> bool: return path[: len(prefix)] == prefix +def check_var_name_collisions( + role_prefix: str, items: Iterable[tuple[tuple[str, ...], Any]] +) -> None: + """Fail if distinct config paths collapse to the same variable name.""" + seen: dict[str, tuple[str, ...]] = {} + for path, _value in items: + path_tuple = tuple(str(p) for p in path) + var_name = make_var_name(role_prefix, path_tuple) + previous = seen.get(var_name) + if previous is not None and previous != path_tuple: + raise VariableNameCollisionError( + "distinct config paths collapse to the same variable name " + f"{var_name!r}: {previous!r} and {path_tuple!r}" + ) + seen[var_name] = path_tuple + + +def _flatten_loop_defaults( + loop_candidates: list[LoopCandidate] | None, +) -> list[tuple[tuple[str, ...], Any]]: + if not loop_candidates: + return [] + return [(candidate.path, candidate.items) for candidate in loop_candidates] + + def generate_ansible_yaml( role_prefix: str, flat_items: list[tuple[tuple[str, ...], Any]], @@ -349,6 +474,10 @@ def generate_ansible_yaml( """ Create Ansible YAML for defaults/main.yml. """ + check_var_name_collisions( + role_prefix, flat_items + _flatten_loop_defaults(loop_candidates) + ) + defaults: dict[str, Any] = {} # Add scalar variables @@ -399,87 +528,7 @@ def generate_jinja2_template( return template -def _template_variable_names( - role_prefix: str, - flat_items: list[tuple[tuple[str, ...], Any]], - loop_candidates: list[LoopCandidate] | None = None, -) -> set[str]: - names = {make_var_name(role_prefix, path) for path, _value in flat_items} - if loop_candidates: - for candidate in loop_candidates: - names.add(make_var_name(role_prefix, candidate.path)) - return names - - -def generate_puppet_hiera_yaml( - role_prefix: str, - flat_items: list[tuple[tuple[str, ...], Any]], - loop_candidates: list[LoopCandidate] | None = None, - *, - puppet_class: str | None = None, -) -> str: - """Create Puppet Hiera data suitable for Automatic Parameter Lookup. - - ``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``. - """ - - klass = puppet_class_name(puppet_class or role_prefix) - data: dict[str, Any] = {} - - for path, value in flat_items: - generated = make_var_name(role_prefix, path) - local = puppet_local_var_name(role_prefix, generated, puppet_class=klass) - data[f"{klass}::{local}"] = value - - if loop_candidates: - 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 - - return dump_yaml(data, sort_keys=True) - - -def generate_erb_template( - fmt: str, - parsed: Any, - role_prefix: str, - *, - original_text: str | None = None, - loop_candidates: list[LoopCandidate] | None = None, - flat_items: list[tuple[tuple[str, ...], Any]] | None = None, - puppet_class: str | None = None, -) -> str: - """Generate a Puppet ERB template from JinjaTurtle's renderer-neutral data. - - The first implementation intentionally translates the Jinja2 subset emitted - by JinjaTurtle's existing format handlers. This keeps parsing, formatting - preservation, and loop detection identical between Jinja2 and ERB output - while still producing Puppet-native ``@parameter`` references. - """ - - jinja_template = generate_jinja2_template( - fmt, - parsed, - role_prefix, - original_text=original_text, - loop_candidates=loop_candidates, - ) - names = _template_variable_names(role_prefix, flat_items or [], loop_candidates) - erb_template = translate_jinja2_to_erb( - jinja_template, - role_prefix=role_prefix, - puppet_class=puppet_class or role_prefix, - variable_names=names, - ) - # Defence in depth: no live Jinja2 delimiter may survive translation. - verify_erb_template_safe(erb_template) - return erb_template - - -def _stringify_timestamps(obj: Any) -> Any: +def _stringify_timestamps(obj: Any, _depth: int = 0) -> Any: """ Recursively walk a parsed config and turn any datetime/date/time objects into plain strings in ISO-8601 form. @@ -489,11 +538,23 @@ def _stringify_timestamps(obj: Any) -> Any: This commonly occurs otherwise with TOML and YAML files, which sees Python automatically convert those sorts of strings into datetime objects. + + Because this is the first walk applied to every parsed config (in + :func:`parse_config`), it also enforces :data:`MAX_CONFIG_DEPTH`. Bounding + nesting here means every downstream recursive walker (flatteners, loop + analyzer) is fed input of safe depth, so a degenerate deeply-nested file + fails with :class:`ConfigTooDeeplyNestedError` instead of an uncaught + ``RecursionError`` somewhere later in the pipeline. """ + if _depth > MAX_CONFIG_DEPTH: + raise ConfigTooDeeplyNestedError( + f"config nests deeper than the supported maximum of " + f"{MAX_CONFIG_DEPTH} levels; refusing to process it" + ) if isinstance(obj, dict): - return {k: _stringify_timestamps(v) for k, v in obj.items()} + return {k: _stringify_timestamps(v, _depth + 1) for k, v in obj.items()} if isinstance(obj, list): - return [_stringify_timestamps(v) for v in obj] + return [_stringify_timestamps(v, _depth + 1) for v in obj] # TOML & YAML both use the standard datetime types if isinstance(obj, datetime.datetime): diff --git a/src/jinjaturtle/erb.py b/src/jinjaturtle/erb.py deleted file mode 100644 index 4da6a6e..0000000 --- a/src/jinjaturtle/erb.py +++ /dev/null @@ -1,243 +0,0 @@ -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() - text = re.sub(r"_+", "_", text) - if not text: - text = fallback - if not re.match(r"^[a-z_]", text): - text = f"{fallback}_{text}" - return text - - -def puppet_class_name(raw: str) -> str: - """Return a conservative Puppet class/Hiera namespace name.""" - - text = _safe_name(raw, fallback="jinjaturtle") - if not re.match(r"^[a-z]", text): - text = f"jinjaturtle_{text}" - return text - - -def _role_prefix_name(raw: str) -> str: - return _safe_name(raw, fallback="jinjaturtle") - - -def puppet_local_var_name( - role_prefix: str, - jinja_var_name: str, - *, - puppet_class: str | None = None, -) -> str: - """Map a generated JinjaTurtle variable to a Puppet class parameter. - - For the common case where ``--role-name php`` also means class ``php``, a - generated Jinja variable such as ``php_memory_limit`` becomes Puppet local - parameter ``memory_limit`` and Hiera key ``php::memory_limit``. - - 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") - prefix = _role_prefix_name(role_prefix) - klass = puppet_class_name(puppet_class or role_prefix) - if klass == prefix and var_name.startswith(prefix + "_"): - stripped = var_name[len(prefix) + 1 :] - return stripped or var_name - return var_name - - -class ErbTranslator: - """Translate the Jinja2 subset emitted by JinjaTurtle into Puppet ERB.""" - - _TOKEN_RE = re.compile(r"({{.*?}}|{%.*?%})", re.S) - - def __init__( - self, - *, - role_prefix: str, - puppet_class: str | None = None, - variable_names: set[str] | None = None, - ) -> None: - self.role_prefix = role_prefix - self.puppet_class = puppet_class or role_prefix - self.variable_names = set(variable_names or set()) - 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: - if not token: - continue - if token.startswith("{{") and token.endswith("}}"): - expr = token[2:-2].strip() - out.append(f"<%= {self.expr_to_ruby(expr)} %>") - continue - if token.startswith("{%") and token.endswith("%}"): - stmt = token[2:-2].strip() - out.append(self.statement_to_erb(stmt)) - continue - out.append(token) - return "".join(out) - - def local_var(self, name: str) -> str: - return puppet_local_var_name( - self.role_prefix, name, puppet_class=self.puppet_class - ) - - def ruby_value(self, expr: str) -> str: - expr = expr.strip() - if expr in {"true", "True"}: - return "true" - if expr in {"false", "False"}: - 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 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 - - 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$", - expr, - ) - if m: - truthy = m.group(2) - cond = self.ruby_value(m.group(3)) - falsy = m.group(5) - 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_\.]*)$", - expr, - ) - if m: - value = self.ruby_value(m.group(3)) - fallback = self.ruby_value(m.group(4)) - return f"{value}.nil? ? 'null' : {fallback}" - - if "|" in expr: - base, *filters = [part.strip() for part in expr.split("|")] - ruby = self.ruby_value(base) - for filt in filters: - if filt.startswith("lower"): - ruby = f"{ruby}.to_s.downcase" - elif filt.startswith("to_json") or filt.startswith("tojson"): - self.needs_json = True - if "indent" in filt: - ruby = f"JSON.pretty_generate({ruby})" - else: - ruby = f"JSON.generate({ruby})" - return ruby - - return self.ruby_value(expr) - - def statement_to_erb(self, stmt: str) -> str: - if stmt.endswith(("-", "+")): - stmt = stmt[:-1].rstrip() - - if stmt.startswith("for "): - m = re.match( - r"^for\s+([A-Za-z_][A-Za-z0-9_]*)\s+in\s+([A-Za-z_][A-Za-z0-9_]*)$", - stmt, - ) - if m: - loop_var, collection = m.groups() - idx_var = f"__jt_idx_{len(self.loop_stack)}" - collection_ruby = self.ruby_value(collection) - self.loop_stack.append((loop_var, idx_var, collection_ruby)) - return f"<% {collection_ruby}.each_with_index do |{loop_var}, {idx_var}| -%>" - - if stmt == "endfor": - if self.loop_stack: - self.loop_stack.pop() - return "<% end %>" - - if stmt.startswith("if "): - cond = stmt[3:].strip() - 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) - 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) - if m: - return f"<% if {self.ruby_value(m.group(1))}.nil? -%>" - return f"<% if {self.expr_to_ruby(cond)} -%>" - - if stmt == "else": - return "<% else -%>" - - if stmt.startswith("elif "): - return f"<% elsif {self.expr_to_ruby(stmt[5:].strip())} -%>" - - if stmt == "endif": - return "<% end -%>" - - # Preserve unknown Jinja statements visibly as an ERB comment so the - # generated template does not contain invalid Jinja syntax. - return f"<%# Unsupported JinjaTurtle statement: {stmt} %>" - - -def translate_jinja2_to_erb( - template_text: str, - *, - role_prefix: str, - puppet_class: str | None = None, - variable_names: set[str] | None = None, -) -> str: - return ErbTranslator( - role_prefix=role_prefix, - puppet_class=puppet_class, - variable_names=variable_names, - ).translate(template_text) diff --git a/src/jinjaturtle/escape.py b/src/jinjaturtle/escape.py index 129c08a..6e68f69 100644 --- a/src/jinjaturtle/escape.py +++ b/src/jinjaturtle/escape.py @@ -1,96 +1,57 @@ from __future__ import annotations -"""Neutralise template metacharacters in text copied verbatim from source files. +"""Neutralise Jinja2 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. +and any line it does not recognise as ``key = value``. Config *values* are +always replaced with ``{{ var }}`` placeholders and stored in the defaults data, +so a payload inside a value remains data. Verbatim text is different: if the +source contains ``{{ ... }}``, ``{% ... %}`` or ``{# ... #}``, that text would +become live Jinja2 template code unless it is neutralised. 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 +The helpers here render those metacharacters as literal text so they survive a +later Jinja2 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 +# 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 and with any -# whitespace-control marker on either side. Jinja2 accepts "-", "+", or no +# whitespace-control marker on either side. Jinja2 accepts "-", "+", or no # marker adjacent to the "%}"/"{%" of a block tag (e.g. "{%endraw%}", # "{% endraw %}", "{%- endraw -%}", "{%+ endraw +%}"), and ALL of these close -# a raw block. The control marker must be matched so a "{%+ endraw %}" in +# a raw block. The control marker must be matched so a "{%+ endraw %}" in # attacker-influenced source text cannot survive defanging and break out of our -# {% raw %} wrapper. [-+]? appears on both sides accordingly. +# {% raw %} wrapper. [-+]? appears on both sides accordingly. _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. + We keep everything literal: the replacement breaks 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 while + never being a parseable raw-closing tag inside our wrapper. """ 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...%}" @@ -103,7 +64,7 @@ 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 + 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): @@ -111,43 +72,6 @@ def escape_jinja_literal(text: str) -> str: 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) +def escape_literal(text: str) -> str: + """Escape verbatim *text* for Jinja2 output.""" return escape_jinja_literal(text) diff --git a/src/jinjaturtle/handlers/base.py b/src/jinjaturtle/handlers/base.py index 14aaec7..71490bf 100644 --- a/src/jinjaturtle/handlers/base.py +++ b/src/jinjaturtle/handlers/base.py @@ -2,6 +2,10 @@ from __future__ import annotations from pathlib import Path from typing import Any, Iterable +import re + + +_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") class BaseHandler: @@ -59,6 +63,13 @@ class BaseHandler: Sanitises parts to lowercase [a-z0-9_] and strips extras. """ role_prefix = role_prefix.strip().lower() + if not _VAR_NAME_RE.fullmatch(role_prefix) or "__" in role_prefix: + raise ValueError( + "role name must be a safe Ansible/Jinja variable prefix: " + "letters, digits and underscores only; must start with a letter " + "or underscore; double underscores are not allowed" + ) + clean_parts: list[str] = [] for part in path: @@ -70,10 +81,13 @@ class BaseHandler: cleaned_chars.append(c.lower()) else: cleaned_chars.append("_") - cleaned_part = "".join(cleaned_chars).strip("_") + cleaned_part = re.sub(r"_+", "_", "".join(cleaned_chars)).strip("_") if cleaned_part: clean_parts.append(cleaned_part) if clean_parts: - return role_prefix + "_" + "_".join(clean_parts) - return role_prefix + var_name = role_prefix + "_" + "_".join(clean_parts) + else: + var_name = role_prefix + + return var_name diff --git a/src/jinjaturtle/handlers/json.py b/src/jinjaturtle/handlers/json.py index 064535a..fe32ac6 100644 --- a/src/jinjaturtle/handlers/json.py +++ b/src/jinjaturtle/handlers/json.py @@ -2,6 +2,7 @@ from __future__ import annotations import json import re +from uuid import uuid4 from pathlib import Path from typing import Any @@ -211,25 +212,26 @@ class JsonHandler(DictLikeHandler): Uses | to_json filter to preserve types (numbers, booleans, null). """ + marker_prefix = f"\ue000JT{uuid4().hex}" + scalar_markers: dict[str, str] = {} + 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()} 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 var_name = self.make_var_name(role_prefix, path) - return f"__SCALAR__{var_name}__" + marker = f"{marker_prefix}SCALAR:{len(scalar_markers)}\ue001" + scalar_markers[marker] = var_name + return marker templated = _walk(data) json_str = json.dumps(templated, indent=2, ensure_ascii=False) - # Replace scalar markers with Jinja expressions using to_json filter - # This preserves types (numbers stay numbers, booleans stay booleans) - json_str = re.sub( - r'"__SCALAR__([a-zA-Z_][a-zA-Z0-9_]*)__"', - lambda m: self._json_value_expr(m.group(1)), - json_str, - ) + for marker, var_name in scalar_markers.items(): + json_str = json_str.replace( + json.dumps(marker, ensure_ascii=False), self._json_value_expr(var_name) + ) return json_str + "\n" @@ -245,66 +247,67 @@ class JsonHandler(DictLikeHandler): Generate a JSON Jinja2 template with for loops where appropriate. """ + marker_prefix = f"\ue000JT{uuid4().hex}" + scalar_markers: dict[str, str] = {} + loop_markers: dict[tuple[str, str], str] = {} + def _walk(obj: Any, current_path: tuple[str, ...] = ()) -> Any: - # Check if this path is a loop candidate if current_path in loop_paths: - # Find the matching candidate candidate = next(c for c in loop_candidates if c.path == current_path) collection_var = self.make_var_name(role_prefix, candidate.path) - item_var = candidate.loop_var if candidate.item_schema == "scalar": - # Simple list of scalars - use special marker that we'll replace - return f"__LOOP_SCALAR__{collection_var}__{item_var}__" - elif candidate.item_schema in ("simple_dict", "nested"): - # List of dicts - use special marker - return f"__LOOP_DICT__{collection_var}__{item_var}__" + marker = f"{marker_prefix}LOOP_SCALAR:{len(loop_markers)}\ue001" + loop_markers[("scalar", collection_var)] = marker + return marker + if candidate.item_schema in ("simple_dict", "nested"): + marker = f"{marker_prefix}LOOP_DICT:{len(loop_markers)}\ue001" + loop_markers[("dict", collection_var)] = marker + return marker if isinstance(obj, dict): return {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: - # Already handled above return _walk(obj, current_path) return [_walk(v, current_path + (str(i),)) for i, v in enumerate(obj)] - # scalar - use marker to preserve type var_name = self.make_var_name(role_prefix, current_path) - return f"__SCALAR__{var_name}__" + marker = f"{marker_prefix}SCALAR:{len(scalar_markers)}\ue001" + scalar_markers[marker] = var_name + return marker templated = _walk(data, path) - - # Convert to JSON string json_str = json.dumps(templated, indent=2, ensure_ascii=False) - # Replace scalar markers with Jinja expressions using to_json filter - json_str = re.sub( - r'"__SCALAR__([a-zA-Z_][a-zA-Z0-9_]*)__"', - lambda m: self._json_value_expr(m.group(1)), - json_str, - ) + for marker, var_name in scalar_markers.items(): + json_str = json_str.replace( + json.dumps(marker, ensure_ascii=False), self._json_value_expr(var_name) + ) - # Post-process to replace loop markers with actual Jinja loops (indent-aware) for candidate in loop_candidates: collection_var = self.make_var_name(role_prefix, candidate.path) item_var = candidate.loop_var if candidate.item_schema == "scalar": - marker = f'"__LOOP_SCALAR__{collection_var}__{item_var}__"' + marker = loop_markers.get(("scalar", collection_var)) + if marker is None: + continue json_str = self._replace_marker_with_pretty_loop( json_str, - marker, + json.dumps(marker, ensure_ascii=False), lambda base, cv=collection_var, iv=item_var, c=candidate: self._generate_json_scalar_loop( cv, iv, c, base ), ) elif candidate.item_schema in ("simple_dict", "nested"): - marker = f'"__LOOP_DICT__{collection_var}__{item_var}__"' + marker = loop_markers.get(("dict", collection_var)) + if marker is None: + continue json_str = self._replace_marker_with_pretty_loop( json_str, - marker, + json.dumps(marker, ensure_ascii=False), lambda base, cv=collection_var, iv=item_var, c=candidate: self._generate_json_dict_loop( cv, iv, c, base ), @@ -364,8 +367,9 @@ 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 = json.dumps(str(key), ensure_ascii=False) dict_lines.append( - f'{field}"{key}": ' f"{j2.to_json(f'{item_var}.{key}')}{comma}" + f"{field}{safe_key}: " f"{j2.to_json(f'{item_var}.{key}')}{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/xml.py b/src/jinjaturtle/handlers/xml.py index 9fde3ec..4339fa4 100644 --- a/src/jinjaturtle/handlers/xml.py +++ b/src/jinjaturtle/handlers/xml.py @@ -4,6 +4,7 @@ from collections import Counter, defaultdict from pathlib import Path from typing import Any import xml.etree.ElementTree as ET # nosec +import defusedxml.ElementTree as DET from .base import BaseHandler from .. import j2 @@ -20,9 +21,12 @@ class XmlHandler(BaseHandler): 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 @@ -231,15 +235,6 @@ 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. @@ -252,30 +247,34 @@ 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. + This function is called immediately after parsing source XML and before + JinjaTurtle appends any of its own marker comments. Therefore every + comment currently in the tree is attacker-controlled source data and must + be escaped, including comments that resemble JinjaTurtle's internal + marker syntax. """ - # 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 "") + 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) - 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() - self._apply_jinja_to_xml_tree(role_prefix, root) - - # Neutralise template metacharacters in any comments preserved from the - # source file before serialising. + # Escape source comments before adding any JinjaTurtle marker comments. self._escape_source_comments(root) + self._apply_jinja_to_xml_tree(role_prefix, root) + indent = getattr(ET, "indent", None) if indent is not None: indent(root, space=" ") # type: ignore[arg-type] @@ -293,19 +292,22 @@ 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 + # Parse with comments preserved. + parser = DET.XMLParser( + target=ET.TreeBuilder(insert_comments=True), + forbid_dtd=True, + forbid_entities=True, + forbid_external=True, + ) parser.feed(body) root = parser.close() + # Escape source comments before adding any JinjaTurtle marker comments. + self._escape_source_comments(root) + # 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: diff --git a/src/jinjaturtle/multi.py b/src/jinjaturtle/multi.py index 195434f..1add120 100644 --- a/src/jinjaturtle/multi.py +++ b/src/jinjaturtle/multi.py @@ -29,7 +29,13 @@ from typing import Any, Iterable import xml.etree.ElementTree as ET # nosec from . import j2 -from .core import dump_yaml, flatten_config, make_var_name, parse_config +from .core import ( + check_var_name_collisions, + dump_yaml, + flatten_config, + make_var_name, + parse_config, +) from .handlers.xml import XmlHandler from .safety import verify_jinja2_template_safe from .escape import escape_jinja_literal @@ -45,7 +51,7 @@ SUPPORTED_SUFFIXES: dict[str, set[str]] = { def is_supported_file(path: Path) -> bool: - if not path.is_file(): + if path.is_symlink() or not path.is_file(): return False suffix = path.suffix.lower() for exts in SUPPORTED_SUFFIXES.values(): @@ -57,13 +63,25 @@ def is_supported_file(path: Path) -> bool: def iter_supported_files(root: Path, recursive: bool) -> list[Path]: if not root.exists(): raise FileNotFoundError(str(root)) + if root.is_symlink(): + return [] if root.is_file(): return [root] if is_supported_file(root) else [] if not root.is_dir(): return [] + resolved_root = root.resolve(strict=True) it = root.rglob("*") if recursive else root.glob("*") - files = [p for p in it if is_supported_file(p)] + files: list[Path] = [] + for p in it: + if not is_supported_file(p): + continue + try: + resolved = p.resolve(strict=True) + resolved.relative_to(resolved_root) + except (OSError, ValueError): + continue + files.append(p) files.sort() return files @@ -684,6 +702,7 @@ def process_directory( for rid, parsed, containers in zip(rel_ids, parsed_list, container_sets): item: dict[str, Any] = {"id": rid} flat = flatten_config(fmt, parsed, loop_candidates=None) + check_var_name_collisions(role_prefix, flat) for path, value in flat: item[make_var_name(role_prefix, path)] = value for cpath in optional_containers: @@ -713,6 +732,7 @@ def process_directory( for rid, parser in zip(rel_ids, parsers): # type: ignore[arg-type] item: dict[str, Any] = {"id": rid} flat = flatten_config(fmt, parser, loop_candidates=None) + check_var_name_collisions(role_prefix, flat) for path, value in flat: item[make_var_name(role_prefix, path)] = value # section presence @@ -759,6 +779,7 @@ def process_directory( for rid, parsed, elems in zip(rel_ids, parsed_list, elem_sets): item: dict[str, Any] = {"id": rid} flat = flatten_config(fmt, parsed, loop_candidates=None) + check_var_name_collisions(role_prefix, flat) for path, value in flat: item[make_var_name(role_prefix, path)] = value for epath in optional_elements: diff --git a/src/jinjaturtle/safety.py b/src/jinjaturtle/safety.py index 07d3b8d..5b2dea3 100644 --- a/src/jinjaturtle/safety.py +++ b/src/jinjaturtle/safety.py @@ -22,7 +22,7 @@ single global property: The check is positive/allowlist-based, which is the safe direction: unknown constructs are rejected, not ignored. It runs at the single choke points in -``core.py`` (``generate_jinja2_template`` / ``generate_erb_template``), so it +``core.py`` (``generate_jinja2_template``), so it covers every current handler and every future one automatically. Why this is robust against the escaper being wrong @@ -42,7 +42,6 @@ import re __all__ = [ "TemplateSafetyError", "verify_jinja2_template_safe", - "verify_erb_template_safe", ] @@ -205,51 +204,3 @@ def verify_jinja2_template_safe(template_text: str) -> None: # tokens) so the reconstructed body matches the source spacing. body_parts.append(value if isinstance(value, str) else str(value)) # raw_begin / raw_end / data tokens outside a tag are inert: skip. - - -# --------------------------------------------------------------------------- # -# ERB gate. -# -# JinjaTurtle's ERB output is produced by translating the (already-verified) -# Jinja2 subset, so the Jinja2 gate is the primary guarantee. As an independent -# ERB-side backstop we confirm that every ERB tag body is one the translator -# emits, and that no Jinja2 delimiters survived into the ERB output (which would -# indicate a raw block the translator failed to recognise -- the historical -# ``{%+ raw %}`` blind spot). -# --------------------------------------------------------------------------- # - -_ERB_TAG_RE = re.compile(r"<%[-=#]?(.*?)[-]?%>", re.S) -_JINJA_DELIMS = ("{{", "}}", "{%", "%}", "{#", "#}") - -# Bodies the ErbTranslator emits. Kept permissive for Ruby method chains it -# constructs (``@var``, ``.each_with_index``, ``JSON.generate(...)`` etc.) but -# anchored so arbitrary attacker text cannot masquerade as one. -_ERB_STMT_PATTERNS = tuple( - re.compile(p) - for p in ( - r"^require 'json'$", - r"^end$", - r"^else$", - r"^@?[A-Za-z_][\w@\.\[\]'\"]*\.each_with_index do \|[A-Za-z_]\w*, __jt_idx_\d+\| $", - r"^if .+$", - r"^elsif .+$", - r"^unless .+\.nil\?$", - r"^# Unsupported JinjaTurtle statement: .*$", - ) -) - - -def verify_erb_template_safe(template_text: str) -> None: - """Validate that *template_text* contains no leftover Jinja2 delimiters. - - The translator is the security-relevant step for ERB; this backstop ensures - no live Jinja construct survived translation (which would mean a raw block - was not recognised and source text passed through untouched). - """ - for delim in _JINJA_DELIMS: - if delim in template_text: - raise TemplateSafetyError( - "refusing to emit ERB template: it still contains the Jinja2 " - f"delimiter {delim!r}, which means source text was not fully " - "translated/neutralised (possible template injection)." - ) diff --git a/tests/test_cli.py b/tests/test_cli.py index e4ac519..c0d5470 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -132,57 +132,3 @@ def test_cli_folder_single_output_file_when_one_format(tmp_path): assert defaults_path.is_file() assert template_path.is_file() assert "to_json" in template_path.read_text(encoding="utf-8") - - -def test_cli_erb_outputs_puppet_hiera_and_erb_template(tmp_path): - cfg = tmp_path / "app.ini" - cfg.write_text("[main]\nport = 8080\n", encoding="utf-8") - data_out = tmp_path / "common.yaml" - template_out = tmp_path / "app.ini.erb" - - exit_code = cli._main( - [ - str(cfg), - "-r", - "php", - "--template-engine", - "erb", - "--defaults-output", - str(data_out), - "--template-output", - str(template_out), - ] - ) - - assert exit_code == 0 - assert "php::main_port: '8080'" in data_out.read_text(encoding="utf-8") - assert "port = <%= @main_port %>" in template_out.read_text(encoding="utf-8") - - -def test_cli_erb_can_use_separate_puppet_class_namespace(tmp_path): - cfg = tmp_path / "app.ini" - cfg.write_text("[main]\nport = 8080\n", encoding="utf-8") - data_out = tmp_path / "node.yaml" - template_out = tmp_path / "app.ini.erb" - - exit_code = cli._main( - [ - str(cfg), - "-r", - "php_etc_app_ini", - "--template-engine", - "erb", - "--puppet-class", - "php", - "--defaults-output", - str(data_out), - "--template-output", - str(template_out), - ] - ) - - assert exit_code == 0 - data = data_out.read_text(encoding="utf-8") - template = template_out.read_text(encoding="utf-8") - assert "php::php_etc_app_ini_main_port: '8080'" in data - assert "port = <%= @php_etc_app_ini_main_port %>" in template diff --git a/tests/test_deep_nesting_dos.py b/tests/test_deep_nesting_dos.py new file mode 100644 index 0000000..0ff976d --- /dev/null +++ b/tests/test_deep_nesting_dos.py @@ -0,0 +1,122 @@ +"""Regression tests for the deeply-nested-input denial-of-service fix. + +Pathologically nested configuration (thousands of container levels) used to +drive the recursive walkers in ``core`` -- or the underlying parser itself -- +past Python's stack limit, raising an uncaught ``RecursionError``. The fix +bounds nesting depth and converts both cases into a clean, catchable +``ConfigTooDeeplyNestedError``. These tests pin that behaviour and guard +against regressions, while confirming normally-nested configs still process. +""" + +from __future__ import annotations + +import tempfile +from pathlib import Path + +import pytest + +from jinjaturtle.core import ( + ConfigTooDeeplyNestedError, + MAX_CONFIG_DEPTH, + analyze_loops, + flatten_config, + generate_ansible_yaml, + generate_jinja2_template, + parse_config, +) + + +def _run_pipeline(content: str, suffix: str) -> None: + with tempfile.NamedTemporaryFile( + "w", suffix=suffix, delete=False, encoding="utf-8" + ) as tf: + tf.write(content) + path = Path(tf.name) + try: + fmt, parsed = parse_config(path) + loops = analyze_loops(fmt, parsed) + flat = flatten_config(fmt, parsed, loops) + generate_jinja2_template( + fmt, parsed, "role", original_text=content, loop_candidates=loops + ) + generate_ansible_yaml("role", flat, loops) + finally: + path.unlink() + + +def _deep_json_objects(depth: int) -> str: + s = "0" + for _ in range(depth): + s = '{"a":' + s + "}" + return s + + +def _deep_json_arrays(depth: int) -> str: + return "[" * depth + "1" + "]" * depth + + +def _deep_xml(depth: int) -> str: + s = "v" + for _ in range(depth): + s = f"{s}" + return s + + +def _deep_yaml(depth: int) -> str: + s = "v" + for _ in range(depth): + s = "{a: " + s + "}" + return s + + +def _deep_toml(depth: int) -> str: + return "a = " + "[" * depth + "1" + "]" * depth + + +@pytest.mark.parametrize( + "content, suffix", + [ + (_deep_json_objects(5000), ".json"), + (_deep_json_arrays(100000), ".json"), + (_deep_xml(5000), ".xml"), + (_deep_yaml(3000), ".yaml"), + (_deep_toml(2000), ".toml"), + ], +) +def test_deeply_nested_input_is_rejected_cleanly(content: str, suffix: str) -> None: + # Must raise our typed error, and crucially must NOT raise RecursionError + # (pytest would report that as an error, but be explicit about intent). + with pytest.raises(ConfigTooDeeplyNestedError): + _run_pipeline(content, suffix) + + +def test_recursion_error_never_escapes() -> None: + # Belt-and-suspenders: ensure a RecursionError is never what surfaces. + content = _deep_yaml(6000) + try: + _run_pipeline(content, ".yaml") + except ConfigTooDeeplyNestedError: + pass + except RecursionError: # pragma: no cover - this is the bug we fixed + pytest.fail("RecursionError escaped instead of ConfigTooDeeplyNestedError") + + +@pytest.mark.parametrize( + "content, suffix", + [ + ('{"name":"web","port":8080,"tags":["a","b"]}', ".json"), + ("db5432", ".xml"), + ("name: web\nport: 8080\nnested:\n a:\n b: 1\n", ".yaml"), + ('[section]\nkey = "value"\nnums = [1, 2, 3]\n', ".toml"), + ("[section]\nkey = value\n", ".ini"), + ], +) +def test_normal_configs_still_process(content: str, suffix: str) -> None: + # No exception expected for ordinary, shallow configurations. + _run_pipeline(content, suffix) + + +def test_just_under_limit_is_accepted() -> None: + # A structure comfortably under the cap must still process successfully. + content = _deep_json_objects(MAX_CONFIG_DEPTH - 10) + _run_pipeline(content, ".json") diff --git a/tests/test_injection_security.py b/tests/test_injection_security.py index a1de483..0b0ee7b 100644 --- a/tests/test_injection_security.py +++ b/tests/test_injection_security.py @@ -2,9 +2,9 @@ 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. +Jinja2 delimiters it must be neutralised, otherwise attacker-influenced +config content becomes live template code when a downstream tool 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 @@ -14,7 +14,6 @@ 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 @@ -25,7 +24,6 @@ import yaml as pyyaml from jinjaturtle.escape import ( escape_jinja_literal, - escape_erb_literal, escape_literal, ) @@ -235,36 +233,10 @@ def test_endraw_whitespace_control_cannot_break_out(endraw): assert rendered == payload -@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. - - -def test_escape_literal_dispatches_by_engine(): - """The public ``escape_literal`` wrapper routes to the right engine.""" +def test_escape_literal_is_jinja_alias(): + """The public ``escape_literal`` wrapper is Jinja2-only.""" payload = "{{ 7*7 }}" - # Default engine is Jinja2. assert escape_literal(payload) == escape_jinja_literal(payload) - assert escape_literal(payload, engine="jinja2") == escape_jinja_literal(payload) - # An unknown engine falls back to the safer Jinja2 escaping. - assert escape_literal(payload, engine="nonsense") == escape_jinja_literal(payload) - # ERB routing. - erb_payload = "<%= system('id') %>" - assert escape_literal(erb_payload, engine="erb") == escape_erb_literal(erb_payload) def test_escape_literal_jinja_output_is_inert(): diff --git a/tests/test_output_safety_gate.py b/tests/test_output_safety_gate.py index 055ceb3..813ded0 100644 --- a/tests/test_output_safety_gate.py +++ b/tests/test_output_safety_gate.py @@ -25,7 +25,6 @@ from jinjaturtle.multi import process_directory from jinjaturtle.safety import ( TemplateSafetyError, verify_jinja2_template_safe, - verify_erb_template_safe, ) @@ -180,20 +179,6 @@ def _strip_raw_blocks(text: str) -> str: ) -# --------------------------------------------------------------------------- # -# ERB gate. -# --------------------------------------------------------------------------- # - - -def test_erb_gate_blocks_leftover_jinja_delimiters(): - with pytest.raises(TemplateSafetyError): - verify_erb_template_safe("ok <%= @x %> but {{ leftover }} here") - - -def test_erb_gate_allows_clean_erb(): - verify_erb_template_safe("memory = <%= @memory_limit %>\n") - - # --------------------------------------------------------------------------- # # End-to-end CLI: fail closed. # --------------------------------------------------------------------------- # diff --git a/tests/test_release_blockers.py b/tests/test_release_blockers.py new file mode 100644 index 0000000..c4d3a63 --- /dev/null +++ b/tests/test_release_blockers.py @@ -0,0 +1,98 @@ +from pathlib import Path + +import pytest +import yaml +from defusedxml.common import DTDForbidden + +from jinjaturtle.core import ( + VariableNameCollisionError, + generate_ansible_yaml, + parse_config, +) +from jinjaturtle.handlers.xml import XmlHandler +from jinjaturtle.multi import iter_supported_files, process_directory + + +def test_defaults_string_values_are_ansible_unsafe(): + output = generate_ansible_yaml( + "role", [(("section", "key"), "{{ lookup('pipe', 'id') }}")] + ) + assert "!unsafe" in output + assert yaml.safe_load(output)["role_section_key"] == "{{ lookup('pipe', 'id') }}" + + +def test_variable_name_collisions_fail_closed(): + with pytest.raises(VariableNameCollisionError): + generate_ansible_yaml( + "role", + [ + (("section", "a-b"), "safe"), + (("section", "a_b"), "evil"), + ], + ) + + +def test_folder_mode_does_not_follow_file_symlinks(tmp_path: Path): + outside = tmp_path / "outside.ini" + outside.write_text("[s]\nsecret = value\n", encoding="utf-8") + root = tmp_path / "root" + root.mkdir() + (root / "link.ini").symlink_to(outside) + (root / "real.ini").write_text("[s]\nname = ok\n", encoding="utf-8") + + files = iter_supported_files(root, recursive=True) + assert files == [root / "real.ini"] + defaults, _outputs = process_directory(root, recursive=True, role_prefix="role") + assert "secret" not in defaults + assert "real.ini" in defaults + + +def test_xml_parser_rejects_dtd_and_entities(tmp_path: Path): + xml_path = tmp_path / "evil.xml" + xml_path.write_text(']>&x;', encoding="utf-8") + + with pytest.raises(DTDForbidden): + parse_config(xml_path) + + +def test_xml_source_comments_cannot_forge_internal_markers(): + source = ( + "ok" + ) + template = XmlHandler().generate_jinja2_template(None, "role", original_text=source) + + assert "{% if role_missing is defined %}" not in template + assert "{% endif %}" not in template + assert "IF:role_missing" in template + assert "ENDIF:role_missing" in template + + +def test_cli_refuses_to_write_output_symlink(tmp_path: Path): + import subprocess + import sys + + src = tmp_path / "input.ini" + src.write_text("[s]\nname = ok\n", encoding="utf-8") + target = tmp_path / "target.yml" + target.write_text("do not overwrite\n", encoding="utf-8") + link = tmp_path / "defaults.yml" + link.symlink_to(target) + + result = subprocess.run( + [ + sys.executable, + "-m", + "jinjaturtle.cli", + str(src), + "-d", + str(link), + ], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env={"PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src")}, + check=False, + ) + + assert result.returncode != 0 + assert target.read_text(encoding="utf-8") == "do not overwrite\n" diff --git a/tests/test_yaml_handler.py b/tests/test_yaml_handler.py index a217e8a..543c759 100644 --- a/tests/test_yaml_handler.py +++ b/tests/test_yaml_handler.py @@ -10,7 +10,6 @@ from jinjaturtle.core import ( analyze_loops, flatten_config, generate_ansible_yaml, - generate_erb_template, generate_jinja2_template, ) from jinjaturtle.handlers.yaml import YamlHandler @@ -178,17 +177,15 @@ def test_yaml_loop_preserves_blank_separator_after_list(tmp_path: Path): "blank", "require:\n - rubocop-performance\n - rubocop-rspec\n\nAllCops:\n NewCops: enable\n", "\n - rubocop-rspec\n\nAllCops:", - "<% end %>\nAllCops:", ), ( "no_blank", "require:\n - rubocop-performance\n - rubocop-rspec\nAllCops:\n NewCops: enable\n", "\n - rubocop-rspec\nAllCops:", - "<% end %>AllCops:", ), ] - for label, text, rendered_expected, erb_expected in cases: + for label, text, rendered_expected in cases: path = tmp_path / f"{label}.yml" path.write_text(text, encoding="utf-8") @@ -205,16 +202,6 @@ def test_yaml_loop_preserves_blank_separator_after_list(tmp_path: Path): rendered = Template(template).render(**defaults) assert rendered_expected in rendered - erb_template = generate_erb_template( - fmt, - parsed, - "role", - original_text=text, - loop_candidates=loop_candidates, - flat_items=flat_items, - ) - assert erb_expected in erb_template - def test_yaml_loop_preserves_following_top_level_comments(tmp_path: Path): from jinja2 import Environment, Template From 661320558ca15de0e0ed74a6b22153f408160d06 Mon Sep 17 00:00:00 2001 From: Miguel Jacq Date: Thu, 25 Jun 2026 17:07:58 +1000 Subject: [PATCH 2/4] Go back to just jinja2 --- README.md | 125 +--------------- src/jinjaturtle/cli.py | 79 ++-------- src/jinjaturtle/core.py | 70 --------- src/jinjaturtle/erb.py | 243 ------------------------------- src/jinjaturtle/escape.py | 57 +------- src/jinjaturtle/safety.py | 4 +- tests/test_cli.py | 54 ------- tests/test_injection_security.py | 47 +----- tests/test_yaml_handler.py | 11 -- 9 files changed, 26 insertions(+), 664 deletions(-) delete mode 100644 src/jinjaturtle/erb.py diff --git a/README.md b/README.md index 2905cfe..1726719 100644 --- a/README.md +++ b/README.md @@ -13,9 +13,6 @@ By default it generates: - an **Ansible defaults YAML** file containing the variables used by that template. -It can also generate **ERB** templates and **Puppet Hiera-style YAML** data for -Puppet workflows. - JinjaTurtle does not try to replace configuration-management tools. Its job is to speed up the boring first pass: take a real config file, discover the values inside it, replace those values with variables, and write the corresponding @@ -37,16 +34,6 @@ For the default Jinja2/Ansible mode: 5. An Ansible defaults YAML file is generated with those variables and the original values. -For ERB/Puppet mode: - -1. The same parse/flatten/loop analysis is used. -2. An ERB template is generated with Puppet-style instance variables such as - `<%= @memory_limit %>`. -3. The variables file is written as Puppet Hiera-style data, such as - `php::memory_limit: 256M`. -4. If `--puppet-class` is supplied, that class name is used as the Hiera - namespace while `--role-name` remains the local variable prefix. - By default, the generated variable data and template are printed to stdout. Use `--defaults-output` and `--template-output` to write them to files. @@ -80,78 +67,6 @@ and defaults data like: php_memory_limit: 256M ``` -## ERB / Puppet example - -Use `--template-engine erb` when you want Puppet ERB output: - -```shell -jinjaturtle php.ini \ - --role-name php \ - --template-engine erb \ - --defaults-output data/common.yaml \ - --template-output templates/php.ini.erb -``` - -Given the same source value: - -```ini -memory_limit = 256M -``` - -JinjaTurtle will produce an ERB template value like: - -```erb -memory_limit = <%= @memory_limit %> -``` - -and Hiera-style data like: - -```yaml -php::memory_limit: 256M -``` - -The `--defaults-output` option name is retained for CLI compatibility, but in -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, should still declare the class parameters and call the template, for -example with Puppet's `template()` function. - -## Using `--puppet-class` - -Most direct usage can simply use the same value for the role name and Puppet -class name: - -```shell -jinjaturtle php.ini --role-name php --template-engine erb -``` - -This creates Hiera keys such as: - -```yaml -php::memory_limit: 256M -``` - -and local ERB variables such as: - -```erb -<%= @memory_limit %> -``` - -For generated systems, it can be useful to make `--role-name` more specific -while keeping the Hiera keys under the real Puppet class. For example: - -```shell -jinjaturtle php.ini \ - --role-name php_etc_php_ini \ - --puppet-class php \ - --template-engine erb -``` - -In that case the variable prefix can stay file-specific, while the Hiera data -is still written under `php::...`. - ## What sort of config files can it handle? JinjaTurtle supports common structured and semi-structured config formats: @@ -176,8 +91,8 @@ homogeneous enough. If it is not confident, it falls back to flattened scalar variables. Some very complex files will still need manual cleanup. The goal is to speed up -conversion into Jinja2 or ERB templates, not to guarantee a perfect final module -without review. +conversion into Jinja2 templates, not to guarantee a perfect final module without +review. ## JSON, quoting, and type preservation @@ -196,17 +111,7 @@ when the correct rendered JSON should be: {"enabled": true} ``` -In Jinja2 mode this uses Ansible-style JSON filters. In ERB mode it emits Ruby -JSON generation where required, for example: - -```erb -<% require 'json' -%> -{ - "enabled": <%= JSON.generate(@enabled) %> -} -``` - -That is expected for JSON ERB templates. +This uses Ansible-style JSON filters. ## Can I convert multiple files at once? @@ -287,8 +192,6 @@ poetry install usage: jinjaturtle [-h] [-r ROLE_NAME] [--recursive] [-f {ini,json,toml,yaml,xml,postfix,systemd,ssh}] [-d DEFAULTS_OUTPUT] [-t TEMPLATE_OUTPUT] - [--template-engine {jinja2,erb}] - [--puppet-class PUPPET_CLASS] config Convert a config file into an Ansible defaults file and Jinja2 template. @@ -302,8 +205,7 @@ options: -h, --help show this help message and exit -r, --role-name ROLE_NAME Role name / variable prefix. In Jinja2 mode this is - usually the Ansible role name. In ERB mode it is used - as the local variable prefix. Defaults to jinjaturtle. + usually the Ansible role name. Defaults to jinjaturtle. --recursive When CONFIG is a folder, recurse into subfolders. -f, --format {ini,json,toml,yaml,xml,postfix,systemd,ssh} Force config format instead of auto-detecting from @@ -314,12 +216,6 @@ options: -t, --template-output TEMPLATE_OUTPUT Path to write the generated config template. If omitted, it is printed to stdout. - --template-engine {jinja2,erb} - Template syntax to generate. Defaults to jinja2. Use - erb for Puppet templates. - --puppet-class PUPPET_CLASS - Puppet class / Hiera namespace to use with - --template-engine erb. Defaults to --role-name. ``` ## Additional supported formats @@ -346,16 +242,16 @@ 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, + separately in the defaults 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. + a value 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. + (`{{ }}`, `{% %}`, `{# #}` ) are escaped so they render as the literal + characters the author wrote, rather than executing. ### Consumer responsibilities @@ -368,11 +264,6 @@ which is the normal case: 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. diff --git a/src/jinjaturtle/cli.py b/src/jinjaturtle/cli.py index 6345fae..0eaf506 100644 --- a/src/jinjaturtle/cli.py +++ b/src/jinjaturtle/cli.py @@ -12,12 +12,10 @@ from .core import ( flatten_config, generate_ansible_yaml, generate_jinja2_template, - generate_puppet_hiera_yaml, - generate_erb_template, ) from .multi import process_directory -from .safety import TemplateSafetyError, verify_erb_template_safe +from .safety import TemplateSafetyError def _build_arg_parser() -> argparse.ArgumentParser: @@ -59,20 +57,6 @@ def _build_arg_parser() -> argparse.ArgumentParser: "--template-output", help="Path to write the generated config template. If omitted, template is printed to stdout.", ) - ap.add_argument( - "--template-engine", - choices=[j2.NAME, "erb"], - default=j2.NAME, - help="Template syntax to generate (default: jinja2). Use erb for Puppet templates.", - ) - ap.add_argument( - "--puppet-class", - help=( - "Puppet class/Hiera namespace to use with --template-engine erb. " - "Defaults to --role-name. This lets tools use a file-specific " - "variable prefix while writing Hiera keys under the real Puppet class." - ), - ) return ap @@ -110,21 +94,7 @@ def _run(argv: list[str] | None = None) -> int: print("# defaults/main.yml") print(defaults_yaml, end="") - # Optionally translate folder-mode templates to ERB. Folder mode keeps - # the existing data shape; single-file mode below is the preferred - # Puppet path because it can produce class-parameter Hiera keys. - if args.template_engine == "erb": - from .erb import translate_jinja2_to_erb - - for o in outputs: - o.template = translate_jinja2_to_erb( - o.template, - role_prefix=args.role_name, - puppet_class=args.puppet_class or args.role_name, - ) - verify_erb_template_safe(o.template) - - template_ext = "erb" if args.template_engine == "erb" else j2.TEMPLATE_EXTENSION + template_ext = j2.TEMPLATE_EXTENSION # Write templates if args.template_output: @@ -161,36 +131,17 @@ def _run(argv: list[str] | None = None) -> int: # Flatten config (excluding loop paths if loops are detected) flat_items = flatten_config(fmt, parsed, loop_candidates) - if args.template_engine == "erb": - ansible_yaml = generate_puppet_hiera_yaml( - args.role_name, - flat_items, - loop_candidates, - puppet_class=args.puppet_class or args.role_name, - ) - template_str = generate_erb_template( - fmt, - parsed, - args.role_name, - original_text=config_text, - loop_candidates=loop_candidates, - flat_items=flat_items, - puppet_class=args.puppet_class or args.role_name, - ) - else: - # Generate defaults YAML (with loop collections if detected) - ansible_yaml = generate_ansible_yaml( - args.role_name, flat_items, loop_candidates - ) + # Generate defaults YAML (with loop collections if detected) + ansible_yaml = generate_ansible_yaml(args.role_name, flat_items, loop_candidates) - # Generate template (with loops if detected) - template_str = generate_jinja2_template( - fmt, - parsed, - args.role_name, - original_text=config_text, - loop_candidates=loop_candidates, - ) + # Generate template (with loops if detected) + template_str = generate_jinja2_template( + fmt, + parsed, + args.role_name, + original_text=config_text, + loop_candidates=loop_candidates, + ) if args.defaults_output: Path(args.defaults_output).write_text(ansible_yaml, encoding="utf-8") @@ -201,11 +152,7 @@ def _run(argv: list[str] | None = None) -> int: if args.template_output: Path(args.template_output).write_text(template_str, encoding="utf-8") else: - print( - "# config.erb" - if args.template_engine == "erb" - else f"# config.{j2.TEMPLATE_EXTENSION}" - ) + print(f"# config.{j2.TEMPLATE_EXTENSION}") print(template_str, end="") return 0 diff --git a/src/jinjaturtle/core.py b/src/jinjaturtle/core.py index 4d35260..8e3d10e 100644 --- a/src/jinjaturtle/core.py +++ b/src/jinjaturtle/core.py @@ -8,9 +8,7 @@ import re import yaml from .loop_analyzer import LoopAnalyzer, LoopCandidate -from .erb import puppet_class_name, puppet_local_var_name, translate_jinja2_to_erb from .safety import ( - verify_erb_template_safe, verify_jinja2_template_safe, ) from .handlers import ( @@ -411,74 +409,6 @@ def _template_variable_names( return names -def generate_puppet_hiera_yaml( - role_prefix: str, - flat_items: list[tuple[tuple[str, ...], Any]], - loop_candidates: list[LoopCandidate] | None = None, - *, - puppet_class: str | None = None, -) -> str: - """Create Puppet Hiera data suitable for Automatic Parameter Lookup. - - ``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``. - """ - - klass = puppet_class_name(puppet_class or role_prefix) - data: dict[str, Any] = {} - - for path, value in flat_items: - generated = make_var_name(role_prefix, path) - local = puppet_local_var_name(role_prefix, generated, puppet_class=klass) - data[f"{klass}::{local}"] = value - - if loop_candidates: - 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 - - return dump_yaml(data, sort_keys=True) - - -def generate_erb_template( - fmt: str, - parsed: Any, - role_prefix: str, - *, - original_text: str | None = None, - loop_candidates: list[LoopCandidate] | None = None, - flat_items: list[tuple[tuple[str, ...], Any]] | None = None, - puppet_class: str | None = None, -) -> str: - """Generate a Puppet ERB template from JinjaTurtle's renderer-neutral data. - - The first implementation intentionally translates the Jinja2 subset emitted - by JinjaTurtle's existing format handlers. This keeps parsing, formatting - preservation, and loop detection identical between Jinja2 and ERB output - while still producing Puppet-native ``@parameter`` references. - """ - - jinja_template = generate_jinja2_template( - fmt, - parsed, - role_prefix, - original_text=original_text, - loop_candidates=loop_candidates, - ) - names = _template_variable_names(role_prefix, flat_items or [], loop_candidates) - erb_template = translate_jinja2_to_erb( - jinja_template, - role_prefix=role_prefix, - puppet_class=puppet_class or role_prefix, - variable_names=names, - ) - # Defence in depth: no live Jinja2 delimiter may survive translation. - verify_erb_template_safe(erb_template) - return erb_template - - def _stringify_timestamps(obj: Any) -> Any: """ Recursively walk a parsed config and turn any datetime/date/time objects diff --git a/src/jinjaturtle/erb.py b/src/jinjaturtle/erb.py deleted file mode 100644 index 4da6a6e..0000000 --- a/src/jinjaturtle/erb.py +++ /dev/null @@ -1,243 +0,0 @@ -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() - text = re.sub(r"_+", "_", text) - if not text: - text = fallback - if not re.match(r"^[a-z_]", text): - text = f"{fallback}_{text}" - return text - - -def puppet_class_name(raw: str) -> str: - """Return a conservative Puppet class/Hiera namespace name.""" - - text = _safe_name(raw, fallback="jinjaturtle") - if not re.match(r"^[a-z]", text): - text = f"jinjaturtle_{text}" - return text - - -def _role_prefix_name(raw: str) -> str: - return _safe_name(raw, fallback="jinjaturtle") - - -def puppet_local_var_name( - role_prefix: str, - jinja_var_name: str, - *, - puppet_class: str | None = None, -) -> str: - """Map a generated JinjaTurtle variable to a Puppet class parameter. - - For the common case where ``--role-name php`` also means class ``php``, a - generated Jinja variable such as ``php_memory_limit`` becomes Puppet local - parameter ``memory_limit`` and Hiera key ``php::memory_limit``. - - 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") - prefix = _role_prefix_name(role_prefix) - klass = puppet_class_name(puppet_class or role_prefix) - if klass == prefix and var_name.startswith(prefix + "_"): - stripped = var_name[len(prefix) + 1 :] - return stripped or var_name - return var_name - - -class ErbTranslator: - """Translate the Jinja2 subset emitted by JinjaTurtle into Puppet ERB.""" - - _TOKEN_RE = re.compile(r"({{.*?}}|{%.*?%})", re.S) - - def __init__( - self, - *, - role_prefix: str, - puppet_class: str | None = None, - variable_names: set[str] | None = None, - ) -> None: - self.role_prefix = role_prefix - self.puppet_class = puppet_class or role_prefix - self.variable_names = set(variable_names or set()) - 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: - if not token: - continue - if token.startswith("{{") and token.endswith("}}"): - expr = token[2:-2].strip() - out.append(f"<%= {self.expr_to_ruby(expr)} %>") - continue - if token.startswith("{%") and token.endswith("%}"): - stmt = token[2:-2].strip() - out.append(self.statement_to_erb(stmt)) - continue - out.append(token) - return "".join(out) - - def local_var(self, name: str) -> str: - return puppet_local_var_name( - self.role_prefix, name, puppet_class=self.puppet_class - ) - - def ruby_value(self, expr: str) -> str: - expr = expr.strip() - if expr in {"true", "True"}: - return "true" - if expr in {"false", "False"}: - 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 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 - - 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$", - expr, - ) - if m: - truthy = m.group(2) - cond = self.ruby_value(m.group(3)) - falsy = m.group(5) - 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_\.]*)$", - expr, - ) - if m: - value = self.ruby_value(m.group(3)) - fallback = self.ruby_value(m.group(4)) - return f"{value}.nil? ? 'null' : {fallback}" - - if "|" in expr: - base, *filters = [part.strip() for part in expr.split("|")] - ruby = self.ruby_value(base) - for filt in filters: - if filt.startswith("lower"): - ruby = f"{ruby}.to_s.downcase" - elif filt.startswith("to_json") or filt.startswith("tojson"): - self.needs_json = True - if "indent" in filt: - ruby = f"JSON.pretty_generate({ruby})" - else: - ruby = f"JSON.generate({ruby})" - return ruby - - return self.ruby_value(expr) - - def statement_to_erb(self, stmt: str) -> str: - if stmt.endswith(("-", "+")): - stmt = stmt[:-1].rstrip() - - if stmt.startswith("for "): - m = re.match( - r"^for\s+([A-Za-z_][A-Za-z0-9_]*)\s+in\s+([A-Za-z_][A-Za-z0-9_]*)$", - stmt, - ) - if m: - loop_var, collection = m.groups() - idx_var = f"__jt_idx_{len(self.loop_stack)}" - collection_ruby = self.ruby_value(collection) - self.loop_stack.append((loop_var, idx_var, collection_ruby)) - return f"<% {collection_ruby}.each_with_index do |{loop_var}, {idx_var}| -%>" - - if stmt == "endfor": - if self.loop_stack: - self.loop_stack.pop() - return "<% end %>" - - if stmt.startswith("if "): - cond = stmt[3:].strip() - 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) - 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) - if m: - return f"<% if {self.ruby_value(m.group(1))}.nil? -%>" - return f"<% if {self.expr_to_ruby(cond)} -%>" - - if stmt == "else": - return "<% else -%>" - - if stmt.startswith("elif "): - return f"<% elsif {self.expr_to_ruby(stmt[5:].strip())} -%>" - - if stmt == "endif": - return "<% end -%>" - - # Preserve unknown Jinja statements visibly as an ERB comment so the - # generated template does not contain invalid Jinja syntax. - return f"<%# Unsupported JinjaTurtle statement: {stmt} %>" - - -def translate_jinja2_to_erb( - template_text: str, - *, - role_prefix: str, - puppet_class: str | None = None, - variable_names: set[str] | None = None, -) -> str: - return ErbTranslator( - role_prefix=role_prefix, - puppet_class=puppet_class, - variable_names=variable_names, - ).translate(template_text) diff --git a/src/jinjaturtle/escape.py b/src/jinjaturtle/escape.py index 129c08a..e55f416 100644 --- a/src/jinjaturtle/escape.py +++ b/src/jinjaturtle/escape.py @@ -9,7 +9,7 @@ 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. +executed when Ansible later renders the template. Because JinjaTurtle is frequently fed harvested, attacker-influenceable config (hostnames, banners, GECOS-derived comments, "Managed by" notes), this is a @@ -30,14 +30,6 @@ Design notes: 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 @@ -70,11 +62,6 @@ def contains_jinja_markup(text: str) -> bool: 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. @@ -109,45 +96,3 @@ def escape_jinja_literal(text: str) -> str: 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/safety.py b/src/jinjaturtle/safety.py index 07d3b8d..713d49a 100644 --- a/src/jinjaturtle/safety.py +++ b/src/jinjaturtle/safety.py @@ -22,8 +22,8 @@ single global property: The check is positive/allowlist-based, which is the safe direction: unknown constructs are rejected, not ignored. It runs at the single choke points in -``core.py`` (``generate_jinja2_template`` / ``generate_erb_template``), so it -covers every current handler and every future one automatically. +``core.py`` (``generate_jinja2_template``) so it covers every current handler and +every future one automatically. Why this is robust against the escaper being wrong --------------------------------------------------- diff --git a/tests/test_cli.py b/tests/test_cli.py index e4ac519..c0d5470 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -132,57 +132,3 @@ def test_cli_folder_single_output_file_when_one_format(tmp_path): assert defaults_path.is_file() assert template_path.is_file() assert "to_json" in template_path.read_text(encoding="utf-8") - - -def test_cli_erb_outputs_puppet_hiera_and_erb_template(tmp_path): - cfg = tmp_path / "app.ini" - cfg.write_text("[main]\nport = 8080\n", encoding="utf-8") - data_out = tmp_path / "common.yaml" - template_out = tmp_path / "app.ini.erb" - - exit_code = cli._main( - [ - str(cfg), - "-r", - "php", - "--template-engine", - "erb", - "--defaults-output", - str(data_out), - "--template-output", - str(template_out), - ] - ) - - assert exit_code == 0 - assert "php::main_port: '8080'" in data_out.read_text(encoding="utf-8") - assert "port = <%= @main_port %>" in template_out.read_text(encoding="utf-8") - - -def test_cli_erb_can_use_separate_puppet_class_namespace(tmp_path): - cfg = tmp_path / "app.ini" - cfg.write_text("[main]\nport = 8080\n", encoding="utf-8") - data_out = tmp_path / "node.yaml" - template_out = tmp_path / "app.ini.erb" - - exit_code = cli._main( - [ - str(cfg), - "-r", - "php_etc_app_ini", - "--template-engine", - "erb", - "--puppet-class", - "php", - "--defaults-output", - str(data_out), - "--template-output", - str(template_out), - ] - ) - - assert exit_code == 0 - data = data_out.read_text(encoding="utf-8") - template = template_out.read_text(encoding="utf-8") - assert "php::php_etc_app_ini_main_port: '8080'" in data - assert "port = <%= @php_etc_app_ini_main_port %>" in template diff --git a/tests/test_injection_security.py b/tests/test_injection_security.py index a1de483..9ff0230 100644 --- a/tests/test_injection_security.py +++ b/tests/test_injection_security.py @@ -3,8 +3,8 @@ 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. +config content becomes live template code that executes when Ansible 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 @@ -14,7 +14,6 @@ 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 @@ -25,8 +24,6 @@ import yaml as pyyaml from jinjaturtle.escape import ( escape_jinja_literal, - escape_erb_literal, - escape_literal, ) TRIP = "__TRIPWIRE_FIRED__" @@ -233,43 +230,3 @@ def test_endraw_whitespace_control_cannot_break_out(endraw): rendered = env.from_string(escaped).render(boom=_Boom()) assert TRIP not in rendered assert rendered == payload - - -@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. - - -def test_escape_literal_dispatches_by_engine(): - """The public ``escape_literal`` wrapper routes to the right engine.""" - payload = "{{ 7*7 }}" - # Default engine is Jinja2. - assert escape_literal(payload) == escape_jinja_literal(payload) - assert escape_literal(payload, engine="jinja2") == escape_jinja_literal(payload) - # An unknown engine falls back to the safer Jinja2 escaping. - assert escape_literal(payload, engine="nonsense") == escape_jinja_literal(payload) - # ERB routing. - erb_payload = "<%= system('id') %>" - assert escape_literal(erb_payload, engine="erb") == escape_erb_literal(erb_payload) - - -def test_escape_literal_jinja_output_is_inert(): - """End-to-end: text routed through escape_literal does not execute.""" - escaped = escape_literal("{{ boom.run('x') }}") - env = jinja2.Environment(undefined=jinja2.ChainableUndefined) - rendered = env.from_string(escaped).render(boom=_Boom()) - assert TRIP not in rendered diff --git a/tests/test_yaml_handler.py b/tests/test_yaml_handler.py index a217e8a..548f5a6 100644 --- a/tests/test_yaml_handler.py +++ b/tests/test_yaml_handler.py @@ -10,7 +10,6 @@ from jinjaturtle.core import ( analyze_loops, flatten_config, generate_ansible_yaml, - generate_erb_template, generate_jinja2_template, ) from jinjaturtle.handlers.yaml import YamlHandler @@ -205,16 +204,6 @@ def test_yaml_loop_preserves_blank_separator_after_list(tmp_path: Path): rendered = Template(template).render(**defaults) assert rendered_expected in rendered - erb_template = generate_erb_template( - fmt, - parsed, - "role", - original_text=text, - loop_candidates=loop_candidates, - flat_items=flat_items, - ) - assert erb_expected in erb_template - def test_yaml_loop_preserves_following_top_level_comments(tmp_path: Path): from jinja2 import Environment, Template From 4a1f2ac15e2ba51c018e8fed8f9a079c3e5d0539 Mon Sep 17 00:00:00 2001 From: Miguel Jacq Date: Sun, 28 Jun 2026 20:33:12 +1000 Subject: [PATCH 3/4] More JSON defenses --- src/jinjaturtle/core.py | 8 +++ src/jinjaturtle/handlers/json.py | 34 +++++++++++-- src/jinjaturtle/multi.py | 4 +- src/jinjaturtle/safety.py | 84 +++++++++++++++++++++++++++++++- tests/test_injection_security.py | 80 ++++++++++++++++++++++++++++++ 5 files changed, 203 insertions(+), 7 deletions(-) diff --git a/src/jinjaturtle/core.py b/src/jinjaturtle/core.py index 8e3d10e..3e9cefd 100644 --- a/src/jinjaturtle/core.py +++ b/src/jinjaturtle/core.py @@ -10,6 +10,7 @@ import yaml from .loop_analyzer import LoopAnalyzer, LoopCandidate from .safety import ( verify_jinja2_template_safe, + verify_no_live_jinja_in_json_keys, ) from .handlers import ( BaseHandler, @@ -394,6 +395,13 @@ def generate_jinja2_template( # verbatim source text, the un-escaped payload shows up here as a live tag # and generation aborts instead of emitting an injectable template. verify_jinja2_template_safe(template) + + # Format-specific backstop: JinjaTurtle never emits Jinja inside a JSON object + # key, so a live construct in key position means source key text leaked into + # the template unescaped. This is independent of per-handler escaping. + if fmt == "json": + verify_no_live_jinja_in_json_keys(template) + return template diff --git a/src/jinjaturtle/handlers/json.py b/src/jinjaturtle/handlers/json.py index 064535a..0268666 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,18 @@ class JsonHandler(DictLikeHandler): chunks: list[str] = [] pos = 0 for path, start, end in spans: - chunks.append(text[pos:start]) + # Text between scalar values (object keys, structural punctuation, + # whitespace, and any comment-like trailing text) is copied verbatim + # from the source file. Like every other text-emitting handler, this + # verbatim text must be neutralised: if it contains Jinja2 markup it + # would otherwise become live template code at apply time. The value + # itself is replaced with a safe placeholder below. ``escape_jinja_literal`` + # is a no-op on text without Jinja markers, so benign JSON is unchanged + # byte-for-byte and a later render reproduces the original characters. + 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 +222,12 @@ 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()} + # Keys are emitted verbatim into the template, so neutralise any + # Jinja markup in them (see _generate_json_template_from_text). + 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 +275,12 @@ 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()} + # Keys are emitted verbatim into the template, so neutralise any + # Jinja markup in them (see _generate_json_template_from_text). + 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 +383,13 @@ 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 "" + # The literal key text is emitted verbatim into the template; escape any + # Jinja markup in it. The value side ({item_var}.{key}) is constrained by + # the output safety gate's dotted-name allowlist, which fails closed on + # anything that is not a plain identifier path. dict_lines.append( - f'{field}"{key}": ' f"{j2.to_json(f'{item_var}.{key}')}{comma}" + f'{field}"{escape_jinja_literal(str(key))}": ' + f"{j2.to_json(f'{item_var}.{key}')}{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/multi.py b/src/jinjaturtle/multi.py index 195434f..c945481 100644 --- a/src/jinjaturtle/multi.py +++ b/src/jinjaturtle/multi.py @@ -31,7 +31,7 @@ import xml.etree.ElementTree as ET # nosec from . import j2 from .core import dump_yaml, flatten_config, make_var_name, parse_config from .handlers.xml import XmlHandler -from .safety import verify_jinja2_template_safe +from .safety import verify_jinja2_template_safe, verify_no_live_jinja_in_json_keys from .escape import escape_jinja_literal @@ -784,5 +784,7 @@ def process_directory( # Any un-neutralised source text that became a live tag aborts generation. for out in outputs: verify_jinja2_template_safe(out.template) + if out.fmt == "json": + verify_no_live_jinja_in_json_keys(out.template) return defaults_yaml, outputs diff --git a/src/jinjaturtle/safety.py b/src/jinjaturtle/safety.py index 713d49a..5ed2fe1 100644 --- a/src/jinjaturtle/safety.py +++ b/src/jinjaturtle/safety.py @@ -43,6 +43,7 @@ __all__ = [ "TemplateSafetyError", "verify_jinja2_template_safe", "verify_erb_template_safe", + "verify_no_live_jinja_in_json_keys", ] @@ -208,7 +209,88 @@ def verify_jinja2_template_safe(template_text: str) -> None: # --------------------------------------------------------------------------- # -# ERB gate. +# JSON-key gate (defence in depth, format-specific). +# +# JinjaTurtle never emits a live Jinja construct inside a JSON *object key*: keys +# are copied verbatim from the source and (after escape.py) are wrapped in +# ``{% raw %}`` if they contain markup, so a key never lexes as a live tag. A +# live construct in key position can therefore only mean source key text leaked +# into the template unescaped (the json-handler blind spot). This gate is +# independent of the escaper: it inspects the finished template, replaces every +# *live* Jinja construct with an inert sentinel (raw-wrapped literal text stays +# literal), and rejects any sentinel that lands in a JSON key slot. +# --------------------------------------------------------------------------- # + +# Sentinel byte that cannot occur in normal generated template text. +_LIVE_SENTINEL = "\x00" + +# A JSON key is a double-quoted string immediately followed (after optional +# whitespace) by a colon. We only need to detect a sentinel *inside* such a +# string, so match a quoted run that ends in `":` and look for the sentinel. +_JSON_KEY_RE = re.compile(r'"((?:[^"\\]|\\.)*)"\s*:', re.S) + +_KEY_JINJA_DELIMS = ("{{", "}}", "{%", "%}", "{#", "#}") + + +def _contains_jinja_delim(text: str) -> bool: + """True if *text* contains any Jinja delimiter (live or escaped-literal).""" + return any(d in text for d in _KEY_JINJA_DELIMS) + + +def verify_no_live_jinja_in_json_keys(template_text: str) -> None: + """Reject a JSON template that carries Jinja markup in an object key. + + JinjaTurtle never templates a JSON *object key*: keys come straight from the + source and a key is an identifier/string, never a value placeholder. Any + Jinja in key position therefore means attacker-influenced source key text + reached the template. This gate fails closed on it, independent of whether a + handler left the markup *live* (a raw ``{{ ... }}`` in the key) or *escaped* + it into a ``{% raw %}`` wrapper -- both indicate a key that should never have + contained templating, so generation aborts rather than emitting it. + + Detection is done on the lexer token stream: we reconstruct the document with + every live tag collapsed to a sentinel and every ``{% raw %}``-wrapped region + also marked, then reject a sentinel that lands inside a JSON key string. + """ + import jinja2 + + env = jinja2.Environment(autoescape=True) + try: + tokens = list(env.lex(template_text)) + except jinja2.TemplateSyntaxError as exc: + raise TemplateSafetyError( + f"generated JSON template does not lex as the JinjaTurtle subset: {exc}" + ) from exc + + out: list[str] = [] + in_tag = False + for _lineno, tok_type, value in tokens: + if tok_type in ("variable_begin", "block_begin", "comment_begin"): + # A live construct: collapse to a sentinel so it is detectable if it + # sits in key position. (raw_begin/raw_end are *not* live; the data + # inside a raw block is preserved verbatim below, so an escaped key + # still shows its literal Jinja delimiters to the key check.) + in_tag = True + out.append(_LIVE_SENTINEL) + elif tok_type in ("variable_end", "block_end", "comment_end"): + in_tag = False + elif not in_tag: + # data, whitespace, raw_begin/raw_end markers, and inert raw content. + out.append(value if isinstance(value, str) else str(value)) + + reconstructed = "".join(out) + + for match in _JSON_KEY_RE.finditer(reconstructed): + key_text = match.group(1) + if _LIVE_SENTINEL in key_text or _contains_jinja_delim(key_text): + raise TemplateSafetyError( + "refusing to emit JSON template: Jinja markup appears inside a " + "JSON object key. JinjaTurtle never templates keys, so this " + "indicates attacker-influenced source key text (possible " + "template injection)." + ) + + # # JinjaTurtle's ERB output is produced by translating the (already-verified) # Jinja2 subset, so the Jinja2 gate is the primary guarantee. As an independent diff --git a/tests/test_injection_security.py b/tests/test_injection_security.py index 9ff0230..1110ef1 100644 --- a/tests/test_injection_security.py +++ b/tests/test_injection_security.py @@ -152,6 +152,86 @@ def test_generated_template_is_renderable(tmp_path, fmt, name, body): _render_jinja(template_text, defaults) +# --- JSON object-key injection ------------------------------------------------ +# +# The JSON handler copies the text *between* scalar values (object keys, +# punctuation) verbatim. A key is never a value placeholder, so Jinja markup in a +# key can only come from attacker-influenced source text. The output gate +# (verify_no_live_jinja_in_json_keys) must fail closed on it -- including the +# "benign-looking name" form (e.g. ``{{ ansible_hostname }}``) that the generic +# allowlist would otherwise accept as an ordinary variable reference, and which +# could leak an in-scope variable's value into the rendered config at apply time. + +JSON_KEY_INJECTION_BODIES = [ + # benign-looking variable reference (the residual bypass: passes the generic + # allowlist but must still be rejected in *key* position) + '{ "{{ ansible_hostname }}": "v" }', + # dotted reference (e.g. dumping another host's vars) + '{ "{{ hostvars.localhost }}": 1 }', + # self-referencing a sibling-derived variable name + '{ "{{ role_port }}": "x", "port": 8080 }', + # classic gadget (already rejected historically; kept as a guard) + '{ "{{ cycler.__init__.__globals__ }}": 1 }', + # statement injection in a key + '{ "{% for x in y %}k{% endfor %}": 1 }', + # nested object key + '{ "ok": { "{{ ansible_hostname }}": 2 } }', +] + + +@pytest.mark.parametrize("body", JSON_KEY_INJECTION_BODIES) +def test_json_key_injection_fails_closed_cli(tmp_path, body): + src = tmp_path / "evil.json" + src.write_text(body, encoding="utf-8") + out = tmp_path / "out.j2" + res = subprocess.run( + [ + sys.executable, + "-m", + "jinjaturtle.cli", + str(src), + "-f", + "json", + "--role-name", + "role", + "-t", + str(out), + ], + capture_output=True, + text=True, + ) + assert res.returncode == 2, f"expected fail-closed, got rc={res.returncode}" + assert "refusing to generate unsafe template" in res.stderr + assert not out.exists(), "no template may be written when the gate refuses" + + +def test_json_benign_keys_still_generate(tmp_path): + src = tmp_path / "ok.json" + src.write_text('{ "host": "localhost", "port": 8080 }', encoding="utf-8") + out = tmp_path / "out.j2" + res = subprocess.run( + [ + sys.executable, + "-m", + "jinjaturtle.cli", + str(src), + "-f", + "json", + "--role-name", + "demo", + "-t", + str(out), + ], + capture_output=True, + text=True, + ) + assert res.returncode == 0, res.stderr + template_text = out.read_text() + # Keys stay literal; values become placeholders. + assert '"host":' in template_text + assert "demo_host" in template_text + + # --- Unit-level guarantees for the escaper itself --------------------------- SSTI_PAYLOADS = [ From 70be0f7e33ff04362036dc03d5403b3984201f5b Mon Sep 17 00:00:00 2001 From: Miguel Jacq Date: Mon, 29 Jun 2026 08:51:53 +1000 Subject: [PATCH 4/4] Hardening: use unsafe for ansible vars, ensure API use of JinjaTurtle uses safe XML parsing, avoid symlinks --- src/jinjaturtle/cli.py | 18 +++-- src/jinjaturtle/core.py | 45 ++++++++++- src/jinjaturtle/handlers/xml.py | 13 ++-- src/jinjaturtle/multi.py | 30 +++++++- src/jinjaturtle/output_safety.py | 124 +++++++++++++++++++++++++++++++ tests/test_injection_security.py | 17 ++++- tests/test_security_hardening.py | 120 ++++++++++++++++++++++++++++++ 7 files changed, 348 insertions(+), 19 deletions(-) create mode 100644 src/jinjaturtle/output_safety.py create mode 100644 tests/test_security_hardening.py diff --git a/src/jinjaturtle/cli.py b/src/jinjaturtle/cli.py index 0eaf506..bdabc18 100644 --- a/src/jinjaturtle/cli.py +++ b/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( - o.template, encoding="utf-8" + write_text_safely( + 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="") diff --git a/src/jinjaturtle/core.py b/src/jinjaturtle/core.py index 3e9cefd..72b42de 100644 --- a/src/jinjaturtle/core.py +++ b/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, diff --git a/src/jinjaturtle/handlers/xml.py b/src/jinjaturtle/handlers/xml.py index 9fde3ec..56e3fd7 100644 --- a/src/jinjaturtle/handlers/xml.py +++ b/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( - target=ET.TreeBuilder(insert_comments=False) - ) # nosec B314 - parser.feed(text) - root = parser.close() + # Security must live in the handler, not only in the CLI entry point: + # callers may import JinjaTurtle as a library and invoke parse_config() + # directly. defusedxml rejects DTD/entity abuse and also discards + # comments by default, matching the previous TreeBuilder behaviour. + root = DET.fromstring(text) return root def flatten(self, parsed: Any) -> list[tuple[tuple[str, ...], Any]]: diff --git a/src/jinjaturtle/multi.py b/src/jinjaturtle/multi.py index c945481..c968d61 100644 --- a/src/jinjaturtle/multi.py +++ b/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("*") diff --git a/src/jinjaturtle/output_safety.py b/src/jinjaturtle/output_safety.py new file mode 100644 index 0000000..9833cde --- /dev/null +++ b/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 diff --git a/tests/test_injection_security.py b/tests/test_injection_security.py index 1110ef1..4afb7a5 100644 --- a/tests/test_injection_security.py +++ b/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 diff --git a/tests/test_security_hardening.py b/tests/test_security_hardening.py new file mode 100644 index 0000000..41a6b05 --- /dev/null +++ b/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( + "]>&xxe;", + 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"