Move inject_credential from awx

.flake8

@@ -110,6 +110,20 @@ per-file-ignores =
   # additionally test docstrings don't need param lists (DAR, DCO020):
   tests/**.py: DAR, DCO020, S101, S105, S108, S404, S603, WPS202, WPS210, WPS430, WPS436, WPS441, WPS442, WPS450
 
+  src/awx_plugins/interfaces/_temporary_private_api.py: ANN001,ANN201,B950,C901,CCR001,D103,E800,LN001,LN002,Q003,WPS110,WPS111,WPS118,WPS125,WPS204,WPS210,WPS211,WPS213,WPS221,WPS226,WPS231,WPS232,WPS319,WPS323,WPS336,WPS337,WPS361,WPS421,WPS429,WPS430,WPS431,WPS436,WPS442,WPS503,WPS507,WPS516
+
+
+
+
+
+
+
+
+
+
+
+
+
 # Count the number of occurrences of each error/warning code and print a report:
 statistics = true
 

dependencies/direct/py.in

@@ -4,7 +4,9 @@ covdefaults
 coverage  # accessed directly from tox
 coverage-enable-subprocess
 hypothesis
+jinja2
 pytest
 pytest-cov
 pytest-mock
 pytest-xdist
+pyyaml

src/awx_plugins/interfaces/_temporary_private_api.py

@@ -4,14 +4,83 @@
 The hope is that it will be refactored into something more standardized.
 """
 
+import os
+import re
+import stat
+import tempfile
 from collections.abc import Callable
 
+from jinja2 import sandbox
+from yaml import safe_dump as yaml_safe_dump
+
+from ._temporary_private_container_api import get_incontainer_path
 from ._temporary_private_credential_api import (  # noqa: WPS436
     Credential as Credential,
     GenericOptionalPrimitiveType,
 )
 
 
+HIDDEN_PASSWORD = '*' * 10
+SENSITIVE_ENV_VAR_NAMES = 'API|TOKEN|KEY|SECRET|PASS'
+
+HIDDEN_PASSWORD_RE = re.compile(SENSITIVE_ENV_VAR_NAMES, re.I)
+HIDDEN_URL_PASSWORD_RE = re.compile('^.*?://[^:]+:(.*?)@.*?$')
+
+ENV_BLOCKLIST = frozenset(
+    (
+        'VIRTUAL_ENV',
+        'PATH',
+        'PYTHONPATH',
+        'JOB_ID',
+        'INVENTORY_ID',
+        'INVENTORY_SOURCE_ID',
+        'INVENTORY_UPDATE_ID',
+        'AD_HOC_COMMAND_ID',
+        'REST_API_URL',
+        'REST_API_TOKEN',
+        'MAX_EVENT_RES',
+        'CALLBACK_QUEUE',
+        'CALLBACK_CONNECTION',
+        'CACHE',
+        'JOB_CALLBACK_DEBUG',
+        'INVENTORY_HOSTVARS',
+        'AWX_HOST',
+        'PROJECT_REVISION',
+        'SUPERVISOR_CONFIG_PATH',
+    )
+)
+
+def build_safe_env(
+    env: dict[str, GenericOptionalPrimitiveType],
+) -> dict[str, GenericOptionalPrimitiveType]:
+    """Obscure potentially sensitive env values.
+
+    Given a set of environment variables, execute a set of heuristics to
+    obscure potentially sensitive environment values.
+
+    :param env: Existing environment variables
+    :returns: Sanitized environment variables.
+    """
+    safe_env = dict(env)
+    for env_k, env_val in safe_env.items():
+        is_special = (
+            env_k == 'AWS_ACCESS_KEY_ID'
+            or (
+                env_k.startswith('ANSIBLE_')
+                and not env_k.startswith('ANSIBLE_NET')
+                and not env_k.startswith('ANSIBLE_GALAXY_SERVER')
+            )
+        )
+        if is_special:
+            continue
+        elif HIDDEN_PASSWORD_RE.search(env_k):
+            safe_env[env_k] = HIDDEN_PASSWORD
+        elif isinstance(env_val, str) and HIDDEN_URL_PASSWORD_RE.match(env_val):
+            safe_env[env_k] = HIDDEN_URL_PASSWORD_RE.sub(
+                HIDDEN_PASSWORD, env_val,
+            )
+    return safe_env
+
 try:
     # pylint: disable-next=unused-import
     from awx.main.models.credential import (  # noqa: WPS433
@@ -21,7 +90,7 @@
     from dataclasses import dataclass  # noqa: WPS433
 
     @dataclass(frozen=True)
-    class ManagedCredentialType:  # type: ignore[no-redef]  # noqa: WPS440
+    class ManagedCredentialType:
         """Managed credential type stub."""
 
         namespace: str
@@ -33,7 +102,7 @@ class ManagedCredentialType:  # type: ignore[no-redef]  # noqa: WPS440
         kind: str
         """Plugin category."""
 
-        inputs: dict[str, list[dict[str, bool | str] | str]]
+        inputs: dict[str, list[dict[str, str | bool]]]
         """UI input fields schema."""
 
         injectors: dict[str, dict[str, str]] | None = None
@@ -50,5 +119,166 @@ class ManagedCredentialType:  # type: ignore[no-redef]  # noqa: WPS440
         ] | None = None
         """Function to call as an alternative to the templated injection."""
 
+        @property
+        def secret_fields(self: 'ManagedCredentialType') -> list[str]:
+            return [
+                str(field['id'])
+                for field in self.inputs.get('fields', [])
+                if field.get('secret', False) is True
+            ]
+
+        def inject_credential(
+            self: 'ManagedCredentialType',
+            credential: Credential,
+            env: dict[str, GenericOptionalPrimitiveType],
+            safe_env: dict[str, GenericOptionalPrimitiveType],
+            args: list[GenericOptionalPrimitiveType],
+            private_data_dir: str,
+        ) -> None:
+            """Inject credential data.
+
+            Inject credential data into the environment variables and
+            arguments passed to `ansible-playbook`
+
+            :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 self.custom_injectors:
+                    injected_env: dict[str, GenericOptionalPrimitiveType] = {}
+                    self.custom_injectors(
+                        credential, injected_env, private_data_dir,
+                    )
+                    env.update(injected_env)
+                    safe_env.update(build_safe_env(injected_env))
+                return
+
+            class TowerNamespace:
+                """Dummy class."""
+
+            tower_namespace = TowerNamespace()
+
+            # maintain a normal namespace for building the ansible-playbook
+            # arguments (env and args)
+            namespace: dict[str, TowerNamespace | GenericOptionalPrimitiveType] = {
+                'tower': tower_namespace,
+            }
+
+            # maintain a sanitized namespace for building the DB-stored arguments
+            # (safe_env)
+            safe_namespace: dict[str, TowerNamespace | GenericOptionalPrimitiveType] = {
+                '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] = value
+                    namespace[field_name] = value
+                    continue
+
+                if field_name in self.secret_fields:
+                    safe_namespace[field_name] = HIDDEN_PASSWORD
+                elif value:
+                    safe_namespace[field_name] = value
+                if value:
+                    namespace[field_name] = value
+
+            for field in self.inputs.get('fields', []):
+                field_id = str(field['id'])
+                # default missing boolean fields to False
+                if field['type'] == 'boolean' and field_id not in credential.get_input_keys():
+                    namespace[field_id] = False
+                    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 str(namespace[field_id]).endswith(
+                            '\n',
+                    ):
+                        namespace[field_id] = str(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()  # type: ignore[misc]
+
+            for file_label, file_tmpl in file_tmpls.items():
+                data: str = sandbox_env.from_string(file_tmpl).render(**namespace)  # type: ignore[misc]
+                env_dir = os.path.join(private_data_dir, 'env')
+                _, path = tempfile.mkstemp(dir=env_dir)
+                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)
+
+            for env_var, tmpl in self.injectors.get('env', {}).items():
+                if env_var in ENV_BLOCKLIST:
+                    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: dict[str, str | list[str]] | list[str] | str) -> dict[str, str] | list[str] | str:
+                    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: str) -> str:
+                    handle, path = tempfile.mkstemp(
+                        dir=os.path.join(private_dir, 'env'),
+                    )
+                    f = os.fdopen(handle, 'w')
+                    f.write(yaml_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])
+
 
 __all__ = ()  # noqa: WPS410