fix(ci): verify Trigger promotion against the deployed image and cutover

Author: TheodoreSpeaks

.github/actions/docker-build/action.yml

@@ -30,6 +30,11 @@ inputs:
       bypass an input `default:` entirely.
     required: false
 
+outputs:
+  digest:
+    description: The image digest returned by the selected build provider.
+    value: ${{ steps.build-blacksmith.outputs.digest || steps.build-github.outputs.digest }}
+
 # Registry logins must precede this action. provenance/sbom stay off: attestation
 # manifests break `imagetools create` retagging in promote-images.
 runs:
@@ -64,6 +69,7 @@ runs:
         cache-key: ${{ steps.cache-key.outputs.value }}
 
     - name: Build and push (Blacksmith)
+      id: build-blacksmith
       if: inputs.provider == '' || inputs.provider == 'blacksmith'
       uses: useblacksmith/build-push-action@fb9e3e6a9299c78462bfadd0d93352c316adc9b8 # v2
       with:
@@ -169,6 +175,7 @@ runs:
 
     # No cache-to: type=gha — it shares the 10 GB repo quota with the cache mounts.
     - name: Build and push (GitHub)
+      id: build-github
       if: inputs.provider != '' && inputs.provider != 'blacksmith'
       uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7
       with:

.github/scripts/get-ecr-image-digest.sh

@@ -0,0 +1,28 @@
+#!/usr/bin/env bash
+# Read one ECR tag. Only ImageNotFound is optional; AWS and response errors fail.
+set -euo pipefail
+REPOSITORY="${1:?repository required}"
+TAG="${2:?tag required}"
+ALLOW_MISSING="${3:-}"
+if [ -n "$ALLOW_MISSING" ] && [ "$ALLOW_MISSING" != '--allow-missing' ]; then
+  echo 'ERROR: expected --allow-missing or no third argument' >&2
+  exit 1
+fi
+export AWS_PAGER=''
+aws ecr batch-get-image --repository-name "$REPOSITORY" --image-ids imageTag="$TAG" --output json |
+  ALLOW_MISSING="$ALLOW_MISSING" python3 -c '
+import json, os, re, sys
+response = json.load(sys.stdin)
+images, failures = response["images"], response["failures"]
+if failures:
+    if not images and len(failures) == 1 and failures[0]["failureCode"] == "ImageNotFound" and os.environ["ALLOW_MISSING"]:
+        print("")
+        sys.exit(0)
+    raise SystemExit("ERROR: ECR image lookup failed: " + ", ".join(f["failureCode"] for f in failures))
+if len(images) != 1:
+    raise SystemExit("ERROR: expected exactly one ECR image")
+digest = images[0]["imageId"]["imageDigest"]
+if not re.fullmatch(r"sha256:[0-9a-f]{64}", digest):
+    raise SystemExit("ERROR: invalid ECR image digest")
+print(digest)
+'

.github/scripts/test-trigger-deploy.py

