bvandeusen
0bf007173b
Global Ansible credentials, applied to every run (manual + alert-triggered): - core/settings.py: ansible.ssh_private_key / become_password / vault_password (plaintext at rest, masked in UI — encryption tracked in #580) + host_key_checking (default off); surfaced via to_ansible_cfg into app.config[ANSIBLE] - executor.py: pure build_credentials() materializes creds into a 0600 temp dir (--private-key, --vault-password-file, become via -e @vars-file so the password never hits argv) cleaned up in finally; pure ansible_env() sets ANSIBLE_HOST_KEY_CHECKING. build_ansible_command stays param-only - settings/routes.py + ansible.html: admin-only Credentials section, masked-update (blank keeps current, explicit Clear checkbox), reload app config on save - tests: unit (build_credentials per cred + none; ansible_env toggle); integration vault round-trip (ansible-vault encrypt a vars file, run via executor with the vault password, assert it decrypted) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
"""Integration: the executor wires a configured vault password into a real run (task 548).
|
|
|
|
Encrypts a vars file with ansible-vault, then runs a playbook through
|
|
executor.start_run with the matching vault_password in app config, and asserts
|
|
the executor's --vault-password-file decrypted it. (SSH-key/become auth need a
|
|
remote host, so they're covered by the unit-level flag construction.)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import subprocess
|
|
import textwrap
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
_NEEDS_DB = pytest.mark.skipif(
|
|
not os.environ.get("STEWARD_DATABASE_URL"),
|
|
reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
if not os.environ.get("STEWARD_DATABASE_URL"):
|
|
pytest.skip("needs Postgres")
|
|
from steward.app import create_app
|
|
return create_app(testing=False)
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_vault_password_decrypts_a_run(app, tmp_path):
|
|
from sqlalchemy import select
|
|
from steward.ansible import executor
|
|
from steward.models.ansible import AnsibleRun, AnsibleRunStatus
|
|
|
|
vault_pw = "test-vault-pw"
|
|
pwfile = tmp_path / "pw"
|
|
pwfile.write_text(vault_pw + "\n")
|
|
|
|
secret = tmp_path / "secret.yml"
|
|
secret.write_text("vaulted_value: hello-from-vault\n")
|
|
subprocess.run(
|
|
["ansible-vault", "encrypt", str(secret), "--vault-password-file", str(pwfile)],
|
|
check=True, capture_output=True,
|
|
)
|
|
|
|
(tmp_path / "play.yml").write_text(textwrap.dedent("""\
|
|
- hosts: localhost
|
|
connection: local
|
|
gather_facts: false
|
|
vars_files:
|
|
- secret.yml
|
|
tasks:
|
|
- debug:
|
|
msg: "VAULT-{{ vaulted_value }}"
|
|
"""))
|
|
(tmp_path / "inv.ini").write_text("localhost ansible_connection=local\n")
|
|
|
|
# Executor reads credentials from app.config["ANSIBLE"].
|
|
app.config["ANSIBLE"] = {"vault_password": vault_pw, "host_key_checking": False}
|
|
|
|
run_id = str(uuid.uuid4())
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
s.add(AnsibleRun(
|
|
id=run_id, playbook_path="play.yml", inventory_path="inv.ini",
|
|
source_name="t", triggered_by=None, status=AnsibleRunStatus.running,
|
|
))
|
|
await executor.start_run(app, run_id, "play.yml", "inv.ini", str(tmp_path))
|
|
async with app.db_sessionmaker() as s:
|
|
return (await s.execute(
|
|
select(AnsibleRun).where(AnsibleRun.id == run_id))).scalar_one()
|
|
|
|
run = asyncio.run(_go())
|
|
assert run.status == AnsibleRunStatus.success, run.output
|
|
assert "VAULT-hello-from-vault" in (run.output or "") # vault password decrypted it
|