Fix untagged ECS task registration and identify failed AWS calls

tcdeploy.py

@@ -208,7 +208,12 @@ def call(self, api_service, operation, **parameters):
         # JSON on stdin avoids credentials and task configuration in argv or files.
         command = ['aws', api_service, operation, '--region', self.region,
                    '--output', 'json', '--no-cli-pager', '--cli-input-json', 'file:///dev/stdin']
-        output = execute(command, self.env, compact(parameters), timeout=120)
+        try:
+            output = execute(command, self.env, compact(parameters), timeout=120)
+        except DeployError as exc:
+            # Identify the reviewed API operation, never its request parameters
+            # or raw CLI output. Keep the error-code suffix for retry handling.
+            raise DeployError(f'AWS {api_service} {operation}: {exc}') from None
         return json.loads(output) if output.strip() else {}
 
 
@@ -266,7 +271,11 @@ def task_request(m, task, tags, images):
     require(platform.get('cpuArchitecture', 'X86_64') == 'X86_64' and platform.get('operatingSystemFamily', 'LINUX') == 'LINUX', 'This release supports Linux/amd64 tasks only')
     require(set(images) == set(m['images']), 'Supply exactly the manifest image set')
     result = {k: deepcopy(v) for k, v in task.items() if k in TASK_FIELDS}
-    result['tags'] = deepcopy(tags)
+    require(isinstance(tags, list), 'Task tags must be a list')
+    # ECS rejects an explicitly empty tag list on registration. Omission keeps
+    # an untagged definition untagged; preserve every tag when any are present.
+    if tags:
+        result['tags'] = deepcopy(tags)
     counts = {name: 0 for name in images}
     for container in result['containerDefinitions']:
         validate_environment(container.get('environment', []))
@@ -463,8 +472,8 @@ def deploy(aws, m, locals_by_name, release, publisher):
     saved = aws.call('ecs', 'describe-task-definition', taskDefinition=target, include=['TAGS'])
     saved_request = {k: v for k, v in saved['taskDefinition'].items() if k in TASK_FIELDS}
     saved_request['tags'] = sorted(saved.get('tags', []), key=lambda t: t['key'])
