From 610774bb5f238a2dda72e32fb57fac27b23be5cb Mon Sep 17 00:00:00 2001 From: "Matthew C. Morgan" Date: Fri, 13 Jun 2025 18:22:25 -0400 Subject: [PATCH] updated ecr for vulns --- .../aws_resource_management/managers/ecr.py | 766 ++++++++++++------ 1 file changed, 534 insertions(+), 232 deletions(-) diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ecr.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ecr.py index 53f5fa42..d52b8d5b 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ecr.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ecr.py @@ -2,24 +2,27 @@ ECR resource manager for AWS Resource Management. """ -from datetime import datetime, timedelta, timezone -from typing import Any, Dict, List, Optional +import threading +from collections import Counter, defaultdict +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional, Tuple +from concurrent.futures import ThreadPoolExecutor, as_completed -import boto3 from aws_resource_management.managers.base import ResourceManager -from aws_resource_management.utils import ( - create_client, - get_config, - parse_iso_datetime, - setup_logging, -) - -logger = setup_logging() +from aws_resource_management.utils.api_utils import create_client +from aws_resource_management.utils.logging_utils import logger +from aws_resource_management.utils.config_utils import get_config +from aws_resource_management.utils.aws_core_utils import create_cache_key, safe_get_dict_value +from aws_resource_management.utils.datetime_utils import parse_iso_datetime + config = get_config() # Default age threshold in days for "old" images DEFAULT_AGE_THRESHOLD_DAYS = 365 # 1 year +# Threading configuration +MAX_WORKERS = config.get("ecr_max_workers", 10) + class ECRManager(ResourceManager): """Manager for Amazon ECR images.""" @@ -34,216 +37,523 @@ def __init__( self, account_id: str, region: str, - credentials: Optional[Dict[str, Any]] = None, + credentials: Dict[str, str], account_name: Optional[str] = None, ): - """ - Initialize ECR manager. - - Args: - account_id: AWS account ID - region: AWS region - credentials: Optional AWS credentials dictionary - account_name: Optional AWS account name - """ - super().__init__( - account_id=account_id, - region=region, - credentials=credentials, - account_name=account_name, - resource_type="ecr" - ) - self.age_threshold = config.get( - "ecr_age_threshold_days", DEFAULT_AGE_THRESHOLD_DAYS - ) - - def get_ecr_client(self, region: str = None) -> boto3.client: - """ - Get an ECR client for the specified region using central utility. - - Args: - region: AWS region (defaults to the manager's region) - - Returns: - Boto3 ECR client - """ - region = region or self.region + """Initialize ECR manager.""" + super().__init__(account_id, region, credentials) + self.account_name = account_name or account_id + self.ecr_client = create_client("ecr", credentials, region) + self.inspector_client = create_client("inspector2", credentials, region) + self.age_threshold = config.get("ecr_age_threshold_days", DEFAULT_AGE_THRESHOLD_DAYS) + + # Cache for vulnerability findings to avoid duplicate API calls + self._vuln_cache = {} + self._cache_lock = threading.Lock() + + def get_ecr_client(self, region: str): + """Get ECR client for specified region.""" return create_client("ecr", self.credentials, region) + + def get_inspector_client(self, region: str): + """Get Inspector2 client for specified region.""" + return create_client("inspector2", self.credentials, region) def discover_resources(self, regions: List[str]) -> List[Dict[str, Any]]: - """ - Discover ECR repositories and their images across regions. - - Args: - regions: List of AWS regions to search - - Returns: - List of ECR image dictionaries - """ + """Discover ECR repositories and their images across regions.""" all_images = [] for region in regions: try: ecr_client = self.get_ecr_client(region) - - # Get all repositories - repositories = [] - paginator = ecr_client.get_paginator("describe_repositories") - for page in paginator.paginate(): - repositories.extend(page.get("repositories", [])) - + repositories = self._get_repositories(ecr_client) logger.info(f"Found {len(repositories)} ECR repositories in {region}") # Get images from each repository for repo in repositories: - repo_name = repo.get("repositoryName", "unknown") + repo_name = safe_get_dict_value(repo, "repositoryName", "unknown") images = self._get_repository_images(ecr_client, repo_name, region) all_images.extend(images) except Exception as e: logger.error(f"Error discovering ECR repositories in {region}: {e}") - # Enrich all images with age and old status + # Enrich all images with age and vulnerability data all_images = self.enrich_resources(all_images) - logger.info( - f"Discovered {len(all_images)} ECR images across {len(regions)} regions" - ) + logger.info(f"Discovered {len(all_images)} ECR images across {len(regions)} regions") return all_images - def _get_repository_images( - self, ecr_client, repo_name: str, region: str - ) -> List[Dict[str, Any]]: - """ - Get all images from a specific ECR repository. + def _get_repositories(self, ecr_client) -> List[Dict[str, Any]]: + """Get all repositories from ECR with pagination.""" + repositories = [] + response = ecr_client.describe_repositories() + repositories.extend(response.get("repositories", [])) + + # Handle pagination + while "nextToken" in response: + response = ecr_client.describe_repositories(nextToken=response["nextToken"]) + repositories.extend(response.get("repositories", [])) - Args: - ecr_client: Boto3 ECR client - repo_name: Name of the ECR repository - region: AWS region + return repositories - Returns: - List of ECR image dictionaries - """ + def _get_repository_images(self, ecr_client, repo_name: str, region: str) -> List[Dict[str, Any]]: + """Get all images from a specific ECR repository.""" images = [] try: - paginator = ecr_client.get_paginator("describe_images") - for page in paginator.paginate(repositoryName=repo_name): - for image in page.get("imageDetails", []): - # Get a valid image tag for the name - image_tag = "untagged" - if image.get("imageTags") and len(image["imageTags"]) > 0: - image_tag = image["imageTags"][0] - - image_dict = { - "id": image.get("imageDigest", "unknown"), - "name": f"{repo_name}:{image_tag}", - "region": region, - "repositoryName": repo_name, - "tags": image.get("imageTags", []), - "imagePushedAt": image.get("imagePushedAt"), - "lastRecordedPullTime": image.get("lastRecordedPullTime"), - "imageSizeInBytes": image.get("imageSizeInBytes", 0), - "status": "AVAILABLE", - # Add account info directly - "accountId": self.account_id, - "accountName": self.account_name or self.account_id, - "accountAlias": self.get_account_alias() or self.account_id, - } - - images.append(image_dict) + # Convert to strings to avoid unhashable type issues + account_name_str = str(self.account_name) + account_alias_str = str(self.get_account_alias() or self.account_id) + account_id_str = str(self.account_id) + repo_name_str = str(repo_name) + region_str = str(region) + + # Get image details with pagination + images_response = ecr_client.describe_images(repositoryName=repo_name_str) + image_details = images_response.get("imageDetails", []) + + while "nextToken" in images_response: + images_response = ecr_client.describe_images( + repositoryName=repo_name_str, + nextToken=images_response["nextToken"], + ) + image_details.extend(images_response.get("imageDetails", [])) + + # Process each image + for image in image_details: + image_tag = "untagged" + if image.get("imageTags") and len(image["imageTags"]) > 0: + image_tag = str(image["imageTags"][0]) + + image_dict = { + "id": str(image.get("imageDigest", "unknown")), + "name": f"{repo_name_str}:{image_tag}", + "region": region_str, + "repositoryName": repo_name_str, + "tags": image.get("imageTags", []), + "imagePushedAt": image.get("imagePushedAt"), + "lastRecordedPullTime": image.get("lastRecordedPullTime"), + "imageSizeInBytes": image.get("imageSizeInBytes", 0), + "status": "AVAILABLE", + "accountId": account_id_str, + "accountName": account_name_str, + "accountAlias": account_alias_str, + } + images.append(image_dict) + + logger.debug(f"Repository {repo_name}: {len(images)} images") except Exception as e: - logger.error(f"Error getting images for repository {repo_name}: {e}") + logger.warning(f"Error processing repository {repo_name}: {e}") return images def enrich_resources(self, resources: List[Dict[str, Any]]) -> List[Dict[str, Any]]: - """ - Enrich ECR images with age information and identify old images. + """Enrich ECR images with age and vulnerability data.""" + if not resources: + return resources - Args: - resources: List of ECR image dictionaries + logger.info(f"Enriching {len(resources)} ECR images with age and vulnerability data") - Returns: - Enriched list of ECR image dictionaries - """ - now = datetime.now(timezone.utc) - threshold = timedelta(days=self.age_threshold) + # Add age information + self._add_age_information(resources) + + # Get vulnerability data for all images + logger.info("🔍 Getting vulnerability data for images...") + vulnerability_data = self._get_vulnerabilities_for_images(resources) + + # Apply vulnerability data to images + self._apply_vulnerability_data(resources, vulnerability_data) + + logger.info("ECR image enrichment completed") + return resources - old_images = [] + def _add_age_information(self, resources: List[Dict[str, Any]]) -> None: + """Add age information to images.""" + now = datetime.now(timezone.utc) for image in resources: - # Calculate age if pushed date exists - if pushed_at := image.get("imagePushedAt"): + pushed_at = image.get("imagePushedAt") + if pushed_at: if isinstance(pushed_at, str): - pushed_at = parse_iso_datetime(pushed_at) - - if pushed_at: - age = now - pushed_at - image["ageInDays"] = age.days - - # Identify images older than threshold - if age > threshold: - image["isOld"] = True - old_images.append(image) - else: - image["isOld"] = False - - logger.info( - f"Found {len(old_images)} images older than {self.age_threshold} days" - ) - return resources + pushed_dt = parse_iso_datetime(pushed_at) + else: + pushed_dt = pushed_at + + if pushed_dt: + age_delta = now - pushed_dt + age_days = age_delta.days + image["ageInDays"] = age_days + image["isOld"] = age_days > self.age_threshold + else: + image["ageInDays"] = 0 + image["isOld"] = False + else: + image["ageInDays"] = 0 + image["isOld"] = False + + def _apply_vulnerability_data(self, resources: List[Dict[str, Any]], vulnerability_data: Dict[str, Dict[str, int]]) -> None: + """Apply vulnerability data to resources.""" + total_vulns_found = sum(1 for vuln_data in vulnerability_data.values() + if vuln_data.get('totalVulnerabilities', 0) > 0) + logger.info(f"📊 Retrieved vulnerability data: {total_vulns_found}/{len(vulnerability_data)} images have vulnerabilities") + + images_with_vulns_applied = 0 + for image in resources: + image_digest = image.get("id", "") + vuln_data = vulnerability_data.get(image_digest, self._get_empty_vulnerability_counts()) + + # Add vulnerability counts to the image + image.update(vuln_data) + + if vuln_data.get('totalVulnerabilities', 0) > 0: + images_with_vulns_applied += 1 + logger.debug(f"✅ Applied {vuln_data['totalVulnerabilities']} vulnerabilities to {image.get('name', 'unknown')}") + + logger.info(f"📊 Applied vulnerability data: {images_with_vulns_applied}/{len(resources)} images now have vulnerability data") + + def _get_vulnerabilities_for_images(self, images: List[Dict[str, Any]]) -> Dict[str, Dict[str, int]]: + """Get vulnerability data for multiple images in parallel.""" + results = {} + + if not images: + return results + + # Group images by region and repository for efficient processing + images_by_region_repo = self._group_images_by_region_repo(images) + + if not images_by_region_repo: + logger.debug("No valid images with region, repository, and digest found") + return results + + # Process all repository/region combinations in parallel + with ThreadPoolExecutor(max_workers=min(MAX_WORKERS, len(images_by_region_repo))) as executor: + future_to_repo = { + executor.submit(self._process_repo_images, region_repo_key, repo_images): region_repo_key + for region_repo_key, repo_images in images_by_region_repo.items() + } + + for future in as_completed(future_to_repo): + region_repo_key = future_to_repo[future] + try: + repo_results = future.result() + for image_digest, vuln_data in repo_results: + results[image_digest] = vuln_data + # Cache the results + region = region_repo_key.split('#')[0] + self._cache_vulnerabilities(image_digest, region, vuln_data) + except Exception as e: + logger.error(f"Error processing repository {region_repo_key}: {e}") + + self._log_vulnerability_summary(results) + return results + + def _group_images_by_region_repo(self, images: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]: + """Group images by region and repository for efficient processing.""" + images_by_region_repo = defaultdict(list) + for image in images: + region = safe_get_dict_value(image, "region", "unknown") + repo_name = safe_get_dict_value(image, "repositoryName", "unknown") + image_digest = safe_get_dict_value(image, "id", "unknown") + + if region and image_digest and repo_name: + key = f"{region}#{repo_name}" + images_by_region_repo[key].append({ + "id": str(image_digest), + "region": str(region), + "repositoryName": str(repo_name), + "name": safe_get_dict_value(image, "name", "unknown") + }) + return images_by_region_repo + + def _process_repo_images(self, region_repo_key: str, repo_images: List[Dict[str, Any]]) -> List[Tuple[str, Dict[str, int]]]: + """Process vulnerability lookups for all images in a repository.""" + region, repo_name = region_repo_key.split('#', 1) + repo_results = [] + + try: + ecr_client = self.get_ecr_client(region) + inspector_client = self.get_inspector_client(region) + + # Set current repo name for ECR scan lookups + original_repo = getattr(self, '_current_repo_name', None) + self._current_repo_name = repo_name + + try: + logger.info(f"🔍 Processing {len(repo_images)} images in {repo_name} ({region})") + + # Process images in batches + batch_size = min(10, len(repo_images)) + for i in range(0, len(repo_images), batch_size): + batch = repo_images[i:i + batch_size] + + with ThreadPoolExecutor(max_workers=min(3, len(batch))) as executor: + future_to_image = { + executor.submit( + self._get_single_image_vulnerabilities, + ecr_client, inspector_client, img["id"], img["repositoryName"], region + ): img["id"] + for img in batch + } + + for future in as_completed(future_to_image): + image_digest = future_to_image[future] + try: + vuln_data = future.result() + repo_results.append((image_digest, vuln_data)) + + total_vulns = vuln_data.get('totalVulnerabilities', 0) + if total_vulns > 0: + logger.info(f" ✅ {image_digest[:20]}: {total_vulns} vulnerabilities") + else: + logger.debug(f" ❌ {image_digest[:20]}: 0 vulnerabilities") + + except Exception as e: + logger.warning(f"Error getting vulnerabilities for {image_digest}: {e}") + default_vulns = self._get_empty_vulnerability_counts() + repo_results.append((image_digest, default_vulns)) + finally: + self._current_repo_name = original_repo + + except Exception as e: + logger.error(f"Error processing vulnerabilities for {region_repo_key}: {e}") + for img in repo_images: + default_vulns = self._get_empty_vulnerability_counts() + repo_results.append((img["id"], default_vulns)) + + return repo_results + + def _get_single_image_vulnerabilities( + self, ecr_client, inspector_client, image_digest: str, repo_name: str, region: str + ) -> Dict[str, int]: + """Get vulnerability data for a single image.""" + # Ensure string types + image_digest = str(image_digest) if image_digest else "unknown" + region = str(region) if region else "unknown" + repo_name = str(repo_name) if repo_name else "unknown" + + # Check cache first + cached_data = self._get_cached_vulnerabilities(image_digest, region) + if cached_data: + return cached_data + + # Set current repo for ECR scanning + original_repo = getattr(self, '_current_repo_name', None) + self._current_repo_name = repo_name + + try: + # Try ECR scan findings first + ecr_vuln_data = self._get_ecr_scan_findings(ecr_client, repo_name, image_digest) + if ecr_vuln_data.get('totalVulnerabilities', 0) > 0: + return ecr_vuln_data + + # Fallback to Inspector2 + return self._get_inspector2_findings(inspector_client, image_digest) + finally: + self._current_repo_name = original_repo + + def _get_ecr_scan_findings(self, ecr_client, repo_name: str, image_digest: str) -> Dict[str, int]: + """Get vulnerability findings from ECR scanning.""" + try: + response = ecr_client.describe_image_scan_findings( + repositoryName=repo_name, + imageId={'imageDigest': image_digest} + ) + + scan_status = response.get('imageScanStatus', {}).get('status') + + if scan_status in ['COMPLETE', 'ACTIVE']: + findings = response.get('imageScanFindings', {}) + + # Try enhanced findings first (Inspector2 format) + enhanced_findings = findings.get('enhancedFindings', []) + if enhanced_findings: + return self._parse_enhanced_findings(findings) + + # Try severity counts summary (Inspector2) + severity_counts = findings.get('findingSeverityCounts', {}) + if severity_counts: + return self._parse_severity_counts(severity_counts) + + # Fallback to basic ECR findings + finding_counts = findings.get('findingCounts', {}) + if finding_counts: + return self._parse_severity_counts(finding_counts) + + except ecr_client.exceptions.ScanNotFoundException: + # Try to trigger a scan + try: + ecr_client.start_image_scan( + repositoryName=repo_name, + imageId={'imageDigest': image_digest} + ) + logger.debug(f"Started ECR scan for {repo_name}:{image_digest[:12]}") + except Exception: + pass # Scan already in progress or other issue + + except Exception as ecr_error: + logger.debug(f"ECR scan lookup failed for {image_digest[:20]}: {ecr_error}") + + return self._get_empty_vulnerability_counts() + + def _get_inspector2_findings(self, inspector_client, image_digest: str) -> Dict[str, int]: + """Get vulnerability findings from Inspector2.""" + try: + # Check if Inspector2 is enabled + config = inspector_client.get_configuration() + ecr_config = config.get('ecrConfiguration', {}) + if not ecr_config.get('scanOnPush', False): + return self._get_empty_vulnerability_counts() + + # Query Inspector2 for findings + filter_criteria = { + "resourceType": [{"value": "ECR_CONTAINER_IMAGE", "comparison": "EQUALS"}], + "ecrImageHash": [{"value": image_digest, "comparison": "EQUALS"}], + } + + response = inspector_client.list_findings( + filterCriteria=filter_criteria, + maxResults=100 + ) + + findings = response.get("findings", []) + if not findings: + return self._get_empty_vulnerability_counts() + + # Count by severity + severity_counter = Counter() + for finding in findings: + severity = finding.get("severity", "UNKNOWN") + if severity != "UNKNOWN": + severity_counter[severity] += 1 + + return self._map_to_standard_format(severity_counter) + + except Exception as e: + logger.debug(f"Error getting Inspector2 findings for {image_digest[:20]}: {e}") + return self._get_empty_vulnerability_counts() + + def _parse_enhanced_findings(self, findings: Dict[str, Any]) -> Dict[str, int]: + """Parse Inspector2 enhanced findings.""" + # Try severity counts first + severity_counts = findings.get('findingSeverityCounts', {}) + if severity_counts: + return self._parse_severity_counts(severity_counts) + + # Fallback to parsing individual findings + enhanced_findings = findings.get('enhancedFindings', []) + if not enhanced_findings: + return self._get_empty_vulnerability_counts() + + severity_counter = Counter() + for finding in enhanced_findings: + severity = finding.get('severity', 'UNKNOWN') + if severity != 'UNKNOWN': + severity_counter[severity] += 1 + + return self._map_to_standard_format(severity_counter) + + def _parse_severity_counts(self, severity_counts: Dict[str, int]) -> Dict[str, int]: + """Parse severity counts to standard format.""" + return { + "CRITICAL": severity_counts.get('CRITICAL', 0), + "HIGH": severity_counts.get('HIGH', 0), + "MEDIUM": severity_counts.get('MEDIUM', 0), + "LOW": severity_counts.get('LOW', 0), + "INFORMATIONAL": severity_counts.get('INFORMATIONAL', 0), + "UNTRIAGED": severity_counts.get('UNTRIAGED', 0), + "totalVulnerabilities": sum(severity_counts.values()), + } + + def _map_to_standard_format(self, severity_counter: Counter) -> Dict[str, int]: + """Map severity counter to standard vulnerability format.""" + return { + "CRITICAL": severity_counter.get('CRITICAL', 0), + "HIGH": severity_counter.get('HIGH', 0), + "MEDIUM": severity_counter.get('MEDIUM', 0), + "LOW": severity_counter.get('LOW', 0), + "INFORMATIONAL": severity_counter.get('INFORMATIONAL', 0), + "UNTRIAGED": severity_counter.get('UNTRIAGED', 0), + "totalVulnerabilities": sum(severity_counter.values()), + } + + def _log_vulnerability_summary(self, results: Dict[str, Dict[str, int]]) -> None: + """Log summary of vulnerability processing results.""" + total_images_processed = len(results) + images_with_vulns = sum(1 for vuln_data in results.values() if vuln_data.get('totalVulnerabilities', 0) > 0) + + logger.info(f"📊 Vulnerability Processing Summary:") + logger.info(f" Total images processed: {total_images_processed}") + logger.info(f" Images with vulnerabilities: {images_with_vulns}") + + # Log sample vulnerability data for debugging + sample_count = 0 + for image_digest, vuln_data in results.items(): + if vuln_data.get('totalVulnerabilities', 0) > 0 and sample_count < 3: + logger.info(f" Sample: {image_digest[:20]} = {vuln_data}") + sample_count += 1 + + if images_with_vulns == 0 and total_images_processed > 0: + logger.warning("🔍 No vulnerabilities found - check ECR scanning configuration") + + def _get_empty_vulnerability_counts(self) -> Dict[str, int]: + """Return empty vulnerability counts structure.""" + return { + "CRITICAL": 0, + "HIGH": 0, + "MEDIUM": 0, + "LOW": 0, + "INFORMATIONAL": 0, + "UNTRIAGED": 0, + "totalVulnerabilities": 0, + } + + def _cache_vulnerabilities(self, image_digest: str, region: str, vuln_data: Dict[str, int]) -> None: + """Cache vulnerability data for an image.""" + try: + cache_key = create_cache_key("ecr_vulns", str(image_digest), str(region)) + with self._cache_lock: + self._vuln_cache[cache_key] = vuln_data + except Exception as e: + logger.warning(f"Error caching vulnerability data for {image_digest}: {e}") + + def _get_cached_vulnerabilities(self, image_digest: str, region: str) -> Optional[Dict[str, int]]: + """Get cached vulnerability data for an image.""" + try: + cache_key = create_cache_key("ecr_vulns", str(image_digest), str(region)) + with self._cache_lock: + return self._vuln_cache.get(cache_key) + except Exception as e: + logger.warning(f"Error retrieving cached vulnerability data for {image_digest}: {e}") + return None @staticmethod def normalize_resources(resources: List[Dict[str, Any]]) -> None: - """ - Normalize ECR resources to ensure consistent format. - - Args: - resources: List of ECR image dictionaries - """ + """Normalize ECR resources to ensure consistent format.""" for resource in resources: - # Ensure status field is present if "status" not in resource: resource["status"] = "AVAILABLE" - - # Ensure region is a string if "region" not in resource: resource["region"] = "unknown" + # Ensure vulnerability fields exist with defaults + vuln_fields = ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW', 'INFORMATIONAL', 'UNTRIAGED', 'totalVulnerabilities'] + for field in vuln_fields: + if field not in resource: + resource[field] = 0 + @staticmethod def update_stats(resources: List[Dict[str, Any]], stats: Dict[str, Any]) -> None: - """ - Update statistics with ECR-specific information. - - Args: - resources: List of ECR image resources - stats: Statistics dictionary to update - """ - # Update the found count in stats + """Update statistics with ECR-specific information.""" stats["ecr_found"] = len(resources) - - # Count old images - old_images = [r for r in resources if r.get("isOld", False)] - stats["ecr_old_images"] = len(old_images) - - # Make sure we increment the global resource counters properly + stats["ecr_old_images"] = sum(1 for r in resources if r.get("isOld", False)) + stats["ecr_images_with_vulnerabilities"] = sum(1 for r in resources if r.get('totalVulnerabilities', 0) > 0) + stats["ecr_total_vulnerabilities"] = sum(r.get('totalVulnerabilities', 0) for r in resources) + if "resources_processed" in stats: stats["resources_processed"] += len(resources) - def stop( - self, resources: List[Dict[str, Any]], dry_run: bool = False - ) -> Dict[str, int]: - """ - Delete old ECR images. For ECR, 'stop' means deleting old images. - - Args: - resources: List of ECR image dictionaries - dry_run: If True, don't actually delete images - - Returns: - Dictionary with success, skipped and error counts - """ + def stop(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """Delete old ECR images.""" result = {"success": 0, "skipped": 0, "errors": 0} old_images = [r for r in resources if r.get("isOld", True)] @@ -251,77 +561,69 @@ def stop( logger.info("No old ECR images found to delete") return result - # Create ECR client for this region - try: - ecr_client = self.get_ecr_client() - - for image in old_images: - if image.get("region") != self.region: - result["skipped"] += 1 - continue - - repo_name = image.get("repositoryName") - image_id = image.get("id") + logger.info(f"Deleting {len(old_images)} old ECR images in {self.region}") + + success_count = 0 + error_count = 0 + + for image in old_images: + if image.get("region") != self.region: + result["skipped"] += 1 + continue - if not repo_name or not image_id: - logger.warning(f"Incomplete image data: {image}") - result["skipped"] += 1 - continue + repo_name = image.get("repositoryName") + image_id = image.get("id") - try: - if dry_run: - self.log_action( - resource_id=image_id, - region=self.region, - action="delete", - resource_name=f"{repo_name}:{image_id[:12]}", - details=f"Age: {image.get('ageInDays', 'unknown')} days", - dry_run=True, - ) - result["success"] += 1 - else: - # Delete the image - ecr_client.batch_delete_image( - repositoryName=repo_name, - imageIds=[{"imageDigest": image_id}], - ) - - self.log_action( - resource_id=image_id, - region=self.region, - action="delete", - resource_name=f"{repo_name}:{image_id[:12]}", - details=f"Age: {image.get('ageInDays', 'unknown')} days", - dry_run=False, - ) - result["success"] += 1 + if not repo_name or not image_id: + logger.warning(f"Incomplete image data: {image}") + result["skipped"] += 1 + continue - except Exception as e: - logger.error(f"Error deleting image {image_id}: {e}") - result["errors"] += 1 + try: + vuln_info = f"Vulns: {image.get('totalVulnerabilities', 'unknown')}" + details = f"Age: {image.get('ageInDays', 'unknown')} days, {vuln_info}" + + if dry_run: + self.log_action( + resource_id=image_id, + region=self.region, + action="delete", + resource_name=f"{repo_name}:{image_id[:12]}", + details=details, + dry_run=True, + ) + success_count += 1 + else: + self.ecr_client.batch_delete_image( + repositoryName=repo_name, + imageIds=[{"imageDigest": image_id}], + ) + self.log_action( + resource_id=image_id, + region=self.region, + action="delete", + resource_name=f"{repo_name}:{image_id[:12]}", + details=details, + dry_run=False, + ) + success_count += 1 - except Exception as e: - logger.error(f"Error creating ECR client for {self.region}: {e}") - result["errors"] += len(old_images) + except Exception as e: + logger.error(f"Error deleting image {image_id}: {e}") + error_count += 1 - # Ensure we update the stats for deleted items properly - # This ensures both "ecr_deleted" and "ecr_stopped" keys are set - # for backward compatibility - result["deleted"] = result["success"] + result["deleted"] = success_count + result["success"] = success_count + result["errors"] = error_count + result["skipped"] = len(old_images) - success_count - error_count return result - def start( - self, resources: List[Dict[str, Any]], dry_run: bool = False - ) -> Dict[str, int]: - """ - Start operation is not applicable for ECR images. - - Args: - resources: List of ECR image dictionaries - dry_run: If True, simulate the operation - - Returns: - Dictionary with success, skipped and error counts - """ + def start(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """Start operation is not applicable for ECR images.""" logger.info("Start operation not applicable for ECR images") return {"success": 0, "skipped": len(resources), "errors": 0} + + # Legacy compatibility method + def count_inspector2_findings(self, inspector_client, image_digest: str) -> Dict[str, int]: + """Legacy method for compatibility - delegates to main vulnerability method.""" + return self._get_inspector2_findings(inspector_client, image_digest)