AWS Security Agent 検証 — GraphQL API に対するペネトレーションテストの検出能力

"""
Deliberately vulnerable GraphQL application for Security Agent penetration testing.
WARNING: Do NOT deploy in production environments.
"""
import os, sqlite3, hashlib, json
from flask import Flask, request, jsonify, g
from ariadne import (
    QueryType, MutationType, ObjectType,
    make_executable_schema, graphql_sync, snake_case_fallback_resolvers,
)
 
app = Flask(__name__)
DATABASE = "/tmp/graphql_vuln.db"
 
# --- Schema ---
type_defs = """
    type Query {
        me: User
        user(id: ID!): User
        users: [User!]!
        post(id: ID!): Post
        posts: [Post!]!
        searchPosts(keyword: String!): [Post!]!
    }
    type Mutation {
        login(username: String!, password: String!): AuthPayload!
        createUser(username: String!, password: String!, email: String!, role: String): User!
        createPost(title: String!, body: String!): Post!
    }
    type AuthPayload { token: String; success: Boolean!; message: String }
    type User {
        id: ID!; username: String!; email: String!; role: String!
        password_hash: String!  # VULN #5: 機密フィールドがスキーマに露出
        posts: [Post!]!
    }
    type Post { id: ID!; title: String!; body: String!; author: User!; comments: [Comment!]! }
    type Comment { id: ID!; text: String!; post: Post!; author: User! }
"""
 
query = QueryType()
mutation = MutationType()
 
# VULN #2: フィールドレベル認可なし — 全ユーザーのデータが取得可能
@query.field("users")
def resolve_users(*_):
    db = get_db()
    return [dict(r) for r in db.execute("SELECT * FROM users").fetchall()]
 
# VULN #3: SQL Injection — f-string で直接 SQL に埋め込み
@query.field("searchPosts")
def resolve_search_posts(_, info, keyword):
    db = get_db()
    sql = f"SELECT * FROM posts WHERE title LIKE '%{keyword}%' OR body LIKE '%{keyword}%'"
    return [dict(r) for r in db.execute(sql).fetchall()]
 
# VULN #4: レート制限なし + トークンが平文ユーザー名
@mutation.field("login")
def resolve_login(_, info, username, password):
    db = get_db()
    pw_hash = hashlib.md5(password.encode()).hexdigest()
    row = db.execute(
        "SELECT * FROM users WHERE username = ? AND password_hash = ?",
        (username, pw_hash),
    ).fetchone()
    if row:
        return {"token": row["username"], "success": True, "message": "Login successful"}
    return {"token": None, "success": False, "message": "Invalid credentials"}
 
# VULN #5: createUser が password_hash を含む完全な User オブジェクトを返す
# + role パラメータを検証なしで受け入れ（Mass Assignment）
@mutation.field("createUser")
def resolve_create_user(_, info, username, password, email, role=None):
    db = get_db()
    pw_hash = hashlib.md5(password.encode()).hexdigest()
    db.execute(
        "INSERT INTO users (username, password_hash, email, role) VALUES (?, ?, ?, ?)",
        (username, pw_hash, email, role or "user"),
    )
    db.commit()
    row = db.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
    return dict(row)  # password_hash を含む
 
# VULN #1: ネストリゾルバに深さ制限なし
# User -> posts -> comments -> author -> posts -> ... (無限再帰可能)
user_type = ObjectType("User")
user_type.set_field("posts", lambda user, *_: [dict(r) for r in get_db().execute(
    "SELECT * FROM posts WHERE author_id = ?", (user["id"],)).fetchall()])
 
post_type = ObjectType("Post")
post_type.set_field("author", lambda post, *_: dict(get_db().execute(
    "SELECT * FROM users WHERE id = ?", (post["author_id"],)).fetchone()))
post_type.set_field("comments", lambda post, *_: [dict(r) for r in get_db().execute(
    "SELECT * FROM comments WHERE post_id = ?", (post["id"],)).fetchall()])
 
comment_type = ObjectType("Comment")
comment_type.set_field("author", lambda c, *_: dict(get_db().execute(
    "SELECT * FROM users WHERE id = ?", (c["author_id"],)).fetchone()))
comment_type.set_field("post", lambda c, *_: dict(get_db().execute(
    "SELECT * FROM posts WHERE id = ?", (c["post_id"],)).fetchone()))
 
schema = make_executable_schema(
    type_defs, query, mutation, user_type, post_type, comment_type,
    snake_case_fallback_resolvers
)

