CHANGE: removed dedicated CLNS layer and replaced it with a better usable solution

CHANGE: made IS-IS more PEP-8 compliant

scapy/contrib/isis.py

@@ -5,7 +5,7 @@
     IS-IS Scapy Extension
     ~~~~~~~~~~~~~~~~~~~~~
 
-    :copyright: 2014 BENOCS GmbH, Berlin (Germany)
+    :copyright: 2014, 2015 BENOCS GmbH, Berlin (Germany)
     :author:    Marcel Patzlaff, mpatzlaff@benocs.com
     :license:   GPLv2
 
@@ -50,7 +50,7 @@ import struct
 from scapy.config import conf
 from scapy.fields import Field, FlagsField, ByteField, ByteEnumField, \
     FieldLenField, ShortField, PacketListField, XIntField, \
-    XShortField, ConditionalField, X3BytesField, IntField, BitField, IPField, \
+    XShortField, ConditionalField, ThreeBytesField, IntField, BitField, IPField, \
     FieldListField, MACField, BitFieldLenField, IPPrefixField, IP6PrefixField,\
     BoundStrLenField
 from scapy.layers.clns import network_layer_protocol_ids, register_cln_protocol
@@ -65,13 +65,14 @@ EXT_VERSION = "v0.0.1"
 
 conf.debug_dissector = True
 
+
 #######################################################################
 ##  ISIS Utilities + Fields                                          ##
 #######################################################################
-
 def isis_area2str(area):
     return "".join(x.decode("hex") for x in area.split("."))
 
+
 def isis_str2area(s):
     if len(s) == 0:
         return ""
@@ -80,21 +81,27 @@ def isis_str2area(s):
     fmt = "%02X" + (".%02X%02X" * (numbytes / 2)) + ("" if (numbytes % 2) == 0 else ".%02X")
     return fmt % tuple(map(ord, s))
 
+
 def isis_sysid2str(sysid):
     return "".join(x.decode("hex") for x in sysid.split("."))
 
+
 def isis_str2sysid(s):
     return ("%02X%02X."*3)[:-1] % tuple(map(ord, s))
 
+
 def isis_nodeid2str(nodeid):
     return "%s%s" % (isis_sysid2str(nodeid[:-3]), nodeid[-2:].decode("hex"))
 
+
 def isis_str2nodeid(s):
     return "%s.%02X" % (isis_str2sysid(s[:-1]), ord(s[-1]))
 
+
 def isis_lspid2str(lspid):
     return "%s%s" % (isis_nodeid2str(lspid[:-3]), lspid[-2:].decode("hex"))
 
+
 def isis_str2lspid(s):
     return "%s-%02X" % (isis_str2nodeid(s[:-1]), ord(s[-1]))
 
@@ -138,6 +145,7 @@ class _ISIS_RandId(RandString):
 
         return self.format % val
 
+
 class _ISIS_RandAreaId(_ISIS_RandId):
     def __init__(self, bytecount= None):
         self.bytecount = random.randint(1, 13) if bytecount is None else bytecount
@@ -199,14 +207,14 @@ class ISIS_LspIdField(_ISIS_IdFieldBase):
 
 
 class ISIS_CircuitTypeField(FlagsField):
-    def __init__(self, name="circuittype", default=2, size=8, names=["L1", "L2", "r0", "r1", "r2", "r3", "r4", "r5"]):
+    def __init__(self, name="circuittype", default=2, size=8,
+                 names=["L1", "L2", "r0", "r1", "r2", "r3", "r4", "r5"]):
         FlagsField.__init__(self, name, default, size, names)
 
 
 #######################################################################
 ##  ISIS TLVs                                                        ##
 #######################################################################
-
 _isis_tlv_classes = { 
     1: "ISIS_AreaTlv",
     2: "ISIS_IsReachabilityTlv",
@@ -275,7 +283,6 @@ _isis_tlv_names= {
 }
 
 
-
 def _ISIS_GuessTlvClass(p, **kargs):
     cls = conf.raw_layer
     if len(p) >= 2:
@@ -327,7 +334,6 @@ class ISIS_BufferSizeTlv(ISIS_GenericTlv):
                    ShortField("lspbuffersize", 1497)]
 
 
