diff --git a/docs/vyos.vyos.vyos_file_module.rst b/docs/vyos.vyos.vyos_file_module.rst index 7c3508cb..a1bef8df 100644 --- a/docs/vyos.vyos.vyos_file_module.rst +++ b/docs/vyos.vyos.vyos_file_module.rst @@ -1,251 +1,259 @@ .. _vyos.vyos.vyos_file_module: ******************* vyos.vyos.vyos_file ******************* **Manage files, directories, and their ownership on VyOS devices** Version added: 6.0.0 .. contents:: :local: :depth: 1 Synopsis -------- - Creates, updates, or removes a file or directory on a VyOS device, optionally pushing content from a local file (*src*) or inline text (*content*), and setting owner/group/mode via sudo chown/chmod. - This module does not touch the configuration tree (config.boot). It manages arbitrary filesystem paths such as certificates or auth files under /config/auth/, which are not tracked by commit/save/rollback. - All logic runs inside this module's main(), using the standard get_connection()/run_commands() pattern shared with vyos_command — there is no dedicated action plugin; this module uses the shared generic vyos action plugin like every other module in the collection. Parameters ---------- .. raw:: html
Parameter Choices/Defaults Comments
become
boolean
    Choices:
  • no
  • yes ←
Whether to prefix remote commands with sudo.
content
string
Inline text content to write to dest. Marked no_log, since this module is commonly used to push credential material. Mutually exclusive with src.
+
Since content is a normal string-type module option, Ansible renders any Jinja expressions in it (e.g. {{ my_var }}) before this module ever runs, the same as any other option value — no special templating support is implemented by this module itself.
dest
path / required
Absolute path to the remote file or directory to manage.
group
string
Name of the group that should own dest.
mode
string
Permission bits for dest, as a string (e.g. '0600'). Compared against stat output after normalizing to 4 digits; '600' and '0600' are treated as equivalent.
owner
string
Name of the user that should own dest.
src
path
-
Path to a local file (on the Ansible controller) whose content should be pushed to dest. Read locally and pushed as base64 via a single CLI command, since network_cli has no SFTP/SCP channel available to this module. Mutually exclusive with content.
+
Path to a local file (on the Ansible controller) whose content should be pushed to dest. Transferred via a real SCP session over the connection's own persistent socket (the same mechanism ansible.netcommon.net_put uses), never placed inside a command string. Mutually exclusive with content.
+
File bytes are uploaded exactly as they exist on disk — Ansible does not render Jinja expressions inside the file's contents for src, only in the option values of the task itself (e.g. a templated path string). To push templated text, render it first with the template lookup and pass the result via content instead.
state
string
    Choices:
  • present ←
  • absent
Whether the path should exist (present) or be removed (absent).

Notes ----- .. note:: - This module works with connection ``ansible.netcommon.network_cli``. - - Tested against VyOS 1.4.2 and 1.5.0. - File state managed by this module is independent of VyOS's config revision system. A rollback to a previous config revision will not revert changes made by this module. - Paths under */config/auth* are deliberately setgid ``vyattacfg`` by VyOS's own config-management convention (see vyos.dev T2713). If *mode* is given with a leading digit of ``0`` (e.g. ``'0750'``), this module compares only the rwx bits and will not report a diff for VyOS's own setgid bit. To manage the setgid/setuid/sticky bit explicitly, pass a non-zero leading digit (e.g. ``'2750'``). - - For more information on using Ansible to manage network devices see the :ref:`Ansible Network Guide ` Examples -------- .. code-block:: yaml - name: ensure the auth directory exists with correct ownership vyos.vyos.vyos_file: dest: /config/auth/office-vpn owner: openvpn group: openvpn mode: '0750' - name: push a client certificate with correct ownership vyos.vyos.vyos_file: dest: /config/auth/office-vpn/client.pem src: files/office-vpn-client.pem owner: openvpn group: openvpn mode: '0600' - name: remove a stale cert vyos.vyos.vyos_file: dest: /config/auth/old-vpn/client.pem state: absent + - name: push templated LDAP auth config (content is rendered by Ansible before this module runs) + vyos.vyos.vyos_file: + dest: /config/auth/office-vpn/ldap-auth.config + content: "{{ lookup('template', 'ldap_auth.config.j2') }}" + owner: openvpn + group: openvpn + mode: '0640' + Return Values ------------- Common return values are documented `here `_, the following are the fields unique to this module: .. raw:: html
Key Returned Description
diff_fields
list / elements=string
always
Fields that differed between requested and actual state and were converged.

Sample:
['owner', 'mode', 'content']


