perf(vault): encrypt in-process instead of spawning ansible-vault per value

VaultHandler.encrypt_string started one ansible-vault process for every single credential. The swarm matrix' rotation gate between PASS 1 and PASS 2 drives 113 application subprocesses over 855 credential keys, so the step cost 1078s in run 33923190220 while the green baseline never ran it at all. Measured per value: 226ms via the CLI, 3.3ms via VaultLib, so 855 keys drop from 193s to 2.8s.

The snippet layout is a contract: encrypt_leaves splits the returned lines and strips the body indent, and existing inventories were written by the CLI. Verified byte-compatible (same 'name: !vault |' header, same 10-space body indent, same $ANSIBLE_VAULT;1.1;AES256 marker), and ansible-vault decrypt reads back what the new path writes.

Reading the password file delegates to ansible's own FileVaultSecret rather than approximating it: ansible strips, then decrypts a vaulted password file, then rejects an empty secret. A hand-rolled .strip() covered only the first of those. _vault stays lazy because callers construct VaultHandler with a password file that need not exist yet.

The tests replace two subprocess.run mocks that asserted against invented output. They now pin the layout encrypt_leaves depends on, and pin the guarantee from 052cea02e2 (values starting with a dash), which landed without a test.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tests/unit/python/cli/administration/inventory/credentials/test_vault.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
-import subprocess
+import tempfile
 import unittest
 import unittest.mock
+from pathlib import Path
+
+from ansible.errors import AnsibleError
 
 from cli.administration.inventory.credentials.vault import (
     _make_vault_scalar_from_text,
@@ -98,27 +101,46 @@ class TestVaultHandlerIntegration(unittest.TestCase):
     """Black-box coverage of the VaultHandler that ``to_vault_block``
     delegates to for plaintext encryption."""
 
-    def test_encrypt_string_success(self):
-        handler = VaultHandler("dummy_pw_file")
-        fake_output = "Encrypted data"
-        completed = subprocess.CompletedProcess(
-            args=["ansible-vault"], returncode=0, stdout=fake_output, stderr=""
-        )
-        with unittest.mock.patch("subprocess.run", return_value=completed) as proc_run:
-            result = handler.encrypt_string("plain_val", "name")
-            proc_run.assert_called_once()
-            self.assertEqual(result, fake_output)
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.password_file = Path(self._tmp.name) / "vault.pw"
+        self.password_file.write_text("unit-test-vault-password\n", encoding="utf-8")
+        self.handler = VaultHandler(str(self.password_file))
 
-    def test_encrypt_string_failure_raises(self):
-        handler = VaultHandler("dummy_pw_file")
-        completed = subprocess.CompletedProcess(
-            args=["ansible-vault"], returncode=1, stdout="", stderr="error"
+    def test_encrypt_string_round_trips_through_the_vault(self):
+        snippet = self.handler.encrypt_string("plain_val", "name")
+        body = "\n".join(line.strip() for line in snippet.splitlines()[1:])
+        self.assertEqual(self.handler._vault.decrypt(body).decode(), "plain_val")
+
+    def test_encrypt_string_keeps_the_layout_encrypt_leaves_unindents(self):
+        lines = self.handler.encrypt_string("plain_val", "name").splitlines()
+        self.assertEqual(
+            lines[0], "name: !vault |", "first line must stay the YAML key and tag"
         )
-        with (
-            unittest.mock.patch("subprocess.run", return_value=completed),
-            self.assertRaises(RuntimeError),
-        ):
-            handler.encrypt_string("plain_val", "name")
+        indents = {len(line) - len(line.lstrip()) for line in lines[1:]}
+        self.assertEqual(
+            indents, {10}, "encrypt_leaves strips a uniform body indent of 10"
+        )
+        self.assertEqual(lines[1].strip(), "$ANSIBLE_VAULT;1.1;AES256")
+
+    def test_values_that_look_like_flags_still_encrypt(self):
+        for value in ("-leading-dash", "--stdin-name"):
+            with self.subTest(value=value):
+                snippet = self.handler.encrypt_string(value, "k")
+                body = "\n".join(line.strip() for line in snippet.splitlines()[1:])
+                self.assertEqual(self.handler._vault.decrypt(body).decode(), value)
+
+    def test_missing_password_file_raises(self):
+        with self.assertRaises(AnsibleError):
+            VaultHandler("dummy_pw_file").encrypt_string("plain_val", "name")
+
+    def test_password_file_whitespace_is_stripped_like_ansible_does(self):
+        padded = Path(self._tmp.name) / "padded.pw"
+        padded.write_text("  unit-test-vault-password  \n", encoding="utf-8")
+        snippet = VaultHandler(str(padded)).encrypt_string("plain_val", "name")
+        body = "\n".join(line.strip() for line in snippet.splitlines()[1:])
+        self.assertEqual(self.handler._vault.decrypt(body).decode(), "plain_val")
 
 
 if __name__ == "__main__":  # pragma: no cover

utils/handler/vault.py

@@ -1,7 +1,9 @@
-import subprocess
+from functools import cached_property
 from typing import Any
 
 import yaml
+from ansible.parsing.dataloader import DataLoader
+from ansible.parsing.vault import FileVaultSecret, VaultLib
 from yaml.dumper import SafeDumper
 from yaml.loader import SafeLoader
 
@@ -31,22 +33,22 @@ class VaultHandler:
     def __init__(self, vault_password_file: str):
         self.vault_password_file = vault_password_file
 
+    @cached_property
+    def _vault(self) -> VaultLib:
+        secret = FileVaultSecret(filename=self.vault_password_file, loader=DataLoader())
+        secret.load()
+        return VaultLib([("default", secret)])
+
     def encrypt_string(self, value: str, name: str) -> str:
-        """Encrypt a string using ansible-vault."""
-        cmd = [
-            "ansible-vault",
-            "encrypt_string",
-            "--stdin-name",
-            name,
-            "--vault-password-file",
-            self.vault_password_file,
-        ]
-        proc = subprocess.run(
-            cmd, input=value, capture_output=True, text=True, check=False
-        )
-        if proc.returncode != 0:
-            raise RuntimeError(f"ansible-vault encrypt_string failed:\n{proc.stderr}")
-        return proc.stdout
+        """Return the ``name: !vault |`` snippet as ``ansible-vault`` lays it out.
+
+        Args:
+            value: plaintext to encrypt.
+            name: key the snippet is emitted under.
+        """
+        body = self._vault.encrypt(value).decode()
+        indented = "\n".join(f"          {line}" for line in body.splitlines())
+        return f"{name}: !vault |\n{indented}\n"
 
     def encrypt_leaves(self, branch: dict[str, Any], vault_pw: str):
         """Recursively encrypt all leaves (plain text values) under the credentials section."""