scapy/contrib/ldp.py

# scapy.contrib.description = Label Distribution Protocol (LDP)
# scapy.contrib.status = loads

# http://git.savannah.gnu.org/cgit/ldpscapy.git/snapshot/ldpscapy-5285b81d6e628043df2a83301b292f24a95f0ba1.tar.gz

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

# Copyright (C) 2010 Florian Duraffourg

from __future__ import absolute_import
import struct

from scapy.packet import *
from scapy.fields import *
from scapy.ansmachine import *
from scapy.layers.inet import UDP
from scapy.layers.inet import TCP
from scapy.base_classes import Net
from scapy.modules.six.moves import range


# Guess payload
def guess_payload(p):
    LDPTypes = {
        0x0001: LDPNotification,
        0x0100: LDPHello,
        0x0200: LDPInit,
        0x0201: LDPKeepAlive,
        0x0300: LDPAddress,
        0x0301: LDPAddressWM,
        0x0400: LDPLabelMM,
        0x0401: LDPLabelReqM,
        0x0404: LDPLabelARM,
        0x0402: LDPLabelWM,
        0x0403: LDPLabelRelM,
        }
    type = struct.unpack("!H",p[0:2])[0]
    type = type & 0x7fff
    if type == 0x0001 and struct.unpack("!H",p[2:4])[0] > 20:
        return LDP
    if type in LDPTypes:
        return LDPTypes[type]
    else:
        return conf.raw_layer

## Fields ##

# 3.4.1. FEC TLV

class FecTLVField(StrField):
    islist=1
    def m2i(self, pkt, x):
        nbr = struct.unpack("!H",x[2:4])[0]
        used = 0
        x=x[4:]
        list=[]
        while x:
            #if x[0] == 1:
            #   list.append('Wildcard')
            #else:
            #mask=ord(x[8*i+3])
            #add=inet_ntoa(x[8*i+4:8*i+8])
            mask=ord(x[3])
            nbroctets = mask / 8
            if mask % 8:
                nbroctets += 1
            add=inet_ntoa(x[4:4+nbroctets]+b"\x00"*(4-nbroctets))
            list.append( (add, mask) )
            used += 4 + nbroctets
            x=x[4+nbroctets:]
        return list
    def i2m(self, pkt, x):
        if isinstance(x, str):
            return x
        s = b"\x01\x00"
        l = 0
        fec = ""
        for o in x:
            fec += b"\x02\x00\x01"
            # mask length
            fec += struct.pack("!B",o[1])
            # Prefix
            fec += inet_aton(o[0])
            l += 8
        s += struct.pack("!H",l)
        s += fec
        return s
    def size(self, s):
        """Get the size of this field"""
        l = 4 + struct.unpack("!H",s[2:4])[0]
        return l
    def getfield(self, pkt, s):
        l = self.size(s)
        return s[l:],self.m2i(pkt, s[:l])
        

# 3.4.2.1. Generic Label TLV

class LabelTLVField(StrField):
    def m2i(self, pkt, x):
        return struct.unpack("!I",x[4:8])[0]
    def i2m(self, pkt, x):
        if isinstance(x, str):
            return x
        s = b"\x02\x00\x00\x04"
        s += struct.pack("!I",x)
        return s
    def size(self, s):
        """Get the size of this field"""
        l = 4 + struct.unpack("!H",s[2:4])[0]
        return l
    def getfield(self, pkt, s):
        l = self.size(s)
        return s[l:],self.m2i(pkt, s[:l])


# 3.4.3. Address List TLV

class AddressTLVField(StrField):
    islist=1
    def m2i(self, pkt, x):
        nbr = struct.unpack("!H",x[2:4])[0] - 2
        nbr /= 4
        x=x[6:]
        list=[]
        for i in range(0, nbr):
            add = x[4*i:4*i+4]
            list.append(inet_ntoa(add))
        return list
    def i2m(self, pkt, x):
        if isinstance(x, str):
            return x
        l=2+len(x)*4
        s = b"\x01\x01"+struct.pack("!H",l)+b"\x00\x01"
        for o in x:
            s += inet_aton(o)
        return s
    def size(self, s):
        """Get the size of this field"""
        l = 4 + struct.unpack("!H",s[2:4])[0]
        return l
    def getfield(self, pkt, s):
        l = self.size(s)
        return s[l:],self.m2i(pkt, s[:l])


