HSRP version 2

scapy/layers/hsrp.py

@@ -3,23 +3,42 @@
 # See https://scapy.net/ for more information
 # Copyright (C)  Mathieu RENARD <mathieu.renard(at)gmail.com>
 
+# scapy.contrib.description = Hot Standby Router Protocol (HSRP)
+# scapy.contrib.status = loads
+
 """
 HSRP (Hot Standby Router Protocol)
 A proprietary redundancy protocol for Cisco routers.
 
-- HSRP Version 1: RFC 2281
+- HSRP Version 1:
+    https://tools.ietf.org/html/rfc2281
 - HSRP Version 2:
     http://www.smartnetworks.jp/2006/02/hsrp_8_hsrp_version_2.html
 """
 
+import struct
 from scapy.config import conf
-from scapy.compat import orb
-from scapy.fields import ByteEnumField, ByteField, IntField, IPField, \
-    ShortEnumField, ShortField, SourceIPField, StrFixedLenField, \
-    XIntField, XShortField
+from scapy.fields import (
+    ByteEnumField,
+    ByteField,
+    ConditionalField,
+    IntField,
+    IP6Field,
+    IPField,
+    MultipleTypeField,
+    ShortEnumField,
+    ShortField,
+    SourceIPField,
+    StrField,
+    StrFixedLenField,
+    XIntField,
+    XShortField,
+)
 from scapy.packet import Packet, bind_layers, bind_bottom_up
-from scapy.layers.inet import DestIPField, UDP
+from scapy.utils import valid_net, valid_net6
 
