diff --git a/README.md b/README.md index 5e104f3..2905cfe 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,9 @@ 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 @@ -34,6 +37,16 @@ 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. @@ -67,6 +80,78 @@ 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: @@ -91,8 +176,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 templates, not to guarantee a perfect final module without -review. +conversion into Jinja2 or ERB templates, not to guarantee a perfect final module +without review. ## JSON, quoting, and type preservation @@ -111,7 +196,17 @@ when the correct rendered JSON should be: {"enabled": true} ``` -This uses Ansible-style JSON filters. +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? @@ -192,6 +287,8 @@ 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. @@ -205,7 +302,8 @@ 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. Defaults to jinjaturtle. + usually the Ansible role name. In ERB mode it is used + as the local 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 @@ -216,6 +314,12 @@ 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 @@ -242,16 +346,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 data. When the template is later rendered, + 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 is inert. + a value (for example `motd = {{ salt['cmd.run']('id') }}`) is inert. 2. **Verbatim text is neutralised.** To preserve formatting, JinjaTurtle copies comments, blank lines, headers and any unrecognised lines from the source into the template. Any template metacharacters in that copied text - (`{{ }}`, `{% %}`, `{# #}` ) are escaped so they render as the literal - characters the author wrote, rather than executing. + (`{{ }}`, `{% %}`, `{# #}` for Jinja2; `<% %>` for ERB) are escaped so they + render as the literal characters the author wrote, rather than executing. ### Consumer responsibilities @@ -264,6 +368,11 @@ 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. @@ -275,4 +384,7 @@ management system! ## Found a bug, have a suggestion? -You can e-mail me; see `pyproject.toml` for details. +You can e-mail me; see `pyproject.toml` for details. You can also contact me on +the Fediverse: + +https://goto.mig5.net/@mig5 diff --git a/debian/changelog b/debian/changelog index be6b123..d32bd4f 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,15 +1,3 @@ -jinjaturtle (0.7.0) unstable; urgency=medium - - * Much hardening. - - -- Miguel Jacq Sun, 5 Jul 2026 10:54:00 +1000 - -jinjaturtle (0.5.7) unstable; urgency=medium - - * More hardening measures - - -- Miguel Jacq Wed, 24 Jun 2026 16:13:00 +1000 - jinjaturtle (0.5.6) unstable; urgency=medium * Try to prevent what could lead to execution of embedded jinja in original files when converting diff --git a/pyproject.toml b/pyproject.toml index 7751303..d856172 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "jinjaturtle" -version = "0.7.0" +version = "0.5.6" description = "Convert config files into Ansible defaults and Jinja2 templates." authors = [ { name = "Miguel Jacq", email = "mig@mig5.net" }, diff --git a/release.sh b/release.sh index e221cef..874e50e 100755 --- a/release.sh +++ b/release.sh @@ -46,6 +46,7 @@ REPO_ROOT="${HOME}/git/repo_rpm" REMOTE="ashpool.mig5.net:/opt/repo_rpm" DISTS=( + fedora:44 fedora:43 ) diff --git a/rpm/jinjaturtle.spec b/rpm/jinjaturtle.spec index 8c8d845..2ed4063 100644 --- a/rpm/jinjaturtle.spec +++ b/rpm/jinjaturtle.spec @@ -1,4 +1,4 @@ -%global upstream_version 0.7.0 +%global upstream_version 0.5.6 Name: jinjaturtle Version: %{upstream_version} @@ -42,10 +42,6 @@ Convert config files into Ansible defaults and Jinja2 templates. %{_bindir}/jinjaturtle %changelog -* Sun Jul 05 2026 Miguel Jacq - %{version}-%{release} -- Much hardening -* Wed Jun 24 2026 Miguel Jacq - %{version}-%{release} -- More hardening * Tue Jun 23 2026 Miguel Jacq - %{version}-%{release} - Try to prevent what could lead to execution of embedded jinja in original files when converting * Sat Jun 20 2026 Miguel Jacq - %{version}-%{release} @@ -59,7 +55,7 @@ Convert config files into Ansible defaults and Jinja2 templates. - Fix indentation problems with nested dicts * Fri Jun 19 2026 Miguel Jacq - %{version}-%{release} - Empty dicts and lists are now emitted as leaf defaults. -* Mon May 11 2026 Miguel Jacq - %{version}-%{release} +* Tue May 11 2026 Miguel Jacq - %{version}-%{release} - Support ssh configs * Tue Jan 06 2026 Miguel Jacq - %{version}-%{release} - Support converting systemd files and postfix main.cf diff --git a/src/jinjaturtle/cli.py b/src/jinjaturtle/cli.py index d901606..0c403fa 100644 --- a/src/jinjaturtle/cli.py +++ b/src/jinjaturtle/cli.py @@ -3,7 +3,6 @@ from __future__ import annotations import argparse import sys from defusedxml import defuse_stdlib -from defusedxml.common import DefusedXmlException from pathlib import Path from . import j2 @@ -13,12 +12,11 @@ from .core import ( flatten_config, generate_ansible_yaml, generate_jinja2_template, - ConfigParseError, + generate_puppet_hiera_yaml, + generate_erb_template, ) 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: @@ -60,47 +58,24 @@ 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 def _main(argv: list[str] | None = None) -> int: - try: - return _run(argv) - except TemplateSafetyError as exc: - # The output safety gate refused to emit a template because it contained - # a construct JinjaTurtle never produces -- i.e. attacker-influenced - # source text became live template code. Fail closed with a clear - # message and a non-zero exit code; never write the unsafe template. - print( - 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 - except DefusedXmlException as exc: - # defusedxml rejected the XML because it attempted a DTD, entity - # expansion, or external reference (XXE / billion-laughs class attack). - # This is a deliberately-blocked attack, not a benign malformed file, so - # core.parse_config lets it propagate unchanged rather than folding it - # into ConfigParseError. Report it as a refused unsafe input with a - # non-zero exit code instead of leaking an internal traceback. - print( - "jinjaturtle: refusing unsafe XML input: the document uses a DTD, " - f"entity expansion, or external reference ({exc.__class__.__name__}). " - "This is blocked to prevent XXE / entity-expansion attacks.", - file=sys.stderr, - ) - return 2 - except ConfigParseError as exc: - # The source file could not be parsed as its (detected or forced) - # format. This is expected for malformed/attacker-influenced input; - # fail cleanly with a non-zero exit code instead of a traceback. - print(f"jinjaturtle: {exc}", file=sys.stderr) - return 1 - - -def _run(argv: list[str] | None = None) -> int: defuse_stdlib() parser = _build_arg_parser() args = parser.parse_args(argv) @@ -115,23 +90,36 @@ def _run(argv: list[str] | None = None) -> int: # Write defaults if args.defaults_output: - write_text_safely(Path(args.defaults_output), defaults_yaml) + Path(args.defaults_output).write_text(defaults_yaml, encoding="utf-8") else: print("# defaults/main.yml") print(defaults_yaml, end="") - template_ext = j2.TEMPLATE_EXTENSION + # 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, + ) + + 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(): - write_text_safely(out_path, outputs[0].template) + out_path.write_text(outputs[0].template, encoding="utf-8") else: - ensure_safe_directory(out_path) + out_path.mkdir(parents=True, exist_ok=True) for o in outputs: - write_text_safely( - out_path / f"config.{o.fmt}.{template_ext}", o.template + (out_path / f"config.{o.fmt}.{template_ext}").write_text( + o.template, encoding="utf-8" ) else: for o in outputs: @@ -157,28 +145,51 @@ 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) - # Generate defaults YAML (with loop collections if detected) - ansible_yaml = generate_ansible_yaml(args.role_name, flat_items, 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 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: - write_text_safely(Path(args.defaults_output), ansible_yaml) + Path(args.defaults_output).write_text(ansible_yaml, encoding="utf-8") else: print("# defaults/main.yml") print(ansible_yaml, end="") if args.template_output: - write_text_safely(Path(args.template_output), template_str) + Path(args.template_output).write_text(template_str, encoding="utf-8") else: - print(f"# config.{j2.TEMPLATE_EXTENSION}") + print( + "# config.erb" + if args.template_engine == "erb" + else f"# config.{j2.TEMPLATE_EXTENSION}" + ) print(template_str, end="") return 0 diff --git a/src/jinjaturtle/core.py b/src/jinjaturtle/core.py index de7b552..1faefca 100644 --- a/src/jinjaturtle/core.py +++ b/src/jinjaturtle/core.py @@ -8,10 +8,7 @@ import re import yaml from .loop_analyzer import LoopAnalyzer, LoopCandidate -from .safety import ( - verify_jinja2_template_safe, - verify_no_live_jinja_in_json_keys, -) +from .erb import puppet_class_name, puppet_local_var_name, translate_jinja2_to_erb from .handlers import ( BaseHandler, IniHandler, @@ -33,22 +30,6 @@ 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. @@ -68,33 +49,7 @@ 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) @@ -126,9 +81,8 @@ 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( - safe_data, + data, Dumper=_TurtleDumper, sort_keys=sort_keys, default_flow_style=False, @@ -305,66 +259,6 @@ def detect_format(path: Path, explicit: str | None = None) -> str: return "ini" -class ConfigParseError(Exception): - """Raised when a source config file cannot be parsed as its format. - - Each underlying parser (json, tomllib, PyYAML, defusedxml/ElementTree, - configparser) raises its own exception type on malformed input. Without a - single normalised error, a malformed file -- which is entirely expected when - JinjaTurtle is pointed at harvested, attacker-influenceable config -- would - escape as an unhandled traceback (e.g. ``xml.etree.ElementTree.ParseError`` - on an XML file whose element name is not well-formed). ``parse_config`` - converts every such failure into this one type so the CLI can fail closed - with a clean message and a non-zero exit code, and so library callers (such - as Enroll, which falls back to copying the raw file) have a single, stable - exception to catch. - - Note: defusedxml's *security* exceptions (``EntitiesForbidden``, - ``DTDForbidden``, ...) are intentionally NOT folded into this type. They - signal an attempted XXE/entity-expansion attack rather than a benign - malformed file, and must propagate unchanged so callers can tell the two - apart. - """ - - -def _build_malformed_config_errors() -> tuple[type[BaseException], ...]: - """Return the concrete "this file is malformed" exception types to catch. - - Deliberately specific. In particular we must avoid catching plain - ``ValueError``: defusedxml's ``EntitiesForbidden``/``DTDForbidden`` subclass - ``ValueError``, and those are security signals that must NOT be swallowed. - """ - - import configparser - import json - from xml.etree.ElementTree import ParseError as _XMLParseError # nosec - - import yaml as _yaml - - errs: list[type[BaseException]] = [ - _XMLParseError, - json.JSONDecodeError, - configparser.Error, - _yaml.YAMLError, - UnicodeDecodeError, - ] - try: - import tomllib - - errs.append(tomllib.TOMLDecodeError) - except ModuleNotFoundError: # pragma: no cover - Python < 3.11 fallback - try: - import tomli # type: ignore - - errs.append(tomli.TOMLDecodeError) - except ModuleNotFoundError: - pass - return tuple(errs) - - -_MALFORMED_CONFIG_ERRORS = _build_malformed_config_errors() - - def parse_config(path: Path, fmt: str | None = None) -> tuple[str, Any]: """ Parse config file into a Python object. @@ -373,38 +267,9 @@ 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}") - try: - parsed = handler.parse(path) - # Make sure datetime objects are treated as strings (TOML, YAML). This - # walks the parsed object recursively, so keep it inside the try where a - # RecursionError from a pathological structure is normalised below. - parsed = _stringify_timestamps(parsed) - except ConfigParseError: - raise - except _MALFORMED_CONFIG_ERRORS as exc: - # Normalise the per-parser "this file is malformed" errors into one - # json.JSONDecodeError / tomllib.TOMLDecodeError (ValueError - # subclasses), PyYAML's YAMLError, configparser.Error, and - # xml.etree.ElementTree.ParseError (raised by defusedxml on XML whose - # structure/element name is not well-formed). A bad input file is - # expected when parsing harvested config, so fail closed with a clean - # error instead of an unhandled traceback. - # - # IMPORTANT: this deliberately does NOT catch defusedxml's security - # exceptions (EntitiesForbidden, DTDForbidden, ...). Those signal an - # attempted XXE/entity-expansion attack and must propagate unchanged so - # callers (and tests) can distinguish "malformed" from "malicious". - raise ConfigParseError(f"could not parse {path} as {fmt}: {exc}") from exc - except RecursionError as exc: - # A deeply-nested or self-referential structure (e.g. a recursive YAML - # anchor) can exhaust the Python stack while walking the parsed object. - # The YAML handler already rejects reference cycles up front; this is a - # format-agnostic backstop so any such input fails closed with a clean - # message instead of a stack-overflow traceback. - raise ConfigParseError( - f"could not parse {path} as {fmt}: input is too deeply nested " - "or self-referential" - ) from exc + parsed = handler.parse(path) + # Make sure datetime objects are treated as strings (TOML, YAML) + parsed = _stringify_timestamps(parsed) return fmt, parsed @@ -513,28 +378,91 @@ def generate_jinja2_template( # Check if handler supports loop-aware generation if hasattr(handler, "generate_jinja2_template_with_loops") and loop_candidates: - template = handler.generate_jinja2_template_with_loops( + return handler.generate_jinja2_template_with_loops( parsed, role_prefix, original_text, loop_candidates ) - else: - # Fallback to original scalar-only generation - template = handler.generate_jinja2_template( - parsed, role_prefix, original_text=original_text - ) - # Defence in depth: independently verify that the finished template contains - # only JinjaTurtle-emitted constructs. If any handler failed to neutralise - # 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) + # Fallback to original scalar-only generation + return handler.generate_jinja2_template( + parsed, role_prefix, original_text=original_text + ) - # 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 +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) + return translate_jinja2_to_erb( + jinja_template, + role_prefix=role_prefix, + puppet_class=puppet_class or role_prefix, + variable_names=names, + ) def _stringify_timestamps(obj: Any) -> Any: diff --git a/src/jinjaturtle/erb.py b/src/jinjaturtle/erb.py new file mode 100644 index 0000000..6eb26ad --- /dev/null +++ b/src/jinjaturtle/erb.py @@ -0,0 +1,241 @@ +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 e55f416..3b8575e 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 Ansible later renders the template. +executed when Salt/Ansible/Puppet later renders the template. Because JinjaTurtle is frequently fed harvested, attacker-influenceable config (hostnames, banners, GECOS-derived comments, "Managed by" notes), this is a @@ -30,6 +30,14 @@ 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 @@ -42,14 +50,9 @@ _JINJA_MARKERS = ("{{", "}}", "{%", "%}", "{#", "#}") _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 -# 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 -# attacker-influenced source text cannot survive defanging and break out of our -# {% raw %} wrapper. [-+]? appears on both sides accordingly. -_ENDRAW_RE = re.compile(r"{%[-+]?\s*endraw\s*[-+]?%}") +# Matches a Jinja2 endraw tag in any internal spacing, e.g. "{%endraw%}", +# "{% endraw %}", "{%- endraw -%}". +_ENDRAW_RE = re.compile(r"{%-?\s*endraw\s*-?%}") # Sentinel inserted between "end" and "raw" to break the endraw keyword without # changing the visible characters. We use a Jinja comment-free approach: insert @@ -62,6 +65,11 @@ 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. @@ -96,3 +104,45 @@ 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/handlers/base.py b/src/jinjaturtle/handlers/base.py index 6be91f0..14aaec7 100644 --- a/src/jinjaturtle/handlers/base.py +++ b/src/jinjaturtle/handlers/base.py @@ -1,6 +1,5 @@ from __future__ import annotations -import re from pathlib import Path from typing import Any, Iterable @@ -58,20 +57,8 @@ class BaseHandler: role_prefix_section_subsection_key Sanitises parts to lowercase [a-z0-9_] and strips extras. - - Consecutive separators are collapsed to a single underscore. This is - required for correctness, not just aesthetics: a source key such as - ``log..level`` or ``cache--size`` would otherwise sanitise to a name - containing a double underscore (``log__level``). The output safety gate - in ``safety.py`` deliberately rejects *any* ``__`` in a generated - identifier because ``__`` is the gateway to every Jinja2 SSTI gadget - (``__class__``/``__globals__``/...). Emitting a dunder here would make - JinjaTurtle's own gate reject JinjaTurtle's own placeholder, aborting - generation on entirely benign config. Collapsing runs keeps every - generated name a plain single-underscore-delimited identifier that the - gate accepts. """ - role_prefix = re.sub(r"_+", "_", role_prefix.strip().lower()) + role_prefix = role_prefix.strip().lower() clean_parts: list[str] = [] for part in path: @@ -83,9 +70,7 @@ class BaseHandler: cleaned_chars.append(c.lower()) else: cleaned_chars.append("_") - # Collapse runs of underscores (from adjacent separators) to a - # single "_" so the result can never contain a forbidden "__". - cleaned_part = re.sub(r"_+", "_", "".join(cleaned_chars)).strip("_") + cleaned_part = "".join(cleaned_chars).strip("_") if cleaned_part: clean_parts.append(cleaned_part) diff --git a/src/jinjaturtle/handlers/json.py b/src/jinjaturtle/handlers/json.py index 41699c9..064535a 100644 --- a/src/jinjaturtle/handlers/json.py +++ b/src/jinjaturtle/handlers/json.py @@ -7,8 +7,7 @@ from typing import Any from . import DictLikeHandler from .. import j2 -from ..escape import escape_jinja_literal -from ..loop_analyzer import LoopCandidate, is_safe_loop_field_key +from ..loop_analyzer import LoopCandidate class JsonHandler(DictLikeHandler): @@ -110,18 +109,10 @@ class JsonHandler(DictLikeHandler): chunks: list[str] = [] pos = 0 for path, start, end in spans: - # 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(text[pos:start]) chunks.append(self._json_value_expr(self.make_var_name(role_prefix, path))) pos = end - chunks.append(escape_jinja_literal(text[pos:])) + chunks.append(text[pos:]) return "".join(chunks) def _collect_json_scalar_spans( @@ -222,12 +213,7 @@ class JsonHandler(DictLikeHandler): def _walk(obj: Any, path: tuple[str, ...] = ()) -> Any: if isinstance(obj, dict): - # 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() - } + 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 @@ -275,12 +261,7 @@ class JsonHandler(DictLikeHandler): return f"__LOOP_DICT__{collection_var}__{item_var}__" if isinstance(obj, dict): - # 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() - } + 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: @@ -383,21 +364,8 @@ 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 "" - # Defence in depth: never interpolate a raw source key into an - # ``item_var.key`` reference. A key such as ``a }}{{ x`` would break - # out of the value placeholder and inject a live construct that the - # output-safety gate cannot distinguish from a legitimate variable. - if not is_safe_loop_field_key(key): - raise ValueError( - f"refusing to emit loop-item field reference for unsafe key: {key!r}" - ) - # 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}"{escape_jinja_literal(str(key))}": ' - f"{j2.to_json(f'{item_var}.{key}')}{comma}" + f'{field}"{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/toml.py b/src/jinjaturtle/handlers/toml.py index b2aa6e5..6d25cff 100644 --- a/src/jinjaturtle/handlers/toml.py +++ b/src/jinjaturtle/handlers/toml.py @@ -6,7 +6,7 @@ from typing import Any from . import DictLikeHandler from .. import j2 from ..escape import escape_jinja_literal -from ..loop_analyzer import LoopCandidate, is_safe_loop_field_key +from ..loop_analyzer import LoopCandidate try: import tomllib @@ -438,18 +438,6 @@ class TomlHandler(DictLikeHandler): for key, value in sample_item.items(): if key == "_key": continue - # Defence in depth: the loop analyzer refuses a - # dict-loop whose items contain a non-identifier - # key, so ``key`` is always a plain identifier - # here. Never interpolate a raw key into an - # ``item_var.key`` reference: a key such as - # ``a }}{{ x`` would break out of the placeholder - # and inject a live construct. - if not is_safe_loop_field_key(key): - raise ValueError( - "refusing to emit loop-item field " - f"reference for unsafe key: {key!r}" - ) if isinstance(value, str): out_lines.append( f"{escape_jinja_literal(str(key))} = " diff --git a/src/jinjaturtle/handlers/xml.py b/src/jinjaturtle/handlers/xml.py index 56e3fd7..9fde3ec 100644 --- a/src/jinjaturtle/handlers/xml.py +++ b/src/jinjaturtle/handlers/xml.py @@ -3,8 +3,7 @@ from __future__ import annotations from collections import Counter, defaultdict from pathlib import Path from typing import Any -import xml.etree.ElementTree as ET # nosec B405 - safe trees only; parsing uses defusedxml -import defusedxml.ElementTree as DET +import xml.etree.ElementTree as ET # nosec from .base import BaseHandler from .. import j2 @@ -21,11 +20,11 @@ class XmlHandler(BaseHandler): def parse(self, path: Path) -> ET.Element: text = path.read_text(encoding="utf-8") - # 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) + parser = ET.XMLParser( + target=ET.TreeBuilder(insert_comments=False) + ) # nosec B314 + parser.feed(text) + root = parser.close() return root def flatten(self, parsed: Any) -> list[tuple[tuple[str, ...], Any]]: diff --git a/src/jinjaturtle/handlers/yaml.py b/src/jinjaturtle/handlers/yaml.py index 9a868a6..d8131b0 100644 --- a/src/jinjaturtle/handlers/yaml.py +++ b/src/jinjaturtle/handlers/yaml.py @@ -7,40 +7,9 @@ from typing import Any from .dict import DictLikeHandler from .. import j2 from ..escape import escape_jinja_literal -from ..loop_analyzer import is_safe_loop_field_key from ..loop_analyzer import LoopCandidate -def _reject_recursive_structure(obj: Any) -> None: - """Raise ``yaml.YAMLError`` if *obj* contains a reference cycle. - - A recursive YAML anchor (``a: &a [*a]``) produces a container that contains - itself. Every consumer in JinjaTurtle walks the parsed object depth-first, - so a cycle would raise ``RecursionError`` deep in unrelated code. Detect it - up front by tracking the ``id()`` of containers on the current descent path; - a repeat means a cycle. ``yaml.YAMLError`` is raised so ``parse_config`` - normalises it into a clean ``ConfigParseError`` like any other malformed - input, rather than surfacing a stack-overflow traceback. - """ - - on_path: set[int] = set() - - def walk(node: Any) -> None: - if isinstance(node, (dict, list)): - marker = id(node) - if marker in on_path: - raise yaml.YAMLError( - "recursive/self-referential YAML structure is not supported" - ) - on_path.add(marker) - children = node.values() if isinstance(node, dict) else node - for child in children: - walk(child) - on_path.discard(marker) - - walk(obj) - - class YamlHandler(DictLikeHandler): """ YAML handler that can generate both scalar templates and loop-based templates. @@ -51,16 +20,7 @@ class YamlHandler(DictLikeHandler): def parse(self, path: Path) -> Any: text = path.read_text(encoding="utf-8") - parsed = yaml.safe_load(text) or {} - # PyYAML's safe_load happily builds *recursive* structures from an anchor - # that references itself (e.g. ``a: &a [*a]``). Downstream flattening, - # timestamp-stringifying and template generation all walk the parsed - # object recursively and would blow the Python stack (RecursionError) on - # such input. JinjaTurtle is regularly pointed at harvested, - # attacker-influenceable config, so reject a self-referential document - # cleanly here rather than crashing later. - _reject_recursive_structure(parsed) - return parsed + return yaml.safe_load(text) or {} def generate_jinja2_template( self, @@ -587,18 +547,6 @@ class YamlHandler(DictLikeHandler): # Special key for dict collections - output as comment or skip continue - # Defence in depth: the loop analyzer already refuses a dict-loop - # whose items contain a non-identifier key (see _analyze_dict_schema), - # so ``key`` should always be a plain identifier here. Assert it - # rather than interpolate a raw key into a Jinja reference: a key such - # as ``a }}{{ x`` would otherwise close the placeholder and inject a - # live construct that the output gate cannot distinguish from a - # legitimate variable. - if not is_safe_loop_field_key(key): - raise ValueError( - f"refusing to emit loop-item field reference for unsafe key: {key!r}" - ) - if first_key and is_list_item: # First key gets the list marker value_expr = self._yaml_value_expr(f"{loop_var}.{key}", value) diff --git a/src/jinjaturtle/j2.py b/src/jinjaturtle/j2.py index e20ed20..13531f5 100644 --- a/src/jinjaturtle/j2.py +++ b/src/jinjaturtle/j2.py @@ -2,6 +2,7 @@ from __future__ import annotations from typing import Any +NAME = "jinja2" TEMPLATE_EXTENSION = "j2" JSON_VALUE_FILTER = "to_json(ensure_ascii=False)" diff --git a/src/jinjaturtle/loop_analyzer.py b/src/jinjaturtle/loop_analyzer.py index 608c8e4..23702d8 100644 --- a/src/jinjaturtle/loop_analyzer.py +++ b/src/jinjaturtle/loop_analyzer.py @@ -7,34 +7,10 @@ instead of flattened scalar variables. from __future__ import annotations -import re from collections import Counter from typing import Any, Literal -# A dict-loop emits per-item field references of the form ``loopvar.``. -# ```` is derived from a source key, which is attacker-influenceable when -# JinjaTurtle is fed harvested config. The output-safety gate only accepts a -# reference whose every hop matches this identifier class, so a key must reduce -# to exactly this shape before it can be used as a loop-item field. Anything else -# (a key containing ``}}``, quotes, ``.``, ``__``, ...) must not be turned into a -# loop; the caller falls back to scalar generation, where every value goes through -# make_var_name() and all verbatim text is escaped. -_SAFE_LOOP_FIELD_RE = re.compile(r"(?!\w*__)[A-Za-z_][A-Za-z0-9_]*\Z") - - -def is_safe_loop_field_key(key: Any) -> bool: - """Return True if *key* is safe to emit as a ``loopvar.`` field access. - - The check mirrors the output-safety gate's identifier class (single - identifier, no double underscore). A key that does not match cannot be - expressed as a dotted loop-item reference without risking template - injection, so a loop candidate containing such a key is rejected upstream. - """ - - return bool(_SAFE_LOOP_FIELD_RE.match(str(key))) - - class LoopCandidate: """ Represents a detected loop opportunity in the config structure. @@ -289,19 +265,6 @@ class LoopAnalyzer: if not dicts: return "heterogeneous" - # Security: a dict-loop emits ``loopvar.`` field references. If any - # item key is not a plain identifier, it cannot be expressed safely as a - # dotted reference (a key such as ``a }}{{ x`` would break out of the - # placeholder and inject a live construct). Refuse the loop so the caller - # falls back to scalar generation, which escapes all verbatim text and - # routes every value through make_var_name(). - for d in dicts: - for k in d.keys(): - if k == "_key": - continue - if not is_safe_loop_field_key(k): - return "heterogeneous" - # Get key sets from each dict key_sets = [set(d.keys()) for d in dicts] diff --git a/src/jinjaturtle/multi.py b/src/jinjaturtle/multi.py index ce54e1c..05f92b3 100644 --- a/src/jinjaturtle/multi.py +++ b/src/jinjaturtle/multi.py @@ -22,26 +22,15 @@ Notes: from collections import Counter, defaultdict from copy import deepcopy -import os import configparser -import stat -import sys from dataclasses import dataclass from pathlib import Path 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, - ConfigParseError, -) +from .core import dump_yaml, flatten_config, make_var_name, parse_config from .handlers.xml import XmlHandler -from .safety import verify_jinja2_template_safe, verify_no_live_jinja_in_json_keys -from .escape import escape_jinja_literal SUPPORTED_SUFFIXES: dict[str, set[str]] = { @@ -53,23 +42,8 @@ SUPPORTED_SUFFIXES: dict[str, set[str]] = { } -def _lstat(path: Path) -> os.stat_result: - return path.lstat() - - def is_supported_file(path: Path) -> bool: - """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): + if not path.is_file(): return False suffix = path.suffix.lower() for exts in SUPPORTED_SUFFIXES.values(): @@ -79,16 +53,11 @@ def is_supported_file(path: Path) -> bool: def iter_supported_files(root: Path, recursive: bool) -> list[Path]: - try: - st = _lstat(root) - except FileNotFoundError: + if not root.exists(): raise FileNotFoundError(str(root)) - - if stat.S_ISLNK(st.st_mode): - raise ValueError(f"refusing to follow symlink: {root}") - if stat.S_ISREG(st.st_mode): + if root.is_file(): return [root] if is_supported_file(root) else [] - if not stat.S_ISDIR(st.st_mode): + if not root.is_dir(): return [] it = root.rglob("*") if recursive else root.glob("*") @@ -191,11 +160,6 @@ def _yaml_render_union( if isinstance(union_obj, dict): for key, val in union_obj.items(): key_path = path + (str(key),) - # The key text is copied verbatim into the template; escape it so an - # attacker-influenced key (e.g. ``{{ 7*7 }}``) cannot become live - # template code. ``key_path`` (used only to build sanitised var - # names) keeps the original key. - safe_key = escape_jinja_literal(str(key)) cond_var = ( defined_var_name(role_prefix, key_path) if key_path in optional_containers @@ -206,13 +170,13 @@ def _yaml_render_union( value = _yaml_scalar_placeholder(role_prefix, key_path, val) if cond_var: lines.append(f"{ind}{j2.if_defined(cond_var)}") - lines.append(f"{ind}{safe_key}: {value}") + lines.append(f"{ind}{key}: {value}") if cond_var: lines.append(f"{ind}{j2.endif()}") else: if cond_var: lines.append(f"{ind}{j2.if_defined(cond_var)}") - lines.append(f"{ind}{safe_key}:") + lines.append(f"{ind}{key}:") lines.extend( _yaml_render_union( role_prefix, @@ -250,7 +214,6 @@ def _yaml_render_union( first = True for k, v in item.items(): kp = item_path + (str(k),) - safe_k = escape_jinja_literal(str(k)) k_cond = ( defined_var_name(role_prefix, kp) if kp in optional_containers @@ -261,14 +224,14 @@ def _yaml_render_union( if first: if k_cond: lines.append(f"{ind}{j2.if_defined(k_cond)}") - lines.append(f"{ind}- {safe_k}: {value}") + lines.append(f"{ind}- {k}: {value}") if k_cond: lines.append(f"{ind}{j2.endif()}") first = False else: if k_cond: lines.append(f"{ind} {j2.if_defined(k_cond)}") - lines.append(f"{ind} {safe_k}: {value}") + lines.append(f"{ind} {k}: {value}") if k_cond: lines.append(f"{ind} {j2.endif()}") else: @@ -276,7 +239,7 @@ def _yaml_render_union( if first: if k_cond: lines.append(f"{ind}{j2.if_defined(k_cond)}") - lines.append(f"{ind}- {safe_k}:") + lines.append(f"{ind}- {k}:") lines.extend( _yaml_render_union( role_prefix, @@ -292,7 +255,7 @@ def _yaml_render_union( else: if k_cond: lines.append(f"{ind} {j2.if_defined(k_cond)}") - lines.append(f"{ind} {safe_k}:") + lines.append(f"{ind} {k}:") lines.extend( _yaml_render_union( role_prefix, @@ -338,7 +301,6 @@ def _toml_render_union( def emit_kv(path: tuple[str, ...], key: str, value: Any) -> None: var_name = make_var_name(role_prefix, path + (key,)) - safe_key = escape_jinja_literal(str(key)) cond = ( defined_var_name(role_prefix, path + (key,)) if (path + (key,)) in optional_containers @@ -347,11 +309,11 @@ def _toml_render_union( if cond: lines.append(f"{j2.if_defined(cond)}") if isinstance(value, str): - lines.append(f"{safe_key} = {j2.quoted_variable(var_name)}") + lines.append(f"{key} = {j2.quoted_variable(var_name)}") elif isinstance(value, bool): - lines.append(f"{safe_key} = {j2.lower(var_name)}") + lines.append(f"{key} = {j2.lower(var_name)}") else: - lines.append(f"{safe_key} = {j2.variable(var_name)}") + lines.append(f"{key} = {j2.variable(var_name)}") if cond: lines.append(j2.endif()) @@ -364,7 +326,7 @@ def _toml_render_union( ) if cond: lines.append(f"{j2.if_defined(cond)}") - lines.append(f"[{'.'.join(escape_jinja_literal(str(p)) for p in path)}]") + lines.append(f"[{'.'.join(path)}]") scalar_items = {k: v for k, v in obj.items() if not isinstance(v, dict)} nested_items = {k: v for k, v in obj.items() if isinstance(v, dict)} @@ -450,11 +412,10 @@ def _ini_render_union( ) if sec_cond: lines.append(f"{j2.if_defined(sec_cond)}") - lines.append(f"[{escape_jinja_literal(str(section))}]") + lines.append(f"[{section}]") for key, raw_val in union.items(section, raw=True): path = (section, key) var = make_var_name(role_prefix, path) - safe_key = escape_jinja_literal(str(key)) key_cond = ( defined_var_name(role_prefix, path) if path in optional_keys else None ) @@ -463,9 +424,9 @@ def _ini_render_union( if key_cond: lines.append(f"{j2.if_defined(key_cond)}") if quoted: - lines.append(f"{safe_key} = {j2.quoted_variable(var)}") + lines.append(f"{key} = {j2.quoted_variable(var)}") else: - lines.append(f"{safe_key} = {j2.variable(var)}") + lines.append(f"{key} = {j2.variable(var)}") if key_cond: lines.append(j2.endif()) lines.append("") @@ -638,13 +599,7 @@ def process_directory( # Parse and group by format grouped: dict[str, list[tuple[Path, Any]]] = defaultdict(list) for p in files: - try: - fmt, parsed = parse_config(p, None) - except ConfigParseError as exc: - # One malformed file should not abort processing of an entire - # directory. Skip it with a warning; the rest still generate. - print(f"jinjaturtle: skipping {p}: {exc}", file=sys.stderr) - continue + fmt, parsed = parse_config(p, None) if fmt not in FOLDER_SUPPORTED_FORMATS: # Directory mode only supports a subset of formats for now. continue @@ -814,12 +769,4 @@ def process_directory( defaults_doc[out.list_var] = out.items defaults_yaml = dump_yaml(defaults_doc, sort_keys=True) - # Defence in depth: folder-mode union templates are built by their own - # renderers (not core.generate_jinja2_template), so gate each one here too. - # 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/output_safety.py b/src/jinjaturtle/output_safety.py deleted file mode 100644 index eac40f9..0000000 --- a/src/jinjaturtle/output_safety.py +++ /dev/null @@ -1,191 +0,0 @@ -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, -refuse symlink parents, reject root-run output through untrusted parent -components, write through a 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.""" - - -# Keep a reference to the real euid getter. Tests can monkeypatch -# _effective_uid directly without changing process-wide os.geteuid behaviour. -_OS_GETEUID = getattr(os, "geteuid", None) - - -def _effective_uid() -> int | None: - if _OS_GETEUID is None: - return None - try: - return int(_OS_GETEUID()) - except OSError: - return None - - -def _absolute(path: Path) -> Path: - expanded = path.expanduser() - return expanded if expanded.is_absolute() else Path.cwd() / expanded - - -def _chmod_private(path: Path) -> None: - try: - os.chmod(path, 0o700) - except OSError: - # Best-effort; mkdir(mode=0o700) is already used for normal filesystems. - pass - - -def _assert_trusted_root_parent(path: Path, st: os.stat_result) -> None: - """Reject root-run output through attacker-controlled parent directories. - - A root-run JinjaTurtle process may write files that an administrator later - applies as configuration-management input. Parent directories controlled by - an unprivileged user are therefore not acceptable output anchors. Root-owned - sticky shared directories such as /tmp are allowed as a boundary, but any - existing child below them must be root-owned and not writable by group/other. - """ - - if _effective_uid() != 0: - return - if not stat.S_ISDIR(st.st_mode): - raise OutputPathError(f"output parent is not a directory: {path}") - if st.st_uid != 0: - raise OutputPathError( - f"output parent is not owned by root; refusing root-run output: {path}" - ) - writable_by_group_or_other = st.st_mode & (stat.S_IWGRP | stat.S_IWOTH) - sticky = st.st_mode & stat.S_ISVTX - if writable_by_group_or_other and not sticky: - raise OutputPathError( - "output parent is writable by group/other; " - f"refusing root-run output: {path}" - ) - - -def _assert_existing_directory_component(path: Path) -> None: - try: - st = path.lstat() - except OSError as exc: - raise OutputPathError(f"unable to inspect output parent: {path}") from exc - if stat.S_ISLNK(st.st_mode): - raise OutputPathError(f"refusing to use symlink parent: {path}") - if not stat.S_ISDIR(st.st_mode): - raise OutputPathError(f"output parent is not a directory: {path}") - _assert_trusted_root_parent(path, st) - - -def _mkdir_safe_dir_tree(path: Path) -> Path: - """Create/validate a directory tree one component at a time. - - pathlib.mkdir(parents=True) can traverse a symlink inserted after a parent - pre-check. Walking one component at a time keeps every existing component - checked before it is used, and newly-created components are immediately - re-inspected. - """ - - out = _absolute(path) - parts = out.parts - if not parts: - return out - - if out.is_absolute(): - cur = Path(parts[0]) - rest = parts[1:] - _assert_existing_directory_component(cur) - else: - # _absolute() currently always returns an absolute path, but keep this - # branch for clarity if that helper is ever relaxed. - cur = Path.cwd() - rest = parts - _assert_existing_directory_component(cur) - - for part in rest: - cur = cur / part - if os.path.lexists(cur): - _assert_existing_directory_component(cur) - continue - try: - os.mkdir(cur, 0o700) - except FileExistsError: - _assert_existing_directory_component(cur) - continue - _chmod_private(cur) - _assert_existing_directory_component(cur) - - return out - - -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 trusted real directory.""" - - _mkdir_safe_dir_tree(parent) - - -def ensure_safe_directory(path: Path) -> None: - """Create or validate a directory tree without accepting unsafe parents.""" - - _mkdir_safe_dir_tree(path) - - -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 is created/validated component by component. Existing - parent symlinks are refused for every user. When running as root, existing - parent components must also be root-owned and not writable by group/other, - except for sticky shared boundaries such as /tmp. Existing final-path - symlinks are refused rather than followed. - """ - - path = _absolute(path) - parent = _mkdir_safe_dir_tree(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(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_parent_components(path.parent) - _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/src/jinjaturtle/safety.py b/src/jinjaturtle/safety.py deleted file mode 100644 index 938f2ad..0000000 --- a/src/jinjaturtle/safety.py +++ /dev/null @@ -1,382 +0,0 @@ -from __future__ import annotations - -"""Output safety gate for generated templates (defence in depth). - -JinjaTurtle's first line of defence is per-handler escaping: every piece of -verbatim source text is meant to be wrapped/neutralised before it reaches the -template (see ``escape.py``). That model is correct but *fragile*: it relies on -every handler remembering to escape at every site, and on the escaper being -exactly right for every delimiter form. A single forgotten call site -- or a -new handler, or a missed delimiter variant -- silently reopens a -template-injection / SSTI path that can become remote code execution on the -configuration-management control node when the template is later rendered. - -This module adds a second, independent line of defence that does **not** depend -on getting every escape right. It inspects the *finished* template and proves a -single global property: - - Every *live* template construct in the output is one that JinjaTurtle itself - legitimately emits. Anything else can only have originated from - un-neutralised source text, so generation fails loudly instead of emitting a - dangerous template. - -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``) so it covers every current handler and -every future one automatically. - -Why this is robust against the escaper being wrong ---------------------------------------------------- -We tokenise with Jinja2's own lexer. The lexer emits a JinjaTurtle -``{% raw %} ... {% endraw %}`` wrapper as ``raw_begin`` / inert ``data`` / -``raw_end``: the wrapped literal text is *not* tokenised as live tags. So the -verifier only ever sees, as live constructs, the tags Jinja2 would actually -execute. If an escaped block was mis-wrapped such that a payload escapes the -raw wrapper (the historical ``{%+ endraw %}`` bug), that payload now appears as a -*live* token here and is rejected -- the gate catches the failure even though -the escaper produced it. -""" - -import re - -__all__ = [ - "TemplateSafetyError", - "verify_jinja2_template_safe", - "verify_erb_template_safe", - "verify_no_live_jinja_in_json_keys", -] - - -class TemplateSafetyError(Exception): - """Raised when a generated template contains a construct JinjaTurtle would - never emit, indicating that un-neutralised source text became live template - code. Generation must abort rather than emit the template.""" - - -# --------------------------------------------------------------------------- # -# Allowlist grammar for JinjaTurtle-emitted Jinja2. -# -# JinjaTurtle emits a deliberately tiny subset of Jinja2. Each pattern below -# describes the *full body* of a tag (the text between ``{%``/``%}`` or -# ``{{``/``}}``), already stripped of surrounding whitespace and of any ``-``/ -# ``+`` whitespace-control markers. Identifiers (variable names, loop vars, -# keys) are restricted to a conservative character class; crucially this class -# excludes characters needed for SSTI gadgets (quotes, parentheses, brackets, -# arithmetic/operator characters, ``%``, ``|`` except in the known filter forms, -# attribute access beyond a single dotted hop, etc.). -# --------------------------------------------------------------------------- # - -# A single identifier. Word characters only, and -- critically -- no -# double-underscore anywhere. ``__`` is the gateway to every classic Jinja2 -# SSTI gadget (``__class__``, ``__init__``, ``__globals__``, ``__builtins__``), -# and JinjaTurtle never emits a dunder, so forbidding ``__`` here removes the -# entire attribute-traversal escape class even if such a token reached output. -_NAME = r"(?!\w*__)[A-Za-z_][A-Za-z0-9_]*" - -# A dotted reference for loop-item field access. JinjaTurtle emits at most -# three hops (``loopvar``, ``loopvar.field``, ``loopvar.key.subkey``), so cap the -# depth rather than allow arbitrary chains. -_DOTTED = rf"{_NAME}(?:\.{_NAME}){{0,2}}" - -# Filters JinjaTurtle is known to emit inside ``{{ ... }}`` expressions. -_KNOWN_FILTERS = ( - r"lower", - r"to_json\(ensure_ascii=(?:True|False)\)", - r"to_json\(indent=\d+,\s*ensure_ascii=(?:True|False)\)", - r"tojson", -) -_FILTER_ALT = "|".join(_KNOWN_FILTERS) - -# Expression bodies allowed inside ``{{ ... }}``. -_EXPR_PATTERNS = tuple( - re.compile(p) - for p in ( - # Plain variable / dotted loop-field reference. - rf"^{_DOTTED}$", - # Filtered reference: ``name | filter`` (one known filter). - rf"^{_DOTTED}\s*\|\s*(?:{_FILTER_ALT})$", - # YAML-preserving boolean ternary emitted by j2.yaml_*_expression: - # 'true' if NAME else 'false' / "true" if NAME else "false" - rf"^(['\"])(?:true|false)\1\s+if\s+{_DOTTED}\s+else\s+(['\"])(?:true|false)\2$", - # YAML-preserving null ternary: - # 'null' if NAME is none else NAME - rf"^(['\"])null\1\s+if\s+{_DOTTED}\s+is\s+none\s+else\s+{_DOTTED}$", - ) -) - -# Statement bodies allowed inside ``{% ... %}``. -_STMT_PATTERNS = tuple( - re.compile(p) - for p in ( - rf"^for\s+{_NAME}\s+in\s+{_DOTTED}$", - r"^endfor$", - rf"^if\s+{_DOTTED}\s+is\s+defined$", - rf"^if\s+{_DOTTED}\s+is\s+none$", - r"^if\s+not\s+loop\.last$", - r"^else$", - r"^endif$", - # ``elif`` is emitted only for the same shapes as the if-conditions - # above; keep it conservative. - rf"^elif\s+{_DOTTED}\s+is\s+(?:defined|none)$", - ) -) - - -def _strip_ws_control(body: str) -> str: - """Remove a leading/trailing Jinja whitespace-control marker and spaces.""" - body = body.strip() - if body[:1] in "-+": - body = body[1:] - if body[-1:] in "-+": - body = body[:-1] - return body.strip() - - -# Built-in Jinja2/Ansible globals that JinjaTurtle never emits as a reference -# head. The allowlist grammar necessarily accepts any bare identifier (it cannot -# tell an injected global from a legitimate generated variable), so these names -# are rejected explicitly: their presence as the head of a live reference means -# source text leaked into a live construct. This is a targeted backstop and does -# not replace the grammar check. -_FORBIDDEN_REFERENCE_HEADS = frozenset( - { - "range", - "dict", - "lipsum", - "cycler", - "joiner", - "namespace", - "config", - "self", - "cycle", - "request", - "get_flashed_messages", - "url_for", - # Ansible global set - "lookup", - "q", - "query", - "now", - "omit", - "undef", - } -) - - -def _reference_head(body: str) -> str: - """Return the leading identifier (before any dot/filter) of an expression.""" - body = _strip_ws_control(body) - m = re.match(r"[A-Za-z_][A-Za-z0-9_]*", body) - return m.group(0) if m else "" - - -def _expr_is_allowed(body: str) -> bool: - body = _strip_ws_control(body) - if _reference_head(body) in _FORBIDDEN_REFERENCE_HEADS: - return False - return any(p.match(body) for p in _EXPR_PATTERNS) - - -def _stmt_is_allowed(body: str) -> bool: - body = _strip_ws_control(body) - # A ``for ... in `` whose collection is a forbidden global is - # rejected for the same reason as an expression head. - m = re.match(r"for\s+[A-Za-z_]\w*\s+in\s+([A-Za-z_][A-Za-z0-9_]*)", body) - if m and m.group(1) in _FORBIDDEN_REFERENCE_HEADS: - return False - return any(p.match(body) for p in _STMT_PATTERNS) - - -def verify_jinja2_template_safe(template_text: str) -> None: - """Validate that *template_text* contains only JinjaTurtle-emitted Jinja2. - - Raises :class:`TemplateSafetyError` on the first live construct that is not - in the allowlist grammar. Text inside JinjaTurtle's own ``{% raw %}`` - wrappers is treated as inert (the lexer does not tokenise it as tags), so - legitimately-escaped source content passes. - """ - # Import lazily so the dependency is only needed when generating Jinja2. - import jinja2 - - env = jinja2.Environment(autoescape=True) - - try: - tokens = list(env.lex(template_text)) - except jinja2.TemplateSyntaxError as exc: - # A syntax error means our own raw-wrapping did not fully contain the - # source text (e.g. an early ``endraw`` breakout left dangling tags). - # That is precisely a safety failure, not a benign parse hiccup. - raise TemplateSafetyError( - f"generated template does not lex as the JinjaTurtle subset: {exc}" - ) from exc - - # Walk the token stream. Jinja2 yields raw blocks as single - # ``raw_begin``/``raw_end`` tokens with inert ``data`` between them, so we - # never see wrapped literal text as live tags. We collect the raw inner - # text of each live ``{% %}`` / ``{{ }}`` construct (preserving original - # spacing) and check it against the allowlist. - mode: str | None = None # None, "block", or "variable" - body_parts: list[str] = [] - - def _flush(kind: str, lineno: int) -> None: - body = "".join(body_parts) - if kind == "variable": - ok = _expr_is_allowed(body) - else: - ok = _stmt_is_allowed(body) - if not ok: - shown = body.strip() - wrapped = ( - "{{ " + shown + " }}" if kind == "variable" else "{% " + shown + " %}" - ) - raise TemplateSafetyError( - "refusing to emit template: unexpected live " - f"{'expression' if kind == 'variable' else 'statement'} " - f"{wrapped!r} at line {lineno}. This construct is not one " - "JinjaTurtle emits, so it likely came from un-neutralised " - "source text (possible template injection)." - ) - - for lineno, tok_type, value in tokens: - if tok_type in ("variable_begin", "block_begin"): - mode = "variable" if tok_type == "variable_begin" else "block" - body_parts = [] - elif tok_type in ("variable_end", "block_end"): - if mode is not None: - _flush(mode, lineno) - mode = None - body_parts = [] - elif mode is not None: - # Preserve the original token text (including its own whitespace - # 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. - - -# --------------------------------------------------------------------------- # -# 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 -# 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 c0d5470..e4ac519 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -132,3 +132,57 @@ 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_config_parse_errors.py b/tests/test_config_parse_errors.py deleted file mode 100644 index 7f4c422..0000000 --- a/tests/test_config_parse_errors.py +++ /dev/null @@ -1,205 +0,0 @@ -"""Regression tests for malformed-input handling (``ConfigParseError``). - -Bug: every parse-layer handler (xml/json/toml/yaml/ini) used to let its -underlying parser's exception escape as an unhandled traceback when given a -malformed file. Pointing JinjaTurtle (or Enroll, which calls it as a library) -at harvested, attacker-influenceable config makes malformed input an entirely -expected condition, so it must fail cleanly instead of crashing. - -These tests pin down that: - - * ``parse_config`` raises the normalised ``ConfigParseError`` for malformed - XML/JSON/TOML/INI; - * defusedxml's *security* exceptions (XXE / ``EntitiesForbidden``) are NOT - swallowed by that normalisation -- they must still propagate so a caller can - distinguish "malformed" from "malicious"; - * an unrelated error (e.g. the TOML "tomllib missing" ``RuntimeError``) is not - captured by the normalisation either; - * the CLI exits non-zero with a clean message (no traceback) on malformed - input; and - * folder mode skips an unparseable file instead of aborting the whole run. -""" - -from __future__ import annotations - -import subprocess -import sys -from pathlib import Path - -import pytest - -from jinjaturtle.core import ConfigParseError, parse_config - - -def _run_cli(args: list[str]) -> subprocess.CompletedProcess[str]: - return subprocess.run( - [sys.executable, "-m", "jinjaturtle.cli", *args], - capture_output=True, - text=True, - ) - - -@pytest.mark.parametrize( - "fmt,filename,content", - [ - # Element name is not well-formed XML -> expat ParseError historically - # escaped as an uncaught traceback. - ("xml", "bad.xml", "<{{tag}}/>"), - ("xml", "bad2.xml", ""), - ("json", "bad.json", "{not valid json"), - ("toml", "bad.toml", "x = = ="), - # A bare key with no value is invalid INI for configparser. - ("ini", "bad.ini", "[s]\nthis line has no equals and no colon\n"), - ], -) -def test_parse_config_raises_configparseerror_on_malformed( - tmp_path: Path, fmt: str, filename: str, content: str -) -> None: - src = tmp_path / filename - src.write_text(content, encoding="utf-8") - with pytest.raises(ConfigParseError): - parse_config(src, fmt=fmt) - - -def test_parse_config_does_not_swallow_xxe_security_exception(tmp_path: Path) -> None: - """defusedxml's EntitiesForbidden must propagate, not become ConfigParseError. - - EntitiesForbidden subclasses ValueError, so a naive ``except ValueError`` / - ``except Exception`` in parse_config would mask an XXE attempt as a benign - "malformed file". The normalisation is deliberately scoped to exclude it. - """ - from defusedxml.common import EntitiesForbidden - - xxe = ( - '\n' - ' ]>\n' - "&xxe;\n" - ) - src = tmp_path / "xxe.xml" - src.write_text(xxe, encoding="utf-8") - with pytest.raises(EntitiesForbidden): - parse_config(src, fmt="xml") - - -def test_parse_config_does_not_swallow_unrelated_runtimeerror( - tmp_path: Path, monkeypatch -) -> None: - """A non-malformation error (tomllib missing) must propagate unchanged.""" - import jinjaturtle.handlers.toml as toml_module - - monkeypatch.setattr(toml_module, "tomllib", None) - src = tmp_path / "x.toml" - src.write_text('a = "b"\n', encoding="utf-8") - with pytest.raises(RuntimeError) as exc: - parse_config(src, fmt="toml") - assert "tomllib/tomli is required" in str(exc.value) - - -def test_cli_fails_cleanly_on_malformed_xml(tmp_path: Path) -> None: - """The CLI must exit non-zero with a clean message, not a traceback.""" - src = tmp_path / "bad.xml" - src.write_text("<{{tag}}/>", encoding="utf-8") - res = _run_cli([str(src), "-f", "xml", "-r", "demo"]) - assert res.returncode != 0 - # Clean, user-facing message -- not a Python traceback. - assert "could not parse" in res.stderr - assert "Traceback (most recent call last)" not in res.stderr - - -@pytest.mark.parametrize( - "content", - [ - # Classic XXE: external SYSTEM entity used to read a local file. - ( - '\n' - ']>\n' - "&xxe;\n" - ), - # Billion-laughs style internal entity expansion. - ( - '\n' - ']>\n' - "&lol2;\n" - ), - # External parameter entity (SSRF / out-of-band XXE vector). - ( - '\n' - ' %ext;]>\n' - "x\n" - ), - ], -) -def test_cli_refuses_xxe_cleanly(tmp_path: Path, content: str) -> None: - """defusedxml's security refusal must surface as a clean non-zero exit. - - ``EntitiesForbidden``/``DTDForbidden``/``ExternalReferenceForbidden`` subclass - ``DefusedXmlException`` (and ``ValueError``). ``parse_config`` deliberately - lets them propagate rather than folding them into ``ConfigParseError`` so a - blocked attack stays distinguishable from a benign malformed file. The CLI - must therefore catch ``DefusedXmlException`` itself and exit cleanly instead - of leaking a Python traceback. - """ - src = tmp_path / "xxe.xml" - src.write_text(content, encoding="utf-8") - res = _run_cli([str(src), "-f", "xml", "-r", "demo"]) - assert res.returncode == 2 - assert "refusing unsafe XML input" in res.stderr - assert "Traceback (most recent call last)" not in res.stderr - - -def test_folder_mode_skips_unparseable_file(tmp_path: Path) -> None: - """One malformed file should not abort processing of the whole directory.""" - from jinjaturtle.multi import process_directory - - good = tmp_path / "good.json" - good.write_text('{ "host": "localhost" }', encoding="utf-8") - bad = tmp_path / "bad.json" - bad.write_text("{not valid json", encoding="utf-8") - - defaults_yaml, outputs = process_directory(tmp_path, False, "demo") - # The good file still produced output despite the bad sibling: its content - # and source id appear, and a template was generated. - assert "localhost" in defaults_yaml - assert "good.json" in defaults_yaml - assert outputs - - -@pytest.mark.parametrize( - "content", - [ - "a: &a [*a]\n", # self-referential sequence alias - "root: &r\n child: *r\n", # self-referential mapping alias - ], -) -def test_recursive_yaml_is_rejected_cleanly(tmp_path: Path, content: str) -> None: - """A recursive YAML anchor must fail as a clean ConfigParseError. - - PyYAML's safe_load builds a self-referential object from a recursive anchor; - JinjaTurtle's downstream walks (flatten, timestamp-stringify, template gen) - would otherwise blow the stack with a RecursionError traceback. Harvested - config is attacker-influenceable, so this must fail closed. - """ - src = tmp_path / "cyclic.yaml" - src.write_text(content, encoding="utf-8") - with pytest.raises(ConfigParseError): - parse_config(src, "yaml") - - -def test_recursive_yaml_cli_exits_nonzero_without_traceback(tmp_path: Path) -> None: - src = tmp_path / "cyclic.yaml" - src.write_text("a: &a [*a]\n", encoding="utf-8") - res = _run_cli([str(src), "-f", "yaml", "-r", "role"]) - assert res.returncode != 0 - assert "Traceback" not in res.stderr - assert "RecursionError" not in res.stderr - - -def test_shared_noncyclic_yaml_aliases_still_work(tmp_path: Path) -> None: - """A shared (acyclic) anchor referenced multiple times is a DAG, not a - cycle, and must still parse and template normally.""" - src = tmp_path / "shared.yaml" - src.write_text("base: &b\n x: 1\na: *b\nc: *b\n", encoding="utf-8") - fmt, parsed = parse_config(src, "yaml") - assert fmt == "yaml" - assert parsed["a"] == {"x": 1} - assert parsed["c"] == {"x": 1} diff --git a/tests/test_core_utils.py b/tests/test_core_utils.py index fda6dd0..c8e41e1 100644 --- a/tests/test_core_utils.py +++ b/tests/test_core_utils.py @@ -191,43 +191,3 @@ def test_flatten_config_unsupported_format(): flatten_config("bogusfmt", parsed=None) assert "Unsupported format" in str(exc.value) - - -def test_make_var_name_collapses_adjacent_separators(): - """Regression: adjacent separators must not produce a forbidden ``__``. - - A source key such as ``log..level`` or ``cache--size`` previously sanitised - to a name containing a double underscore (``..._log__level``). The output - safety gate in safety.py rejects *any* ``__`` in a generated identifier - (it is the gateway to Jinja2 SSTI gadgets), so emitting one made JinjaTurtle - reject its own placeholder and abort generation on entirely benign config. - make_var_name now collapses runs of underscores to a single ``_``. - """ - for raw_key in ("log..level", "cache--size", "a...b", "x.-.y", "a..b--c"): - name = make_var_name("role", ("main", raw_key)) - assert "__" not in name, f"{raw_key!r} produced {name!r}" - # Still a valid Ansible/Jinja identifier. - assert name.replace("_", "").isalnum() or name == "role" - - # The double-underscore-free collapse is stable and predictable. - assert make_var_name("role", ("main", "log..level")) == "role_main_log_level" - assert make_var_name("role", ("main", "cache--size")) == "role_main_cache_size" - # A leading-digit-free prefix with its own repeated separators is collapsed too. - assert make_var_name("My__Role", ("k",)) == "my_role_k" - - -def test_make_var_name_collapsed_names_pass_output_safety_gate(): - """The names make_var_name emits must be accepted by the safety gate. - - This binds the two modules together: whatever identifier make_var_name - produces for a hostile-looking key must lex as the JinjaTurtle subset, so a - real template built from it is not refused. - """ - from jinjaturtle.safety import verify_jinja2_template_safe - - for raw_key in ("log..level", "cache--size", "a...b", "weird..key..name"): - var = make_var_name("demo", ("section", raw_key)) - # Build the kind of expression a handler would emit for this variable. - template = "{{ " + var + " }}" - # Must not raise TemplateSafetyError. - verify_jinja2_template_safe(template) diff --git a/tests/test_injection_security.py b/tests/test_injection_security.py index 4afb7a5..75edc70 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 Ansible later -renders the template. +config content becomes live template code that executes when Salt/Ansible/Puppet +later renders the template. These tests render the *generated* template the way a downstream tool would and assert that an injected payload never executes. A tripwire object is exposed @@ -14,6 +14,7 @@ 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 @@ -24,26 +25,13 @@ import yaml as pyyaml from jinjaturtle.escape import ( escape_jinja_literal, + escape_erb_literal, + escape_literal, ) 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.""" @@ -97,7 +85,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 = _safe_load_defaults(dfl.read_text()) or {} + defaults = pyyaml.safe_load(dfl.read_text()) or {} return tpl.read_text(), defaults @@ -167,86 +155,6 @@ 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 = [ @@ -257,18 +165,6 @@ SSTI_PAYLOADS = [ "text {% endraw %} breakout {{ evil }}", "nested {% endraw %} spacing {{ evil }}", "{%- endraw -%}{{ evil }}", - # Whitespace-control markers: Jinja2 accepts "-", "+" or none adjacent to a - # tag's delimiters, and every variant closes a {% raw %} block. The "+" - # forms in particular were a raw-wrapper breakout vector (the defang regex - # historically only matched "-"), so all combinations must be neutralised. - "{%+ endraw %}{{ evil }}", - "{% endraw +%}{{ evil }}", - "{%+ endraw +%}{{ evil }}", - "{%- endraw +%}{{ evil }}", - "{%+ endraw -%}{{ evil }}", - # Full breakout attempt: close raw early, inject live code, re-open raw to - # swallow our trailing {% endraw %} so the template would otherwise compile. - "{%+ endraw %}{{ evil }}{%+ raw %}", "mixed {{ a }} and {% b %} and {# c #}", "}}{{ orphan delimiters %}{%", ] @@ -301,27 +197,40 @@ def test_escape_jinja_literal_actually_blocks_execution(): @pytest.mark.parametrize( - "endraw", + "payload", [ - "{% endraw %}", - "{% endraw -%}", - "{% endraw +%}", - "{%- endraw %}", - "{%- endraw -%}", - "{%- endraw +%}", - "{%+ endraw %}", - "{%+ endraw -%}", - "{%+ endraw +%}", + "<%= system('id') %>", + "<% require 'open3' %>", + "text <%= 1+1 %> more", + "-%> orphan <%", ], ) -def test_endraw_whitespace_control_cannot_break_out(endraw): - """Every whitespace-control form of endraw closes a {% raw %} block in - Jinja2, so each must be defanged. A payload that closes raw early, injects - a live tripwire call, then re-opens raw to balance the wrapper must still - render inertly back to its original characters.""" +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) - payload = f"{endraw}{{{{ boom.run('x') }}}}{{%+ raw %}}" - escaped = escape_jinja_literal(payload) rendered = env.from_string(escaped).render(boom=_Boom()) assert TRIP not in rendered - assert rendered == payload diff --git a/tests/test_output_safety_gate.py b/tests/test_output_safety_gate.py deleted file mode 100644 index 055ceb3..0000000 --- a/tests/test_output_safety_gate.py +++ /dev/null @@ -1,233 +0,0 @@ -"""Regression tests for the output safety gate (``jinjaturtle.safety``). - -The gate is JinjaTurtle's second, independent line of defence against template -injection. Where the per-handler escaper neutralises verbatim source text, the -gate inspects the *finished* template and refuses to emit it if any live -construct is not one JinjaTurtle itself produces. These tests cover: - - * the gate's allow/deny grammar (unit level); - * the two concrete injection findings that motivated it -- JSON object keys - and folder-mode union keys copied into templates unescaped; - * end-to-end CLI fail-closed behaviour (non-zero exit, no file written). -""" - -from __future__ import annotations - -import json -import subprocess -import sys -from pathlib import Path - -import pytest - -from jinjaturtle import core -from jinjaturtle.multi import process_directory -from jinjaturtle.safety import ( - TemplateSafetyError, - verify_jinja2_template_safe, - verify_erb_template_safe, -) - - -# --------------------------------------------------------------------------- # -# Unit: the allow/deny grammar. -# --------------------------------------------------------------------------- # - -LEGIT_TEMPLATES = [ - "{{ demo_memory_limit }}", - "{{ demo_x | to_json(ensure_ascii=False) }}", - "{{ demo_a | to_json(indent=2, ensure_ascii=False) }}", - "{{ demo_name | lower }}", - "{{ 'true' if demo_flag else 'false' }}", - '{{ "true" if demo_flag else "false" }}', - "{{ 'null' if demo_v is none else demo_v }}", - "{% for server in demo_servers %}{{ server.name }}{% endfor %}", - "{{ server.config.port }}", - "{% if demo_x is defined %}{{ demo_x }}{% endif %}", - "{% if demo_x is none %}x{% endif %}", - "{% if not loop.last %},{% endif %}", - "plain text with no tags", - "{% raw %}# literal {{ not_code }} {% if x %}{% endraw %}", - "{% raw %}{{ 7*7 }}{% endraw %}: value", # escaped key (folder-mode fix) -] - - -@pytest.mark.parametrize("template", LEGIT_TEMPLATES) -def test_gate_allows_jinjaturtle_constructs(template): - # Must not raise. - verify_jinja2_template_safe(template) - - -MALICIOUS_TEMPLATES = [ - "{{ 7*7 }}", - "{{ cycler.__init__.__globals__ }}", - "{{ cycler.__init__.__globals__.os.popen('id').read() }}", - "{{ salt['cmd.run']('id') }}", - "{{ self.__init__ }}", - "{% for x in ().__class__.__base__.__subclasses__() %}{{ x }}{% endfor %}", - "{{ config.items() }}", - '{{ request["application"] }}', - "{% set x = 1 %}", - "{{ lipsum.__globals__ }}", - "{{ a.__class__.__mro__ }}", - "{{ ''.join(['a','b']) }}", - "{% include 'x' %}", - "{% import 'x' as y %}", -] - - -@pytest.mark.parametrize("template", MALICIOUS_TEMPLATES) -def test_gate_blocks_injection(template): - with pytest.raises(TemplateSafetyError): - verify_jinja2_template_safe(template) - - -def test_gate_blocks_double_underscore_anywhere(): - # The no-dunder rule is the backbone of blocking attribute-traversal SSTI. - with pytest.raises(TemplateSafetyError): - verify_jinja2_template_safe("{{ demo__x }}") - with pytest.raises(TemplateSafetyError): - verify_jinja2_template_safe("{{ a.b__c }}") - - -def test_gate_rejects_unlexable_breakout_as_safety_error(): - # A raw-wrapper breakout that leaves dangling tags must surface as a - # TemplateSafetyError, not a raw Jinja2 syntax error. - broken = "{% raw %}{%+ endraw %}{{ 7*7 }}" # unbalanced on purpose - with pytest.raises(TemplateSafetyError): - verify_jinja2_template_safe(broken) - - -# --------------------------------------------------------------------------- # -# Finding 1: JSON object keys copied verbatim into the template. -# --------------------------------------------------------------------------- # - -JSON_KEY_PAYLOADS = [ - '{ "{{ 7*7 }}": "v" }', - '{ "{{cycler.__init__.__globals__}}": 1 }', - "{ \"ok\": { \"{{ salt['cmd.run']('id') }}\": 2 } }", -] - - -@pytest.mark.parametrize("src", JSON_KEY_PAYLOADS) -def test_json_key_injection_is_blocked(src): - parsed = json.loads(src) - with pytest.raises(TemplateSafetyError): - core.generate_jinja2_template("json", parsed, "demo", original_text=src) - - -def test_json_benign_template_still_generates(): - src = '{ "name": "app", "port": 8080 }' - parsed = json.loads(src) - out = core.generate_jinja2_template("json", parsed, "demo", original_text=src) - assert "demo_name" in out - assert "demo_port" in out - - -# --------------------------------------------------------------------------- # -# Finding 2: folder-mode union renderers copied keys verbatim. -# --------------------------------------------------------------------------- # - - -def _write(tmp: Path, name: str, text: str) -> None: - (tmp / name).write_text(text, encoding="utf-8") - - -def test_folder_yaml_key_injection_is_neutralised(tmp_path): - # The malicious key must be escaped (root-cause fix), so the union template - # generates successfully AND passes the gate, rendering the key inertly. - _write(tmp_path, "a.yaml", '"{{ 7*7 }}": value\nnormalkey: ok\n') - _write(tmp_path, "b.yaml", "normalkey: ok2\n") - _defaults, outputs = process_directory(tmp_path, False, "demo") - template = "\n".join(o.template for o in outputs) - # Escaped, not live: - assert "{% raw %}{{ 7*7 }}{% endraw %}" in template - # And the gate (run inside process_directory) did not reject it. - - -def test_folder_ini_section_and_key_injection_neutralised(tmp_path): - _write( - tmp_path, - "a.ini", - "[{{ 7*7 }}]\n{{ evil }} = x\n", - ) - _write(tmp_path, "b.ini", "[normal]\nk = y\n") - _defaults, outputs = process_directory(tmp_path, False, "demo") - template = "\n".join(o.template for o in outputs) - # Section header and key are escaped, not live. - assert "{{ 7*7 }}" not in _strip_raw_blocks(template) - assert "{{ evil }}" not in _strip_raw_blocks(template) - - -def test_folder_toml_key_injection_neutralised(tmp_path): - _write(tmp_path, "a.toml", '"{{ 7*7 }}" = "v"\nok = "y"\n') - _write(tmp_path, "b.toml", 'ok = "z"\n') - _defaults, outputs = process_directory(tmp_path, False, "demo") - template = "\n".join(o.template for o in outputs) - assert "{{ 7*7 }}" not in _strip_raw_blocks(template) - - -def _strip_raw_blocks(text: str) -> str: - """Remove the *contents* of raw blocks so we can assert no LIVE payload - remains outside them.""" - import re - - return re.sub( - r"{%[-+]?\s*raw\s*[-+]?%}.*?{%[-+]?\s*endraw\s*[-+]?%}", - "", - text, - flags=re.S, - ) - - -# --------------------------------------------------------------------------- # -# 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. -# --------------------------------------------------------------------------- # - - -def _run_cli(args, env_extra=None): - import os - - env = {**os.environ, "PYTHONPATH": "src"} - if env_extra: - env.update(env_extra) - return subprocess.run( - [sys.executable, "-m", "jinjaturtle.cli", *args], - capture_output=True, - text=True, - env=env, - ) - - -def test_cli_fails_closed_on_injection(tmp_path): - bad = tmp_path / "evil.json" - bad.write_text('{ "{{ 7*7 }}": "v" }', encoding="utf-8") - out = tmp_path / "out.j2" - result = _run_cli([str(bad), "-r", "demo", "-f", "json", "-t", str(out)]) - assert result.returncode == 2 - assert "refusing to generate unsafe template" in result.stderr - # No template file may be written when the gate refuses. - assert not out.exists() - - -def test_cli_succeeds_on_benign_input(tmp_path): - good = tmp_path / "ok.json" - good.write_text('{ "name": "app" }', encoding="utf-8") - out = tmp_path / "out.j2" - result = _run_cli([str(good), "-r", "demo", "-f", "json", "-t", str(out)]) - assert result.returncode == 0 - assert out.exists() diff --git a/tests/test_security_hardening.py b/tests/test_security_hardening.py deleted file mode 100644 index b5bb855..0000000 --- a/tests/test_security_hardening.py +++ /dev/null @@ -1,136 +0,0 @@ -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" - - -def test_cli_refuses_root_output_through_group_writable_parent( - tmp_path: Path, monkeypatch -): - from jinjaturtle import output_safety - - unsafe_dir = tmp_path / "unsafe" - unsafe_dir.mkdir() - unsafe_dir.chmod(0o777) - monkeypatch.setattr(output_safety, "_effective_uid", lambda: 0) - - with pytest.raises(OutputPathError): - write_text_safely(unsafe_dir / "out.yml", "data\n") - - assert not (unsafe_dir / "out.yml").exists() diff --git a/tests/test_yaml_handler.py b/tests/test_yaml_handler.py index 543c759..a217e8a 100644 --- a/tests/test_yaml_handler.py +++ b/tests/test_yaml_handler.py @@ -10,6 +10,7 @@ from jinjaturtle.core import ( analyze_loops, flatten_config, generate_ansible_yaml, + generate_erb_template, generate_jinja2_template, ) from jinjaturtle.handlers.yaml import YamlHandler @@ -177,15 +178,17 @@ 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 in cases: + for label, text, rendered_expected, erb_expected in cases: path = tmp_path / f"{label}.yml" path.write_text(text, encoding="utf-8") @@ -202,6 +205,16 @@ 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