# 3.4.6. Status TLV

class StatusTLVField(StrField):
    islist=1
    def m2i(self, pkt, x):
        l = []
        statuscode = struct.unpack("!I",x[4:8])[0]
        l.append( (statuscode & 2**31) >> 31)
        l.append( (statuscode & 2**30) >> 30)
        l.append( statuscode & 0x3FFFFFFF )
        l.append( struct.unpack("!I", x[8:12])[0] )
        l.append( struct.unpack("!H", x[12:14])[0] )
        return l
    def i2m(self, pkt, x):
        if isinstance(x, str):
            return x
        s = b"\x03\x00" + struct.pack("!H",10)
        statuscode = 0
        if x[0] != 0:
            statuscode += 2**31
        if x[1] != 0:
            statuscode += 2**30
        statuscode += x[2]
        s += struct.pack("!I",statuscode)
        if len(x) > 3:
            s += struct.pack("!I",x[3])
        else:
            s += b"\x00\x00\x00\x00"
        if len(x) > 4:
            s += struct.pack("!H",x[4])
        else:
            s += b"\x00\x00"
        return s
    def getfield(self, pkt, s):
        l = 14
        return s[l:],self.m2i(pkt, s[:l])


# 3.5.2 Common Hello Parameters TLV
class CommonHelloTLVField(StrField):
    islist = 1
    def m2i(self, pkt, x):
        list = []
        v = struct.unpack("!H",x[4:6])[0]
        list.append(v)
        flags = struct.unpack("B",x[6])[0]
        v = ( flags & 0x80 ) >> 7
        list.append(v)
        v = ( flags & 0x40 ) >> 7
        list.append(v)
        return list
    def i2m(self, pkt, x):
        if isinstance(x, str):
            return x
        s = b"\x04\x00\x00\x04"
        s += struct.pack("!H",x[0])
        byte = 0
        if x[1] == 1:
            byte += 0x80
        if x[2] == 1:
            byte += 0x40
        s += struct.pack("!B",byte)
        s += b"\x00"
        return s
    def getfield(self, pkt, s):
        l = 8
        return s[l:],self.m2i(pkt, s[:l])


# 3.5.3 Common Session Parameters TLV
class CommonSessionTLVField(StrField):
    islist = 1
    def m2i(self, pkt, x):
        l = [struct.unpack("!H", x[6:8])[0]]
        octet = struct.unpack("B",x[8:9])[0]
        l.append( (octet & 2**7 ) >> 7 )
        l.append( (octet & 2**6 ) >> 6 )
        l.append( struct.unpack("B",x[9:10])[0] )
        l.append( struct.unpack("!H",x[10:12])[0] )
        l.append( inet_ntoa(x[12:16]) )
        l.append( struct.unpack("!H",x[16:18])[0] )
        return l
    def i2m(self, pkt, x):
        if isinstance(x, str):
            return x
        s = b"\x05\x00\x00\x0E\x00\x01"
        s += struct.pack("!H",x[0])
        octet = 0
        if x[1] != 0:
            octet += 2**7
        if x[2] != 0:
            octet += 2**6
        s += struct.pack("!B",octet)
        s += struct.pack("!B",x[3])
        s += struct.pack("!H",x[4])
        s += inet_aton(x[5])
        s += struct.pack("!H",x[6])
        return s
    def getfield(self, pkt, s):
        l = 18
        return s[l:],self.m2i(pkt, s[:l])
    


## Messages ##

