Compare commits
3 commits
f63cb92b35
...
70be0f7e33
| Author | SHA1 | Date | |
|---|---|---|---|
| 70be0f7e33 | |||
| 4a1f2ac15e | |||
| 661320558c |
14 changed files with 577 additions and 690 deletions
125
README.md
125
README.md
|
|
@ -13,9 +13,6 @@ By default it generates:
|
|||
- an **Ansible defaults YAML** file containing the variables used by that
|
||||
template.
|
||||
|
||||
It can also generate **ERB** templates and **Puppet Hiera-style YAML** data for
|
||||
Puppet workflows.
|
||||
|
||||
JinjaTurtle does not try to replace configuration-management tools. Its job is
|
||||
to speed up the boring first pass: take a real config file, discover the values
|
||||
inside it, replace those values with variables, and write the corresponding
|
||||
|
|
@ -37,16 +34,6 @@ For the default Jinja2/Ansible mode:
|
|||
5. An Ansible defaults YAML file is generated with those variables and the
|
||||
original values.
|
||||
|
||||
For ERB/Puppet mode:
|
||||
|
||||
1. The same parse/flatten/loop analysis is used.
|
||||
2. An ERB template is generated with Puppet-style instance variables such as
|
||||
`<%= @memory_limit %>`.
|
||||
3. The variables file is written as Puppet Hiera-style data, such as
|
||||
`php::memory_limit: 256M`.
|
||||
4. If `--puppet-class` is supplied, that class name is used as the Hiera
|
||||
namespace while `--role-name` remains the local variable prefix.
|
||||
|
||||
By default, the generated variable data and template are printed to stdout. Use
|
||||
`--defaults-output` and `--template-output` to write them to files.
|
||||
|
||||
|
|
@ -80,78 +67,6 @@ and defaults data like:
|
|||
php_memory_limit: 256M
|
||||
```
|
||||
|
||||
## ERB / Puppet example
|
||||
|
||||
Use `--template-engine erb` when you want Puppet ERB output:
|
||||
|
||||
```shell
|
||||
jinjaturtle php.ini \
|
||||
--role-name php \
|
||||
--template-engine erb \
|
||||
--defaults-output data/common.yaml \
|
||||
--template-output templates/php.ini.erb
|
||||
```
|
||||
|
||||
Given the same source value:
|
||||
|
||||
```ini
|
||||
memory_limit = 256M
|
||||
```
|
||||
|
||||
JinjaTurtle will produce an ERB template value like:
|
||||
|
||||
```erb
|
||||
memory_limit = <%= @memory_limit %>
|
||||
```
|
||||
|
||||
and Hiera-style data like:
|
||||
|
||||
```yaml
|
||||
php::memory_limit: 256M
|
||||
```
|
||||
|
||||
The `--defaults-output` option name is retained for CLI compatibility, but in
|
||||
ERB mode the file is intended to be Puppet Hiera data rather than Ansible role
|
||||
defaults.
|
||||
|
||||
JinjaTurtle does **not** generate Puppet classes or `file` resources. A Puppet
|
||||
module, should still declare the class parameters and call the template, for
|
||||
example with Puppet's `template()` function.
|
||||
|
||||
## Using `--puppet-class`
|
||||
|
||||
Most direct usage can simply use the same value for the role name and Puppet
|
||||
class name:
|
||||
|
||||
```shell
|
||||
jinjaturtle php.ini --role-name php --template-engine erb
|
||||
```
|
||||
|
||||
This creates Hiera keys such as:
|
||||
|
||||
```yaml
|
||||
php::memory_limit: 256M
|
||||
```
|
||||
|
||||
and local ERB variables such as:
|
||||
|
||||
```erb
|
||||
<%= @memory_limit %>
|
||||
```
|
||||
|
||||
For generated systems, it can be useful to make `--role-name` more specific
|
||||
while keeping the Hiera keys under the real Puppet class. For example:
|
||||
|
||||
```shell
|
||||
jinjaturtle php.ini \
|
||||
--role-name php_etc_php_ini \
|
||||
--puppet-class php \
|
||||
--template-engine erb
|
||||
```
|
||||
|
||||
In that case the variable prefix can stay file-specific, while the Hiera data
|
||||
is still written under `php::...`.
|
||||
|
||||
## What sort of config files can it handle?
|
||||
|
||||
JinjaTurtle supports common structured and semi-structured config formats:
|
||||
|
|
@ -176,8 +91,8 @@ homogeneous enough. If it is not confident, it falls back to flattened scalar
|
|||
variables.
|
||||
|
||||
Some very complex files will still need manual cleanup. The goal is to speed up
|
||||
conversion into Jinja2 or ERB templates, not to guarantee a perfect final module
|
||||
without review.
|
||||
conversion into Jinja2 templates, not to guarantee a perfect final module without
|
||||
review.
|
||||
|
||||
## JSON, quoting, and type preservation
|
||||
|
||||
|
|
@ -196,17 +111,7 @@ when the correct rendered JSON should be:
|
|||
{"enabled": true}
|
||||
```
|
||||
|
||||
In Jinja2 mode this uses Ansible-style JSON filters. In ERB mode it emits Ruby
|
||||
JSON generation where required, for example:
|
||||
|
||||
```erb
|
||||
<% require 'json' -%>
|
||||
{
|
||||
"enabled": <%= JSON.generate(@enabled) %>
|
||||
}
|
||||
```
|
||||
|
||||
That is expected for JSON ERB templates.
|
||||
This uses Ansible-style JSON filters.
|
||||
|
||||
## Can I convert multiple files at once?
|
||||
|
||||
|
|
@ -287,8 +192,6 @@ poetry install
|
|||
usage: jinjaturtle [-h] [-r ROLE_NAME] [--recursive]
|
||||
[-f {ini,json,toml,yaml,xml,postfix,systemd,ssh}]
|
||||
[-d DEFAULTS_OUTPUT] [-t TEMPLATE_OUTPUT]
|
||||
[--template-engine {jinja2,erb}]
|
||||
[--puppet-class PUPPET_CLASS]
|
||||
config
|
||||
|
||||
Convert a config file into an Ansible defaults file and Jinja2 template.
|
||||
|
|
@ -302,8 +205,7 @@ options:
|
|||
-h, --help show this help message and exit
|
||||
-r, --role-name ROLE_NAME
|
||||
Role name / variable prefix. In Jinja2 mode this is
|
||||
usually the Ansible role name. In ERB mode it is used
|
||||
as the local variable prefix. Defaults to jinjaturtle.
|
||||
usually the Ansible role name. Defaults to jinjaturtle.
|
||||
--recursive When CONFIG is a folder, recurse into subfolders.
|
||||
-f, --format {ini,json,toml,yaml,xml,postfix,systemd,ssh}
|
||||
Force config format instead of auto-detecting from
|
||||
|
|
@ -314,12 +216,6 @@ options:
|
|||
-t, --template-output TEMPLATE_OUTPUT
|
||||
Path to write the generated config template. If omitted,
|
||||
it is printed to stdout.
|
||||
--template-engine {jinja2,erb}
|
||||
Template syntax to generate. Defaults to jinja2. Use
|
||||
erb for Puppet templates.
|
||||
--puppet-class PUPPET_CLASS
|
||||
Puppet class / Hiera namespace to use with
|
||||
--template-engine erb. Defaults to --role-name.
|
||||
```
|
||||
|
||||
## Additional supported formats
|
||||
|
|
@ -346,16 +242,16 @@ Two guarantees matter:
|
|||
|
||||
1. **Values are data, never code.** Every config *value* is replaced with a
|
||||
`{{ variable }}` placeholder in the template, and the original value is stored
|
||||
separately in the defaults/Hiera data. When the template is later rendered,
|
||||
separately in the defaults data. When the template is later rendered,
|
||||
the placeholder prints the value as a literal string; Jinja2 does not
|
||||
recursively render the *contents* of a variable, so a payload sitting inside
|
||||
a value (for example `motd = {{ salt['cmd.run']('id') }}`) is inert.
|
||||
a value is inert.
|
||||
|
||||
2. **Verbatim text is neutralised.** To preserve formatting, JinjaTurtle copies
|
||||
comments, blank lines, headers and any unrecognised lines from the source
|
||||
into the template. Any template metacharacters in that copied text
|
||||
(`{{ }}`, `{% %}`, `{# #}` for Jinja2; `<% %>` for ERB) are escaped so they
|
||||
render as the literal characters the author wrote, rather than executing.
|
||||
(`{{ }}`, `{% %}`, `{# #}` ) are escaped so they render as the literal
|
||||
characters the author wrote, rather than executing.
|
||||
|
||||
### Consumer responsibilities
|
||||
|
||||
|
|
@ -368,11 +264,6 @@ which is the normal case:
|
|||
default. If you build your own var structures from this data and pass them
|
||||
through additional templating, mark untrusted values with the `!unsafe` tag so
|
||||
they are never re-evaluated.
|
||||
- **Salt**: use the generated file as a `file.managed` template
|
||||
(`template: jinja`). Do **not** place JinjaTurtle output where Salt would
|
||||
render it a second time as part of SLS/pillar rendering, which is a separate
|
||||
Jinja pass and would re-evaluate any embedded expressions.
|
||||
- **Puppet/ERB**: render with the standard `template()`/`epp()` flow.
|
||||
|
||||
In short: render JinjaTurtle output exactly once. Do not feed it back through
|
||||
another templating pass.
|
||||
|
|
|
|||
|
|
@ -12,12 +12,11 @@ from .core import (
|
|||
flatten_config,
|
||||
generate_ansible_yaml,
|
||||
generate_jinja2_template,
|
||||
generate_puppet_hiera_yaml,
|
||||
generate_erb_template,
|
||||
)
|
||||
|
||||
from .multi import process_directory
|
||||
from .safety import TemplateSafetyError, verify_erb_template_safe
|
||||
from .safety import TemplateSafetyError
|
||||
from .output_safety import OutputPathError, ensure_safe_directory, write_text_safely
|
||||
|
||||
|
||||
def _build_arg_parser() -> argparse.ArgumentParser:
|
||||
|
|
@ -59,20 +58,6 @@ def _build_arg_parser() -> argparse.ArgumentParser:
|
|||
"--template-output",
|
||||
help="Path to write the generated config template. If omitted, template is printed to stdout.",
|
||||
)
|
||||
ap.add_argument(
|
||||
"--template-engine",
|
||||
choices=[j2.NAME, "erb"],
|
||||
default=j2.NAME,
|
||||
help="Template syntax to generate (default: jinja2). Use erb for Puppet templates.",
|
||||
)
|
||||
ap.add_argument(
|
||||
"--puppet-class",
|
||||
help=(
|
||||
"Puppet class/Hiera namespace to use with --template-engine erb. "
|
||||
"Defaults to --role-name. This lets tools use a file-specific "
|
||||
"variable prefix while writing Hiera keys under the real Puppet class."
|
||||
),
|
||||
)
|
||||
return ap
|
||||
|
||||
|
||||
|
|
@ -88,6 +73,9 @@ def _main(argv: list[str] | None = None) -> int:
|
|||
f"jinjaturtle: refusing to generate unsafe template: {exc}", file=sys.stderr
|
||||
)
|
||||
return 2
|
||||
except OutputPathError as exc:
|
||||
print(f"jinjaturtle: refusing unsafe output path: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
|
||||
def _run(argv: list[str] | None = None) -> int:
|
||||
|
|
@ -105,37 +93,23 @@ def _run(argv: list[str] | None = None) -> int:
|
|||
|
||||
# Write defaults
|
||||
if args.defaults_output:
|
||||
Path(args.defaults_output).write_text(defaults_yaml, encoding="utf-8")
|
||||
write_text_safely(Path(args.defaults_output), defaults_yaml)
|
||||
else:
|
||||
print("# defaults/main.yml")
|
||||
print(defaults_yaml, end="")
|
||||
|
||||
# Optionally translate folder-mode templates to ERB. Folder mode keeps
|
||||
# the existing data shape; single-file mode below is the preferred
|
||||
# Puppet path because it can produce class-parameter Hiera keys.
|
||||
if args.template_engine == "erb":
|
||||
from .erb import translate_jinja2_to_erb
|
||||
|
||||
for o in outputs:
|
||||
o.template = translate_jinja2_to_erb(
|
||||
o.template,
|
||||
role_prefix=args.role_name,
|
||||
puppet_class=args.puppet_class or args.role_name,
|
||||
)
|
||||
verify_erb_template_safe(o.template)
|
||||
|
||||
template_ext = "erb" if args.template_engine == "erb" else j2.TEMPLATE_EXTENSION
|
||||
template_ext = j2.TEMPLATE_EXTENSION
|
||||
|
||||
# Write templates
|
||||
if args.template_output:
|
||||
out_path = Path(args.template_output)
|
||||
if len(outputs) == 1 and not out_path.is_dir():
|
||||
out_path.write_text(outputs[0].template, encoding="utf-8")
|
||||
write_text_safely(out_path, outputs[0].template)
|
||||
else:
|
||||
out_path.mkdir(parents=True, exist_ok=True)
|
||||
ensure_safe_directory(out_path)
|
||||
for o in outputs:
|
||||
(out_path / f"config.{o.fmt}.{template_ext}").write_text(
|
||||
o.template, encoding="utf-8"
|
||||
write_text_safely(
|
||||
out_path / f"config.{o.fmt}.{template_ext}", o.template
|
||||
)
|
||||
else:
|
||||
for o in outputs:
|
||||
|
|
@ -161,51 +135,28 @@ def _run(argv: list[str] | None = None) -> int:
|
|||
# Flatten config (excluding loop paths if loops are detected)
|
||||
flat_items = flatten_config(fmt, parsed, loop_candidates)
|
||||
|
||||
if args.template_engine == "erb":
|
||||
ansible_yaml = generate_puppet_hiera_yaml(
|
||||
args.role_name,
|
||||
flat_items,
|
||||
loop_candidates,
|
||||
puppet_class=args.puppet_class or args.role_name,
|
||||
)
|
||||
template_str = generate_erb_template(
|
||||
fmt,
|
||||
parsed,
|
||||
args.role_name,
|
||||
original_text=config_text,
|
||||
loop_candidates=loop_candidates,
|
||||
flat_items=flat_items,
|
||||
puppet_class=args.puppet_class or args.role_name,
|
||||
)
|
||||
else:
|
||||
# Generate defaults YAML (with loop collections if detected)
|
||||
ansible_yaml = generate_ansible_yaml(
|
||||
args.role_name, flat_items, loop_candidates
|
||||
)
|
||||
# Generate defaults YAML (with loop collections if detected)
|
||||
ansible_yaml = generate_ansible_yaml(args.role_name, flat_items, loop_candidates)
|
||||
|
||||
# Generate template (with loops if detected)
|
||||
template_str = generate_jinja2_template(
|
||||
fmt,
|
||||
parsed,
|
||||
args.role_name,
|
||||
original_text=config_text,
|
||||
loop_candidates=loop_candidates,
|
||||
)
|
||||
# Generate template (with loops if detected)
|
||||
template_str = generate_jinja2_template(
|
||||
fmt,
|
||||
parsed,
|
||||
args.role_name,
|
||||
original_text=config_text,
|
||||
loop_candidates=loop_candidates,
|
||||
)
|
||||
|
||||
if args.defaults_output:
|
||||
Path(args.defaults_output).write_text(ansible_yaml, encoding="utf-8")
|
||||
write_text_safely(Path(args.defaults_output), ansible_yaml)
|
||||
else:
|
||||
print("# defaults/main.yml")
|
||||
print(ansible_yaml, end="")
|
||||
|
||||
if args.template_output:
|
||||
Path(args.template_output).write_text(template_str, encoding="utf-8")
|
||||
write_text_safely(Path(args.template_output), template_str)
|
||||
else:
|
||||
print(
|
||||
"# config.erb"
|
||||
if args.template_engine == "erb"
|
||||
else f"# config.{j2.TEMPLATE_EXTENSION}"
|
||||
)
|
||||
print(f"# config.{j2.TEMPLATE_EXTENSION}")
|
||||
print(template_str, end="")
|
||||
|
||||
return 0
|
||||
|
|
|
|||
|
|
@ -8,10 +8,9 @@ import re
|
|||
import yaml
|
||||
|
||||
from .loop_analyzer import LoopAnalyzer, LoopCandidate
|
||||
from .erb import puppet_class_name, puppet_local_var_name, translate_jinja2_to_erb
|
||||
from .safety import (
|
||||
verify_erb_template_safe,
|
||||
verify_jinja2_template_safe,
|
||||
verify_no_live_jinja_in_json_keys,
|
||||
)
|
||||
from .handlers import (
|
||||
BaseHandler,
|
||||
|
|
@ -34,6 +33,22 @@ class QuotedString(str):
|
|||
pass
|
||||
|
||||
|
||||
class AnsibleUnsafeString(str):
|
||||
"""Marker type emitted with Ansible's !unsafe YAML tag.
|
||||
|
||||
Ansible recursively templates string values by default. Source-derived
|
||||
config values that contain Jinja delimiters must therefore be marked
|
||||
unsafe in defaults/main.yml, otherwise a harvested value such as
|
||||
``{{ lookup('pipe', 'id') }}`` becomes executable on the Ansible
|
||||
controller when the generated role is applied.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
_JINJA_STARTS = ("{{", "{%", "{#")
|
||||
|
||||
|
||||
def _fallback_str_representer(dumper: yaml.SafeDumper, data: Any):
|
||||
"""
|
||||
Fallback for objects the dumper doesn't know about.
|
||||
|
|
@ -53,7 +68,33 @@ def _quoted_str_representer(dumper: yaml.SafeDumper, data: QuotedString):
|
|||
return dumper.represent_scalar("tag:yaml.org,2002:str", str(data), style='"')
|
||||
|
||||
|
||||
def _ansible_unsafe_str_representer(dumper: yaml.SafeDumper, data: AnsibleUnsafeString):
|
||||
return dumper.represent_scalar("!unsafe", str(data), style="'")
|
||||
|
||||
|
||||
def _needs_ansible_unsafe(value: str) -> bool:
|
||||
return any(marker in value for marker in _JINJA_STARTS)
|
||||
|
||||
|
||||
def _mark_ansible_unsafe_values(obj: Any) -> Any:
|
||||
"""Recursively mark mapping/list values containing Jinja as !unsafe.
|
||||
|
||||
Mapping keys are intentionally left alone: they are variable names or YAML
|
||||
structure, not Ansible-templated values. Values nested in folder-mode item
|
||||
lists, including source-derived ``id`` values, are protected.
|
||||
"""
|
||||
|
||||
if isinstance(obj, dict):
|
||||
return {k: _mark_ansible_unsafe_values(v) for k, v in obj.items()}
|
||||
if isinstance(obj, list):
|
||||
return [_mark_ansible_unsafe_values(v) for v in obj]
|
||||
if isinstance(obj, str) and _needs_ansible_unsafe(obj):
|
||||
return AnsibleUnsafeString(obj)
|
||||
return obj
|
||||
|
||||
|
||||
_TurtleDumper.add_representer(QuotedString, _quoted_str_representer)
|
||||
_TurtleDumper.add_representer(AnsibleUnsafeString, _ansible_unsafe_str_representer)
|
||||
# Use our fallback for any unknown object types
|
||||
_TurtleDumper.add_representer(None, _fallback_str_representer)
|
||||
|
||||
|
|
@ -85,8 +126,9 @@ def dump_yaml(data: Any, *, sort_keys: bool = True) -> str:
|
|||
|
||||
This is used by both the single-file and multi-file code paths.
|
||||
"""
|
||||
safe_data = _mark_ansible_unsafe_values(data)
|
||||
return yaml.dump(
|
||||
data,
|
||||
safe_data,
|
||||
Dumper=_TurtleDumper,
|
||||
sort_keys=sort_keys,
|
||||
default_flow_style=False,
|
||||
|
|
@ -396,6 +438,13 @@ def generate_jinja2_template(
|
|||
# verbatim source text, the un-escaped payload shows up here as a live tag
|
||||
# and generation aborts instead of emitting an injectable template.
|
||||
verify_jinja2_template_safe(template)
|
||||
|
||||
# Format-specific backstop: JinjaTurtle never emits Jinja inside a JSON object
|
||||
# key, so a live construct in key position means source key text leaked into
|
||||
# the template unescaped. This is independent of per-handler escaping.
|
||||
if fmt == "json":
|
||||
verify_no_live_jinja_in_json_keys(template)
|
||||
|
||||
return template
|
||||
|
||||
|
||||
|
|
@ -411,74 +460,6 @@ def _template_variable_names(
|
|||
return names
|
||||
|
||||
|
||||
def generate_puppet_hiera_yaml(
|
||||
role_prefix: str,
|
||||
flat_items: list[tuple[tuple[str, ...], Any]],
|
||||
loop_candidates: list[LoopCandidate] | None = None,
|
||||
*,
|
||||
puppet_class: str | None = None,
|
||||
) -> str:
|
||||
"""Create Puppet Hiera data suitable for Automatic Parameter Lookup.
|
||||
|
||||
``role_prefix`` remains the source variable prefix used by JinjaTurtle while
|
||||
``puppet_class`` is the Puppet class/Hiera namespace. In the normal case
|
||||
they are the same, so ``php_memory_limit`` becomes ``php::memory_limit``.
|
||||
"""
|
||||
|
||||
klass = puppet_class_name(puppet_class or role_prefix)
|
||||
data: dict[str, Any] = {}
|
||||
|
||||
for path, value in flat_items:
|
||||
generated = make_var_name(role_prefix, path)
|
||||
local = puppet_local_var_name(role_prefix, generated, puppet_class=klass)
|
||||
data[f"{klass}::{local}"] = value
|
||||
|
||||
if loop_candidates:
|
||||
for candidate in loop_candidates:
|
||||
generated = make_var_name(role_prefix, candidate.path)
|
||||
local = puppet_local_var_name(role_prefix, generated, puppet_class=klass)
|
||||
data[f"{klass}::{local}"] = candidate.items
|
||||
|
||||
return dump_yaml(data, sort_keys=True)
|
||||
|
||||
|
||||
def generate_erb_template(
|
||||
fmt: str,
|
||||
parsed: Any,
|
||||
role_prefix: str,
|
||||
*,
|
||||
original_text: str | None = None,
|
||||
loop_candidates: list[LoopCandidate] | None = None,
|
||||
flat_items: list[tuple[tuple[str, ...], Any]] | None = None,
|
||||
puppet_class: str | None = None,
|
||||
) -> str:
|
||||
"""Generate a Puppet ERB template from JinjaTurtle's renderer-neutral data.
|
||||
|
||||
The first implementation intentionally translates the Jinja2 subset emitted
|
||||
by JinjaTurtle's existing format handlers. This keeps parsing, formatting
|
||||
preservation, and loop detection identical between Jinja2 and ERB output
|
||||
while still producing Puppet-native ``@parameter`` references.
|
||||
"""
|
||||
|
||||
jinja_template = generate_jinja2_template(
|
||||
fmt,
|
||||
parsed,
|
||||
role_prefix,
|
||||
original_text=original_text,
|
||||
loop_candidates=loop_candidates,
|
||||
)
|
||||
names = _template_variable_names(role_prefix, flat_items or [], loop_candidates)
|
||||
erb_template = translate_jinja2_to_erb(
|
||||
jinja_template,
|
||||
role_prefix=role_prefix,
|
||||
puppet_class=puppet_class or role_prefix,
|
||||
variable_names=names,
|
||||
)
|
||||
# Defence in depth: no live Jinja2 delimiter may survive translation.
|
||||
verify_erb_template_safe(erb_template)
|
||||
return erb_template
|
||||
|
||||
|
||||
def _stringify_timestamps(obj: Any) -> Any:
|
||||
"""
|
||||
Recursively walk a parsed config and turn any datetime/date/time objects
|
||||
|
|
|
|||
|
|
@ -1,243 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from .escape import escape_erb_literal
|
||||
|
||||
|
||||
def _safe_name(raw: str, *, fallback: str = "var") -> str:
|
||||
text = re.sub(r"[^A-Za-z0-9_]+", "_", str(raw or fallback)).strip("_").lower()
|
||||
text = re.sub(r"_+", "_", text)
|
||||
if not text:
|
||||
text = fallback
|
||||
if not re.match(r"^[a-z_]", text):
|
||||
text = f"{fallback}_{text}"
|
||||
return text
|
||||
|
||||
|
||||
def puppet_class_name(raw: str) -> str:
|
||||
"""Return a conservative Puppet class/Hiera namespace name."""
|
||||
|
||||
text = _safe_name(raw, fallback="jinjaturtle")
|
||||
if not re.match(r"^[a-z]", text):
|
||||
text = f"jinjaturtle_{text}"
|
||||
return text
|
||||
|
||||
|
||||
def _role_prefix_name(raw: str) -> str:
|
||||
return _safe_name(raw, fallback="jinjaturtle")
|
||||
|
||||
|
||||
def puppet_local_var_name(
|
||||
role_prefix: str,
|
||||
jinja_var_name: str,
|
||||
*,
|
||||
puppet_class: str | None = None,
|
||||
) -> str:
|
||||
"""Map a generated JinjaTurtle variable to a Puppet class parameter.
|
||||
|
||||
For the common case where ``--role-name php`` also means class ``php``, a
|
||||
generated Jinja variable such as ``php_memory_limit`` becomes Puppet local
|
||||
parameter ``memory_limit`` and Hiera key ``php::memory_limit``.
|
||||
|
||||
When ``puppet_class`` differs from ``role_prefix`` we keep the full
|
||||
generated variable name as the local parameter and only use ``puppet_class``
|
||||
as the Hiera namespace.
|
||||
"""
|
||||
|
||||
var_name = _safe_name(jinja_var_name, fallback="value")
|
||||
prefix = _role_prefix_name(role_prefix)
|
||||
klass = puppet_class_name(puppet_class or role_prefix)
|
||||
if klass == prefix and var_name.startswith(prefix + "_"):
|
||||
stripped = var_name[len(prefix) + 1 :]
|
||||
return stripped or var_name
|
||||
return var_name
|
||||
|
||||
|
||||
class ErbTranslator:
|
||||
"""Translate the Jinja2 subset emitted by JinjaTurtle into Puppet ERB."""
|
||||
|
||||
_TOKEN_RE = re.compile(r"({{.*?}}|{%.*?%})", re.S)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
role_prefix: str,
|
||||
puppet_class: str | None = None,
|
||||
variable_names: set[str] | None = None,
|
||||
) -> None:
|
||||
self.role_prefix = role_prefix
|
||||
self.puppet_class = puppet_class or role_prefix
|
||||
self.variable_names = set(variable_names or set())
|
||||
self.loop_stack: list[tuple[str, str, str]] = []
|
||||
self.needs_json = False
|
||||
|
||||
# Matches a JinjaTurtle ``{% raw %} ... {% endraw %}`` block (non-greedy).
|
||||
# JinjaTurtle emits raw blocks only to carry verbatim, security-escaped
|
||||
# source text (comments and unrecognised lines), so the *contents* must be
|
||||
# treated as literal output, never translated as Jinja tokens.
|
||||
_RAW_BLOCK_RE = re.compile(
|
||||
r"{%[-+]?\s*raw\s*[-+]?%}(.*?){%[-+]?\s*endraw\s*[-+]?%}", re.S
|
||||
)
|
||||
|
||||
def translate(self, template_text: str) -> str:
|
||||
# Split out raw blocks first. Their inner text is literal and must be
|
||||
# carried through as literal ERB (with ERB delimiters re-escaped), rather
|
||||
# than tokenised -- otherwise an escaped Jinja payload inside a comment
|
||||
# would be "re-animated" into live ERB during translation.
|
||||
segments = self._RAW_BLOCK_RE.split(template_text)
|
||||
out: list[str] = []
|
||||
# re.split with one capture group yields: [text, raw_inner, text, ...].
|
||||
for idx, segment in enumerate(segments):
|
||||
if idx % 2 == 1:
|
||||
# Captured raw-block contents: emit as literal ERB text.
|
||||
out.append(escape_erb_literal(segment))
|
||||
else:
|
||||
out.append(self._translate_tokens(segment))
|
||||
|
||||
rendered = "".join(out)
|
||||
if self.needs_json and "require 'json'" not in rendered:
|
||||
rendered = "<% require 'json' -%>\n" + rendered
|
||||
return rendered
|
||||
|
||||
def _translate_tokens(self, template_text: str) -> str:
|
||||
parts = self._TOKEN_RE.split(template_text)
|
||||
out: list[str] = []
|
||||
for token in parts:
|
||||
if not token:
|
||||
continue
|
||||
if token.startswith("{{") and token.endswith("}}"):
|
||||
expr = token[2:-2].strip()
|
||||
out.append(f"<%= {self.expr_to_ruby(expr)} %>")
|
||||
continue
|
||||
if token.startswith("{%") and token.endswith("%}"):
|
||||
stmt = token[2:-2].strip()
|
||||
out.append(self.statement_to_erb(stmt))
|
||||
continue
|
||||
out.append(token)
|
||||
return "".join(out)
|
||||
|
||||
def local_var(self, name: str) -> str:
|
||||
return puppet_local_var_name(
|
||||
self.role_prefix, name, puppet_class=self.puppet_class
|
||||
)
|
||||
|
||||
def ruby_value(self, expr: str) -> str:
|
||||
expr = expr.strip()
|
||||
if expr in {"true", "True"}:
|
||||
return "true"
|
||||
if expr in {"false", "False"}:
|
||||
return "false"
|
||||
if expr in {"none", "None", "null"}:
|
||||
return "nil"
|
||||
if re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", expr):
|
||||
if any(expr == loop_var for loop_var, _idx, _coll in self.loop_stack):
|
||||
return expr
|
||||
return f"@{self.local_var(expr)}"
|
||||
m = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)\.([A-Za-z_][A-Za-z0-9_]*)$", expr)
|
||||
if m:
|
||||
base, key = m.groups()
|
||||
if any(base == loop_var for loop_var, _idx, _coll in self.loop_stack):
|
||||
return f"{base}[{key!r}]"
|
||||
return f"@{self.local_var(base)}[{key!r}]"
|
||||
return expr
|
||||
|
||||
def expr_to_ruby(self, expr: str) -> str:
|
||||
expr = expr.strip()
|
||||
|
||||
# JinjaTurtle emits these YAML-preserving ternaries for booleans/nulls.
|
||||
m = re.match(
|
||||
r"^(['\"])(true|false)\1\s+if\s+([A-Za-z_][A-Za-z0-9_\.]*)\s+else\s+(['\"])(true|false)\4$",
|
||||
expr,
|
||||
)
|
||||
if m:
|
||||
truthy = m.group(2)
|
||||
cond = self.ruby_value(m.group(3))
|
||||
falsy = m.group(5)
|
||||
return f"{cond} ? {truthy!r} : {falsy!r}"
|
||||
|
||||
m = re.match(
|
||||
r"^(['\"])(null)\1\s+if\s+([A-Za-z_][A-Za-z0-9_\.]*)\s+is\s+none\s+else\s+([A-Za-z_][A-Za-z0-9_\.]*)$",
|
||||
expr,
|
||||
)
|
||||
if m:
|
||||
value = self.ruby_value(m.group(3))
|
||||
fallback = self.ruby_value(m.group(4))
|
||||
return f"{value}.nil? ? 'null' : {fallback}"
|
||||
|
||||
if "|" in expr:
|
||||
base, *filters = [part.strip() for part in expr.split("|")]
|
||||
ruby = self.ruby_value(base)
|
||||
for filt in filters:
|
||||
if filt.startswith("lower"):
|
||||
ruby = f"{ruby}.to_s.downcase"
|
||||
elif filt.startswith("to_json") or filt.startswith("tojson"):
|
||||
self.needs_json = True
|
||||
if "indent" in filt:
|
||||
ruby = f"JSON.pretty_generate({ruby})"
|
||||
else:
|
||||
ruby = f"JSON.generate({ruby})"
|
||||
return ruby
|
||||
|
||||
return self.ruby_value(expr)
|
||||
|
||||
def statement_to_erb(self, stmt: str) -> str:
|
||||
if stmt.endswith(("-", "+")):
|
||||
stmt = stmt[:-1].rstrip()
|
||||
|
||||
if stmt.startswith("for "):
|
||||
m = re.match(
|
||||
r"^for\s+([A-Za-z_][A-Za-z0-9_]*)\s+in\s+([A-Za-z_][A-Za-z0-9_]*)$",
|
||||
stmt,
|
||||
)
|
||||
if m:
|
||||
loop_var, collection = m.groups()
|
||||
idx_var = f"__jt_idx_{len(self.loop_stack)}"
|
||||
collection_ruby = self.ruby_value(collection)
|
||||
self.loop_stack.append((loop_var, idx_var, collection_ruby))
|
||||
return f"<% {collection_ruby}.each_with_index do |{loop_var}, {idx_var}| -%>"
|
||||
|
||||
if stmt == "endfor":
|
||||
if self.loop_stack:
|
||||
self.loop_stack.pop()
|
||||
return "<% end %>"
|
||||
|
||||
if stmt.startswith("if "):
|
||||
cond = stmt[3:].strip()
|
||||
if cond == "not loop.last" and self.loop_stack:
|
||||
_loop_var, idx_var, collection_ruby = self.loop_stack[-1]
|
||||
return f"<% if {idx_var} < ({collection_ruby}.length - 1) -%>"
|
||||
m = re.match(r"^([A-Za-z_][A-Za-z0-9_\.]*)\s+is\s+defined$", cond)
|
||||
if m:
|
||||
return f"<% unless {self.ruby_value(m.group(1))}.nil? -%>"
|
||||
m = re.match(r"^([A-Za-z_][A-Za-z0-9_\.]*)\s+is\s+none$", cond)
|
||||
if m:
|
||||
return f"<% if {self.ruby_value(m.group(1))}.nil? -%>"
|
||||
return f"<% if {self.expr_to_ruby(cond)} -%>"
|
||||
|
||||
if stmt == "else":
|
||||
return "<% else -%>"
|
||||
|
||||
if stmt.startswith("elif "):
|
||||
return f"<% elsif {self.expr_to_ruby(stmt[5:].strip())} -%>"
|
||||
|
||||
if stmt == "endif":
|
||||
return "<% end -%>"
|
||||
|
||||
# Preserve unknown Jinja statements visibly as an ERB comment so the
|
||||
# generated template does not contain invalid Jinja syntax.
|
||||
return f"<%# Unsupported JinjaTurtle statement: {stmt} %>"
|
||||
|
||||
|
||||
def translate_jinja2_to_erb(
|
||||
template_text: str,
|
||||
*,
|
||||
role_prefix: str,
|
||||
puppet_class: str | None = None,
|
||||
variable_names: set[str] | None = None,
|
||||
) -> str:
|
||||
return ErbTranslator(
|
||||
role_prefix=role_prefix,
|
||||
puppet_class=puppet_class,
|
||||
variable_names=variable_names,
|
||||
).translate(template_text)
|
||||
|
|
@ -9,7 +9,7 @@ always replaced with ``{{ var }}`` placeholders and parked in the defaults data,
|
|||
so a payload inside a value is inert. Verbatim text is different: if the source
|
||||
contains ``{{ ... }}``, ``{% ... %}`` or ``{# ... #}`` (Jinja2), or ``<%= %>`` /
|
||||
``<% %>`` (ERB), that text becomes *live template code* in the output and is
|
||||
executed when Salt/Ansible/Puppet later renders the template.
|
||||
executed when Ansible later renders the template.
|
||||
|
||||
Because JinjaTurtle is frequently fed harvested, attacker-influenceable config
|
||||
(hostnames, banners, GECOS-derived comments, "Managed by" notes), this is a
|
||||
|
|
@ -30,14 +30,6 @@ Design notes:
|
|||
Jinja construct. The only way to break out of a raw block is a literal
|
||||
``{% endraw %}`` in the source, so we defang the token ``endraw`` (in any
|
||||
internal spacing) before wrapping.
|
||||
* The ERB translator (``erb.py``) is raw-aware: it copies the *contents* of a
|
||||
JinjaTurtle raw block through as literal ERB text and re-escapes any ERB
|
||||
delimiters found there. That keeps a Jinja-escaped comment inert after the
|
||||
Jinja2 -> ERB translation step, instead of the payload being "re-animated"
|
||||
as ERB.
|
||||
* ``escape_erb_literal`` exists for completeness / direct ERB emission: it
|
||||
rewrites each ERB delimiter into an ERB expression that prints the delimiter
|
||||
characters literally.
|
||||
"""
|
||||
|
||||
import re
|
||||
|
|
@ -70,11 +62,6 @@ def contains_jinja_markup(text: str) -> bool:
|
|||
return any(m in text for m in _JINJA_MARKERS)
|
||||
|
||||
|
||||
def contains_erb_markup(text: str) -> bool:
|
||||
"""Return True if *text* contains any ERB delimiter."""
|
||||
return any(m in text for m in (*_ERB_OPEN_MARKERS, *_ERB_CLOSE_MARKERS))
|
||||
|
||||
|
||||
def _defang_endraw(text: str) -> str:
|
||||
"""Rewrite any literal ``{% endraw %}`` so it cannot close our raw wrapper.
|
||||
|
||||
|
|
@ -109,45 +96,3 @@ def escape_jinja_literal(text: str) -> str:
|
|||
if not text or not contains_jinja_markup(text):
|
||||
return text
|
||||
return "{% raw %}" + _defang_endraw(text) + "{% endraw %}"
|
||||
|
||||
|
||||
def escape_erb_literal(text: str) -> str:
|
||||
"""Make *text* render as literal characters under a later ERB render.
|
||||
|
||||
ERB has no ``raw`` block, so each opening/closing delimiter is rewritten as
|
||||
an ERB expression that prints the delimiter literally. Text with no ERB
|
||||
metacharacters is returned unchanged.
|
||||
"""
|
||||
if not text or not contains_erb_markup(text):
|
||||
return text
|
||||
|
||||
result: list[str] = []
|
||||
i = 0
|
||||
n = len(text)
|
||||
while i < n:
|
||||
matched = None
|
||||
for marker in (*_ERB_CLOSE_MARKERS, *_ERB_OPEN_MARKERS):
|
||||
if text.startswith(marker, i):
|
||||
matched = marker
|
||||
break
|
||||
if matched is not None:
|
||||
escaped = matched.replace("\\", "\\\\").replace('"', '\\"')
|
||||
result.append('<%= "' + escaped + '" %>')
|
||||
i += len(matched)
|
||||
else:
|
||||
result.append(text[i])
|
||||
i += 1
|
||||
return "".join(result)
|
||||
|
||||
|
||||
def escape_literal(text: str, *, engine: str = "jinja2") -> str:
|
||||
"""Escape verbatim *text* for the target template *engine*.
|
||||
|
||||
``engine`` is ``"jinja2"`` (default) or ``"erb"``. Unknown engines fall back
|
||||
to Jinja2 escaping. In JinjaTurtle's pipeline ERB output is produced by
|
||||
translating Jinja2 output, and the translator is raw-aware, so handlers can
|
||||
always Jinja-escape and rely on the translator to keep the literal inert.
|
||||
"""
|
||||
if engine == "erb":
|
||||
return escape_erb_literal(text)
|
||||
return escape_jinja_literal(text)
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ from typing import Any
|
|||
|
||||
from . import DictLikeHandler
|
||||
from .. import j2
|
||||
from ..escape import escape_jinja_literal
|
||||
from ..loop_analyzer import LoopCandidate
|
||||
|
||||
|
||||
|
|
@ -109,10 +110,18 @@ class JsonHandler(DictLikeHandler):
|
|||
chunks: list[str] = []
|
||||
pos = 0
|
||||
for path, start, end in spans:
|
||||
chunks.append(text[pos:start])
|
||||
# Text between scalar values (object keys, structural punctuation,
|
||||
# whitespace, and any comment-like trailing text) is copied verbatim
|
||||
# from the source file. Like every other text-emitting handler, this
|
||||
# verbatim text must be neutralised: if it contains Jinja2 markup it
|
||||
# would otherwise become live template code at apply time. The value
|
||||
# itself is replaced with a safe placeholder below. ``escape_jinja_literal``
|
||||
# is a no-op on text without Jinja markers, so benign JSON is unchanged
|
||||
# byte-for-byte and a later render reproduces the original characters.
|
||||
chunks.append(escape_jinja_literal(text[pos:start]))
|
||||
chunks.append(self._json_value_expr(self.make_var_name(role_prefix, path)))
|
||||
pos = end
|
||||
chunks.append(text[pos:])
|
||||
chunks.append(escape_jinja_literal(text[pos:]))
|
||||
return "".join(chunks)
|
||||
|
||||
def _collect_json_scalar_spans(
|
||||
|
|
@ -213,7 +222,12 @@ class JsonHandler(DictLikeHandler):
|
|||
|
||||
def _walk(obj: Any, path: tuple[str, ...] = ()) -> Any:
|
||||
if isinstance(obj, dict):
|
||||
return {k: _walk(v, path + (str(k),)) for k, v in obj.items()}
|
||||
# Keys are emitted verbatim into the template, so neutralise any
|
||||
# Jinja markup in them (see _generate_json_template_from_text).
|
||||
return {
|
||||
escape_jinja_literal(str(k)): _walk(v, path + (str(k),))
|
||||
for k, v in obj.items()
|
||||
}
|
||||
if isinstance(obj, list):
|
||||
return [_walk(v, path + (str(i),)) for i, v in enumerate(obj)]
|
||||
# scalar - use marker that will be replaced with to_json
|
||||
|
|
@ -261,7 +275,12 @@ class JsonHandler(DictLikeHandler):
|
|||
return f"__LOOP_DICT__{collection_var}__{item_var}__"
|
||||
|
||||
if isinstance(obj, dict):
|
||||
return {k: _walk(v, current_path + (str(k),)) for k, v in obj.items()}
|
||||
# Keys are emitted verbatim into the template, so neutralise any
|
||||
# Jinja markup in them (see _generate_json_template_from_text).
|
||||
return {
|
||||
escape_jinja_literal(str(k)): _walk(v, current_path + (str(k),))
|
||||
for k, v in obj.items()
|
||||
}
|
||||
if isinstance(obj, list):
|
||||
# Check if this list is a loop candidate
|
||||
if current_path in loop_paths:
|
||||
|
|
@ -364,8 +383,13 @@ class JsonHandler(DictLikeHandler):
|
|||
] # first line has no indent; we prepend `inner` when emitting
|
||||
for i, key in enumerate(keys):
|
||||
comma = "," if i < len(keys) - 1 else ""
|
||||
# The literal key text is emitted verbatim into the template; escape any
|
||||
# Jinja markup in it. The value side ({item_var}.{key}) is constrained by
|
||||
# the output safety gate's dotted-name allowlist, which fails closed on
|
||||
# anything that is not a plain identifier path.
|
||||
dict_lines.append(
|
||||
f'{field}"{key}": ' f"{j2.to_json(f'{item_var}.{key}')}{comma}"
|
||||
f'{field}"{escape_jinja_literal(str(key))}": '
|
||||
f"{j2.to_json(f'{item_var}.{key}')}{comma}"
|
||||
)
|
||||
# Comma between *items* goes after the closing brace.
|
||||
dict_lines.append(f"{inner}}}{j2.if_not_loop_last()},{j2.endif()}")
|
||||
|
|
|
|||
|
|
@ -3,7 +3,8 @@ from __future__ import annotations
|
|||
from collections import Counter, defaultdict
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
import xml.etree.ElementTree as ET # nosec
|
||||
import xml.etree.ElementTree as ET # nosec B405 - safe trees only; parsing uses defusedxml
|
||||
import defusedxml.ElementTree as DET
|
||||
|
||||
from .base import BaseHandler
|
||||
from .. import j2
|
||||
|
|
@ -20,11 +21,11 @@ class XmlHandler(BaseHandler):
|
|||
|
||||
def parse(self, path: Path) -> ET.Element:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
parser = ET.XMLParser(
|
||||
target=ET.TreeBuilder(insert_comments=False)
|
||||
) # nosec B314
|
||||
parser.feed(text)
|
||||
root = parser.close()
|
||||
# Security must live in the handler, not only in the CLI entry point:
|
||||
# callers may import JinjaTurtle as a library and invoke parse_config()
|
||||
# directly. defusedxml rejects DTD/entity abuse and also discards
|
||||
# comments by default, matching the previous TreeBuilder behaviour.
|
||||
root = DET.fromstring(text)
|
||||
return root
|
||||
|
||||
def flatten(self, parsed: Any) -> list[tuple[tuple[str, ...], Any]]:
|
||||
|
|
|
|||
|
|
@ -22,7 +22,9 @@ Notes:
|
|||
|
||||
from collections import Counter, defaultdict
|
||||
from copy import deepcopy
|
||||
import os
|
||||
import configparser
|
||||
import stat
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
|
|
@ -31,7 +33,7 @@ import xml.etree.ElementTree as ET # nosec
|
|||
from . import j2
|
||||
from .core import dump_yaml, flatten_config, make_var_name, parse_config
|
||||
from .handlers.xml import XmlHandler
|
||||
from .safety import verify_jinja2_template_safe
|
||||
from .safety import verify_jinja2_template_safe, verify_no_live_jinja_in_json_keys
|
||||
from .escape import escape_jinja_literal
|
||||
|
||||
|
||||
|
|
@ -44,8 +46,23 @@ SUPPORTED_SUFFIXES: dict[str, set[str]] = {
|
|||
}
|
||||
|
||||
|
||||
def _lstat(path: Path) -> os.stat_result:
|
||||
return path.lstat()
|
||||
|
||||
|
||||
def is_supported_file(path: Path) -> bool:
|
||||
if not path.is_file():
|
||||
"""Return True only for real regular files with supported suffixes.
|
||||
|
||||
pathlib.Path.is_file() follows symlinks. Folder mode must not follow
|
||||
attacker-controlled symlinks when run over an untrusted tree, especially if
|
||||
an administrator accidentally runs the CLI as root.
|
||||
"""
|
||||
|
||||
try:
|
||||
st = _lstat(path)
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
if not stat.S_ISREG(st.st_mode):
|
||||
return False
|
||||
suffix = path.suffix.lower()
|
||||
for exts in SUPPORTED_SUFFIXES.values():
|
||||
|
|
@ -55,11 +72,16 @@ def is_supported_file(path: Path) -> bool:
|
|||
|
||||
|
||||
def iter_supported_files(root: Path, recursive: bool) -> list[Path]:
|
||||
if not root.exists():
|
||||
try:
|
||||
st = _lstat(root)
|
||||
except FileNotFoundError:
|
||||
raise FileNotFoundError(str(root))
|
||||
if root.is_file():
|
||||
|
||||
if stat.S_ISLNK(st.st_mode):
|
||||
raise ValueError(f"refusing to follow symlink: {root}")
|
||||
if stat.S_ISREG(st.st_mode):
|
||||
return [root] if is_supported_file(root) else []
|
||||
if not root.is_dir():
|
||||
if not stat.S_ISDIR(st.st_mode):
|
||||
return []
|
||||
|
||||
it = root.rglob("*") if recursive else root.glob("*")
|
||||
|
|
@ -784,5 +806,7 @@ def process_directory(
|
|||
# Any un-neutralised source text that became a live tag aborts generation.
|
||||
for out in outputs:
|
||||
verify_jinja2_template_safe(out.template)
|
||||
if out.fmt == "json":
|
||||
verify_no_live_jinja_in_json_keys(out.template)
|
||||
|
||||
return defaults_yaml, outputs
|
||||
|
|
|
|||
124
src/jinjaturtle/output_safety.py
Normal file
124
src/jinjaturtle/output_safety.py
Normal file
|
|
@ -0,0 +1,124 @@
|
|||
from __future__ import annotations
|
||||
|
||||
"""Safer file-output helpers for the JinjaTurtle CLI.
|
||||
|
||||
The CLI is often used by administrators. A plain Path.write_text() follows a
|
||||
final-path symlink and can therefore be dangerous when a root-run invocation
|
||||
writes into an attacker-writable tree. These helpers validate path components,
|
||||
write through a private temporary file in the target directory, and replace the
|
||||
final path atomically. Existing final-path symlinks are refused rather than
|
||||
followed.
|
||||
"""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
import stat
|
||||
import tempfile
|
||||
|
||||
|
||||
class OutputPathError(OSError):
|
||||
"""Raised when a requested output path is unsafe."""
|
||||
|
||||
|
||||
def _absolute(path: Path) -> Path:
|
||||
return path if path.is_absolute() else Path.cwd() / path
|
||||
|
||||
|
||||
def _check_existing_path_not_symlink(path: Path) -> None:
|
||||
try:
|
||||
st = path.lstat()
|
||||
except FileNotFoundError:
|
||||
return
|
||||
if stat.S_ISLNK(st.st_mode):
|
||||
raise OutputPathError(f"refusing to use symlink path: {path}")
|
||||
|
||||
|
||||
def _check_existing_output_file(path: Path) -> None:
|
||||
try:
|
||||
st = path.lstat()
|
||||
except FileNotFoundError:
|
||||
return
|
||||
if stat.S_ISLNK(st.st_mode):
|
||||
raise OutputPathError(f"refusing to write through symlink: {path}")
|
||||
if not stat.S_ISREG(st.st_mode):
|
||||
raise OutputPathError(f"refusing to replace non-regular file: {path}")
|
||||
|
||||
|
||||
def _check_parent_components(parent: Path) -> None:
|
||||
"""Require every existing parent component to be a real directory."""
|
||||
|
||||
parent = _absolute(parent)
|
||||
parts = parent.parts
|
||||
if not parts:
|
||||
return
|
||||
|
||||
cur = Path(parts[0])
|
||||
for part in parts[1:]:
|
||||
cur = cur / part
|
||||
try:
|
||||
st = cur.lstat()
|
||||
except FileNotFoundError as exc:
|
||||
raise OutputPathError(f"output parent does not exist: {cur}") from exc
|
||||
if stat.S_ISLNK(st.st_mode):
|
||||
raise OutputPathError(f"refusing to use symlink parent: {cur}")
|
||||
if not stat.S_ISDIR(st.st_mode):
|
||||
raise OutputPathError(f"output parent is not a directory: {cur}")
|
||||
|
||||
|
||||
def ensure_safe_directory(path: Path) -> None:
|
||||
"""Create or validate a directory tree without accepting symlinks."""
|
||||
|
||||
path = _absolute(path)
|
||||
parts = path.parts
|
||||
if not parts:
|
||||
return
|
||||
|
||||
cur = Path(parts[0])
|
||||
for part in parts[1:]:
|
||||
cur = cur / part
|
||||
try:
|
||||
st = cur.lstat()
|
||||
except FileNotFoundError:
|
||||
cur.mkdir(mode=0o700)
|
||||
st = cur.lstat()
|
||||
if stat.S_ISLNK(st.st_mode):
|
||||
raise OutputPathError(f"refusing to use symlink directory: {cur}")
|
||||
if not stat.S_ISDIR(st.st_mode):
|
||||
raise OutputPathError(f"output path is not a directory: {cur}")
|
||||
|
||||
|
||||
def write_text_safely(path: Path, text: str, *, encoding: str = "utf-8") -> None:
|
||||
"""Write text without following a final-path symlink.
|
||||
|
||||
The target's parent must already exist and every parent component must be a
|
||||
real directory. The write is completed with os.replace(), which atomically
|
||||
swaps the final directory entry and does not dereference a final symlink.
|
||||
"""
|
||||
|
||||
path = _absolute(path)
|
||||
_check_parent_components(path.parent)
|
||||
_check_existing_output_file(path)
|
||||
|
||||
fd = -1
|
||||
tmp_name: str | None = None
|
||||
try:
|
||||
fd, tmp_name = tempfile.mkstemp(
|
||||
prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent), text=True
|
||||
)
|
||||
with os.fdopen(fd, "w", encoding=encoding) as f:
|
||||
fd = -1
|
||||
f.write(text)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
os.chmod(tmp_name, 0o600)
|
||||
_check_existing_output_file(path)
|
||||
os.replace(tmp_name, path)
|
||||
tmp_name = None
|
||||
finally:
|
||||
if fd >= 0:
|
||||
os.close(fd)
|
||||
if tmp_name is not None:
|
||||
try:
|
||||
os.unlink(tmp_name)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
|
@ -22,8 +22,8 @@ single global property:
|
|||
|
||||
The check is positive/allowlist-based, which is the safe direction: unknown
|
||||
constructs are rejected, not ignored. It runs at the single choke points in
|
||||
``core.py`` (``generate_jinja2_template`` / ``generate_erb_template``), so it
|
||||
covers every current handler and every future one automatically.
|
||||
``core.py`` (``generate_jinja2_template``) so it covers every current handler and
|
||||
every future one automatically.
|
||||
|
||||
Why this is robust against the escaper being wrong
|
||||
---------------------------------------------------
|
||||
|
|
@ -43,6 +43,7 @@ __all__ = [
|
|||
"TemplateSafetyError",
|
||||
"verify_jinja2_template_safe",
|
||||
"verify_erb_template_safe",
|
||||
"verify_no_live_jinja_in_json_keys",
|
||||
]
|
||||
|
||||
|
||||
|
|
@ -208,7 +209,88 @@ def verify_jinja2_template_safe(template_text: str) -> None:
|
|||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# ERB gate.
|
||||
# JSON-key gate (defence in depth, format-specific).
|
||||
#
|
||||
# JinjaTurtle never emits a live Jinja construct inside a JSON *object key*: keys
|
||||
# are copied verbatim from the source and (after escape.py) are wrapped in
|
||||
# ``{% raw %}`` if they contain markup, so a key never lexes as a live tag. A
|
||||
# live construct in key position can therefore only mean source key text leaked
|
||||
# into the template unescaped (the json-handler blind spot). This gate is
|
||||
# independent of the escaper: it inspects the finished template, replaces every
|
||||
# *live* Jinja construct with an inert sentinel (raw-wrapped literal text stays
|
||||
# literal), and rejects any sentinel that lands in a JSON key slot.
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
# Sentinel byte that cannot occur in normal generated template text.
|
||||
_LIVE_SENTINEL = "\x00"
|
||||
|
||||
# A JSON key is a double-quoted string immediately followed (after optional
|
||||
# whitespace) by a colon. We only need to detect a sentinel *inside* such a
|
||||
# string, so match a quoted run that ends in `":` and look for the sentinel.
|
||||
_JSON_KEY_RE = re.compile(r'"((?:[^"\\]|\\.)*)"\s*:', re.S)
|
||||
|
||||
_KEY_JINJA_DELIMS = ("{{", "}}", "{%", "%}", "{#", "#}")
|
||||
|
||||
|
||||
def _contains_jinja_delim(text: str) -> bool:
|
||||
"""True if *text* contains any Jinja delimiter (live or escaped-literal)."""
|
||||
return any(d in text for d in _KEY_JINJA_DELIMS)
|
||||
|
||||
|
||||
def verify_no_live_jinja_in_json_keys(template_text: str) -> None:
|
||||
"""Reject a JSON template that carries Jinja markup in an object key.
|
||||
|
||||
JinjaTurtle never templates a JSON *object key*: keys come straight from the
|
||||
source and a key is an identifier/string, never a value placeholder. Any
|
||||
Jinja in key position therefore means attacker-influenced source key text
|
||||
reached the template. This gate fails closed on it, independent of whether a
|
||||
handler left the markup *live* (a raw ``{{ ... }}`` in the key) or *escaped*
|
||||
it into a ``{% raw %}`` wrapper -- both indicate a key that should never have
|
||||
contained templating, so generation aborts rather than emitting it.
|
||||
|
||||
Detection is done on the lexer token stream: we reconstruct the document with
|
||||
every live tag collapsed to a sentinel and every ``{% raw %}``-wrapped region
|
||||
also marked, then reject a sentinel that lands inside a JSON key string.
|
||||
"""
|
||||
import jinja2
|
||||
|
||||
env = jinja2.Environment(autoescape=True)
|
||||
try:
|
||||
tokens = list(env.lex(template_text))
|
||||
except jinja2.TemplateSyntaxError as exc:
|
||||
raise TemplateSafetyError(
|
||||
f"generated JSON template does not lex as the JinjaTurtle subset: {exc}"
|
||||
) from exc
|
||||
|
||||
out: list[str] = []
|
||||
in_tag = False
|
||||
for _lineno, tok_type, value in tokens:
|
||||
if tok_type in ("variable_begin", "block_begin", "comment_begin"):
|
||||
# A live construct: collapse to a sentinel so it is detectable if it
|
||||
# sits in key position. (raw_begin/raw_end are *not* live; the data
|
||||
# inside a raw block is preserved verbatim below, so an escaped key
|
||||
# still shows its literal Jinja delimiters to the key check.)
|
||||
in_tag = True
|
||||
out.append(_LIVE_SENTINEL)
|
||||
elif tok_type in ("variable_end", "block_end", "comment_end"):
|
||||
in_tag = False
|
||||
elif not in_tag:
|
||||
# data, whitespace, raw_begin/raw_end markers, and inert raw content.
|
||||
out.append(value if isinstance(value, str) else str(value))
|
||||
|
||||
reconstructed = "".join(out)
|
||||
|
||||
for match in _JSON_KEY_RE.finditer(reconstructed):
|
||||
key_text = match.group(1)
|
||||
if _LIVE_SENTINEL in key_text or _contains_jinja_delim(key_text):
|
||||
raise TemplateSafetyError(
|
||||
"refusing to emit JSON template: Jinja markup appears inside a "
|
||||
"JSON object key. JinjaTurtle never templates keys, so this "
|
||||
"indicates attacker-influenced source key text (possible "
|
||||
"template injection)."
|
||||
)
|
||||
|
||||
|
||||
#
|
||||
# JinjaTurtle's ERB output is produced by translating the (already-verified)
|
||||
# Jinja2 subset, so the Jinja2 gate is the primary guarantee. As an independent
|
||||
|
|
|
|||
|
|
@ -132,57 +132,3 @@ def test_cli_folder_single_output_file_when_one_format(tmp_path):
|
|||
assert defaults_path.is_file()
|
||||
assert template_path.is_file()
|
||||
assert "to_json" in template_path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_cli_erb_outputs_puppet_hiera_and_erb_template(tmp_path):
|
||||
cfg = tmp_path / "app.ini"
|
||||
cfg.write_text("[main]\nport = 8080\n", encoding="utf-8")
|
||||
data_out = tmp_path / "common.yaml"
|
||||
template_out = tmp_path / "app.ini.erb"
|
||||
|
||||
exit_code = cli._main(
|
||||
[
|
||||
str(cfg),
|
||||
"-r",
|
||||
"php",
|
||||
"--template-engine",
|
||||
"erb",
|
||||
"--defaults-output",
|
||||
str(data_out),
|
||||
"--template-output",
|
||||
str(template_out),
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
assert "php::main_port: '8080'" in data_out.read_text(encoding="utf-8")
|
||||
assert "port = <%= @main_port %>" in template_out.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_cli_erb_can_use_separate_puppet_class_namespace(tmp_path):
|
||||
cfg = tmp_path / "app.ini"
|
||||
cfg.write_text("[main]\nport = 8080\n", encoding="utf-8")
|
||||
data_out = tmp_path / "node.yaml"
|
||||
template_out = tmp_path / "app.ini.erb"
|
||||
|
||||
exit_code = cli._main(
|
||||
[
|
||||
str(cfg),
|
||||
"-r",
|
||||
"php_etc_app_ini",
|
||||
"--template-engine",
|
||||
"erb",
|
||||
"--puppet-class",
|
||||
"php",
|
||||
"--defaults-output",
|
||||
str(data_out),
|
||||
"--template-output",
|
||||
str(template_out),
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
data = data_out.read_text(encoding="utf-8")
|
||||
template = template_out.read_text(encoding="utf-8")
|
||||
assert "php::php_etc_app_ini_main_port: '8080'" in data
|
||||
assert "port = <%= @php_etc_app_ini_main_port %>" in template
|
||||
|
|
|
|||
|
|
@ -3,8 +3,8 @@
|
|||
JinjaTurtle copies parts of the source config (comments, unrecognised lines,
|
||||
structural keys) verbatim into the generated template. If that text contains
|
||||
Jinja2/ERB delimiters it must be neutralised, otherwise attacker-influenced
|
||||
config content becomes live template code that executes when Salt/Ansible/Puppet
|
||||
later renders the template.
|
||||
config content becomes live template code that executes when Ansible later
|
||||
renders the template.
|
||||
|
||||
These tests render the *generated* template the way a downstream tool would and
|
||||
assert that an injected payload never executes. A tripwire object is exposed
|
||||
|
|
@ -14,7 +14,6 @@ the tripwire sentinel, an injected expression executed and the test fails.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
|
@ -25,13 +24,26 @@ 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."""
|
||||
|
||||
|
|
@ -85,7 +97,7 @@ def _run_jinjaturtle(tmp_path: Path, source_name: str, body: str, fmt: str):
|
|||
text=True,
|
||||
)
|
||||
assert res.returncode == 0, f"generation failed: {res.stderr}"
|
||||
defaults = pyyaml.safe_load(dfl.read_text()) or {}
|
||||
defaults = _safe_load_defaults(dfl.read_text()) or {}
|
||||
return tpl.read_text(), defaults
|
||||
|
||||
|
||||
|
|
@ -155,6 +167,86 @@ def test_generated_template_is_renderable(tmp_path, fmt, name, body):
|
|||
_render_jinja(template_text, defaults)
|
||||
|
||||
|
||||
# --- JSON object-key injection ------------------------------------------------
|
||||
#
|
||||
# The JSON handler copies the text *between* scalar values (object keys,
|
||||
# punctuation) verbatim. A key is never a value placeholder, so Jinja markup in a
|
||||
# key can only come from attacker-influenced source text. The output gate
|
||||
# (verify_no_live_jinja_in_json_keys) must fail closed on it -- including the
|
||||
# "benign-looking name" form (e.g. ``{{ ansible_hostname }}``) that the generic
|
||||
# allowlist would otherwise accept as an ordinary variable reference, and which
|
||||
# could leak an in-scope variable's value into the rendered config at apply time.
|
||||
|
||||
JSON_KEY_INJECTION_BODIES = [
|
||||
# benign-looking variable reference (the residual bypass: passes the generic
|
||||
# allowlist but must still be rejected in *key* position)
|
||||
'{ "{{ ansible_hostname }}": "v" }',
|
||||
# dotted reference (e.g. dumping another host's vars)
|
||||
'{ "{{ hostvars.localhost }}": 1 }',
|
||||
# self-referencing a sibling-derived variable name
|
||||
'{ "{{ role_port }}": "x", "port": 8080 }',
|
||||
# classic gadget (already rejected historically; kept as a guard)
|
||||
'{ "{{ cycler.__init__.__globals__ }}": 1 }',
|
||||
# statement injection in a key
|
||||
'{ "{% for x in y %}k{% endfor %}": 1 }',
|
||||
# nested object key
|
||||
'{ "ok": { "{{ ansible_hostname }}": 2 } }',
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("body", JSON_KEY_INJECTION_BODIES)
|
||||
def test_json_key_injection_fails_closed_cli(tmp_path, body):
|
||||
src = tmp_path / "evil.json"
|
||||
src.write_text(body, encoding="utf-8")
|
||||
out = tmp_path / "out.j2"
|
||||
res = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-m",
|
||||
"jinjaturtle.cli",
|
||||
str(src),
|
||||
"-f",
|
||||
"json",
|
||||
"--role-name",
|
||||
"role",
|
||||
"-t",
|
||||
str(out),
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert res.returncode == 2, f"expected fail-closed, got rc={res.returncode}"
|
||||
assert "refusing to generate unsafe template" in res.stderr
|
||||
assert not out.exists(), "no template may be written when the gate refuses"
|
||||
|
||||
|
||||
def test_json_benign_keys_still_generate(tmp_path):
|
||||
src = tmp_path / "ok.json"
|
||||
src.write_text('{ "host": "localhost", "port": 8080 }', encoding="utf-8")
|
||||
out = tmp_path / "out.j2"
|
||||
res = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-m",
|
||||
"jinjaturtle.cli",
|
||||
str(src),
|
||||
"-f",
|
||||
"json",
|
||||
"--role-name",
|
||||
"demo",
|
||||
"-t",
|
||||
str(out),
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert res.returncode == 0, res.stderr
|
||||
template_text = out.read_text()
|
||||
# Keys stay literal; values become placeholders.
|
||||
assert '"host":' in template_text
|
||||
assert "demo_host" in template_text
|
||||
|
||||
|
||||
# --- Unit-level guarantees for the escaper itself ---------------------------
|
||||
|
||||
SSTI_PAYLOADS = [
|
||||
|
|
@ -233,43 +325,3 @@ def test_endraw_whitespace_control_cannot_break_out(endraw):
|
|||
rendered = env.from_string(escaped).render(boom=_Boom())
|
||||
assert TRIP not in rendered
|
||||
assert rendered == payload
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"payload",
|
||||
[
|
||||
"<%= system('id') %>",
|
||||
"<% require 'open3' %>",
|
||||
"text <%= 1+1 %> more",
|
||||
"-%> orphan <%",
|
||||
],
|
||||
)
|
||||
def test_escape_erb_literal_removes_executable_tags(payload):
|
||||
escaped = escape_erb_literal(payload)
|
||||
# No raw executable ERB tag should survive that contains the original code.
|
||||
live = re.findall(r"<%[-=#]?(.*?)-?%>", escaped, re.S)
|
||||
for chunk in live:
|
||||
assert "system" not in chunk
|
||||
assert "require" not in chunk
|
||||
# Numeric/expression payloads must be reduced to literal-string prints.
|
||||
|
||||
|
||||
def test_escape_literal_dispatches_by_engine():
|
||||
"""The public ``escape_literal`` wrapper routes to the right engine."""
|
||||
payload = "{{ 7*7 }}"
|
||||
# Default engine is Jinja2.
|
||||
assert escape_literal(payload) == escape_jinja_literal(payload)
|
||||
assert escape_literal(payload, engine="jinja2") == escape_jinja_literal(payload)
|
||||
# An unknown engine falls back to the safer Jinja2 escaping.
|
||||
assert escape_literal(payload, engine="nonsense") == escape_jinja_literal(payload)
|
||||
# ERB routing.
|
||||
erb_payload = "<%= system('id') %>"
|
||||
assert escape_literal(erb_payload, engine="erb") == escape_erb_literal(erb_payload)
|
||||
|
||||
|
||||
def test_escape_literal_jinja_output_is_inert():
|
||||
"""End-to-end: text routed through escape_literal does not execute."""
|
||||
escaped = escape_literal("{{ boom.run('x') }}")
|
||||
env = jinja2.Environment(undefined=jinja2.ChainableUndefined)
|
||||
rendered = env.from_string(escaped).render(boom=_Boom())
|
||||
assert TRIP not in rendered
|
||||
|
|
|
|||
120
tests/test_security_hardening.py
Normal file
120
tests/test_security_hardening.py
Normal file
|
|
@ -0,0 +1,120 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
import os
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
from defusedxml.common import EntitiesForbidden
|
||||
|
||||
from jinjaturtle import cli
|
||||
from jinjaturtle.core import generate_ansible_yaml, parse_config, flatten_config
|
||||
from jinjaturtle.multi import is_supported_file, iter_supported_files, process_directory
|
||||
from jinjaturtle.output_safety import OutputPathError, write_text_safely
|
||||
|
||||
|
||||
class UnsafeAwareLoader(yaml.SafeLoader):
|
||||
pass
|
||||
|
||||
|
||||
def _unsafe(loader: UnsafeAwareLoader, node: yaml.Node):
|
||||
return loader.construct_scalar(node)
|
||||
|
||||
|
||||
UnsafeAwareLoader.add_constructor("!unsafe", _unsafe)
|
||||
|
||||
|
||||
def test_jinja_values_are_emitted_as_ansible_unsafe(tmp_path: Path):
|
||||
src = tmp_path / "app.ini"
|
||||
src.write_text("[main]\ncmd = {{ lookup('pipe','id') }}\n", encoding="utf-8")
|
||||
|
||||
fmt, parsed = parse_config(src)
|
||||
defaults_yaml = generate_ansible_yaml("role", flatten_config(fmt, parsed))
|
||||
|
||||
assert "role_main_cmd: !unsafe" in defaults_yaml
|
||||
assert "{{ lookup(''pipe'',''id'') }}" in defaults_yaml
|
||||
loaded = yaml.load(defaults_yaml, Loader=UnsafeAwareLoader)
|
||||
assert loaded["role_main_cmd"] == "{{ lookup('pipe','id') }}"
|
||||
|
||||
|
||||
def test_folder_mode_marks_nested_jinja_values_and_ids_unsafe(tmp_path: Path):
|
||||
src = tmp_path / "src"
|
||||
src.mkdir()
|
||||
# Filename ids are source-derived values too.
|
||||
(src / "{{ bad }}.yaml").write_text(
|
||||
"message: \"{{ lookup('pipe','id') }}\"\n", encoding="utf-8"
|
||||
)
|
||||
|
||||
defaults_yaml, _outputs = process_directory(
|
||||
src, recursive=False, role_prefix="role"
|
||||
)
|
||||
|
||||
assert "id: !unsafe" in defaults_yaml
|
||||
assert "role_message: !unsafe" in defaults_yaml
|
||||
loaded = yaml.load(defaults_yaml, Loader=UnsafeAwareLoader)
|
||||
assert loaded["role_items"][0]["id"] == "{{ bad }}.yaml"
|
||||
assert loaded["role_items"][0]["role_message"] == "{{ lookup('pipe','id') }}"
|
||||
|
||||
|
||||
def test_xml_parser_rejects_entities_when_called_as_library(tmp_path: Path):
|
||||
src = tmp_path / "bad.xml"
|
||||
src.write_text(
|
||||
"<!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"
|
||||
|
|
@ -10,7 +10,6 @@ from jinjaturtle.core import (
|
|||
analyze_loops,
|
||||
flatten_config,
|
||||
generate_ansible_yaml,
|
||||
generate_erb_template,
|
||||
generate_jinja2_template,
|
||||
)
|
||||
from jinjaturtle.handlers.yaml import YamlHandler
|
||||
|
|
@ -205,16 +204,6 @@ def test_yaml_loop_preserves_blank_separator_after_list(tmp_path: Path):
|
|||
rendered = Template(template).render(**defaults)
|
||||
assert rendered_expected in rendered
|
||||
|
||||
erb_template = generate_erb_template(
|
||||
fmt,
|
||||
parsed,
|
||||
"role",
|
||||
original_text=text,
|
||||
loop_candidates=loop_candidates,
|
||||
flat_items=flat_items,
|
||||
)
|
||||
assert erb_expected in erb_template
|
||||
|
||||
|
||||
def test_yaml_loop_preserves_following_top_level_comments(tmp_path: Path):
|
||||
from jinja2 import Environment, Template
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue