Support In-situ Operation Administration and Maintenance (IOAM)

scapy/contrib/ioam.py

@@ -0,0 +1,54 @@
+# scapy.contrib.description = In-situ Operation Administration and Maintenance (IOAM)
+# scapy.contrib.status = loads
+
+'''
+In-situ Operation Administration and Maintenance (IOAM)
+
+Notice:
+This is China Mobile Communications Corporation (CMCC) IOAM, instead of Cisco IOAM
+
+IOAM Shim Reference:
+https://datatracker.ietf.org/doc/rfc9486/
+https://datatracker.ietf.org/doc/html/rfc8321
+
+IOAM Report Packet Reference:
+https://datatracker.ietf.org/doc/draft-ietf-netconf-udp-notif/12/
+
+IOAM layer identifier:
+IPv4.proto == 186
+IPv6.NextHeader == 0x0 && IPv6.HBH.option_type == 0x11
+
+Example Packet Format:
+IOAMoIPv4       = Ether/IP/IOAM/Payload
+IOAMoIPv4UDP    = Ether/IP/IOAM/UDP/Payload
+IOAMoIPv4TCP    = Ether/IP/IOAM/TCP/Payload
+IOAMoIPv4VxLAN  = Ether/IP/IOAM/UDP/VXLAN/Ether/IP/TCP/Payload
+IOAMoIPv6IP     = Ether/IPv6/IPv6ExtHdrHopByHop(nh=59, options=[HBHOptIOAM(ioam=ioam)])/Payload # noqa: E501
+IOAMoIPv6UDP    = Ether/IPv6/IPv6ExtHdrHopByHop(nh=socket.IPPROTO_UDP, options=[HBHOptIOAM(ioam=IOAM)])/UDP/Payload # noqa: E501
+IOAMoIPv6TCP    = Ether/IPv6/IPv6ExtHdrHopByHop(nh=socket.IPPROTO_TCP, options=[HBHOptIOAM(ioam=IOAM)])/TCP/Payload # noqa: E501
+IOAMoIPv6VxLAN  = Ether/IPv6/IPv6ExtHdrHopByHop(nh=socket.IPPROTO_UDP, options=[HBHOptIOAM(ioam=IOAM)])/UDP/VXLAN/Ether/IP/TCP/Payload # noqa: E501
+'''
+
+import socket
+from scapy.packet import Packet, bind_layers
+from scapy.fields import BitField, ByteField
+from scapy.layers.inet import IP, TCP, UDP
+
+IPPROTO_IOAM = 186
+
+
+class IOAM(Packet):
+    name = 'IOAM'
+    fields_desc = [
+        BitField('flow_id', 0, 20),
+        BitField('color_val', 0, 1),
+        BitField('delay_en', 0, 1),
+        BitField('color_en', 0, 1),
+        BitField('reserved', 0, 1),
+        ByteField('next_protocol', 0),
+    ]
+
+
+bind_layers(IP, IOAM, proto=IPPROTO_IOAM)
+bind_layers(IOAM, TCP, next_protocol=socket.IPPROTO_TCP)
+bind_layers(IOAM, UDP, next_protocol=socket.IPPROTO_UDP)

scapy/layers/inet.py

@@ -749,11 +749,16 @@ def post_build(self, p, pay):
             dataofs = (dataofs << 4) | orb(p[12]) & 0x0f
             p = p[:12] + chb(dataofs & 0xff) + p[13:]
         if self.chksum is None:
-            if isinstance(self.underlayer, IP):
-                ck = in4_chksum(socket.IPPROTO_TCP, self.underlayer, p)
+            selfdown = self.underlayer
+            from scapy.contrib.ioam import IOAM
+            if isinstance(self.underlayer, IOAM):
+                selfdown = self.underlayer.underlayer
+
+            if isinstance(selfdown, IP):
+                ck = in4_chksum(socket.IPPROTO_TCP, selfdown, p)
                 p = p[:16] + struct.pack("!H", ck) + p[18:]
