如何配置AWS Lambda Python函数跨多账号管理ECR生命周期策略并免手动输账号

1. 配置Lambda Python函数为多账号无生命周期策略的ECR仓库添加策略 #

步骤1：配置跨账号IAM角色 #

在每个目标AWS账号中创建专属IAM角色，允许Lambda所在账号的执行角色通过sts:AssumeRole获取临时权限，同时赋予该角色ECR操作权限：

• 信任策略（允许Lambda执行角色assume）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::[Lambda所在账号ID]:role/[Lambda执行角色名]"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

• 权限策略（ECR操作权限）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "ecr:DescribeRepositories",
        "ecr:GetLifecyclePolicy",
        "ecr:PutLifecyclePolicy"
      ],
      "Resource": "arn:aws:ecr:*:[目标账号ID]:repository/*"
    }
  ]
}

步骤2：编写Lambda Python核心逻辑 #

实现跨账号凭证获取、ECR仓库遍历、策略检查与添加的完整流程：

import boto3

# 自定义生命周期策略（示例：保留最近30个镜像，清理更早版本）
LIFECYCLE_POLICY = {
    "rules": [
        {
            "rulePriority": 1,
            "description": "Keep last 30 images",
            "selection": {
                "tagStatus": "any",
                "countType": "imageCountMoreThan",
                "countNumber": 30
            },
            "action": {
                "type": "expire"
            }
        }
    ]
}

def assume_role(account_id, role_name):
    sts_client = boto3.client('sts')
    return sts_client.assume_role(
        RoleArn=f"arn:aws:iam::{account_id}:role/{role_name}",
        RoleSessionName="ECR-LP-Automation"
    )['Credentials']

def process_ecr_repos(credentials):
    ecr_client = boto3.client(
        'ecr',
        aws_access_key_id=credentials['AccessKeyId'],
        aws_secret_access_key=credentials['SecretAccessKey'],
        aws_session_token=credentials['SessionToken']
    )
    # 分页获取所有ECR仓库
    repos = []
    paginator = ecr_client.get_paginator('describe_repositories')
    for page in paginator.paginate():
        repos.extend(page['repositories'])
    
    for repo in repos:
        repo_name = repo['repositoryName']
        try:
            # 检查是否已有生命周期策略
            ecr_client.get_lifecycle_policy(repositoryName=repo_name)
            print(f"Repo {repo_name} already has LP, skipping")
        except ecr_client.exceptions.LifecyclePolicyNotFoundException:
            # 为无策略的仓库添加预设LP
            ecr_client.put_lifecycle_policy(
                repositoryName=repo_name,
                lifecyclePolicyText=str(LIFECYCLE_POLICY).replace("'", '"')
            )
            print(f"Added LP to repo {repo_name}")

def lambda_handler(event, context):
    # 先手动传入目标账号列表，后续可优化为自动获取
    target_accounts = [
        {"id": "123456789012", "role_name": "ECR-LP-Automation-Role"},
        {"id": "234567890123", "role_name": "ECR-LP-Automation-Role"}
    ]
    
    for account in target_accounts:
        try:
            creds = assume_role(account['id'], account['role_name'])
            process_ecr_repos(creds)
        except Exception as e:
            print(f"Failed to process account {account['id']}: {str(e)}")

步骤3：配置Lambda执行角色权限 #

给Lambda的执行角色添加sts:AssumeRole权限，允许其调用目标账号的IAM角色：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "sts:AssumeRole",
      "Resource": "arn:aws:iam::*:role/ECR-LP-Automation-Role"
    }
  ]
}

2. 配置函数自动遍历所有AWS账号（无需手动输入账号ID） #

前提条件 #

你的Lambda所在账号为AWS Organizations管理账号，或拥有组织账号列表的查询权限。

步骤1：更新Lambda执行角色权限 #

添加组织账号列表查询权限：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "organizations:ListAccounts",
        "organizations:ListAccountsForParent"
      ],
      "Resource": "*"
    }
  ]
}

步骤2：修改代码自动获取组织内所有账号 #

在Lambda函数中添加组织账号查询逻辑，替换手动账号列表：

def get_all_org_accounts():
    org_client = boto3.client('organizations')
    accounts = []
    paginator = org_client.get_paginator('list_accounts')
    for page in paginator.paginate():
        for account in page['Accounts']:
            # 仅处理活跃状态的账号
            if account['Status'] == 'ACTIVE':
                accounts.append({
                    "id": account['Id'],
                    "role_name": "ECR-LP-Automation-Role"
                })
    return accounts

def lambda_handler(event, context):
    # 自动获取组织内所有活跃账号
    target_accounts = get_all_org_accounts()
    
    for account in target_accounts:
        try:
            creds = assume_role(account['id'], account['role_name'])
            process_ecr_repos(creds)
        except Exception as e:
            print(f"Failed to process account {account['id']}: {str(e)}")

步骤3：优化目标账号角色信任策略 #

为了让组织内所有账号的角色都能被Lambda所在的管理账号调用，更新目标账号角色的信任策略：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::[管理账号ID]:role/[Lambda执行角色名]"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "aws:PrincipalOrgID": "[你的组织ID]"
        }
      }
    }
  ]
}

内容的提问来源于stack exchange，提问作者Eduardo Duarte