【提示/解決方法】如何將最新的Boto3與SSM Automation （或Lambda ）配合使用

📅 發布日: 2024/05/19, 更新日: 2024/05/19 ⏱️ 預計閱讀時間約5分鐘 (2,028字)

SSM自動化aws ： executeScript可能無法執行最新的API。
例如， Control Tower的Basseline API現已於2024年12月15日可用，但在撰寫本文時，它在aws: executeScript上不可用。
SSM Automation aws ： executeScript action和Step Functions相同，最新API可能不可用。為了應對這種情況，我們在SSM Automation aws: executeScript操作中檢查了Boto3的版本，並使用了最新的API。
簡單地用pip升級Boto3不起作用，所以我會描述相應的方法。

什麼是自動化？
要點
1. 對於 SSM 自動化
2. 對於拉姆達
驗證結果（針對SSM自動化）
用於驗證的 SSM Automation Terraform 程式碼
用於驗證Lambda的Boto3版本升級的Terraform程式碼

什麼是自動化？

自動化是 AWS Systems Manager 的一項功能，可讓您自動執行操作任務。
您可以使用 Runbook 自動執行各種操作，該 Runbook 定義了自動化執行的任務。
AWS 提供了許多 Runbook，因此您只需傳遞所需引數即可執行任務。
參考： Systems Manager Automation Runbook 參考 - AWS Systems Manager Automation Runbook 參考
您還可以定義自己的 Runbook，透過組合操作實施自定義處理，或呼叫多個 AWS 提供的 Runbook 來建立操作工作流程。
參考： Systems Manager 自動化操作參考 - AWS Systems Manager

要點

對於 SSM 自動化

作為先決條件，您需要在 SSM 自動化的附件部分中指定包含 boto3 和 botocore 包檔案的 zip 檔案。
您可以按照以下兩個步驟使用最新的Boto3：

將 AWS_DATA_PATH 環境變數設定為儲存最新模型檔案的目錄（當前目錄下的 /botocore/data ）。

SSM Automation執行的目錄具有隨機名稱，因此您可以透過在當前目錄下指定 /botocore/data 來使用最新的模型檔案。

import os

os.environ["AWS_DATA_PATH"] = f"{os.getcwd()}/botocore/data"

重新載入 boto3 和 botocore。

import os
import sys
import boto3
import botocore

sys.path.insert(0, os.getcwd())
reload(botocore)
reload(boto3)

對於拉姆達

對於 Lambda，您必須在 Lambda 層中附加包含 boto3 和 botocore 包檔案的 zip 檔案。
您可以透過將 AWS_DATA_PATH 環境變數設定為 /opt/python/botocore/data 來使用最新的Boto3。以下 Python 程式碼在 Lambda 程式碼中設定環境變數，但您也可以在 Lambda 環境設定中設定它。

import os

os.environ["AWS_DATA_PATH"] = "/opt/python/botocore/data"

在 Lambda 中，Lambda 層安裝在 /opt/python 處，並且該層中的模組優先。與 SSM Automation 不同，您不需要重新載入 boto3 或 botocore。

驗證結果（針對SSM自動化）

