Compare commits
No commits in common. "main" and "0.5.6" have entirely different histories.
26 changed files with 722 additions and 1807 deletions
130
README.md
130
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
|
||||
|
|
|
|||
12
debian/changelog
vendored
12
debian/changelog
vendored
|
|
@ -1,15 +1,3 @@
|
|||
jinjaturtle (0.7.0) unstable; urgency=medium
|
||||
|
||||
* Much hardening.
|
||||
|
||||
-- Miguel Jacq <mig@mig5.net> Sun, 5 Jul 2026 10:54:00 +1000
|
||||
|
||||
jinjaturtle (0.5.7) unstable; urgency=medium
|
||||
|
||||
* More hardening measures
|
||||
|
||||
-- Miguel Jacq <mig@mig5.net> 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
|
||||
|
|
|
|||
|
|
@ -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" },
|
||||
|
|
|
|||
|
|
@ -46,6 +46,7 @@ REPO_ROOT="${HOME}/git/repo_rpm"
|
|||
REMOTE="ashpool.mig5.net:/opt/repo_rpm"
|
||||
|
||||
DISTS=(
|
||||
fedora:44
|
||||
fedora:43
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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 <mig@mig5.net> - %{version}-%{release}
|
||||
- Much hardening
|
||||
* Wed Jun 24 2026 Miguel Jacq <mig@mig5.net> - %{version}-%{release}
|
||||
- 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
|
||||
* Sat Jun 20 2026 Miguel Jacq <mig@mig5.net> - %{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 <mig@mig5.net> - %{version}-%{release}
|
||||
- Empty dicts and lists are now emitted as leaf defaults.
|
||||
* Mon May 11 2026 Miguel Jacq <mig@mig5.net> - %{version}-%{release}
|
||||
* Tue May 11 2026 Miguel Jacq <mig@mig5.net> - %{version}-%{release}
|
||||
- Support ssh configs
|
||||
* Tue Jan 06 2026 Miguel Jacq <mig@mig5.net> - %{version}-%{release}
|
||||
- Support converting systemd files and postfix main.cf
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
241
src/jinjaturtle/erb.py
Normal file
241
src/jinjaturtle/erb.py
Normal file
|
|
@ -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)
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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()}")
|
||||
|
|
|
|||
|
|
@ -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))} = "
|
||||
|
|
|
|||
|
|
@ -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]]:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)"
|
||||
|
||||
|
|
|
|||
|
|
@ -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.<field>``.
|
||||
# ``<field>`` 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.<key>`` 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.<key>`` 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]
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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 <collection>`` 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)."
|
||||
)
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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", "<root><{{tag}}/></root>"),
|
||||
("xml", "bad2.xml", "<root><unclosed></root>"),
|
||||
("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 = (
|
||||
'<?xml version="1.0"?>\n'
|
||||
'<!DOCTYPE foo [ <!ENTITY xxe SYSTEM "file:///etc/passwd"> ]>\n'
|
||||
"<foo>&xxe;</foo>\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("<root><{{tag}}/></root>", 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.
|
||||
(
|
||||
'<?xml version="1.0"?>\n'
|
||||
'<!DOCTYPE root [<!ENTITY xxe SYSTEM "file:///etc/hostname">]>\n'
|
||||
"<root>&xxe;</root>\n"
|
||||
),
|
||||
# Billion-laughs style internal entity expansion.
|
||||
(
|
||||
'<?xml version="1.0"?>\n'
|
||||
'<!DOCTYPE lolz [<!ENTITY lol "lol"><!ENTITY lol2 "&lol;&lol;&lol;">]>\n'
|
||||
"<lolz>&lol2;</lolz>\n"
|
||||
),
|
||||
# External parameter entity (SSRF / out-of-band XXE vector).
|
||||
(
|
||||
'<?xml version="1.0"?>\n'
|
||||
'<!DOCTYPE root [<!ENTITY % ext SYSTEM "http://127.0.0.1:9/x.dtd"> %ext;]>\n'
|
||||
"<root>x</root>\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}
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -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(
|
||||
"<!DOCTYPE root [<!ENTITY xxe SYSTEM 'file:///etc/passwd'>]><root>&xxe;</root>",
|
||||
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()
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue