From 9b0c2d1b31bbc550ce183f8c4f3546c5920921f4 Mon Sep 17 00:00:00 2001 From: ErwanLF Date: Wed, 9 Dec 2020 21:49:09 +0100 Subject: [PATCH] Added support for HSRP version 2 and updated HSRP unit test. AI-Assisted: no --- scapy/layers/hsrp.py | 186 ++++++++++++++++++++++++++++++++----- test/scapy/layers/hsrp.uts | 25 ++++- 2 files changed, 185 insertions(+), 26 deletions(-) diff --git a/scapy/layers/hsrp.py b/scapy/layers/hsrp.py index b64b9425d16..341ea206fbd 100644 --- a/scapy/layers/hsrp.py +++ b/scapy/layers/hsrp.py @@ -3,23 +3,42 @@ # See https://scapy.net/ for more information # Copyright (C) Mathieu RENARD +# 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) diff --git a/test/scapy/layers/hsrp.uts b/test/scapy/layers/hsrp.uts index 00bbbb7340d..e0bd1676448 100644 --- a/test/scapy/layers/hsrp.uts +++ b/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"