dcerpc: fragment vt_trailer and frag_len properly (#5161)

AI-Assisted: no

Author: gpotter2

scapy/layers/dcerpc.py

@@ -2969,6 +2969,15 @@ def _fragment(self, pkt, body):
                 if not body:
                     # It's the last one
                     pkt_frag.pfc_flags += "PFC_LAST_FRAG"
+                else:
+                    # [MS-RPCE] sect 2.2.2.13 - Verification Trailer
+                    # "only the last PDU of the request MUST have a verification
+                    # trailer"
+                    pkt_frag.vt_trailer = None
+
+                # Update payload for frag_len calculation
+                pkt_frag.payload.payload = conf.raw_layer(load=b"\x00" * len(cur))
+
                 yield pkt_frag, cur
         else:
             yield pkt, body

test/scapy/layers/dcerpc.uts

@@ -65,22 +65,6 @@ f.addfield(None, b'', f.default) == hex_bytes('0123456789abcdef0123456789abcdef'
 
 + DCE/RPC v5
 
-= Require a verifier for requests after packet-protection negotiation
-
-protected_session = DceRpcSession(
-    sspcontext=object(),
-    auth_level=RPC_C_AUTHN_LEVEL.PKT_PRIVACY,
-)
-for protected_payload in (DceRpc5Request(), DceRpc5Response()):
-    unsigned_packet = DceRpc5(
-        pfc_flags="PFC_FIRST_FRAG+PFC_LAST_FRAG",
-    ) / protected_payload
-    try:
-        protected_session.in_pkt(unsigned_packet)
-        assert False, "The unsigned packet was accepted"
-    except ValueError as exc:
-        assert str(exc) == "DCE/RPC packet protection is required !"
-
 = Dissect DCE/RPC v5 Request with Kerberos GSSAPI/RFC1964
 
 pkt = DceRpc(b"\x05\x00\x00\x03\x10\x00\x00\x00\xcd\x00-\x00\x01\x00\x00\x00x\x00\x00\x00\x00\x00\x00\x00j\x87\xb4\xa8DrE3\xfa\xc1\x1d\x9e\xb7\x8a_\xffr\xbe\x13\xc4<\x85\xf0\xf2'y\x84t%u|e\xef/\x04\xb0m\x98\xb1\xd2\x00KwW#P\x8f2\xecB\x81\x19\xf3g\xd2o[\x07L-\xb8\x89\x05\xcf?\xcf\t\xeb\xb3&&6\xb7\x84\xb6\xcd8Ao\x8c\x94\xca\x03\xe3\x0e\x86'-\xfaHj\xcez\xf0A\x83\x9dX\r\xe8\x96\x07Bs\xaf\x9c[=2\x9eS\xb1\x18\x84 \xb4y\n9\xdf\x92\x1c\xd8\xe2e\xd3^,\t\x06\x08\x00pj\x8f\x04`+\x06\t*\x86H\x86\xf7\x12\x01\x02\x02\x02\x01\x11\x00\x10\x00\xff\xffp\xc0\\m\xfe\xa4\xe1!\xf7\xdf\xbf\xa4\xad\xdf\xcb\x16\x1e\xb5+{\x97\xaf\xd5~")
@@ -171,6 +155,41 @@ assert DceRpc5Request in pkt2
 assert conf.padding_layer not in pkt2
 assert pkt2.vt_trailer.commands[1].InterfaceId == pkt2.object
 
+= Build and dissect DCE/RPC - Fragment test
+
+from unittest import mock
+from scapy.layers.msrpce.raw.ept import ept_lookup_Request
+
+# 1. Build a KerberosSSP ready to encrypt
+ssp = KerberosSSP()
+ctx = KerberosSSP.CONTEXT(IsAcceptor=False, req_flags=GSS_C_FLAGS.GSS_C_CONF_FLAG)
+ctx.KrbSessionKey = Key(EncryptionType.AES128_CTS_HMAC_SHA1_96, key=bytes.fromhex("3705D96080C17728A0E800EAB6E0D23C"))
+ctx.SendSeqNum = 0x60cbacd3
+Confounder = bytes.fromhex("5256f3fb630cf12a")
+session = DceRpcSession(
+    auth_level=RPC_C_AUTHN_LEVEL.PKT_PRIVACY,
+    ssp=ssp,
+)
+session.sspcontext = ctx
+
+# 2. Make a big packet go out, this will trigger fragmentation
+pkt = DceRpc5(call_id=12, vt_trailer=DceRpcSecVT(commands=[DceRpcSecVTCommand(SEC_VT_COMMAND_END=1)/DceRpcSecVTPcontext()])) / DceRpc5Request() / (b"\x11" * 5000)
+with mock.patch('scapy.layers.kerberos.os.urandom', side_effect=lambda x: Confounder):
+    pkt = session.out_pkt(pkt)
+
+# Checks on the fragmented packet
+assert len(pkt) == 2
+
+assert pkt[0].pfc_flags == 1
+assert pkt[0].vt_trailer is None
+assert bytes(pkt[0].auth_verifier.auth_value) == b'\x05\x04\x06\xff\x00\x10\x00\x1c\x00\x00\x00\x00`\xcb\xac\xd3\xc3\xb9\r\xf4F\xf5\x17\xc2\x9cvA\xd2,\x15\x13\xb4\xdcI\x06\x8b^\x057\xe1\xa0\x18\xbd\t=\xba)\xa3\xe1\xa1y\x81\xc7b\xdc\xbd\xd8\xf2(\xd1q?\xd6#\x8e\xd2^\xa0'
+assert len(pkt[0].load) == 4176
+
+assert pkt[1].pfc_flags == 2
+assert pkt[1].vt_trailer is not None
+assert bytes(pkt[1].auth_verifier.auth_value) == b'\x05\x04\x06\xff\x00\x10\x00\x1c\x00\x00\x00\x00`\xcb\xac\xd4\x9c_--ht\xc4\xc3\xfd\x8e!\x04;\x8d\xe1\xe1\x8c\xf3b\xb9\xb4\xb5%J\xff\xc5\xcb\xc1\x87\xdc]\x85S\x18\x07\x1d~e\x8f\xa6v]G\xa1q?\xd6#\x8e\xd2^\xa0'
+assert len(pkt[1].load) == 824
+
 + Check DCE/RPC 4 layer
 
 = DCE/RPC 4 default values
@@ -617,15 +636,6 @@ assert pkts[22][ept_map_Response].valueof("ITowers")[0].tower_octet_string == b'
 
 % The fact that all of this actually works is crazy to me.
 
-= Server faults a request received before bind
-from scapy.layers.dcerpc import DCERPC_Transport, DceRpc5Fault
-from scapy.layers.msrpce.rpcserver import DCERPC_Server
-
-server = DCERPC_Server(DCERPC_Transport.NCACN_IP_TCP, port=135, verb=False)
-request = hex_bytes("050000031000000018000000000000000000000000000000")
-server.recv(request)
-assert DceRpc5Fault in server.get_response()
-
 = Functional: Define a MS-RPC server
 % Same as in dcerpc.rst
 
@@ -670,27 +680,6 @@ class MyRPCServer(DCERPC_Server):
             ndr64=self.ndr64,
         )
 
-= Auth3 cannot lower the protection level negotiated during bind
-
-class TestSSP:
-    def GSS_Accept_sec_context(self, context, token):
-        return object(), None, 0
-
-server = DCERPC_Server(
-    DCERPC_Transport.NCACN_IP_TCP,
-    verb=False,
-    ssp=TestSSP(),
-)
-server.session.rpc_bind_interface = find_dcerpc_interface("wkssvc")
-server.session.auth_level = DCE_C_AUTHN_LEVEL.PKT_INTEGRITY
-auth3 = DceRpc5() / DceRpc5Auth3()
-auth3.auth_verifier = CommonAuthVerifier(
-    auth_level=DCE_C_AUTHN_LEVEL.CONNECT,
-    auth_value=Raw(b"authenticate"),
-)
-server.recv(auth3)
-assert server.session.auth_level == DCE_C_AUTHN_LEVEL.PKT_INTEGRITY
-
 = Functional: Define wrapper over samba's rpcclient
 ~ linux samba
 
@@ -1025,3 +1014,36 @@ rpcserver.close()
 = Restore conf.debug_dissector
 
 conf.debug_dissector = old_debug_dissector
+
++ Security regression tests
+
+= Security - Auth3 cannot lower the protection level negotiated during bind
+
+class TestSSP:
+    def GSS_Accept_sec_context(self, context, token):
+        return object(), None, 0
+
+server = DCERPC_Server(
+    DCERPC_Transport.NCACN_IP_TCP,
+    verb=False,
+    ssp=TestSSP(),
+)
+server.session.rpc_bind_interface = find_dcerpc_interface("wkssvc")
+server.session.auth_level = DCE_C_AUTHN_LEVEL.PKT_INTEGRITY
+auth3 = DceRpc5() / DceRpc5Auth3()
+auth3.auth_verifier = CommonAuthVerifier(
+    auth_level=DCE_C_AUTHN_LEVEL.CONNECT,
+    auth_value=Raw(b"authenticate"),
+)
+server.recv(auth3)
+assert server.session.auth_level == DCE_C_AUTHN_LEVEL.PKT_INTEGRITY
+
+= Security - Server faults a request received before bind
+
+from scapy.layers.dcerpc import DCERPC_Transport, DceRpc5Fault
+from scapy.layers.msrpce.rpcserver import DCERPC_Server
+
+server = DCERPC_Server(DCERPC_Transport.NCACN_IP_TCP, port=135, verb=False)
+request = hex_bytes("050000031000000018000000000000000000000000000000")
+server.recv(request)
+assert DceRpc5Fault in server.get_response()