diff --git a/gradientql/scanner/actions/__init__.py b/gradientql/scanner/actions/__init__.py index f69cd65..3a08320 100644 --- a/gradientql/scanner/actions/__init__.py +++ b/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). diff --git a/gradientql/scanner/actions/arsenal.py b/gradientql/scanner/actions/arsenal.py index e76c84a..658f34f 100644 --- a/gradientql/scanner/actions/arsenal.py +++ b/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) diff --git a/gradientql/scanner/prompt.py b/gradientql/scanner/prompt.py index d69c884..ff8b864 100644 --- a/gradientql/scanner/prompt.py +++ b/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 diff --git a/gradientql/utils/gqlws.py b/gradientql/utils/gqlws.py index 9cac678..b0fe70e 100644 --- a/gradientql/utils/gqlws.py +++ b/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 diff --git a/gradientql/utils/graphql_sse.py b/gradientql/utils/graphql_sse.py new file mode 100644 index 0000000..2303038 --- /dev/null +++ b/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} diff --git a/tests/test_gqlws.py b/tests/test_gqlws.py index b8d1d53..87361ac 100644 --- a/tests/test_gqlws.py +++ b/tests/test_gqlws.py @@ -98,3 +98,64 @@ def fake_probe(url, headers=None, sub_field=None, timeout=6): res = dispatch("subscribe", ctx, {}) assert any("Auth Bypass" in v["vuln_type"] for v in ctx.vulns) assert "⚠" in res.observation + + +import websocket # noqa: E402 + + +class _IdleWS: + def settimeout(self, t): + pass + + def recv(self): + raise websocket.WebSocketTimeoutException() + + def close(self): + pass + + +class _ClosedWS(_IdleWS): + def recv(self): + raise websocket.WebSocketConnectionClosedException() + + +def test_init_timeout_held_open_is_a_finding(monkeypatch): + _patch_connections(monkeypatch, [_IdleWS()]) + r = gqlws.probe_init_timeout("ws://t/graphql", per_recv=1, max_wait=2) + assert r and "connection_init Timeout" in r[0] + + +def test_init_timeout_enforced_no_finding(monkeypatch): + _patch_connections(monkeypatch, [_ClosedWS()]) + assert gqlws.probe_init_timeout("ws://t/graphql", per_recv=1, max_wait=2) is None + + +def test_subscription_cap_no_limit_records(monkeypatch): + ack = json.dumps({"type": "connection_ack"}) + nxt = json.dumps({"type": "next", "payload": {"data": {"__typename": "Subscription"}}}) + _patch_connections(monkeypatch, [_FakeWS("graphql-transport-ws", [ack, nxt, nxt, nxt])]) + r = gqlws.probe_subscription_cap("http://t/graphql", n=3) + assert any("No Subscription-Count Limiting" in vt for vt, _ in r["findings"]) + + +def test_subscription_cap_capped_no_finding(monkeypatch): + ack = json.dumps({"type": "connection_ack"}) + nxt = json.dumps({"type": "next", "payload": {"data": {}}}) + err = json.dumps({"type": "error"}) + _patch_connections(monkeypatch, [_FakeWS("graphql-transport-ws", [ack, nxt, err])]) + assert gqlws.probe_subscription_cap("http://t/graphql", n=3)["findings"] == [] + + +def test_stale_auth_confirmed(monkeypatch): + ack = json.dumps({"type": "connection_ack"}) + snap = json.dumps({"type": "next", "payload": {"data": {"x": 1}}}) + after = json.dumps({"type": "next", "payload": {"data": {"x": 2}}}) + _patch_connections(monkeypatch, [_FakeWS("graphql-transport-ws", [ack, snap, after])]) + + def execute(q): + if q == "query { me { id } }": # verify -> now blocked + return {"_status_code": 401, "data": None, "errors": [{"message": "unauthorized"}]} + return {"_status_code": 200, "data": {"ok": True}, "errors": []} + finding, note = gqlws.probe_stale_auth("http://t/graphql", {}, execute, + "query { me { id } }", "mutation { logout }", "newMsg") + assert finding and "Stale Auth over Persistent Subscription" in finding[0] diff --git a/tests/test_graphql_sse.py b/tests/test_graphql_sse.py new file mode 100644 index 0000000..6cf17c4 --- /dev/null +++ b/tests/test_graphql_sse.py @@ -0,0 +1,29 @@ +"""GraphQL-over-SSE support + unauthenticated-subscription probe.""" + +from __future__ import annotations + +from gradientql.utils import graphql_sse + + +def test_probe_sse_support_and_unauth_subscription(monkeypatch): + def fake_stream(url, body, session=None, headers=None, timeout=12): + q = body.get("query", "") + if "subscription" in q: + return "text/event-stream", 'event: next\ndata: {"payload":{"data":{"msg":"hi"}}}\n' + return "text/event-stream", 'event: next\ndata: {"data":{"__typename":"Query"}}\n' + monkeypatch.setattr(graphql_sse, "_stream", fake_stream) + sm = {"_subscription_type": "Subscription", "Subscription": {"newMsg": {"return_type": "Msg", "args": []}}} + vts = [vt for vt, _ in graphql_sse.probe_sse("http://t/graphql", sm)["findings"]] + assert any("SSE" in vt for vt in vts) + assert any("Broken Authorization over GraphQL SSE Subscription" in vt for vt in vts) + + +def test_probe_sse_no_support(monkeypatch): + monkeypatch.setattr(graphql_sse, "_stream", lambda *a, **k: ("application/json", '{"data":{}}')) + assert graphql_sse.probe_sse("http://t/graphql", {})["findings"] == [] + + +def test_sse_data_parsing(): + assert graphql_sse._sse_data('data: {"payload":{"data":{"x":1}}}') == {"x": 1} + assert graphql_sse._sse_data('event: complete\ndata: {"data":{}}') is None # empty data + assert graphql_sse._sse_data('not sse') is None diff --git a/tests/test_scanner_actions.py b/tests/test_scanner_actions.py index 1fcc53d..c6f3487 100644 --- a/tests/test_scanner_actions.py +++ b/tests/test_scanner_actions.py @@ -662,7 +662,7 @@ def test_disabled_actions_defaults_and_safe_mode(): # dos defaults OFF with no config (matches the shipped settings.yaml); safe_mode offs the # destructive trio regardless from gradientql.scanner.actions import disabled_actions - assert disabled_actions({}) == {"dos"} + assert disabled_actions({}) == {"dos", "sub_flood"} # sub_flood rides the dos toggle assert disabled_actions({"scanner": {"attacks": {"dos": True}}}) == set() assert {"dos", "smuggle", "batch_brute"} <= disabled_actions( {"scanner": {"safe_mode": True, "attacks": {"dos": True}}})