<s> import abc <EOL> import six <EOL> import struct <EOL> from . import packet_base <EOL> from . import icmpv6 <EOL> from . import tcp <EOL> from . import udp <EOL> from . import sctp <EOL> from . import in_proto as inet <EOL> from ryu . lib import addrconv <EOL> from ryu . lib import stringify <EOL> IPV6_ADDRESS_PACK_STR = '<STR_LIT>' <EOL> IPV6_ADDRESS_LEN = struct . calcsize ( IPV6_ADDRESS_PACK_STR ) <EOL> IPV6_PSEUDO_HEADER_PACK_STR = '<STR_LIT>' <EOL> class ipv6 ( packet_base . PacketBase ) : <EOL> """<STR_LIT>""" <EOL> _PACK_STR = '<STR_LIT>' <EOL> _MIN_LEN = struct . calcsize ( _PACK_STR ) <EOL> _IPV6_EXT_HEADER_TYPE = { } <EOL> _TYPE = { <EOL> '<STR_LIT:ascii>' : [ <EOL> '<STR_LIT:src>' , '<STR_LIT>' <EOL> ] <EOL> } <EOL> @ staticmethod <EOL> def register_header_type ( type_ ) : <EOL> def _register_header_type ( cls ) : <EOL> ipv6 . _IPV6_EXT_HEADER_TYPE [ type_ ] = cls <EOL> return cls <EOL> return _register_header_type <EOL> def __init__ ( self , version = <NUM_LIT:6> , traffic_class = <NUM_LIT:0> , flow_label = <NUM_LIT:0> , <EOL> payload_length = <NUM_LIT:0> , nxt = inet . IPPROTO_TCP , hop_limit = <NUM_LIT:255> , <EOL> src = '<STR_LIT>' , dst = '<STR_LIT>' , ext_hdrs = None ) : <EOL> super ( ipv6 , self ) . __init__ ( ) <EOL> self . version = version <EOL> self . traffic_class = traffic_class <EOL> self . flow_label = flow_label <EOL> self . payload_length = payload_length <EOL> self . nxt = nxt <EOL> self . hop_limit = hop_limit <EOL> self . src = src <EOL> self . dst = dst <EOL> ext_hdrs = ext_hdrs or [ ] <EOL> assert isinstance ( ext_hdrs , list ) <EOL> for ext_hdr in ext_hdrs : <EOL> assert isinstance ( ext_hdr , header ) <EOL> self . ext_hdrs = ext_hdrs <EOL> @ classmethod <EOL> def parser ( cls , buf ) : <EOL> ( v_tc_flow , payload_length , nxt , hlim , src , dst ) = struct . unpack_from ( <EOL> cls . _PACK_STR , buf ) <EOL> version = v_tc_flow >> <NUM_LIT> <EOL> traffic_class = ( v_tc_flow >> <NUM_LIT:20> ) & <NUM_LIT> <EOL> flow_label = v_tc_flow & <NUM_LIT> <EOL> hop_limit = hlim <EOL> offset = cls . _MIN_LEN <EOL> last = nxt <EOL> ext_hdrs = [ ] <EOL> while True : <EOL> cls_ = cls . _IPV6_EXT_HEADER_TYPE . get ( last ) <EOL> if not cls_ : <EOL> break <EOL> hdr = cls_ . parser ( buf [ offset : ] ) <EOL> ext_hdrs . append ( hdr ) <EOL> offset += len ( hdr ) <EOL> last = hdr . nxt <EOL> msg = cls ( version , traffic_class , flow_label , payload_length , <EOL> nxt , hop_limit , addrconv . ipv6 . bin_to_text ( src ) , <EOL> addrconv . ipv6 . bin_to_text ( dst ) , ext_hdrs ) <EOL> return ( msg , ipv6 . get_packet_type ( last ) , <EOL> buf [ offset : offset + payload_length ] ) <EOL> def serialize ( self , payload , prev ) : <EOL> hdr = bytearray ( <NUM_LIT> ) <EOL> v_tc_flow = ( self . version << <NUM_LIT> | self . traffic_class << <NUM_LIT:20> | <EOL> self . flow_label ) <EOL> struct . pack_into ( ipv6 . _PACK_STR , hdr , <NUM_LIT:0> , v_tc_flow , <EOL> self . payload_length , self . nxt , self . hop_limit , <EOL> addrconv . ipv6 . text_to_bin ( self . src ) , <EOL> addrconv . ipv6 . text_to_bin ( self . dst ) ) <EOL> if self . ext_hdrs : <EOL> for ext_hdr in self . ext_hdrs : <EOL> hdr . extend ( ext_hdr . serialize ( ) ) <EOL> if <NUM_LIT:0> == self . payload_length : <EOL> payload_length = len ( payload ) <EOL> for ext_hdr in self . ext_hdrs : <EOL> payload_length += len ( ext_hdr ) <EOL> self . payload_length = payload_length <EOL> struct . pack_into ( '<STR_LIT>' , hdr , <NUM_LIT:4> , self . payload_length ) <EOL> return hdr <EOL> def __len__ ( self ) : <EOL> ext_hdrs_len = <NUM_LIT:0> <EOL> for ext_hdr in self . ext_hdrs : <EOL> ext_hdrs_len += len ( ext_hdr ) <EOL> return self . _MIN_LEN + ext_hdrs_len <EOL> ipv6 . register_packet_type ( icmpv6 . icmpv6 , inet . IPPROTO_ICMPV6 ) <EOL> ipv6 . register_packet_type ( tcp . tcp , inet . IPPROTO_TCP ) <EOL> ipv6 . register_packet_type ( udp . udp , inet . IPPROTO_UDP ) <EOL> ipv6 . register_packet_type ( sctp . sctp , inet . IPPROTO_SCTP ) <EOL> @ six . add_metaclass ( abc . ABCMeta ) <EOL> class header ( stringify . StringifyMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , nxt ) : <EOL> self . nxt = nxt <EOL> @ classmethod <EOL> @ abc . abstractmethod <EOL> def parser ( cls , buf ) : <EOL> pass <EOL> @ abc . abstractmethod <EOL> def serialize ( self ) : <EOL> pass <EOL> @ abc . abstractmethod <EOL> def __len__ ( self ) : <EOL> pass <EOL> class opt_header ( header ) : <EOL> """<STR_LIT>""" <EOL> _PACK_STR = '<STR_LIT>' <EOL> _MIN_LEN = struct . calcsize ( _PACK_STR ) <EOL> _FIX_SIZE = <NUM_LIT:8> <EOL> _class_prefixes = [ '<STR_LIT>' ] <EOL> @ abc . abstractmethod <EOL> def __init__ ( self , nxt , size , data ) : <EOL> super ( opt_header , self ) . __init__ ( nxt ) <EOL> assert not ( size % <NUM_LIT:8> ) <EOL> self . size = size <EOL> self . data = data <EOL> @ classmethod <EOL> def parser ( cls , buf ) : <EOL> ( nxt , len_ ) = struct . unpack_from ( cls . _PACK_STR , buf ) <EOL> data_len = cls . _FIX_SIZE + int ( len_ ) <EOL> data = [ ] <EOL> size = cls . _MIN_LEN <EOL> while size < data_len : <EOL> ( type_ , ) = struct . unpack_from ( '<STR_LIT>' , buf [ size : ] ) <EOL> if type_ == <NUM_LIT:0> : <EOL> opt = option ( type_ , - <NUM_LIT:1> , None ) <EOL> size += <NUM_LIT:1> <EOL> else : <EOL> opt = option . parser ( buf [ size : ] ) <EOL> size += len ( opt ) <EOL> data . append ( opt ) <EOL> return cls ( nxt , len_ , data ) <EOL> def serialize ( self ) : <EOL> buf = struct . pack ( self . _PACK_STR , self . nxt , self . size ) <EOL> buf = bytearray ( buf ) <EOL> if self . data is None : <EOL> self . data = [ option ( type_ = <NUM_LIT:1> , len_ = <NUM_LIT:4> , <EOL> data = b'<STR_LIT>' ) ] <EOL> for opt in self . data : <EOL> buf . extend ( opt . serialize ( ) ) <EOL> return buf <EOL> def __len__ ( self ) : <EOL> return self . _FIX_SIZE + self . size <EOL> @ ipv6 . register_header_type ( inet . IPPROTO_HOPOPTS ) <EOL> class hop_opts ( opt_header ) : <EOL> """<STR_LIT>""" <EOL> TYPE = inet . IPPROTO_HOPOPTS <EOL> def __init__ ( self , nxt = inet . IPPROTO_TCP , size = <NUM_LIT:0> , data = None ) : <EOL> super ( hop_opts , self ) . __init__ ( nxt , size , data ) <EOL> @ ipv6 . register_header_type ( inet . IPPROTO_DSTOPTS ) <EOL> class dst_opts ( opt_header ) : <EOL> """<STR_LIT>""" <EOL> TYPE = inet . IPPROTO_DSTOPTS <EOL> def __init__ ( self , nxt = inet . IPPROTO_TCP , size = <NUM_LIT:0> , data = None ) : <EOL> super ( dst_opts , self ) . __init__ ( nxt , size , data ) <EOL> class option ( stringify . StringifyMixin ) : <EOL> """<STR_LIT>""" <EOL> _PACK_STR = '<STR_LIT>' <EOL> _MIN_LEN = struct . calcsize ( _PACK_STR ) <EOL> def __init__ ( self , type_ = <NUM_LIT:0> , len_ = - <NUM_LIT:1> , data = None ) : <EOL> self . type_ = type_ <EOL> self . len_ = len_ <EOL> self . data = data <EOL> @ classmethod <EOL> def parser ( cls , buf ) : <EOL> ( type_ , ) = struct . unpack_from ( '<STR_LIT>' , buf ) <EOL> if not type_ : <EOL> cls_ = cls ( type_ , - <NUM_LIT:1> , None ) <EOL> else : <EOL> data = None <EOL> ( type_ , len_ ) = struct . unpack_from ( cls . _PACK_STR , buf ) <EOL> if len_ : <EOL> form = "<STR_LIT>" % len_ <EOL> ( data , ) = struct . unpack_from ( form , buf , cls . _MIN_LEN ) <EOL> cls_ = cls ( type_ , len_ , data ) <EOL> return cls_ <EOL> def serialize ( self ) : <EOL> data = None <EOL> if not self . type_ : <EOL> data = struct . pack ( '<STR_LIT>' , self . type_ ) <EOL> elif not self . len_ : <EOL> data = struct . pack ( self . _PACK_STR , self . type_ , self . len_ ) <EOL> else : <EOL> form = "<STR_LIT>" % self . len_ <EOL> data = struct . pack ( self . _PACK_STR + form , self . type_ , <EOL> self . len_ , self . data ) <EOL> return data <EOL> def __len__ ( self ) : <EOL> return self . _MIN_LEN + self . len_ <EOL> @ ipv6 . register_header_type ( inet . IPPROTO_ROUTING ) <EOL> class routing ( header ) : <EOL> """<STR_LIT>""" <EOL> TYPE = inet . IPPROTO_ROUTING <EOL> _OFFSET_LEN = struct . calcsize ( '<STR_LIT>' ) <EOL> ROUTING_TYPE_2 = <NUM_LIT> <EOL> ROUTING_TYPE_3 = <NUM_LIT> <EOL> @ classmethod <EOL> def parser ( cls , buf ) : <EOL> ( type_ , ) = struct . unpack_from ( '<STR_LIT>' , buf , cls . _OFFSET_LEN ) <EOL> switch = { <EOL> cls . ROUTING_TYPE_2 : None , <EOL> cls . ROUTING_TYPE_3 : routing_type3 <EOL> } <EOL> cls_ = switch . get ( type_ ) <EOL> if cls_ : <EOL> return cls_ . parser ( buf ) <EOL> else : <EOL> return None <EOL> class routing_type3 ( header ) : <EOL> """<STR_LIT>""" <EOL> _PACK_STR = '<STR_LIT>' <EOL> _MIN_LEN = struct . calcsize ( _PACK_STR ) <EOL> _TYPE = { <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' <EOL> ] <EOL> } <EOL> def __init__ ( self , nxt = inet . IPPROTO_TCP , size = <NUM_LIT:0> , <EOL> type_ = <NUM_LIT:3> , seg = <NUM_LIT:0> , cmpi = <NUM_LIT:0> , cmpe = <NUM_LIT:0> , adrs = None ) : <EOL> super ( routing_type3 , self ) . __init__ ( nxt ) <EOL> self . size = size <EOL> self . type_ = type_ <EOL> self . seg = seg <EOL> self . cmpi = cmpi <EOL> self . cmpe = cmpe <EOL> adrs = adrs or [ ] <EOL> assert isinstance ( adrs , list ) <EOL> self . adrs = adrs <EOL> self . _pad = ( <NUM_LIT:8> - ( ( len ( self . adrs ) - <NUM_LIT:1> ) * ( <NUM_LIT:16> - self . cmpi ) + <EOL> ( <NUM_LIT:16> - self . cmpe ) % <NUM_LIT:8> ) ) % <NUM_LIT:8> <EOL> @ classmethod <EOL> def _get_size ( cls , size ) : <EOL> return ( int ( size ) + <NUM_LIT:1> ) * <NUM_LIT:8> <EOL> @ classmethod <EOL> def parser ( cls , buf ) : <EOL> ( nxt , size , type_ , seg , cmp_ , pad ) = struct . unpack_from ( <EOL> cls . _PACK_STR , buf ) <EOL> data = cls . _MIN_LEN <EOL> header_len = cls . _get_size ( size ) <EOL> cmpi = int ( cmp_ >> <NUM_LIT:4> ) <EOL> cmpe = int ( cmp_ & <NUM_LIT> ) <EOL> pad = int ( pad >> <NUM_LIT:4> ) <EOL> adrs = [ ] <EOL> if size : <EOL> adrs_len_i = <NUM_LIT:16> - cmpi <EOL> adrs_len_e = <NUM_LIT:16> - cmpe <EOL> form_i = "<STR_LIT>" % adrs_len_i <EOL> form_e = "<STR_LIT>" % adrs_len_e <EOL> while data < ( header_len - ( adrs_len_e + pad ) ) : <EOL> ( adr , ) = struct . unpack_from ( form_i , buf [ data : ] ) <EOL> adr = ( b'<STR_LIT:\x00>' * cmpi ) + adr <EOL> adrs . append ( addrconv . ipv6 . bin_to_text ( adr ) ) <EOL> data += adrs_len_i <EOL> ( adr , ) = struct . unpack_from ( form_e , buf [ data : ] ) <EOL> adr = ( b'<STR_LIT:\x00>' * cmpe ) + adr <EOL> adrs . append ( addrconv . ipv6 . bin_to_text ( adr ) ) <EOL> return cls ( nxt , size , type_ , seg , cmpi , cmpe , adrs ) <EOL> def serialize ( self ) : <EOL> if self . size == <NUM_LIT:0> : <EOL> self . size = ( ( len ( self . adrs ) - <NUM_LIT:1> ) * ( <NUM_LIT:16> - self . cmpi ) + <EOL> ( <NUM_LIT:16> - self . cmpe ) + self . _pad ) // <NUM_LIT:8> <EOL> buf = struct . pack ( self . _PACK_STR , self . nxt , self . size , <EOL> self . type_ , self . seg , ( self . cmpi << <NUM_LIT:4> ) | self . cmpe , <EOL> self . _pad << <NUM_LIT:4> ) <EOL> buf = bytearray ( buf ) <EOL> if self . size : <EOL> form_i = "<STR_LIT>" % ( <NUM_LIT:16> - self . cmpi ) <EOL> form_e = "<STR_LIT>" % ( <NUM_LIT:16> - self . cmpe ) <EOL> slice_i = slice ( self . cmpi , <NUM_LIT:16> ) <EOL> slice_e = slice ( self . cmpe , <NUM_LIT:16> ) <EOL> for adr in self . adrs [ : - <NUM_LIT:1> ] : <EOL> buf . extend ( <EOL> struct . pack ( <EOL> form_i , addrconv . ipv6 . text_to_bin ( adr ) [ slice_i ] ) ) <EOL> buf . extend ( struct . pack ( <EOL> form_e , <EOL> addrconv . ipv6 . text_to_bin ( self . adrs [ - <NUM_LIT:1> ] ) [ slice_e ] ) ) <EOL> return buf <EOL> def __len__ ( self ) : <EOL> return routing_type3 . _get_size ( self . size ) <EOL> @ ipv6 . register_header_type ( inet . IPPROTO_FRAGMENT ) <EOL> class fragment ( header ) : <EOL> """<STR_LIT>""" <EOL> TYPE = inet . IPPROTO_FRAGMENT <EOL> _PACK_STR = '<STR_LIT>' <EOL> _MIN_LEN = struct . calcsize ( _PACK_STR ) <EOL> def __init__ ( self , nxt = inet . IPPROTO_TCP , offset = <NUM_LIT:0> , more = <NUM_LIT:0> , id_ = <NUM_LIT:0> ) : <EOL> super ( fragment , self ) . __init__ ( nxt ) <EOL> self . offset = offset <EOL> self . more = more <EOL> self . id_ = id_ <EOL> @ classmethod <EOL> def parser ( cls , buf ) : <EOL> ( nxt , off_m , id_ ) = struct . unpack_from ( cls . _PACK_STR , buf ) <EOL> offset = off_m >> <NUM_LIT:3> <EOL> more = off_m & <NUM_LIT> <EOL> return cls ( nxt , offset , more , id_ ) <EOL> def serialize ( self ) : <EOL> off_m = ( self . offset << <NUM_LIT:3> | self . more ) <EOL> buf = struct . pack ( self . _PACK_STR , self . nxt , off_m , self . id_ ) <EOL> return buf <EOL> def __len__ ( self ) : <EOL> return self . _MIN_LEN <EOL> @ ipv6 . register_header_type ( inet . IPPROTO_AH ) <EOL> class auth ( header ) : <EOL> """<STR_LIT>""" <EOL> TYPE = inet . IPPROTO_AH <EOL> _PACK_STR = '<STR_LIT>' <EOL> _MIN_LEN = struct . calcsize ( _PACK_STR ) <EOL> def __init__ ( self , nxt = inet . IPPROTO_TCP , size = <NUM_LIT:2> , spi = <NUM_LIT:0> , seq = <NUM_LIT:0> , <EOL> data = b'<STR_LIT>' ) : <EOL> super ( auth , self ) . __init__ ( nxt ) <EOL> assert data is not None <EOL> self . size = size <EOL> self . spi = spi <EOL> self . seq = seq <EOL> self . data = data <EOL> @ classmethod <EOL> def _get_size ( cls , size ) : <EOL> return ( int ( size ) + <NUM_LIT:2> ) * <NUM_LIT:4> <EOL> @ classmethod <EOL> def parser ( cls , buf ) : <EOL> ( nxt , size , spi , seq ) = struct . unpack_from ( cls . _PACK_STR , buf ) <EOL> form = "<STR_LIT>" % ( cls . _get_size ( size ) - cls . _MIN_LEN ) <EOL> ( data , ) = struct . unpack_from ( form , buf , cls . _MIN_LEN ) <EOL> return cls ( nxt , size , spi , seq , data ) <EOL> def serialize ( self ) : <EOL> buf = struct . pack ( self . _PACK_STR , self . nxt , self . size , self . spi , <EOL> self . seq ) <EOL> buf = bytearray ( buf ) <EOL> form = "<STR_LIT>" % ( auth . _get_size ( self . size ) - self . _MIN_LEN ) <EOL> buf . extend ( struct . pack ( form , self . data ) ) <EOL> return buf <EOL> def __len__ ( self ) : <EOL> return auth . _get_size ( self . size ) <EOL> ipv6 . set_classes ( ipv6 . _IPV6_EXT_HEADER_TYPE ) </s>
