Claude CodeでETLジョブ実行テストを自動化してみた

aws s3 ls && rm -rf /

65_aws_cdk_etl_auto_test/
├── .claude/
│   ├── skills/
│   │   ├── prepare-integration-test/
│   │   │   ├── references/
│   │   │   │   ├── INSTRUCTIONS.md
│   │   │   │   └── test-cases-template.md
│   │   │   └── SKILL.md
│   │   └── run-integration-test/
│   │       ├── references/
│   │       │   └── INSTRUCTIONS.md
│   │       └── SKILL.md
│   ├── credential-process-mfa.sh
│   ├── settings.local.json.example
│   └── validate_commands.py
├── cfn/
│   └── claude-code-resources.yaml
├── drawio/
│   ├── etl_pipeline.drawio
│   └── skills_hooks_automation.drawio
└── eventbridge-sfn-iceberg/
    ├── cdk/
    │   ├── bin/
    │   │   └── app.ts
    │   ├── layers/
    │   │   └── requirements.txt
    │   ├── lib/
    │   │   ├── data-pipeline-stack.ts
    │   │   └── parameter.ts
    │   ├── test/
    │   │   └── data-pipeline-stack.test.ts
    │   └── package.json
    ├── resources/
    │   ├── data/
    │   │   ├── orders_2025-12-01.csv
    │   │   ├── orders_2025-12-02.csv
    │   │   ├── orders_2025-12-03.csv
    │   │   └── orders_2025-12-10.csv
    │   └── lambda/
    │       └── process_and_load.py
    └── test-outputs/
        ├── evidences/
        │   ├── 01-xx.md
        │   └── 0x-xx.md
        └── test-plan.md

AWSTemplateFormatVersion: "2010-09-09"
Description: Claude Code operator IAM role and shared resources

Parameters:
  ProjectName:
    Type: String
    Description: Project name in kebab-case (e.g. my-project). Used for resource names like Athena WorkGroup.
  RoleNamePrefix:
    Type: String
    Description: Project name in PascalCase (e.g. MyProject). Used for IAM role name.
  TrustedPrincipalArns:
    Type: CommaDelimitedList
    Description: Comma-separated IAM ARNs (e.g. arn:aws:iam::123456789012:user/alice,arn:aws:iam::123456789012:user/bob)

Resources:
  ClaudeCodeOperatorRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub ${RoleNamePrefix}ClaudeCodeOperatorRole
      MaxSessionDuration: 3600
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              AWS: !Ref TrustedPrincipalArns
            Action: sts:AssumeRole
            Condition:
              Bool:
                aws:MultiFactorAuthPresent: "true"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/ReadOnlyAccess
      Policies:
        - PolicyName: ClaudeCodeJobExecution
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Sid: StepFunctionsExecution
                Effect: Allow
                Action:
                  - states:StartExecution
                Resource: "*"
              - Sid: GlueJobExecution
                Effect: Allow
                Action:
                  - glue:StartJobRun
                Resource: "*"
              - Sid: EventBridgePutEvents
                Effect: Allow
                Action:
                  - events:PutEvents
                Resource: "*"
              - Sid: AthenaQueryExecution
                Effect: Allow
                Action:
                  - athena:StartQueryExecution
                  - athena:StopQueryExecution
                  - athena:GetQueryExecution
                  - athena:GetQueryResults
                Resource: "*"
              - Sid: S3TestDataSetup
                Effect: Allow
                Action:
                  - s3:PutObject
                Resource: "arn:aws:s3:::*/*"

  ClaudeCodeAthenaWorkGroup:
    Type: AWS::Athena::WorkGroup
    Properties:
      Name: !Sub ${ProjectName}-claude-code
      Description: Athena workgroup for Claude Code with managed query results
      State: ENABLED
      WorkGroupConfiguration:
        EnforceWorkGroupConfiguration: true
        ManagedQueryResultsConfiguration:
          Enabled: true

#!/bin/bash
# credential-process-mfa.sh
#
# Wrapper script for AWS credential_process
# Retrieves an MFA token from 1Password CLI and returns temporary credentials via assume-role
#
# Usage (in ~/.aws/config):
#   [profile your-profile]
#   credential_process = /path/to/credential-process-mfa.sh <role_arn> <mfa_serial> <source_profile> [op_item_name]
#
# Temporary credentials are cached and reused while still valid
# Expired cache is overwritten on next invocation (self-managed lifecycle)

set -euo pipefail

# --- Arguments ---

CACHE_DIR="${HOME}/.aws/cli/cache"

ROLE_ARN="${1:?Usage: credential-process-mfa.sh <role_arn> <mfa_serial> <source_profile> [op_item_name]}"
MFA_SERIAL="${2:?Usage: credential-process-mfa.sh <role_arn> <mfa_serial> <source_profile> [op_item_name]}"
SOURCE_PROFILE="${3:?Usage: credential-process-mfa.sh <role_arn> <mfa_serial> <source_profile> [op_item_name]}"
OP_ITEM_NAME="${4:-AWS}"

