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,51 +4,275 @@
 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 dataclasses import dataclass  # noqa: WPS433
 
+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,
 )
 
 
-try:
-    # pylint: disable-next=unused-import
-    from awx.main.models.credential import (  # noqa: WPS433
-        ManagedCredentialType as ManagedCredentialType,
+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',
     )
-except ImportError:  # FIXME: eventually, this should not exist
-    from dataclasses import dataclass  # noqa: WPS433
+)
+
+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
+
+
+@dataclass(frozen=True)
+class ManagedCredentialType:
+    """Managed credential type stub."""
+
+    namespace: str
+    """Plugin namespace."""
+
+    name: str
+    """Plugin name within the namespace."""
+
+    kind: str
+    """Plugin category."""
+
+    inputs: dict[str, list[dict[str, str | bool]]]
+    """UI input fields schema."""
+
+    injectors: dict[str, dict[str, str]] | None = None
+    """Injector hook parameters."""
+
+    managed: bool = False
+    """Flag for whether this plugin instance is managed."""
+
+    custom_injectors: Callable[
+        [
+            Credential,
+            dict[str, GenericOptionalPrimitiveType], str,
+        ], str | None,
+    ] | 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'
 
-    @dataclass(frozen=True)
-    class ManagedCredentialType:  # type: ignore[no-redef]  # noqa: WPS440
-        """Managed credential type stub."""
+        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
 
-        namespace: str
-        """Plugin namespace."""
+        sandbox_env = sandbox.ImmutableSandboxedEnvironment()  # type: ignore[misc]
 
-        name: str
-        """Plugin name within the namespace."""
+        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)
 
-        kind: str
-        """Plugin category."""
+            # 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)
 
-        inputs: dict[str, list[dict[str, bool | str] | str]]
-        """UI input fields schema."""
+        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)
 
-        injectors: dict[str, dict[str, str]] | None = None
-        """Injector hook parameters."""
+        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)
 
-        managed: bool = False
-        """Flag for whether this plugin instance is managed."""
+            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
 
-        custom_injectors: Callable[
-            [
-                Credential,
-                dict[str, GenericOptionalPrimitiveType], str,
-            ], str | None,
-        ] | None = None
-        """Function to call as an alternative to the templated injection."""
+            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