Add WS/SSE transport engines: SSE, init-timeout, stale-auth, subscrip…

gradientql/scanner/actions/__init__.py

@@ -24,7 +24,7 @@ def _register(fn: Handler) -> Handler:
 _ATTACK_TOGGLE = {
     "dos": "dos", "smuggle": "smuggle", "csrf": "csrf",
     "forge_jwt": "jwt", "batch_brute": "brute", "auth_test": "bola",
-    "evade": "evade",
+    "evade": "evade", "sub_flood": "dos",
 }
 _SAFE_MODE_OFF = ("dos", "smuggle", "brute", "evade")
 # Toggles default ON; dos defaults OFF (matches the shipped config: it can knock a target over).

gradientql/scanner/actions/arsenal.py

@@ -400,6 +400,49 @@ def _first_subscription_field(schema_map: dict) -> str | None:
     return None
 
 
+@action("sse")
+def handle_sse(ctx: ActionContext, args: dict) -> Result:
+    """GraphQL-over-SSE (Server-Sent Events): detect SSE support and unauthenticated data over an SSE
+    subscription (pure HTTP streaming, no WebSocket engine). No args."""
+    from ...utils.graphql_sse import probe_sse
+    try:
+        result = probe_sse(ctx.target_url, ctx.schema_map, session=getattr(ctx.client, "session", None),
+                           headers=dict(ctx.identity or {}))
+    except Exception as e:  # noqa: BLE001
+        obs = f"sse probe failed: {str(e)[:120]}"
+        ctx.log(f"[{ctx.step}] sse -> {obs}")
+        return Result(observation=obs, touched_target=True)
+    for vt, evidence in result["findings"]:
+        ctx.record(vt, "endpoint", evidence, 3.0 if "Broken Authorization" in vt else 2.0)
+    ctx.interactions.append({"target_node": "sse", "reason": "agent_sse", "response_status": 0,
+                             "score": 0.0, "timestamp": datetime.now(timezone.utc).isoformat()})
+    head = ("⚠ " + "; ".join(vt for vt, _ in result["findings"]) + "\n  ") if result["findings"] else ""
+    obs = (head + " | ".join(result["observations"])) or "no SSE signals"
+    ctx.log(f"[{ctx.step}] sse -> {obs[:200]}")
+    return Result(observation=obs[:700], touched_target=True)
+
+
+@action("sub_flood")
+def handle_sub_flood(ctx: ActionContext, args: dict) -> Result:
+    """Open ~40 subscriptions on one WS connection to check for a per-connection subscription cap (a
+    resource-exhaustion surface). DoS-flavored - gated by the dos toggle. No args."""
+    from ...utils.gqlws import probe_subscription_cap
+    try:
+        result = probe_subscription_cap(ctx.target_url, headers=dict(ctx.identity or {}))
+    except Exception as e:  # noqa: BLE001
+        obs = f"sub_flood probe failed: {str(e)[:120]}"
+        ctx.log(f"[{ctx.step}] sub_flood -> {obs}")
+        return Result(observation=obs, touched_target=True)
+    for vt, evidence in result["findings"]:
+        ctx.record(vt, "endpoint", evidence, 2.5)
+    ctx.interactions.append({"target_node": "sub_flood", "reason": "agent_sub_flood", "response_status": 0,
+                             "score": 0.0, "timestamp": datetime.now(timezone.utc).isoformat()})
+    head = ("⚠ " + "; ".join(vt for vt, _ in result["findings"]) + "\n  ") if result["findings"] else ""
+    obs = (head + " | ".join(result["observations"])) or "no subscription-cap signal"
+    ctx.log(f"[{ctx.step}] sub_flood -> {obs[:200]}")
+    return Result(observation=obs[:700], touched_target=True)
+
+
 @action("subscribe")
 def handle_subscribe(ctx: ActionContext, args: dict) -> Result:
     """Probe GraphQL subscriptions over WebSocket - the transport the HTTP loop can't drive.
@@ -418,12 +461,21 @@ def handle_subscribe(ctx: ActionContext, args: dict) -> Result:
         return Result(observation=obs, touched_target=True)
     for vt, evidence in result.get("findings", []):
         ctx.record(vt, "subscription", evidence, 3.0)
+    vq, rq = str(args.get("verify_query", "")).strip(), str(args.get("revoke_query", "")).strip()
+    stale_note = ""
+    if vq and rq:  # model-driven stale-auth check: revoke over HTTP, then re-poll the open WS
+        from ...utils.gqlws import probe_stale_auth
+        finding, stale_note = probe_stale_auth(ctx.target_url, dict(ctx.identity or {}),
+                                               ctx.client.execute, vq, rq, field)
+        if finding:
+            ctx.record(finding[0], "subscription", finding[1], 3.5)
     ctx.interactions.append({"target_node": "subscribe", "reason": "agent_subscribe",
                              "response_status": 0, "score": 0.0,
                              "timestamp": datetime.now(timezone.utc).isoformat()})
     head = ("⚠ " + "; ".join(vt for vt, _ in result["findings"]) + "\n  "
             if result.get("findings") else "")
-    obs = head + " | ".join(result.get("observations", [])) or "no WS observations"
+    obs = head + " | ".join(result.get("observations", []) + ([stale_note] if stale_note else [])) \
+        or "no WS observations"
     ctx.log(f"[{ctx.step}] subscribe -> {obs[:200]}")
     return Result(observation=obs[:700], touched_target=True)
 

gradientql/scanner/prompt.py

@@ -262,11 +262,17 @@
   redeem/withdraw/claim/apply mutation. Reports how many raced copies succeeded with no serializing
   error - if a SINGLE-USE op succeeds >1×, that's a limit-overrun/double-spend race (verify the effect
   before report_finding). Racing a plain read or idempotent op proves nothing.""",
