<s> """<STR_LIT>""" <EOL> __author__ = '<STR_LIT>' <EOL> import datetime <EOL> import logging <EOL> import nacaddr <EOL> import re <EOL> from string import Template <EOL> import aclgenerator <EOL> class Term ( aclgenerator . Term ) : <EOL> """<STR_LIT>""" <EOL> _PLATFORM = '<STR_LIT>' <EOL> _POSTJUMP_FORMAT = None <EOL> _PREJUMP_FORMAT = Template ( '<STR_LIT>' ) <EOL> _TERM_FORMAT = Template ( '<STR_LIT>' ) <EOL> _COMMENT_FORMAT = Template ( '<STR_LIT>' ) <EOL> _FILTER_TOP_FORMAT = Template ( '<STR_LIT>' ) <EOL> _LOG_FORMAT = Template ( '<STR_LIT>' ) <EOL> _ACTION_TABLE = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> _PROTO_TABLE = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:all>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> _TCP_FLAGS_TABLE = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:all>' : '<STR_LIT>' , <EOL> '<STR_LIT:none>' : '<STR_LIT>' , <EOL> } <EOL> _KNOWN_OPTIONS_MATCHERS = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def __init__ ( self , term , filter_name , trackstate , filter_action , af = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> super ( Term , self ) . __init__ ( term ) <EOL> self . trackstate = trackstate <EOL> self . term = term <EOL> self . filter = filter_name <EOL> self . default_action = filter_action <EOL> self . options = [ ] <EOL> self . af = af <EOL> if af == '<STR_LIT>' : <EOL> self . _all_ips = nacaddr . IPv6 ( '<STR_LIT>' ) <EOL> self . _ACTION_TABLE [ '<STR_LIT>' ] = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> self . _all_ips = nacaddr . IPv4 ( '<STR_LIT>' ) <EOL> self . _ACTION_TABLE [ '<STR_LIT>' ] = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . term_name = '<STR_LIT>' % ( self . filter [ : <NUM_LIT:1> ] , self . term . name ) <EOL> def __str__ ( self ) : <EOL> if self . term . platform : <EOL> if self . _PLATFORM not in self . term . platform : <EOL> return '<STR_LIT>' <EOL> if self . term . platform_exclude : <EOL> if self . _PLATFORM in self . term . platform_exclude : <EOL> return '<STR_LIT>' <EOL> ret_str = [ ] <EOL> if ( ( self . af == '<STR_LIT>' and '<STR_LIT>' in self . term . protocol ) or <EOL> ( self . af == '<STR_LIT>' and '<STR_LIT>' in self . term . protocol ) ) : <EOL> logging . debug ( self . NO_AF_LOG_PROTO . substitute ( term = self . term . name , <EOL> proto = self . term . protocol , <EOL> af = self . af ) ) <EOL> return '<STR_LIT>' <EOL> if self . term . verbatim : <EOL> for next_verbatim in self . term . verbatim : <EOL> if next_verbatim . value [ <NUM_LIT:0> ] == self . _PLATFORM : <EOL> ret_str . append ( str ( next_verbatim . value [ <NUM_LIT:1> ] ) ) <EOL> return '<STR_LIT:\n>' . join ( ret_str ) <EOL> if self . term . ether_type : <EOL> raise UnsupportedFilterError ( '<STR_LIT>' % ( <EOL> '<STR_LIT>' , self . _PLATFORM , <EOL> '<STR_LIT>' , self . term . name ) ) <EOL> if self . term . address : <EOL> raise UnsupportedFilterError ( '<STR_LIT>' % ( <EOL> '<STR_LIT>' , self . _PLATFORM , <EOL> '<STR_LIT>' , '<STR_LIT>' , self . term . name ) ) <EOL> if self . term . port : <EOL> raise UnsupportedFilterError ( '<STR_LIT>' % ( <EOL> '<STR_LIT>' , self . _PLATFORM , <EOL> '<STR_LIT>' , '<STR_LIT>' , self . term . name ) ) <EOL> self . _SetDefaultAction ( ) <EOL> if self . _TERM_FORMAT : <EOL> ret_str . append ( self . _TERM_FORMAT . substitute ( term = self . term_name ) ) <EOL> if self . _PREJUMP_FORMAT : <EOL> ret_str . append ( self . _PREJUMP_FORMAT . substitute ( filter = self . filter , <EOL> term = self . term_name ) ) <EOL> if self . term . owner : <EOL> self . term . comment . append ( '<STR_LIT>' % self . term . owner ) <EOL> comment_max_width = <NUM_LIT> - len ( self . term_name ) <EOL> if comment_max_width < <NUM_LIT> : <EOL> comment_max_width = <NUM_LIT> <EOL> comments = aclgenerator . WrapWords ( self . term . comment , comment_max_width ) <EOL> if comments and comments [ <NUM_LIT:0> ] : <EOL> for line in comments : <EOL> if not line : <EOL> continue <EOL> ret_str . append ( self . _COMMENT_FORMAT . substitute ( filter = self . filter , <EOL> term = self . term_name , <EOL> comment = str ( line ) ) ) <EOL> if self . term . source_prefix or self . term . destination_prefix : <EOL> if str ( self . term . action [ <NUM_LIT:0> ] ) not in set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) : <EOL> raise UnsupportedFilterError ( '<STR_LIT>' % ( <EOL> '<STR_LIT>' , self . term . name , '<STR_LIT>' , str ( self . term . action [ <NUM_LIT:0> ] ) , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , self . _PLATFORM , '<STR_LIT>' ) ) <EOL> return ( '<STR_LIT>' % <EOL> self . term . name ) <EOL> if self . term . protocol : <EOL> protocol = self . term . protocol <EOL> else : <EOL> protocol = [ '<STR_LIT:all>' ] <EOL> if self . term . protocol_except : <EOL> raise UnsupportedFilterError ( '<STR_LIT>' % ( <EOL> '<STR_LIT:\n>' , self . term . name , <EOL> '<STR_LIT>' ) ) <EOL> ( term_saddr , exclude_saddr , <EOL> term_daddr , exclude_daddr ) = self . _CalculateAddresses ( <EOL> self . term . source_address , self . term . source_address_exclude , <EOL> self . term . destination_address , self . term . destination_address_exclude ) <EOL> if not term_saddr : <EOL> logging . debug ( self . NO_AF_LOG_ADDR . substitute ( term = self . term . name , <EOL> direction = '<STR_LIT:source>' , <EOL> af = self . af ) ) <EOL> return '<STR_LIT>' <EOL> if not term_daddr : <EOL> logging . debug ( self . NO_AF_LOG_ADDR . substitute ( term = self . term . name , <EOL> direction = '<STR_LIT>' , <EOL> af = self . af ) ) <EOL> return '<STR_LIT>' <EOL> source_port = [ ] <EOL> destination_port = [ ] <EOL> if self . term . source_port : <EOL> source_port = self . term . source_port <EOL> if self . term . destination_port : <EOL> destination_port = self . term . destination_port <EOL> icmp_types = [ '<STR_LIT>' ] <EOL> if self . term . icmp_type : <EOL> icmp_types = self . NormalizeIcmpTypes ( self . term . icmp_type , protocol , <EOL> self . af ) <EOL> source_interface = '<STR_LIT>' <EOL> if self . term . source_interface : <EOL> source_interface = self . term . source_interface <EOL> destination_interface = '<STR_LIT>' <EOL> if self . term . destination_interface : <EOL> destination_interface = self . term . destination_interface <EOL> log_hits = False <EOL> if self . term . logging : <EOL> log_hits = True <EOL> tcp_flags = [ ] <EOL> tcp_track_options = [ ] <EOL> for next_opt in [ str ( x ) for x in self . term . option ] : <EOL> if ( ( next_opt . find ( '<STR_LIT>' ) == <NUM_LIT:0> or <EOL> next_opt . find ( '<STR_LIT>' ) == <NUM_LIT:0> ) <EOL> and '<STR_LIT>' not in [ x . strip ( ) for x in self . options ] ) : <EOL> if next_opt . find ( '<STR_LIT>' ) == <NUM_LIT:0> and protocol != [ '<STR_LIT>' ] : <EOL> raise TcpEstablishedError ( '<STR_LIT>' % ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , self . term . name ) ) <EOL> if self . trackstate : <EOL> self . options . append ( '<STR_LIT>' ) <EOL> elif protocol == [ '<STR_LIT>' ] : <EOL> tcp_track_options = [ ( [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , [ '<STR_LIT>' ] ) ] <EOL> for next_flag in self . _TCP_FLAGS_TABLE : <EOL> if next_opt . find ( next_flag ) == <NUM_LIT:0> : <EOL> tcp_flags . append ( self . _TCP_FLAGS_TABLE . get ( next_flag ) ) <EOL> if next_opt in self . _KNOWN_OPTIONS_MATCHERS : <EOL> self . options . append ( self . _KNOWN_OPTIONS_MATCHERS [ next_opt ] ) <EOL> if self . term . packet_length : <EOL> self . options . append ( '<STR_LIT>' % <EOL> self . term . packet_length . replace ( '<STR_LIT:->' , '<STR_LIT::>' ) ) <EOL> if self . term . fragment_offset : <EOL> self . options . append ( '<STR_LIT>' % <EOL> self . term . fragment_offset . replace ( '<STR_LIT:->' , '<STR_LIT::>' ) ) <EOL> for saddr in exclude_saddr : <EOL> ret_str . extend ( self . _FormatPart ( <EOL> '<STR_LIT>' , saddr , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> self . _ACTION_TABLE . get ( '<STR_LIT>' ) ) ) <EOL> for daddr in exclude_daddr : <EOL> ret_str . extend ( self . _FormatPart ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , daddr , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> self . _ACTION_TABLE . get ( '<STR_LIT>' ) ) ) <EOL> for saddr in term_saddr : <EOL> for daddr in term_daddr : <EOL> for icmp in icmp_types : <EOL> for proto in protocol : <EOL> for tcp_matcher in tcp_track_options or ( ( [ ] , [ ] ) , ) : <EOL> ret_str . extend ( self . _FormatPart ( <EOL> str ( proto ) , <EOL> saddr , <EOL> source_port , <EOL> daddr , <EOL> destination_port , <EOL> self . options , <EOL> tcp_flags , <EOL> icmp , <EOL> tcp_matcher , <EOL> source_interface , <EOL> destination_interface , <EOL> log_hits , <EOL> self . _ACTION_TABLE . get ( str ( self . term . action [ <NUM_LIT:0> ] ) ) <EOL> ) ) <EOL> if self . _POSTJUMP_FORMAT : <EOL> ret_str . append ( self . _POSTJUMP_FORMAT . substitute ( filter = self . filter , <EOL> term = self . term_name ) ) <EOL> return '<STR_LIT:\n>' . join ( str ( v ) for v in ret_str if v is not '<STR_LIT>' ) <EOL> def _CalculateAddresses ( self , term_saddr , exclude_saddr , <EOL> term_daddr , exclude_daddr ) : <EOL> """<STR_LIT>""" <EOL> term_saddr_excluded = [ ] <EOL> if not term_saddr : <EOL> term_saddr = [ self . _all_ips ] <EOL> if exclude_saddr : <EOL> term_saddr_excluded . extend ( nacaddr . ExcludeAddrs ( term_saddr , <EOL> exclude_saddr ) ) <EOL> term_daddr_excluded = [ ] <EOL> if not term_daddr : <EOL> term_daddr = [ self . _all_ips ] <EOL> if exclude_daddr : <EOL> term_daddr_excluded . extend ( nacaddr . ExcludeAddrs ( term_daddr , <EOL> exclude_daddr ) ) <EOL> bailout_count = len ( exclude_saddr ) + len ( exclude_daddr ) + ( <EOL> ( len ( self . term . source_address ) or <NUM_LIT:1> ) * <EOL> ( len ( self . term . destination_address ) or <NUM_LIT:1> ) ) <EOL> exclude_count = ( ( len ( term_saddr_excluded ) or <NUM_LIT:1> ) * <EOL> ( len ( term_daddr_excluded ) or <NUM_LIT:1> ) ) <EOL> if exclude_count < bailout_count : <EOL> exclude_saddr = [ ] <EOL> exclude_daddr = [ ] <EOL> if term_saddr_excluded : <EOL> term_saddr = term_saddr_excluded <EOL> if term_daddr_excluded : <EOL> term_daddr = term_daddr_excluded <EOL> v4_src_count = len ( [ x for x in term_saddr if x . version == <NUM_LIT:4> ] ) <EOL> v4_dst_count = len ( [ x for x in term_daddr if x . version == <NUM_LIT:4> ] ) <EOL> v6_src_count = len ( [ x for x in term_saddr if x . version == <NUM_LIT:6> ] ) <EOL> v6_dst_count = len ( [ x for x in term_daddr if x . version == <NUM_LIT:6> ] ) <EOL> num_pairs = v4_src_count * v4_dst_count + v6_src_count * v6_dst_count <EOL> if num_pairs > <NUM_LIT:100> : <EOL> new_exclude_source = nacaddr . ExcludeAddrs ( [ self . _all_ips ] , term_saddr ) <EOL> new_exclude_dest = nacaddr . ExcludeAddrs ( [ self . _all_ips ] , term_daddr ) <EOL> if len ( new_exclude_source ) < len ( new_exclude_dest ) and not exclude_saddr : <EOL> if len ( new_exclude_source ) + len ( term_daddr ) < num_pairs : <EOL> exclude_saddr = new_exclude_source <EOL> term_saddr = [ self . _all_ips ] <EOL> elif not exclude_daddr : <EOL> if len ( new_exclude_dest ) + len ( term_saddr ) < num_pairs : <EOL> exclude_daddr = new_exclude_dest <EOL> term_daddr = [ self . _all_ips ] <EOL> term_saddr = [ x for x in term_saddr <EOL> if x . version == self . AF_MAP [ self . af ] ] <EOL> exclude_saddr = [ x for x in exclude_saddr <EOL> if x . version == self . AF_MAP [ self . af ] ] <EOL> term_daddr = [ x for x in term_daddr <EOL> if x . version == self . AF_MAP [ self . af ] ] <EOL> exclude_daddr = [ x for x in exclude_daddr <EOL> if x . version == self . AF_MAP [ self . af ] ] <EOL> return ( term_saddr , exclude_saddr , term_daddr , exclude_daddr ) <EOL> def _FormatPart ( self , protocol , saddr , sport , daddr , dport , options , <EOL> tcp_flags , icmp_type , track_flags , sint , dint , log_hits , <EOL> action ) : <EOL> """<STR_LIT>""" <EOL> src , dst = self . _GenerateAddressStatement ( saddr , daddr ) <EOL> filter_top = self . _FILTER_TOP_FORMAT . substitute ( filter = self . filter , <EOL> term = self . term_name ) <EOL> source_int = '<STR_LIT>' <EOL> if sint : <EOL> source_int = '<STR_LIT>' % sint <EOL> destination_int = '<STR_LIT>' <EOL> if dint : <EOL> destination_int = '<STR_LIT>' % dint <EOL> log_jump = '<STR_LIT>' <EOL> if log_hits : <EOL> log_jump = self . _LOG_FORMAT . substitute ( term = self . term . name ) <EOL> if not options : <EOL> options = [ ] <EOL> proto = self . _PROTO_TABLE . get ( str ( protocol ) ) <EOL> if protocol and not proto : <EOL> proto = '<STR_LIT>' % str ( protocol ) <EOL> if protocol == '<STR_LIT>' : <EOL> proto = '<STR_LIT>' <EOL> options . append ( '<STR_LIT>' ) <EOL> if protocol == '<STR_LIT>' : <EOL> proto = '<STR_LIT>' <EOL> options . append ( '<STR_LIT>' ) <EOL> if self . trackstate : <EOL> already_stateful = False <EOL> for option in options : <EOL> if '<STR_LIT:state>' in option : <EOL> already_stateful = True <EOL> if not already_stateful : <EOL> if '<STR_LIT>' in action : <EOL> options . append ( '<STR_LIT>' ) <EOL> if tcp_flags or ( track_flags and track_flags [ <NUM_LIT:0> ] ) : <EOL> check_fields = '<STR_LIT:U+002C>' . join ( sorted ( set ( tcp_flags + track_flags [ <NUM_LIT:0> ] ) ) ) <EOL> set_fields = '<STR_LIT:U+002C>' . join ( sorted ( set ( tcp_flags + track_flags [ <NUM_LIT:1> ] ) ) ) <EOL> flags = '<STR_LIT>' % ( check_fields , set_fields ) <EOL> else : <EOL> flags = '<STR_LIT>' <EOL> icmp_type = str ( icmp_type ) <EOL> if not icmp_type : <EOL> icmp = '<STR_LIT>' <EOL> elif str ( protocol ) == '<STR_LIT>' : <EOL> icmp = '<STR_LIT>' % icmp_type <EOL> else : <EOL> icmp = '<STR_LIT>' % icmp_type <EOL> sports = dports = [ '<STR_LIT>' ] <EOL> if sport : <EOL> sports = self . _GeneratePortStatement ( sport , source = True ) <EOL> if dport : <EOL> dports = self . _GeneratePortStatement ( dport , dest = True ) <EOL> ret_lines = [ ] <EOL> for sport in sports : <EOL> for dport in dports : <EOL> rval = [ filter_top ] <EOL> if re . search ( '<STR_LIT>' , sport ) and not re . search ( '<STR_LIT>' , dport ) : <EOL> dport , sport = sport , dport <EOL> if str ( protocol ) == '<STR_LIT>' : <EOL> options . extend ( ( proto , icmp ) ) <EOL> proto = '<STR_LIT>' <EOL> icmp = '<STR_LIT>' <EOL> if ( str ( self . af ) == '<STR_LIT>' and <EOL> str ( protocol ) == '<STR_LIT:all>' and <EOL> '<STR_LIT>' in str ( action ) ) : <EOL> proto = '<STR_LIT>' <EOL> for value in ( proto , flags , sport , dport , icmp , src , dst , <EOL> '<STR_LIT:U+0020>' . join ( options ) , source_int , destination_int ) : <EOL> if value : <EOL> rval . append ( str ( value ) ) <EOL> if log_jump : <EOL> ret_lines . append ( '<STR_LIT:U+0020>' . join ( rval + [ log_jump ] ) ) <EOL> ret_lines . append ( '<STR_LIT:U+0020>' . join ( rval + [ action ] ) ) <EOL> return ret_lines <EOL> def _GenerateAddressStatement ( self , saddr , daddr ) : <EOL> """<STR_LIT>""" <EOL> src = '<STR_LIT>' <EOL> dst = '<STR_LIT>' <EOL> if not saddr or saddr == self . _all_ips : <EOL> src = '<STR_LIT>' <EOL> else : <EOL> src = '<STR_LIT>' % ( saddr . ip , saddr . prefixlen ) <EOL> if not daddr or daddr == self . _all_ips : <EOL> dst = '<STR_LIT>' <EOL> else : <EOL> dst = '<STR_LIT>' % ( daddr . ip , daddr . prefixlen ) <EOL> return ( src , dst ) <EOL> def _GeneratePortStatement ( self , ports , source = False , dest = False ) : <EOL> """<STR_LIT>""" <EOL> if not ports : <EOL> return '<STR_LIT>' <EOL> direction = '<STR_LIT>' <EOL> if source and dest : <EOL> raise BadPortsError ( '<STR_LIT>' ) <EOL> elif source : <EOL> direction = '<STR_LIT:s>' <EOL> elif dest : <EOL> direction = '<STR_LIT:d>' <EOL> else : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> max_ports = <NUM_LIT> <EOL> norm_ports = [ ] <EOL> portstrings = [ ] <EOL> count = <NUM_LIT:0> <EOL> for port in ports : <EOL> if port [ <NUM_LIT:0> ] == port [ <NUM_LIT:1> ] : <EOL> norm_ports . append ( str ( port [ <NUM_LIT:0> ] ) ) <EOL> count += <NUM_LIT:1> <EOL> else : <EOL> norm_ports . append ( '<STR_LIT>' % ( port [ <NUM_LIT:0> ] , port [ <NUM_LIT:1> ] ) ) <EOL> count += <NUM_LIT:2> <EOL> if count >= max_ports : <EOL> count = <NUM_LIT:0> <EOL> portstrings . append ( '<STR_LIT>' % ( direction , <EOL> '<STR_LIT:U+002C>' . join ( norm_ports ) ) ) <EOL> norm_ports = [ ] <EOL> if norm_ports : <EOL> if len ( norm_ports ) == <NUM_LIT:1> : <EOL> portstrings . append ( '<STR_LIT>' % ( direction , norm_ports [ <NUM_LIT:0> ] ) ) <EOL> else : <EOL> portstrings . append ( '<STR_LIT>' % <EOL> ( direction , '<STR_LIT:U+002C>' . join ( norm_ports ) ) ) <EOL> return portstrings <EOL> def _SetDefaultAction ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . term . action : <EOL> self . term . action [ <NUM_LIT:0> ] . value = self . default_action <EOL> class Iptables ( aclgenerator . ACLGenerator ) : <EOL> """<STR_LIT>""" <EOL> _PLATFORM = '<STR_LIT>' <EOL> _DEFAULT_PROTOCOL = '<STR_LIT:all>' <EOL> _SUFFIX = '<STR_LIT>' <EOL> _RENDER_PREFIX = None <EOL> _RENDER_SUFFIX = None <EOL> _DEFAULTACTION_FORMAT = '<STR_LIT>' <EOL> _DEFAULTACTION_FORMAT_CUSTOM_CHAIN = '<STR_LIT>' <EOL> _DEFAULT_ACTION = '<STR_LIT>' <EOL> _TERM = Term <EOL> _TERM_MAX_LENGTH = <NUM_LIT> <EOL> _OPTIONAL_SUPPORTED_KEYWORDS = set ( [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] ) <EOL> _GOOD_FILTERS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> def _WarnIfCustomTarget ( self , target ) : <EOL> """<STR_LIT>""" <EOL> if target not in self . _GOOD_FILTERS : <EOL> logging . warn ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , target ) <EOL> def _TranslatePolicy ( self , pol , exp_info ) : <EOL> """<STR_LIT>""" <EOL> self . iptables_policies = [ ] <EOL> current_date = datetime . date . today ( ) <EOL> exp_info_date = current_date + datetime . timedelta ( weeks = exp_info ) <EOL> default_action = None <EOL> good_default_actions = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> good_afs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> good_options = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> all_protocols_stateful = True <EOL> for header , terms in pol . filters : <EOL> filter_type = None <EOL> if self . _PLATFORM not in header . platforms : <EOL> continue <EOL> filter_options = header . FilterOptions ( self . _PLATFORM ) [ <NUM_LIT:1> : ] <EOL> filter_name = header . FilterName ( self . _PLATFORM ) <EOL> self . _WarnIfCustomTarget ( filter_name ) <EOL> for opt in filter_options : <EOL> if opt not in good_default_actions + good_afs + good_options : <EOL> raise UnsupportedTargetOption ( '<STR_LIT>' % ( <EOL> '<STR_LIT>' , self . _PLATFORM , <EOL> '<STR_LIT>' , opt ) ) <EOL> if '<STR_LIT>' in filter_options : <EOL> all_protocols_stateful = False <EOL> for address_family in good_afs : <EOL> if address_family in filter_options : <EOL> if filter_type is not None : <EOL> raise UnsupportedFilterError ( '<STR_LIT>' % ( <EOL> '<STR_LIT>' , good_afs , '<STR_LIT>' , <EOL> filter_options ) ) <EOL> filter_type = address_family <EOL> if filter_type is None : <EOL> filter_type = '<STR_LIT>' <EOL> if self . _PLATFORM == '<STR_LIT>' and filter_name == '<STR_LIT>' : <EOL> default_action = '<STR_LIT>' <EOL> for next_target in header . target : <EOL> if next_target . platform == self . _PLATFORM : <EOL> if len ( next_target . options ) > <NUM_LIT:1> : <EOL> for arg in next_target . options : <EOL> if arg in good_default_actions : <EOL> default_action = arg <EOL> if default_action and default_action not in good_default_actions : <EOL> raise UnsupportedDefaultAction ( '<STR_LIT>' % ( <EOL> '<STR_LIT>' , '<STR_LIT:U+002CU+0020>' . join ( good_default_actions ) , <EOL> '<STR_LIT>' , default_action , '<STR_LIT>' ) ) <EOL> new_terms = [ ] <EOL> term_names = set ( ) <EOL> for term in terms : <EOL> term . name = self . FixTermLength ( term . name , <EOL> '<STR_LIT>' in filter_options , <EOL> '<STR_LIT>' in filter_options ) <EOL> if term . name in term_names : <EOL> raise aclgenerator . DuplicateTermError ( <EOL> '<STR_LIT>' % term . name ) <EOL> term_names . add ( term . name ) <EOL> term = self . FixHighPorts ( term , af = filter_type , <EOL> all_protocols_stateful = all_protocols_stateful ) <EOL> if not term : <EOL> continue <EOL> if term . expiration : <EOL> if term . expiration <= exp_info_date : <EOL> logging . info ( '<STR_LIT>' <EOL> '<STR_LIT>' , term . name , filter_name ) <EOL> if term . expiration <= current_date : <EOL> logging . warn ( '<STR_LIT>' <EOL> '<STR_LIT>' , term . name , filter_name ) <EOL> continue <EOL> new_terms . append ( self . _TERM ( term , filter_name , all_protocols_stateful , <EOL> default_action , filter_type ) ) <EOL> self . iptables_policies . append ( ( header , filter_name , filter_type , <EOL> default_action , new_terms ) ) <EOL> def SetTarget ( self , target , action = None ) : <EOL> """<STR_LIT>""" <EOL> pol = list ( self . iptables_policies [ <NUM_LIT:0> ] ) <EOL> pol [ <NUM_LIT:1> ] = target <EOL> self . _WarnIfCustomTarget ( target ) <EOL> if action : <EOL> pol [ <NUM_LIT:3> ] = action <EOL> self . iptables_policies [ <NUM_LIT:0> ] = tuple ( pol ) <EOL> def __str__ ( self ) : <EOL> target = [ ] <EOL> pretty_platform = '<STR_LIT>' % ( self . _PLATFORM [ <NUM_LIT:0> ] . upper ( ) , self . _PLATFORM [ <NUM_LIT:1> : ] ) <EOL> if self . _RENDER_PREFIX : <EOL> target . append ( self . _RENDER_PREFIX ) <EOL> for ( header , filter_name , filter_type , default_action , terms <EOL> ) in self . iptables_policies : <EOL> target . append ( '<STR_LIT>' % ( pretty_platform , <EOL> header . FilterName ( self . _PLATFORM ) ) ) <EOL> comments = aclgenerator . WrapWords ( header . comment , <NUM_LIT> ) <EOL> if comments and comments [ <NUM_LIT:0> ] : <EOL> for line in comments : <EOL> target . append ( '<STR_LIT>' % line ) <EOL> target . append ( '<STR_LIT:#>' ) <EOL> target . extend ( aclgenerator . AddRepositoryTags ( '<STR_LIT>' ) ) <EOL> target . append ( '<STR_LIT>' + filter_type ) <EOL> if filter_name in self . _GOOD_FILTERS : <EOL> if default_action : <EOL> target . append ( self . _DEFAULTACTION_FORMAT % ( filter_name , <EOL> default_action ) ) <EOL> elif self . _PLATFORM == '<STR_LIT>' : <EOL> target . append ( <EOL> self . _DEFAULTACTION_FORMAT % ( filter_name , self . _DEFAULT_ACTION ) ) <EOL> else : <EOL> target . append ( self . _DEFAULTACTION_FORMAT_CUSTOM_CHAIN % filter_name ) <EOL> for term in terms : <EOL> term_str = str ( term ) <EOL> if term_str : <EOL> target . append ( term_str ) <EOL> if self . _RENDER_SUFFIX : <EOL> target . append ( self . _RENDER_SUFFIX ) <EOL> target . append ( '<STR_LIT>' ) <EOL> return '<STR_LIT:\n>' . join ( target ) <EOL> class Error ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class BadPortsError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class UnsupportedFilterError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class NoIptablesPolicyError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class TcpEstablishedError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class EstablishedError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class UnsupportedDefaultAction ( Error ) : <EOL> """<STR_LIT>""" <EOL> class UnsupportedTargetOption ( Error ) : <EOL> """<STR_LIT>""" </s>