-    request['tags'] = sorted(request['tags'], key=lambda t: t['key'])
-    require(saved_request == request,
+    expected_request = {**request, 'tags': sorted(request.get('tags', []), key=lambda t: t['key'])}
+    require(saved_request == expected_request,
             'Registered task configuration differs from requested configuration')
     current = read_service(aws, m)
     require(stable(current, definition), 'Concurrent deployment detected before service update')

tests/test_security.py

@@ -62,6 +62,7 @@ def __init__(self, m):
         self.m = m
         self.calls = []
         self.task = task(m)
+        self.tags = [{'key': 'owner', 'value': 'team'}]
         self.service = service(m)
         self.registered = None
         self.before = None
@@ -74,10 +75,12 @@ def call(self, api, operation, **params):
         if (api, operation) == ('ecs', 'describe-services'):
             return {'services': [deepcopy(self.service)]}
         if (api, operation) == ('ecs', 'describe-task-definition'):
-            return {'taskDefinition': deepcopy(self.registered or self.task), 'tags': [{'key': 'owner', 'value': 'team'}]}
+            return {'taskDefinition': deepcopy(self.registered or self.task), 'tags': deepcopy(self.tags)}
         if (api, operation) == ('ecs', 'register-task-definition'):
+            if 'tags' in params and not params['tags']:
+                raise d.DeployError('Tags can not be empty.')
             self.registered = deepcopy(params)
-            self.registered.pop('tags')
+            self.tags = self.registered.pop('tags', [])
             self.registered.update(status='ACTIVE', revision=2, taskDefinitionArn=self.task['taskDefinitionArn'][:-1] + '2')
             return {'taskDefinition': deepcopy(self.registered)}
         if (api, operation) == ('ecs', 'update-service'):
@@ -344,6 +347,24 @@ def test_command_error_does_not_include_secrets(self):
         self.assertNotIn('dummy-secret', str(caught.exception))
         self.assertIn('AccessDenied', str(caught.exception))
 
+    def test_aws_error_identifies_operation_without_exposing_payloads(self):
+        result = subprocess.CompletedProcess([], 1, 'response-secret',
+            'An error occurred (ClientException) request-secret response-secret')
+        with patch.object(d.subprocess, 'run', return_value=result):
+            with self.assertRaises(d.DeployError) as caught:
+                d.AWS('us-east-1').call('ecs', 'register-task-definition',
+                                     containerDefinitions=[{'environment': [{'value': 'request-secret'}]}])
+        self.assertIn('AWS ecs register-task-definition:', str(caught.exception))
+        self.assertTrue(str(caught.exception).endswith(': ClientException'))
+        self.assertNotIn('request-secret', str(caught.exception))
+        self.assertNotIn('response-secret', str(caught.exception))
+
+    def test_aws_scan_error_preserves_retry_suffix(self):
+        with patch.object(d, 'execute', side_effect=d.DeployError('External command failed: ScanNotFoundException')):
+            with self.assertRaises(d.DeployError) as caught:
+                d.AWS('us-east-1').call('ecr', 'describe-image-scan-findings')
+        self.assertTrue(str(caught.exception).endswith(': ScanNotFoundException'))
+
 
 class DeploymentTests(unittest.TestCase):
     def setUp(self):
@@ -363,6 +384,53 @@ def test_full_release_never_calls_ssm_and_preserves_roles_secrets(self):
         self.assertEqual(self.aws.registered['containerDefinitions'][0]['secrets'], original['containerDefinitions'][0]['secrets'])
         self.assertEqual(self.aws.registered['volumes'], original['volumes'])
 
+    def test_untagged_task_omits_tags_and_passes_readback(self):
+        for include_tags_field in (True, False):
+            with self.subTest(include_tags_field=include_tags_field):
+                self.aws = FakeAWS(self.m)
+                self.aws.tags = []
+                call = self.aws.call
+                def describe_without_tags(api, operation, **params):
+                    response = call(api, operation, **params)
+                    if operation == 'describe-task-definition' and not include_tags_field:
+                        response.pop('tags')
+                    return response
+                with patch.object(self.aws, 'call', side_effect=describe_without_tags):
+                    result = self.deploy()
+                self.assertTrue(result['taskDefinition'].endswith(':2'))
+                registration = next(p for _, op, p in self.aws.calls if op == 'register-task-definition')
+                self.assertNotIn('tags', registration)
+                self.assertEqual(self.aws.tags, [])
+
+    def test_existing_tags_and_empty_tag_values_are_preserved(self):
+        self.aws.tags = [{'key': 'owner', 'value': 'team'}, {'key': 'environment', 'value': ''}]
+        original = deepcopy(self.aws.tags)
+        self.deploy()
+        registration = next(p for _, op, p in self.aws.calls if op == 'register-task-definition')
+        self.assertEqual(registration['tags'], original)
+
+    def test_malformed_tags_stop_before_publication(self):
+        for tags in (None, {}, '', False):
+            with self.subTest(tags=tags):
+                self.aws.tags = tags
+                with self.assertRaises(d.DeployError):
+                    self.deploy()
+                self.assertEqual(self.publisher.calls, [])
+
+    def test_changed_tags_stop_before_service_update(self):
+        for original, changed in [([], [{'key': 'unexpected', 'value': 'tag'}]),
+                                  ([{'key': 'owner', 'value': 'team'}], [])]:
+            with self.subTest(original=original, changed=changed):
+                self.aws = FakeAWS(self.m)
+                self.aws.tags = original
+                def tamper(api, operation, params):
+                    if operation == 'describe-task-definition' and self.aws.registered:
+                        self.aws.tags = changed
+                self.aws.before = tamper
+                with self.assertRaises(d.DeployError):
+                    self.deploy()
+                self.assertNotIn('update-service', [op for _, op, _ in self.aws.calls])
+
     def test_fargate_role_never_overwritten(self):
         self.aws.task['executionRoleArn'] = 'arn:aws:iam::111111111111:role/ecsTaskExecutionRole'
         with self.assertRaises(d.DeployError):