• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

ansible / awx_plugins.interfaces / 27741604131

08 Jun 2026 06:54PM UTC coverage: 98.749% (+0.09%) from 98.655%
27741604131

push

github

web-flow
Merge pull request #42 from TheRealHaoLiu/fix-custom-credential-file-injection

Fix custom credential file injectors

3 of 3 branches covered (100.0%)

Branch coverage included in aggregate %.

85 of 87 new or added lines in 2 files covered. (97.7%)

1102 of 1116 relevant lines covered (98.75%)

5.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.08
/src/awx_plugins/interfaces/_temporary_private_inject_api.py
1
"""Injectors exercise plugins."""
2

3
import os
6✔
4
import pathlib
6✔
5
import re
6✔
6
import stat
6✔
7
import tempfile
6✔
8
from collections.abc import Mapping
6✔
9
from types import SimpleNamespace
6✔
10

11
from jinja2.sandbox import ImmutableSandboxedEnvironment
6✔
12
from yaml import safe_dump as yaml_safe_dump
6✔
13

14
from awx_plugins.interfaces._temporary_private_container_api import (
6✔
15
    get_incontainer_path,
16
)
17

18
from ._temporary_private_api import EnvVarsType, ManagedCredentialType
6✔
19
from ._temporary_private_credential_api import (
6✔
20
    Credential,
21
    GenericOptionalPrimitiveType,
22
)
23

24

25
# pylint: disable-next=too-few-public-methods
26
class TowerNamespace(SimpleNamespace):
6✔
27
    """Dummy class."""
28

29
    filename: str | SimpleNamespace | None = None
6✔
30

31

32
TowerNamespaceValueType = TowerNamespace | GenericOptionalPrimitiveType
6✔
33
ExtraVarsType = Mapping[str, 'ExtraVarsType'] | list['ExtraVarsType'] | str
6✔
34

35
ArgsType = list[str]
6✔
36

37
HIDDEN_PASSWORD = '*' * 10
6✔
38
SENSITIVE_ENV_VAR_NAMES = 'API|TOKEN|KEY|SECRET|PASS'
6✔
39

40
HIDDEN_PASSWORD_RE = re.compile(SENSITIVE_ENV_VAR_NAMES, re.I)
6✔
41
HIDDEN_URL_PASSWORD_RE = re.compile('^.*?://[^:]+:(.*?)@.*?$')
6✔
42

43
ENV_BLOCKLIST = frozenset(
6✔
44
    (
45
        'VIRTUAL_ENV',
6✔
46
        'PATH',
6✔
47
        'PYTHONPATH',
6✔
48
        'JOB_ID',
6✔
49
        'INVENTORY_ID',
6✔
50
        'INVENTORY_SOURCE_ID',
6✔
51
        'INVENTORY_UPDATE_ID',
6✔
52
        'AD_HOC_COMMAND_ID',
6✔
53
        'REST_API_URL',
6✔
54
        'REST_API_TOKEN',
6✔
55
        'MAX_EVENT_RES',
6✔
56
        'CALLBACK_QUEUE',
6✔
57
        'CALLBACK_CONNECTION',
6✔
58
        'CACHE',
6✔
59
        'JOB_CALLBACK_DEBUG',
6✔
60
        'INVENTORY_HOSTVARS',
6✔
61
        'AWX_HOST',
6✔
62
        'PROJECT_REVISION',
6✔
63
        'SUPERVISOR_CONFIG_PATH',
6✔
64
    ),
65
)
66

67

68
def build_safe_env(
6✔
69
    env: EnvVarsType,
70
) -> EnvVarsType:
71
    """Obscure potentially sensitive environment values.
72

73
    Given a set of environment variables, execute a set of heuristics to
74
    obscure potentially sensitive environment values.
75

76
    :param env: Existing environment variables
77
    :returns: Sanitized environment variables.
78
    """
79
    safe_env = dict(env)
6✔
80
    for env_k, env_v in safe_env.items():
6✔
81
        is_special = env_k == 'AWS_ACCESS_KEY_ID' or (
6✔
82
            env_k.startswith('ANSIBLE_')
6✔
83
            and not env_k.startswith('ANSIBLE_NET')
6✔
84
            and not env_k.startswith('ANSIBLE_GALAXY_SERVER')
6✔
85
        )
86
        if is_special:
6✔
87
            continue
6✔
88
        if HIDDEN_PASSWORD_RE.search(env_k):
6✔
89
            safe_env[env_k] = HIDDEN_PASSWORD
6✔
90
        elif isinstance(env_v, str) and HIDDEN_URL_PASSWORD_RE.match(env_v):
6✔
91
            safe_env[env_k] = HIDDEN_URL_PASSWORD_RE.sub(
6✔
92
                HIDDEN_PASSWORD,
6✔
93
                env_v,
6✔
94
            )
95
    return safe_env
6✔
96

97

98
def secret_fields(cred_type: ManagedCredentialType) -> list[str]:
6✔
99
    """List of fields that are sensitive from the credential type.
100

101
    :param cred_type: Where the secret field descriptions live
102
    :return: list of secret field names
103
    """
