如何通过Python判断AWS ECR镜像扫描任务是否完成？

你可以用AWS的Python SDK boto3 实现扫描启动+状态轮询的自动化流程，核心思路是启动扫描后，定期调用ECR的查询接口检查扫描状态，直到状态变为COMPLETE或FAILED。

步骤1：准备依赖 #

先确保安装boto3：

pip install boto3

同时确保你的Python环境已配置AWS凭证（可通过环境变量、~/.aws/credentials文件或IAM角色授权）。

步骤2：完整实现代码 #

import boto3
import time

def start_ecr_image_scan(repository_name, image_digest=None, image_tag=None):
    """启动ECR镜像扫描"""
    ecr_client = boto3.client('ecr')
    try:
        # 必须指定镜像的digest或tag其中一个
        if image_digest:
            response = ecr_client.start_image_scan(
                repositoryName=repository_name,
                imageId={'imageDigest': image_digest}
            )
        elif image_tag:
            response = ecr_client.start_image_scan(
                repositoryName=repository_name,
                imageId={'imageTag': image_tag}
            )
        else:
            raise ValueError("必须提供image_digest或image_tag")
        
        print(f"扫描已启动，当前状态: {response['imageScanStatus']['status']}")
        return response
    except ecr_client.exceptions.ImageScanInProgressException:
        print("该镜像的扫描任务已在进行中")
        return None
    except Exception as e:
        print(f"启动扫描失败: {str(e)}")
        return None

def wait_for_scan_completion(repository_name, image_digest=None, image_tag=None, poll_interval=30):
    """轮询等待扫描完成，返回扫描结果"""
    ecr_client = boto3.client('ecr')
    while True:
        try:
            if image_digest:
                response = ecr_client.describe_image_scan_findings(
                    repositoryName=repository_name,
                    imageId={'imageDigest': image_digest}
                )
            elif image_tag:
                response = ecr_client.describe_image_scan_findings(
                    repositoryName=repository_name,
                    imageId={'imageTag': image_tag}
                )
            else:
                raise ValueError("必须提供image_digest或image_tag")
            
            scan_status = response['imageScanStatus']['status']
            if scan_status == 'COMPLETE':
                print("扫描已完成")
                return response['imageScanFindings']
            elif scan_status == 'FAILED':
                print(f"扫描失败，原因: {response['imageScanStatus']['description']}")
                return None
            else:
                print(f"扫描进行中，当前状态: {scan_status}，{poll_interval}秒后再次检查...")
                time.sleep(poll_interval)
        except Exception as e:
            print(f"查询扫描状态失败: {str(e)}")
            time.sleep(poll_interval)

# 示例调用
if __name__ == "__main__":
    REPO_NAME = "your-repository-name"
    IMAGE_TAG = "latest"  # 或者用IMAGE_DIGEST = "sha256:xxxxxxx"
    
    # 启动扫描
    start_ecr_image_scan(REPO_NAME, image_tag=IMAGE_TAG)
    # 等待扫描完成并获取结果
    scan_findings = wait_for_scan_completion(REPO_NAME, image_tag=IMAGE_TAG, poll_interval=60)
    
    if scan_findings:
        print(f"发现漏洞总数: {scan_findings['findingSeverityCounts']}")
        # 可在此添加后续处理逻辑，比如导出漏洞报告

关键说明 #

• 状态判断：通过describe_image_scan_findings接口返回的imageScanStatus.status字段判断状态，常见值：
  • IN_PROGRESS：扫描中
  • COMPLETE：扫描完成
  • FAILED：扫描失败（比如镜像不存在、权限不足）
• 轮询间隔：根据镜像大小调整poll_interval参数，大镜像可设为60秒，小镜像可缩短到10-15秒
• 重复扫描处理：如果调用start_image_scan时扫描已在进行，会抛出ImageScanInProgressException，代码里已做捕获处理

内容的提问来源于stack exchange，提问作者Mark P