HMAC 認証を実施する HCP Terraform Run Tasks を AWS Lambda で実装する

## はじめに

HCP Terraform の Run Tasks 機能を使用すると、Terraform 実行時に外部サービスによる検証を組み込むことができます。

本記事では、AWS Lambda を使用して Run Tasks を実装する方法を、具体的なコードと共に解説します。
この実装により、Terraform の実行フローに柔軟なカスタム検証を組み込むことができます。
また、Lambda は HMAC 認証を実施する構成にしているため、意図しないアクセスに対しては処理しないように構成しています。

### 想定読者

- Terraform を使用したインフラ構築の経験がある方
- AWS Lambda での開発経験がある方
- HCP Terraform でのガバナンス強化を検討している方

### 環境要件

- HCP Terraform アカウント
  - Organization と Project が構成済み
- AWS アカウント
- Python 3.13

## HMAC 認証について

Run Tasks と Lambda 関数間の通信を [HMAC 認証](https://www.nic.ad.jp/ja/tech/ipa/RFC2104JA.html) で保護します。

通信元と通信先で共通の秘密鍵を持っておき、通信元からのリクエストボディ全体のハッシュ値が通信元から送られた HMAC 署名と一致するか検証します。

HMAC 署名はセキュリティ保護の仕組みとして Run Task が標準で提供しています。

https://developer.hashicorp.com/terraform/cloud-docs/api-docs/notification-configurations#notification-authenticity

上記のドキュメントでは Ruby のサンプルが記載されていますが、Python の場合も比較的簡単に HMAC 認証を実現できます。

## アーキテクチャ概要

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3769407/6ccc38e3-2ef3-69eb-7265-2b2918c30bf0.png)

本実装は以下の 2 つの主要コンポーネントで構成されています：

### Run Task 実行基盤（run_task_resources）

AWS 上に構築される検証基盤として、以下のリソースを含みます：

- **Lambda 関数**: 実際の検証ロジックを実行

  - HMAC 認証による安全な通信を実装
  - カスタマイズ可能な検証ルールを定義
  - Parameter Store からキーを読み込み

- **Parameter Store**: シークレット管理
  - HMAC キーの安全な管理

### HCP Terraform の Workspace の設定

Terraform 実行環境の設定として、以下を含みます：

- **Run Tasks 設定**: Plan 後の自動検証を実行
  - 警告チェックとして設定（enforcement_level = "advisory"）
  - Plan 後の実行
  - Workspace への関連付け

## サンプルコード

下記に用意しましたのでご活用ください。

https://github.com/hirokimatsueda/hcp-terraform-run-task-validator

サンプル実装として、Plan 結果のタイムスタンプの「分」が偶数なら問題なし、奇数ならエラーとするようにしました。

**Plan 結果のタイムスタンプの「分」が偶数の場合の実行結果**

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3769407/03035837-b748-3796-04b0-9cf4eceb03fe.png)

-> Status が `Passed` で問題なしとなります

**Plan 結果のタイムスタンプの「分」が奇数の場合の実行結果**

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3769407/2ea3eee5-3803-4121-ed7e-a77fcc0c4535.png)

-> Status が `Failed` でエラーとなります

## Lambda のコードの実装ポイント

### シークレット値の読み込み

