6.2The SemanticFuzzer Class

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"