mig5/jinjaturtle
1
0
Fork 0
Code Issues Pull requests Projects Releases 16 Packages Wiki Activity Actions

Byebye
Some checks failed
CI / test (push) Successful in 48s
Details
CI / test (debian, docker.io/library/debian:13, python3) (push) Successful in 1m21s
Details
Lint / test (push) Failing after 38s
Details

Browse source
This commit is contained in:
Miguel Jacq 2026-06-24 20:29:20 +10:00
parent 094c4d2274
commit f63cb92b35
Signed by: mig5
GPG key ID: 03906B4110AAD3B8
18 changed files with 593 additions and 886 deletions
Download patch file Download diff file Expand all files Collapse all files

178
README.md
View file

@ -1,5 +1,11 @@
# JinjaTurtle
## ABANDONED
Project is abandoned. To many potential security issues. Please uninstall it. Sorry for wasting your time.
____
<div align="center">
<img src="https://git.mig5.net/mig5/jinjaturtle/raw/branch/main/jinjaturtle.svg" alt="JinjaTurtle logo" width="240" />
</div>
@ -7,15 +13,12 @@
JinjaTurtle is a command-line tool that helps turn existing native
configuration files into reusable configuration-management templates.
By default it generates:
It generates:
- a **Jinja2** template; and
- an **Ansible defaults YAML** file containing the variables used by that
template.
It can also generate **ERB** templates and **Puppet Hiera-style YAML** data for
Puppet workflows.
JinjaTurtle does not try to replace configuration-management tools. Its job is
to speed up the boring first pass: take a real config file, discover the values
inside it, replace those values with variables, and write the corresponding
@ -26,8 +29,6 @@ variable data beside the template.
JinjaTurtle examines a source config file and keeps the original structure as
much as possible.
For the default Jinja2/Ansible mode:
1. The config file is parsed.
2. Variable names are generated from the config keys and paths.
3. Those variable names are prefixed with `--role-name`, which should usually
@ -37,16 +38,6 @@ For the default Jinja2/Ansible mode:
5. An Ansible defaults YAML file is generated with those variables and the
original values.
For ERB/Puppet mode:
1. The same parse/flatten/loop analysis is used.
2. An ERB template is generated with Puppet-style instance variables such as
`<%= @memory_limit %>`.
3. The variables file is written as Puppet Hiera-style data, such as
`php::memory_limit: 256M`.
4. If `--puppet-class` is supplied, that class name is used as the Hiera
namespace while `--role-name` remains the local variable prefix.
By default, the generated variable data and template are printed to stdout. Use
`--defaults-output` and `--template-output` to write them to files.
@ -80,78 +71,6 @@ and defaults data like:
php_memory_limit: 256M
```
## ERB / Puppet example
Use `--template-engine erb` when you want Puppet ERB output:
```shell
jinjaturtle php.ini \
--role-name php \
--template-engine erb \
--defaults-output data/common.yaml \
--template-output templates/php.ini.erb
```
Given the same source value:
```ini
memory_limit = 256M
```
JinjaTurtle will produce an ERB template value like:
```erb
memory_limit = <%= @memory_limit %>
```
and Hiera-style data like:
```yaml
php::memory_limit: 256M
```
The `--defaults-output` option name is retained for CLI compatibility, but in
ERB mode the file is intended to be Puppet Hiera data rather than Ansible role
defaults.
JinjaTurtle does **not** generate Puppet classes or `file` resources. A Puppet
module, should still declare the class parameters and call the template, for
example with Puppet's `template()` function.
## Using `--puppet-class`
Most direct usage can simply use the same value for the role name and Puppet
class name:
```shell
jinjaturtle php.ini --role-name php --template-engine erb
```
This creates Hiera keys such as:
```yaml
php::memory_limit: 256M
```
and local ERB variables such as:
```erb
<%= @memory_limit %>
```
For generated systems, it can be useful to make `--role-name` more specific
while keeping the Hiera keys under the real Puppet class. For example:
```shell
jinjaturtle php.ini \
--role-name php_etc_php_ini \
--puppet-class php \
--template-engine erb
```
In that case the variable prefix can stay file-specific, while the Hiera data
is still written under `php::...`.
## What sort of config files can it handle?
JinjaTurtle supports common structured and semi-structured config formats:
@ -176,15 +95,15 @@ homogeneous enough. If it is not confident, it falls back to flattened scalar
variables.
Some very complex files will still need manual cleanup. The goal is to speed up
conversion into Jinja2 or ERB templates, not to guarantee a perfect final module
without review.
conversion into Jinja2 templates, not to guarantee a perfect final role without
review.
## JSON, quoting, and type preservation
JinjaTurtle tries to preserve rendered config types.
For JSON, it uses JSON-aware expressions rather than plain string substitution.
This avoids generating invalid JSON such as:
For JSON, it uses JSON-aware Jinja2 expressions rather than plain string
substitution. This avoids generating invalid JSON such as:
```json
{"enabled": True}
@ -196,18 +115,6 @@ when the correct rendered JSON should be:
{"enabled": true}
```
In Jinja2 mode this uses Ansible-style JSON filters. In ERB mode it emits Ruby
JSON generation where required, for example:
```erb
<% require 'json' -%>
{
"enabled": <%= JSON.generate(@enabled) %>
}
```
That is expected for JSON ERB templates.
## Can I convert multiple files at once?
Yes. Pass a directory instead of a single file and JinjaTurtle will convert the
@ -287,8 +194,6 @@ poetry install
usage: jinjaturtle [-h] [-r ROLE_NAME] [--recursive]
[-f {ini,json,toml,yaml,xml,postfix,systemd,ssh}]
[-d DEFAULTS_OUTPUT] [-t TEMPLATE_OUTPUT]
[--template-engine {jinja2,erb}]
[--puppet-class PUPPET_CLASS]
config
Convert a config file into an Ansible defaults file and Jinja2 template.
@ -301,25 +206,18 @@ positional arguments:
options:
-h, --help show this help message and exit
-r, --role-name ROLE_NAME
Role name / variable prefix. In Jinja2 mode this is
usually the Ansible role name. In ERB mode it is used
as the local variable prefix. Defaults to jinjaturtle.
Ansible role name, used as variable prefix. Defaults
to jinjaturtle.
--recursive When CONFIG is a folder, recurse into subfolders.
-f, --format {ini,json,toml,yaml,xml,postfix,systemd,ssh}
Force config format instead of auto-detecting from
filename.
-d, --defaults-output DEFAULTS_OUTPUT
Path to write the generated variable YAML. If omitted,
it is printed to stdout.
Path to write defaults/main.yml. If omitted, defaults
YAML is printed to stdout.
-t, --template-output TEMPLATE_OUTPUT
Path to write the generated config template. If omitted,
it is printed to stdout.
--template-engine {jinja2,erb}
Template syntax to generate. Defaults to jinja2. Use
erb for Puppet templates.
--puppet-class PUPPET_CLASS
Puppet class / Hiera namespace to use with
--template-engine erb. Defaults to --role-name.
Path to write the generated config template. If
omitted, template is printed to stdout.
```
## Additional supported formats
@ -345,43 +243,37 @@ code.
Two guarantees matter:
1. **Values are data, never code.** Every config *value* is replaced with a
`{{ variable }}` placeholder in the template, and the original value is stored
separately in the defaults/Hiera data. When the template is later rendered,
the placeholder prints the value as a literal string; Jinja2 does not
recursively render the *contents* of a variable, so a payload sitting inside
a value (for example `motd = {{ salt['cmd.run']('id') }}`) is inert.
`{{ variable }}` placeholder in the template, and the original value is
stored separately in the defaults data. String values in generated defaults
are tagged with Ansible's `!unsafe` tag by default, so a payload sitting
inside a value (for example `motd = {{ lookup('pipe', 'id') }}`) remains
inert even if downstream playbooks accidentally place that value in another
templating context.
2. **Verbatim text is neutralised.** To preserve formatting, JinjaTurtle copies
comments, blank lines, headers and any unrecognised lines from the source
into the template. Any template metacharacters in that copied text
(`{{ }}`, `{% %}`, `{# #}` for Jinja2; `<% %>` for ERB) are escaped so they
render as the literal characters the author wrote, rather than executing.
into the template. Any Jinja2 metacharacters in that copied text (`{{ }}`,
`{% %}`, `{# #}`) are escaped so they render as the literal characters the
author wrote, rather than executing.
### Consumer responsibilities
The value guarantee above relies on the downstream renderer being single-pass,
which is the normal case:
Treat generated defaults as untrusted input. JinjaTurtle emits extracted
string values with Ansible's `!unsafe` tag, but you should preserve that tag if
you merge the generated data into larger defaults files or transform it with
other tooling.
- **Ansible**: rendering a template with `template:`/`ansible.builtin.template`
is single-pass. For defence in depth, treat the generated defaults as
untrusted input — Ansible already does not re-template variable *contents* by
default. If you build your own var structures from this data and pass them
through additional templating, mark untrusted values with the `!unsafe` tag so
they are never re-evaluated.
- **Salt**: use the generated file as a `file.managed` template
(`template: jinja`). Do **not** place JinjaTurtle output where Salt would
render it a second time as part of SLS/pillar rendering, which is a separate
Jinja pass and would re-evaluate any embedded expressions.
- **Puppet/ERB**: render with the standard `template()`/`epp()` flow.
In short: render JinjaTurtle output exactly once. Do not strip `!unsafe` tags or
feed the data back through another templating pass.
In short: render JinjaTurtle output exactly once. Do not feed it back through
another templating pass.
Directory mode deliberately ignores symlinks and will not traverse outside the
input directory. This prevents a malicious folder tree from smuggling in
configuration files through links to unrelated filesystem locations.
**IMPORTANT**: Always review both the original config files, then the resulting
templates generated by JinjaTurtle, before integrating them into your config
management system!
## Found a bug, have a suggestion?
You can e-mail me; see `pyproject.toml` for details. You can also contact me on

