feat(ca-trust): unify trust injection + strict templating

Move `with-ca-trust.sh` to `sys-ca-selfsigned` as the single source of truth, add `CA_TRUST_NAME`/`trust_name`, and pass it through `compose_ca_inject.py` + `compose_ca_inject_cmd`. Introduce `module_utils/templating.py` for strict multi-pass rendering with fallback env lookup, update TLS self-signed regen logic, purge script, and extend unit tests accordingly.

https://chatgpt.com/share/697c876d-7140-800f-9981-762dfeeafdfe
This commit is contained in:
kevinveenbirkenbach committed 2026-01-30 11:26:46 +01:00
1 parent aedce7a9cd
commit 2678090934
14 files changed
+644 -190

No files matched your search

+3 -1
View File
@@ -2,7 +2,9 @@
- Implement multi language
- Implement rbac administration interface
- [Enable IP6 for docker](https://chatgpt.com/share/68a0acb8-db20-800f-9d2c-b34e38b5cdee).
-. backup docker to local für ca optimieren
# Testing
- re-run witrh different credentials/configuration
- run over all distros for each app
- run over all distros for each app
+4 -3
View File
@@ -41,6 +41,7 @@ CA_ROOT:
CN: "{{ DOMAIN_PRIMARY | default('infinito.local') }} Root CA"
CA_TRUST:
cert_host: "{{ CA_ROOT.cert_host }}"
wrapper_host: "/etc/{{ SOFTWARE_NAME | lower }}/bin/with-ca-trust.sh"
inject_script: "/etc/{{ SOFTWARE_NAME | lower }}/bin/compose_ca_inject.py"
cert_host: "{{ CA_ROOT.cert_host }}"
wrapper_host: "/usr/local/bin/with-ca-trust.sh"
inject_script: "/usr/local/bin/compose_ca_inject.py"
trust_name: "{{ DOMAIN_PRIMARY | lower }}"
+180
View File
@@ -0,0 +1,180 @@
# module_utils/templating.py
from __future__ import annotations
import os
import re
from typing import Any
from ansible.errors import AnsibleError
_RE_LOOKUP_ENV = re.compile(
r"""^lookup\(\s*['"]env['"]\s*,\s*['"]([^'"]+)['"]\s*\)\s*$"""
)
_RE_VARPATH = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z0-9_]+)*$")
_RE_JINJA_BLOCK = re.compile(r"\{\{\s*(.*?)\s*\}\}", re.DOTALL)
def _get_by_path(variables: dict, path: str) -> Any:
cur: Any = variables
for part in path.split("."):
if not isinstance(cur, dict) or part not in cur:
raise KeyError(path)
cur = cur[part]
return cur
def _apply_filter(value: Any, filt: str) -> Any:
f = filt.strip()
if f == "lower":
return str(value).lower()
if f == "upper":
return str(value).upper()
# default('x', true) -> treat None/"" as default
if f.startswith("default(") and f.endswith(")"):
inner = f[len("default(") : -1].strip()
default_val: Any = ""
if inner.startswith(("'", '"')):
q = inner[0]
end = inner.find(q, 1)
default_val = inner[1:end] if end != -1 else ""
else:
default_val = inner.split(",", 1)[0].strip()
if value is None:
return default_val
if str(value) == "":
return default_val
return value
# Unknown filter -> no-op
return value
def _fallback_eval_expr(expr: str, variables: dict) -> str:
"""
Evaluate a single Jinja expression (no surrounding {{ }}).
Supported:
- lookup('env','NAME')
- VAR / VAR.path
- filters: | lower, | upper, | default('x', true)
This is a fallback when Ansible templar can't or won't evaluate.
"""
parts = [p.strip() for p in expr.split("|")]
head = parts[0]
m = _RE_LOOKUP_ENV.match(head)
if m:
key = m.group(1)
val: Any = os.environ.get(key)
else:
if not _RE_VARPATH.match(head):
raise ValueError(f"unsupported expression: {head}")
try:
val = _get_by_path(variables, head)
except KeyError:
val = None
for filt in parts[1:]:
val = _apply_filter(val, filt)
return "" if val is None else str(val)
def _fallback_render_embedded(s: str, variables: dict) -> str:
def repl(m: re.Match) -> str:
expr = (m.group(1) or "").strip()
return _fallback_eval_expr(expr, variables)
return _RE_JINJA_BLOCK.sub(repl, s)
def _templar_render_best_effort(templar: Any, s: str, variables: dict) -> str:
"""
Render with Ansible templar across versions.
Compatibility:
- Do NOT pass variables=... to templar.template() (breaks on some versions)
- Instead set templar.available_variables temporarily.
"""
if templar is None:
return _fallback_render_embedded(s, variables)
prev = None
if hasattr(templar, "available_variables"):
try:
prev = templar.available_variables
templar.available_variables = variables
except Exception:
prev = None
try:
try:
out = templar.template(s, fail_on_undefined=True)
except TypeError:
out = templar.template(s)
finally:
if prev is not None and hasattr(templar, "available_variables"):
try:
templar.available_variables = prev
except Exception:
pass
out_s = "" if out is None else str(out)
# If templar didn't change anything while Jinja exists, fallback for embedded patterns.
if out_s.strip() == s.strip() and ("{{" in s or "{%" in s):
return _fallback_render_embedded(s, variables)
return out_s
def render_ansible_strict(
*,
templar: Any,
raw: Any,
var_name: str,
err_prefix: str,
variables: dict,
max_rounds: int = 6,
) -> str:
"""
Strict rendering helper for lookup/filter plugins.
- Renders via Ansible templar when possible (lookup(), filters, vars).
- If templar can't/won't render, applies safe fallback for embedded {{ ... }}.
- Re-renders multiple rounds because intermediate results can still contain Jinja.
- Hard-fails if output is empty or still contains unresolved Jinja markers.
"""
if raw is None:
raise AnsibleError(f"{err_prefix}: {var_name} resolved to None")
s = str(raw)
out = s
for _ in range(max_rounds):
if ("{{" not in out) and ("{%" not in out):
break
out2 = _templar_render_best_effort(templar, out, variables)
if out2 == out:
break
out = out2
out = "" if out is None else str(out).strip()
if not out:
raise AnsibleError(
f"{err_prefix}: {var_name} rendered to empty string. Raw: {s}"
)
if ("{{" in out) or ("{%" in out):
raise AnsibleError(
f"{err_prefix}: {var_name} still contains unresolved Jinja. Rendered: {out}. Raw: {s}"
)
return out
@@ -273,12 +273,17 @@ def render_override(
env: Dict[str, str],
ca_host: str,
wrapper_host: str,
trust_name: str,
) -> Dict[str, Any]:
"""
Generate a docker-compose override that injects CA trust into every service by:
- mounting CA cert + wrapper script
- setting entrypoint to the wrapper
- setting command to the ORIGINAL effective command (final entrypoint + final command)
Also sets:
- CA_TRUST_CERT (container path)
- CA_TRUST_NAME (trust anchor name; required by with-ca-trust.sh)
"""
ca_container = "/tmp/infinito/ca/root-ca.crt"
wrapper_container = "/tmp/infinito/bin/with-ca-trust.sh"
@@ -331,7 +336,10 @@ def render_override(
f"{ca_host}:{ca_container}:ro",
f"{wrapper_host}:{wrapper_container}:ro",
],
"environment": {"CA_TRUST_CERT": ca_container},
"environment": {
"CA_TRUST_CERT": ca_container,
"CA_TRUST_NAME": trust_name,
},
"entrypoint": [wrapper_container],
"command": effective_cmd,
}
@@ -362,6 +370,11 @@ def main() -> int:
required=True,
help="Host path to wrapper script (bind-mounted)",
)
ap.add_argument(
"--trust-name",
required=True,
help="Trust anchor name for CA installation inside containers (CA_TRUST_NAME)",
)
args = ap.parse_args()
cwd = Path(args.chdir)
@@ -374,10 +387,14 @@ def main() -> int:
ca_host = str(args.ca_host).strip()
wrapper_host = str(args.wrapper_host).strip()
trust_name = str(args.trust_name).strip()
if not ca_host:
die("--ca-host must be non-empty")
if not wrapper_host:
die("--wrapper-host must be non-empty")
if not trust_name:
die("--trust-name must be non-empty")
env = dict(os.environ)
@@ -443,6 +460,7 @@ def main() -> int:
env=env,
ca_host=ca_host,
wrapper_host=wrapper_host,
trust_name=trust_name,
)
try:
@@ -1,39 +0,0 @@
#!/usr/bin/env sh
set -eu
: "${CA_TRUST_CERT:?CA_TRUST_CERT env var must be set (path inside container to the mounted CA cert)}"
if [ ! -r "$CA_TRUST_CERT" ]; then
echo "[with-ca-trust] CA certificate not readable: $CA_TRUST_CERT" >&2
exit 2
fi
install_ca_debian_like() {
dst="/usr/local/share/ca-certificates/infinito-root-ca.crt"
mkdir -p "$(dirname "$dst")"
cp -f "$CA_TRUST_CERT" "$dst"
update-ca-certificates >/dev/null
}
install_ca_rhel_like() {
dst="/etc/pki/ca-trust/source/anchors/infinito-root-ca.crt"
mkdir -p "$(dirname "$dst")"
cp -f "$CA_TRUST_CERT" "$dst"
update-ca-trust extract >/dev/null
}
if command -v update-ca-certificates >/dev/null 2>&1; then
install_ca_debian_like
elif command -v update-ca-trust >/dev/null 2>&1; then
install_ca_rhel_like
else
echo "[with-ca-trust] No CA update tool found (update-ca-certificates/update-ca-trust)." >&2
exit 3
fi
if [ "$#" -lt 1 ]; then
echo "[with-ca-trust] No command provided to execute after CA installation." >&2
exit 4
fi
exec "$@"
-22
View File
@@ -3,19 +3,6 @@
name: sys-ca-selfsigned
when: run_once_sys_ca_selfsigned is not defined
- name: "CA trust | assert CA_TRUST structure"
ansible.builtin.assert:
that:
- CA_TRUST is mapping
- CA_TRUST.cert_host is defined
- CA_TRUST.wrapper_host is defined
- CA_TRUST.inject_script is defined
- (CA_TRUST.cert_host | length) > 0
- (CA_TRUST.wrapper_host | length) > 0
- (CA_TRUST.inject_script | length) > 0
fail_msg: >-
CA_TRUST must be defined with keys cert_host, wrapper_host, inject_script.
- name: "CA trust | ensure directories exist"
become: true
ansible.builtin.file:
@@ -40,15 +27,6 @@
Provision it before running docker-compose CA injection.
when: not _ca_cert_stat.stat.exists
- name: "CA trust | install wrapper script"
become: true
ansible.builtin.copy:
src: "with-ca-trust.sh"
dest: "{{ CA_TRUST.wrapper_host }}"
mode: "0755"
owner: root
group: root
- name: "CA trust | install compose CA inject script"
become: true
ansible.builtin.copy:
@@ -1,23 +1,4 @@
# roles/docker-compose/lookup_plugins/compose_ca_inject_cmd.py
#
# Build the command to generate docker-compose.ca.override.yml via compose_ca_inject.py
#
# HARD FAIL PRINCIPLE:
# - No silent defaults.
# - Exactly one term is required: application_id.
#
# Conventions (Infinito.Nexus):
# - docker compose project name is ALWAYS the entity name:
# project = get_entity_name(application_id)
#
# This lookup calls compose_ca_inject.py, which runs `docker compose config` using:
# - base docker-compose.yml
# - optional docker-compose.override.yml (ONLY if the role provides it)
# - NEVER the CA override during injection bootstrap (because it does not exist yet)
#
# Source of truth for "-f ..." composition:
# - compose_f_args(application_id, include_ca=False)
from __future__ import annotations
import os
@@ -29,7 +10,7 @@ from ansible.plugins.lookup import LookupBase
from ansible.plugins.loader import lookup_loader
from module_utils.entity_name_utils import get_entity_name
from module_utils.jinja_strict import render_strict
from module_utils.templating import render_ansible_strict
def _as_str(v: Any) -> str:
@@ -52,16 +33,10 @@ def _require(d: Any, key: str, expected_type: Any, *, label: str) -> Any:
def _shell_quote(s: str) -> str:
# Safe for sh: wrap in single quotes and escape embedded single quotes.
return "'" + s.replace("'", "'\"'\"'") + "'"
def _coerce_to_dict(v: Any, label: str) -> dict:
"""
Coerce a value into a dict:
- accept dict directly
- accept YAML/JSON encoded string and parse it
"""
if isinstance(v, dict):
return v
@@ -85,10 +60,6 @@ def _coerce_to_dict(v: Any, label: str) -> dict:
def _maybe_template(templar: Any, value: Any) -> Any:
"""
Render via Ansible templar if available.
Unit tests may not inject a templar -> then we skip templating.
"""
if isinstance(value, (dict, list, tuple, int, float, bool)) or value is None:
return value
if templar is None:
@@ -103,7 +74,6 @@ class LookupModule(LookupBase):
def run(self, terms, variables: Optional[dict] = None, **kwargs):
variables = variables or {}
# 1) Require application_id as the single lookup term
if not terms or len(terms) != 1:
raise AnsibleError(
"compose_ca_inject_cmd: exactly one term required (application_id)"
@@ -113,12 +83,10 @@ class LookupModule(LookupBase):
if not application_id:
raise AnsibleError("compose_ca_inject_cmd: application_id is empty")
# 2) Resolve project name from application_id via module utility (no Jinja templating)
project = _as_str(get_entity_name(application_id))
if not project:
raise AnsibleError("compose_ca_inject_cmd: resolved project is empty")
# 3) Read docker_compose mapping from variables
raw_dc = variables.get("docker_compose", None)
if raw_dc is None:
raise AnsibleError(
@@ -136,54 +104,57 @@ class LookupModule(LookupBase):
"compose_ca_inject_cmd: docker_compose.directories/files must be dicts"
)
instance_dir = render_strict(
dirs.get("instance"),
variables=variables,
instance_dir = render_ansible_strict(
templar=templar,
raw=dirs.get("instance"),
var_name="docker_compose.directories.instance",
err_prefix="compose_ca_inject_cmd",
)
env_file = render_strict(
files.get("env"),
variables=variables,
)
env_file = render_ansible_strict(
templar=templar,
raw=files.get("env"),
var_name="docker_compose.files.env",
err_prefix="compose_ca_inject_cmd",
)
out_file = render_strict(
files.get("docker_compose_ca_override"),
variables=variables,
)
out_file = render_ansible_strict(
templar=templar,
raw=files.get("docker_compose_ca_override"),
var_name="docker_compose.files.docker_compose_ca_override",
err_prefix="compose_ca_inject_cmd",
variables=variables,
)
if not _as_str(instance_dir):
raise AnsibleError(
"compose_ca_inject_cmd: docker_compose.directories.instance is required"
)
if not _as_str(out_file):
raise AnsibleError(
"compose_ca_inject_cmd: docker_compose.files.docker_compose_ca_override is required"
)
# 4) CA_TRUST inputs
ca_trust = _require(variables, "CA_TRUST", dict, label="variable")
script_path = render_strict(
_require(ca_trust, "inject_script", str, label="CA_TRUST"),
variables=variables,
script_path = render_ansible_strict(
templar=templar,
raw=_require(ca_trust, "inject_script", str, label="CA_TRUST"),
var_name="CA_TRUST.inject_script",
err_prefix="compose_ca_inject_cmd",
)
ca_host = render_strict(
_require(ca_trust, "cert_host", str, label="CA_TRUST"),
variables=variables,
)
ca_host = render_ansible_strict(
templar=templar,
raw=_require(ca_trust, "cert_host", str, label="CA_TRUST"),
var_name="CA_TRUST.cert_host",
err_prefix="compose_ca_inject_cmd",
)
wrapper_host = render_strict(
_require(ca_trust, "wrapper_host", str, label="CA_TRUST"),
variables=variables,
)
wrapper_host = render_ansible_strict(
templar=templar,
raw=_require(ca_trust, "wrapper_host", str, label="CA_TRUST"),
var_name="CA_TRUST.wrapper_host",
err_prefix="compose_ca_inject_cmd",
variables=variables,
)
trust_name = render_ansible_strict(
templar=templar,
raw=_require(ca_trust, "trust_name", str, label="CA_TRUST"),
var_name="CA_TRUST.trust_name",
err_prefix="compose_ca_inject_cmd",
variables=variables,
)
out_basename = os.path.basename(out_file)
@@ -192,19 +163,15 @@ class LookupModule(LookupBase):
"compose_ca_inject_cmd: output basename resolved to empty"
)
# 5) Canonical compose file resolver:
# - same logic as runtime, but WITHOUT CA file (bootstrap)
compose_f_args_lkp = lookup_loader.get(
"compose_f_args", self._loader, self._templar
)
compose_files = compose_f_args_lkp.run(
[application_id], variables=variables, include_ca=False
)[0]
if not _as_str(compose_files):
raise AnsibleError("compose_ca_inject_cmd: compose_f_args returned empty")
# 6) Build command (compose-files is already a string like "-f a.yml -f b.yml")
cmd = [
"python3",
_shell_quote(script_path),
@@ -226,6 +193,8 @@ class LookupModule(LookupBase):
_shell_quote(ca_host),
"--wrapper-host",
_shell_quote(wrapper_host),
"--trust-name",
_shell_quote(trust_name),
]
return [" ".join(cmd)]
@@ -0,0 +1,54 @@
#!/usr/bin/env sh
set -eu
: "${CA_TRUST_CERT:?CA_TRUST_CERT env var must be set (path to CA cert)}"
: "${CA_TRUST_NAME:?CA_TRUST_NAME env var must be set (trust anchor name)}"
if [ ! -r "$CA_TRUST_CERT" ]; then
echo "[with-ca-trust] CA certificate not readable: $CA_TRUST_CERT" >&2
exit 2
fi
# Sanitize trust name to a safe filename
name="$(printf '%s' "$CA_TRUST_NAME" | tr -c 'A-Za-z0-9._-' '_' )"
if [ -z "$name" ]; then
echo "[with-ca-trust] CA_TRUST_NAME resolved to empty after sanitization." >&2
exit 4
fi
install_ca_debian_like() {
dst="/usr/local/share/ca-certificates/${name}.crt"
mkdir -p "$(dirname "$dst")"
cp -f "$CA_TRUST_CERT" "$dst"
update-ca-certificates >/dev/null
}
install_ca_rhel_like() {
dst="/etc/pki/ca-trust/source/anchors/${name}.crt"
mkdir -p "$(dirname "$dst")"
cp -f "$CA_TRUST_CERT" "$dst"
update-ca-trust extract >/dev/null
}
install_ca_arch_like() {
dst="/etc/ca-certificates/trust-source/anchors/${name}.crt"
mkdir -p "$(dirname "$dst")"
cp -f "$CA_TRUST_CERT" "$dst"
trust extract-compat >/dev/null
}
if command -v update-ca-certificates >/dev/null 2>&1; then
install_ca_debian_like
elif command -v update-ca-trust >/dev/null 2>&1; then
install_ca_rhel_like
elif command -v trust >/dev/null 2>&1; then
install_ca_arch_like
else
echo "[with-ca-trust] No CA update tool found." >&2
exit 3
fi
# Optional: run a command (container use-case)
if [ "$#" -gt 0 ]; then
exec "$@"
fi
+22
View File
@@ -133,4 +133,26 @@
Generated certificate does not appear to be a CA certificate (CA:TRUE missing).
Please inspect {{ CA_ROOT.cert_host }}.
- name: "CA | install trust helper script (SPOT) on host"
become: true
ansible.builtin.copy:
src: "with-ca-trust.sh"
dest: "{{ CA_TRUST.wrapper_host }}"
owner: root
group: root
mode: "0755"
- name: "CA | (optional) install Root CA into host trust store"
become: true
ansible.builtin.command:
argv:
- "{{ CA_TRUST.wrapper_host }}"
environment:
CA_TRUST_CERT: "{{ CA_ROOT.cert_host }}"
CA_TRUST_NAME: "{{ DOMAIN_PRIMARY | lower }}"
changed_when: true
when:
- (CA_ROOT.install_to_host_trust | default(false)) | bool
- _ca_regen_needed | bool
- include_tasks: utils/once/flag.yml
@@ -37,7 +37,15 @@
_ss_key: "{{ _cert.files.key }}"
_ss_dir: "{{ _cert.files.cert | dirname }}"
_ss_conf: "{{ (_cert.files.cert | dirname) ~ '/openssl.cnf' }}"
_ss_cn: "{{ (tls_selfsigned_subject.CN | default('') | length > 0) | ternary(tls_selfsigned_subject.CN, _tls.domains.primary) }}"
_ss_cn: >-
{{
(_cert.scope == 'global')
| ternary(
(DOMAIN_PRIMARY | upper) ~ ' Global TLS',
((tls_selfsigned_subject.CN | default('') | length > 0)
| ternary(tls_selfsigned_subject.CN, _tls.domains.primary))
)
}}
changed_when: false
- name: "TLS | (self_signed) ensure directory exists"
@@ -100,45 +108,41 @@
}}
changed_when: false
- name: "TLS | (self_signed) decide whether regeneration is needed"
ansible.builtin.set_fact:
_ss_regen_needed: >-
- block:
- name: "TLS | (self_signed) remove outdated cert/key if SAN drift detected"
ansible.builtin.file:
path: "{{ item }}"
state: absent
loop:
- "{{ _ss_cert }}"
- "{{ _ss_key }}"
- name: "TLS | (self_signed) generate key+cert if missing or drift detected"
ansible.builtin.command:
argv:
- openssl
- req
- -x509
- -nodes
- -days
- "{{ tls_selfsigned_days | string }}"
- -newkey
- "rsa:{{ tls_selfsigned_key_bits }}"
- -keyout
- "{{ _ss_key }}"
- -out
- "{{ _ss_cert }}"
- -config
- "{{ _ss_conf }}"
- -extensions
- req_ext
args:
creates: "{{ _ss_cert }}"
when: >-
{{
(not (_ss_cert_stat.stat.exists | default(false))) or
((_ss_wanted_sans | difference(_ss_existing_sans) | length) > 0)
}}
changed_when: false
- name: "TLS | (self_signed) remove outdated cert/key if SAN drift detected"
ansible.builtin.file:
path: "{{ item }}"
state: absent
loop:
- "{{ _ss_cert }}"
- "{{ _ss_key }}"
when: _ss_regen_needed | bool
- name: "TLS | (self_signed) generate key+cert if missing or drift detected"
ansible.builtin.command:
argv:
- openssl
- req
- -x509
- -nodes
- -days
- "{{ tls_selfsigned_days | string }}"
- -newkey
- "rsa:{{ tls_selfsigned_key_bits }}"
- -keyout
- "{{ _ss_key }}"
- -out
- "{{ _ss_cert }}"
- -config
- "{{ _ss_conf }}"
- -extensions
- req_ext
args:
creates: "{{ _ss_cert }}"
when: _ss_regen_needed | bool
- include_tasks: utils/once/flag.yml
+2
View File
@@ -1,3 +1,5 @@
#!/bin/bash
rm -rv /etc/nginx/
rm -rv /etc/infinito.nexus/ca
rm -rv /etc/infinito.nexus/selfsigned
( cd /opt/docker/openresty/ && docker compose down -v )
@@ -43,14 +43,11 @@ class TestComposeCaInjectCmd(unittest.TestCase):
"inject_script": "/etc/infinito.nexus/bin/compose_ca_inject.py",
"cert_host": "/etc/infinito.nexus/ca/root-ca.crt",
"wrapper_host": "/etc/infinito.nexus/bin/with-ca-trust.sh",
"trust_name": "infinito.nexus",
},
}
def _patch_compose_f_args(self, ret: str):
"""
Patch the lookup_loader.get('compose_f_args', ...) call within the lookup plugin
so tests do not depend on Ansible loader internals.
"""
fake = type(
"FakeComposeFArgs",
(),
@@ -59,7 +56,6 @@ class TestComposeCaInjectCmd(unittest.TestCase):
return patch.object(self.m.lookup_loader, "get", return_value=fake)
def test_builds_command_string(self):
# project is derived from module_utils.entity_name_utils.get_entity_name(application_id)
with (
patch.object(self.m, "get_entity_name", return_value="myproj"),
self._patch_compose_f_args(
@@ -71,48 +67,42 @@ class TestComposeCaInjectCmd(unittest.TestCase):
self.assertIn("python3", out)
self.assertIn("/etc/infinito.nexus/bin/compose_ca_inject.py", out)
# Must include derived project (entity name)
self.assertIn("--project", out)
self.assertIn("'myproj'", out)
# Must include base + override only (NOT the CA override)
self.assertIn("--compose-files", out)
self.assertIn(
"-f /opt/docker/app/docker-compose.yml -f /opt/docker/app/docker-compose.override.yml",
out,
)
self.assertNotIn(
"docker-compose.ca.override.yml -f", out
) # ensure CA override not appended as -f
self.assertNotIn("docker-compose.ca.override.yml -f", out)
# Must include env-file, out basename, and CA args
self.assertIn("--env-file", out)
self.assertIn("/opt/docker/app/.env/env", out)
self.assertIn("--out", out)
# out must be basename of docker_compose_ca_override
self.assertIn("'docker-compose.ca.override.yml'", out)
self.assertIn("--ca-host", out)
self.assertIn("/etc/infinito.nexus/ca/root-ca.crt", out)
self.assertIn("--wrapper-host", out)
self.assertIn("/etc/infinito.nexus/bin/with-ca-trust.sh", out)
self.assertIn("--trust-name", out)
self.assertIn("'infinito.nexus'", out)
def test_requires_application_id_term(self):
# No terms
with self.assertRaises(AnsibleError):
self.lookup.run([], variables=self.vars)
# More than one term
with self.assertRaises(AnsibleError):
self.lookup.run(["a", "b"], variables=self.vars)
# Empty term
with self.assertRaises(AnsibleError):
self.lookup.run([""], variables=self.vars)
def test_requires_project_non_empty(self):
# If entity name resolution returns empty, this must hard-fail.
with patch.object(self.m, "get_entity_name", return_value=""):
with self.assertRaises(AnsibleError):
self.lookup.run(["web-app-foo"], variables=self.vars)
@@ -138,7 +128,6 @@ class TestComposeCaInjectCmd(unittest.TestCase):
self.lookup.run(["web-app-foo"], variables=v)
def test_requires_compose_f_args_non_empty(self):
# compose_f_args returning empty must hard-fail
with (
patch.object(self.m, "get_entity_name", return_value="p"),
self._patch_compose_f_args(""),
+223
View File
@@ -0,0 +1,223 @@
# tests/unit/module_utils/test_templating.py
import os
import unittest
from pathlib import Path
from unittest.mock import patch
from ansible.errors import AnsibleError
def _repo_root() -> Path:
# __file__ = tests/unit/module_utils/test_templating.py
return Path(__file__).resolve().parents[3]
# Ensure repo root is importable so `module_utils.*` resolves in all runners
import sys # noqa: E402
sys.path.insert(0, str(_repo_root())) # noqa: E402
from module_utils.templating import ( # noqa: E402
render_ansible_strict,
)
class _FakeTemplarLookupUndefined:
"""
Simulates the earlier failure:
- templar exists
- but cannot evaluate lookup()
- returns string unchanged => must trigger fallback evaluator
"""
def __init__(self):
self.available_variables = {}
def template(self, s, *args, **kwargs):
# behave like templar that doesn't expand anything
return s
class _FakeTemplarRejectsVariablesKwarg:
"""
Simulates the earlier failure:
Templar.template() got an unexpected keyword argument 'variables'
We ensure our helper does NOT pass variables kwarg.
"""
def __init__(self):
self.available_variables = {}
def template(self, s, fail_on_undefined=False): # supports only these args
# return something simple if variables were set via available_variables
# We mimic minimal Jinja for a very specific case:
if s.strip() == "{{ CA_ROOT.cert_host }}":
ca_root = self.available_variables.get("CA_ROOT", {})
return ca_root.get("cert_host", s)
return s
class TestTemplatingRenderStrict(unittest.TestCase):
def test_plain_string_passthrough(self):
out = render_ansible_strict(
templar=None,
raw="/etc/infinito/ca/root-ca.crt",
var_name="x",
err_prefix="t",
variables={},
)
self.assertEqual(out, "/etc/infinito/ca/root-ca.crt")
def test_none_hard_fails(self):
with self.assertRaises(AnsibleError):
render_ansible_strict(
templar=None,
raw=None,
var_name="x",
err_prefix="t",
variables={},
)
def test_empty_after_render_hard_fails(self):
with self.assertRaises(AnsibleError):
render_ansible_strict(
templar=None,
raw=" ",
var_name="x",
err_prefix="t",
variables={},
)
def test_env_lookup_fallback_when_lookup_undefined_like(self):
# Previously: 'lookup' is undefined -> string remained unchanged.
# Now: fallback must resolve lookup('env', 'domain') (case-insensitive quoting)
with patch.dict(os.environ, {"domain": "infinito.localhost"}, clear=False):
out = render_ansible_strict(
templar=_FakeTemplarLookupUndefined(),
raw="{{ lookup('env', 'domain') | default('infinito.localhost', true) }}",
var_name="CA_TRUST.trust_name",
err_prefix="compose_ca_inject_cmd",
variables={},
)
self.assertEqual(out, "infinito.localhost")
def test_env_lookup_default_when_env_missing(self):
with patch.dict(os.environ, {}, clear=True):
out = render_ansible_strict(
templar=_FakeTemplarLookupUndefined(),
raw="{{ lookup('env', 'DOMAIN') | default('infinito.localhost', true) }}",
var_name="CA_TRUST.trust_name",
err_prefix="compose_ca_inject_cmd",
variables={},
)
self.assertEqual(out, "infinito.localhost")
def test_ca_root_nested_var_resolves_without_templar(self):
# This is the scenario that kept failing:
# Rendered: /etc/{{ SOFTWARE_NAME | lower }}/ca/root-ca.crt
# Raw: {{ CA_ROOT.cert_host }}
#
# We simulate CA_ROOT.cert_host itself being a Jinja string that requires another variable.
variables = {
"SOFTWARE_NAME": "Infinito.Nexus",
"CA_ROOT": {
"cert_host": "/etc/{{ SOFTWARE_NAME | lower }}/ca/root-ca.crt",
},
}
out = render_ansible_strict(
templar=None, # no templar available -> must still resolve via fallback
raw="{{ CA_ROOT.cert_host }}",
var_name="CA_TRUST.cert_host",
err_prefix="compose_ca_inject_cmd",
variables=variables,
)
self.assertEqual(out, "/etc/infinito.nexus/ca/root-ca.crt")
def test_templar_available_variables_is_used_no_variables_kwarg(self):
# This covers the earlier crash:
# Templar.template() got an unexpected keyword argument 'variables'
#
# Our helper must not pass variables=..., but instead set available_variables.
variables = {"CA_ROOT": {"cert_host": "/etc/infinito/ca/root-ca.crt"}}
out = render_ansible_strict(
templar=_FakeTemplarRejectsVariablesKwarg(),
raw="{{ CA_ROOT.cert_host }}",
var_name="CA_TRUST.cert_host",
err_prefix="compose_ca_inject_cmd",
variables=variables,
)
self.assertEqual(out, "/etc/infinito/ca/root-ca.crt")
def test_unresolved_jinja_hard_fails(self):
# If we cannot resolve, strict mode must hard fail (this matches your runtime failures).
with self.assertRaises(AnsibleError):
render_ansible_strict(
templar=_FakeTemplarLookupUndefined(),
raw="{{ DOES_NOT_EXIST }}",
var_name="x",
err_prefix="t",
variables={},
)
def test_multi_round_rendering(self):
# Ensure multi-pass rendering works:
# raw -> "{{ CA_ROOT.cert_host }}"
# CA_ROOT.cert_host -> "/etc/{{ SOFTWARE_NAME | lower }}/ca/root-ca.crt"
variables = {
"SOFTWARE_NAME": "Infinito.Nexus",
"CA_ROOT": {"cert_host": "/etc/{{ SOFTWARE_NAME | lower }}/ca/root-ca.crt"},
}
out = render_ansible_strict(
templar=_FakeTemplarLookupUndefined(), # returns unchanged => fallback kicks in
raw="{{ CA_ROOT.cert_host }}",
var_name="x",
err_prefix="t",
variables=variables,
)
self.assertEqual(out, "/etc/infinito.nexus/ca/root-ca.crt")
def test_embedded_expression_in_string(self):
# Make sure embedded {{ ... }} inside a larger string is handled.
variables = {"SOFTWARE_NAME": "Infinito.Nexus"}
out = render_ansible_strict(
templar=None,
raw="/etc/{{ SOFTWARE_NAME | lower }}/ca/root-ca.crt",
var_name="x",
err_prefix="t",
variables=variables,
)
self.assertEqual(out, "/etc/infinito.nexus/ca/root-ca.crt")
def test_filters_lower_upper_default(self):
variables = {"NAME": "MiXeD", "EMPTY": ""}
out1 = render_ansible_strict(
templar=None,
raw="{{ NAME | lower }}",
var_name="x",
err_prefix="t",
variables=variables,
)
self.assertEqual(out1, "mixed")
out2 = render_ansible_strict(
templar=None,
raw="{{ NAME | upper }}",
var_name="x",
err_prefix="t",
variables=variables,
)
self.assertEqual(out2, "MIXED")
out3 = render_ansible_strict(
templar=None,
raw="{{ EMPTY | default('fallback', true) }}",
var_name="x",
err_prefix="t",
variables=variables,
)
self.assertEqual(out3, "fallback")
if __name__ == "__main__":
unittest.main()
@@ -1,5 +1,6 @@
# tests/unit/roles/docker-compose-ca/files/test_compose_ca_inject.py
import importlib.util
import json
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -58,23 +59,39 @@ class TestComposeCaInject(unittest.TestCase):
def test_main_generates_override(
self, _mkdir, _write_text, _read_text, _is_dir, _exists
):
# The new code scans compose files to discover profiles.
# So we must provide read_text() content for those files.
"""
The implementation:
- parses compose files to discover profiles (reads files from --compose-files)
- runs `docker compose ... config` (default + per-profile)
- inspects images via `docker image inspect`
- writes an override that injects CA_TRUST_CERT + CA_TRUST_NAME
Therefore we must:
- provide readable compose file content (for profile discovery)
- mock run() for compose config + image inspect
- include the new required arg: --trust-name
"""
# Content for compose files referenced by -f ... arguments (profile discovery).
# Keep it minimal; no profiles needed, but valid YAML mapping with services.
_read_text.return_value = "services:\n app:\n image: myimage:latest\n"
def fake_run(cmd, *, cwd, env):
# docker compose ... config
if (
len(cmd) >= 1
len(cmd) >= 3
and cmd[0:2] == ["docker", "compose"]
and cmd[-1] == "config"
):
yml = "services:\n app:\n image: myimage:latest\n"
return 0, yml, ""
# docker image inspect <image> (used both for exists-check and for config inspection)
# docker image inspect <image>
if cmd[:3] == ["docker", "image", "inspect"]:
json_out = '[{"Config":{"Entrypoint":["/entry"],"Cmd":["run"]}}]'
# compose_ca_inject expects `run()` to return stdout as JSON string here.
json_out = json.dumps(
[{"Config": {"Entrypoint": ["/entry"], "Cmd": ["run"]}}]
)
return 0, json_out, ""
return 1, "", "unexpected"
@@ -94,6 +111,8 @@ class TestComposeCaInject(unittest.TestCase):
"/etc/infinito/ca/root-ca.crt",
"--wrapper-host",
"/etc/infinito/bin/with-ca-trust.sh",
"--trust-name",
"infinito.local",
]
with patch("sys.argv", argv):
rc = self.m.main()
@@ -101,6 +120,38 @@ class TestComposeCaInject(unittest.TestCase):
self.assertEqual(rc, 0)
self.assertTrue(_write_text.called)
# Optional extra sanity: ensure the override contains CA_TRUST_NAME
# (We can't easily read the written content since write_text is mocked,
# but we *can* inspect its call args.)
args, kwargs = _write_text.call_args
# signature: Path.write_text(self, data, encoding=...)
written = args[1] if len(args) > 1 else ""
self.assertIn("CA_TRUST_NAME", written)
self.assertIn("infinito.local", written)
@patch.object(Path, "exists", autospec=True, return_value=True)
@patch.object(Path, "is_dir", autospec=True, return_value=True)
def test_main_requires_trust_name(self, _is_dir, _exists):
argv = [
"compose_ca_inject.py",
"--chdir",
"/tmp/app",
"--project",
"p",
"--compose-files",
"-f docker-compose.yml",
"--out",
"docker-compose.ca.override.yml",
"--ca-host",
"/etc/infinito/ca/root-ca.crt",
"--wrapper-host",
"/etc/infinito/bin/with-ca-trust.sh",
# missing: --trust-name
]
with patch("sys.argv", argv):
with self.assertRaises(SystemExit):
self.m.main()
if __name__ == "__main__":
unittest.main()