From 80c437b8d0e57fdbf0a8d2289db09c96b2a10ecb Mon Sep 17 00:00:00 2001 From: "Matthew C. Morgan" Date: Mon, 15 Sep 2025 13:03:52 -0400 Subject: [PATCH] added more data extract from ecr --- .../aws_resource_management/managers/ecr.py | 419 +++++++++++++----- 1 file changed, 320 insertions(+), 99 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 de0a98b3..3210a68d 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 @@ -22,6 +22,10 @@ # Threading configuration MAX_WORKERS = config.get("ecr_max_workers", 10) +# New: optional usage discovery settings +DISCOVER_USAGE = config.get("ecr_discover_usage", False) +USAGE_SCAN_LIMIT = int(config.get("ecr_usage_scan_limit", 50)) + class ECRManager(ResourceManager): """Manager for Amazon ECR images.""" @@ -125,6 +129,22 @@ def _get_repository_images(self, ecr_client, repo_name: str, region: str) -> Lis if image.get("imageTags") and len(image["imageTags"]) > 0: image_tag = str(image["imageTags"][0]) + # New: get last scan time if available from summary + last_scan_time = None + try: + summary = image.get("imageScanFindingsSummary", {}) + lst = summary.get("imageScanCompletedAt") + if lst: + last_scan_time = lst.isoformat() if hasattr(lst, "isoformat") else str(lst) + except Exception: + pass + + # New: last used time from lastRecordedPullTime + last_used_time = None + if image.get("lastRecordedPullTime"): + lrt = image.get("lastRecordedPullTime") + last_used_time = lrt.isoformat() if hasattr(lrt, "isoformat") else str(lrt) + image_dict = { "id": str(image.get("imageDigest", "unknown")), "name": f"{repo_name_str}:{image_tag}", @@ -138,6 +158,16 @@ def _get_repository_images(self, ecr_client, repo_name: str, region: str) -> Lis "accountId": account_id_str, "accountName": account_name_str, "accountAlias": account_alias_str, + # New fields (defaults set here; enrichment fills them in): + "lastScanTime": last_scan_time, + "lastUsedTime": last_used_time, + "architecture": None, + "baseImage": None, + "fixAvailable": False, + "exploitAvailable": False, + "usedInECS": False, + "usedInLambda": False, + "usedByResourceArn": None, } images.append(image_dict) @@ -161,9 +191,16 @@ def enrich_resources(self, resources: List[Dict[str, Any]]) -> List[Dict[str, An logger.info("🔍 Getting vulnerability data for images...") vulnerability_data = self._get_vulnerabilities_for_images(resources) - # Apply vulnerability data to images + # Apply vulnerability data to images (also maps extra fields if present) self._apply_vulnerability_data(resources, vulnerability_data) + # Optional: discover usage in ECS/Lambda (by repo and region) + if DISCOVER_USAGE: + try: + self._discover_usage(resources) + except Exception as e: + logger.warning(f"Error discovering image usage: {e}") + logger.info("ECR image enrichment completed") return resources @@ -202,7 +239,18 @@ def _apply_vulnerability_data(self, resources: List[Dict[str, Any]], vulnerabili vuln_data = vulnerability_data.get(image_digest, self._get_empty_vulnerability_counts()) # Add vulnerability counts to the image - image.update(vuln_data) + image.update({k: v for k, v in vuln_data.items() if k in [ + 'CRITICAL','HIGH','MEDIUM','LOW','INFORMATIONAL','UNTRIAGED','totalVulnerabilities' + ]}) + # New: copy extra fields when provided by vuln_data + if 'lastScanTime' in vuln_data and not image.get('lastScanTime'): + image['lastScanTime'] = vuln_data['lastScanTime'] + if 'architecture' in vuln_data and not image.get('architecture'): + image['architecture'] = vuln_data['architecture'] + if 'fixAvailable' in vuln_data: + image['fixAvailable'] = bool(vuln_data['fixAvailable']) + if 'exploitAvailable' in vuln_data: + image['exploitAvailable'] = bool(vuln_data['exploitAvailable']) if vuln_data.get('totalVulnerabilities', 0) > 0: images_with_vulns_applied += 1 @@ -359,6 +407,12 @@ def _get_ecr_scan_findings(self, ecr_client, repo_name: str, image_digest: str) ) scan_status = response.get('imageScanStatus', {}).get('status') + # New: capture lastScanTime if available + scan_completed_at = response.get('imageScanFindings', {}).get('imageScanCompletedAt') \ + or response.get('imageScanStatus', {}).get('scanCompletedAt') + last_scan_time_iso = None + if scan_completed_at: + last_scan_time_iso = scan_completed_at.isoformat() if hasattr(scan_completed_at, "isoformat") else str(scan_completed_at) if scan_status in ['COMPLETE', 'ACTIVE']: findings = response.get('imageScanFindings', {}) @@ -366,17 +420,26 @@ def _get_ecr_scan_findings(self, ecr_client, repo_name: str, image_digest: str) # Try enhanced findings first (Inspector2 format) enhanced_findings = findings.get('enhancedFindings', []) if enhanced_findings: - return self._parse_enhanced_findings(findings) + parsed = self._parse_enhanced_findings(findings) + if last_scan_time_iso: + parsed['lastScanTime'] = last_scan_time_iso + return parsed # Try severity counts summary (Inspector2) severity_counts = findings.get('findingSeverityCounts', {}) if severity_counts: - return self._parse_severity_counts(severity_counts) + parsed = self._parse_severity_counts(severity_counts) + if last_scan_time_iso: + parsed['lastScanTime'] = last_scan_time_iso + return parsed # Fallback to basic ECR findings finding_counts = findings.get('findingCounts', {}) if finding_counts: - return self._parse_severity_counts(finding_counts) + parsed = self._parse_severity_counts(finding_counts) + if last_scan_time_iso: + parsed['lastScanTime'] = last_scan_time_iso + return parsed except ecr_client.exceptions.ScanNotFoundException: # Try to trigger a scan @@ -403,14 +466,11 @@ def _get_inspector2_findings(self, inspector_client, image_digest: str) -> Dict[ 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, + filterCriteria={ + "resourceType": [{"value": "ECR_CONTAINER_IMAGE", "comparison": "EQUALS"}], + "ecrImageHash": [{"value": image_digest, "comparison": "EQUALS"}], + }, maxResults=100 ) @@ -418,112 +478,273 @@ def _get_inspector2_findings(self, inspector_client, image_digest: str) -> Dict[ if not findings: return self._get_empty_vulnerability_counts() - # Count by severity + # Count by severity and aggregate fix/exploit availability severity_counter = Counter() + fix_available = False + exploit_available = False + arch = None + last_scan_time_iso = None + for finding in findings: - severity = finding.get("severity", "UNKNOWN") - if severity != "UNKNOWN": - severity_counter[severity] += 1 - - return self._map_to_standard_format(severity_counter) - + # Severity + sev = finding.get("severity", "UNKNOWN") + if sev != "UNKNOWN": + severity_counter[sev] += 1 + # Fix available + fav = finding.get("fixAvailable") + if isinstance(fav, str): + fix_available = fix_available or fav.upper() == "YES" + elif isinstance(fav, bool): + fix_available = fix_available or fav + # Exploit available + exa = finding.get("exploitAvailable") + if isinstance(exa, str): + exploit_available = exploit_available or exa.upper() == "YES" + elif isinstance(exa, bool): + exploit_available = exploit_available or exa + # Architecture (from resource details if present) + try: + res = finding.get("resources", []) + if res: + details = res[0].get("details", {}) + ecrd = details.get("awsEcrContainerImage", {}) + if ecrd.get("architecture"): + arch = arch or ecrd.get("architecture") + except Exception: + pass + # Last scan/observed time (best-effort) + ts = finding.get("firstObservedAt") or finding.get("lastObservedAt") + if ts and not last_scan_time_iso: + last_scan_time_iso = ts.isoformat() if hasattr(ts, "isoformat") else str(ts) + + mapped = self._map_to_standard_format(severity_counter) + mapped["fixAvailable"] = fix_available + mapped["exploitAvailable"] = exploit_available + if arch: + mapped["architecture"] = arch + if last_scan_time_iso: + mapped["lastScanTime"] = last_scan_time_iso + return mapped + 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.""" + # New: vulnerability helpers and caching + def _get_empty_vulnerability_counts(self) -> Dict[str, int]: + """Return a standard empty vulnerability payload.""" 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()), + 'CRITICAL': 0, + 'HIGH': 0, + 'MEDIUM': 0, + 'LOW': 0, + 'INFORMATIONAL': 0, + 'UNTRIAGED': 0, + 'totalVulnerabilities': 0, + # Optional extras default + 'fixAvailable': False, + 'exploitAvailable': False, } 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()), - } + """Map a Counter to the standard severity dict with totals.""" + result = self._get_empty_vulnerability_counts() + for sev in ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW', 'INFORMATIONAL', 'UNTRIAGED']: + result[sev] = int(severity_counter.get(sev, 0)) + result['totalVulnerabilities'] = sum(result[s] for s in ['CRITICAL','HIGH','MEDIUM','LOW','INFORMATIONAL','UNTRIAGED']) + return result - 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 _parse_severity_counts(self, counts: Dict[str, int]) -> Dict[str, int]: + """Parse severity count dicts from ECR/Inspector summaries.""" + result = self._get_empty_vulnerability_counts() + for sev, val in (counts or {}).items(): + sev_up = str(sev).upper() + if sev_up in result: + result[sev_up] = int(val or 0) + result['totalVulnerabilities'] = sum(result[s] for s in ['CRITICAL','HIGH','MEDIUM','LOW','INFORMATIONAL','UNTRIAGED']) + return result - 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 _parse_enhanced_findings(self, findings: Dict[str, Any]) -> Dict[str, Any]: + """Parse enhanced findings block from ECR describe_image_scan_findings.""" + try: + enhanced = findings.get('enhancedFindings', []) or [] + counter = Counter() + fix_available = False + exploit_available = False + arch = None + for f in enhanced: + sev = f.get('severity', 'UNKNOWN') + sev_up = str(sev).upper() + if sev_up in ['CRITICAL','HIGH','MEDIUM','LOW','INFORMATIONAL','UNTRIAGED']: + counter[sev_up] += 1 + # Aggregate fix/exploit availability when present + fav = f.get('fixAvailable') + if isinstance(fav, str): + fix_available = fix_available or fav.upper() == 'YES' + elif isinstance(fav, bool): + fix_available = fix_available or fav + exa = f.get('exploitAvailable') + if isinstance(exa, str): + exploit_available = exploit_available or exa.upper() == 'YES' + elif isinstance(exa, bool): + exploit_available = exploit_available or exa + # Try architecture from resource details if present + try: + res = f.get("resources", []) + if res: + details = res[0].get("details", {}) + ecrd = details.get("awsEcrContainerImage", {}) + if ecrd.get("architecture"): + arch = arch or ecrd.get("architecture") + except Exception: + pass + + mapped = self._map_to_standard_format(counter) + mapped['fixAvailable'] = fix_available + mapped['exploitAvailable'] = exploit_available + if arch: + mapped['architecture'] = arch + return mapped + except Exception: + return self._get_empty_vulnerability_counts() + + def _cache_key_vuln(self, image_digest: str, region: str) -> str: + """Create a stable cache key for vulnerability lookups.""" + return create_cache_key("vuln", image_digest or "unknown", region or "unknown") - def _cache_vulnerabilities(self, image_digest: str, region: str, vuln_data: Dict[str, int]) -> None: + def _get_cached_vulnerabilities(self, image_digest: str, region: str) -> Optional[Dict[str, Any]]: + """Get cached vulnerability data if available.""" + key = self._cache_key_vuln(image_digest, region) + with self._cache_lock: + return self._vuln_cache.get(key) + + def _cache_vulnerabilities(self, image_digest: str, region: str, data: Dict[str, Any]) -> None: """Cache vulnerability data for an image.""" + key = self._cache_key_vuln(image_digest, region) + with self._cache_lock: + self._vuln_cache[key] = data + + def _log_vulnerability_summary(self, results: Dict[str, Dict[str, Any]]) -> None: + """Log an aggregate summary for diagnostics.""" + if not results: + logger.info("No vulnerability data collected") + return + totals = Counter() + images_with_vulns = 0 + for _, data in results.items(): + for sev in ['CRITICAL','HIGH','MEDIUM','LOW','INFORMATIONAL','UNTRIAGED']: + totals[sev] += int(data.get(sev, 0)) + if data.get('totalVulnerabilities', 0) > 0: + images_with_vulns += 1 + total_images = len(results) + logger.info( + f"Vulnerability summary: images_with_vulns={images_with_vulns}/{total_images}, " + f"CRIT={totals['CRITICAL']}, HIGH={totals['HIGH']}, MED={totals['MEDIUM']}, " + f"LOW={totals['LOW']}, INFO={totals['INFORMATIONAL']}, UNTRIAGED={totals['UNTRIAGED']}" + ) + + def _discover_usage(self, resources: List[Dict[str, Any]]) -> None: + """ + Discover usage of images in ECS task definitions and Lambda functions. + Matches repository + tags to mark usedInECS, usedInLambda and usedByResourceArn. + """ + # Group images by (region, repository) + by_region_repo = {} + for img in resources: + key = (img.get("region"), img.get("repositoryName")) + by_region_repo.setdefault(key, []).append(img) + + for (region, repo), imgs in by_region_repo.items(): + if not region or not repo: + continue + try: + self._discover_usage_for_group(region, repo, imgs) + except Exception as e: + logger.debug(f"Usage discovery failed for {repo} in {region}: {e}") + + def _discover_usage_for_group(self, region: str, repo: str, images: List[Dict[str, Any]]) -> None: + """Discover ECS and Lambda usage for a single (region, repository) group.""" + # Build quick lookup by tag and digest + tags_to_images = {} + digests_to_images = {} + for img in images: + for t in img.get("tags", []) or []: + tags_to_images.setdefault(t, []).append(img) + dg = img.get("id") + if dg: + digests_to_images.setdefault(dg, []).append(img) + + # Discover ECS task definition usage try: - cache_key = create_cache_key("ecr_vulns", str(image_digest), str(region)) - with self._cache_lock: - self._vuln_cache[cache_key] = vuln_data + ecs = create_client("ecs", self.credentials, region) + tds = ecs.list_task_definitions(status="ACTIVE", sort="DESC", maxResults=min(USAGE_SCAN_LIMIT, 100)).get("taskDefinitionArns", []) + # Describe in batches + for i in range(0, len(tds), 10): + batch = tds[i:i+10] + for td_arn in batch: + td = ecs.describe_task_definition(taskDefinition=td_arn).get("taskDefinition", {}) + for c in td.get("containerDefinitions", []): + img_uri = c.get("image", "") + if not img_uri or f"/{repo}" not in img_uri: + continue + # Try tag match + if ":" in img_uri and "@sha256:" not in img_uri: + tag = img_uri.rsplit(":", 1)[-1] + for im in tags_to_images.get(tag, []): + im["usedInECS"] = True + im["usedByResourceArn"] = im.get("usedByResourceArn") or td_arn + # Try digest match + if "@sha256:" in img_uri: + digest = "sha256:" + img_uri.split("@sha256:")[-1] + for im in digests_to_images.get(digest, []): + im["usedInECS"] = True + im["usedByResourceArn"] = im.get("usedByResourceArn") or td_arn except Exception as e: - logger.warning(f"Error caching vulnerability data for {image_digest}: {e}") + logger.debug(f"ECS usage discovery error in {region}/{repo}: {e}") - def _get_cached_vulnerabilities(self, image_digest: str, region: str) -> Optional[Dict[str, int]]: - """Get cached vulnerability data for an image.""" + # Discover Lambda image usage try: - cache_key = create_cache_key("ecr_vulns", str(image_digest), str(region)) - with self._cache_lock: - return self._vuln_cache.get(cache_key) + lmb = create_client("lambda", self.credentials, region) + paginator = lmb.get_paginator("list_functions") + count = 0 + for page in paginator.paginate(): + for fn in page.get("Functions", []): + if count >= USAGE_SCAN_LIMIT: + break + count += 1 + if fn.get("PackageType") != "Image": + continue + img_uri = fn.get("Code", {}).get("ImageUri", "") + if not img_uri or f"/{repo}" not in img_uri: + continue + # Tag-based + if ":" in img_uri and "@sha256:" not in img_uri: + tag = img_uri.rsplit(":", 1)[-1] + for im in tags_to_images.get(tag, []): + im["usedInLambda"] = True + im["usedByResourceArn"] = im.get("usedByResourceArn") or fn.get("FunctionArn") + # if no architecture yet, try lambda's + archs = fn.get("Architectures", []) + if archs and not im.get("architecture"): + im["architecture"] = archs[0] + # Digest-based + if "@sha256:" in img_uri: + digest = "sha256:" + img_uri.split("@sha256:")[-1] + for im in digests_to_images.get(digest, []): + im["usedInLambda"] = True + im["usedByResourceArn"] = im.get("usedByResourceArn") or fn.get("FunctionArn") + archs = fn.get("Architectures", []) + if archs and not im.get("architecture"): + im["architecture"] = archs[0] except Exception as e: - logger.warning(f"Error retrieving cached vulnerability data for {image_digest}: {e}") - return None + logger.debug(f"Lambda usage discovery error in {region}/{repo}: {e}") + + # Add compatibility aliases after discovery + for im in images: + if "usedinECS" not in im: + im["usedinECS"] = bool(im.get("usedInECS", False)) + if "usedinLambda" not in im: + im["usedinLambda"] = bool(im.get("usedInLambda", False)) @staticmethod def normalize_resources(resources: List[Dict[str, Any]]) -> None: