dcommap.py

import json
import argparse
from typing import Dict
from datetime import datetime
from rich import get_console

from impacket.uuid import bin_to_string, string_to_bin
from impacket.examples.utils import parse_target
from impacket.smbconnection import SMBConnection
from impacket.dcerpc.v5 import rrp, dcomrt, transport, rpcrt

from utils.reg import RemoteOperations, yield_key, Key
from utils.dcom import prepare_guid, generate_clsid

CONSOLE = get_console()

def parse_args():
    parser = argparse.ArgumentParser(description="Windows DCOM Mapping Utility")
    parser.add_argument('target', action='store', help='[[domain/]username[:password]@]<targetName or address>')
    
    group = parser.add_argument_group('Authentication')

    group.add_argument('--hashes', action="store", metavar="LMHASH:NTHASH", help='NTLM hashes, format is LMHASH:NTHASH')
    group.add_argument('--no-pass', action="store_true", help='don\'t ask for password (useful for -k)')
    group.add_argument('-k', action="store_true",
                       help='Use Kerberos authentication. Grabs credentials from ccache file (KRB5CCNAME) based on '
                            'target parameters. If valid credentials cannot be found, it will use the ones specified '
                            'in the command line')
    group.add_argument('--aes-key', action="store", metavar="hex key",
                       help='AES key to use for Kerberos Authentication (128 or 256 bits)')

    group = parser.add_argument_group('Connection')

    group.add_argument('--dc-ip', action='store', metavar="ip address",
                       help='IP Address of the domain controller. If omitted it will use the domain part (FQDN) specified in '
                            'the target parameter.')
    group.add_argument('--kdc', action='store', metavar="ip address",
                       help='IP Address of the Key Distribution Center used for kerberos authentication. If omitted it will use the domain part (FQDN) specified in '
                            'the target parameter.')
    group.add_argument('--target-ip', action='store', metavar="ip address",
                       help='IP Address of the target machine. If omitted it will use whatever was specified as target. '
                            'This is useful when target is the NetBIOS name and you cannot resolve it')
    group.add_argument('-p', '--port', nargs='?', type=int, default=445, metavar="destination port",
                       help='Destination port to connect to SMB Server')
    
    group = parser.add_argument_group('Scan')
    
    group.add_argument('-i', '--interface', metavar='GUID', default=bin_to_string(dcomrt.IID_IRemUnknown), help="The GUID of the interface (IID) to try and bind to when scanning, by default it's IRemUnknown.")
    
    options = parser.parse_args()

    domain, username, password, remoteName = parse_target(options.target)

    if options.target_ip is None:
        options.target_ip = remoteName

    if domain is None:
        domain = ''

    if options.aes_key is not None:
        options.k = True

    if password == '' and username != '' and options.hashes is None and options.no_pass is False and options.aes_key is None:
        from getpass import getpass

        password = getpass("Password:")
    
    return options, remoteName, domain, username, password
    

def scan_clsid(dcom: dcomrt.DCOMConnection, clsid: dcomrt.CLSID, interface: bytes) -> Dict[str, str]:
    try:
        dcom.CoCreateInstanceEx(clsid, interface)
    except dcomrt.DCERPCSessionError as err:
        if 'REGDB_E_CLASSNOTREG' in str(err):
            return {
                'result': 'Not exposed externally.',
                'summary': 'Not Accessible :('
            }
        else:
            return {
                'result': str(err),
                'summary': 'Accessible!'
            }
    except TimeoutError:
        return {
            'result': 'Timed out - COM Server Missing?',
            'summary': 'Not Accessible :('
        }
    else:
        return {
            'result': f'Successfully Created Instance of Interface ({bin_to_string(interface)})!',
            'summary': 'Accessible!'
        }


def main():
    lmhash = ''
    nthash = ''
    options, remoteName, domain, username, password = parse_args()
    interface = string_to_bin(prepare_guid(options.interface))
    
    CONSOLE.log("Creating SMB Connection to remote host...")
    
    smb = SMBConnection(remoteName, options.target_ip)
    
    if options.hashes is not None:
        lmhash, nthash = options.hashes.split(':')
        
    if options.k:
        smb.kerberosLogin(username, password, domain, lmhash, nthash, options.aes_key, options.kdc)
    else:
        smb.login(username, password, domain, lmhash, nthash)

    CONSOLE.log("Creating DCOM Connection to remote host...")
    
    dcom_connection = dcomrt.DCOMConnection(
        options.target_ip,
        username,
        password,
        domain,
        oxidResolver=True,
    )
    portmap: transport.DCERPC_v5 = dcom_connection.get_dce_rpc()
    portmap.set_auth_level(rpcrt.RPC_C_AUTHN_LEVEL_NONE)
    portmap.connect()
    ox = dcomrt.IObjectExporter(portmap)
    bindings = ox.ServerAlive2()
    portmap.disconnect()
    portmap.set_auth_level(rpcrt.RPC_C_AUTHN_LEVEL_PKT_INTEGRITY)
    portmap.connect()
    CONSOLE.log(f"Connected To Host!")
    
    for b in bindings:
        CONSOLE.log(f"\t- {b['aNetworkAddr']}")
    try:
        results = {}
        with RemoteOperations(smb, doKerberos=False) as remops:
            rrp_dcerpc = remops.rrp_dcerpc
            hkcr = rrp.hOpenClassesRoot(rrp_dcerpc)['phKey']
            with Key(rrp_dcerpc, hkcr, "CLSID") as hkey:
                key_gen = yield_key(rrp_dcerpc, hkey)
                count = 0
                CONSOLE.log("Reading CLSIDs from registy using [MS-RRP], this may take a while...\nMaybe go take a shower? you stink :P")
                with CONSOLE.status(f"Scanning CLSIDs from HKCR (total: {count})...") as s:
                    for key in key_gen:
                        key_str = str(key['lpNameOut'][:-1])
                        try:
                            clsid = prepare_guid(key_str) # Test key for validity.
                        except ValueError:
                            continue
                        
                        try:
                            s.update(f"Testing key '{key_str}' CLSID for DCOM usability (total: {count})...")
                            results[clsid] = scan_clsid(dcom_connection, generate_clsid(key_str), interface)
                        finally:
                            # Must reset bind context for some reason - maybe find a way to make it better?
                            portmap.disconnect()
                            portmap.connect()
                            count += 1
                            if count == 100:
                                break
    finally:
        dcom_connection.disconnect()
        smb.close()

    timestamp = datetime.now().isoformat(sep='_').rsplit('.')[0]
    with open(f"dcom_clsid_scan_{timestamp}.json", 'w') as f:
        f.write(json.dumps(results))

if __name__ == '__main__':
    main()