127 lines
5.4 KiB
Python
127 lines
5.4 KiB
Python
"""Strict validation of user-facing variable dicts against role argument specs.
|
|
|
|
Each spec file's ``argument_specs.inputs.options`` declares top-level variables. Every one the host
|
|
defines is checked without Ansible's lenient coercion: a string is never a list, bool or dict. A dict
|
|
option whose only sub-option is named ``<name>`` is a collection keyed by name; each entry is checked
|
|
against that sub-option, and a null entry means the entry is not managed. Messages name the key path
|
|
and the types involved but never a value, so secrets cannot leak into job output.
|
|
"""
|
|
|
|
from collections.abc import Mapping
|
|
|
|
from ansible.errors import AnsibleError
|
|
from ansible.parsing.vault import EncryptedString
|
|
from ansible.plugins.action import ActionBase
|
|
|
|
KEYED = "<name>"
|
|
SPEC_KEYS = {"type", "required", "default", "choices", "elements", "options", "description", "no_log"}
|
|
UNRESOLVED = object()
|
|
|
|
# Numbers pass where a string is declared and digit strings where an int is: the roles read those
|
|
# through `| string` and `| int` unchanged. A string posing as a list, bool or dict never passes.
|
|
TYPES = {
|
|
"str": lambda v: isinstance(v, (str, int, float, EncryptedString)) and not isinstance(v, bool),
|
|
"int": lambda v: (isinstance(v, int) and not isinstance(v, bool)) or (isinstance(v, str) and v.isdigit()),
|
|
"float": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool),
|
|
"bool": lambda v: isinstance(v, bool),
|
|
"list": lambda v: isinstance(v, list),
|
|
"dict": lambda v: isinstance(v, Mapping),
|
|
"raw": lambda _: True,
|
|
}
|
|
|
|
|
|
def _kind(value):
|
|
if value is None:
|
|
return "null"
|
|
# bool before int: bool is an int subclass.
|
|
for name, cls in (("bool", bool), ("dict", Mapping), ("str", (str, EncryptedString)), ("int", int),
|
|
("float", float), ("list", list)):
|
|
if isinstance(value, cls):
|
|
return name
|
|
return type(value).__name__
|
|
|
|
|
|
def _check_spec(options, path):
|
|
for key, spec in options.items():
|
|
unknown = set(spec) - SPEC_KEYS
|
|
if unknown:
|
|
raise AnsibleError(f"{path}.{key}: unsupported spec attributes {', '.join(sorted(unknown))}")
|
|
if spec.get("type", "str") not in TYPES:
|
|
raise AnsibleError(f"{path}.{key}: unsupported spec type {spec['type']}")
|
|
if "options" in spec:
|
|
_check_spec(spec["options"], f"{path}.{key}")
|
|
|
|
|
|
class _Validator:
|
|
def __init__(self, templar):
|
|
self.templar = templar
|
|
self.errors = []
|
|
|
|
def resolve(self, value):
|
|
# Task vars arrive untemplated. Render one node at a time so a vaulted leaf is never decrypted;
|
|
# a node that cannot render here (no vault secret, a runtime-only variable) is left unchecked.
|
|
if isinstance(value, str) and self.templar.is_template(value):
|
|
try:
|
|
return self.templar.template(value)
|
|
except AnsibleError:
|
|
return UNRESOLVED
|
|
return value
|
|
|
|
def check(self, spec, value, path):
|
|
value = self.resolve(value)
|
|
if value is UNRESOLVED:
|
|
return
|
|
kind = spec.get("type", "str")
|
|
if not TYPES[kind](value):
|
|
self.errors.append(f"{path}: expected {kind}, got {_kind(value)}")
|
|
return
|
|
if "choices" in spec and value not in spec["choices"]:
|
|
shown = "" if spec.get("no_log") else f", got {value!r}"
|
|
self.errors.append(f"{path}: must be one of {', '.join(map(str, spec['choices']))}{shown}")
|
|
if kind == "dict" and "options" in spec:
|
|
self.check_options(spec["options"], value, path)
|
|
elif kind == "list" and "elements" in spec:
|
|
element = {"type": spec["elements"]}
|
|
if "options" in spec:
|
|
element["options"] = spec["options"]
|
|
for index, item in enumerate(value):
|
|
self.check(element, item, f"{path}[{index}]")
|
|
|
|
def check_options(self, options, value, path):
|
|
if set(options) == {KEYED}:
|
|
for name, entry in value.items():
|
|
if entry is not None:
|
|
self.check(options[KEYED], entry, f"{path}.{name}")
|
|
return
|
|
for key in value:
|
|
if key not in options:
|
|
self.errors.append(f"{path}.{key}: unsupported key, {path} accepts {', '.join(sorted(options))}")
|
|
for key, spec in options.items():
|
|
if key in value:
|
|
self.check(spec, value[key], f"{path}.{key}")
|
|
elif spec.get("required"):
|
|
self.errors.append(f"{path}.{key}: required key is missing")
|
|
|
|
|
|
class ActionModule(ActionBase):
|
|
_requires_connection = False
|
|
|
|
def run(self, tmp=None, task_vars=None):
|
|
result = super().run(tmp, task_vars)
|
|
task_vars = task_vars or {}
|
|
validator = _Validator(self._templar)
|
|
for path in self._task.args["specs"]:
|
|
options = self._loader.load_from_file(path)["argument_specs"]["inputs"]["options"]
|
|
_check_spec(options, path)
|
|
for name, spec in options.items():
|
|
if name in task_vars:
|
|
validator.check(spec, task_vars[name], name)
|
|
elif spec.get("required"):
|
|
validator.errors.append(f"{name}: required variable is not defined")
|
|
if validator.errors:
|
|
result.update(failed=True, errors=validator.errors,
|
|
msg="Invalid input variables:\n" + "\n".join(validator.errors))
|
|
else:
|
|
result.update(changed=False, msg="All input variables are valid")
|
|
return result
|