diff --git a/scapy/layers/msrpce/rpcserver.py b/scapy/layers/msrpce/rpcserver.py index 8b65164c2cc..de177563039 100644 --- a/scapy/layers/msrpce/rpcserver.py +++ b/scapy/layers/msrpce/rpcserver.py @@ -18,6 +18,7 @@ from scapy.volatile import RandShort from scapy.layers.dcerpc import ( + COM_INTERFACES, CommonAuthVerifier, DCE_RPC_INTERFACES, DCERPC_Transport, @@ -248,6 +249,8 @@ def recv(self, data): pad = req[conf.padding_layer].load req[conf.padding_layer].underlayer.remove_payload() # Ask the DCE/RPC session to process it (match interface, etc.) + previous_interface = self.session.rpc_bind_interface + previous_interface_is_com = self.session.rpc_bind_is_com req = self.session.in_pkt(req) hdr = DceRpc5( endian=req.endian, @@ -325,6 +328,7 @@ def recv(self, data): # Process bind contexts and answer to them results = [] + accepted_context = None for ctx in req.context_elem: # Get name name = ctx.transfer_syntaxes[0].sprintf("%if_uuid%") @@ -336,6 +340,7 @@ def recv(self, data): (name == "NDR 2.0" and not self.ndr64) ): # Acceptance + accepted_context = ctx results.append( DceRpc5Result( result=0, @@ -404,6 +409,18 @@ def recv(self, data): + ("NDR64" if self.ndr64 else "NDR32") ) ) + if accepted_context is None: + self.session.rpc_bind_interface = previous_interface + self.session.rpc_bind_is_com = previous_interface_is_com + elif len(req.context_elem) > 1: + abstract_syntax = accepted_context.abstract_syntax + interface = DCE_RPC_INTERFACES.get( + (abstract_syntax.if_uuid, abstract_syntax.if_version) + ) + self.session.rpc_bind_interface = interface or COM_INTERFACES.get( + abstract_syntax.if_uuid + ) + self.session.rpc_bind_is_com = interface is None elif DceRpc5Request in req: if self.verb: print( diff --git a/test/scapy/layers/dcerpc.uts b/test/scapy/layers/dcerpc.uts index 484e52a606b..b81f4d5416d 100644 --- a/test/scapy/layers/dcerpc.uts +++ b/test/scapy/layers/dcerpc.uts @@ -974,6 +974,65 @@ except OSError: rpcserver.close() += A rejected presentation context cannot select an RPC interface + +# Take the classes from the registry rather than from the module. A second +# import of an already-registered interface returns before binding opnum and +# intf, so the module-level names are not always the registered objects. +_ept_free = find_dcerpc_interface("ept").opnums[4] +ept_lookup_handle_free_Request = _ept_free.request +ept_lookup_handle_free_Response = _ept_free.response + +class PresentationContextServer(DCERPC_Server): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.handler_calls = 0 + @DCERPC_Server.answer(ept_lookup_handle_free_Request) + def operation(self, request): + self.handler_calls += 1 + return ept_lookup_handle_free_Response(status=0, ndr64=False) + +def drive_presentation_context(transfer_syntax): + interface = find_dcerpc_interface("ept") + syntax = next( + syntax + for syntax, name in DCE_RPC_TRANSFER_SYNTAXES.items() + if name == transfer_syntax + ) + context = DceRpc5Context( + abstract_syntax=DceRpc5AbstractSyntax( + if_uuid=interface.uuid, + if_version=interface.if_version, + ), + transfer_syntaxes=[ + DceRpc5TransferSyntax( + if_uuid=syntax, + if_version=2 if transfer_syntax == "NDR 2.0" else 0, + ) + ], + ) + server = PresentationContextServer( + DCERPC_Transport.NCACN_IP_TCP, + verb=False, + port=12345, + ) + server.recv(bytes(DceRpc5(call_id=1) / DceRpc5Bind(context_elem=[context]))) + result = int(server.get_response().results[0].result) + try: + server.recv( + bytes( + DceRpc5(call_id=2) + / DceRpc5Request(opnum=4) + / ept_lookup_handle_free_Request(ndr64=False) + ) + ) + except AttributeError: + pass + return result, server.handler_calls + +assert drive_presentation_context("NULL") == (2, 0) +assert drive_presentation_context("NDR 2.0") == (0, 1) + + Cleanup = Restore conf.debug_dissector