Status ------ Authors ~~~~~~~ - VyOS maintainers and contributors (@vyos) diff --git a/plugins/modules/vyos_file.py b/plugins/modules/vyos_file.py index 4e0d49b5..2dd4533c 100644 --- a/plugins/modules/vyos_file.py +++ b/plugins/modules/vyos_file.py @@ -1,430 +1,497 @@ #!/usr/bin/python # -*- coding: utf-8 -*- # Copyright: (c) 2026, VyOS maintainers and contributors # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) from __future__ import absolute_import, division, print_function __metaclass__ = type DOCUMENTATION = """ module: vyos_file short_description: Manage files, directories, and their ownership on VyOS devices description: - Creates, updates, or removes a file or directory on a VyOS device, optionally pushing content from a local file (I(src)) or inline text (I(content)), and setting owner/group/mode via sudo chown/chmod. - This module does not touch the configuration tree (config.boot). It manages arbitrary filesystem paths such as certificates or auth files under /config/auth/, which are not tracked by commit/save/rollback. - All logic runs inside this module's main(), using the standard get_connection()/run_commands() pattern shared with vyos_command — there is no dedicated action plugin; this module uses the shared generic vyos action plugin like every other module in the collection. version_added: "6.0.0" author: - VyOS maintainers and contributors (@vyos) -extends_documentation_fragment: - - vyos.vyos.vyos options: dest: description: Absolute path to the remote file or directory to manage. type: path required: true state: description: Whether the path should exist (present) or be removed (absent). type: str choices: [present, absent] default: present src: description: - Path to a local file (on the Ansible controller) whose content should be - pushed to I(dest). Read locally and pushed as base64 via a single CLI - command, since network_cli has no SFTP/SCP channel available to this - module. Mutually exclusive with I(content). + pushed to I(dest). Transferred via a real SCP session over the + connection's own persistent socket (the same mechanism + M(ansible.netcommon.net_put) uses), never placed inside a command + string. Mutually exclusive with I(content). + - File bytes are uploaded exactly as they exist on disk — Ansible does + not render Jinja expressions inside the file's contents for I(src), + only in the option values of the task itself (e.g. a templated path + string). To push templated text, render it first with the C(template) + lookup and pass the result via I(content) instead. type: path content: description: - Inline text content to write to I(dest). Marked no_log, since this module is commonly used to push credential material. Mutually exclusive with I(src). + - Since I(content) is a normal string-type module option, Ansible renders + any Jinja expressions in it (e.g. C({{ my_var }})) before this module + ever runs, the same as any other option value — no special templating + support is implemented by this module itself. type: str owner: description: Name of the user that should own I(dest). type: str group: description: Name of the group that should own I(dest). type: str mode: description: - Permission bits for I(dest), as a string (e.g. '0600'). Compared against stat output after normalizing to 4 digits; '600' and '0600' are treated as equivalent. type: str become: description: Whether to prefix remote commands with sudo. type: bool default: true notes: - This module works with connection C(ansible.netcommon.network_cli). - - Tested against VyOS 1.4.2 and 1.5.0. - File state managed by this module is independent of VyOS's config revision system. A rollback to a previous config revision will not revert changes made by this module. - Paths under I(/config/auth) are deliberately setgid C(vyattacfg) by VyOS's own config-management convention (see vyos.dev T2713). If I(mode) is given with a leading digit of C(0) (e.g. C('0750')), this module compares only the rwx bits and will not report a diff for VyOS's own setgid bit. To manage the setgid/setuid/sticky bit explicitly, pass a non-zero leading digit (e.g. C('2750')). """ EXAMPLES = """ - name: ensure the auth directory exists with correct ownership vyos.vyos.vyos_file: dest: /config/auth/office-vpn owner: openvpn group: openvpn mode: '0750' - name: push a client certificate with correct ownership vyos.vyos.vyos_file: dest: /config/auth/office-vpn/client.pem src: files/office-vpn-client.pem owner: openvpn group: openvpn mode: '0600' - name: remove a stale cert vyos.vyos.vyos_file: dest: /config/auth/old-vpn/client.pem state: absent + +- name: push templated LDAP auth config (content is rendered by Ansible before this module runs) + vyos.vyos.vyos_file: + dest: /config/auth/office-vpn/ldap-auth.config + content: "{{ lookup('template', 'ldap_auth.config.j2') }}" + owner: openvpn + group: openvpn + mode: '0640' """ RETURN = """ diff_fields: description: Fields that differed between requested and actual state and were converged. returned: always type: list elements: str sample: ["owner", "mode", "content"] """ -import base64 import hashlib import os import re import shlex +import tempfile from ansible.module_utils.basic import AnsibleModule from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.vyos import ( + get_connection, run_commands, ) from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.vyos_file import ( build_want, diff_want_have, parse_stat, ) ARGUMENT_SPEC = dict( dest=dict(type="path", required=True), state=dict(type="str", choices=["present", "absent"], default="present"), src=dict(type="path"), content=dict(type="str", no_log=True), owner=dict(type="str"), group=dict(type="str"), mode=dict(type="str"), become=dict(type="bool", default=True), ) def get_have(module, become, dest, need_content_hash=False): quoted_dest = shlex.quote(dest) # check_rc=False is required here: a missing path is a normal, expected # outcome on first-run creation, not a failure. With the default # check_rc=True, run_commands() would call module.fail_json() on every # "file doesn't exist yet" case, which is exactly the case we need to # handle gracefully to build `have`. responses = run_commands( module, ["{0}stat --format='%a %U %G %s' {1}".format(become, quoted_dest)], check_rc=False, ) out = responses[0] if responses else "" if not out: return None if "No such file" in out: return None have = parse_stat(out) if have is None: # Anything that isn't the specific "doesn't exist" message and # doesn't parse as valid stat output is a real problem — permission # denied, I/O error, unexpected format, etc. Fail loudly rather than # silently treating it as "create it", which could otherwise lead # this module to attempt mkdir/chown/chmod against a path it # actually has no real visibility into. module.fail_json( msg="vyos_file: unexpected stat output for {0}: {1}".format(dest, out.strip()), ) if need_content_hash: # Only hash when content comparison actually matters (src/content # given) — no need to pay this cost for plain directory/ownership # management. Without this, `have["content_hash"]` would always be # None, so `content` would show as "different" forever, even right # after a successful write. hash_responses = run_commands( module, ["{0}sha256sum {1}".format(become, quoted_dest)], check_rc=False, ) hash_out = hash_responses[0] if hash_responses else "" # sha256sum output format: " " parts = hash_out.strip().split() if parts and len(parts[0]) == 64 and all(c in "0123456789abcdef" for c in parts[0].lower()): have["content_hash"] = parts[0] # else: leave content_hash unset — a malformed/errored sha256sum # (e.g. the file vanished in a race between stat and sha256sum) # should surface as a real diff on the next comparison, not get # silently recorded as a bogus "hash". return have _OCTAL_DIGIT_TO_SYMBOLIC = { "0": "", "1": "x", "2": "w", "3": "wx", "4": "r", "5": "rx", "6": "rw", "7": "rwx", } def _rwx_digits_to_symbolic_mode(mode4): """Convert the last 3 digits of a normalized 4-digit mode string into a symbolic chmod argument (e.g. "0750" -> "u=rwx,g=rx,o="). Symbolic mode assignment for u/g/o only touches those classes — unlike any numeric chmod form, it leaves existing setuid/setgid/sticky bits untouched unless explicitly referenced (u+s, g+s, +t), which is exactly the "special bits are unmanaged for implicit mode requests" guarantee this module's docs and diff logic already promise but a plain numeric chmod would silently violate. """ u, g, o = mode4[-3], mode4[-2], mode4[-1] return "u={0},g={1},o={2}".format( _OCTAL_DIGIT_TO_SYMBOLIC[u], _OCTAL_DIGIT_TO_SYMBOLIC[g], _OCTAL_DIGIT_TO_SYMBOLIC[o], ) def _build_chmod_command(become, mode4, quoted_dest): if mode4[0] == "0": # Implicit special bits (caller didn't ask for them): use symbolic # mode so existing setuid/setgid/sticky bits survive. A numeric # chmod here — even a bare 3-digit form — always explicitly sets # the special-bits digit to 0, silently clearing e.g. VyOS's own # setgid convention on /config/auth (vyos.dev T2713) the moment any # rwx change is needed, rather than genuinely leaving it unmanaged. symbolic = _rwx_digits_to_symbolic_mode(mode4) return "{0}chmod {1} {2}".format(become, shlex.quote(symbolic), quoted_dest) # Explicit non-zero leading digit: caller wants exact control over # special bits too, so a plain numeric chmod is correct here. return "{0}chmod {1} {2}".format(become, shlex.quote(mode4), quoted_dest) def local_content_hash(params): if params.get("src"): h = hashlib.sha256() with open(params["src"], "rb") as f: for chunk in iter(lambda: f.read(65536), b""): h.update(chunk) return h.hexdigest() if params.get("content") is not None: return hashlib.sha256(params["content"].encode()).hexdigest() return None def read_local_bytes(params): if params.get("src"): with open(params["src"], "rb") as f: return f.read() if params.get("content") is not None: return params["content"].encode() return None +def push_content_via_scp(module, connection, dest, params): + # Real SCP transfer over the connection's own persistent SSH session — + # content/src bytes never appear inside a command string sent through + # run_commands(). The earlier base64-in-a-shell-command approach was + # only ever encoded, not encrypted, and remained fully readable to + # anything logging connection traffic (e.g. persistent connection + # logging), regardless of no_log on the task — a real problem given + # this module's actual purpose (VPN certs, LDAP credentials). + # + # net_put's own action plugin uses this exact mechanism — connection + # here is get_connection(module), the same Connection(module._socket_path) + # JSON-RPC proxy net_put builds via Connection(socket_path) — so this is + # not action-plugin-only, despite that being true historically for some + # other network_cli file-transfer patterns. + cleanup_local = False + if params.get("src"): + local_path = params["src"] + else: + data = read_local_bytes(params) + fd, local_path = tempfile.mkstemp(prefix="vyos_file_") + cleanup_local = True + try: + with os.fdopen(fd, "wb") as f: + f.write(data) + except Exception: + os.remove(local_path) + raise + + try: + timeout = connection.get_option("persistent_command_timeout") + connection.copy_file( + source=local_path, + destination=dest, + proto="scp", + timeout=timeout, + ) + finally: + if cleanup_local: + os.remove(local_path) + + def converge(module, become, dest, want, diff, params): cmds = [] quoted_dest = shlex.quote(dest) if want["state"] == "absent": cmds.append("{0}rm -rf {1}".format(become, quoted_dest)) run_commands(module, cmds) post_have = get_have(module, become, dest) if post_have is not None: module.fail_json( msg="vyos_file: removal of {0} did not take effect".format(dest), ) return if "content" in diff: - data = read_local_bytes(params) - b64 = base64.b64encode(data).decode() - # Build the whole inner script as one plain string, quoting dest - # within it, then quote the ENTIRE script once as a single argument - # to `sh -c`. Do not nest a shlex.quote()'d fragment inside a - # separately-quoted outer string (e.g. double quotes) — if dest - # contains a single quote, shlex.quote()'s escaping introduces a - # literal double quote into its output, which would prematurely - # close a surrounding double-quoted wrapper and reintroduce the - # exact shell-injection risk quoting was meant to prevent. - script = "echo {0} | base64 -d > {1}".format(b64, shlex.quote(dest)) - cmds.append("{0}sh -c {1}".format(become, shlex.quote(script))) + connection = get_connection(module) + push_content_via_scp(module, connection, dest, params) elif "state" in diff and have_is_missing(diff): cmds.append("{0}mkdir -p {1}".format(become, quoted_dest)) if "owner" in diff and "group" in diff: cmds.append( "{0}chown {1}:{2} {3}".format( become, shlex.quote(want["owner"]), shlex.quote(want["group"]), quoted_dest, ), ) elif "owner" in diff: cmds.append( "{0}chown {1} {2}".format(become, shlex.quote(want["owner"]), quoted_dest), ) elif "group" in diff: cmds.append( "{0}chgrp {1} {2}".format(become, shlex.quote(want["group"]), quoted_dest), ) if "mode" in diff: cmds.append(_build_chmod_command(become, want["mode"], quoted_dest)) if cmds: run_commands(module, cmds) # run_commands() only confirms the CLI accepted each command line # syntactically — it does NOT confirm the underlying binary succeeded. # A chown against a nonexistent group, for example, prints an error to # stdout but the CLI wrapper still reports the line as "executed"; we # would otherwise report changed=true for a write that silently did # nothing. Re-stat and compare against `want` to catch this class of # failure before returning success. post_have = get_have( module, become, dest, need_content_hash=want.get("content_hash") is not None, ) post_diff = diff_want_have(want, post_have) if post_diff: module.fail_json( msg=( "vyos_file converged but post-check found remaining " "differences — one or more commands likely failed silently " "at the OS level (e.g. chown to a nonexistent user/group): " "{0}".format(post_diff) ), ) def have_is_missing(diff): return diff.get("state") == (None, "present") def validate_dest(module, dest): # dest is type=path in ARGUMENT_SPEC, which expands ~ and env vars but # does NOT enforce absoluteness — a relative value would resolve against # whatever the underlying shell's cwd happens to be, an unintended and # unpredictable target. And since this module issues raw `rm -rf`, # `chmod`, `chown` against dest with no config-tree safety net, a # dest of "/" (or anything that normalizes to it) combined with # state=absent would attempt to recursively remove the entire # filesystem. Both must be rejected before any stat/converge runs. if not os.path.isabs(dest): module.fail_json( msg="vyos_file: dest must be an absolute path, got {0!r}".format(dest), ) normalized = os.path.normpath(dest) # normalized == "/" alone is insufficient: os.path.normpath preserves # "//" as-is (a POSIX quirk permitting implementation-defined behavior # for exactly two leading slashes), so dest="//" would otherwise bypass # this check entirely. Stripping all slashes catches "/", "//", "///", # etc. uniformly. if normalized.strip("/") == "": module.fail_json( msg=( "vyos_file: refusing to manage the root filesystem path " "(dest normalized to {0!r}): {1!r}".format(normalized, dest) ), ) _MODE_RE = re.compile(r"^[0-7]{3,4}$") def validate_mode(module, mode): # _normalize_mode() (module_utils) does str(mode).zfill(4)[-4:], which # for genuinely invalid input silently mangles it into something that # LOOKS valid rather than rejecting it — e.g. "10640" (5 digits, an # obvious typo for a 4-digit mode) becomes "0640" by truncation, and # the module would silently apply permissions the caller never actually # asked for. Validate strictly here, before that normalization ever # runs, so malformed input fails loudly instead of being reinterpreted. if mode is None: return if not _MODE_RE.match(mode): module.fail_json( msg=( "vyos_file: mode must be an octal string of 3 or 4 digits " "(0-7 only), got {0!r}".format(mode) ), ) +def validate_src(module, src): + # local_content_hash()/read_local_bytes() do plain open(src, "rb") + # calls with no existence/type/permission check. A missing file, a + # directory passed where a file is expected, or an unreadable path + # would otherwise surface as an unhandled Python traceback instead of + # a clean module error — and this happens even under check_mode, since + # content-hashing runs before the check-mode short-circuit. + if src is None: + return + if not os.path.exists(src): + module.fail_json(msg="vyos_file: src not found: {0!r}".format(src)) + if os.path.isdir(src): + module.fail_json( + msg="vyos_file: src is a directory, expected a file: {0!r}".format(src), + ) + if not os.access(src, os.R_OK): + module.fail_json(msg="vyos_file: src is not readable: {0!r}".format(src)) + + def main(): module = AnsibleModule( argument_spec=ARGUMENT_SPEC, mutually_exclusive=[["src", "content"]], supports_check_mode=True, ) dest = module.params["dest"] validate_dest(module, dest) validate_mode(module, module.params.get("mode")) + validate_src(module, module.params.get("src")) become = "sudo " if module.params.get("become", True) else "" want = build_want(module.params, local_content_hash(module.params)) have = get_have( module, become, dest, need_content_hash=want.get("content_hash") is not None, ) diff = diff_want_have(want, have) result = {"changed": bool(diff), "diff_fields": list(diff.keys())} if module.check_mode or not diff: module.exit_json(**result) converge(module, become, dest, want, diff, module.params) module.exit_json(**result) if __name__ == "__main__": main() diff --git a/tests/unit/modules/network/vyos/test_vyos_file.py b/tests/unit/modules/network/vyos/test_vyos_file.py index ee2cf6af..64963173 100644 --- a/tests/unit/modules/network/vyos/test_vyos_file.py +++ b/tests/unit/modules/network/vyos/test_vyos_file.py @@ -1,411 +1,506 @@ # tests/unit/modules/network/vyos/test_vyos_file.py # # Mocks run_commands() directly — the real call path this module uses via # get_connection()/run_commands() in module_utils/network/vyos/vyos.py. # This replaces an earlier draft that mocked a bespoke ActionModule; that # design was abandoned once it turned out every module in this collection # (vyos_command, vyos_config, etc.) shares one generic action plugin and # puts real logic inside main() instead. from __future__ import absolute_import, division, print_function __metaclass__ = type import hashlib import json import os import tempfile -from unittest.mock import patch +from unittest.mock import MagicMock, patch from ansible_collections.vyos.vyos.plugins.modules import vyos_file from ansible_collections.vyos.vyos.tests.unit.modules.network.vyos.vyos_module import ( TestVyosModule, ) from ansible_collections.vyos.vyos.tests.unit.modules.utils import ( AnsibleExitJson, AnsibleFailJson, set_module_args, ) class TestVyosFileModule(TestVyosModule): module = vyos_file def setUp(self): super(TestVyosFileModule, self).setUp() self.mock_run_commands = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_file.run_commands", ) self.run_commands = self.mock_run_commands.start() + # content/src transfer now goes through a real SCP call via + # get_connection(module).copy_file(...) — never through + # run_commands() — so it needs its own mock, separate from the + # command-based stat/chown/chmod/rm path above. + self.mock_get_connection = patch( + "ansible_collections.vyos.vyos.plugins.modules.vyos_file.get_connection", + ) + self.get_connection = self.mock_get_connection.start() + self.mock_connection = MagicMock() + self.mock_connection.get_option.return_value = 30 + self.get_connection.return_value = self.mock_connection + def tearDown(self): super(TestVyosFileModule, self).tearDown() self.mock_run_commands.stop() + self.mock_get_connection.stop() # ---- helpers ----------------------------------------------------- def _queue(self, *responses): """Queue successive return values, one per run_commands() call.""" self.run_commands.side_effect = list(responses) def _run(self, args, expect_fail=False): set_module_args(args) exc = AnsibleFailJson if expect_fail else AnsibleExitJson with self.assertRaises(exc) as ctx: vyos_file.main() return ctx.exception.args[0] # ---- idempotency core --------------------------------------------- def test_creates_when_absent(self): # get_have() issues ONE stat call; converge() batches mkdir+chown+ # chmod into a SINGLE run_commands() call (not one call per # command); the post-check issues one more stat call. Three total # run_commands() invocations, matching the module's actual batching. self._queue( ["stat: cannot statx '/config/auth/x': No such file or directory"], ["", "", ""], # mkdir, chown, chmod — one batched call ["750 vyos vyattacfg 4096"], # post-check stat ) result = self._run( {"dest": "/config/auth/x", "owner": "vyos", "group": "vyattacfg", "mode": "0750"}, ) self.assertTrue(result["changed"]) self.assertIn("state", result["diff_fields"]) self.assertIn("owner", result["diff_fields"]) def test_noop_when_converged(self): self._queue(["750 vyos vyattacfg 4096"]) result = self._run( {"dest": "/config/auth/x", "owner": "vyos", "group": "vyattacfg", "mode": "0750"}, ) self.assertFalse(result["changed"]) self.assertEqual(result["diff_fields"], []) def test_setgid_ignored_when_mode_leading_digit_is_zero(self): # /config/auth is deliberately setgid vyattacfg (vyos.dev T2713). # Requesting mode '0750' (leading digit 0) must NOT be reported as # different from an actual mode of 2750. self._queue(["2750 vyos vyattacfg 4096"]) result = self._run({"dest": "/config/auth/x", "mode": "0750"}) self.assertFalse(result["changed"], result.get("diff_fields")) def test_setgid_respected_when_explicitly_requested(self): # Explicit non-zero leading digit means the caller does care about # the special bits — since 2750 is requested and 2750 is already # there, this should be a no-op (only the initial stat call fires). self._queue(["2750 vyos vyattacfg 4096"]) result = self._run({"dest": "/config/auth/x", "mode": "2750"}) self.assertFalse(result["changed"]) def test_mode_change_detected(self): # Only mode differs, so converge() batches a single chmod command # (one run_commands() call), then the post-check stat is a second. self._queue( ["600 vyos vyattacfg 10"], [""], # chmod — the only mutating command needed ["640 vyos vyattacfg 10"], ) result = self._run({"dest": "/config/auth/x/hello.txt", "mode": "0640"}) self.assertTrue(result["changed"]) self.assertEqual(result["diff_fields"], ["mode"]) def test_mode_string_normalization(self): # "00640" (5 digits) is deliberately excluded here — it's now # correctly rejected by the strict [0-7]{3,4} validation (see # test_rejects_mode_with_extra_leading_digit), even though its # value is harmless. Only genuinely valid 3-4 digit forms of the # same value are expected to normalize equivalently. for requested in ("640", "0640"): with self.subTest(requested=requested): self._queue(["640 vyos vyattacfg 10"]) result = self._run({"dest": "/config/auth/x/hello.txt", "mode": requested}) self.assertFalse( result["changed"], "mode {0!r} incorrectly compared unequal to stat's '640'".format(requested), ) def test_implicit_mode_uses_symbolic_chmod_preserving_special_bits(self): # Real bug found in review: a plain numeric chmod ALWAYS explicitly # sets the special-bits digit (even a bare 3-digit form implies a # leading 0), so it would silently clear an existing setgid/setuid # bit the moment any rwx change is needed — directly contradicting # the "special bits are unmanaged for implicit mode" guarantee this # module's own diff comparison already promises. Symbolic chmod # (u=,g=,o=) is the only form that genuinely leaves them untouched. self._queue( ["2770 vyos vyattacfg 10"], # existing: setgid + rwxrwx--- [""], # the single batched chmod command ["2750 vyos vyattacfg 10"], # post-check: rwx fixed, setgid survived ) result = self._run({"dest": "/x", "mode": "0750"}) self.assertTrue(result["changed"]) self.assertEqual(result["diff_fields"], ["mode"]) converge_call = self.run_commands.call_args_list[1] chmod_cmd = converge_call.args[1][0] self.assertIn("u=", chmod_cmd, "expected symbolic chmod for an implicit mode request") self.assertNotRegex( chmod_cmd, r"chmod\s+0?750\b", "must not use a numeric chmod for an implicit mode request — it would " "clear the existing setgid bit", ) def test_explicit_mode_uses_numeric_chmod(self): # A non-zero leading digit means the caller explicitly wants control # over special bits too — numeric chmod is correct here, unlike the # implicit case above. self._queue( ["0750 vyos vyattacfg 10"], # existing: no special bits [""], ["2750 vyos vyattacfg 10"], # post-check: matches the explicit request ) result = self._run({"dest": "/x", "mode": "2750"}) self.assertTrue(result["changed"]) converge_call = self.run_commands.call_args_list[1] chmod_cmd = converge_call.args[1][0] self.assertIn("2750", chmod_cmd) self.assertNotIn("u=", chmod_cmd, "explicit mode should use numeric chmod, not symbolic") # ---- content --------------------------------------------------------- def test_content_push_detected_and_verified(self): - # need_content_hash is True (content was given), but get_have() - # only actually hashes when the path already exists — on the - # initial (absent) lookup it's skipped, so that first call is a - # single stat. converge() batches base64-write + chown into one - # call. The post-check, now that the file exists, issues stat AND - # sha256sum as two separate calls. The queued hash must be the - # REAL sha256("hi\n") hex digest, or the module's own post-check - # will (correctly) call fail_json on a genuine mismatch. + # Content transfer now goes through connection.copy_file() (mocked + # via self.mock_connection), not run_commands() — so the converge + # batch here only contains chown+chmod (2 items), not the old + # 3-item base64-write+chown+chmod batch. real_hash = "98ea6e4f216f2fb4b69fff9b3a44842c38686ca685f3f55dc48c5d3fb1107be4" self._queue( ["stat: cannot statx '/config/auth/x/hello.txt': No such file or directory"], - ["", "", ""], # base64 write, chown, chmod — batched (mode was also requested) + ["", ""], # chown, chmod — batched (content push is separate now) ["600 vyos vyattacfg 10"], # post-check stat ["{0} /config/auth/x/hello.txt".format(real_hash)], # post-check sha256sum ) result = self._run( { "dest": "/config/auth/x/hello.txt", "content": "hi\n", "owner": "vyos", "mode": "0600", }, ) self.assertTrue(result["changed"]) self.assertIn("content", result["diff_fields"]) + self.mock_connection.copy_file.assert_called_once() + self.assertEqual( + self.mock_connection.copy_file.call_args.kwargs["destination"], + "/config/auth/x/hello.txt", + ) def test_src_upload_reads_local_file_and_pushes_content(self): # src takes a different code path from content (read_local_bytes() # opens the local path rather than encoding an inline string), and # had no direct test coverage — this exercises that path explicitly # using a real temporary file, since local_content_hash()/ # read_local_bytes() do plain open() calls that aren't mockable # through run_commands. with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as f: f.write("-----BEGIN CERTIFICATE-----\nfakecertdata\n-----END CERTIFICATE-----\n") local_path = f.name try: real_hash = hashlib.sha256( b"-----BEGIN CERTIFICATE-----\nfakecertdata\n-----END CERTIFICATE-----\n", ).hexdigest() self._queue( ["stat: cannot statx '/config/auth/x/client.pem': No such file or directory"], - ["", "", ""], # base64 write, chown, chmod + ["", ""], # chown, chmod — batched (transfer is via copy_file, separate) ["600 vyos vyattacfg 10"], ["{0} /config/auth/x/client.pem".format(real_hash)], ) result = self._run( { "dest": "/config/auth/x/client.pem", "src": local_path, "owner": "vyos", "mode": "0600", }, ) self.assertTrue(result["changed"]) self.assertIn("content", result["diff_fields"]) + self.mock_connection.copy_file.assert_called_once() + self.assertEqual( + self.mock_connection.copy_file.call_args.kwargs["source"], + local_path, + ) finally: os.unlink(local_path) def test_src_upload_idempotent_on_matching_remote_content(self): data = b"identical content\n" with tempfile.NamedTemporaryFile(mode="wb", suffix=".txt", delete=False) as f: f.write(data) local_path = f.name try: real_hash = hashlib.sha256(data).hexdigest() self._queue( ["600 vyos vyattacfg 10"], ["{0} /x".format(real_hash)], ) result = self._run({"dest": "/x", "src": local_path, "owner": "vyos", "mode": "0600"}) self.assertFalse(result["changed"], result.get("diff_fields")) finally: os.unlink(local_path) def test_content_hash_looked_up_only_when_relevant(self): # plain ownership/mode management on an existing path should never # trigger a sha256sum call — that's the whole point of the # need_content_hash gate. self._queue(["750 vyos vyattacfg 4096"]) self._run({"dest": "/config/auth/x", "mode": "0750"}) called_commands = [c.args[1] for c in self.run_commands.call_args_list] joined = " ".join(str(c) for c in called_commands) self.assertNotIn("sha256sum", joined) # ---- absent state ---------------------------------------------------- def test_absent_on_existing_removes(self): self._queue( ["600 vyos vyattacfg 10"], [""], # rm -rf ["stat: cannot statx '/config/auth/x/hello.txt': No such file or directory"], ) result = self._run({"dest": "/config/auth/x/hello.txt", "state": "absent"}) self.assertTrue(result["changed"]) self.assertEqual(result["diff_fields"], ["state"]) def test_absent_noop_when_already_gone(self): self._queue(["stat: cannot statx '/x': No such file or directory"]) result = self._run({"dest": "/x", "state": "absent"}) self.assertFalse(result["changed"]) def test_real_stat_error_fails_loudly_instead_of_treated_as_missing(self): # Permission denied (or any other real stat failure) must NOT be # silently treated the same as "doesn't exist" — that could lead # the module to attempt mkdir/chown/chmod against a path it # actually has no real visibility into. self._queue(["stat: cannot statx '/x': Permission denied"]) result = self._run({"dest": "/x", "mode": "0750"}, expect_fail=True) self.assertIn("unexpected stat output", result["msg"]) # ---- destination path validation --------------------------------- def test_rejects_relative_path(self): # No run_commands() calls should even be attempted for an invalid # dest — validation must happen before any stat/converge logic. result = self._run({"dest": "relative/path"}, expect_fail=True) self.assertIn("absolute path", result["msg"]) self.assertEqual(self.run_commands.call_count, 0) def test_rejects_root_path(self): result = self._run({"dest": "/", "state": "absent"}, expect_fail=True) self.assertIn("root filesystem", result["msg"]) self.assertEqual(self.run_commands.call_count, 0) def test_rejects_double_slash_root_bypass(self): # os.path.normpath preserves "//" as-is (a POSIX quirk for exactly # two leading slashes) rather than collapsing it to "/" — a naive # `normalized == "/"` check would miss this and let it through. result = self._run({"dest": "//", "state": "absent"}, expect_fail=True) self.assertIn("root filesystem", result["msg"]) self.assertEqual(self.run_commands.call_count, 0) def test_rejects_dot_path_that_normalizes_to_root(self): result = self._run({"dest": "/.", "state": "absent"}, expect_fail=True) self.assertIn("root filesystem", result["msg"]) self.assertEqual(self.run_commands.call_count, 0) # ---- mode validation ----------------------------------------------- def test_rejects_mode_with_extra_leading_digit(self): # The exact real bug found in review: _normalize_mode()'s # zfill(4)[-4:] would silently truncate "10640" into "0640" rather # than rejecting an obviously malformed 5-digit value — applying # permissions the caller never actually asked for. result = self._run({"dest": "/x", "mode": "10640"}, expect_fail=True) self.assertIn("octal string", result["msg"]) self.assertEqual(self.run_commands.call_count, 0) def test_rejects_non_octal_digits(self): result = self._run({"dest": "/x", "mode": "0890"}, expect_fail=True) self.assertIn("octal string", result["msg"]) self.assertEqual(self.run_commands.call_count, 0) def test_rejects_non_numeric_mode(self): result = self._run({"dest": "/x", "mode": "abcd"}, expect_fail=True) self.assertIn("octal string", result["msg"]) self.assertEqual(self.run_commands.call_count, 0) def test_rejects_too_short_mode(self): result = self._run({"dest": "/x", "mode": "07"}, expect_fail=True) self.assertIn("octal string", result["msg"]) self.assertEqual(self.run_commands.call_count, 0) def test_accepts_valid_3_and_4_digit_modes(self): # Sanity check that the new strict validation doesn't reject # legitimate input alongside the malformed cases above. for valid_mode in ("750", "0750", "2750", "0000", "7777"): with self.subTest(valid_mode=valid_mode): self._queue([" ".join([valid_mode.zfill(4), "vyos", "vyattacfg", "10"])]) result = self._run({"dest": "/x", "mode": valid_mode}) self.assertFalse(result["changed"]) + # ---- src validation -------------------------------------------------- + + def test_rejects_missing_src_file(self): + # Without this check, open() inside local_content_hash() would + # raise an unhandled FileNotFoundError instead of a clean module + # error — and this happens even under check_mode, since content + # hashing runs before the check-mode short-circuit. + result = self._run( + {"dest": "/x", "src": "/definitely/does/not/exist/x.pem"}, + expect_fail=True, + ) + self.assertIn("src not found", result["msg"]) + self.assertEqual(self.run_commands.call_count, 0) + + def test_rejects_src_that_is_a_directory(self): + result = self._run({"dest": "/x", "src": "/tmp"}, expect_fail=True) + self.assertIn("directory", result["msg"]) + self.assertEqual(self.run_commands.call_count, 0) + def test_malformed_sha256sum_output_does_not_get_recorded_as_a_hash(self): # If sha256sum itself errors (e.g. a race where the file vanished # between stat and sha256sum), the garbage output must not be # silently trusted as a real content hash — that would corrupt the # comparison instead of surfacing as a real, visible diff. # check_mode=True keeps this isolated to have/diff computation only, # without needing to model a full converge cycle. self._queue( ["600 vyos vyattacfg 10"], ["sha256sum: /x: No such file or directory"], ) set_module_args({"dest": "/x", "content": "hi\n", "_ansible_check_mode": True}) with self.assertRaises(AnsibleExitJson) as ctx: vyos_file.main() result = ctx.exception.args[0] # have.content_hash stays unset -> compared against a real want hash # -> reported as a genuine diff, not silently accepted as converged. self.assertIn("content", result.get("diff_fields", [])) # ---- silent-failure detection (the real bug this caught on hardware) -- def test_post_check_fails_module_when_chown_silently_no_ops(self): # Reproduces the real failure found on hardware: chown to a # nonexistent group prints an error but the CLI still reports the # line as "executed" with rc 0 — have must be re-verified. # converge() batches mkdir+chown into one call (mode wasn't # requested, so no chmod); the second queued item represents that # single batched call's two responses. self._queue( ["stat: cannot statx '/x': No such file or directory"], ["", "chown: invalid group: 'x:bogus'"], # mkdir ok, chown failed ["644 root nogroup 4096"], # post-check: neither owner nor group took ) result = self._run({"dest": "/x", "owner": "vyos", "group": "bogus"}, expect_fail=True) self.assertIn("post-check", result["msg"]) # ---- check_mode -------------------------------------------------------- def test_check_mode_reports_diff_without_converging(self): self._queue(["600 vyos vyattacfg 10"]) set_module_args({"dest": "/x", "mode": "0640", "_ansible_check_mode": True}) with self.assertRaises(AnsibleExitJson) as ctx: vyos_file.main() result = ctx.exception.args[0] self.assertTrue(result["changed"]) self.assertEqual(result["diff_fields"], ["mode"]) # only the initial stat call should have happened — no chmod self.assertEqual(self.run_commands.call_count, 1) # ---- secrets discipline ------------------------------------------------ + def test_content_push_via_scp_creates_and_cleans_up_temp_file_for_inline_content(self): + # For `content` (no src), a real local temp file must be created to + # hand to copy_file() (SCP needs a real source path — it can't + # stream an in-memory string), and that temp file must be removed + # again afterward regardless of outcome, since it briefly holds + # secret material on the controller's local disk. + captured_path = {} + + def fake_copy_file(source, destination, proto, timeout): + captured_path["source"] = source + # the temp file must exist at the moment copy_file is invoked + self.assertTrue(os.path.exists(source)) + with open(source, "rb") as f: + self.assertEqual(f.read(), b"hi\n") + + self.mock_connection.copy_file.side_effect = fake_copy_file + self._queue( + ["stat: cannot statx '/x': No such file or directory"], + ["600 vyos vyattacfg 10"], + [hashlib.sha256(b"hi\n").hexdigest() + " /x"], + ) + self._run({"dest": "/x", "content": "hi\n"}) + # cleaned up after the transfer completes — nothing sensitive left + # sitting on the controller's local disk + self.assertFalse(os.path.exists(captured_path["source"])) + + def test_content_push_via_scp_uses_src_path_directly_without_a_temp_file(self): + # When src is given, the provided path IS the source — no temp + # file should be created or deleted for it. + with tempfile.NamedTemporaryFile(mode="w", delete=False) as f: + f.write("real file content\n") + local_path = f.name + try: + self._queue( + ["stat: cannot statx '/x': No such file or directory"], + ["600 vyos vyattacfg 10"], + [hashlib.sha256(b"real file content\n").hexdigest() + " /x"], + ) + self._run({"dest": "/x", "src": local_path}) + self.assertEqual( + self.mock_connection.copy_file.call_args.kwargs["source"], + local_path, + ) + # the caller's own file must still exist — this module must + # never delete a user-provided src path + self.assertTrue(os.path.exists(local_path)) + finally: + os.unlink(local_path) + def test_content_not_echoed_in_result(self): + # No owner/group/mode requested here, so converge()'s cmds list + # stays empty (content push is separate, via copy_file) — meaning + # the old "chown/chmod batch" run_commands call doesn't happen at + # all in this scenario. Only 3 run_commands calls total: initial + # stat, post-check stat, post-check sha256sum. real_hash = "03767fbe485736bb40cc5d85e4c9bb10b12a415674b46faf005aa22188a39a10" self._queue( ["stat: cannot statx '/x': No such file or directory"], - [""], # single batched command: base64 write only (no owner/mode given) ["600 root root 4"], # post-check stat ["{0} /x".format(real_hash)], # post-check sha256sum ) result = self._run({"dest": "/x", "content": "super-secret-value"}) self.assertNotIn("super-secret-value", json.dumps(result)) + # the secret must not leak into the copy_file() call args either — + # only a real local temp-file path should appear there + for call in self.mock_connection.copy_file.call_args_list: + self.assertNotIn("super-secret-value", str(call))