REGION=ap-northeast-1
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
AGENT_SPACE_ID=<your-agent-space-id>
SERVICE_ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role/SecurityAgentPentestRole"
 
# デフォルト VPC・サブネット取得
VPC_ID=$(aws ec2 describe-vpcs --filters "Name=isDefault,Values=true" \
  --query "Vpcs[0].VpcId" --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)
 
# セキュリティグループ作成（ポート 80 を VPC CIDR から許可）
SG_ID=$(aws ec2 create-security-group \
  --group-name "graphql-vuln-app-sg" \
  --description "Security group for GraphQL vulnerable app" \
  --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 172.31.0.0/16 --region $REGION
 
# EC2 起動（Amazon Linux 2023、t3.small）
AMI_ID=$(aws ec2 describe-images --owners amazon \
  --filters "Name=name,Values=al2023-ami-2023.*-x86_64" "Name=state,Values=available" \
  --query "sort_by(Images, &CreationDate)[-1].ImageId" \
  --output text --region $REGION)
INSTANCE_ID=$(aws ec2 run-instances \
  --image-id $AMI_ID --instance-type t3.small \
  --subnet-id $SUBNET_ID --security-group-ids $SG_ID \
  --tag-specifications "ResourceType=instance,Tags=[{Key=Name,Value=graphql-vuln-app-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)
 
# SSM 経由でアプリデプロイ
aws ssm send-command --instance-ids $INSTANCE_ID \
  --document-name "AWS-RunShellScript" \
  --parameters commands='[
    "sudo dnf install -y python3.11 python3.11-pip",
    "sudo python3.11 -m pip install flask==3.1.0 ariadne==0.24.0 gunicorn",
    "sudo mkdir -p /opt/graphql-vuln-app"
  ]' --region $REGION
 
# app.py を転送（base64 エンコード経由）
APP_B64=$(base64 -w0 app.py)
aws ssm send-command --instance-ids $INSTANCE_ID \
  --document-name "AWS-RunShellScript" \
  --parameters commands="[
    \"echo '$APP_B64' | base64 -d | sudo tee /opt/graphql-vuln-app/app.py > /dev/null\",
    \"cd /opt/graphql-vuln-app && sudo python3.11 -c \\\"exec(open('app.py').read().split('if __name__')[0]); init_db()\\\"\",
    \"sudo systemd-run --unit=graphql-vuln-app --working-directory=/opt/graphql-vuln-app gunicorn --bind 0.0.0.0:80 --workers 4 --timeout 120 app:app\"
  ]" --region $REGION

# Target Domain 登録
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 verify-target-domain \
  --target-domain-id $TARGET_DOMAIN_ID --region $REGION
# プライベート DNS は UNREACHABLE になるが、VPC Config 経由のペンテストでは問題ない
 
# Agent Space 更新（新しい Target Domain を追加）
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
 
# ペンテスト作成（条件 A: URL のみ）
LOG_GROUP="/aws/securityagent/graphql-pentest"
aws logs create-log-group --log-group-name $LOG_GROUP --region $REGION 2>/dev/null
PENTEST_ID=$(aws securityagent create-pentest \
  --title "graphql-vuln-app-pentest-url-only" \
  --agent-space-id $AGENT_SPACE_ID \
  --assets "{\"endpoints\": [{\"uri\": \"http://${PRIVATE_DNS}\"}]}" \
  --service-role $SERVICE_ROLE_ARN \
  --log-config "{\"logGroup\": \"$LOG_GROUP\"}" \
  --vpc-config "{\"vpcArn\": \"$VPC_ID\", \"securityGroupArns\": [\"$SG_ID\"], \"subnetArns\": [\"$SUBNET_ID\"]}" \
  --region $REGION --query "pentestId" --output text)
 
# ペンテスト実行
aws securityagent start-pentest-job \
  --pentest-id $PENTEST_ID \
  --agent-space-id $AGENT_SPACE_ID \
  --region $REGION
 
# ステータス確認
aws securityagent batch-get-pentest-jobs \
  --pentest-job-ids "<pentest-job-id>" \
  --agent-space-id $AGENT_SPACE_ID \
  --region $REGION \
  --query "pentestJobs[0].{Status:status}" --output table

# S3 にスキーマ・API ドキュメントをアップロード
# schema.graphql はソースコード内の type_defs と同じ内容
# api-documentation.md は各 Query/Mutation の説明と認証要件を記載
S3_BUCKET="security-agent-graphql-docs-${ACCOUNT_ID}"
aws s3 mb "s3://${S3_BUCKET}" --region $REGION
aws s3 cp schema.graphql "s3://${S3_BUCKET}/" --region $REGION
aws s3 cp api-documentation.md "s3://${S3_BUCKET}/" --region $REGION
 