-    "subscribe": """- subscribe: args {field?} - probe GraphQL SUBSCRIPTIONS over WebSocket (the one
-  transport regular graphql calls can't reach). Auto-tests: legacy graphql-ws subprotocol DOWNGRADE,
-  PRE-HANDSHAKE auth bypass (a subscribe accepted before connection_init), and UNAUTHENTICATED data
-  over a subscription (subscription resolvers often skip the auth queries/mutations enforce). Use it
-  whenever KNOWN says a subscription root exists.""",
+    "subscribe": """- subscribe: args {field?, verify_query?, revoke_query?} - probe GraphQL SUBSCRIPTIONS
+  over WebSocket (the one transport regular graphql calls can't reach). Auto-tests: legacy graphql-ws
+  subprotocol DOWNGRADE, PRE-HANDSHAKE auth bypass (a subscribe accepted before connection_init),
+  UNAUTHENTICATED data over a subscription, and missing connection_init TIMEOUT. Pass verify_query (a
+  query that returns your data) + revoke_query (a logout/revoke mutation) to test STALE-AUTH: HTTP
+  honours revocation but the open subscription keeps delivering. Use whenever KNOWN says a subscription
+  root exists.""",
+    "sse": """- sse: no args - GraphQL-over-SSE (Server-Sent Events) probe (pure HTTP streaming, no WS). Detects
+  SSE incremental-delivery support and unauthenticated data over an SSE subscription.""",
+    "sub_flood": """- sub_flood: no args - open ~40 subscriptions on one WS connection to check for a missing
+  per-connection subscription cap (resource-exhaustion surface). DoS-flavored, gated by the dos toggle.""",
     "defer": """- defer: args {query?} - probe @defer/@stream incremental delivery. Detects whether the
   server streams a multipart/mixed response (auto-builds a probe query if you don't pass one). If
   SUPPORTED it's a DoS-amplification (many `... @defer` fragments -> many chunks), response-desync, and

gradientql/utils/gqlws.py

@@ -131,4 +131,140 @@ def probe_subscriptions(url: str, headers: dict | None = None, sub_field: str |
     except Exception as e:  # noqa: BLE001
         obs.append(f"init probe failed ({str(e)[:50]})")
 
+    # 4. connection_init timeout: a server should close an un-initialised socket (4408); holding it
+    #    open leaves the pre-handshake auth gate holdable and lets idle sockets accumulate.
+    it = probe_init_timeout(ws_url, headers, per_recv=min(timeout, 3))
+    if it:
+        findings.append(it)
+        obs.append("no connection_init timeout - un-initialised socket held open")
+
     return {"available": True, "observations": obs, "findings": findings}
+
+
+def probe_init_timeout(ws_url: str, headers: dict | None = None, per_recv: int = 3,
+                       max_wait: int = 12) -> tuple[str, str] | None:
+    """Open a WS, send nothing, and watch for a server-side close (4408). Returns a finding tuple if the
+    socket is still open after `max_wait`s, else None (the timeout is enforced)."""
+    if not _WS_AVAILABLE:
+        return None
+    try:
+        ws = _connect(ws_url, _PROTO_MODERN, headers, per_recv)
+    except Exception:  # noqa: BLE001
+        return None
+    held = True
+    try:
+        ws.settimeout(per_recv)
+        for _ in range(max(1, int(max_wait // per_recv))):
+            try:
+                if ws.recv() in ("", None):
+                    held = False
+                    break
+            except websocket.WebSocketTimeoutException:
+                continue                                  # idle: socket still open, keep waiting
+            except Exception:  # noqa: BLE001
+                held = False                              # closed / connection reset
+                break
+    finally:
+        try:
+            ws.close()
+        except Exception:  # noqa: BLE001
+            pass
+    if held:
+        return ("No connection_init Timeout (pre-handshake socket held open)",
+                f"the WS server did not close an un-initialised socket within {max_wait}s (no 4408 close) - "
+                "the pre-handshake auth gate is holdable and idle un-authed sockets accumulate.")
+    return None
+
+
+def probe_subscription_cap(url: str, headers: dict | None = None, timeout: int = 6,
+                           n: int = 40) -> dict[str, Any]:
+    """Open N subscriptions on one connection (hard-capped, not a flood). No per-connection cap = a
+    resource-exhaustion surface. Returns {findings, observations}."""
+    findings: list[tuple[str, str]] = []
+    obs: list[str] = []
+    if not _WS_AVAILABLE:
+        return {"findings": findings, "observations": ["websocket-client not installed"]}
+    try:
+        ws = _connect(_ws_url(url), _PROTO_MODERN, headers, timeout)
+        ws.send(json.dumps({"type": "connection_init", "payload": {}}))
+        _recv_json(ws)
+    except Exception as e:  # noqa: BLE001
+        return {"findings": findings, "observations": [f"subscription-cap: connect failed ({str(e)[:50]})"]}
+    accepted = 0
+    try:
+        for i in range(n):
+            ws.send(json.dumps({"id": str(i), "type": "subscribe",
+                                "payload": {"query": "subscription { __typename }"}}))
+            m = _recv_json(ws)
+            if m is None or m.get("type") in ("error", "connection_error"):
+                break                                     # a cap / limit kicked in
+            accepted += 1
+    finally:
+        try:
+            ws.close()
+        except Exception:  # noqa: BLE001
+            pass
+    if accepted >= n:
+        findings.append(("No Subscription-Count Limiting (resource exhaustion surface)",
+                         f"the server accepted {n} concurrent subscriptions on a single connection with no "
+                         "per-connection cap or error - unbounded subscriptions can exhaust server resources."))
+        obs.append(f"subscription cap: {n} accepted, no limit")
+    else:
+        obs.append(f"subscription cap: server stopped/capped at {accepted} subscriptions")
+    return {"findings": findings, "observations": obs}
+
+
+def probe_stale_auth(url: str, headers: dict | None, execute: Any, verify_query: str,
+                     revoke_query: str, sub_field: str, timeout: int = 6) -> tuple[Any, str]:
+    """Record iff HTTP honours token revocation but a subscription on the still-open WS keeps delivering.
+
+    `execute(query)` runs an HTTP GraphQL request under the current identity. Model-driven: the caller
+    supplies the revoke (logout) mutation and a verify query that returns data pre-revoke. Returns
+    (finding_tuple_or_None, note); non-recording and honest about inconclusive preconditions.
+    """
+    if not _WS_AVAILABLE or not (verify_query and revoke_query and sub_field):
+        return None, "stale-auth: needs {verify_query, revoke_query} and a subscription field"
+    try:
+        ws = _connect(_ws_url(url), _PROTO_MODERN, headers, timeout)
+        ws.send(json.dumps({"type": "connection_init", "payload": dict(headers or {})}))
+        _recv_json(ws)
+        ws.send(_start_msg(_PROTO_MODERN, sub_field))
+        snap = _recv_json(ws)
+    except Exception as e:  # noqa: BLE001
+        return None, f"stale-auth: WS subscribe failed ({str(e)[:50]})"
+    if not (snap and snap.get("type") in ("next", "data") and (snap.get("payload") or {}).get("data")):
+        _safe_close(ws)
+        return None, "stale-auth: no immediate subscription snapshot (inconclusive, not a miss)"
+    try:
+        execute(revoke_query)
+        after = execute(verify_query) or {}
+    except Exception:  # noqa: BLE001
+        _safe_close(ws)
+        return None, "stale-auth: revoke/verify HTTP call failed"
+    errs = after.get("errors") or []
+    msgs = " ".join(str(e.get("message", "")).lower() for e in errs)
+    blocked = (after.get("_status_code") in (401, 403)
+               or any(m in msgs for m in ("not authori", "unauthori", "forbidden", "expired", "invalid token"))
+               or (after.get("data") is None and bool(errs)))
+    if not blocked:
+        _safe_close(ws)
+        return None, "stale-auth: revocation did not kill the token on HTTP (stateless JWT?) - inconclusive"
+    try:
+        ws.send(json.dumps({"id": "2", "type": "subscribe",
+                            "payload": {"query": f"subscription {{ {sub_field} }}"}}))
+        after_frame = _recv_json(ws)
+    finally:
+        _safe_close(ws)
+    if after_frame and after_frame.get("type") in ("next", "data") and (after_frame.get("payload") or {}).get("data"):
+        return (("Stale Auth over Persistent Subscription",
+                 f"HTTP honoured the session revocation but the already-open WS subscription to `{sub_field}` "
+                 "kept delivering data - the persistent connection is never re-checked against revocation."),
+                "stale-auth CONFIRMED")
+    return None, "stale-auth: post-revoke subscription stopped (revocation honoured on the WS too)"
+
+
+def _safe_close(ws: Any) -> None:
+    try:
+        ws.close()
+    except Exception:  # noqa: BLE001
+        pass

gradientql/utils/graphql_sse.py

@@ -0,0 +1,87 @@
+"""GraphQL-over-SSE (Server-Sent Events) probe: SSE support and unauthenticated data over a
+subscription, using pure `requests` streaming - no websocket engine, no browser.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+_SSE_MARKERS = ("event: next", "event:next", "event: complete", "data: {", "data:{")
+
+
+def first_subscription_field(schema_map: dict[str, Any]) -> str | None:
+    st = schema_map.get("_subscription_type")
+    fields = schema_map.get(st) if st else None
+    if isinstance(fields, dict):
+        return next((f for f in fields if not str(f).startswith("_")), None)
+    return None
+
+
+def _stream(url: str, body: dict, session: Any = None, headers: dict | None = None,
+            timeout: int = 12) -> tuple[str, str]:
+    """POST with an event-stream Accept header and read a bounded chunk of the streamed body."""
+    import requests
+    h = {"Accept": "text/event-stream", "Content-Type": "application/json"}
+    h.update(headers or {})
+    http = session or requests
+    try:
+        r = http.post(url, json=body, headers=h, stream=True, timeout=timeout)
+        ct = (r.headers.get("Content-Type", "") or "").lower()
+        buf = ""
+        for line in r.iter_lines(decode_unicode=True):
+            buf += (line or "") + "\n"
+            if len(buf) > 4000:
+                break
+        r.close()
+        return ct, buf
+    except requests.RequestException:
+        return "", ""
+
+
+def _sse_data(body: str) -> Any:
+    """Return the first non-empty JSON `data.*` payload carried on an SSE `data:` line, else None."""
+    for line in body.splitlines():
+        line = line.strip()
+        if not line.lower().startswith("data:"):
+            continue
+        try:
+            j = json.loads(line[5:].strip())
+        except Exception:  # noqa: BLE001
+            continue
+        payload = j.get("payload") if isinstance(j, dict) and "payload" in j else j
+        data = payload.get("data") if isinstance(payload, dict) else None
+        if isinstance(data, dict) and any(v not in (None, {}, []) for v in data.values()):
+            return data
+    return None
+
+
+def probe_sse(url: str, schema_map: dict[str, Any], session: Any = None,
+              headers: dict | None = None) -> dict[str, Any]:
+    """Detect SSE support (low-sev surface) and unauthenticated data over an SSE subscription."""
+    findings: list[tuple[str, str]] = []
+    obs: list[str] = []
+    ct, body = _stream(url, {"query": "{ __typename }"}, session, headers)
+    supported = "text/event-stream" in ct or any(m in body.lower() for m in _SSE_MARKERS)
+    if not supported:
+        obs.append("SSE: server did not stream text/event-stream for an event-stream Accept")
+        return {"findings": findings, "observations": obs}
+    findings.append(("GraphQL-over-SSE Incremental Delivery Enabled",
+                     f"the endpoint streams Server-Sent Events (content-type {ct[:40]}) - a subscription/"
+                     "incremental-delivery transport that bypasses the WebSocket handshake and its auth gate."))
+    obs.append(f"SSE supported (ct={ct[:40]})")
+    sub = first_subscription_field(schema_map)
+    if sub:
+        import requests
+        _ct, body2 = _stream(url, {"query": f"subscription {{ {sub} }}"},
+                             session=requests.Session(), headers={})   # fresh session, no auth
+        data = _sse_data(body2)
+        if data:
+            findings.append(("Broken Authorization over GraphQL SSE Subscription",
+                             f"an unauthenticated SSE subscription to `{sub}` streamed data "
+                             f"({json.dumps(data)[:160]}) - the subscription resolver skips the auth applied "
+                             "to queries/mutations."))
+            obs.append(f"unauth SSE subscription `{sub}` -> DATA (BROKEN AUTHZ)")
+        else:
+            obs.append(f"unauth SSE subscription `{sub}` -> no immediate data (event-driven)")
+    return {"findings": findings, "observations": obs}