如下面的類方法所示，當在 boto3 中建立客戶端時，botocore 從 ${botocore 根目錄}/data 、 ~/.aws/models 和 AWS_DATA_PATH 載入模型檔案。
參考：【如何將新的AWS服務與舊的boto3一起使用|DevelopersIO](https://dev.classmethod.jp/articles/use-new-aws-service-with-old-boto3)
botocore的安裝目錄具有以下模型檔案檔案結構：

# 【提示/解決方法】如何將最新的Boto3與SSM Automation （或Lambda ）配合使用
botocore/data
└── controltower
     └── 2018-05-10
         ├── endpoint-rule-set-1.json.gz
         ├── paginators-1.json # List of available APIs
         └── service-2.json.gz

如果您設定 AWS_DATA_PATH 環境變數，則以最高優先順序載入最新的模型檔案。
參考：配置 - Boto3 1.34.108 文件
但是，即使您設定此環境變數，boto3 和 botocore 本身也不是 SSM Automation 中的最新版本。
在 Lambda 中，Lambda 層中的模組優先，但在 SSM Automation 中，zip 檔案（附件）中的模組不具有優先順序。
在 SSM Automation 中，附件檔案被提取到“/tmp/${random string}”。以下輸出顯示了“sys.path”的 JSON 格式，其中附件中的檔案具有最低優先順序。

{
    "sys path": [
        "/var/task",
        "/opt/python/lib/python3.8/site-packages",
        "/opt/python",
        "/var/runtime",
        "/var/lang/lib/python38.zip",
        "/var/lang/lib/python3.8",
        "/var/lang/lib/python3.8/lib-dynload",
        "/var/lang/lib/python3.8/site-packages",
        "/opt/python/lib/python3.8/site-packages",
        "/opt/python",
        "/tmp/4aa790cf-cb0b-48b6-80f0-3ed7f82f228c-2024-03-17-07-15-31"
    ]
}

因此，在SSM Automation中，您需要將當前目錄新增到 sys.path 並重新載入boto3和botocore。

import os
import sys
import boto3
import botocore

sys.path.insert(0, os.getcwd())
reload(botocore)
reload(boto3)

用於驗證的 SSM Automation Terraform 程式碼

為了升級 SSM Automation 中的 boto3 版本，我使用了以下 Terraform 檔案結構：

├── src
│   ├── customer_script.py
│   └── modules
│       └── requirements.txt
└── ssm.tf

src/customer_script.py

SSM Automation中，除了升級boto3之外，還考慮了日誌輸出。
您可以使用 ListHandler 類將日誌輸出儲存在列表中並在最後返回它，以便日誌顯示在SSM自動化執行螢幕中。

import logging
import json
import os
import sys
from importlib import reload

import boto3
import botocore

AUTOMATION_NAME = "sample-automation"
logger = logging.getLogger(__name__)

class ListHandler(logging.Handler):
    def __init__(self):
        super().__init__()
        self.log_messages = []
        log_format = json.dumps(
            {
                "logLevel": "%(levelname)s",
                "time": "%(asctime)s",
                "line": "%(filename)s: %(lineno)d",
                "message": "%(message)s",
            },
            indent=2,
            ensure_ascii=False,
        )
        self.formatter = logging.Formatter(log_format)

    def emit(self, record):
        self.log_messages.append(json.loads(self.format(record)))

def handler(event, context):
    list_handler = ListHandler()
    logger.addHandler(list_handler)
    logger.setLevel(logging.DEBUG)

    os.environ["AWS_DATA_PATH"] = f"{os.getcwd()}/botocore/data"
    session = boto3.Session()
    logger.debug({"botocore paths": session._loader.search_paths})

    sys.path.insert(0, os.getcwd())
    reload(botocore)
    reload(boto3)

    logger.debug({"sys path": sys.path})
    logger.debug({"current directory": os.getcwd()})
    logger.debug({"child directory": os.listdir()})
    logger.debug({"botocore path": os.path.dirname(botocore.__file__)})
    logger.debug({"boto3 path": boto3.__file__})
    logger.debug({"boto3 version": boto3.__version__})
    logger.debug({"botocore path": botocore.__file__})
    logger.debug({"botocore version": botocore.__version__})

    try:
        client = session.client("controltower")
        logger.info(client.list_baselines())

    except Exception as e:
        logger.error(f"Failed to list_baselines: {e}")
        raise Exception(list_handler.log_messages)

    # The max length of SSM Automation output is 100 KB (102400 characters)
    return list_handler.log_messages

if __name__ == "__main__":
    handler(event={}, context={})

src/modules/requirements.txt

此處指定用於 SSM 自動化驗證的 boto3 和 botocore 版本。
未設定具體版本；使用最新版本（撰寫本文時為 1.34.64）。

boto3
botocore

ssm.tf

這是用於驗證 SSM 自動化中升級 boto3 的 Terraform 程式碼。
像“#tfsec:ignore:aws-s3-encryption-customer-key”這樣的註釋將禁用 TFSec 檢查。
使用“null_resource”、“archive_file”和“aws_s3_object”，建立包含 boto3 和 botocore 的 zip 檔案並將其附加到 SSM Automation。
對於 CloudFormation，不支援“aws cloudformation package”命令，因此您必須手動建立 zip 檔案並獲取 SHA256 雜湊。Terraform 對於此目的更加靈活。

data "aws_caller_identity" "current" {}

#tfsec:ignore:aws-s3-encryption-customer-key
#tfsec:ignore:aws-s3-enable-bucket-encryption
#tfsec:ignore:aws-s3-enable-bucket-logging
resource "aws_s3_bucket" "bucket" {
  bucket = "sample-automation-${data.aws_caller_identity.current.account_id}"
}

resource "aws_s3_bucket_versioning" "bucket_versioning" {
  bucket = aws_s3_bucket.bucket.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_public_access_block" "bucket_public_access_block" {
  bucket = aws_s3_bucket.bucket.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

resource "null_resource" "install_requirements" {
  triggers = {
    always_run = "${timestamp()}"
  }

  provisioner "local-exec" {
    command = "pip install -r ${path.root}/src/modules/requirements.txt -t ${path.root}/src/modules"
  }
}

data "archive_file" "file" {
  depends_on = [ null_resource.install_requirements ]
  type        = "zip"
  source_dir  = "${path.root}/src/modules"
  output_path = "${path.root}/src/modules.zip"
}

resource "aws_s3_object" "object" {
  bucket = aws_s3_bucket.bucket.id
  key    = "modules.zip"
  source = data.archive_file.file.output_path
}

resource "aws_ssm_document" "runbook" {
  depends_on = [ aws_s3_object.object ]

  name            = "sample-automation"
  document_type   = "Automation"
  document_format = "JSON"
  attachments_source {
    key = "S3FileUrl"
    name = "modules.zip"
    values = [
      "https://${aws_s3_bucket.bucket.bucket}.s3.amazonaws.com/modules.zip"
    ]
  }

  content = jsonencode({
    assumeRole      = aws_iam_role.role.arn
    schemaVersion = "0.3"
    description   = "TestBoto3VersionUp"
    mainSteps = [
      {
        name   = "TestBoto3VersionUp"
        action = "aws:executeScript"
        inputs = {
          Runtime = "python3.8"
          Handler = "handler"
          Script  = "${file("${path.root}/src/customer_script.py")}"
          Attachment = "modules.zip"
        }
      }
    ]
    files = {
      "modules.zip" = {
        checksums = {
          sha256 = filesha256(data.archive_file.file.output_path)
        }
      }
    }
  })
}

resource "aws_iam_role" "role" {
  name = "sample-automation-role"
  assume_role_policy = jsonencode({
    Version = "2012-10-17",
    Statement = [
      {
        Action = "sts:AssumeRole",
        Effect = "Allow",
        Principal = {
          Service = "ssm.amazonaws.com"
        }
      }
    ]
  })
  managed_policy_arns = [
    "arn:aws:iam::aws:policy/service-role/AmazonSSMAutomationRole",
    "arn:aws:iam::aws:policy/AdministratorAccess"
  ]
}

用於驗證Lambda的Boto3版本升級的Terraform程式碼

我使用Terraform程式碼和以下檔案配置來執行Lambda的Boto3版本升級驗證。

├── src
│   ├── function
│   │   └── main.py
│   └── layer
│   │   ├── python # Store the package file of Boto3 / Botocore by processing Terraform under this directory
│   │   └── requirements.txt
└── lambda.tf

src/函式/main.py

Lambda的Python程式碼很簡單。

import logging
import os
import sys

import boto3
import botocore

logger = logging.getLogger(__name__)

def handler(event, context):
    logger.setLevel(logging.DEBUG)

    os.environ["AWS_DATA_PATH"] = "/opt/python/botocore/data"
    session = boto3.Session()
    logger.debug({"botocore paths": session._loader.search_paths})

    logger.debug({"sys path": sys.path})
    logger.debug({"botocore path": os.path.dirname(botocore.__file__)})
    logger.debug({"boto3 path": boto3.__file__})
    logger.debug({"boto3 version": boto3.__version__})
    logger.debug({"botocore path": botocore.__file__})
    logger.debug({"botocore version": botocore.__version__})

    client = session.client("controltower")
    logger.info(client.list_baselines())

if __name__ == "__main__":
    handler(event={}, context={})

src/layer/requirements.txt

描述了要包含在 Lambda Layer 中的包檔案。
它不是特定版本，但使用最新版本（撰寫本文時為 1.34.64）。

boto3
botocore

拉姆達.tf

這是一段Terraform程式碼，用於驗證Lambda的Boto3版本升級。
與 SSM Automation 一樣，我們使用 null_resource 和 data.archive_file 建立 ZIP 檔案，包括 Lambda 層中的 Boto3 / Botocore 包檔案。

data "archive_file" "function" {
  depends_on  = [null_resource.install_requirements]
  type        = "zip"
  source_dir  = "${path.root}/src/function"
  output_path = "${path.root}/src/function.zip"
}

resource "aws_lambda_function" "function" {
  function_name    = "sample-function"
  filename         = data.archive_file.function.output_path
  source_code_hash = data.archive_file.function.output_base64sha256
  handler          = "main.handler"
  runtime          = "python3.11"
  role             = aws_iam_role.role.arn
  # os.environ["AWS_DATA_PATH"] is OK even if you set the following environment variables
  # environment {
  #   variables = {
  #     AWS_DATA_PATH = "/opt/python/botocore/data"
  #   }
  # }
  layers        = [aws_lambda_layer_version.layer_version.arn]
}

resource "null_resource" "install_requirements" {
  triggers = {
    always_run = "${timestamp()}"
  }

  provisioner "local-exec" {
    command = "pip install -r ${path.root}/src/layer/requirements.txt -t ${path.root}/src/layer/python"
  }
}

data "archive_file" "layer" {
  depends_on  = [null_resource.install_requirements]

  type        = "zip"
  source_dir  = "${path.root}/src/layer"
  output_path = "${path.root}/src/layer.zip"
}

resource "aws_lambda_layer_version" "layer_version" {
  layer_name = "sample-layer"
  filename  = data.archive_file.layer.output_path
  compatible_runtimes = ["python3.11"]
  source_code_hash = data.archive_file.layer.output_base64sha256
}

resource "aws_iam_role" "role" {
  name = "sample-lambda-role"
  assume_role_policy = jsonencode({
    Version = "2012-10-17",
    Statement = [
      {
        Action = "sts:AssumeRole",
        Effect = "Allow",
        Principal = {
          Service = "lambda.amazonaws.com"
        }
      }
    ]
  })
  managed_policy_arns = [
    "arn:aws:iam::aws:policy/AdministratorAccess"
  ]
}