+from scapy.layers.l2 import SourceMACField
+from scapy.layers.inet import DestIPField, UDP
 
 _HSRP_OPCODES = {0: "Hello", 1: "Coup", 2: "Resign", 3: "Advertise"}
 _HSRP_STATES = {
@@ -35,7 +54,11 @@
 
 
 class HSRP(Packet):
-    name = "HSRP"
+    """
+    HSRP version 1
+    """
+
+    name = "HSRPv1"
     fields_desc = [
         ByteField("version", 0),
         ByteEnumField("opcode", 0, _HSRP_OPCODES),
@@ -45,15 +68,24 @@ class HSRP(Packet):
         ByteField("priority", 120),
         ByteField("group", 1),
         ByteField("reserved", 0),
-        StrFixedLenField("auth", b"cisco" + b"\00" * 3, 8),
-        IPField("virtualIP", "192.168.1.1")
+        StrFixedLenField("auth", b"cisco" + b"\x00" * 3, 8),
+        IPField("virtualIP", "192.168.1.1"),
     ]
 
     @classmethod
     def dispatch_hook(cls, _pkt=None, *args, **kargs):
-        if _pkt and len(_pkt) >= 2 and orb(_pkt[1:2]) == 3:
-            return HSRPAdvertise
-        return cls
+        """
+        Dissects proper version of HSRP packet
+        based on the first byte.
+        """
+        if _pkt:
+            if _pkt[0] == 0:
+                if len(_pkt) >= 2 and _pkt[1] == 3:
+                    return HSRPAdvertise
+                return HSRP
+            if _pkt[0] == 1:
+                return HSRPv2
+        return HSRP
 
     def guess_payload_class(self, payload):
         if self.underlayer.len > 28:
@@ -78,37 +110,145 @@ class HSRPAdvertise(Packet):
 
     @classmethod
     def dispatch_hook(cls, _pkt=None, *args, **kargs):
-        if _pkt and len(_pkt) >= 2 and orb(_pkt[1:2]) != 3:
+        if _pkt and len(_pkt) >= 2 and _pkt[1] != 3:
             return HSRP
         return cls
 
 
+class HSRPv2(Packet):
+    """
+    HSRP version 2.
+    """
+
+    name = "HSRPv2"
+    fields_desc = [
+        ByteEnumField("type", 1, {1: "Group state TLV"}),
+        ByteField("len", None),
+        ByteField("version", 2),
+        ByteEnumField(
+            "opcode", 0, {0: "Hello", 1: "Coup", 2: "Resign", 3: "Advertise"}
+        ),
+        ByteEnumField(
+            "state",
+            16,
+            {
+                0: "Initial",
+                1: "Learn",
+                2: "Listen",
+                4: "Speak",
+                8: "Standby",
+                16: "Active",
+            },
+        ),
+        ByteEnumField("ipVer", None, {4: "IPv4", 6: "IPv6"}),
+        XShortField("group", 1),
+        SourceMACField("identifier"),
+        IntField("priority", 100),
+        IntField("hellotime", 3000),
+        IntField("holdtime", 10000),
+        MultipleTypeField(
+            [
+                (
+                    IPField("virtualIP", "0.0.0.0"),
+                    (
+                        lambda p: p.ipVer == 4,
+                        lambda p, val: p.ipVer != 6 and (val is None or valid_net(val)),
+                    ),
+                ),
+                (
+                    IP6Field("virtualIP", "::"),
+                    (
+                        lambda p: p.ipVer == 6,
+                        lambda p, val: p.ipVer != 4
+                        and (val is None or valid_net6(val)),
+                    ),
+                ),
+            ],
+            StrField("virtualIP", None),  # By default
+        ),
+        # The virtualIP field's expected size is always the size of an IPv6
+        # address. If IPv4 is used, padding is required.
+        ConditionalField(
+            StrFixedLenField("padding", b"\x00" * 12, 12),
+            lambda pkt: valid_net(pkt.virtualIP),
+        ),
+    ]
+
+    def post_build(self, pkt, pay):
+        if self.ipVer is None:
+            ip_ver = 4
+
+            if valid_net6(self.virtualIP):
+                ip_ver = 6
+
+            pkt = pkt[:5] + struct.pack("B", ip_ver) + pkt[6:]
+
+        if self.len is None:
+            pkt = pkt[:1] + struct.pack("B", len(pkt) - 2) + pkt[2:]
+        return pkt + pay
+
+    def guess_payload_class(self, payload):
+        if payload:
+            hsrp_auth_payload_type = payload[0]
+            if hsrp_auth_payload_type == 3:
+                return HSRPv2TextAuth
+            elif hsrp_auth_payload_type == 4:
+                return HSRPmd5
+        return Packet.guess_payload_class(self, payload)
+
+
 class HSRPmd5(Packet):
+    """
+    MD5 Authentication header for HSRP (version 1 and 2).
+    """
+
     name = "HSRP MD5 Authentication"
     fields_desc = [
         ByteEnumField("type", 4, {4: "MD5 authentication"}),
         ByteField("len", None),
-        ByteEnumField("algo", 0, {1: "MD5"}),
+        ByteEnumField("algo", 1, {1: "MD5"}),
         ByteField("padding", 0x00),
         XShortField("flags", 0x00),
         SourceIPField("sourceip"),
         XIntField("keyid", 0x00),
-        StrFixedLenField("authdigest", b"\00" * 16, 16)]
+        StrFixedLenField("authdigest", b"\x00" * 16, 16),
+    ]
+
+    def post_build(self, pkt, pay):
+        if self.len is None:
+            pkt = pkt[:1] + struct.pack("B", len(pkt) - 2) + pkt[2:]
+        return pkt + pay
+
 
-    def post_build(self, p, pay):
-        if self.len is None and pay:
-            tmp_len = len(pay)
-            p = p[:1] + hex(tmp_len)[30:] + p[30:]
-        return p
+class HSRPv2TextAuth(Packet):
+    """
+    Default plain text authentication header.
+    This is only used with HSRP version 2.
+    """
+
+    name = "HSRP Authentication"
+    fields_desc = [
+        ByteEnumField("type", 3, {3: "Text Authentication TLV"}),
+        ByteField("len", 8),
+        StrFixedLenField("auth", b"cisco" + b"\x00" * 3, 8),
+    ]
 
 
 bind_bottom_up(UDP, HSRP, dport=1985)
 bind_bottom_up(UDP, HSRP, sport=1985)