# Agent Space に S3 バケットを登録
aws securityagent update-agent-space \
  --agent-space-id $AGENT_SPACE_ID --name "pentest-verification" \
  --aws-resources "{\"vpcs\": [...], \"iamRoles\": [...], \"s3Buckets\": [\"$S3_BUCKET\"]}" \
  --region $REGION
 
# ペンテスト作成（documents 付き）
aws securityagent create-pentest \
  --title "graphql-vuln-app-pentest-with-docs" \
  --agent-space-id $AGENT_SPACE_ID \
  --assets "{\"endpoints\": [{\"uri\": \"http://${PRIVATE_DNS}\"}], \"documents\": [{\"s3Location\": \"s3://${S3_BUCKET}/schema.graphql\"}, {\"s3Location\": \"s3://${S3_BUCKET}/api-documentation.md\"}]}" \
  --service-role $SERVICE_ROLE_ARN \
  --log-config "{\"logGroup\": \"/aws/securityagent/graphql-pentest-docs\"}" \
  --vpc-config "{...}" --region $REGION

# endpoints の uri を /graphql に変更
aws securityagent create-pentest \
  --title "graphql-vuln-app-pentest-explicit-endpoint" \
  --agent-space-id $AGENT_SPACE_ID \
  --assets "{\"endpoints\": [{\"uri\": \"http://${PRIVATE_DNS}/graphql\"}], \"documents\": [...]}" \
  --service-role $SERVICE_ROLE_ARN \
  --log-config "{\"logGroup\": \"/aws/securityagent/graphql-pentest-explicit\"}" \
  --vpc-config "{...}" --region $REGION

@app.route("/")
def index():
    return """<!DOCTYPE html>
<html><head><title>Vulnerable App</title></head>
<body>
<h1>GraphQL Vulnerable App</h1>
<ul>
  <li><a href="/graphql">GraphQL API Explorer</a></li>
  <li><a href="/api/users">Users API</a></li>
  <li><a href="/api/posts">Posts API</a></li>
  <li><a href="/api/search?q=test">Search API</a></li>
</ul>
</body></html>"""
 
@app.route("/graphql", methods=["GET"])
def graphql_playground():
    return """<!DOCTYPE html>
<html><head><title>GraphQL API</title></head>
<body>
<h1>GraphQL API Explorer</h1>
<form method="POST" action="/graphql" enctype="application/x-www-form-urlencoded">
  <textarea name="query" rows="6" cols="60">{ users { id username email role } }</textarea><br>
  <input type="submit" value="Execute Query">
</form>
</body></html>"""
 
@app.route("/graphql", methods=["POST"])
def graphql_server():
    if request.content_type and "json" in request.content_type:
        data = request.get_json()
    else:
        data = {"query": request.form.get("query", "")}
    success, result = graphql_sync(schema, data, context_value={"request": request})
    return jsonify(result), 200 if success else 400
 
# REST 風ルート（クローラーの発見用）
@app.route("/api/users")
def api_users():
    db = get_db()
    users = [dict(r) for r in db.execute("SELECT id, username, email, role FROM users").fetchall()]
    return jsonify({"users": users})
 
@app.route("/api/search")
def api_search():
    q = request.args.get("q", "")
    db = get_db()
    sql = f"SELECT * FROM posts WHERE title LIKE '%{q}%' OR body LIKE '%{q}%'"
    return jsonify({"results": [dict(r) for r in db.execute(sql).fetchall()]})

REGION=ap-northeast-1
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
 
# EC2 インスタンス削除
aws ec2 terminate-instances --instance-ids i-0c33e55680c014e35 --region $REGION
aws ec2 wait instance-terminated --instance-ids i-0c33e55680c014e35 --region $REGION
 
# セキュリティグループ削除
aws ec2 delete-security-group --group-id sg-015279938a1092769 --region $REGION
 
# S3 バケット削除
aws s3 rb s3://security-agent-graphql-docs-${ACCOUNT_ID} --force --region $REGION
 
# CloudWatch ロググループ削除
for LG in graphql-pentest graphql-pentest-docs graphql-pentest-explicit graphql-pentest-v2; do
  aws logs delete-log-group --log-group-name /aws/securityagent/$LG --region $REGION 2>/dev/null
done
 
# Target Domain 削除
aws securityagent delete-target-domain \
  --target-domain-id td-038a9d82-90d2-5897-bf51-44b88ece712f --region $REGION