Lambda 関数では [AWS Parameters and Secrets Lambda Extension](https://docs.aws.amazon.com/ja_jp/secretsmanager/latest/userguide/retrieving-secrets_lambda.html) を使用して Parameter Store のシークレットを読み込むようにしています。
これにより、キャッシュを経由して HMAC 認証に使用するキーを読み込めます。

```python
def get_parameter(name: str) -> str:
    """Parameter Store Extension API からパラメータを取得

    AWS Lambda Extension for Parameter Store を使用して、
    Parameter Store からパラメータを取得します。
    withDecryption=true を指定することで、SecureString 型の値を復号化して取得します。

    Args:
        name: パラメータ名

    Returns:
        パラメータの値（SecureString の場合は復号化された値）
    """
    url = f"http://localhost:2773/systemsmanager/parameters/get?name={name}&withDecryption=true"
    headers = {
        "X-Aws-Parameters-Secrets-Token": os.environ.get("AWS_SESSION_TOKEN", "")
    }
    req = urllib.request.Request(url, headers=headers)
    with urllib.request.urlopen(req) as res:
        response = json.loads(res.read())
        return response["Parameter"]["Value"]
```

ちなみに[以前紹介した Requests ライブラリ](https://qiita.com/pokapu/items/2ba221568a59ef6f72de)は使用していません。
外部ライブラリを使用するには Lambda のレイヤー追加が必要となり、terraform コードが複雑化するためサンプル実装には適さないと判断しました。

https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/python-layers.html

### HMAC 検証

リクエストヘッダーから HMAC 認証として渡された SHA512 ハッシュ値を取得し、Python の標準ライブラリを用いて内容を検証します。

```python
def verify_hmac(payload: str, signature: str, secret: str) -> bool:
    """HMAC 署名を検証

    HCP Terraform からのリクエストの HMAC 署名を検証します。
    SHA-512 ハッシュアルゴリズムを使用し、タイミング攻撃を防ぐために
    hmac.compare_digest を使用して比較を行います。

    Args:
        payload: 署名対象のペイロード
        signature: リクエストヘッダーの HMAC 署名
        secret: HMAC 署名の秘密鍵

    Returns:
        署名が有効な場合は True、それ以外は False
    """
    computed = hmac.new(secret.encode(), payload.encode(), hashlib.sha512).hexdigest()
    return hmac.compare_digest(computed, signature)


def handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    ...
    try:
        # パラメータストアからシークレット取得
        hmac_key = get_parameter(os.environ["HMAC_SECRET_KEY_PARAM"])

        # リクエストの解析
        body = json.loads(event.get("body", "{}"))
        headers = event.get("headers", {})
        signature = headers.get("x-tfc-task-signature", "")

        # HMAC署名の検証
        if not verify_hmac(event["body"], signature, hmac_key):
        ...
```

意図した HMAC 認証キーで SHA512 ハッシュ値を生成していることを確認できたら認証成功と判断し、後続処理を実施しています。

### 初期設定の判断

Run Task の初期設定時にも検証用のリクエストが送られてきて、200 応答しないと Run Task が構成できません。
初期設定の際はリクエストボディにはダミーの URL が渡ってくるだけで検証ロジックを実行する必要は無いので、HMAC 認証後はすぐに応答するようにします。

```python
        if body.get("task_result_enforcement_level") == "test":
            return {
                "statusCode": 200,
                "body": json.dumps(
                    {
                        "data": {
                            "type": "task-results",
                            "attributes": {
                                "status": "passed",
                                "message": "Configuration successful",
                            },
                        }
                    }
                ),
            }

```

### Plan 結果の読み込み

Plan 結果はリクエストボディの `plan_json_api_url` に記載した URL から取得できます。
取得する際はリクエストボディの `access_token` に記載された Bearer トークンを使用する必要があります。

```python
def get_plan_json(request_body: Any):
    """Plan 結果の Json の取得

    Args:
        request_body: リクエストボディ

    Returns:
        Plan 結果の Json
    """
    url = request_body.get("plan_json_api_url")
    access_token = request_body.get("access_token")

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-type": "application/vnd.api+json",
    }
    req = urllib.request.Request(url, headers=headers)
    with urllib.request.urlopen(req) as res:
        response = json.loads(res.read())
        return response
```

### 結果の応答

検証結果はリクエストボディの `task_result_callback_url` に記載した URL に PATCH リクエストで応答する必要があります。
こちらもリクエストボディの `access_token` に記載された Bearer トークンを使用する必要があります。
送信するデータのフォーマットは指定があり、`data.attributes.status` に `passed` を記載すれば問題なし、`failed` ならエラーとなります。

https://developer.hashicorp.com/terraform/cloud-docs/api-docs/run-tasks/run-tasks-integration#request-body-1

```python
def notify_result(request_body: Any, status: str, message: str) -> None:
    """HCP Terraform に Run Task 結果を通知

    Run Task の実行結果を HCP Terraform に通知します。
    コールバック URL とアクセストークンは、リクエストペイロードから取得した値を使用します。

    Args:
        request_body: リクエストボディ
        status: 実行結果のステータス（"passed" or "failed"）
        message: 実行結果のメッセージ
    """
    callback_url = request_body.get("task_result_callback_url")
    access_token = request_body.get("access_token")

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/vnd.api+json",
    }
    payload = {
        "data": {
            "type": "task-results",
            "attributes": {"status": status, "message": message},
        }
    }
    data = json.dumps(payload).encode()
    req = urllib.request.Request(
        url=callback_url, data=data, headers=headers, method="PATCH"
    )
    urllib.request.urlopen(req)
```

## インフラコード実装ポイント

### Lambda の関数 URL の使用

Lambda 関数は、[関数 URL](https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/urls-configuration.html) を使用して HTTP エンドポイントを用意しています。
認証は独自実装 (HMAC 認証)となるので、`authorization_type` は `NONE` を指定します。

```terraform
resource "aws_lambda_function_url" "run_task" {
  function_name      = aws_lambda_function.run_task.function_name
  authorization_type = "NONE"

  cors {
    allow_origins = ["*"]
    allow_methods = ["POST"]
    max_age       = 86400
  }
}
```

### AWS Parameters and Secrets Lambda Extension のレイヤー設定

AWS Parameters and Secrets Lambda Extension を使用するために、Lambda のレイヤー設定をする必要があります。
使用する ARN は[リージョンごとに異なる](https://docs.aws.amazon.com/ja_jp/systems-manager/latest/userguide/ps-integration-lambda-extensions.html#ps-integration-lambda-extensions-add)ため、リージョンによって場合分けが必要です。

```terraform
locals {
  # リージョンごとのレイヤー情報マッピング
  extension_layers = {
    "ap-northeast-1" = {
      account_id = "133490724326"
      version    = "12"
    }
   ...
  }

  extension_layer_arn = "arn:aws:lambda:${var.aws_region}:${local.extension_layers[var.aws_region].account_id}:layer:AWS-Parameters-and-Secrets-Lambda-Extension:${local.extension_layers[var.aws_region].version}"
}

# Lambda関数
resource "aws_lambda_function" "run_task" {
  ...
  layers = [local.extension_layer_arn]
  ...
}
```

とはいえ、構築先のリージョンが限定されているケースが多いと思いますので、実際はここまでの実装はせずに、ARN をベタ書きしても構わないと思います。

## terraform 実装上のポイント

### 別の Workspace の state の参照

Workspace 設定のポイントは下記の Run Task 関連の設定部分です。
Lambda の関数 URL は Lambda 構築後に決定するため、何らかの方法で得る必要があります。
また、Lambda で使用する HMAC 認証キーと同一のキーを Run Task に設定する必要があります。
今回は、Run Task 用 Lambda の構築時の state に出力された値を利用するようにしています。

```hcl
# 01_run_task_resourcesのステートを参照
data "terraform_remote_state" "run_task" {
  backend = "remote"

  config = {
    organization = var.tfc_organization_name
    workspaces = {
      name = var.run_task_workspace_name
    }
  }
}

...

resource "tfe_organization_run_task" "validator" {
  organization = var.tfc_organization_name
  url          = data.terraform_remote_state.run_task.outputs.function_url
  name         = "terraform-validator"
  enabled      = true
  hmac_key     = data.terraform_remote_state.run_task.outputs.hmac_secret_key
}

# WorkspaceへのRun Task関連付け
resource "tfe_workspace_run_task" "validator" {
  workspace_id      = tfe_workspace.example.id
  task_id           = tfe_organization_run_task.validator.id
  enforcement_level = "advisory"
  stages            = ["post_plan"]
}
```

Run Task 用 Lambda の構築時の state には下記のように出力されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3769407/a25ee2a2-1265-ee59-f885-56d762ce11ec.png)

他の Workspace の state の参照になるので、参照先の Workspace の設定で Remote state sharing の設定を行う必要があります。

## 考慮が必要な事項

本構成を採用する前に、以下の観点での検討が推奨されます。

### 可用性への影響

Run Task の導入により、本実装の通り Lambda による検証を導入すると、Lambda 関数の SLA が影響してきます。

これが受け入れ可能か確認し、必要に応じてマルチリージョン構成等で可用性の向上を検討します。

### 攻撃対象の増加

Lambda はインターネット上に公開するエンドポイントを用意することになるので、攻撃対象となる場所が増加します。

対策としては例えば、WAF による保護を検討します。