AWS Security Agent Verification — Penetration Testing Detection Capability Against GraphQL APIs

"""
Deliberately vulnerable GraphQL application for Security Agent penetration testing.
WARNING: Do NOT deploy in production environments.
"""
import sqlite3, hashlib
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/vuln_graphql.db"
 
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 User {
        id: ID!
        username: String!
        password_hash: String!
        email: String!
        role: String!
        posts: [Post!]!
    }
    type Post {
        id: ID!
        title: String!
        body: String!
        author: User!
        comments: [Comment!]!
    }
    type Comment {
        id: ID!
        text: String!
        post: Post!
        author: User!
    }
    type AuthPayload {
        token: String
        success: Boolean!
        message: String
    }
"""
 
query = QueryType()
mutation = MutationType()
user_type = ObjectType("User")
post_type = ObjectType("Post")
comment_type = ObjectType("Comment")
 
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_hash TEXT NOT NULL,
            email TEXT,
            role TEXT DEFAULT 'user');
        CREATE TABLE IF NOT EXISTS posts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT, body TEXT, author_id INTEGER);
        CREATE TABLE IF NOT EXISTS comments (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            text TEXT, post_id INTEGER, author_id INTEGER);
    """)
    for name, email, role in [("admin", "admin@internal.corp", "admin"), ("alice", "alice@example.com", "editor"), ("bob", "bob@example.com", "user")]:
        pw = hashlib.md5(f"{name}123".encode()).hexdigest()
        db.execute(
            "INSERT OR IGNORE INTO users (username, password_hash, email, role) "
            "VALUES (?, ?, ?, ?)",
            (name, pw, email, role))
    for i in range(1, 4):
        db.execute(
            "INSERT OR IGNORE INTO posts (id, title, body, author_id) VALUES (?, ?, ?, ?)",
            (i, f"Post {i}", f"Body of post {i}", 1))
    db.commit()
    db.close()
 
def _get_current_user():
    """Extract user from Authorization header (simple token = username)."""
    auth = request.headers.get("Authorization", "")
    if auth.startswith("Bearer "):
        username = auth[7:]
        row = get_db().execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
        if row:
            return dict(row)
    return None
 
# --- Vulnerability 1: No depth limit on nested queries ---
@user_type.field("posts")
def resolve_user_posts(user, *_):
    rows = get_db().execute("SELECT * FROM posts WHERE author_id = ?", (user["id"],)).fetchall()
    return [dict(r) for r in rows]
 
@post_type.field("author")
def resolve_post_author(post, *_):
    row = get_db().execute("SELECT * FROM users WHERE id = ?", (post["author_id"],)).fetchone()
    return dict(row) if row else None
 
@post_type.field("comments")
def resolve_post_comments(post, *_):
    rows = get_db().execute("SELECT * FROM comments WHERE post_id = ?", (post["id"],)).fetchall()
    return [dict(r) for r in rows]
 
@comment_type.field("author")
def resolve_comment_author(comment, *_):
    row = get_db().execute("SELECT * FROM users WHERE id = ?", (comment["author_id"],)).fetchone()
    return dict(row) if row else None
 
# --- Vulnerability 2: No field-level authorization (password_hash exposed) ---
@query.field("user")
def resolve_user(_, info, id):
    row = get_db().execute("SELECT * FROM users WHERE id = ?", (id,)).fetchone()
    return dict(row) if row else None
 
@query.field("users")
def resolve_users(*_):
    return [dict(r) for r in get_db().execute("SELECT * FROM users").fetchall()]
 
@query.field("posts")
def resolve_posts(*_):
    return [dict(r) for r in get_db().execute("SELECT * FROM posts").fetchall()]
 
# --- Vulnerability 3: SQL Injection via f-string ---
@query.field("searchPosts")
def resolve_search_posts(_, info, keyword):
    sql = f"SELECT * FROM posts WHERE title LIKE '%{keyword}%' OR body LIKE '%{keyword}%'"
    try:
        return [dict(r) for r in get_db().execute(sql).fetchall()]
    except Exception:
        return []
 
# --- Vulnerability 4: Login returns username as plaintext token ---
@mutation.field("login")
def resolve_login(_, info, username, password):
    pw = hashlib.md5(password.encode()).hexdigest()
    row = get_db().execute(
        "SELECT * FROM users WHERE username = ? AND password_hash = ?", (username, pw)
    ).fetchone()
    if not row:
        return {"token": None, "success": False, "message": "Invalid credentials"}
    return {"token": username, "success": True, "message": "Login successful"}
 
