Remove inject_credential from awx

* Consume inject_credential from its new home, awx_plugins.interfaces
This commit is contained in:
Chris Meyers 2024-12-06 14:24:05 -05:00 committed by Chris Meyers
parent ac34e14228
commit bd96000494
3 changed files with 5 additions and 392 deletions

View File

@ -4,15 +4,10 @@ from contextlib import nullcontext
import functools
import inspect
import logging
import os
from importlib.metadata import entry_points
import re
import stat
import tempfile
from types import SimpleNamespace
# Jinja2
from jinja2 import sandbox
# Django
from django.apps.config import AppConfig
@ -26,8 +21,6 @@ from django.utils.functional import cached_property
from django.utils.timezone import now
from django.contrib.auth.models import User
# Shared code for the AWX platform
from awx_plugins.interfaces._temporary_private_container_api import get_incontainer_path
# DRF
from awx.main.utils.pglock import advisory_lock
@ -43,7 +36,6 @@ from awx.main.fields import (
DynamicCredentialInputField,
)
from awx.main.utils import decrypt_field, classproperty, set_environ
from awx.main.utils.safe_yaml import safe_dump
from awx.main.validators import validate_ssh_private_key
from awx.main.models.base import CommonModelNameNotUnique, PasswordFieldsModel, PrimordialModel
from awx.main.models.mixins import ResourceMixin
@ -446,7 +438,7 @@ class CredentialType(CommonModelNameNotUnique):
native = ManagedCredentialType.registry[instance.namespace]
instance.inputs = native.inputs
instance.injectors = native.injectors
instance.custom_injectors = native.custom_injectors
instance.custom_injectors = getattr(native, 'custom_injectors', None)
return instance
def get_absolute_url(self, request=None):
@ -552,133 +544,9 @@ class CredentialType(CommonModelNameNotUnique):
ManagedCredentialType(namespace=ns, name=plugin.name, kind='external', inputs=plugin.inputs)
def inject_credential(self, credential, env, safe_env, args, private_data_dir):
"""
Inject credential data into the environment variables and arguments
passed to `ansible-playbook`
from awx_plugins.interfaces._temporary_private_inject_api import inject_credential
:param credential: a :class:`awx.main.models.Credential` instance
:param env: a dictionary of environment variables used in
the `ansible-playbook` call. This method adds
additional environment variables based on
custom `env` injectors defined on this
CredentialType.
:param safe_env: a dictionary of environment variables stored
in the database for the job run
(`UnifiedJob.job_env`); secret values should
be stripped
:param args: a list of arguments passed to
`ansible-playbook` in the style of
`subprocess.call(args)`. This method appends
additional arguments based on custom
`extra_vars` injectors defined on this
CredentialType.
:param private_data_dir: a temporary directory to store files generated
by `file` injectors (like config files or key
files)
"""
if not self.injectors:
if self.managed and credential.credential_type.custom_injectors:
injected_env = {}
credential.credential_type.custom_injectors(credential, injected_env, private_data_dir)
env.update(injected_env)
safe_env.update(build_safe_env(injected_env))
return
class TowerNamespace:
pass
tower_namespace = TowerNamespace()
# maintain a normal namespace for building the ansible-playbook arguments (env and args)
namespace = {'tower': tower_namespace}
# maintain a sanitized namespace for building the DB-stored arguments (safe_env)
safe_namespace = {'tower': tower_namespace}
# build a normal namespace with secret values decrypted (for
# ansible-playbook) and a safe namespace with secret values hidden (for
# DB storage)
for field_name in credential.get_input_keys():
value = credential.get_input(field_name)
if type(value) is bool:
# boolean values can't be secret/encrypted/external
safe_namespace[field_name] = namespace[field_name] = value
continue
if field_name in self.secret_fields:
safe_namespace[field_name] = '**********'
elif len(value):
safe_namespace[field_name] = value
if len(value):
namespace[field_name] = value
for field in self.inputs.get('fields', []):
# default missing boolean fields to False
if field['type'] == 'boolean' and field['id'] not in credential.inputs.keys():
namespace[field['id']] = safe_namespace[field['id']] = False
# make sure private keys end with a \n
if field.get('format') == 'ssh_private_key':
if field['id'] in namespace and not namespace[field['id']].endswith('\n'):
namespace[field['id']] += '\n'
file_tmpls = self.injectors.get('file', {})
# If any file templates are provided, render the files and update the
# special `tower` template namespace so the filename can be
# referenced in other injectors
sandbox_env = sandbox.ImmutableSandboxedEnvironment()
for file_label, file_tmpl in file_tmpls.items():
data = sandbox_env.from_string(file_tmpl).render(**namespace)
_, path = tempfile.mkstemp(dir=os.path.join(private_data_dir, 'env'))
with open(path, 'w') as f:
f.write(data)
os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
container_path = get_incontainer_path(path, private_data_dir)
# determine if filename indicates single file or many
if file_label.find('.') == -1:
tower_namespace.filename = container_path
else:
if not hasattr(tower_namespace, 'filename'):
tower_namespace.filename = TowerNamespace()
file_label = file_label.split('.')[1]
setattr(tower_namespace.filename, file_label, container_path)
injector_field = self._meta.get_field('injectors')
for env_var, tmpl in self.injectors.get('env', {}).items():
try:
injector_field.validate_env_var_allowed(env_var)
except ValidationError as e:
logger.error('Ignoring prohibited env var {}, reason: {}'.format(env_var, e))
continue
env[env_var] = sandbox_env.from_string(tmpl).render(**namespace)
safe_env[env_var] = sandbox_env.from_string(tmpl).render(**safe_namespace)
if 'INVENTORY_UPDATE_ID' not in env:
# awx-manage inventory_update does not support extra_vars via -e
def build_extra_vars(node):
if isinstance(node, dict):
return {build_extra_vars(k): build_extra_vars(v) for k, v in node.items()}
elif isinstance(node, list):
return [build_extra_vars(x) for x in node]
else:
return sandbox_env.from_string(node).render(**namespace)
def build_extra_vars_file(vars, private_dir):
handle, path = tempfile.mkstemp(dir=os.path.join(private_dir, 'env'))
f = os.fdopen(handle, 'w')
f.write(safe_dump(vars))
f.close()
os.chmod(path, stat.S_IRUSR)
return path
extra_vars = build_extra_vars(self.injectors.get('extra_vars', {}))
if extra_vars:
path = build_extra_vars_file(extra_vars, private_data_dir)
container_path = get_incontainer_path(path, private_data_dir)
args.extend(['-e', '@%s' % container_path])
inject_credential(self, credential, env, safe_env, args, private_data_dir)
class ManagedCredentialType(SimpleNamespace):
@ -688,7 +556,6 @@ class ManagedCredentialType(SimpleNamespace):
for k in ('inputs', 'injectors'):
if k not in kwargs:
kwargs[k] = {}
kwargs.setdefault('custom_injectors', None)
super(ManagedCredentialType, self).__init__(namespace=namespace, **kwargs)
if namespace in ManagedCredentialType.registry:
raise ValueError(
@ -710,7 +577,7 @@ class ManagedCredentialType(SimpleNamespace):
def create(self):
res = CredentialType(**self.get_creation_params())
res.custom_injectors = self.custom_injectors
res.custom_injectors = getattr(self, 'custom_injectors', None)
return res

View File

@ -10,7 +10,6 @@ import fcntl
from unittest import mock
import pytest
import yaml
import jinja2
from awx_plugins.interfaces._temporary_private_container_api import CONTAINER_ROOT
@ -1152,259 +1151,6 @@ class TestJobCredentials(TestJobExecution):
credentials = f.read()
assert credentials == gce_backend_credentials
def test_custom_environment_injectors_with_jinja_syntax_error(self, private_data_dir, mock_me):
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'api_token', 'label': 'API Token', 'type': 'string'}]},
injectors={'env': {'MY_CLOUD_API_TOKEN': '{{api_token.foo()}}'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'api_token': 'ABC123'})
with pytest.raises(jinja2.exceptions.UndefinedError):
credential.credential_type.inject_credential(credential, {}, {}, [], private_data_dir)
def test_custom_environment_injectors(self, private_data_dir, mock_me):
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'api_token', 'label': 'API Token', 'type': 'string'}]},
injectors={'env': {'MY_CLOUD_API_TOKEN': '{{api_token}}'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'api_token': 'ABC123'})
env = {}
credential.credential_type.inject_credential(credential, env, {}, [], private_data_dir)
assert env['MY_CLOUD_API_TOKEN'] == 'ABC123'
def test_custom_environment_injectors_with_boolean_env_var(self, private_data_dir, mock_me):
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'turbo_button', 'label': 'Turbo Button', 'type': 'boolean'}]},
injectors={'env': {'TURBO_BUTTON': '{{turbo_button}}'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'turbo_button': True})
env = {}
credential.credential_type.inject_credential(credential, env, {}, [], private_data_dir)
assert env['TURBO_BUTTON'] == str(True)
def test_custom_environment_injectors_with_reserved_env_var(self, private_data_dir, job, mock_me):
task = jobs.RunJob()
task.instance = job
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'api_token', 'label': 'API Token', 'type': 'string'}]},
injectors={'env': {'JOB_ID': 'reserved'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'api_token': 'ABC123'})
job.credentials.add(credential)
env = task.build_env(job, private_data_dir)
assert env['JOB_ID'] == str(job.pk)
def test_custom_environment_injectors_with_secret_field(self, private_data_dir, mock_me):
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'password', 'label': 'Password', 'type': 'string', 'secret': True}]},
injectors={'env': {'MY_CLOUD_PRIVATE_VAR': '{{password}}'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'password': 'SUPER-SECRET-123'})
credential.inputs['password'] = encrypt_field(credential, 'password')
env = {}
safe_env = {}
credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir)
assert env['MY_CLOUD_PRIVATE_VAR'] == 'SUPER-SECRET-123'
assert 'SUPER-SECRET-123' not in safe_env.values()
assert safe_env['MY_CLOUD_PRIVATE_VAR'] == HIDDEN_PASSWORD
def test_custom_environment_injectors_with_extra_vars(self, private_data_dir, job, mock_me):
task = jobs.RunJob()
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'api_token', 'label': 'API Token', 'type': 'string'}]},
injectors={'extra_vars': {'api_token': '{{api_token}}'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'api_token': 'ABC123'})
job.credentials.add(credential)
args = task.build_args(job, private_data_dir, {})
credential.credential_type.inject_credential(credential, {}, {}, args, private_data_dir)
extra_vars = parse_extra_vars(args, private_data_dir)
assert extra_vars["api_token"] == "ABC123"
assert hasattr(extra_vars["api_token"], '__UNSAFE__')
def test_custom_environment_injectors_with_boolean_extra_vars(self, job, private_data_dir, mock_me):
task = jobs.RunJob()
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'turbo_button', 'label': 'Turbo Button', 'type': 'boolean'}]},
injectors={'extra_vars': {'turbo_button': '{{turbo_button}}'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'turbo_button': True})
job.credentials.add(credential)
args = task.build_args(job, private_data_dir, {})
credential.credential_type.inject_credential(credential, {}, {}, args, private_data_dir)
extra_vars = parse_extra_vars(args, private_data_dir)
assert extra_vars["turbo_button"] == "True"
def test_custom_environment_injectors_with_nested_extra_vars(self, private_data_dir, job, mock_me):
task = jobs.RunJob()
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'host', 'label': 'Host', 'type': 'string'}]},
injectors={'extra_vars': {'auth': {'host': '{{host}}'}}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'host': 'example.com'})
job.credentials.add(credential)
args = task.build_args(job, private_data_dir, {})
credential.credential_type.inject_credential(credential, {}, {}, args, private_data_dir)
extra_vars = parse_extra_vars(args, private_data_dir)
assert extra_vars["auth"]["host"] == "example.com"
def test_custom_environment_injectors_with_templated_extra_vars_key(self, private_data_dir, job, mock_me):
task = jobs.RunJob()
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'environment', 'label': 'Environment', 'type': 'string'}, {'id': 'host', 'label': 'Host', 'type': 'string'}]},
injectors={'extra_vars': {'{{environment}}_auth': {'host': '{{host}}'}}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'environment': 'test', 'host': 'example.com'})
job.credentials.add(credential)
args = task.build_args(job, private_data_dir, {})
credential.credential_type.inject_credential(credential, {}, {}, args, private_data_dir)
extra_vars = parse_extra_vars(args, private_data_dir)
assert extra_vars["test_auth"]["host"] == "example.com"
def test_custom_environment_injectors_with_complicated_boolean_template(self, job, private_data_dir, mock_me):
task = jobs.RunJob()
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'turbo_button', 'label': 'Turbo Button', 'type': 'boolean'}]},
injectors={'extra_vars': {'turbo_button': '{% if turbo_button %}FAST!{% else %}SLOW!{% endif %}'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'turbo_button': True})
job.credentials.add(credential)
args = task.build_args(job, private_data_dir, {})
credential.credential_type.inject_credential(credential, {}, {}, args, private_data_dir)
extra_vars = parse_extra_vars(args, private_data_dir)
assert extra_vars["turbo_button"] == "FAST!"
def test_custom_environment_injectors_with_secret_extra_vars(self, job, private_data_dir, mock_me):
"""
extra_vars that contain secret field values should be censored in the DB
"""
task = jobs.RunJob()
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'password', 'label': 'Password', 'type': 'string', 'secret': True}]},
injectors={'extra_vars': {'password': '{{password}}'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'password': 'SUPER-SECRET-123'})
credential.inputs['password'] = encrypt_field(credential, 'password')
job.credentials.add(credential)
args = task.build_args(job, private_data_dir, {})
credential.credential_type.inject_credential(credential, {}, {}, args, private_data_dir)
extra_vars = parse_extra_vars(args, private_data_dir)
assert extra_vars["password"] == "SUPER-SECRET-123"
def test_custom_environment_injectors_with_file(self, private_data_dir, mock_me):
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'api_token', 'label': 'API Token', 'type': 'string'}]},
injectors={'file': {'template': '[mycloud]\n{{api_token}}'}, 'env': {'MY_CLOUD_INI_FILE': '{{tower.filename}}'}},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'api_token': 'ABC123'})
env = {}
credential.credential_type.inject_credential(credential, env, {}, [], private_data_dir)
path = to_host_path(env['MY_CLOUD_INI_FILE'], private_data_dir)
with open(path, 'r') as f:
assert f.read() == '[mycloud]\nABC123'
def test_custom_environment_injectors_with_unicode_content(self, private_data_dir, mock_me):
value = 'Iñtërnâtiônàlizætiøn'
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': []},
injectors={'file': {'template': value}, 'env': {'MY_CLOUD_INI_FILE': '{{tower.filename}}'}},
)
credential = Credential(
pk=1,
credential_type=some_cloud,
)
env = {}
credential.credential_type.inject_credential(credential, env, {}, [], private_data_dir)
path = to_host_path(env['MY_CLOUD_INI_FILE'], private_data_dir)
with open(path, 'r') as f:
assert f.read() == value
def test_custom_environment_injectors_with_files(self, private_data_dir, mock_me):
some_cloud = CredentialType(
kind='cloud',
name='SomeCloud',
managed=False,
inputs={'fields': [{'id': 'cert', 'label': 'Certificate', 'type': 'string'}, {'id': 'key', 'label': 'Key', 'type': 'string'}]},
injectors={
'file': {'template.cert': '[mycert]\n{{cert}}', 'template.key': '[mykey]\n{{key}}'},
'env': {'MY_CERT_INI_FILE': '{{tower.filename.cert}}', 'MY_KEY_INI_FILE': '{{tower.filename.key}}'},
},
)
credential = Credential(pk=1, credential_type=some_cloud, inputs={'cert': 'CERT123', 'key': 'KEY123'})
env = {}
credential.credential_type.inject_credential(credential, env, {}, [], private_data_dir)
cert_path = to_host_path(env['MY_CERT_INI_FILE'], private_data_dir)
key_path = to_host_path(env['MY_KEY_INI_FILE'], private_data_dir)
with open(cert_path, 'r') as f:
assert f.read() == '[mycert]\nCERT123'
with open(key_path, 'r') as f:
assert f.read() == '[mykey]\nKEY123'
def test_multi_cloud(self, private_data_dir, mock_me):
gce = CredentialType.defaults['gce']()
gce_credential = Credential(pk=1, credential_type=gce, inputs={'username': 'bob', 'project': 'some-project', 'ssh_key_data': self.EXAMPLE_PRIVATE_KEY})

View File

@ -3,4 +3,4 @@ git+https://github.com/ansible/system-certifi.git@devel#egg=certifi
git+https://github.com/ansible/ansible-runner.git@devel#egg=ansible-runner
django-ansible-base @ git+https://github.com/ansible/django-ansible-base@devel#egg=django-ansible-base[rest-filters,jwt_consumer,resource-registry,rbac]
awx-plugins-core @ git+https://git@github.com/ansible/awx-plugins.git@devel#egg=awx-plugins-core
awx_plugins.interfaces @ git+https://github.com/chrismeyersfsu/awx_plugins.interfaces.git@AAP-35749-move-inject-credential-actual
awx_plugins.interfaces @ git+https://github.com/ansible/awx_plugins.interfaces.git