-bind_bottom_up(UDP, HSRP, dport=2029)
-bind_bottom_up(UDP, HSRP, sport=2029)
+bind_bottom_up(UDP, HSRPv2, dport=1985)
+bind_bottom_up(UDP, HSRPv2, sport=1985)
+bind_bottom_up(UDP, HSRPv2, dport=2029)
+bind_bottom_up(UDP, HSRPv2, sport=2029)
+
 bind_layers(UDP, HSRP, dport=1985, sport=1985)
-bind_layers(UDP, HSRP, dport=2029, sport=2029)
+bind_layers(UDP, HSRPv2, dport=1985, sport=1985)  # HSRP v2 with IPv4
+bind_layers(UDP, HSRPv2, dport=2029, sport=2029)  # HSRP v2 with IPv6
+
 DestIPField.bind_addr(UDP, "224.0.0.2", dport=1985)
+DestIPField.bind_addr(UDP, "224.0.0.102", dport=1985)
+
 if conf.ipv6_enabled:
     from scapy.layers.inet6 import DestIP6Field
+
     DestIP6Field.bind_addr(UDP, "ff02::66", dport=2029)

test/scapy/layers/hsrp.uts

@@ -5,14 +5,16 @@
 ############
 + HSRP tests
 
-= HSRP - build & dissection
+= HSRP V1 - build & dissection
 defaddr = conf.route.route('0.0.0.0')[1]
+
 pkt = IP(raw(IP()/UDP(dport=1985, sport=1985)/HSRP()/HSRPmd5()))
 assert pkt[IP].dst == "224.0.0.2" and pkt[UDP].sport == pkt[UDP].dport == 1985
-assert pkt[HSRP].opcode == 0 and pkt[HSRP].state == 16
+assert pkt[HSRP].version == 0 and pkt[HSRP].opcode == 0 and pkt[HSRP].state == 16
+assert pkt[HSRP].hellotime == 3 and pkt[HSRP].holdtime == 10
 assert pkt[HSRPmd5].type == 4 and pkt[HSRPmd5].sourceip == defaddr
 
-= HSRP - Advertise build & dissection
+= HSRP V1 - Advertise build & dissection
 advertise_raw = b"\x00\x03\x00\x01\x00\x0e\x02\x00\x00\x00\x00\x01o\x00\x00\x00"
 pkt = HSRP(advertise_raw)
 assert isinstance(pkt, HSRPAdvertise)
@@ -32,3 +34,20 @@ assert raw(
         reserved2=0x6f000000,
     )
 ) == advertise_raw
+
+
+= HSRP V2 with IPv4 and MD5 Auth - build & dissection
+pkt = IP(raw(IP()/UDP(dport=1985, sport=1985)/HSRPv2(version = 2)/HSRPmd5()))
+assert pkt[IP].dst == "224.0.0.2" and pkt[UDP].sport == pkt[UDP].dport == 1985
+assert pkt[HSRPv2].ipVer == 4 and pkt[HSRPv2].padding == b"\x00" * 12
+assert pkt[HSRPv2].version == 2 and pkt[HSRPv2].opcode == 0 and pkt[HSRPv2].state == 16
+assert pkt[HSRPv2].hellotime == 3000 and pkt[HSRPv2].holdtime == 10000
+assert pkt[HSRPmd5].type == 4 and pkt[HSRPmd5].sourceip == defaddr
+
+= HSRP V2 with IPv6 and default text auth - build & dissection
+pkt = IPv6(raw(IPv6()/UDP(dport=2029, sport=2029)/HSRPv2(version = 2, virtualIP="FE80::01")/HSRPv2TextAuth()))
+assert pkt[IPv6].dst == "ff02::66" and pkt[UDP].sport == pkt[UDP].dport == 2029
+assert pkt[HSRPv2].ipVer == 6 and pkt[HSRPv2].padding == None
+assert pkt[HSRPv2].version == 2 and pkt[HSRPv2].opcode == 0 and pkt[HSRPv2].state == 16
+assert pkt[HSRPv2].hellotime == 3000 and pkt[HSRPv2].holdtime == 10000
+assert pkt[HSRPv2TextAuth].type == 3 and pkt[HSRPv2TextAuth].auth == b"cisco\x00\x00\x00"