""" Ansible module to run a restic backup or restore against an SFTP-backed repository. """ from __future__ import annotations import dataclasses import os import subprocess # nosec B404 import tempfile from typing import Any from ansible.module_utils.basic import AnsibleModule # type: ignore[import-untyped] def run_restic_cli(cmd: list[str], env: dict[str, str]) -> str: """ Run a restic CLI command and return its stdout, raising ValueError on a non-zero exit code. """ local_env = os.environ.copy() local_env.update(env) command_out = subprocess.run(cmd, capture_output=True, check=False, env=local_env, encoding="utf-8") # nosec B603 if command_out.returncode != 0: raise ValueError( f"\nError in executing command: {' '.join(cmd)}\nreturncode: {command_out.returncode}\n" f"stdout: {command_out.stdout}\nstderr: {command_out.stderr}" ) return command_out.stdout @dataclasses.dataclass class ResticOperationParams: """ Parameters for a restic backup or restore operation. """ ops: str restic_bin: str local_dir: str restic_host: str env: dict[str, str] extra_args: list[str] | None = None def run_restic_backup_restore(params: ResticOperationParams) -> dict[str, Any]: """ Run a restic backup or restore operation, then validate the repository with C(restic snapshots) and C(restic check). """ extra_args = params.extra_args or [] restic_bin = params.restic_bin env = params.env if params.ops == "backup": operation_stdout = run_restic_cli( [restic_bin, "backup", "--verbose", *extra_args, "--host", params.restic_host, params.local_dir], env ) else: operation_stdout = run_restic_cli( [ restic_bin, "restore", "--verbose", "latest", "--target", params.local_dir, *extra_args, "--host", params.restic_host, ], env, ) snapshots_stdout = run_restic_cli([restic_bin, "snapshots", "--verbose", *extra_args], env) check_stdout = run_restic_cli([restic_bin, "check", "--verbose", *extra_args], env) return { "operation_stdout": operation_stdout, "snapshots_stdout_lines": snapshots_stdout.splitlines(), "check_stdout_lines": check_stdout.splitlines(), } DOCUMENTATION = r""" --- module: restic_backup_restore_sftp short_description: Run a restic backup or restore against an SFTP-backed repository. version_added: "1.4.0" description: - Runs C(restic backup) or C(restic restore latest) against an SFTP-backed restic repository. - Writes O(private_key) to a mode C(0600) temporary file for the duration of the SFTP session and removes it afterwards, regardless of outcome. - Runs C(restic snapshots) and C(restic check) after a successful backup or restore to validate the SFTP repository. - Fails with the command output if any restic invocation returns a non-zero exit code. options: ops: description: Whether to back up to, or restore from, the SFTP repository. required: true type: str choices: [backup, restore] local_dir: description: Local directory backed up to, or restored from, the SFTP repository. required: true type: str restic_host: description: Value passed to C(--host) for the SFTP backup or restore. required: true type: str ssh_user: description: SSH user for the SFTP repository. required: true type: str ssh_host: description: SSH hostname or IP for the SFTP repository. required: true type: str repo_path: description: Remote repository path on the SFTP host. required: true type: str private_key: description: Ed25519 private key (PEM) used for SFTP authentication. required: true type: str repo_password: description: Encryption password of the SFTP-backed restic repository. required: true type: str extends_documentation_fragment: - restic_module_common """ EXAMPLES = r""" - name: Backup app data via SFTP restic_backup_restore_sftp: ops: backup local_dir: /var/lib/my-app restic_host: "{{ inventory_hostname }}" ssh_user: "{{ restic_repo_access.user }}" ssh_host: "{{ restic_repo_access.host }}" repo_path: "{{ restic_repo_access.path }}" private_key: "{{ restic_repo_access.private_key }}" repo_password: "{{ restic_repo_password.repository_password }}" """ RETURN = r""" operation_stdout: description: Standard output of the SFTP backup or restore command. type: str returned: always snapshots_stdout_lines: description: Line-split standard output of C(restic snapshots) for the SFTP repository. type: list elements: str returned: always check_stdout_lines: description: Line-split standard output of C(restic check) for the SFTP repository. type: list elements: str returned: always """ def run_module() -> None: """ Ansible main module """ module_args = { "ops": {"type": "str", "required": True, "choices": ["backup", "restore"]}, "local_dir": {"type": "str", "required": True}, "restic_host": {"type": "str", "required": True}, "ssh_user": {"type": "str", "required": True}, "ssh_host": {"type": "str", "required": True}, "repo_path": {"type": "str", "required": True}, "private_key": {"type": "str", "required": True, "no_log": True}, "repo_password": {"type": "str", "required": True, "no_log": True}, "restic_bin": {"type": "str", "required": False, "default": "/usr/local/bin/restic"}, } module = AnsibleModule(argument_spec=module_args, supports_check_mode=False) env = { "RESTIC_PASSWORD": module.params["repo_password"], "RESTIC_REPOSITORY": f"sftp:{module.params['ssh_user']}@{module.params['ssh_host']}" f":{module.params['repo_path']}", } try: with tempfile.NamedTemporaryFile(mode="w", delete=True) as key_file: key_file.write(module.params["private_key"]) key_file.flush() os.chmod(key_file.name, 0o600) sftp_args = f"sftp.args=-o IdentityFile={key_file.name} -o StrictHostKeyChecking=no" result = run_restic_backup_restore( ResticOperationParams( ops=module.params["ops"], restic_bin=module.params["restic_bin"], local_dir=module.params["local_dir"], restic_host=module.params["restic_host"], env=env, extra_args=["-o", sftp_args], ) ) module.exit_json(changed=True, **result) # pyright: ignore[reportUnknownMemberType] except ValueError as e: module.fail_json(msg=f"Error running restic: {e}", changed=False) # pyright: ignore[reportUnknownMemberType] def main() -> None: """ Python Main Module """ run_module() if __name__ == "__main__": main()