Detect TCPError/UDPerror/SCTPerror after IPv6 Extension Headers

Author: guedou

scapy/layers/inet6.py

@@ -693,11 +693,23 @@ def in6_chksum(nh, u, p):
 #############################################################################
 #############################################################################
 
+nh_clserror = {socket.IPPROTO_TCP: TCPerror,
+               socket.IPPROTO_UDP: UDPerror}
+
 
 # Inherited by all extension header classes
 class _IPv6ExtHdr(_IPv6GuessPayload, Packet):
     name = 'Abstract IPv6 Option Header'
-    aliastypes = [IPv6, IPerror6]  # TODO ...
+    aliastypes = [IPv6]
+
+    def guess_payload_class(self, payload):
+        if self.nh in nh_clserror:
+            underlayer = self.underlayer
+            while underlayer:
+                if isinstance(underlayer, IPerror6):
+                    return nh_clserror[self.nh]
+                underlayer = underlayer.underlayer
+        return super(_IPv6ExtHdr, self).guess_payload_class(payload)
 
 
 #                    IPv6 options for Extension Headers                     #

scapy/layers/sctp.py

@@ -37,7 +37,7 @@
 )
 from scapy.data import SCTP_SERVICES
 from scapy.layers.inet import IP, IPerror
-from scapy.layers.inet6 import IP6Field, IPv6, IPerror6
+from scapy.layers.inet6 import IP6Field, IPv6, IPerror6, nh_clserror
 
 IPPROTO_SCTP = 132
 
@@ -304,6 +304,8 @@ def mysummary(self):
         return Packet.mysummary(self)
 
 
+nh_clserror[IPPROTO_SCTP] = SCTPerror
+
 # SCTP Chunk variable params
 
 

test/scapy/layers/inet6.uts

@@ -1996,7 +1996,7 @@ def test_getmacbyip6(mock_route6, mock_neighsol):
 
 test_getmacbyip6() == "05:04:03:02:01:00"
 
-= IPv6 - IPerror6 & UDPerror & _ICMPv6Error
+= IPv6 - IPerror6 & UDPerror & _ICMPv6Error & TCPerror
 
 query = IPv6(dst="2001:db8::1", src="2001:db8::2", hlim=1)/UDP()/DNS()
 answer = IPv6(dst="2001:db8::2", src="2001:db8::1", hlim=1)/ICMPv6TimeExceeded()/IPerror6(dst="2001:db8::1", src="2001:db8::2", hlim=0)/UDPerror()/DNS()
@@ -2007,6 +2007,13 @@ from scapy.layers.inet6 import _ICMPv6Error
 assert _ICMPv6Error().guess_payload_class(None) == IPerror6
 assert _ICMPv6Error().hashret() == b''
 
+# Test with extension header
+# From:
+# pkt = IPv6() / ICMPv6DestUnreach() / IPerror6 () / IPv6ExtHdrFragment(nh=6) / TCPerror()
+raw_pkt = b'`\x00\x00\x00\x00L:@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x01\x00l5\x00\x00\x00\x00`\x00\x00\x00\x00\x1c,@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x06\x00\x00\x00\x00\x00\x00\x00\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x8f}\x00\x00'
+IPv6(raw_pkt).summary()
+assert TCPerror in IPv6(raw_pkt)
+
 = reset routes properly
 
 conf.ifaces.reload()