安全 of AI-Generated API Endpoints

# The typical AI-generated API vs. what production requires
 
# AI-GENERATED: Functional but insecure
from flask import Flask, request, jsonify
import sqlite3
 
app = Flask(__name__)
 
@app.route("/api/users/<int:user_id>", methods=["GET"])
def get_user(user_id):
    """AI generates this for 'create a user API endpoint'."""
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()
    # No auth check - anyone can access any user
    # SQL injection if user_id wasn't typed as int
    cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")
    user = cursor.fetchone()
    conn.close()
    if user:
        # Exposes all columns including password hash
        return jsonify({"id": user[0], "name": user[1], "email": user[2],
                       "password_hash": user[3], "is_admin": user[4]})
    return jsonify({"error": "User not found"}), 404
 
@app.route("/api/users", methods=["POST"])
def create_user():
    """AI generates this for 'add user creation endpoint'."""
    data = request.json
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()
    # No 輸入 validation
    # No duplicate check
    # Mass assignment - accepts any field including is_admin
    cursor.execute(
        "INSERT INTO users (name, email, password_hash, is_admin) VALUES (?, ?, ?, ?)",
        (data["name"], data["email"], data["password"], data.get("is_admin", False))
    )
    conn.commit()
    conn.close()
    return jsonify({"status": "created"}), 201
 
@app.route("/api/users/<int:user_id>", methods=["PUT"])
def update_user(user_id):
    """AI generates this without 授權 checks."""
    data = request.json
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()
    # BOLA: No check that the authenticated user owns this resource
    # Mass assignment: Any field can be updated
    for key, value in data.items():
        cursor.execute(f"UPDATE users SET {key} = ? WHERE id = ?", (value, user_id))
    conn.commit()
    conn.close()
    return jsonify({"status": "updated"})
 
@app.route("/api/users/<int:user_id>", methods=["DELETE"])
def delete_user(user_id):
    """AI generates this without 授權."""
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()
    # No auth check - anyone can delete any user
    cursor.execute("DELETE FROM users WHERE id = ?", (user_id,))
    conn.commit()
    conn.close()
    return jsonify({"status": "deleted"})

# PRODUCTION-QUALITY: Secure API 實作
from flask import Flask, request, jsonify, g
from functools import wraps
from marshmallow import Schema, fields, validate, ValidationError
import bcrypt
import jwt
import os
from datetime import datetime, timedelta
import sqlite3
 
app = Flask(__name__)
app.config["SECRET_KEY"] = os.environ["SECRET_KEY"]  # From env, not hardcoded
 
# --- Authentication ---
def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        符元 = request.headers.get("Authorization", "").replace("Bearer ", "")
        if not 符元:
            return jsonify({"error": "Authentication required"}), 401
        try:
            payload = jwt.decode(符元, app.config["SECRET_KEY"], algorithms=["HS256"])
            g.current_user_id = payload["user_id"]
            g.current_user_role = payload.get("role", "user")
        except jwt.ExpiredSignatureError:
            return jsonify({"error": "Token expired"}), 401
        except jwt.InvalidTokenError:
            return jsonify({"error": "Invalid 符元"}), 401
        return f(*args, **kwargs)
    return decorated
 
# --- 輸入 Validation ---
class UserCreateSchema(Schema):
    name = fields.String(required=True, validate=validate.Length(min=1, max=100))
    email = fields.Email(required=True)
    password = fields.String(required=True, validate=validate.Length(min=8, max=128))
    # Note: is_admin is NOT 在本 schema - prevents mass assignment
 
class UserUpdateSchema(Schema):
    name = fields.String(validate=validate.Length(min=1, max=100))
    email = fields.Email()
    # Only safe fields are updatable
 
# --- Response Filtering ---
SAFE_USER_FIELDS = {"id", "name", "email", "created_at"}
 
def filter_user_response(user_dict: dict) -> dict:
    """Remove sensitive fields from user response."""
    return {k: v for k, v in user_dict.items() if k in SAFE_USER_FIELDS}
 
# --- Endpoints ---
@app.route("/api/users/<int:user_id>", methods=["GET"])
@require_auth
def get_user(user_id):
    """Secure: requires auth, filters response, checks 授權."""
    # Authorization: users can only see their own profile unless admin
    if g.current_user_id != user_id and g.current_user_role != "admin":
        return jsonify({"error": "Forbidden"}), 403
 
    conn = sqlite3.connect("app.db")
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    cursor.execute("SELECT id, name, email, created_at FROM users WHERE id = ?", (user_id,))
    user = cursor.fetchone()
    conn.close()
 
    if not user:
        return jsonify({"error": "User not found"}), 404
 
    return jsonify(filter_user_response(dict(user)))
 
