Реализация фаззинга

Updated Jul 2026

Создание полноценного семантического фаззера

Этот файл содержит полную реализацию семантического фаззера на базе ИИ, которую вы можете адаптировать для тестирования своего API.

Класс SemanticFuzzer

import json
import httpx
import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class FuzzPayload:
    payload: dict
    category: str
    expected_vulnerability: str

@dataclass
class FuzzResult:
    payload: FuzzPayload
    response_status: int
    response_body: str
    response_time_ms: float
    anomaly: Optional[str] = None
    severity: Optional[str] = None

@dataclass
class CampaignReport:
    campaign_id: str
    endpoint: str
    method: str
    total_payloads: int
    anomalies_found: int
    findings: list[FuzzResult]
    duration_seconds: float
    token_cost: float = 0.0


class SemanticFuzzer:
    def __init__(self, llm, schema: dict, base_url: str, auth_headers: dict = None):
        self.llm = llm
        self.schema = schema
        self.base_url = base_url
        self.auth_headers = auth_headers or {}

    def generate_fuzz_payloads(self, endpoint: str, count: int = 50) -> list[FuzzPayload]:
        """Generate semantically meaningful fuzz payloads."""
        endpoint_schema = self.schema["paths"].get(endpoint, {})

        prompt = f"""
        Given this API endpoint schema:
        {json.dumps(endpoint_schema, indent=2)}

        Generate {count} test payloads designed to find bugs. Categories:

        1. SQL INJECTION (5 payloads)
           - In string fields, include SQL fragments
        2. XSS (5 payloads)
           - In string fields, include HTML/JS
        3. BOUNDARY VALUES (10 payloads)
           - Min/max for numbers, empty/max-length strings
        4. TYPE CONFUSION (10 payloads)
           - String where number expected, array where object expected
        5. UNICODE EDGE CASES (5 payloads)
           - RTL characters, zero-width joiners, emoji, null bytes
        6. BUSINESS LOGIC (10 payloads)
           - Valid structure but nonsensical values (negative quantity,
             price of $0.001, dates in the past)
        7. AUTHORIZATION BYPASS (5 payloads)
           - Payloads that try to escalate privileges or access
             other users' data (IDOR patterns)

        Output as a JSON array of objects. Each object has:
        - "payload": the request body
        - "category": which category above
        - "expected_vulnerability": what bug this might find
        """
        raw = self.llm.generate_json(prompt)
        return [FuzzPayload(**item) for item in raw]

    def execute_fuzz_campaign(
        self, endpoint: str, method: str = "POST", timeout: float = 10.0
    ) -> CampaignReport:
        """Run a complete fuzz campaign against an endpoint."""
        start_time = time.time()
        payloads = self.generate_fuzz_payloads(endpoint)
        results = []

        for payload_info in payloads:
            try:
                req_start = time.time()
                response = httpx.request(
                    method,
                    f"{self.base_url}{endpoint}",
                    json=payload_info.payload,
                    headers=self.auth_headers,
                    timeout=timeout,
                )
                response_time = (time.time() - req_start) * 1000

                anomaly = self.detect_anomaly(response, payload_info)
                severity = self.classify_severity(anomaly, payload_info) if anomaly else None

                result = FuzzResult(
                    payload=payload_info,
                    response_status=response.status_code,
                    response_body=response.text[:500],
                    response_time_ms=response_time,
                    anomaly=anomaly,
                    severity=severity,
                )
                results.append(result)

            except httpx.TimeoutException:
                results.append(FuzzResult(
                    payload=payload_info,
                    response_status=0,
                    response_body="TIMEOUT",
                    response_time_ms=timeout * 1000,
                    anomaly=f"Request timeout on {payload_info.category} payload",
                    severity="MEDIUM",
                ))
            except Exception as e:
                results.append(FuzzResult(
                    payload=payload_info,
                    response_status=0,
                    response_body=str(e),
                    response_time_ms=0,
                    anomaly=f"Request exception: {type(e).__name__}: {e}",
                    severity="HIGH",
                ))

        anomaly_results = [r for r in results if r.anomaly]

        return CampaignReport(
            campaign_id=f"fuzz-{int(time.time())}",
            endpoint=endpoint,
            method=method,
            total_payloads=len(payloads),
            anomalies_found=len(anomaly_results),
            findings=anomaly_results,
            duration_seconds=time.time() - start_time,
        )

    def detect_anomaly(self, response, payload_info: FuzzPayload) -> Optional[str]:
        """Detect if the response indicates a potential vulnerability."""
        # 500 errors on any input = server-side failure
        if response.status_code >= 500:
            return f"Server error ({response.status_code}) on {payload_info.category} payload"

        # XSS reflection
        if payload_info.category == "XSS":
            dangerous_patterns = ["<script>", "onerror=", "javascript:", "onclick="]
            if any(p in response.text for p in dangerous_patterns):
                return "XSS reflection detected in response"

        # SQL error disclosure
        sql_indicators = ["syntax error", "mysql", "postgresql", "sqlite",
                         "ORA-", "SQLSTATE", "unterminated"]
        if any(ind in response.text.lower() for ind in sql_indicators):
            return "SQL error message disclosed in response"

        # Stack trace leak
        if "Traceback" in response.text or "at Object." in response.text:
            return "Stack trace leaked in response body"

        # Timeout on specific payload (potential DoS)
        if hasattr(response, 'elapsed') and response.elapsed.total_seconds() > 5:
            return f"Slow response ({response.elapsed.total_seconds():.1f}s) on {payload_info.category} payload"

        return None

    def classify_severity(self, anomaly: str, payload: FuzzPayload) -> str:
        """Classify the severity of a detected anomaly."""
        if "XSS reflection" in anomaly:
            return "HIGH"
        if "SQL error" in anomaly:
            return "HIGH"
        if "Stack trace" in anomaly:
            return "MEDIUM"
        if "Server error" in anomaly:
            if payload.category in ["SQL_INJECTION", "AUTHORIZATION_BYPASS"]:
                return "HIGH"
            return "MEDIUM"
        if "Slow response" in anomaly:
            return "LOW"
        if "Timeout" in anomaly:
            return "MEDIUM"
        return "LOW"