@@ -0,0 +1,187 @@
+"""Exercise the deployment gates with scripted AWS responses; no live mutations."""
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+SCRIPTS = Path(__file__).resolve().parent
+DIGEST = 'sha256:' + 'a' * 64
+OTHER_DIGEST = 'sha256:' + 'b' * 64
+
+
+def execution(start=1000, digest=DIGEST, identifier='execution-current'):
+    return {
+        'startTime': start,
+        'pipelineExecutionId': identifier,
+        'sourceRevisions': [{'actionName': 'ECR_Source', 'revisionId': digest}],
+    }
+
+
+class DeploymentGateTests(unittest.TestCase):
+    def run_script(self, script, args, responses):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            fixture = root / 'responses.json'
+            fixture.write_text(json.dumps(responses))
+            (root / 'aws').write_text('''#!/usr/bin/env python3
+import json, os, pathlib, sys
+root = pathlib.Path(os.environ['FIXTURE_DIR'])
+args = sys.argv[1:]
+if args[0] == '--cli-connect-timeout':
+    args = args[4:]
+service, operation = args[:2]
+key = operation
+if operation == 'get-deployment-target':
+    key += ':' + args[args.index('--target-id') + 1]
+with (root / 'calls').open('a') as stream:
+    stream.write(' '.join(args) + '\\n')
+responses = json.loads((root / 'responses.json').read_text())
+if key not in responses:
+    raise SystemExit('Unexpected AWS call: ' + key)
+response = responses[key]
+if isinstance(response, list):
+    response = response.pop(0)
+    responses[key] = response if not responses[key] else responses[key]
+    (root / 'responses.json').write_text(json.dumps(responses))
+if response.get('error'):
+    sys.stderr.write(response['error'])
+    sys.exit(254)
+print(response.get('text', json.dumps(response.get('json'))))
+''')
+            (root / 'date').write_text('''#!/usr/bin/env python3
+import os, pathlib
+path = pathlib.Path(os.environ['FIXTURE_DIR']) / 'clock'
+value = int(path.read_text()) if path.exists() else 1000
+path.write_text(str(value + 1))
+print(value)
+''')
+            (root / 'sleep').write_text('#!/bin/sh\nexit 0\n')
+            for name in ('aws', 'date', 'sleep'):
+                (root / name).chmod(0o755)
+            result = subprocess.run(
+                ['bash', str(SCRIPTS / script), *args],
+                env={**os.environ, 'PATH': f'{root}:{os.environ["PATH"]}',
+                     'FIXTURE_DIR': str(root), 'POLL_INTERVAL': '1', 'OVERALL_TIMEOUT': '12'},
+                capture_output=True, text=True, timeout=10,
+            )
+            calls = (root / 'calls').read_text() if (root / 'calls').exists() else ''
+            return result, calls
+
+    def poll(self, updates=None, since='1000'):
+        responses = {
+            'list-pipeline-executions': {'json': [execution()]},
+            'get-pipeline-execution': {'text': 'InProgress'},
+            'list-action-executions': {'text': 'd-current'},
+            'get-deployment': {'text': 'InProgress'},
+            'list-deployment-targets': {'text': 'target-one\ttarget-two'},
+            'get-deployment-target:target-one': {'text': 'Succeeded'},
+            'get-deployment-target:target-two': {'text': 'Succeeded'},
+        }
+        responses.update(updates or {})
+        return self.run_script('wait-for-ecs-cutover.sh', ['app-pipeline', DIGEST, since], responses)
+
+    def test_waits_for_every_target(self):
+        result, calls = self.poll({'get-deployment-target:target-two': [
+            {'text': 'InProgress'}, {'text': 'Succeeded'}]})
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(calls.count('--target-id target-two'), 2)
+
+    def test_rejects_stale_execution_inside_former_clock_skew_window(self):
+        result, calls = self.poll({'list-pipeline-executions': {'json': [execution(start=999)]}})
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn('timed out', result.stdout)
+        self.assertNotIn('get-pipeline-execution ', calls)
+
+    def test_chooses_newest_matching_execution(self):
+        result, calls = self.poll({'list-pipeline-executions': {'json': [
+            execution(1000, identifier='execution-old'), execution(1001)]}})
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn('--pipeline-execution-id execution-current', calls)
+
+    def test_iso_timestamps(self):
+        result, _ = self.poll({'list-pipeline-executions': {'json': [
+            execution('1970-01-01T00:16:40+00:00')]}})
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_access_denial_fails_immediately(self):
+        result, calls = self.poll({'list-pipeline-executions': {'error': 'AccessDeniedException'}})
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn('AccessDeniedException', result.stderr)
+        self.assertEqual(len(calls.splitlines()), 1)
+
+    def test_credentials_expiring_during_target_poll_fail(self):
+        result, _ = self.poll({'get-deployment-target:target-two': {'error': 'ExpiredToken'}})
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn('ExpiredToken', result.stderr)
+
+    def test_failed_and_superseded_pipeline_never_reach_deployment(self):
+        for status in ('Failed', 'Stopped', 'Superseded'):
+            with self.subTest(status=status):
+                result, calls = self.poll({'get-pipeline-execution': {'text': status}})
+                self.assertNotEqual(result.returncode, 0)
+                self.assertNotIn('get-deployment ', calls)
+
+    def test_waits_for_queued_deploy_action(self):
+        result, calls = self.poll({'list-action-executions': [{'text': 'None'}, {'text': 'd-current'}]})
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(calls.count('list-action-executions '), 2)
+
+    def test_failed_deployment_never_accepts_old_cutover(self):
+        result, calls = self.poll({'get-deployment': {'text': 'Failed'}})
+        self.assertNotEqual(result.returncode, 0)
+        self.assertNotIn('get-deployment-target ', calls)
+
+    def test_empty_targets_cannot_satisfy_gate(self):
+        result, _ = self.poll({'list-deployment-targets': {'text': ''}})
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn('timed out', result.stdout)
+
+    def test_failed_target_fails_immediately(self):
+        result, _ = self.poll({'get-deployment-target:target-two': {'text': 'Failed'}})
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn('cutover status Failed', result.stdout)
+
+    def test_unchanged_image_verifies_existing_cutover(self):
+        result, calls = self.poll({'list-pipeline-executions': {'json': [execution(start=900)]}}, since='0')
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn('get-deployment-target ', calls)
+
+    def test_unchanged_image_rejects_latest_different_deploy(self):
+        result, calls = self.poll({'list-pipeline-executions': {'json': [
+            execution(start=900), execution(start=999, digest=OTHER_DIGEST)]}}, since='0')
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn('cutover is unverified', result.stderr)
+        self.assertNotIn('get-deployment ', calls)
+
+    def test_unchanged_image_rejects_failed_previous_deploy(self):
+        result, _ = self.poll({'get-deployment': {'text': 'Failed'}}, since='0')
+        self.assertNotEqual(result.returncode, 0)
+
+    def test_invalid_metadata_fails_before_aws(self):
+        result, calls = self.poll(since='corrupted')
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(calls, '')
+
+    def test_ecr_digest_and_missing_tag(self):
+        result, _ = self.run_script('get-ecr-image-digest.sh', ['app', 'deploy'], {
+            'batch-get-image': {'json': {'images': [{'imageId': {'imageDigest': DIGEST}}], 'failures': []}}})
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout.strip(), DIGEST)
+        missing = {'batch-get-image': {'json': {'images': [], 'failures': [{'failureCode': 'ImageNotFound'}]}}}
+        result, _ = self.run_script('get-ecr-image-digest.sh', ['app', 'deploy', '--allow-missing'], missing)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout.strip(), '')
+        result, _ = self.run_script('get-ecr-image-digest.sh', ['app', 'deploy'], missing)
+        self.assertNotEqual(result.returncode, 0)
+
+    def test_ecr_response_failures_are_not_missing_images(self):
+        for response in ({'error': 'AccessDeniedException'}, {'json': {'images': [], 'failures': [{'failureCode': 'KmsError'}]}}, {'json': {'images': [], 'failures': []}}):
+            with self.subTest(response=response):
+                result, _ = self.run_script('get-ecr-image-digest.sh', ['app', 'deploy', '--allow-missing'], {'batch-get-image': response})
+                self.assertNotEqual(result.returncode, 0)
+
+
+if __name__ == '__main__':
+    unittest.main()