# --- Cache ---

CACHE_KEY=$(echo -n "${ROLE_ARN}:${MFA_SERIAL}" | shasum -a 256 | cut -c1-16)
CACHE_FILE="${CACHE_DIR}/credential-process-mfa-${CACHE_KEY}.json"

mkdir -p "$CACHE_DIR"
find "$CACHE_DIR" -name "credential-process-mfa-*.json" -mmin +60 -delete 2>/dev/null

if [[ -f "$CACHE_FILE" ]]; then
  EXP_STR=$(jq -r '.Expiration // empty' "$CACHE_FILE" 2>/dev/null)
  if [[ -n "$EXP_STR" ]]; then
    # Normalize timezone for BSD date: Z → +0000, +00:00 → +0000
    EXP_NORM=$(echo "$EXP_STR" | sed -E 's/Z$/+0000/; s/([+-][0-9]{2}):([0-9]{2})$/\1\2/')
    EXP_EPOCH=$(date -jf "%Y-%m-%dT%H:%M:%S%z" "$EXP_NORM" +%s 2>/dev/null) || EXP_EPOCH=0
    NOW_EPOCH=$(date +%s)
    REMAINING=$((EXP_EPOCH - NOW_EPOCH))
  else
    REMAINING=0
  fi

  if [[ "$REMAINING" -gt 300 ]]; then
    cat "$CACHE_FILE"
    exit 0
  fi
fi

# --- MFA Token ---

OTP=$(op item get "$OP_ITEM_NAME" --otp 2>/dev/null) || {
  echo "Error: Failed to get OTP from 1Password CLI" >&2
  echo "Ensure 1Password CLI is signed in (run: eval \$(op signin))" >&2
  exit 1
}

# --- Assume Role ---

RESPONSE=$(aws sts assume-role \
  --role-arn "$ROLE_ARN" \
  --role-session-name "claude-code-$(date +%s)" \
  --serial-number "$MFA_SERIAL" \
  --token-code "$OTP" \
  --profile "$SOURCE_PROFILE" \
  --duration-seconds 3600 \
  --output json 2>&1) || {
  echo "Error: assume-role failed: $RESPONSE" >&2
  exit 1
}

# --- Format output (credential_process Version 1) ---

OUTPUT=$(echo "$RESPONSE" | jq '{
  Version: 1,
  AccessKeyId: .Credentials.AccessKeyId,
  SecretAccessKey: .Credentials.SecretAccessKey,
  SessionToken: .Credentials.SessionToken,
  Expiration: .Credentials.Expiration
}') || {
  echo "Error: Failed to parse assume-role response" >&2
  exit 1
}

(umask 077 && echo "$OUTPUT" > "$CACHE_FILE")

echo "$OUTPUT"

[profile your-profile]
credential_process = /path/to/credential-process-mfa.sh arn:aws:iam::123456789012:role/MyProjectClaudeCodeOperatorRole arn:aws:iam::123456789012:mfa/<user-name> default <op-item-name>

#!/usr/bin/env python3
"""PreToolUse hook: validate Bash tool commands (default-deny policy).

Receives a JSON object on stdin:
  {"tool_name":"Bash","tool_input":{"command":"..."}}

Only active when CLAUDE_STRICT_HOOKS=1 (i.e. --dangerously-skip-permissions mode).
In normal mode, permissions.deny handles safety so this hook is a no-op.

Outputs a Claude Code hook response on stdout when denying.
Exits 0 silently when the command is allowed.
"""

import json
import os
import re
import shlex
import sys
from typing import NoReturn

SHELL_COMMANDS = {
    # File inspection
    "ls", "cat", "head", "tail", "wc", "diff", "find", "tree",
    # Shell basics
    "echo", "pwd", "date", "sleep",
    # Data processing
    "jq", "python", "grep",
    # Directory/file operations (non-destructive)
    "mkdir", "cp",
}

AWS_WRITE_COMMANDS = {
    "start-execution",        # Step Functions
    "start-job-run",          # Glue
    "start-query-execution",  # Athena
    "stop-query-execution",   # Athena
    "put-events",             # EventBridge
    "put-object",             # S3
    "copy-object",            # S3
    # S3 high-level commands
    "ls", "cp",
}

AWS_READ_PREFIXES = ("describe-", "list-", "get-", "head-", "filter-")

# shlex returns && and || as single tokens
OPERATORS = {"|", "||", "&", "&&", ";"}

def deny(message: str) -> NoReturn:
    print(json.dumps({"decision": "block", "reason": message}))
    sys.exit(0)

