AWS Security Agent 検証 — オンデマンド・ペネトレーションテストの検出品質とコスト

ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
REGION=ap-northeast-1
 
# Security Agent 用サービスロール
cat > securityagent-trust-policy.json << EOF
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": { "Service": "securityagent.amazonaws.com" },
    "Action": "sts:AssumeRole",
    "Condition": {
      "StringEquals": { "aws:SourceAccount": "${ACCOUNT_ID}" }
    }
  }]
}
EOF
 
aws iam create-role \
  --role-name SecurityAgentPentestRole \
  --assume-role-policy-document file://securityagent-trust-policy.json
 
# VPC アクセス用ポリシー（ENI 作成に必要）
aws iam attach-role-policy \
  --role-name SecurityAgentPentestRole \
  --policy-arn arn:aws:iam::aws:policy/AmazonVPCFullAccess
 
# CloudWatch Logs ポリシー（ドキュメント未記載だが必須）
cat > logs-policy.json << EOF
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
    "Resource": "arn:aws:logs:${REGION}:${ACCOUNT_ID}:log-group:/aws/securityagent/*"
  }]
}
EOF
 
aws iam put-role-policy \
  --role-name SecurityAgentPentestRole \
  --policy-name SecurityAgentLogsAccess \
  --policy-document file://logs-policy.json

AGENT_SPACE_ID=$(aws securityagent create-agent-space \
  --name "pentest-verification" \
  --region $REGION \
  --query "agentSpaceId" --output text)
 
echo "Agent Space: $AGENT_SPACE_ID"

TARGET_DOMAIN_ID=$(aws securityagent create-target-domain \
  --target-domain-name "$PRIVATE_DNS" \
  --verification-method HTTP_ROUTE \
  --region $REGION \
  --query "targetDomainId" --output text)
 
# 検証トークンを取得
aws securityagent batch-get-target-domains \
  --target-domain-ids $TARGET_DOMAIN_ID \
  --region $REGION \
  --query "targetDomains[0].verificationDetails.httpRoute"

{
  "token": "Vdm6-_8EVhQZR3Xf0hOwug",
  "routePath": ".well-known/aws/securityagent-domain-verification.json"
}

SERVICE_ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role/SecurityAgentPentestRole"
 
aws securityagent update-agent-space \
  --agent-space-id $AGENT_SPACE_ID \
  --name "pentest-verification" \
  --aws-resources "{
    \"vpcs\": [{
      \"vpcArn\": \"$VPC_ID\",
      \"securityGroupArns\": [\"$SG_ID\"],
      \"subnetArns\": [\"$SUBNET_ID\"]
    }],
    \"iamRoles\": [\"$SERVICE_ROLE_ARN\"]
  }" \
  --target-domain-ids "$TARGET_DOMAIN_ID" \
  --region $REGION

PENTEST_ID=$(aws securityagent create-pentest \
  --title "vulnerable-flask-app-pentest" \
  --agent-space-id $AGENT_SPACE_ID \
  --service-role $SERVICE_ROLE_ARN \
  --assets "{
    \"endpoints\": [{\"uri\": \"http://${PRIVATE_DNS}\"}]
  }" \
  --vpc-config "{
    \"vpcArn\": \"$VPC_ID\",
    \"securityGroupArns\": [\"$SG_ID\"],
    \"subnetArns\": [\"$SUBNET_ID\"]
  }" \
  --region $REGION \
  --query "pentestId" --output text)
 
echo "Pentest: $PENTEST_ID"

"""
Deliberately vulnerable Flask application for Security Agent penetration testing.
WARNING: Do NOT deploy in production environments.
"""
import os, sqlite3, hashlib, json, base64
from flask import Flask, request, jsonify, g, send_file
 
app = Flask(__name__)
app.config["SECRET_KEY"] = "hardcoded-secret-key-123"
DATABASE = "/tmp/vuln_app.db"
UPLOAD_DIR = "/tmp/uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)
 
def get_db():
    if "db" not in g:
        g.db = sqlite3.connect(DATABASE)
        g.db.row_factory = sqlite3.Row
    return g.db
 
@app.teardown_appcontext
def close_db(exception):
    db = g.pop("db", None)
    if db is not None:
        db.close()
 