-
 class ISIS_ChecksumTlv(ISIS_GenericTlv):
     name = "ISIS Optional Checksum TLV"
     fields_desc = [ByteEnumField("type", 12, _isis_tlv_names),
@@ -342,7 +348,6 @@ class ISIS_DynamicHostnameTlv(ISIS_GenericTlv):
                    BoundStrLenField("hostname", "", length_from=lambda pkt: pkt.len)]
 
 
-
 class ISIS_GenericSubTlv(Packet):
     name = "ISIS Generic Sub-TLV"
     fields_desc = [ByteField("type", 0),
@@ -352,23 +357,27 @@ class ISIS_GenericSubTlv(Packet):
     def guess_payload_class(self, p):
         return conf.padding_layer
 
+
 def _isis_guess_subtlv_cls(p, **kargs):
     return ISIS_GenericSubTlv(p, **kargs)
 
 
 class ISIS_ExtendedIpPrefix(Packet):
     name = "ISIS Extended IP Prefix"
-    fields_desc = [IntField("metric", 1),
+    fields_desc = [
+        IntField("metric", 1),
         BitField("updown", 0, 1),
         BitField("subtlvindicator", 0, 1),
         BitFieldLenField("pfxlen", None, 6, length_of="pfx"),
         IPPrefixField("pfx", None, wordbytes=1, length_from=lambda x: x.pfxlen),
         ConditionalField(FieldLenField("subtlvslen", None, length_of=lambda x: x.subtlvs, fmt= "B"), lambda pkt: pkt.subtlvindicator == 1), 
-                   ConditionalField(PacketListField("subtlvs", [], _isis_guess_subtlv_cls, length_from=lambda x: x.subtlvslen), lambda pkt: pkt.subtlvindicator == 1)]
+        ConditionalField(PacketListField("subtlvs", [], _isis_guess_subtlv_cls, length_from=lambda x: x.subtlvslen), lambda pkt: pkt.subtlvindicator == 1)
+    ]
 
     def extract_padding(self, s):
         return "", s
 
+ 
 class ISIS_ExtendedIpReachabilityTlv(ISIS_GenericTlv):
     name = "ISIS Extended IP Reachability TLV"
     fields_desc = [ByteEnumField("type", 135, _isis_tlv_names),
@@ -379,13 +388,14 @@ class ISIS_ExtendedIpReachabilityTlv(ISIS_GenericTlv):
 class ISIS_ExtendedIsNeighbourEntry(Packet):
     name = "ISIS Extended IS Neighbour Entry"
     fields_desc = [ISIS_NodeIdField("neighbourid", "0102.0304.0506.07"),
-                   X3BytesField("metric", 1),
+                   ThreeBytesField("metric", 1),
                    FieldLenField("subtlvslen", None, length_of="subtlvs", fmt= "B"),
                    ConditionalField(PacketListField("subtlvs", [], _isis_guess_subtlv_cls, length_from=lambda x: x.subtlvslen), lambda pkt: pkt.subtlvslen > 0)]
 
     def extract_padding(self, s):
         return "", s
 
+    
 class ISIS_ExtendedIsReachabilityTlv(ISIS_GenericTlv):
     name = "ISIS Extended IS Reachability TLV"
     fields_desc = [ByteEnumField("type", 22, _isis_tlv_names),
@@ -402,15 +412,17 @@ class ISIS_IpInterfaceAddressTlv(ISIS_GenericTlv):
 
 class ISIS_Ipv6InterfaceAddressTlv(ISIS_GenericTlv):
     name = "ISIS IPv6 Interface Address TLV"
-    fields_desc = [ByteEnumField("type", 232, _isis_tlv_names),
+    fields_desc = [
+        ByteEnumField("type", 232, _isis_tlv_names),
         FieldLenField("len", None, length_of="addresses", fmt="B"),
-                   IP6ListField("addresses", [], count_from= lambda pkt: pkt.len / 16)]
-
+        IP6ListField("addresses", [], count_from=lambda pkt: pkt.len / 16)
+    ]
 
 
 class ISIS_Ipv6Prefix(Packet):
     name = "ISIS IPv6 Prefix"
-    fields_desc = [IntField("metric", 1),
+    fields_desc = [
+        IntField("metric", 1),
         BitField("updown", 0, 1),
         BitField("external", 0, 1),
         BitField("subtlvindicator", 0, 1),
@@ -418,17 +430,20 @@ class ISIS_Ipv6Prefix(Packet):
         FieldLenField("pfxlen", None, length_of="pfx", fmt="B"),
         IP6PrefixField("pfx", None, wordbytes=1, length_from=lambda x: x.pfxlen),
         ConditionalField(FieldLenField("subtlvslen", None, length_of=lambda x: x.subtlvs, fmt= "B"), lambda pkt: pkt.subtlvindicator == 1), 
-               ConditionalField(PacketListField("subtlvs", [], _isis_guess_subtlv_cls, length_from=lambda x: x.subtlvslen), lambda pkt: pkt.subtlvindicator == 1)]
+        ConditionalField(PacketListField("subtlvs", [], _isis_guess_subtlv_cls, length_from=lambda x: x.subtlvslen), lambda pkt: pkt.subtlvindicator == 1)
+    ]
 
     def extract_padding(self, s):
         return "", s
 
+
 class ISIS_Ipv6ReachabilityTlv(ISIS_GenericTlv):
     name= "ISIS IPv6 Reachability TLV"
     fields_desc = [ByteEnumField("type", 236, _isis_tlv_names),
                    FieldLenField("len", None, length_of= "pfxs", fmt="B"),
                    PacketListField("pfxs", [], ISIS_Ipv6Prefix, length_from= lambda pkt: pkt.len)]
 
+
 class ISIS_IsNeighbourTlv(ISIS_GenericTlv):
     name = "ISIS IS Neighbour TLV"
     fields_desc = [ByteEnumField("type", 6, _isis_tlv_names),
@@ -449,9 +464,11 @@ class ISIS_LspEntry(Packet):
 
 class ISIS_LspEntryTlv(ISIS_GenericTlv):
     name = "ISIS LSP Entry TLV"
-    fields_desc = [ByteEnumField("type", 9, _isis_tlv_names),
+    fields_desc = [
+        ByteEnumField("type", 9, _isis_tlv_names),
         FieldLenField("len", None, length_of="entries", fmt="B"),
-               PacketListField("entries", [], ISIS_LspEntry, count_from= lambda pkt: pkt.len / 16)]
+        PacketListField("entries", [], ISIS_LspEntry, count_from=lambda pkt: pkt.len / 16)
+    ]
 
 
 class _AdjacencyStateTlvLenField(Field):
@@ -467,6 +484,7 @@ class _AdjacencyStateTlvLenField(Field):
 
         return 1
 
+
 class ISIS_P2PAdjacencyStateTlv(ISIS_GenericTlv):
     name = "ISIS P2P Adjacency State TLV"
     fields_desc = [ByteEnumField("type", 240, _isis_tlv_names),
@@ -476,19 +494,24 @@ class ISIS_P2PAdjacencyStateTlv(ISIS_GenericTlv):
                ConditionalField(ISIS_SystemIdField("neighboursystemid", None), lambda pkt: pkt.len >= 11),
                ConditionalField(IntField("neighbourextlocalcircuitid", None), lambda pkt: pkt.len == 15)]
 
+
 # TODO dynamically allocate sufficient size
 class ISIS_PaddingTlv(ISIS_GenericTlv):
     name = "ISIS Padding TLV"
-    fields_desc = [ByteEnumField("type", 8, _isis_tlv_names),
+    fields_desc = [
+        ByteEnumField("type", 8, _isis_tlv_names),
         FieldLenField("len", None, length_of="padding", fmt="B"),
-               BoundStrLenField("padding", "", length_from=lambda pkt: pkt.len)]
+        BoundStrLenField("padding", "", length_from=lambda pkt: pkt.len)
+    ]
 
 
 class ISIS_ProtocolsSupportedTlv(ISIS_GenericTlv):
     name = "ISIS Protocols Supported TLV"
-    fields_desc = [ByteEnumField("type", 129, _isis_tlv_names),
+    fields_desc = [
+        ByteEnumField("type", 129, _isis_tlv_names),
         FieldLenField("len", None, count_of="nlpids", fmt="B"),
-                   FieldListField("nlpids", [], ByteEnumField("", "IPv4", network_layer_protocol_ids), count_from= lambda pkt: pkt.len)]
+        FieldListField("nlpids", [], ByteEnumField("", "IPv4", network_layer_protocol_ids), count_from=lambda pkt: pkt.len)
+    ]
 
 
 #######################################################################
@@ -507,17 +530,23 @@ class ISIS_IpReachabilityEntry(Packet):
     def extract_padding(self, s):
         return "", s
 
+
 class ISIS_InternalIpReachabilityTlv(ISIS_GenericTlv):
     name = "ISIS Internal IP Reachability TLV"
-    fields_desc = [ByteEnumField("type", 128, _isis_tlv_names),
+    fields_desc = [
+        ByteEnumField("type", 128, _isis_tlv_names),
         FieldLenField("len", None, length_of="entries", fmt="B"),
-                   PacketListField("entries", [], ISIS_IpReachabilityEntry, count_from= lambda x: x.len / 12)]
+        PacketListField("entries", [], ISIS_IpReachabilityEntry, count_from=lambda x: x.len / 12)
+    ]
+
 
 class ISIS_ExternalIpReachabilityTLV(ISIS_GenericTlv):
     name = "ISIS External IP Reachability TLV"
-    fields_desc = [ByteEnumField("type", 130, _isis_tlv_names),
+    fields_desc = [
+        ByteEnumField("type", 130, _isis_tlv_names),
         FieldLenField("len", None, length_of="entries", fmt="B"),
-                   PacketListField("entries", [], ISIS_IpReachabilityEntry, count_from= lambda x: x.len / 12)]
+        PacketListField("entries", [], ISIS_IpReachabilityEntry, count_from=lambda x: x.len / 12)
+    ]
 
 
 class ISIS_IsReachabilityEntry(Packet):
@@ -531,13 +560,15 @@ class ISIS_IsReachabilityEntry(Packet):
     def extract_padding(self, s):
         return "", s
 
+
 class ISIS_IsReachabilityTlv(ISIS_GenericTlv):
     name = "ISIS IS Reachability TLV"
-    fields_desc = [ByteEnumField("type", 2, _isis_tlv_names),
+    fields_desc = [
+        ByteEnumField("type", 2, _isis_tlv_names),
         FieldLenField("len", None, fmt="B", length_of="neighbours", adjust=lambda pkt,x: x+1),
         ByteField("virtual", 0),
-                   PacketListField("neighbours", [], ISIS_IsReachabilityEntry, count_from= lambda x: (x.len - 1) / 11)]
-
+        PacketListField("neighbours", [], ISIS_IsReachabilityEntry, count_from=lambda x: (x.len - 1) / 11)
+    ]
 
 #######################################################################
 ##  ISIS PDU Packets                                                 ##
@@ -554,6 +585,7 @@ _isis_pdu_names = {
     27: "L2 PSNP"
 }
 
+
 class ISIS_CommonHdr(Packet):
     name = "ISIS Common Header"
     fields_desc = [
@@ -605,6 +637,7 @@ class _ISIS_TlvListField(PacketListField):
     def __init__(self):
         PacketListField.__init__(self, "tlvs", [], _ISIS_GuessTlvClass, count_from= None, length_from= lambda pkt: pkt.pdulength - pkt.underlayer.hdrlen)
 
+
 class _ISIS_LAN_HelloBase(_ISIS_PduBase):
     fields_desc = [
         ISIS_CircuitTypeField(),
@@ -616,9 +649,11 @@ class _ISIS_LAN_HelloBase(_ISIS_PduBase):
         _ISIS_TlvListField()
     ]
 
+
 class ISIS_L1_LAN_Hello(_ISIS_LAN_HelloBase):
     name = "ISIS L1 LAN Hello PDU"
 
+
 class ISIS_L2_LAN_Hello(_ISIS_LAN_HelloBase):
     name = "ISIS L2 LAN Hello PDU"
 
@@ -653,16 +688,19 @@ class _ISIS_LSP_Base(_ISIS_PduBase):
 
         return (12, 24)
 
+
 def _lsp_answers(lsp, other, clsname):
     # TODO
     return 0
 
+
 class ISIS_L1_LSP(_ISIS_LSP_Base):
     name = "ISIS L1 Link State PDU"
 
     def answers(self, other):
         return _lsp_answers(self, other, "ISIS_L1_PSNP")
 
+
 class ISIS_L2_LSP(_ISIS_LSP_Base):
     name = "ISIS L2 Link State PDU"
 
@@ -679,22 +717,26 @@ class _ISIS_CSNP_Base(_ISIS_PduBase):
         _ISIS_TlvListField()
     ]
 
+
 def _snp_answers(snp, other, clsname):
     # TODO
     return 0
 
+
 class ISIS_L1_CSNP(_ISIS_CSNP_Base):
     name = "ISIS L1 Complete Sequence Number Packet"
 
     def answers(self, other):
         return _snp_answers(self, other, "ISIS_L1_LSP")
 
+
 class ISIS_L2_CSNP(_ISIS_CSNP_Base):
     name = "ISIS L2 Complete Sequence Number Packet"
 
     def answers(self, other):
         return _snp_answers(self, other, "ISIS_L2_LSP")
 
+
 class _ISIS_PSNP_Base(_ISIS_PduBase):
     fields_desc = [
         _ISIS_PduLengthField(),
@@ -702,12 +744,14 @@ class _ISIS_PSNP_Base(_ISIS_PduBase):
         _ISIS_TlvListField()
     ]
 
+
 class ISIS_L1_PSNP(_ISIS_PSNP_Base):
     name = "ISIS L1 Partial Sequence Number Packet"
 
     def answers(self, other):
         return _snp_answers(self, other, "ISIS_L1_LSP")
 
+
 class ISIS_L2_PSNP(_ISIS_PSNP_Base):
     name = "ISIS L2 Partial Sequence Number Packet"
 

scapy/layers/clns.py

@@ -2,7 +2,7 @@
     CLNS Extension
     ~~~~~~~~~~~~~~~~~~~~~
 
-    :copyright: 2014 BENOCS GmbH, Berlin (Germany)
+    :copyright: 2014, 2015 BENOCS GmbH, Berlin (Germany)
     :author:    Marcel Patzlaff, mpatzlaff@benocs.com
     :license:   GPLv2
 
@@ -18,22 +18,15 @@
 
     :description:
 
-        This module provides a layer and registration function for
-        OSI Connectionless-mode Network Services (such as IS-IS).
-                
-    :TODO:
-
-        - rework this if a better way is found/implemented to bind
-          protocols such as IS-IS (or if IS-IS remains the sole CLN
-          protocol)
-
+        This module provides a registration function and a generic PDU
+        for OSI Connectionless-mode Network Services (such as IS-IS).
 """
 import struct
 
 from scapy.config import conf
 from scapy.fields import ByteEnumField, PacketField
 from scapy.layers.l2 import LLC
-from scapy.packet import Packet, bind_layers
+from scapy.packet import Packet, bind_top_down, bind_bottom_up
 
 
 network_layer_protocol_ids = {
@@ -55,6 +48,7 @@ network_layer_protocol_ids= {
 
 _cln_protocols = {}
 
+
 class _GenericClnsPdu(Packet):
     name = "Generic CLNS PDU"
     fields_desc = [
@@ -63,24 +57,28 @@ class _GenericClnsPdu(Packet):
     ]
 
 
-class ConnectionlessNetworkService(Packet):
-    name= "Connectionless-mode Network Service"
+def _create_cln_pdu(s, **kwargs):
+    pdu_cls = conf.raw_layer
 
-    def guess_payload_class(self, p):
-        cls= conf.raw_layer
+    if len(s) >= 1:
+        nlpid = struct.unpack("!B", s[0])[0]
+        pdu_cls = _cln_protocols.get(nlpid, _GenericClnsPdu)
 
-        if len(p) >= 1:
-            nlpid = struct.unpack("!B", p[0])[0]
-            cls= _cln_protocols.get(nlpid, _GenericClnsPdu)
+    return pdu_cls(s, **kwargs)
 
-        return cls
 
 @conf.commands.register
 def register_cln_protocol(nlpid, cln_protocol_class):
     if nlpid is None or cln_protocol_class is None:
         return
 
+    chk = _cln_protocols.get(nlpid, None)
+    if chk is not None and chk != cln_protocol_class:
+        raise ValueError("different protocol already registered!")
+
     _cln_protocols[nlpid] = cln_protocol_class
+    bind_top_down(LLC, cln_protocol_class, dsap=0xfe, ssap=0xfe, ctrl=3)
 
 
-bind_layers(LLC, ConnectionlessNetworkService, dsap=0xfe, ssap=0xfe, ctrl=3)
+bind_top_down(LLC, _GenericClnsPdu, dsap=0xfe, ssap=0xfe, ctrl=3)
\ No newline at end of file
+bind_bottom_up(LLC, _create_cln_pdu, dsap=0xfe, ssap=0xfe, ctrl=3)