{"uuid": "24211bf0-f2d4-40cd-ad68-805ce3aff96f", "vulnerability_lookup_origin": "1a89b78e-f703-45f3-bb86-59eb712668bd", "author": "9f56dd64-161d-43a6-b9c3-555944290a09", "vulnerability": "cve-2026-18963", "type": "seen", "source": "https://gist.github.com/asifnawazminhas/ef499fdd55a5bf49b5056c3e3a7eddbb", "content": "#!/usr/bin/env python3\n\"\"\"Hunt CVE-2026-18963-style password resets via the Keycloak Admin Events API.\n\nCVE-2026-18963 is a reset-credentials flow bypass in Keycloak 26.0.0 through\n26.7.1 (fixed in 26.7.2, 26.4.15, and 26.6.6). An attacker who knows a username\ncan reach UPDATE_PASSWORD without clicking the email action-token link.\nSee https://access.redhat.com/security/cve/cve-2026-18963 , issue #51833,\nand fix PR #51844.\n\nFingerprint: SEND_RESET_PASSWORD or SEND_RESET_PASSWORD_ERROR, then\nUPDATE_PASSWORD or UPDATE_CREDENTIAL (password) on the same code_id, with\ndelta <= --threshold-ms (default 5000). EXECUTE_ACTION_TOKEN is not a reliable\nsuccess signal for this path. This is a heuristic, not proof of exploit: a\nlegitimate reset completed very quickly can also match.\n\nRequires Direct Access Grants on the OIDC client (admin-cli by default) and\nrealm-management/view-events. If listing realms is forbidden, only\n--auth-realm is scanned.\n\nExamples:\n  python3 cve-2026-18963-hunt.py --base https://keycloak.example --admin alice --since 7d --skip-master\n  python3 cve-2026-18963-hunt.py --base https://keycloak.example --auth-realm myrealm --admin alice --realm myrealm --since 24h\n\nEnvironment: KC_BASE, KC_AUTH_REALM, KC_CLIENT, KC_CLIENT_SECRET, KC_ADMIN,\nKC_ADMIN_PASS, KC_TOTP, KC_SINCE, THRESHOLD_MS.\n\nExit codes: 0 no pair under threshold, 1 at least one SUSPECT, 2 login failure.\n\"\"\"\nfrom __future__ import annotations\n\nimport argparse\nimport datetime as dt\nimport getpass\nimport json\nimport os\nimport sys\nimport time\nimport urllib.error\nimport urllib.parse\nimport urllib.request\nfrom typing import Any, Dict, Iterable, List, Optional, Tuple\n\nSEND_TYPES = (\"SEND_RESET_PASSWORD\", \"SEND_RESET_PASSWORD_ERROR\")\nUPDATE_TYPES = (\"UPDATE_PASSWORD\", \"UPDATE_CREDENTIAL\")\nPAGE = 100\n\n\nCOLOR_LOG = sys.stderr.isatty()\n\n\ndef log(msg: str) -> None:\n    ts = dt.datetime.now().strftime(\"%H:%M:%S\")\n    prefix = f\"[{ts}]\"\n    if COLOR_LOG:\n        prefix = f\"\\033[2m{prefix}\\033[0m\"\n    sys.stderr.write(f\"{prefix} {msg}\\n\")\n    sys.stderr.flush()\n\n\ndef parse_since(value: str) -> int:\n    \"\"\"Return epoch milliseconds. Accepts 90m / 24h / 7d / ISO-8601 / epoch-ms.\"\"\"\n    raw = value.strip()\n    now = time.time()\n    if raw.isdigit():\n        n = int(raw)\n        return n if n > 10_000_000_000 else int(n * 1000)\n    unit = raw[-1].lower()\n    if unit in \"smhd\" and raw[:-1].replace(\".\", \"\", 1).isdigit():\n        qty = float(raw[:-1])\n        mult = {\"s\": 1, \"m\": 60, \"h\": 3600, \"d\": 86400}[unit]\n        return int((now - qty * mult) * 1000)\n    iso = raw.replace(\"Z\", \"+00:00\")\n    parsed = dt.datetime.fromisoformat(iso)\n    if parsed.tzinfo is None:\n        parsed = parsed.replace(tzinfo=dt.timezone.utc)\n    return int(parsed.timestamp() * 1000)\n\n\ndef iso_utc(ms: int) -> str:\n    return dt.datetime.fromtimestamp(ms / 1000, dt.timezone.utc).strftime(\"%Y-%m-%d %H:%M:%S UTC\")\n\n\ndef http_json(\n    method: str,\n    url: str,\n    *,\n    token: Optional[str] = None,\n    form: Optional[Dict[str, str]] = None,\n    timeout: int = 60,\n) -> Tuple[int, Any]:\n    data = None\n    headers = {\"Accept\": \"application/json\"}\n    if form is not None:\n        data = urllib.parse.urlencode(form).encode()\n        headers[\"Content-Type\"] = \"application/x-www-form-urlencoded\"\n    if token:\n        headers[\"Authorization\"] = f\"Bearer {token}\"\n    req = urllib.request.Request(url, data=data, headers=headers, method=method)\n    try:\n        with urllib.request.urlopen(req, timeout=timeout) as resp:\n            body = resp.read().decode(\"utf-8\", errors=\"replace\")\n            return resp.status, json.loads(body) if body else None\n    except urllib.error.HTTPError as e:\n        body = e.read().decode(\"utf-8\", errors=\"replace\")\n        try:\n            parsed = json.loads(body) if body else {}\n        except json.JSONDecodeError:\n            parsed = {\"error\": body}\n        return e.code, parsed\n\n\ndef login(\n    base: str,\n    auth_realm: str,\n    user: str,\n    password: str,\n    totp: Optional[str],\n    client_id: str,\n    client_secret: Optional[str],\n) -> str:\n    form = {\n        \"client_id\": client_id,\n        \"username\": user,\n        \"password\": password,\n        \"grant_type\": \"password\",\n    }\n    if totp:\n        form[\"totp\"] = totp\n    if client_secret:\n        form[\"client_secret\"] = client_secret\n    url = f\"{base}/realms/{urllib.parse.quote(auth_realm)}/protocol/openid-connect/token\"\n    status, body = http_json(\"POST\", url, form=form)\n    if status == 200 and isinstance(body, dict) and body.get(\"access_token\"):\n        return str(body[\"access_token\"])\n    desc = \"\"\n    if isinstance(body, dict):\n        desc = str(body.get(\"error_description\") or body.get(\"error\") or body)\n    raise RuntimeError(f\"Admin login failed on realm {auth_realm} (HTTP {status}): {desc}\")\n\n\ndef needs_otp(exc: RuntimeError) -> bool:\n    msg = str(exc).lower()\n    return any(k in msg for k in (\"otp\", \"totp\", \"mfa\", \"invalid user credentials\", \"invalid_grant\"))\n\n\ndef fetch_events(base: str, token: str, realm: str, event_type: str, date_from_ms: Optional[int], cap: int) -> List[dict]:\n    out: List[dict] = []\n    first = 0\n    while first < cap:\n        q = {\n            \"type\": event_type,\n            \"first\": str(first),\n            \"max\": str(min(PAGE, cap - first)),\n        }\n        if date_from_ms is not None:\n            q[\"dateFrom\"] = str(date_from_ms)\n        url = f\"{base}/admin/realms/{urllib.parse.quote(realm)}/events?{urllib.parse.urlencode(q)}\"\n        status, body = http_json(\"GET\", url, token=token)\n        if status == 403:\n            log(f\"  skip {realm}/{event_type}: HTTP 403 (no view-events?)\")\n            return []\n        if status != 200 or not isinstance(body, list):\n            log(f\"  skip {realm}/{event_type}: HTTP {status} {body}\")\n            return []\n        out.extend(body)\n        if len(body) < PAGE:\n            break\n        first += PAGE\n    return out\n\n\ndef is_password_update(event: dict) -> bool:\n    if event.get(\"type\") == \"UPDATE_PASSWORD\":\n        return True\n    details = event.get(\"details\") or {}\n    return details.get(\"credential_type\") == \"password\" or details.get(\"custom_required_action\") == \"UPDATE_PASSWORD\"\n\n\ndef code_id(event: dict) -> str:\n    return str((event.get(\"details\") or {}).get(\"code_id\") or \"\")\n\n\ndef username(event: dict) -> str:\n    return str((event.get(\"details\") or {}).get(\"username\") or event.get(\"userId\") or \"?\")\n\n\ndef ts(event: dict) -> int:\n    return int(event.get(\"time\") or 0)\n\n\ndef pairs_for_realm(sends: Iterable[dict], updates: Iterable[dict], threshold_ms: int) -> List[dict]:\n    by_code: Dict[str, List[dict]] = {}\n    for e in sends:\n        cid = code_id(e)\n        if cid:\n            by_code.setdefault(cid, []).append(e)\n\n    rows = []\n    seen = set()\n    for u in sorted(updates, key=ts):\n        cid = code_id(u)\n        if not cid or cid in seen:\n            continue\n        cands = [s for s in by_code.get(cid, []) if ts(s) <= ts(u)]\n        if not cands:\n            continue\n        send = max(cands, key=ts)\n        delta = ts(u) - ts(send)\n        seen.add(cid)\n        rows.append(\n            {\n                \"delta_ms\": delta,\n                \"suspect\": delta <= threshold_ms,\n                \"username\": username(u),\n                \"send\": send.get(\"type\"),\n                \"update_utc\": iso_utc(ts(u)),\n                \"code_id\": cid,\n                \"user_id\": u.get(\"userId\") or \"\",\n            }\n        )\n    return rows\n\n\ndef term_width() -> int:\n    try:\n        return max(80, os.get_terminal_size().columns)\n    except OSError:\n        return 120\n\n\ndef paint(text: str, color: str, enabled: bool) -> str:\n    if not enabled:\n        return text\n    codes = {\"red\": \"31\", \"green\": \"32\", \"yellow\": \"33\", \"bold\": \"1\", \"dim\": \"2\"}\n    return f\"\\033[{codes[color]}m{text}\\033[0m\"\n\n\ndef print_table(rows: List[dict], color: bool) -> None:\n    if not rows:\n        print(\"  (no send->update pair on the same code_id)\")\n        return\n    headers = [\"delta_ms\", \"verdict\", \"username\", \"send\", \"update (UTC)\", \"code_id\"]\n    cells = []\n    for r in rows:\n        verdict = \"SUSPECT\" if r[\"suspect\"] else \"ok\"\n        cells.append(\n            [\n                str(r[\"delta_ms\"]),\n                verdict,\n                r[\"username\"],\n                str(r[\"send\"]),\n                r[\"update_utc\"],\n                r[\"code_id\"],\n            ]\n        )\n    widths = [len(h) for h in headers]\n    for row in cells:\n        for i, cell in enumerate(row):\n            widths[i] = max(widths[i], len(cell))\n    budget = term_width() - 4 - 3 * (len(headers) - 1)\n    extra = sum(widths) - budget\n    if extra > 0:\n        widths[-1] = max(12, widths[-1] - extra)\n\n    def fmt_row(cols: List[str], *, header: bool = False, suspect: bool = False) -> str:\n        padded = []\n        for i, col in enumerate(cols):\n            val = col if len(col) <= widths[i] else col[: max(0, widths[i] - 3)] + \"...\"\n            padded.append(val.ljust(widths[i]) if i else val.rjust(widths[i]))\n        line = \"  \" + \"   \".join(padded)\n        if header:\n            return paint(line, \"bold\", color)\n        if suspect:\n            return paint(line, \"red\", color)\n        return line\n\n    print(fmt_row(headers, header=True))\n    print(\"  \" + paint(\"-\" * min(term_width() - 2, sum(widths) + 3 * (len(headers) - 1)), \"dim\", color))\n    for r, cols in zip(rows, cells):\n        print(fmt_row(cols, suspect=r[\"suspect\"]))\n\n\ndef main() -> int:\n    p = argparse.ArgumentParser(\n        formatter_class=argparse.RawDescriptionHelpFormatter,\n        description=__doc__,\n    )\n    p.add_argument(\n        \"--base\",\n        default=os.environ.get(\"KC_BASE\"),\n        required=not os.environ.get(\"KC_BASE\"),\n        help=\"Keycloak base URL (or $KC_BASE)\",\n    )\n    p.add_argument(\"--realm\", action=\"append\", dest=\"realms\", help=\"Realm to scan (repeatable). Default: all, or auth-realm if listing fails\")\n    p.add_argument(\"--auth-realm\", default=os.environ.get(\"KC_AUTH_REALM\", \"master\"), help=\"Realm used for the password grant (default: master)\")\n    p.add_argument(\"--client\", default=os.environ.get(\"KC_CLIENT\", \"admin-cli\"), help=\"Client id for password grant (default: admin-cli)\")\n    p.add_argument(\"--client-secret\", default=os.environ.get(\"KC_CLIENT_SECRET\"), help=\"Client secret if the client is confidential\")\n    p.add_argument(\"--skip-master\", action=\"store_true\", help=\"Skip master when scanning all realms\")\n    p.add_argument(\n        \"--admin\",\n        default=os.environ.get(\"KC_ADMIN\"),\n        help=\"Admin username for the password grant (or $KC_ADMIN). Prompted if missing\",\n    )\n    p.add_argument(\"--admin-pass\", default=os.environ.get(\"KC_ADMIN_PASS\"))\n    p.add_argument(\"--totp\", default=os.environ.get(\"KC_TOTP\"), help=\"OTP code (or $KC_TOTP). Prompted if login requires it\")\n    p.add_argument(\"--since\", default=os.environ.get(\"KC_SINCE\", \"7d\"), help=\"Window: 90m, 24h, 7d, or ISO-8601 (default: 7d)\")\n    p.add_argument(\"--since-hours\", type=float, help=\"Deprecated alias of --since Nh\")\n    p.add_argument(\"--threshold-ms\", type=int, default=int(os.environ.get(\"THRESHOLD_MS\", \"5000\")))\n    p.add_argument(\"--max\", type=int, default=5000, dest=\"cap\", help=\"Max events per type per realm\")\n    p.add_argument(\"--no-color\", action=\"store_true\")\n    args = p.parse_args()\n\n    color = sys.stdout.isatty() and not args.no_color\n    global COLOR_LOG\n    COLOR_LOG = sys.stderr.isatty() and not args.no_color\n    base = args.base.rstrip(\"/\")\n    since = f\"{args.since_hours}h\" if args.since_hours is not None else args.since\n    date_from = parse_since(since)\n    admin = args.admin or input(\"Admin username: \").strip()\n    if not admin:\n        log(\"admin username required\")\n        return 2\n    password = args.admin_pass or getpass.getpass(f\"Password for {admin}: \")\n    totp = args.totp\n\n    log(f\"login {args.client} as {admin} on realm {args.auth_realm} ({base})\")\n    try:\n        token = login(base, args.auth_realm, admin, password, totp, args.client, args.client_secret)\n    except RuntimeError as e:\n        if totp or not needs_otp(e):\n            log(str(e))\n            return 2\n        totp = getpass.getpass(\"OTP: \")\n        log(\"retry login with OTP\")\n        try:\n            token = login(base, args.auth_realm, admin, password, totp, args.client, args.client_secret)\n        except RuntimeError as e2:\n            log(str(e2))\n            return 2\n    log(\"token OK\")\n\n    if args.realms:\n        realms = args.realms\n    else:\n        status, body = http_json(\"GET\", f\"{base}/admin/realms\", token=token)\n        if status == 200 and isinstance(body, list):\n            realms = [str(r.get(\"realm\")) for r in body if r.get(\"realm\")]\n            if args.skip_master:\n                realms = [r for r in realms if r != \"master\"]\n            log(f\"{len(realms)} realm(s): {', '.join(realms)}\")\n        else:\n            log(f\"cannot list all realms (HTTP {status}); scan auth-realm only: {args.auth_realm}\")\n            realms = [args.auth_realm]\n\n    print()\n    print(paint(\"CVE-2026-18963 reset timing hunt\", \"bold\", color))\n    print(f\"  instance   {base}\")\n    print(f\"  auth-realm {args.auth_realm}  client={args.client}\")\n    print(f\"  since      {iso_utc(date_from)}  ({since})\")\n    print(f\"  threshold  {args.threshold_ms} ms\")\n    print()\n\n    all_rows: List[Tuple[str, dict]] = []\n    for realm in realms:\n        log(f\"scan realm {realm}\")\n        sends: List[dict] = []\n        for t in SEND_TYPES:\n            sends.extend(fetch_events(base, token, realm, t, date_from, args.cap))\n        updates: List[dict] = []\n        for t in UPDATE_TYPES:\n            updates.extend(e for e in fetch_events(base, token, realm, t, date_from, args.cap) if is_password_update(e))\n        rows = pairs_for_realm(sends, updates, args.threshold_ms)\n        suspects = sum(1 for r in rows if r[\"suspect\"])\n        log(f\"  {realm}: sends={len(sends)} updates={len(updates)} pairs={len(rows)} suspect={suspects}\")\n        print(paint(f\"> {realm}\", \"bold\", color) + paint(f\"   {len(sends)} send | {len(updates)} update | {len(rows)} pair\", \"dim\", color))\n        print_table(rows, color)\n        print()\n        for r in rows:\n            all_rows.append((realm, r))\n\n    suspects = [(realm, r) for realm, r in all_rows if r[\"suspect\"]]\n    print(paint(\"summary\", \"bold\", color))\n    print(f\"  realms={len(realms)}  pairs={len(all_rows)}  suspect={len(suspects)}\")\n    if suspects:\n        print(paint(\"  -> review SUSPECT rows (send then password change on same code_id).\", \"yellow\", color))\n        return 1\n    print(paint(\"  -> no pair under threshold.\", \"green\", color))\n    return 0\n\n\nif __name__ == \"__main__\":\n    sys.exit(main())", "creation_timestamp": "2026-08-24T18:16:44.821002Z"}