# 3.5.1. Notification Message
class LDPNotification(Packet):
    name = "LDPNotification"
    fields_desc = [ BitField("u",0,1),
                    BitField("type", 0x0001, 15),
                    ShortField("len", None),
                    IntField("id", 0) ,
                    StatusTLVField("status",(0,0,0,0,0)) ]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay  
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.5.2. Hello Message
class LDPHello(Packet):
    name = "LDPHello"
    fields_desc = [ BitField("u",0,1),
                    BitField("type", 0x0100, 15),
                    ShortField("len", None),
                    IntField("id", 0) ,
                    CommonHelloTLVField("params",[180,0,0]) ]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay  
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.5.3. Initialization Message
class LDPInit(Packet):
    name = "LDPInit"
    fields_desc = [ BitField("u",0,1),
                    XBitField("type", 0x0200, 15),
                    ShortField("len", None),
                    IntField("id", 0),
                    CommonSessionTLVField("params",None)]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay  
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.5.4. KeepAlive Message
class LDPKeepAlive(Packet):
    name = "LDPKeepAlive"
    fields_desc = [ BitField("u",0,1),
                    XBitField("type", 0x0201, 15),
                    ShortField("len", None),
                    IntField("id", 0)]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay  
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.5.5. Address Message

class LDPAddress(Packet):
    name = "LDPAddress"
    fields_desc = [ BitField("u",0,1),
                    XBitField("type", 0x0300, 15),
                    ShortField("len", None),
                    IntField("id", 0),
                    AddressTLVField("address",None) ]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.5.6. Address Withdraw Message

class LDPAddressWM(Packet):
    name = "LDPAddressWM"
    fields_desc = [ BitField("u",0,1),
                    XBitField("type", 0x0301, 15),
                    ShortField("len", None),
                    IntField("id", 0),
                    AddressTLVField("address",None) ]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.5.7. Label Mapping Message

class LDPLabelMM(Packet):
    name = "LDPLabelMM"
    fields_desc = [ BitField("u",0,1),
                    XBitField("type", 0x0400, 15),
                    ShortField("len", None),
                    IntField("id", 0),
                    FecTLVField("fec",None),
                    LabelTLVField("label",0)]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay  
    def guess_payload_class(self, p):
        return guess_payload(p)

# 3.5.8. Label Request Message

class LDPLabelReqM(Packet):
    name = "LDPLabelReqM"
    fields_desc = [ BitField("u",0,1),
                    XBitField("type", 0x0401, 15),
                    ShortField("len", None),
                    IntField("id", 0),
                    FecTLVField("fec",None)]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay  
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.5.9. Label Abort Request Message

class LDPLabelARM(Packet):
    name = "LDPLabelARM"
    fields_desc = [ BitField("u",0,1),
                    XBitField("type", 0x0404, 15),
                    ShortField("len", None),
                    IntField("id", 0),
                    FecTLVField("fec",None),
                    IntField("labelRMid",0)]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay  
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.5.10. Label Withdraw Message

class LDPLabelWM(Packet):
    name = "LDPLabelWM"
    fields_desc = [ BitField("u",0,1),
                    XBitField("type", 0x0402, 15),
                    ShortField("len", None),
                    IntField("id", 0),
                    FecTLVField("fec",None),
                    LabelTLVField("label",0)]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay  
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.5.11. Label Release Message

class LDPLabelRelM(Packet):
    name = "LDPLabelRelM"
    fields_desc = [ BitField("u",0,1),
                    XBitField("type", 0x0403, 15),
                    ShortField("len", None),
                    IntField("id", 0),
                    FecTLVField("fec",None),
                    LabelTLVField("label",0)]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p) - 4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay  
    def guess_payload_class(self, p):
        return guess_payload(p)


# 3.1. LDP PDUs
class LDP(Packet):
    name = "LDP"
    fields_desc = [ ShortField("version",1),
                    ShortField("len", None),
                    IPField("id","127.0.0.1"),
                    ShortField("space",0) ]
    def post_build(self, p, pay):
        if self.len is None:
            l = len(p)+len(pay)-4
            p = p[:2]+struct.pack("!H", l)+p[4:]
        return p+pay
    def guess_payload_class(self, p):
        return guess_payload(p)

bind_layers( TCP, LDP, sport=646, dport=646 )
bind_layers( UDP, LDP, sport=646, dport=646 )