-            elif conf.ipv6_enabled and isinstance(self.underlayer, scapy.layers.inet6.IPv6) or isinstance(self.underlayer, scapy.layers.inet6._IPv6ExtHdr):  # noqa: E501
-                ck = scapy.layers.inet6.in6_chksum(socket.IPPROTO_TCP, self.underlayer, p)  # noqa: E501
+            elif conf.ipv6_enabled and isinstance(selfdown, scapy.layers.inet6.IPv6) or isinstance(selfdown, scapy.layers.inet6._IPv6ExtHdr):  # noqa: E501
+                ck = scapy.layers.inet6.in6_chksum(socket.IPPROTO_TCP, selfdown, p)  # noqa: E501
                 p = p[:16] + struct.pack("!H", ck) + p[18:]
             else:
                 log_runtime.info(
@@ -821,14 +826,19 @@ def post_build(self, p, pay):
             tmp_len = len(p)
             p = p[:4] + struct.pack("!H", tmp_len) + p[6:]
         if self.chksum is None:
-            if isinstance(self.underlayer, IP):
-                ck = in4_chksum(socket.IPPROTO_UDP, self.underlayer, p)
+            selfdown = self.underlayer
+            from scapy.contrib.ioam import IOAM
+            if isinstance(self.underlayer, IOAM):
+                selfdown = self.underlayer.underlayer
+
+            if isinstance(selfdown, IP):
+                ck = in4_chksum(socket.IPPROTO_UDP, selfdown, p)
                 # According to RFC768 if the result checksum is 0, it should be set to 0xFFFF  # noqa: E501
                 if ck == 0:
                     ck = 0xFFFF
                 p = p[:6] + struct.pack("!H", ck) + p[8:]
-            elif isinstance(self.underlayer, scapy.layers.inet6.IPv6) or isinstance(self.underlayer, scapy.layers.inet6._IPv6ExtHdr):  # noqa: E501
-                ck = scapy.layers.inet6.in6_chksum(socket.IPPROTO_UDP, self.underlayer, p)  # noqa: E501
+            elif isinstance(selfdown, scapy.layers.inet6.IPv6) or isinstance(selfdown, scapy.layers.inet6._IPv6ExtHdr):  # noqa: E501
+                ck = scapy.layers.inet6.in6_chksum(socket.IPPROTO_UDP, selfdown, p)  # noqa: E501
                 # According to RFC2460 if the result checksum is 0, it should be set to 0xFFFF  # noqa: E501
                 if ck == 0:
                     ck = 0xFFFF

scapy/layers/inet6.py

@@ -47,6 +47,7 @@
     LongField,
     MACField,
     MayEnd,
+    PacketField,
     PacketLenField,
     PacketListField,
     ShortEnumField,
@@ -674,6 +675,7 @@ class _IPv6ExtHdr(_IPv6GuessPayload, Packet):
             0x04: "Tunnel Encapsulation Limit",
             0x05: "Router Alert",
             0x06: "Quick-Start",
+            0x11: "IOAM",
             0xc2: "Jumbo Payload",
             0xc9: "Home Address Option"}
 
@@ -783,6 +785,23 @@ def extract_padding(self, p):
         return b"", p
 
 