104
    return [
105
        str(field['id'])
6✔
106
        for field in cred_type.inputs.get('fields', [])
6✔
107
        if field.get('secret', False) is True
6✔
108
    ]
109

110

111
def _build_extra_vars(
6✔
112
    sandbox: ImmutableSandboxedEnvironment,
113
    namespace: dict[str, TowerNamespaceValueType],
114
    node: Mapping[str, str | list[str]] | list[str] | str,
115
) -> ExtraVarsType:
116
    """Execute template to generate extra vars.
117

118
    :param sandbox: jinja2 sandbox environment
119
    :param namespace: variables available to the jinja2 sandbox
120
    :param node: extra vars for this iteration
121
    :return: filled in extra vars node
122
    """
123
    if isinstance(node, Mapping):
6✔
124
        return {
125
            str(
6✔
126
                _build_extra_vars(sandbox, namespace, entry),
6✔
127
            ): _build_extra_vars(
6✔
128
                sandbox,
6✔
129
                namespace,
6✔
130
                v,
6✔
131
            )
132
            for entry, v in node.items()
6✔
133
        }
134
    if isinstance(node, list):
6✔
135
        return [_build_extra_vars(sandbox, namespace, entry) for entry in node]
6✔
136
    return sandbox.from_string(node).render(**namespace)
×
137

138

139
def _build_extra_vars_file(
6✔
140
    extra_vars: ExtraVarsType,
141
    private_dir: str,
142
) -> str:
143
    """Serialize extra vars out to a file.
144

145
    :param extra_vars: python dict to serialize
146
    :param private_dir: base directory to create file in
147
    :return: path to the file
148
    """
149
    handle, path = tempfile.mkstemp(
6✔
150
        dir=os.path.join(private_dir, 'env'),
6✔
151
    )
152
    f = os.fdopen(handle, 'w')
6✔
153
    f.write(yaml_safe_dump(extra_vars))
6✔
154
    f.close()
6✔
155
    os.chmod(path, stat.S_IRUSR)
6✔
156
    return path
6✔
157

158

159
# pylint: disable-next=too-many-arguments,too-many-positional-arguments,too-many-locals,too-many-branches,too-many-statements
160
def inject_credential(
6✔
161
    cred_type: ManagedCredentialType,
162
    credential: Credential,
163
    env: EnvVarsType,
164
    safe_env: EnvVarsType,
165
    args: ArgsType,
166
    private_data_dir: str,
167
    *,
168
    container_root: os.PathLike[str] | str | None = None,
6✔
169
) -> None:
170
    # pylint: disable=unidiomatic-typecheck
171
    """Inject credential data.
172

173
    Inject credential data into the environment variables and arguments
174
    passed to ansible-playbook
175

176
    :param cred_type: an instance of ManagedCredentialType
177
    :param credential: credential holding the input to be used
178
    :param env: a dictionary of environment variables used in the
179
        ansible-playbook call. This method adds additional environment
180
        variables based on custom environment injectors defined on this
181
        CredentialType.
182
    :param safe_env: a dictionary of environment variables stored in the
183
        database for the job run secret values should be stripped
184
    :param args: a list of arguments passed to ansible-playbook in the
185
        style of subprocess.call(). This method appends additional
186
        arguments based on custom extra_vars injectors defined on this
187
        CredentialType.
188
    :param private_data_dir: a temporary directory to store files
189
        generated by file injectors (like configuration files or key
190
        files)
191
    :param container_root: root directory inside the container to mount
192
        the private data directory to.
193
    :raises ValueError: if file injectors mix bare ``template`` and
194
        dotted ``template.<x>`` labels
195
    :returns: None
196
    """
197
    if not cred_type.injectors:
6✔
198
        if cred_type.managed and cred_type.custom_injectors:
6✔
199
            injected_env: EnvVarsType = {}
6✔
200
            cred_type.custom_injectors(
6✔
201
                credential,
6✔
202
                injected_env,
6✔
203
                private_data_dir,
6✔
204
            )
205
            env.update(injected_env)
6✔
206
            safe_env.update(build_safe_env(injected_env))
6✔
207
        return
208

209
    tower_namespace = TowerNamespace()
6✔
210

211
    # maintain a normal namespace for building the ansible-playbook
212
    # arguments (env and args)
213
    namespace: dict[str, TowerNamespaceValueType] = {
6✔
214
        'tower': tower_namespace,
6✔
215
    }
216

217
    # maintain a sanitized namespace for building the DB-stored arguments
218
    # (safe_env)
219
    safe_namespace: dict[str, TowerNamespaceValueType] = {
6✔
220
        'tower': tower_namespace,
6✔
221
    }
222

223
    # build a normal namespace with secret values decrypted (for
224
    # ansible-playbook) and a safe namespace with secret values hidden (for
225
    # DB storage)
226
    for field_name in credential.get_input_keys():
6✔
227
        value = credential.get_input(field_name)
6✔
228

229
        if type(value) is bool:
