Refactor: Fixes on AVDVaultHandler (#6509)

Author: carl-baillargeon

ansible_collections/arista/avd/plugins/action/validate_inputs.py

@@ -174,15 +174,15 @@ def main(self, task_vars: dict[str, Any]) -> None:
         mp_workers, mt_workers = get_workers(len(hosts_to_process), task_vars["ansible_forks"])
         templated_path, validated_path = get_role_tmp_paths(role_name=SCHEMA_MAP[plugin_args.schema_name], tmp_dir=plugin_args.tmp_dir, clean=True)
 
-        # Create vault and file handlers.
+        # Create Vault and file handlers.
         vault_handler = AVDVaultHandler(self._loader, vault_id=plugin_args.vault_id)
         file_handler = AVDFileHandler(vault_handler)
 
-        # Check if vault is configured for encrypting temporary files.
-        if vault_handler.has_vault:
-            self.logger.info("Ansible Vault is configured - temporary files will be encrypted")
+        # Check if Vault secrets are configured for encrypting temporary files.
+        if vault_handler.has_vault_secrets:
+            self.logger.info("Ansible Vault secrets are configured - temporary files will be encrypted")
         else:
-            self.logger.info("Ansible Vault is not configured - temporary files will not be encrypted")
+            self.logger.info("Ansible Vault secrets are not configured - temporary files will not be encrypted")
 
         # Track worker failures globally for the task.
         self.crashed_hosts = set()
@@ -521,7 +521,7 @@ def _validate_host_worker(
         if not input_file_path.exists():
             return WorkerFailure(hostname=hostname, error=f"Missing input data file: {input_file_path}")
 
-        # Load file content (decrypted if vaulted).
+        # Load file content (decrypted if Vault encrypted).
         file_content = file_handler.read_file(input_file_path)
 
         if input_suffix in {"yml", "yaml"}:

ansible_collections/arista/avd/plugins/plugin_utils/utils/avd_file_handler.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2026 Arista Networks, Inc.
 # Use of this source code is governed by the Apache License 2.0
 # that can be found in the LICENSE file.
-"""File handler with automatic vault support for AVD plugins."""
+"""File handler with automatic Ansible Vault support for AVD plugins."""
 
 from __future__ import annotations
 
@@ -14,7 +14,7 @@
 
 
 class AVDFileHandler:
-    """Handles file operations with automatic vault support."""
+    """Handles file operations with automatic Ansible Vault support."""
 
     _vault_handler: AVDVaultHandler
 
@@ -23,30 +23,30 @@ def __init__(self, vault_handler: AVDVaultHandler) -> None:
         Initialize the file handler.
 
         Args:
-            vault_handler: The AVDVaultHandler instance to use for vault operations.
+            vault_handler: The AVDVaultHandler instance to use for Vault operations.
         """
         self._vault_handler = vault_handler
 
     def read_file(self, file_path: Path | str) -> bytes:
         """
-        Read a file with automatic vault decryption, returning raw bytes.
+        Read a file with automatic Vault decryption, returning raw bytes.
 
-        This method reads a file and automatically decrypts it if it is vault encrypted.
+        This method reads a file and automatically decrypts it if it is Vault encrypted.
 
         Args:
             file_path: Path to the file to read.
 
         Returns:
-            Raw file content as bytes (decrypted if vaulted).
+            Raw file content as bytes (decrypted if Vault encrypted).
         """
         file_content = Path(file_path).read_bytes()
         return self._vault_handler.decrypt_if_needed(file_content)
 
     def write_file(self, file_path: Path | str, data: bytes) -> None:
         """
-        Write bytes to a file with automatic vault encryption.
+        Write bytes to a file with automatic Vault encryption.
 
-        This method encrypts the data if vault is configured, then writes to the file.
+        This method encrypts the data if Vault secrets are configured, then writes to the file.
 
         Args:
             file_path: Path to the file to write.
@@ -57,7 +57,7 @@ def write_file(self, file_path: Path | str, data: bytes) -> None:
 
     def load_json(self, file_path: Path | str) -> Any:
         """
-        Load and parse a JSON file with automatic vault decryption.
+        Load and parse a JSON file with automatic Vault decryption.
 
         Args:
             file_path: Path to the JSON file to load.

ansible_collections/arista/avd/plugins/plugin_utils/utils/avd_vault_handler.py

@@ -11,71 +11,64 @@
 
 if TYPE_CHECKING:
     from ansible.parsing.dataloader import DataLoader
-    from ansible.parsing.vault import VaultSecret
 
 
 class AVDVaultHandler:
     """Handles Ansible Vault encryption and decryption operations."""
 
-    _loader: DataLoader
-    _encrypt_vault_id: str | None
-    _encrypt_secret: VaultSecret | None
+    __slots__ = ("__encrypt_secret", "__encrypt_vault_id", "__loader")
 
     def __init__(self, loader: DataLoader, vault_id: str | None = None) -> None:
         """
-        Initialize the vault handler.
+        Initialize the Vault handler.
 
         Args:
             loader: The Ansible DataLoader instance.
-            vault_id: Optional vault ID to use for encryption. If None, uses the first vault ID in the list (default Ansible behavior).
+            vault_id: Optional Vault ID to use for encryption. If None, uses the first Vault ID in the list (default Ansible behavior).
         """
-        self._loader = loader
+        self.__loader = loader
 
-        # Pre-compute encryption secret if vault is configured
+        # Pre-compute encryption credentials if Vault secrets are configured.
         if loader._vault.secrets:
-            self._encrypt_vault_id, self._encrypt_secret = match_encrypt_secret(loader._vault.secrets, vault_id)
+            self.__encrypt_vault_id, self.__encrypt_secret = match_encrypt_secret(loader._vault.secrets, vault_id)
         else:
-            self._encrypt_vault_id = None
-            self._encrypt_secret = None
+            self.__encrypt_vault_id = None
+            self.__encrypt_secret = None
 
     @property
-    def has_vault(self) -> bool:
-        """Whether vault secrets are configured."""
-        return bool(self._loader._vault.secrets)
+    def has_vault_secrets(self) -> bool:
+        """Whether Vault secrets are configured."""
+        return self.__encrypt_secret is not None
 
     def encrypt_if_needed(self, data: bytes) -> bytes:
         """
-        Encrypt data if vault secrets are configured.
+        Encrypt data if needed.
 
         Args:
             data: Data to potentially encrypt.
 
         Returns:
-            Encrypted data if vault is configured, otherwise the original data.
-
+            Encrypted data if Vault secrets are configured, otherwise the original data.
         """
-        if not self.has_vault:
+        if not self.has_vault_secrets:
             return data
 
-        return self._loader._vault.encrypt(data, secret=self._encrypt_secret, vault_id=self._encrypt_vault_id)
+        return self.__loader._vault.encrypt(data, secret=self.__encrypt_secret, vault_id=self.__encrypt_vault_id)
 
     def decrypt_if_needed(self, data: bytes) -> bytes:
         """
-        Decrypt data if it is vault encrypted.
+        Decrypt data if needed.
 
         Args:
             data: Data to potentially decrypt.
 
         Returns:
-            Decrypted data if it was vault encrypted, otherwise the original data.
+            Decrypted data if Vault secrets are configured and data is encrypted, otherwise the original data.
 
         Raises:
-            AnsibleVaultError: If vault decryption fails.
+            AnsibleVaultError: If Vault decryption fails.
         """
-        if self.has_vault and self._loader._vault.is_encrypted(data):
-            # Vault is configured and data is encrypted, decrypt it
-            decrypted_data, _vault_id, _vault_secret = self._loader._vault.decrypt_and_get_vault_id(data)
-            return decrypted_data
+        if self.has_vault_secrets and self.__loader._vault.is_encrypted(data):
+            return self.__loader._vault.decrypt(data)
 
-        # Data is not vaulted or no vault configured - return data as-is
         return data