153 lines
7.1 KiB
Python
153 lines
7.1 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 a placeholder such as ``<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. A ``raw``
|
|
option that declares ``options`` is a role toggle: ``true``, ``false``, or the dict those options
|
|
describe. Messages name the key path
|
|
and the types involved but never a value, so secrets cannot leak into job output.
|
|
"""
|
|
|
|
import glob
|
|
from collections.abc import Mapping
|
|
|
|
from ansible.errors import AnsibleError
|
|
from ansible.parsing.vault import EncryptedString
|
|
from ansible.plugins.action import ActionBase
|
|
|
|
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. An empty string is the contract's unset int, as in
|
|
# `system.vlan: ""`. 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 == "" or 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 kind == "raw" and "options" in spec and not isinstance(value, (bool, Mapping)):
|
|
self.errors.append(f"{path}: expected bool or dict, got {_kind(value)}")
|
|
return
|
|
if not TYPES[kind](value):
|
|
keyed = len(spec.get("options", {})) == 1 and next(iter(spec["options"])).startswith("<")
|
|
self.errors.append(f"{path}: expected {'dict keyed by name' if keyed else 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}")
|
|
# Keyed on the value, not the declared type, so a `raw` option ("false, or a dict of...") is
|
|
# still checked whenever it holds a dict or a list.
|
|
if "options" in spec and isinstance(value, Mapping):
|
|
self.check_options(spec["options"], value, path)
|
|
elif "elements" in spec and isinstance(value, list):
|
|
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 len(options) == 1 and next(iter(options)).startswith("<"):
|
|
entry_spec = next(iter(options.values()))
|
|
for name, entry in value.items():
|
|
if entry is not None:
|
|
self.check(entry_spec, entry, f"{path}.{name}")
|
|
return
|
|
for key in value:
|
|
if key not in options:
|
|
accepted = ", ".join(sorted(options)) or "no keys"
|
|
self.errors.append(f"{path}.{key}: unsupported key, {path} accepts {accepted}")
|
|
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)
|
|
declared = {}
|
|
paths = []
|
|
# A pattern (`.../*/roles/*/meta/argument_specs.yml`) picks up every role's spec without a list
|
|
# to keep in sync; one that matches nothing is a broken path, not an empty set.
|
|
specs = self._task.args["specs"]
|
|
if not isinstance(specs, list):
|
|
raise AnsibleError(f"specs must be a list of spec file paths, got {type(specs).__name__}")
|
|
for entry in specs:
|
|
matches = sorted(glob.glob(entry)) if glob.has_magic(entry) else [entry]
|
|
if not matches:
|
|
raise AnsibleError(f"no spec file matches {entry}")
|
|
paths.extend(matches)
|
|
for path in paths:
|
|
options = self._loader.load_from_file(path)["argument_specs"]["inputs"]["options"]
|
|
_check_spec(options, path)
|
|
for name, spec in options.items():
|
|
if name in declared:
|
|
raise AnsibleError(f"{name} is declared by both {declared[name]} and {path}")
|
|
declared[name] = path
|
|
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
|