1
debian/changelog vendored
View file

@ -1,5 +1,6 @@
jinjaturtle (0.5.7) unstable; urgency=medium
* Remove ERB/Puppet output support and keep Jinja2-only generation.
* More hardening measures
-- Miguel Jacq <mig@mig5.net> Wed, 24 Jun 2026 16:13:00 +1000

1
rpm/jinjaturtle.spec
View file

@ -43,6 +43,7 @@ Convert config files into Ansible defaults and Jinja2 templates.
%changelog
* Wed Jun 24 2026 Miguel Jacq <mig@mig5.net> - %{version}-%{release}
- Remove ERB/Puppet output support and keep Jinja2-only generation.
- More hardening
* Tue Jun 23 2026 Miguel Jacq <mig@mig5.net> - %{version}-%{release}
- Try to prevent what could lead to execution of embedded jinja in original files when converting

105
src/jinjaturtle/cli.py
View file

@ -1,6 +1,7 @@
from __future__ import annotations
import argparse
import os
import sys
from defusedxml import defuse_stdlib
from pathlib import Path
@ -12,12 +13,29 @@ from .core import (
flatten_config,
generate_ansible_yaml,
generate_jinja2_template,
generate_puppet_hiera_yaml,
generate_erb_template,
)
from .multi import process_directory
from .safety import TemplateSafetyError, verify_erb_template_safe
from .safety import TemplateSafetyError
def _write_text_no_symlink(path: Path, text: str) -> None:
"""Write a user-requested output file without following a final symlink."""
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
fd = os.open(path, flags, 0o600)
except OSError as exc:
raise OSError(f"refusing or unable to write output file {path}: {exc}") from exc
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(text)
def _ensure_output_dir(path: Path) -> None:
if path.exists() and path.is_symlink():
raise OSError(f"refusing to use symlink as output directory: {path}")
path.mkdir(parents=True, exist_ok=True)
def _build_arg_parser() -> argparse.ArgumentParser:
@ -59,20 +77,6 @@ def _build_arg_parser() -> argparse.ArgumentParser:
"--template-output",
help="Path to write the generated config template. If omitted, template is printed to stdout.",
)
ap.add_argument(
"--template-engine",
choices=[j2.NAME, "erb"],
default=j2.NAME,
help="Template syntax to generate (default: jinja2). Use erb for Puppet templates.",
)
ap.add_argument(
"--puppet-class",
help=(
"Puppet class/Hiera namespace to use with --template-engine erb. "
"Defaults to --role-name. This lets tools use a file-specific "
"variable prefix while writing Hiera keys under the real Puppet class."
),
)
return ap
@ -88,6 +92,9 @@ def _main(argv: list[str] | None = None) -> int:
f"jinjaturtle: refusing to generate unsafe template: {exc}", file=sys.stderr
)
return 2
except (OSError, TypeError, ValueError) as exc:
print(f"jinjaturtle: error: {exc}", file=sys.stderr)
return 1
def _run(argv: list[str] | None = None) -> int:
@ -105,51 +112,36 @@ def _run(argv: list[str] | None = None) -> int:
# Write defaults
if args.defaults_output:
Path(args.defaults_output).write_text(defaults_yaml, encoding="utf-8")
_write_text_no_symlink(Path(args.defaults_output), defaults_yaml)
else:
print("# defaults/main.yml")
print(defaults_yaml, end="")
# Optionally translate folder-mode templates to ERB. Folder mode keeps
# the existing data shape; single-file mode below is the preferred
# Puppet path because it can produce class-parameter Hiera keys.
if args.template_engine == "erb":
from .erb import translate_jinja2_to_erb
for o in outputs:
o.template = translate_jinja2_to_erb(
o.template,
role_prefix=args.role_name,
puppet_class=args.puppet_class or args.role_name,
)
verify_erb_template_safe(o.template)
template_ext = "erb" if args.template_engine == "erb" else j2.TEMPLATE_EXTENSION
# Write templates
if args.template_output:
out_path = Path(args.template_output)
if len(outputs) == 1 and not out_path.is_dir():
out_path.write_text(outputs[0].template, encoding="utf-8")
_write_text_no_symlink(out_path, outputs[0].template)
else:
out_path.mkdir(parents=True, exist_ok=True)
_ensure_output_dir(out_path)
for o in outputs:
(out_path / f"config.{o.fmt}.{template_ext}").write_text(
o.template, encoding="utf-8"
_write_text_no_symlink(
out_path / f"config.{o.fmt}.{j2.TEMPLATE_EXTENSION}",
o.template,
)
else:
for o in outputs:
name = (
f"config.{template_ext}"
f"config.{j2.TEMPLATE_EXTENSION}"
if len(outputs) == 1
else f"config.{o.fmt}.{template_ext}"
else f"config.{o.fmt}.{j2.TEMPLATE_EXTENSION}"
)
print(f"# {name}")
print(o.template, end="")
return 0
# Single-file mode (existing behaviour)
# Single-file mode
config_text = config_path.read_text(encoding="utf-8")
# Parse the config
@ -161,27 +153,8 @@ def _run(argv: list[str] | None = None) -> int:
# Flatten config (excluding loop paths if loops are detected)
flat_items = flatten_config(fmt, parsed, loop_candidates)
if args.template_engine == "erb":
ansible_yaml = generate_puppet_hiera_yaml(
args.role_name,
flat_items,
loop_candidates,
puppet_class=args.puppet_class or args.role_name,
)
template_str = generate_erb_template(
fmt,
parsed,
args.role_name,
original_text=config_text,
loop_candidates=loop_candidates,
flat_items=flat_items,
puppet_class=args.puppet_class or args.role_name,
)
else:
# Generate defaults YAML (with loop collections if detected)
ansible_yaml = generate_ansible_yaml(
args.role_name, flat_items, loop_candidates
)
ansible_yaml = generate_ansible_yaml(args.role_name, flat_items, loop_candidates)
# Generate template (with loops if detected)
template_str = generate_jinja2_template(
@ -193,19 +166,15 @@ def _run(argv: list[str] | None = None) -> int:
)
if args.defaults_output:
Path(args.defaults_output).write_text(ansible_yaml, encoding="utf-8")
_write_text_no_symlink(Path(args.defaults_output), ansible_yaml)
else:
print("# defaults/main.yml")
print(ansible_yaml, end="")
if args.template_output:
Path(args.template_output).write_text(template_str, encoding="utf-8")
_write_text_no_symlink(Path(args.template_output), template_str)
else:
print(
"# config.erb"
if args.template_engine == "erb"
else f"# config.{j2.TEMPLATE_EXTENSION}"
)
print(f"# config.{j2.TEMPLATE_EXTENSION}")
print(template_str, end="")
return 0

241
src/jinjaturtle/core.py
View file

@ -7,12 +7,39 @@ import datetime
import re
import yaml
class AnsibleUnsafeString(str):
"""Marker for strings emitted with Ansible's !unsafe YAML tag."""
pass
class VariableNameCollisionError(ValueError):
"""Raised when distinct config paths collapse to the same variable name."""
pass
class ConfigTooDeeplyNestedError(ValueError):
"""Raised when a parsed config nests deeper than ``MAX_CONFIG_DEPTH``.
Pathologically nested input (thousands of levels) would otherwise drive the
recursive walkers in this module past Python's stack limit and raise an
uncaught ``RecursionError``. We bound the depth here and fail with a clear,
catchable error instead. The limit is well above any plausible real config
while staying comfortably under the interpreter's default recursion ceiling.
"""
pass
# Maximum container-nesting depth accepted from a parsed config. Real-world
# configuration files are nowhere near this deep; the cap exists purely to turn
# a hostile/degenerate input into a clean error rather than a stack overflow.
MAX_CONFIG_DEPTH = 200
from .loop_analyzer import LoopAnalyzer, LoopCandidate
from .erb import puppet_class_name, puppet_local_var_name, translate_jinja2_to_erb
from .safety import (
verify_erb_template_safe,
verify_jinja2_template_safe,
)
from .safety import verify_jinja2_template_safe
from .handlers import (
BaseHandler,
IniHandler,
@ -53,7 +80,38 @@ def _quoted_str_representer(dumper: yaml.SafeDumper, data: QuotedString):
return dumper.represent_scalar("tag:yaml.org,2002:str", str(data), style='"')
def _ansible_unsafe_str_representer(dumper: yaml.SafeDumper, data: AnsibleUnsafeString):
return dumper.represent_scalar("!unsafe", str(data))
def _ansible_unsafe_str_constructor(
loader: yaml.SafeLoader, node: yaml.nodes.ScalarNode
) -> str:
return loader.construct_scalar(node)
def _mark_ansible_unsafe_values(data: Any) -> Any:
"""Recursively tag string *values* as Ansible !unsafe.
Mapping keys remain ordinary strings because they are variable names in the
generated defaults document, not attacker-controlled config values.
"""
if isinstance(data, AnsibleUnsafeString):
return data
if isinstance(data, str):
return AnsibleUnsafeString(data)
if isinstance(data, dict):
return {k: _mark_ansible_unsafe_values(v) for k, v in data.items()}
if isinstance(data, list):
return [_mark_ansible_unsafe_values(v) for v in data]
if isinstance(data, tuple):
return tuple(_mark_ansible_unsafe_values(v) for v in data)
return data
_TurtleDumper.add_representer(QuotedString, _quoted_str_representer)
_TurtleDumper.add_representer(AnsibleUnsafeString, _ansible_unsafe_str_representer)
yaml.SafeLoader.add_constructor("!unsafe", _ansible_unsafe_str_constructor)
# Use our fallback for any unknown object types
_TurtleDumper.add_representer(None, _fallback_str_representer)
@ -83,10 +141,12 @@ _HANDLERS["ssh"] = _SSH_HANDLER
def dump_yaml(data: Any, *, sort_keys: bool = True) -> str:
"""Dump YAML using JinjaTurtle's dumper settings.
This is used by both the single-file and multi-file code paths.
String values extracted from input configuration are emitted as Ansible
``!unsafe`` values so a downstream playbook cannot accidentally re-template
malicious Jinja syntax stored in defaults. Mapping keys are left untouched.
"""
return yaml.dump(
data,
_mark_ansible_unsafe_values(data),
Dumper=_TurtleDumper,
sort_keys=sort_keys,
default_flow_style=False,
@ -263,6 +323,31 @@ def detect_format(path: Path, explicit: str | None = None) -> str:
return "ini"
def _check_xml_depth(elem: Any, _depth: int = 0) -> None:
"""Enforce :data:`MAX_CONFIG_DEPTH` on an ElementTree element.
XML parses to an ``Element`` rather than dict/list containers, so it is not
covered by the depth bound in :func:`_stringify_timestamps`. Without this
check a pathologically nested document would drive the XML flattener and the
loop analyzer's ``_analyze_xml`` / ``_xml_elem_to_dict`` walkers into an
uncaught ``RecursionError``. We walk the tree iteratively (an explicit
stack, so the check itself cannot overflow) and fail loudly at the limit.
"""
# (element, depth) pairs; iterative traversal avoids its own recursion.
stack = [(elem, 0)]
while stack:
node, depth = stack.pop()
if depth > MAX_CONFIG_DEPTH:
raise ConfigTooDeeplyNestedError(
f"XML config nests deeper than the supported maximum of "
f"{MAX_CONFIG_DEPTH} levels; refusing to process it"
)
for child in list(node):
# Skip comment/PI nodes whose tag is not a str.
if isinstance(getattr(child, "tag", None), str):
stack.append((child, depth + 1))
def parse_config(path: Path, fmt: str | None = None) -> tuple[str, Any]:
"""
Parse config file into a Python object.
@ -271,7 +356,22 @@ def parse_config(path: Path, fmt: str | None = None) -> tuple[str, Any]:
handler = _HANDLERS.get(fmt)
if handler is None:
raise ValueError(f"Unsupported config format: {fmt}")
try:
parsed = handler.parse(path)
except RecursionError as exc:
# Some underlying parsers (e.g. PyYAML, json) recurse while *building*
# the object, so they can exhaust the stack before our own depth guards
# below ever run. Convert that into the same clean, catchable error.
raise ConfigTooDeeplyNestedError(
f"config in {path} is nested too deeply for the parser to handle; "
"refusing to process it"
) from exc
# Bound nesting depth before any recursive walker runs, turning degenerate
# deeply-nested input into a clean, catchable error instead of a stack
# overflow. dict/list configs are bounded inside _stringify_timestamps;
# XML (an Element tree) is bounded explicitly here.
if hasattr(parsed, "tag"): # ElementTree Element
_check_xml_depth(parsed)
# Make sure datetime objects are treated as strings (TOML, YAML)
parsed = _stringify_timestamps(parsed)
@ -341,6 +441,31 @@ def _path_starts_with(path: tuple[str, ...], prefix: tuple[str, ...]) -> bool:
return path[: len(prefix)] == prefix
def check_var_name_collisions(
role_prefix: str, items: Iterable[tuple[tuple[str, ...], Any]]
) -> None:
"""Fail if distinct config paths collapse to the same variable name."""
seen: dict[str, tuple[str, ...]] = {}
for path, _value in items:
path_tuple = tuple(str(p) for p in path)
var_name = make_var_name(role_prefix, path_tuple)
previous = seen.get(var_name)
if previous is not None and previous != path_tuple:
raise VariableNameCollisionError(
"distinct config paths collapse to the same variable name "
f"{var_name!r}: {previous!r} and {path_tuple!r}"
)
seen[var_name] = path_tuple
def _flatten_loop_defaults(
loop_candidates: list[LoopCandidate] | None,
) -> list[tuple[tuple[str, ...], Any]]:
if not loop_candidates:
return []
return [(candidate.path, candidate.items) for candidate in loop_candidates]
def generate_ansible_yaml(
role_prefix: str,
flat_items: list[tuple[tuple[str, ...], Any]],
@ -349,6 +474,10 @@ def generate_ansible_yaml(
"""
Create Ansible YAML for defaults/main.yml.
"""
check_var_name_collisions(
role_prefix, flat_items + _flatten_loop_defaults(loop_candidates)
)
defaults: dict[str, Any] = {}
# Add scalar variables
@ -399,87 +528,7 @@ def generate_jinja2_template(
return template
def _template_variable_names(
role_prefix: str,
flat_items: list[tuple[tuple[str, ...], Any]],
loop_candidates: list[LoopCandidate] | None = None,
) -> set[str]:
names = {make_var_name(role_prefix, path) for path, _value in flat_items}
if loop_candidates:
for candidate in loop_candidates:
names.add(make_var_name(role_prefix, candidate.path))
return names
def generate_puppet_hiera_yaml(
role_prefix: str,
flat_items: list[tuple[tuple[str, ...], Any]],
loop_candidates: list[LoopCandidate] | None = None,
*,
puppet_class: str | None = None,
) -> str:
"""Create Puppet Hiera data suitable for Automatic Parameter Lookup.
``role_prefix`` remains the source variable prefix used by JinjaTurtle while
``puppet_class`` is the Puppet class/Hiera namespace. In the normal case
they are the same, so ``php_memory_limit`` becomes ``php::memory_limit``.
"""
klass = puppet_class_name(puppet_class or role_prefix)
data: dict[str, Any] = {}
for path, value in flat_items:
generated = make_var_name(role_prefix, path)
local = puppet_local_var_name(role_prefix, generated, puppet_class=klass)
data[f"{klass}::{local}"] = value
if loop_candidates:
for candidate in loop_candidates:
generated = make_var_name(role_prefix, candidate.path)
local = puppet_local_var_name(role_prefix, generated, puppet_class=klass)
data[f"{klass}::{local}"] = candidate.items
return dump_yaml(data, sort_keys=True)
def generate_erb_template(
fmt: str,
parsed: Any,
role_prefix: str,
*,
original_text: str | None = None,
loop_candidates: list[LoopCandidate] | None = None,
flat_items: list[tuple[tuple[str, ...], Any]] | None = None,
puppet_class: str | None = None,
) -> str:
"""Generate a Puppet ERB template from JinjaTurtle's renderer-neutral data.
The first implementation intentionally translates the Jinja2 subset emitted
by JinjaTurtle's existing format handlers. This keeps parsing, formatting
preservation, and loop detection identical between Jinja2 and ERB output
while still producing Puppet-native ``@parameter`` references.
"""
jinja_template = generate_jinja2_template(
fmt,
parsed,
role_prefix,
original_text=original_text,
loop_candidates=loop_candidates,
)
names = _template_variable_names(role_prefix, flat_items or [], loop_candidates)
erb_template = translate_jinja2_to_erb(
jinja_template,
role_prefix=role_prefix,
puppet_class=puppet_class or role_prefix,
variable_names=names,
)
# Defence in depth: no live Jinja2 delimiter may survive translation.
verify_erb_template_safe(erb_template)
return erb_template
def _stringify_timestamps(obj: Any) -> Any:
def _stringify_timestamps(obj: Any, _depth: int = 0) -> Any:
"""
Recursively walk a parsed config and turn any datetime/date/time objects
into plain strings in ISO-8601 form.
@ -489,11 +538,23 @@ def _stringify_timestamps(obj: Any) -> Any:
This commonly occurs otherwise with TOML and YAML files, which sees
Python automatically convert those sorts of strings into datetime objects.
Because this is the first walk applied to every parsed config (in
:func:`parse_config`), it also enforces :data:`MAX_CONFIG_DEPTH`. Bounding
nesting here means every downstream recursive walker (flatteners, loop
analyzer) is fed input of safe depth, so a degenerate deeply-nested file
fails with :class:`ConfigTooDeeplyNestedError` instead of an uncaught
``RecursionError`` somewhere later in the pipeline.
"""
if _depth > MAX_CONFIG_DEPTH:
raise ConfigTooDeeplyNestedError(
f"config nests deeper than the supported maximum of "
f"{MAX_CONFIG_DEPTH} levels; refusing to process it"
)
if isinstance(obj, dict):
return {k: _stringify_timestamps(v) for k, v in obj.items()}
return {k: _stringify_timestamps(v, _depth + 1) for k, v in obj.items()}
if isinstance(obj, list):
return [_stringify_timestamps(v) for v in obj]
return [_stringify_timestamps(v, _depth + 1) for v in obj]
# TOML & YAML both use the standard datetime types
if isinstance(obj, datetime.datetime):

243
src/jinjaturtle/erb.py
View file

@ -1,243 +0,0 @@
from __future__ import annotations
import re
from .escape import escape_erb_literal
def _safe_name(raw: str, *, fallback: str = "var") -> str:
text = re.sub(r"[^A-Za-z0-9_]+", "_", str(raw or fallback)).strip("_").lower()
text = re.sub(r"_+", "_", text)
if not text:
text = fallback
if not re.match(r"^[a-z_]", text):
text = f"{fallback}_{text}"
return text
def puppet_class_name(raw: str) -> str:
"""Return a conservative Puppet class/Hiera namespace name."""
text = _safe_name(raw, fallback="jinjaturtle")
if not re.match(r"^[a-z]", text):
text = f"jinjaturtle_{text}"
return text
def _role_prefix_name(raw: str) -> str:
return _safe_name(raw, fallback="jinjaturtle")
def puppet_local_var_name(
role_prefix: str,
jinja_var_name: str,
*,
puppet_class: str | None = None,
) -> str:
"""Map a generated JinjaTurtle variable to a Puppet class parameter.
For the common case where ``--role-name php`` also means class ``php``, a
generated Jinja variable such as ``php_memory_limit`` becomes Puppet local
parameter ``memory_limit`` and Hiera key ``php::memory_limit``.
When ``puppet_class`` differs from ``role_prefix`` we keep the full
generated variable name as the local parameter and only use ``puppet_class``
as the Hiera namespace.
"""
var_name = _safe_name(jinja_var_name, fallback="value")
prefix = _role_prefix_name(role_prefix)
klass = puppet_class_name(puppet_class or role_prefix)
if klass == prefix and var_name.startswith(prefix + "_"):
stripped = var_name[len(prefix) + 1 :]
return stripped or var_name
return var_name
class ErbTranslator:
"""Translate the Jinja2 subset emitted by JinjaTurtle into Puppet ERB."""
_TOKEN_RE = re.compile(r"({{.*?}}|{%.*?%})", re.S)
def __init__(
self,
*,
role_prefix: str,
puppet_class: str | None = None,
variable_names: set[str] | None = None,
) -> None:
self.role_prefix = role_prefix
self.puppet_class = puppet_class or role_prefix
self.variable_names = set(variable_names or set())
self.loop_stack: list[tuple[str, str, str]] = []
self.needs_json = False
# Matches a JinjaTurtle ``{% raw %} ... {% endraw %}`` block (non-greedy).
# JinjaTurtle emits raw blocks only to carry verbatim, security-escaped
# source text (comments and unrecognised lines), so the *contents* must be
# treated as literal output, never translated as Jinja tokens.
_RAW_BLOCK_RE = re.compile(
r"{%[-+]?\s*raw\s*[-+]?%}(.*?){%[-+]?\s*endraw\s*[-+]?%}", re.S
)
def translate(self, template_text: str) -> str:
# Split out raw blocks first. Their inner text is literal and must be
# carried through as literal ERB (with ERB delimiters re-escaped), rather
# than tokenised -- otherwise an escaped Jinja payload inside a comment
# would be "re-animated" into live ERB during translation.
segments = self._RAW_BLOCK_RE.split(template_text)
out: list[str] = []
# re.split with one capture group yields: [text, raw_inner, text, ...].
for idx, segment in enumerate(segments):
if idx % 2 == 1:
# Captured raw-block contents: emit as literal ERB text.
out.append(escape_erb_literal(segment))
else:
out.append(self._translate_tokens(segment))
rendered = "".join(out)
if self.needs_json and "require 'json'" not in rendered:
rendered = "<% require 'json' -%>\n" + rendered
return rendered
def _translate_tokens(self, template_text: str) -> str:
parts = self._TOKEN_RE.split(template_text)
out: list[str] = []
for token in parts:
if not token:
continue
if token.startswith("{{") and token.endswith("}}"):
expr = token[2:-2].strip()
out.append(f"<%= {self.expr_to_ruby(expr)} %>")
continue
if token.startswith("{%") and token.endswith("%}"):
stmt = token[2:-2].strip()
out.append(self.statement_to_erb(stmt))
continue
out.append(token)
return "".join(out)
def local_var(self, name: str) -> str:
return puppet_local_var_name(
self.role_prefix, name, puppet_class=self.puppet_class
)
def ruby_value(self, expr: str) -> str:
expr = expr.strip()
if expr in {"true", "True"}:
return "true"
if expr in {"false", "False"}:
return "false"
if expr in {"none", "None", "null"}:
return "nil"
if re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", expr):
if any(expr == loop_var for loop_var, _idx, _coll in self.loop_stack):
return expr
return f"@{self.local_var(expr)}"
m = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)\.([A-Za-z_][A-Za-z0-9_]*)$", expr)
if m:
base, key = m.groups()
if any(base == loop_var for loop_var, _idx, _coll in self.loop_stack):
return f"{base}[{key!r}]"
return f"@{self.local_var(base)}[{key!r}]"
return expr
def expr_to_ruby(self, expr: str) -> str:
expr = expr.strip()
# JinjaTurtle emits these YAML-preserving ternaries for booleans/nulls.
m = re.match(
r"^(['\"])(true|false)\1\s+if\s+([A-Za-z_][A-Za-z0-9_\.]*)\s+else\s+(['\"])(true|false)\4$",
expr,
)
if m:
truthy = m.group(2)
cond = self.ruby_value(m.group(3))
falsy = m.group(5)
return f"{cond} ? {truthy!r} : {falsy!r}"
m = re.match(
r"^(['\"])(null)\1\s+if\s+([A-Za-z_][A-Za-z0-9_\.]*)\s+is\s+none\s+else\s+([A-Za-z_][A-Za-z0-9_\.]*)$",
expr,
)
if m:
value = self.ruby_value(m.group(3))
fallback = self.ruby_value(m.group(4))
return f"{value}.nil? ? 'null' : {fallback}"
if "|" in expr:
base, *filters = [part.strip() for part in expr.split("|")]
ruby = self.ruby_value(base)
for filt in filters:
if filt.startswith("lower"):
ruby = f"{ruby}.to_s.downcase"
elif filt.startswith("to_json") or filt.startswith("tojson"):
self.needs_json = True
if "indent" in filt:
ruby = f"JSON.pretty_generate({ruby})"
else:
ruby = f"JSON.generate({ruby})"
return ruby
return self.ruby_value(expr)
def statement_to_erb(self, stmt: str) -> str:
if stmt.endswith(("-", "+")):
stmt = stmt[:-1].rstrip()
if stmt.startswith("for "):
m = re.match(
r"^for\s+([A-Za-z_][A-Za-z0-9_]*)\s+in\s+([A-Za-z_][A-Za-z0-9_]*)$",
stmt,
)
if m:
loop_var, collection = m.groups()
idx_var = f"__jt_idx_{len(self.loop_stack)}"
collection_ruby = self.ruby_value(collection)
self.loop_stack.append((loop_var, idx_var, collection_ruby))
return f"<% {collection_ruby}.each_with_index do |{loop_var}, {idx_var}| -%>"
if stmt == "endfor":
if self.loop_stack:
self.loop_stack.pop()
return "<% end %>"
if stmt.startswith("if "):
cond = stmt[3:].strip()
if cond == "not loop.last" and self.loop_stack:
_loop_var, idx_var, collection_ruby = self.loop_stack[-1]
return f"<% if {idx_var} < ({collection_ruby}.length - 1) -%>"
m = re.match(r"^([A-Za-z_][A-Za-z0-9_\.]*)\s+is\s+defined$", cond)
if m:
return f"<% unless {self.ruby_value(m.group(1))}.nil? -%>"
m = re.match(r"^([A-Za-z_][A-Za-z0-9_\.]*)\s+is\s+none$", cond)
if m:
return f"<% if {self.ruby_value(m.group(1))}.nil? -%>"
return f"<% if {self.expr_to_ruby(cond)} -%>"
if stmt == "else":
return "<% else -%>"
if stmt.startswith("elif "):
return f"<% elsif {self.expr_to_ruby(stmt[5:].strip())} -%>"
if stmt == "endif":
return "<% end -%>"
# Preserve unknown Jinja statements visibly as an ERB comment so the
# generated template does not contain invalid Jinja syntax.
return f"<%# Unsupported JinjaTurtle statement: {stmt} %>"
def translate_jinja2_to_erb(
template_text: str,
*,
role_prefix: str,
puppet_class: str | None = None,
variable_names: set[str] | None = None,
) -> str:
return ErbTranslator(
role_prefix=role_prefix,
puppet_class=puppet_class,
variable_names=variable_names,
).translate(template_text)

102
src/jinjaturtle/escape.py
View file

@ -1,43 +1,23 @@
from __future__ import annotations
"""Neutralise template metacharacters in text copied verbatim from source files.
"""Neutralise Jinja2 metacharacters in text copied verbatim from source files.
JinjaTurtle preserves formatting by copying parts of the *original* config file
straight into the generated template: comments, blank lines, section headers,
and any line it does not recognise as ``key = value``. Config *values* are
always replaced with ``{{ var }}`` placeholders and parked in the defaults data,
so a payload inside a value is inert. Verbatim text is different: if the source
contains ``{{ ... }}``, ``{% ... %}`` or ``{# ... #}`` (Jinja2), or ``<%= %>`` /
``<% %>`` (ERB), that text becomes *live template code* in the output and is
executed when Salt/Ansible/Puppet later renders the template.
always replaced with ``{{ var }}`` placeholders and stored in the defaults data,
so a payload inside a value remains data. Verbatim text is different: if the
source contains ``{{ ... }}``, ``{% ... %}`` or ``{# ... #}``, that text would
become live Jinja2 template code unless it is neutralised.
Because JinjaTurtle is frequently fed harvested, attacker-influenceable config
(hostnames, banners, GECOS-derived comments, "Managed by" notes), this is a
template-injection / SSTI vector that can lead to remote code execution on the
configuration-management control node.
The functions here render those metacharacters as literal text so they survive a
later template render as the characters the source author actually wrote, rather
The helpers here render those metacharacters as literal text so they survive a
later Jinja2 render as the characters the source author actually wrote, rather
than as executable template syntax.
Design notes:
* We only ever escape text that originates from the *source file*. We never
pass JinjaTurtle's own generated placeholders (``{{ role_var }}``) through
these helpers, so the placeholders keep working.
* Jinja2 literal text is wrapped in a single ``{% raw %} ... {% endraw %}``
block. ``raw`` disables *all* tag interpretation inside it -- expressions,
statements and ``{# #}`` comments alike -- so one wrap neutralises every
Jinja construct. The only way to break out of a raw block is a literal
``{% endraw %}`` in the source, so we defang the token ``endraw`` (in any
internal spacing) before wrapping.
* The ERB translator (``erb.py``) is raw-aware: it copies the *contents* of a
JinjaTurtle raw block through as literal ERB text and re-escapes any ERB
delimiters found there. That keeps a Jinja-escaped comment inert after the
Jinja2 -> ERB translation step, instead of the payload being "re-animated"
as ERB.
* ``escape_erb_literal`` exists for completeness / direct ERB emission: it
rewrites each ERB delimiter into an ERB expression that prints the delimiter
characters literally.
"""
import re
@ -46,10 +26,6 @@ import re
# configures custom delimiters.
_JINJA_MARKERS = ("{{", "}}", "{%", "%}", "{#", "#}")
# ERB delimiters. Longer markers first so "<%=" matches before "<%".
_ERB_OPEN_MARKERS = ("<%=", "<%-", "<%#", "<%")
_ERB_CLOSE_MARKERS = ("-%>", "%>")
# Matches a Jinja2 endraw tag in any internal spacing and with any
# whitespace-control marker on either side. Jinja2 accepts "-", "+", or no
# marker adjacent to the "%}"/"{%" of a block tag (e.g. "{%endraw%}",
@ -59,38 +35,23 @@ _ERB_CLOSE_MARKERS = ("-%>", "%>")
# {% raw %} wrapper. [-+]? appears on both sides accordingly.
_ENDRAW_RE = re.compile(r"{%[-+]?\s*endraw\s*[-+]?%}")
# Sentinel inserted between "end" and "raw" to break the endraw keyword without
# changing the visible characters. We use a Jinja comment-free approach: insert
# the two halves across a raw boundary so the literal text still reads "endraw"
# to a human but is never a valid tag. See escape_jinja_literal for usage.
def contains_jinja_markup(text: str) -> bool:
"""Return True if *text* contains any Jinja2 delimiter."""
return any(m in text for m in _JINJA_MARKERS)
def contains_erb_markup(text: str) -> bool:
"""Return True if *text* contains any ERB delimiter."""
return any(m in text for m in (*_ERB_OPEN_MARKERS, *_ERB_CLOSE_MARKERS))
def _defang_endraw(text: str) -> str:
"""Rewrite any literal ``{% endraw %}`` so it cannot close our raw wrapper.
We turn each endraw tag into ``{% endraw %}{{ '{% endraw %}' }}{% raw %}``...
no -- that would re-introduce live tags. Instead we keep everything literal:
we break the keyword by emitting the tag's text in two raw segments split
inside the word ``endraw``. The result, when later rendered, reproduces the
exact original characters ``{% endraw %}`` while never being a parseable tag.
We keep everything literal: the replacement breaks the keyword by emitting
the tag's text in two raw segments split inside the word ``endraw``. The
result, when later rendered, reproduces the exact original characters while
never being a parseable raw-closing tag inside our wrapper.
"""
def _replace(match: re.Match[str]) -> str:
tag = match.group(0)
# Split the keyword "endraw" as "end" + "raw"; close and reopen the raw
# block between them. Each half is plain text inside a raw block, so the
# reconstructed output is byte-identical to the original tag, but at no
# point does the token "{% endraw %}" exist contiguously to close raw.
idx = tag.lower().index("endraw")
head = tag[: idx + 3] # up to and including "end"
tail = tag[idx + 3 :] # "raw...%}"
@ -111,43 +72,6 @@ def escape_jinja_literal(text: str) -> str:
return "{% raw %}" + _defang_endraw(text) + "{% endraw %}"
def escape_erb_literal(text: str) -> str:
"""Make *text* render as literal characters under a later ERB render.
ERB has no ``raw`` block, so each opening/closing delimiter is rewritten as
an ERB expression that prints the delimiter literally. Text with no ERB
metacharacters is returned unchanged.
"""
if not text or not contains_erb_markup(text):
return text
result: list[str] = []
i = 0
n = len(text)
while i < n:
matched = None
for marker in (*_ERB_CLOSE_MARKERS, *_ERB_OPEN_MARKERS):
if text.startswith(marker, i):
matched = marker
break
if matched is not None:
escaped = matched.replace("\\", "\\\\").replace('"', '\\"')
result.append('<%= "' + escaped + '" %>')
i += len(matched)
else:
result.append(text[i])
i += 1
return "".join(result)
def escape_literal(text: str, *, engine: str = "jinja2") -> str:
"""Escape verbatim *text* for the target template *engine*.
``engine`` is ``"jinja2"`` (default) or ``"erb"``. Unknown engines fall back
to Jinja2 escaping. In JinjaTurtle's pipeline ERB output is produced by
translating Jinja2 output, and the translator is raw-aware, so handlers can
always Jinja-escape and rely on the translator to keep the literal inert.
"""
if engine == "erb":
return escape_erb_literal(text)
def escape_literal(text: str) -> str:
"""Escape verbatim *text* for Jinja2 output."""
return escape_jinja_literal(text)

20
src/jinjaturtle/handlers/base.py
View file

@ -2,6 +2,10 @@ from __future__ import annotations
from pathlib import Path
from typing import Any, Iterable
import re
_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
class BaseHandler:
@ -59,6 +63,13 @@ class BaseHandler:
Sanitises parts to lowercase [a-z0-9_] and strips extras.
"""
role_prefix = role_prefix.strip().lower()
if not _VAR_NAME_RE.fullmatch(role_prefix) or "__" in role_prefix:
raise ValueError(
"role name must be a safe Ansible/Jinja variable prefix: "
"letters, digits and underscores only; must start with a letter "
"or underscore; double underscores are not allowed"
)
clean_parts: list[str] = []
for part in path:
@ -70,10 +81,13 @@ class BaseHandler:
cleaned_chars.append(c.lower())
else:
cleaned_chars.append("_")
cleaned_part = "".join(cleaned_chars).strip("_")
cleaned_part = re.sub(r"_+", "_", "".join(cleaned_chars)).strip("_")
if cleaned_part:
clean_parts.append(cleaned_part)
if clean_parts:
return role_prefix + "_" + "_".join(clean_parts)
return role_prefix
var_name = role_prefix + "_" + "_".join(clean_parts)
else:
var_name = role_prefix
return var_name

70
src/jinjaturtle/handlers/json.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
import json
import re
from uuid import uuid4
from pathlib import Path
from typing import Any
@ -211,24 +212,25 @@ class JsonHandler(DictLikeHandler):
Uses | to_json filter to preserve types (numbers, booleans, null).
"""
marker_prefix = f"\ue000JT{uuid4().hex}"
scalar_markers: dict[str, str] = {}
def _walk(obj: Any, path: tuple[str, ...] = ()) -> Any:
if isinstance(obj, dict):
return {k: _walk(v, path + (str(k),)) for k, v in obj.items()}
if isinstance(obj, list):
return [_walk(v, path + (str(i),)) for i, v in enumerate(obj)]
# scalar - use marker that will be replaced with to_json
var_name = self.make_var_name(role_prefix, path)
return f"__SCALAR__{var_name}__"
marker = f"{marker_prefix}SCALAR:{len(scalar_markers)}\ue001"
scalar_markers[marker] = var_name
return marker
templated = _walk(data)
json_str = json.dumps(templated, indent=2, ensure_ascii=False)
# Replace scalar markers with Jinja expressions using to_json filter
# This preserves types (numbers stay numbers, booleans stay booleans)
json_str = re.sub(
r'"__SCALAR__([a-zA-Z_][a-zA-Z0-9_]*)__"',
lambda m: self._json_value_expr(m.group(1)),
json_str,
for marker, var_name in scalar_markers.items():
json_str = json_str.replace(
json.dumps(marker, ensure_ascii=False), self._json_value_expr(var_name)
)
return json_str + "\n"
@ -245,66 +247,67 @@ class JsonHandler(DictLikeHandler):
Generate a JSON Jinja2 template with for loops where appropriate.
"""
marker_prefix = f"\ue000JT{uuid4().hex}"
scalar_markers: dict[str, str] = {}
loop_markers: dict[tuple[str, str], str] = {}
def _walk(obj: Any, current_path: tuple[str, ...] = ()) -> Any:
# Check if this path is a loop candidate
if current_path in loop_paths:
# Find the matching candidate
candidate = next(c for c in loop_candidates if c.path == current_path)
collection_var = self.make_var_name(role_prefix, candidate.path)
item_var = candidate.loop_var
if candidate.item_schema == "scalar":
# Simple list of scalars - use special marker that we'll replace
return f"__LOOP_SCALAR__{collection_var}__{item_var}__"
elif candidate.item_schema in ("simple_dict", "nested"):
# List of dicts - use special marker
return f"__LOOP_DICT__{collection_var}__{item_var}__"
marker = f"{marker_prefix}LOOP_SCALAR:{len(loop_markers)}\ue001"
loop_markers[("scalar", collection_var)] = marker
return marker
if candidate.item_schema in ("simple_dict", "nested"):
marker = f"{marker_prefix}LOOP_DICT:{len(loop_markers)}\ue001"
loop_markers[("dict", collection_var)] = marker
return marker
if isinstance(obj, dict):
return {k: _walk(v, current_path + (str(k),)) for k, v in obj.items()}
if isinstance(obj, list):
# Check if this list is a loop candidate
if current_path in loop_paths:
# Already handled above
return _walk(obj, current_path)
return [_walk(v, current_path + (str(i),)) for i, v in enumerate(obj)]
# scalar - use marker to preserve type
var_name = self.make_var_name(role_prefix, current_path)
return f"__SCALAR__{var_name}__"
marker = f"{marker_prefix}SCALAR:{len(scalar_markers)}\ue001"
scalar_markers[marker] = var_name
return marker
templated = _walk(data, path)
# Convert to JSON string
json_str = json.dumps(templated, indent=2, ensure_ascii=False)
# Replace scalar markers with Jinja expressions using to_json filter
json_str = re.sub(
r'"__SCALAR__([a-zA-Z_][a-zA-Z0-9_]*)__"',
lambda m: self._json_value_expr(m.group(1)),
json_str,
for marker, var_name in scalar_markers.items():
json_str = json_str.replace(
json.dumps(marker, ensure_ascii=False), self._json_value_expr(var_name)
)
# Post-process to replace loop markers with actual Jinja loops (indent-aware)
for candidate in loop_candidates:
collection_var = self.make_var_name(role_prefix, candidate.path)
item_var = candidate.loop_var
if candidate.item_schema == "scalar":
marker = f'"__LOOP_SCALAR__{collection_var}__{item_var}__"'
marker = loop_markers.get(("scalar", collection_var))
if marker is None:
continue
json_str = self._replace_marker_with_pretty_loop(
json_str,
marker,
json.dumps(marker, ensure_ascii=False),
lambda base, cv=collection_var, iv=item_var, c=candidate: self._generate_json_scalar_loop(
cv, iv, c, base
),
)
elif candidate.item_schema in ("simple_dict", "nested"):
marker = f'"__LOOP_DICT__{collection_var}__{item_var}__"'
marker = loop_markers.get(("dict", collection_var))
if marker is None:
continue
json_str = self._replace_marker_with_pretty_loop(
json_str,
marker,
json.dumps(marker, ensure_ascii=False),
lambda base, cv=collection_var, iv=item_var, c=candidate: self._generate_json_dict_loop(
cv, iv, c, base
),
@ -364,8 +367,9 @@ class JsonHandler(DictLikeHandler):
] # first line has no indent; we prepend `inner` when emitting
for i, key in enumerate(keys):
comma = "," if i < len(keys) - 1 else ""
safe_key = json.dumps(str(key), ensure_ascii=False)
dict_lines.append(
f'{field}"{key}": ' f"{j2.to_json(f'{item_var}.{key}')}{comma}"
f"{field}{safe_key}: " f"{j2.to_json(f'{item_var}.{key}')}{comma}"
)
# Comma between *items* goes after the closing brace.
dict_lines.append(f"{inner}}}{j2.if_not_loop_last()},{j2.endif()}")

60
src/jinjaturtle/handlers/xml.py
View file

@ -4,6 +4,7 @@ from collections import Counter, defaultdict
from pathlib import Path
from typing import Any
import xml.etree.ElementTree as ET # nosec
import defusedxml.ElementTree as DET
from .base import BaseHandler
from .. import j2
@ -20,9 +21,12 @@ class XmlHandler(BaseHandler):
def parse(self, path: Path) -> ET.Element:
text = path.read_text(encoding="utf-8")
parser = ET.XMLParser(
target=ET.TreeBuilder(insert_comments=False)
) # nosec B314
parser = DET.XMLParser(
target=ET.TreeBuilder(insert_comments=False),
forbid_dtd=True,
forbid_entities=True,
forbid_external=True,
)
parser.feed(text)
root = parser.close()
return root
@ -231,15 +235,6 @@ class XmlHandler(BaseHandler):
walk(root, ())
# Internal marker prefixes used by JinjaTurtle's own comment nodes. These
# must NOT be escaped (they are converted into real Jinja control structures
# downstream). Source-file comments have none of these prefixes.
_MARKER_PREFIXES = ("LOOP:", "IF:", "ENDIF:")
def _is_jt_marker(self, comment_text: str) -> bool:
stripped = (comment_text or "").lstrip()
return any(stripped.startswith(p) for p in self._MARKER_PREFIXES)
def _escape_source_comments(self, root: ET.Element) -> None:
"""Escape template metacharacters in comments preserved from the source.
@ -252,30 +247,34 @@ class XmlHandler(BaseHandler):
placeholders, so comments (and the prolog, handled separately) are the
only XML injection vector.
JinjaTurtle's own internal marker comments are left untouched so they can
be converted into real loops/conditionals later.
This function is called immediately after parsing source XML and before
JinjaTurtle appends any of its own marker comments. Therefore every
comment currently in the tree is attacker-controlled source data and must
be escaped, including comments that resemble JinjaTurtle's internal
marker syntax.
"""
# ET represents comments with a callable tag (ET.Comment). Iterate all
# descendants and escape comment text that is not one of our markers.
for elem in root.iter():
if elem.tag is ET.Comment:
if not self._is_jt_marker(elem.text or ""):
elem.text = escape_jinja_literal(elem.text or "")
def _generate_xml_template_from_text(self, role_prefix: str, text: str) -> str:
"""Generate scalar-only Jinja2 template."""
prolog, body = self._split_xml_prolog(text)
parser = ET.XMLParser(target=ET.TreeBuilder(insert_comments=True)) # nosec B314
parser = DET.XMLParser(
target=ET.TreeBuilder(insert_comments=True),
forbid_dtd=True,
forbid_entities=True,
forbid_external=True,
)
parser.feed(body)
root = parser.close()
self._apply_jinja_to_xml_tree(role_prefix, root)
# Neutralise template metacharacters in any comments preserved from the
# source file before serialising.
# Escape source comments before adding any JinjaTurtle marker comments.
self._escape_source_comments(root)
self._apply_jinja_to_xml_tree(role_prefix, root)
indent = getattr(ET, "indent", None)
if indent is not None:
indent(root, space=" ") # type: ignore[arg-type]
@ -293,19 +292,22 @@ class XmlHandler(BaseHandler):
prolog, body = self._split_xml_prolog(text)
# Parse with comments preserved
parser = ET.XMLParser(target=ET.TreeBuilder(insert_comments=True)) # nosec B314
# Parse with comments preserved.
parser = DET.XMLParser(
target=ET.TreeBuilder(insert_comments=True),
forbid_dtd=True,
forbid_entities=True,
forbid_external=True,
)
parser.feed(body)
root = parser.close()
# Escape source comments before adding any JinjaTurtle marker comments.
self._escape_source_comments(root)
# Apply Jinja transformations (including loop markers)
self._apply_jinja_to_xml_tree(role_prefix, root, loop_candidates)
# Escape comments preserved from the source. JinjaTurtle's own
# LOOP/IF/ENDIF marker comments are recognised and left intact so they
# can be converted into real Jinja control structures below.
self._escape_source_comments(root)
# Convert to string
indent = getattr(ET, "indent", None)
if indent is not None:

27
src/jinjaturtle/multi.py
View file

@ -29,7 +29,13 @@ from typing import Any, Iterable
import xml.etree.ElementTree as ET # nosec
from . import j2
from .core import dump_yaml, flatten_config, make_var_name, parse_config
from .core import (
check_var_name_collisions,
dump_yaml,
flatten_config,
make_var_name,
parse_config,
)
from .handlers.xml import XmlHandler
from .safety import verify_jinja2_template_safe
from .escape import escape_jinja_literal
@ -45,7 +51,7 @@ SUPPORTED_SUFFIXES: dict[str, set[str]] = {
def is_supported_file(path: Path) -> bool:
if not path.is_file():
if path.is_symlink() or not path.is_file():
return False
suffix = path.suffix.lower()
for exts in SUPPORTED_SUFFIXES.values():
@ -57,13 +63,25 @@ def is_supported_file(path: Path) -> bool:
def iter_supported_files(root: Path, recursive: bool) -> list[Path]:
if not root.exists():
raise FileNotFoundError(str(root))
if root.is_symlink():
return []
if root.is_file():
return [root] if is_supported_file(root) else []
if not root.is_dir():
return []
resolved_root = root.resolve(strict=True)
it = root.rglob("*") if recursive else root.glob("*")
files = [p for p in it if is_supported_file(p)]
files: list[Path] = []
for p in it:
if not is_supported_file(p):
continue
try:
resolved = p.resolve(strict=True)
resolved.relative_to(resolved_root)
except (OSError, ValueError):
continue
files.append(p)
files.sort()
return files
@ -684,6 +702,7 @@ def process_directory(
for rid, parsed, containers in zip(rel_ids, parsed_list, container_sets):
item: dict[str, Any] = {"id": rid}
flat = flatten_config(fmt, parsed, loop_candidates=None)
check_var_name_collisions(role_prefix, flat)
for path, value in flat:
item[make_var_name(role_prefix, path)] = value
for cpath in optional_containers:
@ -713,6 +732,7 @@ def process_directory(
for rid, parser in zip(rel_ids, parsers): # type: ignore[arg-type]
item: dict[str, Any] = {"id": rid}
flat = flatten_config(fmt, parser, loop_candidates=None)
check_var_name_collisions(role_prefix, flat)
for path, value in flat:
item[make_var_name(role_prefix, path)] = value
# section presence
@ -759,6 +779,7 @@ def process_directory(
for rid, parsed, elems in zip(rel_ids, parsed_list, elem_sets):
item: dict[str, Any] = {"id": rid}
flat = flatten_config(fmt, parsed, loop_candidates=None)
check_var_name_collisions(role_prefix, flat)
for path, value in flat:
item[make_var_name(role_prefix, path)] = value
for epath in optional_elements:

51
src/jinjaturtle/safety.py
View file

@ -22,7 +22,7 @@ single global property:
The check is positive/allowlist-based, which is the safe direction: unknown
constructs are rejected, not ignored. It runs at the single choke points in
``core.py`` (``generate_jinja2_template`` / ``generate_erb_template``), so it
``core.py`` (``generate_jinja2_template``), so it
covers every current handler and every future one automatically.
Why this is robust against the escaper being wrong
@ -42,7 +42,6 @@ import re
__all__ = [
"TemplateSafetyError",
"verify_jinja2_template_safe",
"verify_erb_template_safe",
]
@ -205,51 +204,3 @@ def verify_jinja2_template_safe(template_text: str) -> None:
# tokens) so the reconstructed body matches the source spacing.
body_parts.append(value if isinstance(value, str) else str(value))
# raw_begin / raw_end / data tokens outside a tag are inert: skip.
# --------------------------------------------------------------------------- #
# ERB gate.
#
# JinjaTurtle's ERB output is produced by translating the (already-verified)
# Jinja2 subset, so the Jinja2 gate is the primary guarantee. As an independent
# ERB-side backstop we confirm that every ERB tag body is one the translator
# emits, and that no Jinja2 delimiters survived into the ERB output (which would
# indicate a raw block the translator failed to recognise -- the historical
# ``{%+ raw %}`` blind spot).
# --------------------------------------------------------------------------- #
_ERB_TAG_RE = re.compile(r"<%[-=#]?(.*?)[-]?%>", re.S)
_JINJA_DELIMS = ("{{", "}}", "{%", "%}", "{#", "#}")
# Bodies the ErbTranslator emits. Kept permissive for Ruby method chains it
# constructs (``@var``, ``.each_with_index``, ``JSON.generate(...)`` etc.) but
# anchored so arbitrary attacker text cannot masquerade as one.
_ERB_STMT_PATTERNS = tuple(
re.compile(p)
for p in (
r"^require 'json'$",
r"^end$",
r"^else$",
r"^@?[A-Za-z_][\w@\.\[\]'\"]*\.each_with_index do \|[A-Za-z_]\w*, __jt_idx_\d+\| $",
r"^if .+$",
r"^elsif .+$",
r"^unless .+\.nil\?$",
r"^# Unsupported JinjaTurtle statement: .*$",
)
)
def verify_erb_template_safe(template_text: str) -> None:
"""Validate that *template_text* contains no leftover Jinja2 delimiters.
The translator is the security-relevant step for ERB; this backstop ensures
no live Jinja construct survived translation (which would mean a raw block
was not recognised and source text passed through untouched).
"""
for delim in _JINJA_DELIMS:
if delim in template_text:
raise TemplateSafetyError(
"refusing to emit ERB template: it still contains the Jinja2 "
f"delimiter {delim!r}, which means source text was not fully "
"translated/neutralised (possible template injection)."
)

54
tests/test_cli.py
View file

@ -132,57 +132,3 @@ def test_cli_folder_single_output_file_when_one_format(tmp_path):
assert defaults_path.is_file()
assert template_path.is_file()
assert "to_json" in template_path.read_text(encoding="utf-8")
def test_cli_erb_outputs_puppet_hiera_and_erb_template(tmp_path):
cfg = tmp_path / "app.ini"
cfg.write_text("[main]\nport = 8080\n", encoding="utf-8")
data_out = tmp_path / "common.yaml"
template_out = tmp_path / "app.ini.erb"
exit_code = cli._main(
[
str(cfg),
"-r",
"php",
"--template-engine",
"erb",
"--defaults-output",
str(data_out),
"--template-output",
str(template_out),
]
)
assert exit_code == 0
assert "php::main_port: '8080'" in data_out.read_text(encoding="utf-8")
assert "port = <%= @main_port %>" in template_out.read_text(encoding="utf-8")
def test_cli_erb_can_use_separate_puppet_class_namespace(tmp_path):
cfg = tmp_path / "app.ini"
cfg.write_text("[main]\nport = 8080\n", encoding="utf-8")
data_out = tmp_path / "node.yaml"
template_out = tmp_path / "app.ini.erb"
exit_code = cli._main(
[
str(cfg),
"-r",
"php_etc_app_ini",
"--template-engine",
"erb",
"--puppet-class",
"php",
"--defaults-output",
str(data_out),
"--template-output",
str(template_out),
]
)
assert exit_code == 0
data = data_out.read_text(encoding="utf-8")
template = template_out.read_text(encoding="utf-8")
assert "php::php_etc_app_ini_main_port: '8080'" in data
assert "port = <%= @php_etc_app_ini_main_port %>" in template

122
tests/test_deep_nesting_dos.py Normal file
View file

@ -0,0 +1,122 @@
"""Regression tests for the deeply-nested-input denial-of-service fix.
Pathologically nested configuration (thousands of container levels) used to
drive the recursive walkers in ``core`` -- or the underlying parser itself --
past Python's stack limit, raising an uncaught ``RecursionError``. The fix
bounds nesting depth and converts both cases into a clean, catchable
``ConfigTooDeeplyNestedError``. These tests pin that behaviour and guard
against regressions, while confirming normally-nested configs still process.
"""
from __future__ import annotations
import tempfile
from pathlib import Path
import pytest
from jinjaturtle.core import (
ConfigTooDeeplyNestedError,
MAX_CONFIG_DEPTH,
analyze_loops,
flatten_config,
generate_ansible_yaml,
generate_jinja2_template,
parse_config,
)
def _run_pipeline(content: str, suffix: str) -> None:
with tempfile.NamedTemporaryFile(
"w", suffix=suffix, delete=False, encoding="utf-8"
) as tf:
tf.write(content)
path = Path(tf.name)
try:
fmt, parsed = parse_config(path)
loops = analyze_loops(fmt, parsed)
flat = flatten_config(fmt, parsed, loops)
generate_jinja2_template(
fmt, parsed, "role", original_text=content, loop_candidates=loops
)
generate_ansible_yaml("role", flat, loops)
finally:
path.unlink()
def _deep_json_objects(depth: int) -> str:
s = "0"
for _ in range(depth):
s = '{"a":' + s + "}"
return s
def _deep_json_arrays(depth: int) -> str:
return "[" * depth + "1" + "]" * depth
def _deep_xml(depth: int) -> str:
s = "v"
for _ in range(depth):
s = f"<a>{s}</a>"
return s
def _deep_yaml(depth: int) -> str:
s = "v"
for _ in range(depth):
s = "{a: " + s + "}"
return s
def _deep_toml(depth: int) -> str:
return "a = " + "[" * depth + "1" + "]" * depth
@pytest.mark.parametrize(
"content, suffix",
[
(_deep_json_objects(5000), ".json"),
(_deep_json_arrays(100000), ".json"),
(_deep_xml(5000), ".xml"),
(_deep_yaml(3000), ".yaml"),
(_deep_toml(2000), ".toml"),
],
)
def test_deeply_nested_input_is_rejected_cleanly(content: str, suffix: str) -> None:
# Must raise our typed error, and crucially must NOT raise RecursionError
# (pytest would report that as an error, but be explicit about intent).
with pytest.raises(ConfigTooDeeplyNestedError):
_run_pipeline(content, suffix)
def test_recursion_error_never_escapes() -> None:
# Belt-and-suspenders: ensure a RecursionError is never what surfaces.
content = _deep_yaml(6000)
try:
_run_pipeline(content, ".yaml")
except ConfigTooDeeplyNestedError:
pass
except RecursionError: # pragma: no cover - this is the bug we fixed
pytest.fail("RecursionError escaped instead of ConfigTooDeeplyNestedError")
@pytest.mark.parametrize(
"content, suffix",
[
('{"name":"web","port":8080,"tags":["a","b"]}', ".json"),
("<config><host>db</host><port>5432</port></config>", ".xml"),
("name: web\nport: 8080\nnested:\n a:\n b: 1\n", ".yaml"),
('[section]\nkey = "value"\nnums = [1, 2, 3]\n', ".toml"),
("[section]\nkey = value\n", ".ini"),
],
)
def test_normal_configs_still_process(content: str, suffix: str) -> None:
# No exception expected for ordinary, shallow configurations.
_run_pipeline(content, suffix)
def test_just_under_limit_is_accepted() -> None:
# A structure comfortably under the cap must still process successfully.
content = _deep_json_objects(MAX_CONFIG_DEPTH - 10)
_run_pipeline(content, ".json")

38
tests/test_injection_security.py
View file

@ -2,9 +2,9 @@
JinjaTurtle copies parts of the source config (comments, unrecognised lines,
structural keys) verbatim into the generated template. If that text contains
Jinja2/ERB delimiters it must be neutralised, otherwise attacker-influenced
config content becomes live template code that executes when Salt/Ansible/Puppet
later renders the template.
Jinja2 delimiters it must be neutralised, otherwise attacker-influenced
config content becomes live template code when a downstream tool renders the
template.
These tests render the *generated* template the way a downstream tool would and
assert that an injected payload never executes. A tripwire object is exposed
@ -14,7 +14,6 @@ the tripwire sentinel, an injected expression executed and the test fails.
from __future__ import annotations
import re
import subprocess
import sys
from pathlib import Path
@ -25,7 +24,6 @@ import yaml as pyyaml
from jinjaturtle.escape import (
escape_jinja_literal,
escape_erb_literal,
escape_literal,
)
@ -235,36 +233,10 @@ def test_endraw_whitespace_control_cannot_break_out(endraw):
assert rendered == payload
@pytest.mark.parametrize(
"payload",
[
"<%= system('id') %>",
"<% require 'open3' %>",
"text <%= 1+1 %> more",
"-%> orphan <%",
],
)
def test_escape_erb_literal_removes_executable_tags(payload):
escaped = escape_erb_literal(payload)
# No raw executable ERB tag should survive that contains the original code.
live = re.findall(r"<%[-=#]?(.*?)-?%>", escaped, re.S)
for chunk in live:
assert "system" not in chunk
assert "require" not in chunk
# Numeric/expression payloads must be reduced to literal-string prints.
def test_escape_literal_dispatches_by_engine():
"""The public ``escape_literal`` wrapper routes to the right engine."""
def test_escape_literal_is_jinja_alias():
"""The public ``escape_literal`` wrapper is Jinja2-only."""
payload = "{{ 7*7 }}"
# Default engine is Jinja2.
assert escape_literal(payload) == escape_jinja_literal(payload)
assert escape_literal(payload, engine="jinja2") == escape_jinja_literal(payload)
# An unknown engine falls back to the safer Jinja2 escaping.
assert escape_literal(payload, engine="nonsense") == escape_jinja_literal(payload)
# ERB routing.
erb_payload = "<%= system('id') %>"
assert escape_literal(erb_payload, engine="erb") == escape_erb_literal(erb_payload)
def test_escape_literal_jinja_output_is_inert():

15
tests/test_output_safety_gate.py
View file

@ -25,7 +25,6 @@ from jinjaturtle.multi import process_directory
from jinjaturtle.safety import (
TemplateSafetyError,
verify_jinja2_template_safe,
verify_erb_template_safe,
)
@ -180,20 +179,6 @@ def _strip_raw_blocks(text: str) -> str:
)
# --------------------------------------------------------------------------- #
# ERB gate.
# --------------------------------------------------------------------------- #
def test_erb_gate_blocks_leftover_jinja_delimiters():
with pytest.raises(TemplateSafetyError):
verify_erb_template_safe("ok <%= @x %> but {{ leftover }} here")
def test_erb_gate_allows_clean_erb():
verify_erb_template_safe("memory = <%= @memory_limit %>\n")
# --------------------------------------------------------------------------- #
# End-to-end CLI: fail closed.
# --------------------------------------------------------------------------- #

98
tests/test_release_blockers.py Normal file
View file

@ -0,0 +1,98 @@
from pathlib import Path
import pytest
import yaml
from defusedxml.common import DTDForbidden
from jinjaturtle.core import (
VariableNameCollisionError,
generate_ansible_yaml,
parse_config,
)
from jinjaturtle.handlers.xml import XmlHandler
from jinjaturtle.multi import iter_supported_files, process_directory
def test_defaults_string_values_are_ansible_unsafe():
output = generate_ansible_yaml(
"role", [(("section", "key"), "{{ lookup('pipe', 'id') }}")]
)
assert "!unsafe" in output
assert yaml.safe_load(output)["role_section_key"] == "{{ lookup('pipe', 'id') }}"
def test_variable_name_collisions_fail_closed():
with pytest.raises(VariableNameCollisionError):
generate_ansible_yaml(
"role",
[
(("section", "a-b"), "safe"),
(("section", "a_b"), "evil"),
],
)
def test_folder_mode_does_not_follow_file_symlinks(tmp_path: Path):
outside = tmp_path / "outside.ini"
outside.write_text("[s]\nsecret = value\n", encoding="utf-8")
root = tmp_path / "root"
root.mkdir()
(root / "link.ini").symlink_to(outside)
(root / "real.ini").write_text("[s]\nname = ok\n", encoding="utf-8")
files = iter_supported_files(root, recursive=True)
assert files == [root / "real.ini"]
defaults, _outputs = process_directory(root, recursive=True, role_prefix="role")
assert "secret" not in defaults
assert "real.ini" in defaults
def test_xml_parser_rejects_dtd_and_entities(tmp_path: Path):
xml_path = tmp_path / "evil.xml"
xml_path.write_text('<!DOCTYPE r [<!ENTITY x "boom">]><r>&x;</r>', encoding="utf-8")
with pytest.raises(DTDForbidden):
parse_config(xml_path)
def test_xml_source_comments_cannot_forge_internal_markers():
source = (
"<root><!--IF:role_missing--><name>ok</name><!--ENDIF:role_missing--></root>"
)
template = XmlHandler().generate_jinja2_template(None, "role", original_text=source)
assert "{% if role_missing is defined %}" not in template
assert "{% endif %}" not in template
assert "IF:role_missing" in template
assert "ENDIF:role_missing" in template
def test_cli_refuses_to_write_output_symlink(tmp_path: Path):
import subprocess
import sys
src = tmp_path / "input.ini"
src.write_text("[s]\nname = ok\n", encoding="utf-8")
target = tmp_path / "target.yml"
target.write_text("do not overwrite\n", encoding="utf-8")
link = tmp_path / "defaults.yml"
link.symlink_to(target)
result = subprocess.run(
[
sys.executable,
"-m",
"jinjaturtle.cli",
str(src),
"-d",
str(link),
],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env={"PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src")},
check=False,
)
assert result.returncode != 0
assert target.read_text(encoding="utf-8") == "do not overwrite\n"

15
tests/test_yaml_handler.py
View file

@ -10,7 +10,6 @@ from jinjaturtle.core import (
analyze_loops,
flatten_config,
generate_ansible_yaml,
generate_erb_template,
generate_jinja2_template,
)
from jinjaturtle.handlers.yaml import YamlHandler
@ -178,17 +177,15 @@ def test_yaml_loop_preserves_blank_separator_after_list(tmp_path: Path):
"blank",
"require:\n - rubocop-performance\n - rubocop-rspec\n\nAllCops:\n NewCops: enable\n",
"\n - rubocop-rspec\n\nAllCops:",
"<% end %>\nAllCops:",
),
(
"no_blank",
"require:\n - rubocop-performance\n - rubocop-rspec\nAllCops:\n NewCops: enable\n",
"\n - rubocop-rspec\nAllCops:",
"<% end %>AllCops:",
),
]
for label, text, rendered_expected, erb_expected in cases:
for label, text, rendered_expected in cases:
path = tmp_path / f"{label}.yml"
path.write_text(text, encoding="utf-8")
@ -205,16 +202,6 @@ def test_yaml_loop_preserves_blank_separator_after_list(tmp_path: Path):
rendered = Template(template).render(**defaults)
assert rendered_expected in rendered
erb_template = generate_erb_template(
fmt,
parsed,
"role",
original_text=text,
loop_candidates=loop_candidates,
flat_items=flat_items,
)
assert erb_expected in erb_template
def test_yaml_loop_preserves_following_top_level_comments(tmp_path: Path):
from jinja2 import Environment, Template