Запуск фаззинга в CI

# .github/workflows/api-fuzz.yml
name: API Fuzzing

on:
  schedule:
    - cron: '0 2 * * *'  # Nightly at 2 AM

jobs:
  fuzz:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Start staging environment
        run: docker-compose up -d

      - name: Wait for API readiness
        run: |
          for i in $(seq 1 30); do
            curl -s http://localhost:8080/health && break
            sleep 2
          done

      - name: Run fuzz campaign
        run: |
          python -m api_fuzzer \
            --schema docs/openapi.yaml \
            --base-url http://localhost:8080 \
            --output fuzz-report.json \
            --endpoints "/api/v2/products,/api/v2/orders"

      - name: Check for critical findings
        run: |
          CRITICAL=$(jq '[.findings[] | select(.severity=="HIGH" or .severity=="CRITICAL")] | length' fuzz-report.json)
          if [ "$CRITICAL" -gt 0 ]; then
            echo "::error::Found $CRITICAL high/critical findings"
            jq '.findings[] | select(.severity=="HIGH" or .severity=="CRITICAL")' fuzz-report.json
            exit 1
          fi

      - name: Upload fuzz report
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: fuzz-report
          path: fuzz-report.json

Планирование фаззинг-кампаний

Интерпретация результатов

Снижение ложных срабатываний

Не каждая аномалия -- реальная уязвимость. Типичные ложные срабатывания:

Процесс триажа

1. Sort findings by severity (CRITICAL → HIGH → MEDIUM → LOW)
2. For each HIGH/CRITICAL:
   - Reproduce manually with curl
   - Verify it is not a false positive
   - File a security bug if confirmed
3. For each MEDIUM:
   - Reproduce, assess business impact
   - File as bug or tech debt based on impact
4. For each LOW:
   - Log for awareness
   - Fix opportunistically (do not block releases)

Ключевой вывод

Полноценный семантический фаззер сочетает сгенерированные ИИ данные с автоматическим обнаружением аномалий и классификацией критичности. Запускайте его ночью в CI на staging-окружении, сортируйте находки по критичности и используйте результаты для выявления уязвимостей безопасности, которые традиционное тестирование пропускает. Реализация проста -- ценность заключается в способности ИИ генерировать контекстно релевантные атакующие данные.