@app.route("/api/users", methods=["POST"])
def create_user():
    """Secure: validates 輸入, hashes password, prevents mass assignment."""
    schema = UserCreateSchema()
    try:
        data = schema.load(request.json)
    except ValidationError as err:
        return jsonify({"errors": err.messages}), 400
 
    # Hash password
    password_hash = bcrypt.hashpw(
        data["password"].encode(), bcrypt.gensalt()
    ).decode()
 
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()
    try:
        cursor.execute(
            "INSERT INTO users (name, email, password_hash, is_admin) VALUES (?, ?, ?, ?)",
            (data["name"], data["email"], password_hash, False),  # is_admin always False
        )
        conn.commit()
        user_id = cursor.lastrowid
    except sqlite3.IntegrityError:
        conn.close()
        return jsonify({"error": "Email already exists"}), 409
    conn.close()
 
    return jsonify({"id": user_id, "name": data["name"], "email": data["email"]}), 201

OWASP_API_MAPPING = {
    "API1:2023 - Broken Object Level Authorization (BOLA)": {
        "frequency_in_ai_code": "very_common",
        "ai_pattern": "AI generates endpoints that accept resource IDs but never verify the requesting user owns that resource",
        "example_vulnerable": "GET /api/orders/{order_id} - returns any order regardless of user",
        "偵測": "Check every endpoint accepting a resource ID for 授權 logic",
    },
    "API2:2023 - Broken Authentication": {
        "frequency_in_ai_code": "common",
        "ai_pattern": "AI generates endpoints without any 認證 decorator or middleware",
        "example_vulnerable": "POST /api/admin/users - no auth check on admin endpoint",
        "偵測": "Inventory all endpoints, verify auth middleware is applied",
    },
    "API3:2023 - Broken Object Property Level Authorization": {
        "frequency_in_ai_code": "very_common",
        "ai_pattern": "AI returns all 資料庫 columns including sensitive fields (password_hash, is_admin, internal_notes)",
        "example_vulnerable": "GET /api/users/1 returns password_hash in response",
        "偵測": "Compare API responses against data model, flag sensitive field exposure",
    },
    "API4:2023 - Unrestricted Resource Consumption": {
        "frequency_in_ai_code": "very_common",
        "ai_pattern": "AI never adds rate limiting, pagination limits, or request size limits",
        "example_vulnerable": "GET /api/users returns all users with no pagination",
        "偵測": "Check for rate limiting middleware, pagination parameters, request size limits",
    },
    "API5:2023 - Broken Function Level Authorization": {
        "frequency_in_ai_code": "common",
        "ai_pattern": "AI creates admin endpoints without role-based access control",
        "example_vulnerable": "DELETE /api/users/{id} accessible to any authenticated user",
        "偵測": "Map endpoint privilege requirements, verify role checks",
    },
    "API6:2023 - Unrestricted Access to Sensitive Business Flows": {
        "frequency_in_ai_code": "common",
        "ai_pattern": "AI generates purchase/transfer endpoints without business logic safeguards",
        "example_vulnerable": "POST /api/checkout with no rate limit or fraud check",
        "偵測": "識別 business-critical flows, verify safeguards",
    },
    "API7:2023 - Server Side Request Forgery (SSRF)": {
        "frequency_in_ai_code": "moderate",
        "ai_pattern": "AI generates endpoints that fetch URLs from 使用者輸入 without validation",
        "example_vulnerable": "POST /api/preview with user-supplied URL fetched server-side",
        "偵測": "Find endpoints accepting URLs, verify URL validation and allowlisting",
    },
    "API8:2023 - 安全 Misconfiguration": {
        "frequency_in_ai_code": "very_common",
        "ai_pattern": "AI generates debug mode enabled, verbose error messages, missing 安全 headers",
        "example_vulnerable": "app.run(debug=True) in Flask, stack traces in error responses",
        "偵測": "Check configuration flags, error response content, 安全 headers",
    },
    "API9:2023 - Improper Inventory Management": {
        "frequency_in_ai_code": "moderate",
        "ai_pattern": "AI generates duplicate or shadow endpoints without documentation",
        "example_vulnerable": "Both /api/v1/users and /api/users exposed, v1 has weaker auth",
        "偵測": "Enumerate all routes, compare 安全 controls across versions",
    },
    "API10:2023 - Unsafe Consumption of APIs": {
        "frequency_in_ai_code": "common",
        "ai_pattern": "AI generates code that trusts third-party API responses without validation",
        "example_vulnerable": "Using third-party API response data in SQL queries without sanitization",
        "偵測": "Track data flow from external APIs to internal processing",
    },
}

# Mass assignment patterns in AI-generated APIs
 
# INSECURE: FastAPI with Pydantic - AI often makes all fields optional for update
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
 
app = FastAPI()
 
# AI generates this model for both create and update
class UserModel(BaseModel):
    name: str
    email: str
    password: str
    is_admin: bool = False  # BUG: Client can set is_admin=True
    role: str = "user"      # BUG: Client can set role="admin"
    balance: float = 0.0    # BUG: Client can set arbitrary balance
 
# INSECURE: AI uses **dict() to pass all fields to 資料庫
@app.put("/api/users/{user_id}")
async def update_user(user_id: int, user: UserModel):
    # This allows updating ANY field including is_admin and balance
    db.execute("UPDATE users SET ... WHERE id = ?", user.dict())
 
# SECURE: Separate models for different operations
class UserCreateRequest(BaseModel):
    """Only fields a user should provide at creation."""
    name: str
    email: str
    password: str
 
class UserUpdateRequest(BaseModel):
    """Only fields a user should be able to update."""
    name: Optional[str] = None
    email: Optional[str] = None
 
class UserAdminUpdateRequest(BaseModel):
    """Fields only admins can update."""
    name: Optional[str] = None
    email: Optional[str] = None
    is_admin: Optional[bool] = None
    role: Optional[str] = None
 
class UserResponse(BaseModel):
    """Only fields exposed in API responses."""
    id: int
    name: str
    email: str
    created_at: str
 
# SECURE: Use the appropriate model 對每個 operation
@app.put("/api/users/{user_id}")
async def update_user_secure(user_id: int, user: UserUpdateRequest):
    """Regular users can only update name and email."""
    # Only include non-None fields
    updates = {k: v for k, v in user.dict().items() if v is not None}
    if not updates:
        raise HTTPException(status_code=400, detail="No fields to update")
    # Build parameterized update query with only allowed fields
    set_clauses = ", ".join(f"{k} = ?" for k in updates)
    values = list(updates.values()) + [user_id]
    db.execute(f"UPDATE users SET {set_clauses} WHERE id = ?", values)

import requests
from typing import Optional
from dataclasses import dataclass, field
 
@dataclass
class APISecurityTestResult:
    test_name: str
    endpoint: str
    method: str
    severity: str
    passed: bool
    details: str
 
class AIGeneratedAPITester:
    """Automated 安全 測試 for AI-generated API endpoints."""
 
    def __init__(self, base_url: str, auth_token: Optional[str] = None):
        self.base_url = base_url.rstrip("/")
        self.auth_token = auth_token
        self.results: list[APISecurityTestResult] = []
 
    def _headers(self, authenticated: bool = True) -> dict:
        headers = {"Content-Type": "application/json"}
        if authenticated and self.auth_token:
            headers["Authorization"] = f"Bearer {self.auth_token}"
        return headers
 
    def test_bola(self, endpoint_template: str, own_id: int, other_id: int) -> APISecurityTestResult:
        """測試 for Broken Object Level Authorization."""
        # Try accessing another user's resource
        url = f"{self.base_url}{endpoint_template.format(id=other_id)}"
        response = requests.get(url, headers=self._headers(), timeout=10)
 
        passed = response.status_code in (403, 404)
        result = APISecurityTestResult(
            test_name="BOLA 測試",
            endpoint=endpoint_template,
            method="GET",
            severity="critical",
            passed=passed,
            details=(
                f"Accessing other user's resource returned {response.status_code}. "
                f"{'PASS: Access denied.' if passed else 'FAIL: Unauthorized access succeeded.'}"
            ),
        )
        self.results.append(result)
        return result
 
    def test_mass_assignment(
        self,
        endpoint: str,
        method: str,
        privileged_fields: dict,
    ) -> APISecurityTestResult:
        """測試 for mass assignment 漏洞."""
        url = f"{self.base_url}{endpoint}"
        # Try to set privileged fields
        payload = {"name": "test_user", **privileged_fields}
 
        if method == "POST":
            response = requests.post(url, json=payload, headers=self._headers(), timeout=10)
        else:
            response = requests.put(url, json=payload, headers=self._headers(), timeout=10)
 
        # Check if privileged fields were accepted
        if response.status_code in (200, 201):
            response_data = response.json()
            accepted_privileged = {
                k: v for k, v in privileged_fields.items()
                if k in response_data and response_data[k] == v
            }
            passed = len(accepted_privileged) == 0
        else:
            passed = True
 
        result = APISecurityTestResult(
            test_name="Mass Assignment 測試",
            endpoint=endpoint,
            method=method,
            severity="high",
            passed=passed,
            details=(
                f"Attempted to set privileged fields: {list(privileged_fields.keys())}. "
                f"{'PASS: Privileged fields rejected.' if passed else 'FAIL: Privileged fields accepted.'}"
            ),
        )
        self.results.append(result)
        return result
 
    def test_unauthenticated_access(self, endpoints: list[dict]) -> list[APISecurityTestResult]:
        """測試 for endpoints accessible without 認證."""
        results = []
        for ep in endpoints:
            url = f"{self.base_url}{ep['path']}"
            method = ep.get("method", "GET")
 
            try:
                if method == "GET":
                    response = requests.get(url, headers={"Content-Type": "application/json"}, timeout=10)
                elif method == "POST":
                    response = requests.post(url, json={}, headers={"Content-Type": "application/json"}, timeout=10)
                elif method == "DELETE":
                    response = requests.delete(url, headers={"Content-Type": "application/json"}, timeout=10)
                else:
                    continue
 
                # Should get 401, not 200/400/500
                passed = response.status_code == 401
                result = APISecurityTestResult(
                    test_name="Unauthenticated Access 測試",
                    endpoint=ep["path"],
                    method=method,
                    severity="critical" if not passed and ep.get("requires_auth", True) else "info",
                    passed=passed,
                    details=(
                        f"Unauthenticated {method} returned {response.status_code}. "
                        f"{'PASS' if passed else 'FAIL: Endpoint accessible without auth.'}"
                    ),
                )
                results.append(result)
                self.results.append(result)
            except requests.exceptions.RequestException:
                pass
 
        return results
 
    def test_excessive_data_exposure(
        self, endpoint: str, sensitive_fields: list[str]
    ) -> APISecurityTestResult:
        """測試 for sensitive data in API responses."""
        url = f"{self.base_url}{endpoint}"
        response = requests.get(url, headers=self._headers(), timeout=10)
 
        exposed_fields = []
        if response.status_code == 200:
            data = response.json()
            # Check top-level and nested fields
            data_str = str(data).lower()
            for field_name in sensitive_fields:
                if field_name.lower() in data_str:
                    exposed_fields.append(field_name)
 
        passed = len(exposed_fields) == 0
        result = APISecurityTestResult(
            test_name="Excessive Data Exposure 測試",
            endpoint=endpoint,
            method="GET",
            severity="high" if not passed else "info",
            passed=passed,
            details=(
                f"Checked for sensitive fields: {sensitive_fields}. "
                f"{'PASS: No sensitive data exposed.' if passed else f'FAIL: Exposed: {exposed_fields}'}"
            ),
        )
        self.results.append(result)
        return result
 
    def generate_report(self) -> dict:
        """Generate a summary report of all 測試 results."""
        total = len(self.results)
        passed = sum(1 for r in self.results if r.passed)
        failed = total - passed
        critical_failures = sum(
            1 for r in self.results
            if not r.passed and r.severity == "critical"
        )
 
        return {
            "total_tests": total,
            "passed": passed,
            "failed": failed,
            "critical_failures": critical_failures,
            "pass_rate": f"{(passed / total * 100):.1f}%" if total > 0 else "N/A",
            "failures": [
                {
                    "測試": r.test_name,
                    "endpoint": r.endpoint,
                    "severity": r.severity,
                    "details": r.details,
                }
                for r in self.results if not r.passed
            ],
        }

SEMGREP_API_RULES = """
rules:
  - id: flask-endpoint-no-auth
    patterns:
      - pattern: |
          @app.route("...", ...)
          def $FUNC(...):
              ...
      - pattern-not: |
          @app.route("...", ...)
          @require_auth
          def $FUNC(...):
              ...
      - pattern-not: |
          @app.route("...", ...)
          @login_required
          def $FUNC(...):
              ...
    message: >
      Flask endpoint without 認證 decorator. AI-generated
      endpoints commonly omit 認證. Add @require_auth or
      @login_required.
    languages: [python]
    severity: WARNING
    metadata:
      owasp: API2:2023
      category: ai-generated
 
  - id: flask-debug-mode
    pattern: app.run(debug=True, ...)
    message: >
      Flask running in debug mode. AI assistants frequently enable
      debug mode for convenience. Disable in production.
    languages: [python]
    severity: ERROR
    metadata:
      owasp: API8:2023
 
  - id: fastapi-no-response-model
    patterns:
      - pattern: |
          @app.$METHOD("...", ...)
          async def $FUNC(...):
              ...
              return $DATA
      - pattern-not: |
          @app.$METHOD("...", response_model=..., ...)
          async def $FUNC(...):
              ...
    message: >
      FastAPI endpoint without response_model. Without a response model,
      all 資料庫 fields may be exposed. AI tools often omit this.
    languages: [python]
    severity: WARNING
    metadata:
      owasp: API3:2023
"""