6✔
230
            # boolean values can't be secret/encrypted/external
231
            safe_namespace[field_name] = value
6✔
232
            namespace[field_name] = value
6✔
233
            continue
6✔
234

235
        if field_name in secret_fields(cred_type):
6✔
236
            safe_namespace[field_name] = HIDDEN_PASSWORD
6✔
237
        elif value:
6✔
238
            safe_namespace[field_name] = value
6✔
239
        if value:
6✔
240
            namespace[field_name] = value
6✔
241

242
    for field in cred_type.inputs.get('fields', []):
6✔
243
        field_id = str(field['id'])
6✔
244
        field_type_is_bool = field['type'] == 'boolean'
6✔
245
        # default missing boolean fields to False
246
        if field_type_is_bool and field_id not in credential.get_input_keys():
6✔
247
            namespace[field_id] = False
6✔
248
            safe_namespace[field_id] = False
6✔
249
        # make sure private keys end with a \n
250
        if field.get('format') == 'ssh_private_key':
6✔
251
            if field_id in namespace and not str(namespace[field_id]).endswith(
6✔
252
                '\n',
6✔
253
            ):
254
                namespace[field_id] = str(namespace[field_id]) + '\n'
6✔
255

256
    file_tmpls = cred_type.injectors.get('file', {})
6✔
257
    # If any file templates are provided, render the files and update the
258
    # special `tower` template namespace so the filename can be
259
    # referenced in other injectors
260

261
    has_bare = any('.' not in label for label in file_tmpls)
6✔
262
    has_dotted = any('.' in label for label in file_tmpls)
6✔
263
    if has_bare and has_dotted:
6✔
264
        msg = (
6✔
265
            "Credential type file injectors cannot mix bare 'template' "
6✔
266
            "and dotted 'template.<x>' labels — the bare value would "
267
            'overwrite the dotted namespace or vice-versa'
268
        )
269
        raise ValueError(msg)
6✔
270

271
    sandbox_env = ImmutableSandboxedEnvironment()
6✔
272

273
    # Pass 1: create all files to establish paths, so that
274
    # tower.filename.* is fully populated before any template
275
    # might need to cross-reference another file's path.
276
    file_paths: dict[str, tuple[str, str]] = {}
6✔
277
    file: str | None = None
6✔
278

279
    for file_label in file_tmpls:
6✔
280
        env_dir = os.path.join(private_data_dir, 'env')
6✔
281
        path = tempfile.mkstemp(dir=env_dir)[1]
6✔
282
        os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
6✔
283
        container_path = get_incontainer_path(path, private_data_dir)
6✔
284
        file_paths[file_label] = (path, container_path)
6✔
285

286
        if '.' in file_label:
6✔
287
            # SimpleNamespace.__getattribute__ is typed as -> Any in
288
            # typeshed, so mypy sees this isinstance() as containing Any.
289
            if not isinstance(tower_namespace.filename, SimpleNamespace):  # type: ignore[misc]
6✔
290
                tower_namespace.filename = SimpleNamespace()
6✔
NEW
291
            setattr(
×
292
                tower_namespace.filename,
6✔
293
                file_label.split('.')[1],
6✔
294
                container_path,
6✔
295
            )
296
        else:
297
            file = container_path
6✔
298

299
    if file is not None:
6✔
300
        tower_namespace.filename = file
6✔
301

302
    # Pass 2: render templates and write files now that all
303
    # tower.filename.* paths are available for cross-references.
304
    for tmpl_label, file_tmpl in file_tmpls.items():
6✔
NEW
305
        data: str = sandbox_env.from_string(file_tmpl).render(
×
306
            **namespace,
6✔
307
        )
308
        host_path = file_paths[tmpl_label][0]
6✔
309
        pathlib.Path(host_path).write_text(data, encoding='utf-8')
6✔
310

311
    for env_var, tmpl in cred_type.injectors.get('env', {}).items():
6✔
312
        if env_var in ENV_BLOCKLIST:
6✔
313
            continue
6✔
314
        env[env_var] = sandbox_env.from_string(tmpl).render(**namespace)
×
315
        safe_env[env_var] = sandbox_env.from_string(
×
316
            tmpl,
6✔
317
        ).render(**safe_namespace)
6✔
318

319
    if 'INVENTORY_UPDATE_ID' not in env:
6✔
320
        extra_vars = _build_extra_vars(
6✔
321
            sandbox_env,
6✔
322
            namespace,
6✔
323
            cred_type.injectors.get(
6✔
324
                'extra_vars',
6✔
325
                {},
326
            ),
327
        )
328
        if extra_vars:
6✔
329
            path = _build_extra_vars_file(extra_vars, private_data_dir)
6✔
330
            container_path = get_incontainer_path(
6✔
331
                path,
6✔
332
                private_data_dir,
6✔
333
                container_root=container_root,
6✔
334
            )
335
            args.extend(
6✔
336
                # pylint: disable-next=consider-using-f-string
337
                ['-e', '@%s' % container_path],
6✔
338
            )
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc