Fix AWS CLI JSON parsing during OIDC deployment

.github/workflows/security-tests.yml

@@ -13,6 +13,12 @@ jobs:
         with:
           persist-credentials: false
       - run: python3 -m unittest discover -s tests -v
+      - name: Test with the pinned CircleCI AWS CLI executor
+        run: |
+          docker run --rm --network none --entrypoint python3 \
+            -v "$PWD:/suite:ro" -w /suite \
+            cimg/aws@sha256:4cf3768db00ddf9bc439b8c84e8a60c11a14edf81c8baf76f74132144f8340d5 \
+            -m unittest discover -s tests -v
       - name: Check shell syntax
         run: |
           python3 - <<'PY'

README.md

@@ -54,4 +54,6 @@ Use a unique CircleCI build number when retrying publication to immutable tags.
 python3 -m unittest discover -s tests -v
 ```
 
-Tests mock AWS/Docker and require no credentials, account, network or external Python packages. Passing them does not replace IAM simulation and a real dev rollout. GitHub CI runs them with read-only repository permissions.
+Tests mock cloud API calls and Docker and require no credentials, account, network or external Python packages. An offline AWS CLI parsing test runs when the CLI is installed; CI also runs the suite inside the pinned CircleCI executor with networking disabled. Passing these tests does not replace IAM simulation and a real dev rollout. GitHub CI runs them with read-only repository permissions.
+
+Subprocess input uses an anonymous Linux memory file so AWS CLI can reopen `file:///dev/stdin` during JSON parsing. Tokens and request bodies stay out of command arguments and named filesystem files; the input descriptor closes on success or failure. Regression tests exercise repeated reads using a real subprocess.

tcdeploy.py

@@ -6,6 +6,7 @@
 """
 import argparse
 import base64
+from contextlib import ExitStack
 from copy import deepcopy
 from datetime import datetime, timezone
 import json
@@ -161,8 +162,20 @@ def clean_environment(region):
 
 def execute(command, env, input_text=None, timeout=120):
     try:
-        result = subprocess.run(command, env=env, input=input_text, text=True,
-                                capture_output=True, timeout=timeout, check=False)
+        with ExitStack() as stack:
+            stdin = None
+            if input_text is not None:
+                # AWS CLI can open file:///dev/stdin more than once. A pipe is
+                # exhausted after the first read; an anonymous Linux memory file
+                # is independently reopenable without putting secrets in argv or
+                # a named filesystem file. Close it on success and every failure.
+                stdin = stack.enter_context(os.fdopen(
+                    os.memfd_create('tcdeploy-input', os.MFD_CLOEXEC), 'w+', encoding='utf-8'))
+                stdin.write(input_text)
+                stdin.flush()
+                stdin.seek(0)
+            result = subprocess.run(command, env=env, stdin=stdin, text=True,
+                                    capture_output=True, timeout=timeout, check=False)
     except (OSError, subprocess.TimeoutExpired):
         raise DeployError('External command failed or timed out; no credentials were printed') from None
     # Never echo stdout/stderr or full argv on failures: AWS responses and Docker

tests/test_security.py

@@ -5,7 +5,9 @@
 import json
 import os
 from pathlib import Path
+import shutil
 import subprocess
+import sys
 import tempfile
 import unittest
 from unittest.mock import patch
@@ -271,6 +273,48 @@ def test_aws_json_on_stdin_never_argv(self):
         self.assertNotIn('secret-token', str(execute.call_args.args[0]))
         self.assertIn('secret-token', execute.call_args.args[2])
 
+    def test_subprocess_can_reopen_json_stdin(self):
+        # Exercise real file descriptors: a subprocess PIPE fails on the second
+        # open, just like AWS CLI's repeated loading of file:///dev/stdin.
+        script = '''
+import json, pathlib
+first = json.loads(pathlib.Path('/dev/stdin').read_text())
+second = json.loads(pathlib.Path('/dev/stdin').read_text())
+assert first == second and first['WebIdentityToken'] == 'dummy-secret'
+print('ok')
+'''
+        result = d.execute([sys.executable, '-c', script], d.clean_environment('us-east-1'),
+                           json.dumps({'WebIdentityToken': 'dummy-secret'}))
+        self.assertEqual(result.strip(), 'ok')
+
+    @unittest.skipUnless(shutil.which('aws'), 'AWS CLI integration runs in the pinned executor')
+    def test_aws_cli_parses_stdin_offline(self):
+        command = ['aws', 'ecr', 'describe-images', '--region', 'us-east-1',
+                   '--cli-input-json', 'file:///dev/stdin', '--generate-cli-skeleton', 'output']
+        result = d.execute(command, d.clean_environment('us-east-1'),
+                           json.dumps({'repositoryName': 'example', 'imageIds': [{'imageTag': 'test'}]}))
+        self.assertIsInstance(json.loads(result)['imageDetails'], list)
+
+    def test_input_descriptor_closed_after_success_and_failure(self):
+        for fails in (False, True):
+            descriptors = []
+            def run(command, **kwargs):
+                stream = kwargs['stdin']
+                descriptors.append(stream)
+                self.assertNotIn('dummy-secret', str(command))
+                self.assertIn('/memfd:', os.readlink('/proc/self/fd/' + str(stream.fileno())))
+                self.assertEqual(stream.read(), 'dummy-secret')
+                if fails:
+                    raise subprocess.TimeoutExpired(command, 1)
+                return subprocess.CompletedProcess(command, 0, '', '')
+            with patch.object(d.subprocess, 'run', side_effect=run):
+                if fails:
+                    with self.assertRaises(d.DeployError):
+                        d.execute(['aws'], {}, 'dummy-secret')
+                else:
+                    d.execute(['aws'], {}, 'dummy-secret')
+            self.assertTrue(descriptors[0].closed)
+
     def test_no_secret_api_even_on_explicit_call(self):
         with patch.object(d, 'execute') as execute:
             for api in ('ssm', 'secretsmanager', 'kms', 'iam', 'cloudformation'):