+class HBHOptIOAM(Packet):  # IPv6 Hop-By-Hop Option
+    from scapy.contrib.ioam import IOAM
+    name = "HBHOptIOAM"
+    fields_desc = [_OTypeField("otype", 0x11, _hbhopts),
+                   ByteField("optlen", 4),
+                   PacketField("ioam", None, IOAM)]
+
+    def alignment_delta(self, curpos):  # alignment requirement : 4n+2
+        x = 4
+        y = 0
+        delta = x * ((curpos - y + x - 1) // x) + y - curpos
+        return delta
+
+    def extract_padding(self, p):
+        return b"", p
+
+
 class Jumbo(Packet):  # IPv6 Hop-By-Hop Option
     name = "Jumbo Payload"
     fields_desc = [_OTypeField("otype", 0xC2, _hbhopts),
@@ -818,6 +837,7 @@ def extract_padding(self, p):
 _hbhoptcls = {0x00: Pad1,
               0x01: PadN,
               0x05: RouterAlert,
+              0x11: HBHOptIOAM,
               0xC2: Jumbo,
               0xC9: HAO}
 

test/contrib/ioam.uts

@@ -0,0 +1,139 @@
+% IOAM unit test
+
+#
+# execute test:
+# > test/run_tests -P "load_contrib('ioam')" -t test/contrib/ioam.uts
+#
+
++ IOAM testsuit
+
+
+= Build & Dissect, IOAM Over IPv4
+
+def check_ioam(pkt:Packet):
+    protocols = [0, socket.IPPROTO_UDP, socket.IPPROTO_TCP]
+    assert pkt[IOAM].flow_id == 100
+    assert pkt[IOAM].color_val == 1
+    assert pkt[IOAM].delay_en == 0
+    assert pkt[IOAM].color_en == 1
+    assert (((not (pkt.haslayer(UDP) or pkt.haslayer(UDP))) and (pkt[IOAM].next_protocol == 0)) or
+            (pkt.haslayer(HBHOptIOAM) and (pkt[IOAM].next_protocol == 0)) or
+            (pkt.haslayer(UDP) and (pkt[IOAM].next_protocol == socket.IPPROTO_UDP)) or
+            (pkt.haslayer(TCP) and (pkt[IOAM].next_protocol == socket.IPPROTO_TCP)))
+
+ioam    = IOAM(flow_id=100, color_val=1, delay_en=0, color_en=1)
+o_eth   = Ether(src="b6:18:00:11:11:00", dst="b6:18:00:22:22:00")
+o_ip4   = IP(src="198.1.1.17", dst="198.1.2.18", ttl=63)
+o_ip6   = IPv6(src="2000::1", dst="5000::2", hlim=63)
+o_udp   = UDP(sport=4196, dport=9028)
+o_tcp   = TCP(sport=4196, dport=9028)
+o_vxlan = VXLAN(vni=5000)
+i_eth   = Ether(dst='b6:18:00:88:88:00', src='b6:18:00:99:99:00')
+i_ip4   = IP(src='192.168.1.5', dst='192.168.6.9', ttl=128)
+i_tcp   = TCP(sport=6677, dport=8899)
+payload = Raw('a'*64)
+
+pkt=o_eth/o_ip4/ioam/payload
+pkt=Ether(raw(pkt))
+# pkt.show2()
+assert pkt[IP].proto == IPPROTO_IOAM
+assert pkt[IP].chksum == 0xebc5
+check_ioam(pkt)
+
+
+= Build & Dissect, IOAM Over IPv4 UDP
+
+pkt=o_eth/o_ip4/ioam/o_udp/payload
+pkt=Ether(raw(pkt))
+# pkt.show2()
+assert pkt[IP].proto == IPPROTO_IOAM
+assert pkt[IP].chksum == 0xebbd
+check_ioam(pkt)
+assert pkt[UDP].sport == 4196
+assert pkt[UDP].dport == 9028
+assert pkt[UDP].chksum == 0x1064
+
+
+= Build & Dissect, IOAM Over IPv4 TCP
+
+pkt=o_eth/o_ip4/ioam/o_tcp/payload
+pkt=Ether(raw(pkt))
+# pkt.show2()
+assert pkt[IP].proto == IPPROTO_IOAM
+assert pkt[IP].chksum == 0xebb1
+check_ioam(pkt)
+assert pkt[TCP].sport == 4196
+assert pkt[TCP].dport == 9028
+assert pkt[TCP].chksum == 0xa0a8
+
+
+= Build & Dissect, IOAM Over IPv4 VXLAN
+
+pkt=o_eth/o_ip4/ioam/UDP(sport=4196, dport=4789)/o_vxlan/i_eth/i_ip4/i_tcp/payload
+pkt=Ether(raw(pkt))
+# pkt.show2()
+assert pkt[IP].proto == IPPROTO_IOAM
+assert pkt[IP].chksum == 0xeb7f
+check_ioam(pkt)
+assert pkt[UDP].sport == 4196
+assert pkt[UDP].dport == 4789
+assert pkt[UDP].chksum == 0xaaf2
+assert pkt[VXLAN].vni == 5000
+
+
+= Build & Dissect, IOAM Over IPv6
+
+pkt=o_eth/o_ip6/IPv6ExtHdrHopByHop(nh=59, options=[HBHOptIOAM(ioam=ioam)])/payload
+pkt=Ether(raw(pkt))
+# pkt.show2()
+assert pkt[IPv6].nh == 0
+assert pkt[IPv6ExtHdrHopByHop].nh == 59
+assert pkt[HBHOptIOAM].otype == 0x11
+assert pkt[HBHOptIOAM].optlen == 4
+check_ioam(pkt)
+
+
+= Build & Dissect, IOAM Over IPv6 UDP
+
+pkt=o_eth/o_ip6/IPv6ExtHdrHopByHop(nh=socket.IPPROTO_UDP, options=[HBHOptIOAM(ioam=ioam)])/o_udp/payload
+pkt=Ether(raw(pkt))
+# pkt.show2()
+assert pkt[IPv6].nh == 0
+assert pkt[IPv6ExtHdrHopByHop].nh == socket.IPPROTO_UDP
+assert pkt[HBHOptIOAM].otype == 0x11
+assert pkt[HBHOptIOAM].optlen == 4
+check_ioam(pkt)
+assert pkt[UDP].sport == 4196
+assert pkt[UDP].dport == 9028
+assert pkt[UDP].chksum == 0x2f87
+
+
+= Build & Dissect, IOAM Over IPv6 TCP
+
+pkt=o_eth/o_ip6/IPv6ExtHdrHopByHop(nh=socket.IPPROTO_TCP, options=[HBHOptIOAM(ioam=ioam)])/o_tcp/payload
+pkt=Ether(raw(pkt))
+# pkt.show2()
+assert pkt[IPv6].nh == 0
+assert pkt[IPv6ExtHdrHopByHop].nh == socket.IPPROTO_TCP
+assert pkt[HBHOptIOAM].otype == 0x11
+assert pkt[HBHOptIOAM].optlen == 4
+check_ioam(pkt)
+assert pkt[TCP].sport == 4196
+assert pkt[TCP].dport == 9028
+assert pkt[TCP].chksum == 0xbfcb
+
+
+= Build & Dissect, IOAM Over IPv6 VXLAN
+
+pkt=o_eth/o_ip6/IPv6ExtHdrHopByHop(nh=socket.IPPROTO_UDP, options=[HBHOptIOAM(ioam=ioam)])/UDP(sport=4196, dport=4789)/o_vxlan/i_eth/i_ip4/i_tcp/payload
+pkt=Ether(raw(pkt))
+# pkt.show2()
+assert pkt[IPv6].nh == 0
+assert pkt[IPv6ExtHdrHopByHop].nh == socket.IPPROTO_UDP
+assert pkt[HBHOptIOAM].otype == 0x11
+assert pkt[HBHOptIOAM].optlen == 4
+check_ioam(pkt)
+assert pkt[UDP].sport == 4196
+assert pkt[UDP].dport == 4789
+assert pkt[UDP].chksum == 0xca15
+assert pkt[VXLAN].vni == 5000