<s> import operator <EOL> import re <EOL> import ipaddr <EOL> from django . core . exceptions import ObjectDoesNotExist , MultipleObjectsReturned <EOL> from django . core . exceptions import ValidationError <EOL> from django . db . models import Q <EOL> from mozdns . address_record . models import AddressRecord <EOL> from mozdns . cname . models import CNAME <EOL> from mozdns . domain . models import Domain <EOL> from mozdns . mx . models import MX <EOL> from mozdns . nameserver . models import Nameserver <EOL> from mozdns . ptr . models import PTR <EOL> from mozdns . srv . models import SRV <EOL> from mozdns . soa . models import SOA <EOL> from mozdns . sshfp . models import SSHFP <EOL> from mozdns . txt . models import TXT <EOL> from mozdns . view . models import View <EOL> from core . interface . static_intr . models import StaticInterface <EOL> from core . site . models import Site <EOL> from core . utils import IPFilter , one_to_two <EOL> from core . vlan . models import Vlan <EOL> from core . utils import start_end_filter , resolve_ip_type <EOL> from systems . models import System <EOL> class BadDirective ( Exception ) : <EOL> pass <EOL> searchables = ( <EOL> ( '<STR_LIT:A>' , AddressRecord ) , <EOL> ( '<STR_LIT>' , CNAME ) , <EOL> ( '<STR_LIT>' , Domain ) , <EOL> ( '<STR_LIT>' , MX ) , <EOL> ( '<STR_LIT>' , Nameserver ) , <EOL> ( '<STR_LIT>' , PTR ) , <EOL> ( '<STR_LIT>' , SOA ) , <EOL> ( '<STR_LIT>' , SRV ) , <EOL> ( '<STR_LIT>' , SSHFP ) , <EOL> ( '<STR_LIT>' , StaticInterface ) , <EOL> ( '<STR_LIT>' , System ) , <EOL> ( '<STR_LIT>' , TXT ) , <EOL> ) <EOL> class _Filter ( object ) : <EOL> """<STR_LIT>""" <EOL> ntype = "<STR_LIT>" <EOL> def __str__ ( self ) : <EOL> return self . value <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" . format ( self . __class__ , self ) <EOL> def compile_Q ( self ) : <EOL> pass <EOL> def build_filter ( filter_ , fields , filter_type ) : <EOL> final_filter = Q ( ) <EOL> for t in fields : <EOL> final_filter = final_filter | Q ( <EOL> ** { "<STR_LIT>" . format ( t , filter_type ) : filter_ } ) <EOL> return final_filter <EOL> class TextFilter ( _Filter ) : <EOL> def __init__ ( self , rvalue ) : <EOL> self . value = rvalue <EOL> def compile_Q ( self ) : <EOL> result = [ ] <EOL> for name , Klass in searchables : <EOL> result . append ( <EOL> build_filter ( self . value , Klass . search_fields , '<STR_LIT>' ) ) <EOL> return result <EOL> class REFilter ( TextFilter ) : <EOL> num_match = re . compile ( "<STR_LIT>" ) <EOL> def _expand_number_regex ( self , value ) : <EOL> """<STR_LIT>""" <EOL> matches = self . num_match . findall ( value ) <EOL> for low , high in matches : <EOL> padding = min ( len ( low ) , len ( high ) ) <EOL> if int ( low ) >= int ( high ) : <EOL> continue <EOL> new_value = "<STR_LIT>" <EOL> for i in xrange ( int ( low ) , int ( high ) + <NUM_LIT:1> ) : <EOL> new_value += "<STR_LIT>" . format ( str ( i ) . rjust ( padding , '<STR_LIT:0>' ) ) <EOL> new_value = '<STR_LIT:(>' + new_value . strip ( '<STR_LIT:|>' ) + '<STR_LIT:)>' <EOL> value = value . replace ( '<STR_LIT>' . format ( low , high ) , new_value ) <EOL> return value <EOL> def compile_Q ( self ) : <EOL> result = [ ] <EOL> value = self . _expand_number_regex ( self . value ) <EOL> for name , Klass in searchables : <EOL> result . append ( build_filter ( value , Klass . search_fields , '<STR_LIT>' ) ) <EOL> return result <EOL> class DirectiveFilter ( _Filter ) : <EOL> def __init__ ( self , directive , dvalue ) : <EOL> self . directive = directive <EOL> self . dvalue = dvalue <EOL> def compile_Q ( self ) : <EOL> if self . directive == '<STR_LIT>' : <EOL> return build_view_qsets ( self . dvalue ) <EOL> elif self . directive == '<STR_LIT>' : <EOL> return build_network_qsets ( self . dvalue ) <EOL> elif self . directive == '<STR_LIT>' : <EOL> return build_vlan_qsets ( self . dvalue ) <EOL> elif self . directive == '<STR_LIT>' : <EOL> return build_zone_qsets ( self . dvalue ) <EOL> elif self . directive == '<STR_LIT>' : <EOL> return build_range_qsets ( self . dvalue ) <EOL> elif self . directive == '<STR_LIT:type>' : <EOL> return build_rdtype_qsets ( self . dvalue ) <EOL> elif self . directive == '<STR_LIT>' : <EOL> return build_site_qsets ( self . dvalue ) <EOL> elif self . directive == '<STR_LIT>' : <EOL> return build_ip_qsets ( self . dvalue ) <EOL> else : <EOL> raise BadDirective ( <EOL> "<STR_LIT>" . format ( self . directive ) <EOL> ) <EOL> def build_rdtype_qsets ( rdtype ) : <EOL> """<STR_LIT>""" <EOL> rdtype = rdtype . upper ( ) <EOL> select = Q ( pk__gt = - <NUM_LIT:1> ) <EOL> no_select = Q ( pk__lte = - <NUM_LIT:1> ) <EOL> result = [ ] <EOL> for name , Klass in searchables : <EOL> if name == rdtype : <EOL> result . append ( select ) <EOL> else : <EOL> result . append ( no_select ) <EOL> return result <EOL> def build_view_qsets ( view_name ) : <EOL> """<STR_LIT>""" <EOL> view_name = view_name . lower ( ) <EOL> try : <EOL> view = View . objects . get ( name = view_name ) <EOL> except ObjectDoesNotExist : <EOL> raise BadDirective ( "<STR_LIT>" . format ( view_name ) ) <EOL> view_filter = Q ( views = view ) <EOL> q_sets = [ ] <EOL> select = Q ( pk__gt = - <NUM_LIT:1> ) <EOL> for name , Klass in searchables : <EOL> if name == '<STR_LIT>' : <EOL> q_sets . append ( select ) <EOL> elif hasattr ( Klass , '<STR_LIT>' ) : <EOL> q_sets . append ( view_filter ) <EOL> else : <EOL> q_sets . append ( None ) <EOL> return q_sets <EOL> def build_ipf_qsets ( q ) : <EOL> """<STR_LIT>""" <EOL> q_sets = [ ] <EOL> for name , Klass in searchables : <EOL> if name == '<STR_LIT:A>' or name == '<STR_LIT>' or name == '<STR_LIT>' : <EOL> q_sets . append ( q ) <EOL> else : <EOL> q_sets . append ( None ) <EOL> return q_sets <EOL> def build_range_qsets ( range_ ) : <EOL> try : <EOL> start , end = range_ . split ( '<STR_LIT:U+002C>' ) <EOL> except ValueError : <EOL> raise BadDirective ( "<STR_LIT>" ) <EOL> start_ip_type , _ = resolve_ip_type ( start ) <EOL> end_ip_type , _ = resolve_ip_type ( end ) <EOL> if start_ip_type != end_ip_type or not start_ip_type or not end_ip_type : <EOL> raise BadDirective ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( start , end ) ) <EOL> try : <EOL> istart , iend , ipf_q = start_end_filter ( start , end , start_ip_type ) <EOL> except ( ValidationError , ipaddr . AddressValueError ) , e : <EOL> raise BadDirective ( str ( e ) ) <EOL> return build_ipf_qsets ( ipf_q ) <EOL> def build_ip_qsets ( ip_str ) : <EOL> ip_type , Klass = resolve_ip_type ( ip_str ) <EOL> try : <EOL> ip = Klass ( ip_str ) <EOL> ip_upper , ip_lower = one_to_two ( int ( ip ) ) <EOL> except ( ipaddr . AddressValueError ) : <EOL> raise BadDirective ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( ip_str ) ) <EOL> return build_ipf_qsets ( Q ( ip_upper = ip_upper , ip_lower = ip_lower ) ) <EOL> def build_network_qsets ( network_str ) : <EOL> ip_type , Klass = resolve_ip_type ( network_str ) <EOL> try : <EOL> network = Klass ( network_str ) <EOL> ipf = IPFilter ( network . network , network . broadcast , ip_type ) <EOL> except ( ipaddr . AddressValueError , ipaddr . NetmaskValueError ) : <EOL> raise BadDirective ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( network_str ) ) <EOL> return build_ipf_qsets ( ipf . Q ) <EOL> def build_site_qsets ( site_name ) : <EOL> try : <EOL> site = Site . objects . get ( name = site_name ) <EOL> except ObjectDoesNotExist : <EOL> raise BadDirective ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( site_name ) ) <EOL> return build_ipf_qsets ( site . compile_Q ( ) ) <EOL> def build_vlan_qsets ( vlan_name ) : <EOL> try : <EOL> if vlan_name . isdigit ( ) : <EOL> vlan = Vlan . objects . get ( number = vlan_name ) <EOL> else : <EOL> vlan = Vlan . objects . get ( name = vlan_name ) <EOL> except ObjectDoesNotExist : <EOL> raise BadDirective ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( vlan_name ) ) <EOL> except MultipleObjectsReturned : <EOL> raise BadDirective ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( vlan_name ) ) <EOL> return build_ipf_qsets ( vlan . compile_Q ( ) ) <EOL> def build_zone_qsets ( zone ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> root_domain = Domain . objects . get ( name = zone ) <EOL> except ObjectDoesNotExist : <EOL> raise BadDirective ( "<STR_LIT>" . format ( zone ) ) <EOL> if not root_domain . soa : <EOL> raise BadDirective ( "<STR_LIT>" . format ( zone ) ) <EOL> def _get_zone_domains ( domain ) : <EOL> domains = [ domain ] <EOL> for sub_domain in domain . domain_set . filter ( soa = domain . soa ) : <EOL> domains += _get_zone_domains ( sub_domain ) <EOL> return domains <EOL> domains = _get_zone_domains ( root_domain ) <EOL> if root_domain . is_reverse : <EOL> domains = [ Q ( reverse_domain = domain ) for domain in domains ] <EOL> else : <EOL> domains = [ Q ( domain = domain ) for domain in domains ] <EOL> zone_query = reduce ( operator . or_ , domains , Q ( ) ) <EOL> result = [ ] <EOL> for name , Klass in searchables : <EOL> if hasattr ( Klass , '<STR_LIT>' ) and not root_domain . is_reverse : <EOL> result . append ( zone_query ) <EOL> elif hasattr ( Klass , '<STR_LIT>' ) and root_domain . is_reverse : <EOL> result . append ( zone_query ) <EOL> elif name == '<STR_LIT>' : <EOL> result . append ( Q ( pk = root_domain . soa . pk ) ) <EOL> else : <EOL> result . append ( None ) <EOL> return result </s>