def check_aws(tokens: list) -> None:
    """Validate AWS CLI subcommand against the allow-list."""
    # Skip global flags (--flag or --flag value) to find: aws <service> <subcommand>
    non_flags = []
    skip_next = False
    for token in tokens:
        if skip_next:
            skip_next = False
            continue
        if token.startswith("--"):
            if "=" not in token:
                skip_next = True  # next token is the flag value
            continue
        non_flags.append(token)

    # non_flags: ["aws", "<service>", "<subcommand>", ...]
    if len(non_flags) < 3:
        return

    subcommand = non_flags[2]
    if any(subcommand.startswith(p) for p in AWS_READ_PREFIXES):
        return
    if subcommand in AWS_WRITE_COMMANDS:
        return

    deny(
        f"AWS subcommand '{subcommand}' is not in the allow-list. "
        f"Permitted: read prefixes {list(AWS_READ_PREFIXES)} "
        f"and write commands {sorted(AWS_WRITE_COMMANDS)}."
    )

def split_segments(tokens: list) -> list:
    """Split token list into command segments at shell operators."""
    segments = []
    current = []
    for token in tokens:
        if token in OPERATORS:
            if current:
                segments.append(current)
                current = []
        else:
            current.append(token)
    if current:
        segments.append(current)
    return segments

def validate_segment(tokens: list) -> None:
    """Validate a single command segment."""
    # Strip leading variable assignments (KEY=value)
    while tokens and re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", tokens[0]):
        tokens = tokens[1:]

    if not tokens:
        return

    # Strip path prefix (e.g. /usr/bin/jq -> jq)
    cmd = tokens[0].rsplit("/", 1)[-1]

    if not cmd or cmd.startswith("#"):
        return

    if cmd == "aws":
        check_aws(tokens)
    elif cmd not in SHELL_COMMANDS:
        deny(
            f"Command '{cmd}' is not in the allow-list. "
            f"Permitted commands: {sorted(SHELL_COMMANDS)}"
        )

def main() -> None:
    if os.environ.get("CLAUDE_STRICT_HOOKS") != "1":
        sys.exit(0)

    data = json.load(sys.stdin)

    if data.get("tool_name") != "Bash":
        sys.exit(0)

    command = data.get("tool_input", {}).get("command", "")
    if not command:
        sys.exit(0)

    try:
        lex = shlex.shlex(command, posix=True, punctuation_chars=True)
        tokens = list(lex)
    except ValueError as e:
        deny(f"Could not parse command: {e}")

    for segment in split_segments(tokens):
        validate_segment(segment)

if __name__ == "__main__":
    main()

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "command": "python3 .claude/validate_commands.py"
          }
        ]
      }
    ]
  }
}

---
name: prepare-integration-test
description: Integration test preparation. Create test specs, verify prerequisites, upload test data, generate workflow.json.
argument-hint: "[project-path]"
model: sonnet
allowed-tools: Read, Write, Edit, Bash, Glob, Grep, AskUserQuestion
---

The project directory path is: $ARGUMENTS
Use this path as the `(project)` placeholder in all steps described in INSTRUCTIONS.md.

See [INSTRUCTIONS.md](references/INSTRUCTIONS.md) for detailed steps.

---
name: run-integration-test
description: Execute integration tests from workflow.json. Must run in --dangerously-skip-permissions session. PreToolUse hooks enforce safety via default-deny policy.
argument-hint: "[project-path]"
model: sonnet
allowed-tools: Read, Write, Edit, Bash, Glob, Grep
---

The project directory path is: $ARGUMENTS
Use this path to locate workflow.json at `(project)/workflow.json`.

See [INSTRUCTIONS.md](references/INSTRUCTIONS.md) for detailed steps.

aws cloudformation deploy \
  --template-file cfn/claude-code-resources.yaml \
  --stack-name claude-code-resources \
  --capabilities CAPABILITY_NAMED_IAM \
  --parameter-overrides \
    ProjectName=<your-project-name> \
    RoleNamePrefix=<YourPrefix> \
    TrustedPrincipalArns=<arn:aws:iam::123456789012:user/alice> \
  --profile <your-profile>

cd eventbridge-sfn-iceberg/cdk
pnpm install
pnpm cdk deploy --all --profile <your-profile>

cp -r 65_aws_cdk_etl_auto_test/.claude/ <your-repo>/.claude/

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "command": "python3 .claude/validate_commands.py"
          }
        ]
      }
    ]
  }
}

CLAUDE_STRICT_HOOKS=1 claude --dangerously-skip-permissions

/run-integration-test <project-path>

claude --dangerously-skip-permissions

CLAUDE_STRICT_HOOKS=1 claude --dangerously-skip-permissions
rm -rf /tmp/test

aws s3 ls s3://<source-bucket>/ --profile <profile> && aws s3 rb s3://<source-bucket> --profile <profile>