def init_db():
    db = sqlite3.connect(DATABASE)
    db.executescript("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            password TEXT NOT NULL,
            email TEXT, role TEXT DEFAULT 'user');
        CREATE TABLE IF NOT EXISTS comments (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER, content TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP);
    """)
    db.execute("INSERT OR IGNORE INTO users (username, password, email, role) VALUES (?, ?, ?, ?)",
               ("admin", hashlib.md5(b"admin123").hexdigest(), "admin@example.com", "admin"))
    db.execute("INSERT OR IGNORE INTO users (username, password, email, role) VALUES (?, ?, ?, ?)",
               ("alice", hashlib.md5(b"password1").hexdigest(), "alice@example.com", "user"))
    db.execute("INSERT OR IGNORE INTO users (username, password, email, role) VALUES (?, ?, ?, ?)",
               ("bob", hashlib.md5(b"password2").hexdigest(), "bob@example.com", "user"))
    db.commit()
    db.close()
 
# --- Vulnerability 1: SQL Injection ---
@app.route("/api/login", methods=["POST"])
def login():
    data = request.get_json()
    username, password = data.get("username", ""), data.get("password", "")
    password_hash = hashlib.md5(password.encode()).hexdigest()
    query = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password_hash}'"
    try:
        user = get_db().execute(query).fetchone()
    except Exception as e:
        return jsonify({"error": str(e)}), 500
    if user:
        payload = base64.b64encode(json.dumps(
            {"user_id": user["id"], "username": user["username"], "role": user["role"]}
        ).encode()).decode()
        return jsonify({"token": f"header.{payload}.nosignature", "user_id": user["id"]})
    return jsonify({"error": "Invalid credentials"}), 401
 
# --- Vulnerability 2: Stored XSS ---
@app.route("/api/comments", methods=["GET"])
def get_comments():
    comments = get_db().execute(
        "SELECT c.id, c.content, c.created_at, u.username "
        "FROM comments c JOIN users u ON c.user_id = u.id ORDER BY c.created_at DESC"
    ).fetchall()
    return jsonify([dict(c) for c in comments])
 
@app.route("/api/comments", methods=["POST"])
def post_comment():
    token = request.headers.get("Authorization", "")
    user = decode_token(token)
    if not user:
        return jsonify({"error": "Unauthorized"}), 401
    content = request.get_json().get("content", "")
    get_db().execute("INSERT INTO comments (user_id, content) VALUES (?, ?)", (user["user_id"], content))
    get_db().commit()
    return jsonify({"message": "Comment posted"}), 201
 
# --- Vulnerability 3: IDOR ---
@app.route("/api/users/<int:user_id>", methods=["GET"])
def get_user_profile(user_id):
    token = request.headers.get("Authorization", "")
    user = decode_token(token)
    if not user:
        return jsonify({"error": "Unauthorized"}), 401
    profile = get_db().execute("SELECT id, username, email, role FROM users WHERE id = ?",
                               (user_id,)).fetchone()
    return jsonify(dict(profile)) if profile else (jsonify({"error": "Not found"}), 404)
 
# --- Vulnerability 4: JWT signature not verified ---
def decode_token(auth_header):
    try:
        token = auth_header.split("Bearer ")[1]
        payload = json.loads(base64.b64decode(token.split(".")[1] + "=="))
        return payload
    except Exception:
        return None
 
# --- Vulnerability 5: Path Traversal ---
@app.route("/api/files/download", methods=["GET"])
def download_file():
    filename = request.args.get("filename", "")
    filepath = os.path.join(UPLOAD_DIR, filename)
    return send_file(filepath) if os.path.exists(filepath) else (jsonify({"error": "Not found"}), 404)
 
@app.route("/health")
def health():
    return jsonify({"status": "ok"})
 
@app.route("/")
def index():
    comments = get_db().execute(
        "SELECT c.content, u.username FROM comments c JOIN users u ON c.user_id = u.id"
    ).fetchall()
    comments_html = "".join(f"<div class='comment'><b>{c['username']}</b>: {c['content']}</div>" for c in comments)
    return f"<!DOCTYPE html><html><head><title>Vulnerable App</title></head><body>"  \
           f"<h1>Comment Board</h1>{comments_html}</body></html>"
 
if __name__ == "__main__":
    init_db()
    app.run(host="0.0.0.0", port=80)

REGION=ap-northeast-1
VPC_ID=$(aws ec2 describe-vpcs --filters "Name=isDefault,Values=true" \
  --query "Vpcs[0].VpcId" --output text --region $REGION)
VPC_CIDR=$(aws ec2 describe-vpcs --vpc-ids $VPC_ID \
  --query "Vpcs[0].CidrBlock" --output text --region $REGION)
SUBNET_ID=$(aws ec2 describe-subnets --filters "Name=vpc-id,Values=$VPC_ID" \
  --query "Subnets[0].SubnetId" --output text --region $REGION)
AMI_ID=$(aws ssm get-parameters \
  --names /aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-x86_64 \
  --query "Parameters[0].Value" --output text --region $REGION)

# SSM 接続用ロール
cat > ec2-trust-policy.json << 'EOF'
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": { "Service": "ec2.amazonaws.com" },
    "Action": "sts:AssumeRole"
  }]
}
EOF
 
aws iam create-role --role-name SecurityAgentPentestEC2Role \
  --assume-role-policy-document file://ec2-trust-policy.json
aws iam attach-role-policy --role-name SecurityAgentPentestEC2Role \
  --policy-arn arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore
aws iam create-instance-profile \
  --instance-profile-name SecurityAgentPentestEC2Profile
aws iam add-role-to-instance-profile \
  --instance-profile-name SecurityAgentPentestEC2Profile \
  --role-name SecurityAgentPentestEC2Role
 
sleep 10  # IAM 伝播待ち

# SG: VPC 内からの HTTP 80 のみ許可
SG_ID=$(aws ec2 create-security-group \
  --group-name security-agent-pentest-target \
  --description "Pentest target - HTTP from VPC only" \
  --vpc-id $VPC_ID --region $REGION --query "GroupId" --output text)
 
aws ec2 authorize-security-group-ingress \
  --group-id $SG_ID --protocol tcp --port 80 \
  --cidr $VPC_CIDR --region $REGION
 
# EC2 起動
INSTANCE_ID=$(aws ec2 run-instances \
  --image-id $AMI_ID --instance-type t3.small \
  --subnet-id $SUBNET_ID --security-group-ids $SG_ID \
  --iam-instance-profile Name=SecurityAgentPentestEC2Profile \
  --associate-public-ip-address \
  --tag-specifications "ResourceType=instance,Tags=[{Key=Name,Value=security-agent-pentest-target}]" \
  --region $REGION --query "Instances[0].InstanceId" --output text)
 
aws ec2 wait instance-running --instance-ids $INSTANCE_ID --region $REGION
 
PRIVATE_DNS=$(aws ec2 describe-instances --instance-ids $INSTANCE_ID \
  --query "Reservations[0].Instances[0].PrivateDnsName" --output text --region $REGION)
echo "Private DNS: $PRIVATE_DNS"

# SSM Agent オンライン待ち
while true; do
  STATUS=$(aws ssm describe-instance-information \
    --filters "Key=InstanceIds,Values=${INSTANCE_ID}" \
    --query "InstanceInformationList[0].PingStatus" --output text --region $REGION 2>/dev/null)
  [ "$STATUS" = "Online" ] && break
  echo "Waiting for SSM Agent..." && sleep 10
done
 
# app.py を EC2 に転送（base64 エンコード経由）
APP_B64=$(base64 -w0 app.py)
aws ssm send-command --instance-ids $INSTANCE_ID \
  --document-name "AWS-RunShellScript" \
  --parameters "{\"commands\":[
    \"sudo mkdir -p /opt/vulnerable-app\",
    \"echo '$APP_B64' | base64 -d | sudo tee /opt/vulnerable-app/app.py > /dev/null\"
  ]}" --region $REGION
 
# gunicorn インストール + DB 初期化 + 起動
aws ssm send-command --instance-ids $INSTANCE_ID \
  --document-name "AWS-RunShellScript" \
  --timeout-seconds 120 \
  --parameters '{"commands":[
    "sudo python3 -m ensurepip --upgrade",
    "sudo python3 -m pip install flask==3.1.0 gunicorn",
    "cd /opt/vulnerable-app && sudo python3 -c \"exec(open('app.py').read().split('if __name__')[0]); init_db()\"",
    "sudo systemd-run --unit=vuln-app --working-directory=/opt/vulnerable-app gunicorn --bind 0.0.0.0:80 --workers 4 --timeout 120 app:app"
  ]}' --region $REGION
 
sleep 5
# 動作確認
aws ssm send-command --instance-ids $INSTANCE_ID \
  --document-name "AWS-RunShellScript" \
  --parameters '{"commands":["curl -s http://localhost/health"]}' \
  --region $REGION

# ペネトレーションテスト削除
aws securityagent batch-delete-pentests \
  --agent-space-id $AGENT_SPACE_ID \
  --pentest-ids $PENTEST_ID \
  --region $REGION
 
# Target Domain 削除
aws securityagent delete-target-domain \
  --target-domain-id $TARGET_DOMAIN_ID \
  --region $REGION
 
# Agent Space 削除
aws securityagent delete-agent-space \
  --agent-space-id $AGENT_SPACE_ID \
  --region $REGION
 
# EC2 インスタンス終了
aws ec2 terminate-instances --instance-ids $INSTANCE_ID --region $REGION
aws ec2 wait instance-terminated --instance-ids $INSTANCE_ID --region $REGION
aws ec2 delete-security-group --group-id $SG_ID --region $REGION
 
# IAM ロール削除
aws iam detach-role-policy --role-name SecurityAgentPentestRole \
  --policy-arn arn:aws:iam::aws:policy/AmazonVPCFullAccess
aws iam delete-role-policy --role-name SecurityAgentPentestRole \
  --policy-name SecurityAgentLogsAccess 2>/dev/null
aws iam delete-role --role-name SecurityAgentPentestRole
 
aws iam detach-role-policy --role-name SecurityAgentPentestEC2Role \
  --policy-arn arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore
aws iam delete-instance-profile --instance-profile-name SecurityAgentPentestEC2Profile
aws iam delete-role --role-name SecurityAgentPentestEC2Role