diff --git a/.github/workflows/security-tests.yml b/.github/workflows/security-tests.yml index 0548955..69abc36 100644 --- a/.github/workflows/security-tests.yml +++ b/.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' diff --git a/README.md b/README.md index e351a2e..9275598 100644 --- a/README.md +++ b/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. diff --git a/tcdeploy.py b/tcdeploy.py index f285508..15c2325 100644 --- a/tcdeploy.py +++ b/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 diff --git a/tests/test_security.py b/tests/test_security.py index 0a8d0db..d3bcf20 100644 --- a/tests/test_security.py +++ b/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'):