Accept verified empty ECR scan results without severity counts

README.md

@@ -44,6 +44,8 @@ In a separately authorized deploy job, load the workflow-specific image archive,
 
 The deployer verifies account/role, live service configuration, source label, ECR immutability, local/remote configuration digest, a scan completed within 24 hours with no MEDIUM/HIGH/CRITICAL/UNDEFINED findings, and the exact healthy running revision/digests. Only selected images change; roles, secret references, sidecars, resources, volumes and health checks remain intact. Non-root runtime images are required and `RESET_DB` cannot be enabled. The source label is a consistency check, not cryptographic build provenance; protect the pipeline and artifact writers.
 
+ECR may omit its optional severity-count map for a clean image. That response is accepted only after the same completion/freshness checks, with an exact registry/repository/digest match, explicitly empty findings lists, and no remaining page. An absent summary without that evidence, nonempty findings without a summary, or malformed counts fails closed. This requires no additional IAM permissions or changes to registry scanning. See the [ECR response schema](https://docs.aws.amazon.com/AmazonECR/latest/APIReference/API_ImageScanFindings.html).
+
 Use a unique CircleCI build number when retrying publication to immutable tags. A failed rollout remains a failure even if ECS rolls back. The script never blindly reverts over another deployment.
 
 ## Tests

tcdeploy.py

@@ -313,6 +313,31 @@ def verify_running(aws, m, definition, images, desired):
     return len(tasks) >= desired
 
 
+def scan_counts(scan, m, repository, digest):
+    """Accept an omitted summary only when ECR explicitly returns no findings."""
+    findings = scan['imageScanFindings']
+    if 'findingSeverityCounts' not in findings:
+        # ECR can omit the optional counts map for a clean image. An absent
+        # summary alone is not evidence of a clean scan: require empty result
+        # lists, no remaining page, and the exact requested image identity.
+        image = scan.get('imageId')
+        require(scan.get('registryId') == m['accountId']
+                and scan.get('repositoryName') == repository
+                and isinstance(image, dict) and image.get('imageDigest') == digest,
+                'Empty scan response does not identify the requested image')
+        lists = [findings[key] for key in ('findings', 'enhancedFindings') if key in findings]
+        require(lists and all(isinstance(items, list) and not items for items in lists)
+                and scan.get('nextToken') in (None, ''),
+                'Missing image scan severity counts without complete empty findings')
+        return {}
+    counts = findings['findingSeverityCounts']
+    severities = {'INFORMATIONAL', 'LOW', 'MEDIUM', 'HIGH', 'CRITICAL', 'UNDEFINED'}
+    require(isinstance(counts, dict) and set(counts) <= severities
+            and all(type(count) is int and count >= 0 for count in counts.values()),
+            'Invalid image scan severity counts')
+    return counts
+
+
 class Publisher:
     def __init__(self, aws, m, directory):
         self.aws, self.m = aws, m
@@ -383,8 +408,7 @@ def publish(self, name, local, tag, deadline):
                 else:
                     completed = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
                 require(0 <= (datetime.now(timezone.utc) - completed).total_seconds() < 86400, 'Image scan must have completed within the last 24 hours')
-                counts = findings.get('findingSeverityCounts')
-                require(isinstance(counts, dict), 'Missing image scan severity counts')
+                counts = scan_counts(scan, self.m, repository, digest)
                 require(not any(counts.get(level, 0) for level in ('CRITICAL', 'HIGH', 'MEDIUM', 'UNDEFINED')), 'Image scan exceeds the release threshold')
                 return uri + '@' + digest
             require(status in ('IN_PROGRESS', 'PENDING', 'ACTIVE'), 'Image scan failed or is unsupported')

tests/test_security.py

@@ -491,6 +491,86 @@ def test_failed_scan(self):
         with self.assertRaises(d.DeployError):
             self.publish()
 
+    def empty_scan(self, status='ACTIVE'):
+        # Shape returned by ECR for an enhanced scan with zero findings.
+        self.scan = {'registryId': self.m['accountId'], 'repositoryName': 'example-v6',
+            'imageId': {'imageDigest': DIGEST}, 'imageScanStatus': {'status': status},
+            'imageScanFindings': {'imageScanCompletedAt': datetime.now(timezone.utc).isoformat(),
+                                  'findings': []}}
+
+    def test_completed_empty_scan_without_counts(self):
+        for status in ('ACTIVE', 'COMPLETE'):
+            for key in ('findings', 'enhancedFindings'):
+                with self.subTest(status=status, key=key):
+                    self.empty_scan(status)
+                    self.scan['imageScanFindings'].pop('findings')
+                    self.scan['imageScanFindings'][key] = []
+                    result, _ = self.publish()
+                    self.assertTrue(result.endswith('@' + DIGEST))
+
+    def test_missing_counts_and_missing_findings_rejected(self):
+        self.empty_scan()
+        self.scan['imageScanFindings'].pop('findings')
+        with self.assertRaises(d.DeployError):
+            self.publish()
+
+    def test_missing_counts_with_nonempty_or_malformed_findings_rejected(self):
+        for key in ('findings', 'enhancedFindings'):
+            for value in ([{'severity': 'HIGH'}], [{'severity': 'LOW'}], None, {}, ''):
+                with self.subTest(key=key, value=value):
+                    self.empty_scan()
+                    self.scan['imageScanFindings'][key] = value
+                    with self.assertRaises(d.DeployError):
+                        self.publish()
+
+    def test_empty_first_page_without_counts_rejected(self):
+        self.empty_scan()
+        self.scan['nextToken'] = 'another-page'
+        with self.assertRaises(d.DeployError):
+            self.publish()
+
+    def test_empty_scan_must_match_requested_image(self):
+        for key, wrong in [('registryId', '222222222222'), ('repositoryName', 'other'),
+                           ('imageId', {'imageDigest': 'sha256:' + 'f' * 64})]:
+            for value in (wrong, None):
+                with self.subTest(key=key, value=value):
+                    self.empty_scan()
+                    if value is None:
+                        self.scan.pop(key)
+                    else:
+                        self.scan[key] = value
+                    with self.assertRaises(d.DeployError):
+                        self.publish()
+
+    def test_empty_scan_still_requires_fresh_completion(self):
+        for timestamp in ('2000-01-01T00:00:00Z',
+                          (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()):
+            with self.subTest(timestamp=timestamp):
+                self.empty_scan()
+                self.scan['imageScanFindings']['imageScanCompletedAt'] = timestamp
+                with self.assertRaises(d.DeployError):
+                    self.publish()
+
+    def test_empty_scan_rejects_malformed_image_identity(self):
+        for image in (None, [], ''):
+            with self.subTest(image=image):
+                self.empty_scan()
+                self.scan['imageId'] = image
+                with self.assertRaises(d.DeployError):
+                    self.publish()
+
+    def test_empty_findings_do_not_override_failed_status_or_summary(self):
+        self.empty_scan('FAILED')
+        with self.assertRaises(d.DeployError):
+            self.publish()
+        for counts in ({'HIGH': 1}, None, [], {'LOW': -1}, {'LOW': True},
+                       {'LOW': '0'}, {'UNKNOWN': 0}):
+            with self.subTest(counts=counts):
+                self.empty_scan()
+                self.scan['imageScanFindings']['findingSeverityCounts'] = counts
+                with self.assertRaises(d.DeployError):
+                    self.publish()
+
     def test_severity_gate(self):
         for level in ('CRITICAL', 'HIGH', 'MEDIUM', 'UNDEFINED'):
             with self.subTest(level=level):