# --- Vulnerability 5: createUser accepts role + returns password_hash (mass assignment) ---
@mutation.field("createUser")
def resolve_create_user(_, info, username, password, email=None, role=None):
    pw = hashlib.md5(password.encode()).hexdigest()
    db = get_db()
    db.execute(
        "INSERT INTO users (username, password_hash, email, role) VALUES (?, ?, ?, ?)",
        (username, pw, email, role or "user"))
    db.commit()
    row = db.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
    return dict(row)
 
@mutation.field("createPost")
def resolve_create_post(_, info, title, body):
    user = _get_current_user()
    if not user:
        raise Exception("Authentication required")
    db = get_db()
    db.execute("INSERT INTO posts (title, body, author_id) VALUES (?, ?, ?)",
               (title, body, user["id"]))
    db.commit()
    row = db.execute("SELECT * FROM posts ORDER BY id DESC LIMIT 1").fetchone()
    return dict(row)
 
schema = make_executable_schema(
    type_defs, query, mutation, user_type, post_type, comment_type,
    snake_case_fallback_resolvers,
)
 
@app.route("/graphql", methods=["POST"])
def graphql_endpoint():
    content_type = request.content_type or ""
    if "application/json" in content_type:
        data = request.get_json(force=True)
    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
 
@app.route("/graphql", methods=["GET"])
def graphql_get():
    return jsonify({"message": "GraphQL endpoint. Use POST with query."})
 
@app.route("/health")
def health():
    return jsonify({"status": "ok"})
 
@app.route("/")
def index():
    return jsonify({"app": "GraphQL Vulnerable App", "graphql_endpoint": "/graphql"})
 
if __name__ == "__main__":
    init_db()
    app.run(host="0.0.0.0", port=80)

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"
 
# Get default VPC and subnet
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)
 
# Create security group (allow port 80 from 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
 
# Launch 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)
 
# Deploy app via 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_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

# Register 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
# Private DNS will show UNREACHABLE — this is expected for VPC-based pentests
 
# Update Agent Space with new 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
 
# Condition A: URL only
aws securityagent create-pentest \
  --title "graphql-pentest-url-only" \
  --agent-space-id $AGENT_SPACE_ID \
  --assets "{\"endpoints\": [{\"uri\": \"http://${PRIVATE_DNS}\"}]}" \
  --service-role $SERVICE_ROLE_ARN \
  --vpc-config "{\"vpcArn\": \"$VPC_ID\", \"securityGroupArns\": [\"$SG_ID\"], \"subnetArns\": [\"$SUBNET_ID\"]}" \
  --region $REGION
 
# Condition B: URL + documents (upload schema first)
# schema.graphql = same content as type_defs in app.py
# api-documentation.md = description of each Query/Mutation with auth requirements
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
 
aws securityagent create-pentest \
  --title "graphql-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 \
  --vpc-config "{...}" --region $REGION
 
# Condition C: explicit /graphql endpoint + documents
aws securityagent create-pentest \
  --title "graphql-pentest-explicit-endpoint" \
  --agent-space-id $AGENT_SPACE_ID \
  --assets "{\"endpoints\": [{\"uri\": \"http://${PRIVATE_DNS}/graphql\"}], \"documents\": [...]}" \
  --service-role $SERVICE_ROLE_ARN \
  --vpc-config "{...}" --region $REGION
 
# Condition D: modify app to add HTML links, then run with same params as Condition B

@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-style routes (for crawler discoverability)
@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
 
# Terminate EC2 instance
aws ec2 terminate-instances --instance-ids i-0c33e55680c014e35 --region $REGION
aws ec2 wait instance-terminated --instance-ids i-0c33e55680c014e35 --region $REGION
 
# Delete security group
aws ec2 delete-security-group --group-id sg-015279938a1092769 --region $REGION
 
# Delete S3 bucket
aws s3 rb s3://security-agent-graphql-docs-381492023699 --force --region $REGION
 
# Delete target domain
aws securityagent delete-target-domain \
  --target-domain-id td-038a9d82-90d2-5897-bf51-44b88ece712f --region $REGION
 
# Delete CloudWatch log groups
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