<s> """<STR_LIT>""" <EOL> from __future__ import generators <EOL> import sys <EOL> import dns . exception <EOL> import dns . name <EOL> import dns . node <EOL> import dns . rdataclass <EOL> import dns . rdatatype <EOL> import dns . rdata <EOL> import dns . rrset <EOL> import dns . tokenizer <EOL> import dns . ttl <EOL> class BadZone ( dns . exception . DNSException ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class NoSOA ( BadZone ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class NoNS ( BadZone ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class UnknownOrigin ( BadZone ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Zone ( object ) : <EOL> """<STR_LIT>""" <EOL> node_factory = dns . node . Node <EOL> __slots__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> def __init__ ( self , origin , rdclass = dns . rdataclass . IN , relativize = True ) : <EOL> """<STR_LIT>""" <EOL> self . rdclass = rdclass <EOL> self . origin = origin <EOL> self . nodes = { } <EOL> self . relativize = relativize <EOL> def __eq__ ( self , other ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( other , Zone ) : <EOL> return False <EOL> if self . rdclass != other . rdclass or self . origin != other . origin or self . nodes != other . nodes : <EOL> return False <EOL> return True <EOL> def __ne__ ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return not self . __eq__ ( other ) <EOL> def _validate_name ( self , name ) : <EOL> if isinstance ( name , ( str , unicode ) ) : <EOL> name = dns . name . from_text ( name , None ) <EOL> elif not isinstance ( name , dns . name . Name ) : <EOL> raise KeyError ( "<STR_LIT>" ) <EOL> if name . is_absolute ( ) : <EOL> if not name . is_subdomain ( self . origin ) : <EOL> raise KeyError ( "<STR_LIT>" ) <EOL> if self . relativize : <EOL> name = name . relativize ( self . origin ) <EOL> return name <EOL> def __getitem__ ( self , key ) : <EOL> key = self . _validate_name ( key ) <EOL> return self . nodes [ key ] <EOL> def __setitem__ ( self , key , value ) : <EOL> key = self . _validate_name ( key ) <EOL> self . nodes [ key ] = value <EOL> def __delitem__ ( self , key ) : <EOL> key = self . _validate_name ( key ) <EOL> del self . nodes [ key ] <EOL> def __iter__ ( self ) : <EOL> return self . nodes . iterkeys ( ) <EOL> def iterkeys ( self ) : <EOL> return self . nodes . iterkeys ( ) <EOL> def keys ( self ) : <EOL> return self . nodes . keys ( ) <EOL> def itervalues ( self ) : <EOL> return self . nodes . itervalues ( ) <EOL> def values ( self ) : <EOL> return self . nodes . values ( ) <EOL> def iteritems ( self ) : <EOL> return self . nodes . iteritems ( ) <EOL> def items ( self ) : <EOL> return self . nodes . items ( ) <EOL> def get ( self , key ) : <EOL> key = self . _validate_name ( key ) <EOL> return self . nodes . get ( key ) <EOL> def __contains__ ( self , other ) : <EOL> return other in self . nodes <EOL> def find_node ( self , name , create = False ) : <EOL> """<STR_LIT>""" <EOL> name = self . _validate_name ( name ) <EOL> node = self . nodes . get ( name ) <EOL> if node is None : <EOL> if not create : <EOL> raise KeyError <EOL> node = self . node_factory ( ) <EOL> self . nodes [ name ] = node <EOL> return node <EOL> def get_node ( self , name , create = False ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> node = self . find_node ( name , create ) <EOL> except KeyError : <EOL> node = None <EOL> return node <EOL> def delete_node ( self , name ) : <EOL> """<STR_LIT>""" <EOL> name = self . _validate_name ( name ) <EOL> if self . nodes . has_key ( name ) : <EOL> del self . nodes [ name ] <EOL> def find_rdataset ( self , name , rdtype , covers = dns . rdatatype . NONE , <EOL> create = False ) : <EOL> """<STR_LIT>""" <EOL> name = self . _validate_name ( name ) <EOL> if isinstance ( rdtype , str ) : <EOL> rdtype = dns . rdatatype . from_text ( rdtype ) <EOL> if isinstance ( covers , str ) : <EOL> covers = dns . rdatatype . from_text ( covers ) <EOL> node = self . find_node ( name , create ) <EOL> return node . find_rdataset ( self . rdclass , rdtype , covers , create ) <EOL> def get_rdataset ( self , name , rdtype , covers = dns . rdatatype . NONE , <EOL> create = False ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> rdataset = self . find_rdataset ( name , rdtype , covers , create ) <EOL> except KeyError : <EOL> rdataset = None <EOL> return rdataset <EOL> def delete_rdataset ( self , name , rdtype , covers = dns . rdatatype . NONE ) : <EOL> """<STR_LIT>""" <EOL> name = self . _validate_name ( name ) <EOL> if isinstance ( rdtype , str ) : <EOL> rdtype = dns . rdatatype . from_text ( rdtype ) <EOL> if isinstance ( covers , str ) : <EOL> covers = dns . rdatatype . from_text ( covers ) <EOL> node = self . get_node ( name ) <EOL> if not node is None : <EOL> node . delete_rdataset ( self . rdclass , rdtype , covers ) <EOL> if len ( node ) == <NUM_LIT:0> : <EOL> self . delete_node ( name ) <EOL> def replace_rdataset ( self , name , replacement ) : <EOL> """<STR_LIT>""" <EOL> if replacement . rdclass != self . rdclass : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> node = self . find_node ( name , True ) <EOL> node . replace_rdataset ( replacement ) <EOL> def find_rrset ( self , name , rdtype , covers = dns . rdatatype . NONE ) : <EOL> """<STR_LIT>""" <EOL> name = self . _validate_name ( name ) <EOL> if isinstance ( rdtype , str ) : <EOL> rdtype = dns . rdatatype . from_text ( rdtype ) <EOL> if isinstance ( covers , str ) : <EOL> covers = dns . rdatatype . from_text ( covers ) <EOL> rdataset = self . nodes [ name ] . find_rdataset ( self . rdclass , rdtype , covers ) <EOL> rrset = dns . rrset . RRset ( name , self . rdclass , rdtype , covers ) <EOL> rrset . update ( rdataset ) <EOL> return rrset <EOL> def get_rrset ( self , name , rdtype , covers = dns . rdatatype . NONE ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> rrset = self . find_rrset ( name , rdtype , covers ) <EOL> except KeyError : <EOL> rrset = None <EOL> return rrset <EOL> def iterate_rdatasets ( self , rdtype = dns . rdatatype . ANY , <EOL> covers = dns . rdatatype . NONE ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( rdtype , str ) : <EOL> rdtype = dns . rdatatype . from_text ( rdtype ) <EOL> if isinstance ( covers , str ) : <EOL> covers = dns . rdatatype . from_text ( covers ) <EOL> for ( name , node ) in self . iteritems ( ) : <EOL> for rds in node : <EOL> if rdtype == dns . rdatatype . ANY or ( rds . rdtype == rdtype and rds . covers == covers ) : <EOL> yield ( name , rds ) <EOL> def iterate_rdatas ( self , rdtype = dns . rdatatype . ANY , <EOL> covers = dns . rdatatype . NONE ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( rdtype , str ) : <EOL> rdtype = dns . rdatatype . from_text ( rdtype ) <EOL> if isinstance ( covers , str ) : <EOL> covers = dns . rdatatype . from_text ( covers ) <EOL> for ( name , node ) in self . iteritems ( ) : <EOL> for rds in node : <EOL> if rdtype == dns . rdatatype . ANY or ( rds . rdtype == rdtype and rds . covers == covers ) : <EOL> for rdata in rds : <EOL> yield ( name , rds . ttl , rdata ) <EOL> def to_file ( self , f , sorted = True , relativize = True , nl = None ) : <EOL> """<STR_LIT>""" <EOL> if sys . hexversion >= <NUM_LIT> : <EOL> str_type = basestring <EOL> else : <EOL> str_type = str <EOL> if nl is None : <EOL> opts = '<STR_LIT:w>' <EOL> else : <EOL> opts = '<STR_LIT:wb>' <EOL> if isinstance ( f , str_type ) : <EOL> f = file ( f , opts ) <EOL> want_close = True <EOL> else : <EOL> want_close = False <EOL> try : <EOL> if sorted : <EOL> names = self . keys ( ) <EOL> names . sort ( ) <EOL> else : <EOL> names = self . iterkeys ( ) <EOL> for n in names : <EOL> l = self [ n ] . to_text ( n , origin = self . origin , <EOL> relativize = relativize ) <EOL> if nl is None : <EOL> print >> f , l <EOL> else : <EOL> f . write ( l ) <EOL> f . write ( nl ) <EOL> finally : <EOL> if want_close : <EOL> f . close ( ) <EOL> def check_origin ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . relativize : <EOL> name = dns . name . empty <EOL> else : <EOL> name = self . origin <EOL> if self . get_rdataset ( name , dns . rdatatype . SOA ) is None : <EOL> raise NoSOA <EOL> if self . get_rdataset ( name , dns . rdatatype . NS ) is None : <EOL> raise NoNS <EOL> class _MasterReader ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , tok , origin , rdclass , relativize , zone_factory = Zone , <EOL> allow_include = False , check_origin = True ) : <EOL> if isinstance ( origin , ( str , unicode ) ) : <EOL> origin = dns . name . from_text ( origin ) <EOL> self . tok = tok <EOL> self . current_origin = origin <EOL> self . relativize = relativize <EOL> self . ttl = <NUM_LIT:0> <EOL> self . last_name = None <EOL> self . zone = zone_factory ( origin , rdclass , relativize = relativize ) <EOL> self . saved_state = [ ] <EOL> self . current_file = None <EOL> self . allow_include = allow_include <EOL> self . check_origin = check_origin <EOL> def _eat_line ( self ) : <EOL> while <NUM_LIT:1> : <EOL> token = self . tok . get ( ) <EOL> if token . is_eol_or_eof ( ) : <EOL> break <EOL> def _rr_line ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . current_origin is None : <EOL> raise UnknownOrigin <EOL> token = self . tok . get ( want_leading = True ) <EOL> if not token . is_whitespace ( ) : <EOL> self . last_name = dns . name . from_text ( token . value , self . current_origin ) <EOL> else : <EOL> token = self . tok . get ( ) <EOL> if token . is_eol_or_eof ( ) : <EOL> return <EOL> self . tok . unget ( token ) <EOL> name = self . last_name <EOL> if not name . is_subdomain ( self . zone . origin ) : <EOL> self . _eat_line ( ) <EOL> return <EOL> if self . relativize : <EOL> name = name . relativize ( self . zone . origin ) <EOL> token = self . tok . get ( ) <EOL> if not token . is_identifier ( ) : <EOL> raise dns . exception . SyntaxError <EOL> try : <EOL> ttl = dns . ttl . from_text ( token . value ) <EOL> token = self . tok . get ( ) <EOL> if not token . is_identifier ( ) : <EOL> raise dns . exception . SyntaxError <EOL> except dns . ttl . BadTTL : <EOL> ttl = self . ttl <EOL> try : <EOL> rdclass = dns . rdataclass . from_text ( token . value ) <EOL> token = self . tok . get ( ) <EOL> if not token . is_identifier ( ) : <EOL> raise dns . exception . SyntaxError <EOL> except dns . exception . SyntaxError : <EOL> raise dns . exception . SyntaxError <EOL> except : <EOL> rdclass = self . zone . rdclass <EOL> if rdclass != self . zone . rdclass : <EOL> raise dns . exception . SyntaxError ( "<STR_LIT>" ) <EOL> try : <EOL> rdtype = dns . rdatatype . from_text ( token . value ) <EOL> except : <EOL> raise dns . exception . SyntaxError ( "<STR_LIT>" % token . value ) <EOL> n = self . zone . nodes . get ( name ) <EOL> if n is None : <EOL> n = self . zone . node_factory ( ) <EOL> self . zone . nodes [ name ] = n <EOL> try : <EOL> rd = dns . rdata . from_text ( rdclass , rdtype , self . tok , <EOL> self . current_origin , False ) <EOL> except dns . exception . SyntaxError : <EOL> ( ty , va ) = sys . exc_info ( ) [ : <NUM_LIT:2> ] <EOL> raise va <EOL> except : <EOL> ( ty , va ) = sys . exc_info ( ) [ : <NUM_LIT:2> ] <EOL> raise dns . exception . SyntaxError ( "<STR_LIT>" % ( str ( ty ) , str ( va ) ) ) <EOL> rd . choose_relativity ( self . zone . origin , self . relativize ) <EOL> covers = rd . covers ( ) <EOL> rds = n . find_rdataset ( rdclass , rdtype , covers , True ) <EOL> rds . add ( rd , ttl ) <EOL> def read ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> while <NUM_LIT:1> : <EOL> token = self . tok . get ( True , True ) . unescape ( ) <EOL> if token . is_eof ( ) : <EOL> if not self . current_file is None : <EOL> self . current_file . close ( ) <EOL> if len ( self . saved_state ) > <NUM_LIT:0> : <EOL> ( self . tok , <EOL> self . current_origin , <EOL> self . last_name , <EOL> self . current_file , <EOL> self . ttl ) = self . saved_state . pop ( - <NUM_LIT:1> ) <EOL> continue <EOL> break <EOL> elif token . is_eol ( ) : <EOL> continue <EOL> elif token . is_comment ( ) : <EOL> self . tok . get_eol ( ) <EOL> continue <EOL> elif token . value [ <NUM_LIT:0> ] == '<STR_LIT:$>' : <EOL> u = token . value . upper ( ) <EOL> if u == '<STR_LIT>' : <EOL> token = self . tok . get ( ) <EOL> if not token . is_identifier ( ) : <EOL> raise dns . exception . SyntaxError ( "<STR_LIT>" ) <EOL> self . ttl = dns . ttl . from_text ( token . value ) <EOL> self . tok . get_eol ( ) <EOL> elif u == '<STR_LIT>' : <EOL> self . current_origin = self . tok . get_name ( ) <EOL> self . tok . get_eol ( ) <EOL> if self . zone . origin is None : <EOL> self . zone . origin = self . current_origin <EOL> elif u == '<STR_LIT>' and self . allow_include : <EOL> token = self . tok . get ( ) <EOL> if not token . is_quoted_string ( ) : <EOL> raise dns . exception . SyntaxError ( "<STR_LIT>" ) <EOL> filename = token . value <EOL> token = self . tok . get ( ) <EOL> if token . is_identifier ( ) : <EOL> new_origin = dns . name . from_text ( token . value , self . current_origin ) <EOL> self . tok . get_eol ( ) <EOL> elif not token . is_eol_or_eof ( ) : <EOL> raise dns . exception . SyntaxError ( "<STR_LIT>" ) <EOL> else : <EOL> new_origin = self . current_origin <EOL> self . saved_state . append ( ( self . tok , <EOL> self . current_origin , <EOL> self . last_name , <EOL> self . current_file , <EOL> self . ttl ) ) <EOL> self . current_file = file ( filename , '<STR_LIT:r>' ) <EOL> self . tok = dns . tokenizer . Tokenizer ( self . current_file , <EOL> filename ) <EOL> self . current_origin = new_origin <EOL> else : <EOL> raise dns . exception . SyntaxError ( "<STR_LIT>" + u + "<STR_LIT:'>" ) <EOL> continue <EOL> self . tok . unget ( token ) <EOL> self . _rr_line ( ) <EOL> except dns . exception . SyntaxError , detail : <EOL> ( filename , line_number ) = self . tok . where ( ) <EOL> if detail is None : <EOL> detail = "<STR_LIT>" <EOL> raise dns . exception . SyntaxError ( "<STR_LIT>" % ( filename , line_number , detail ) ) <EOL> if self . check_origin : <EOL> self . zone . check_origin ( ) <EOL> def from_text ( text , origin = None , rdclass = dns . rdataclass . IN , <EOL> relativize = True , zone_factory = Zone , filename = None , <EOL> allow_include = False , check_origin = True ) : <EOL> """<STR_LIT>""" <EOL> if filename is None : <EOL> filename = '<STR_LIT>' <EOL> tok = dns . tokenizer . Tokenizer ( text , filename ) <EOL> reader = _MasterReader ( tok , origin , rdclass , relativize , zone_factory , <EOL> allow_include = allow_include , <EOL> check_origin = check_origin ) <EOL> reader . read ( ) <EOL> return reader . zone <EOL> def from_file ( f , origin = None , rdclass = dns . rdataclass . IN , <EOL> relativize = True , zone_factory = Zone , filename = None , <EOL> allow_include = True , check_origin = True ) : <EOL> """<STR_LIT>""" <EOL> if sys . hexversion >= <NUM_LIT> : <EOL> str_type = basestring <EOL> opts = '<STR_LIT>' <EOL> else : <EOL> str_type = str <EOL> opts = '<STR_LIT:r>' <EOL> if isinstance ( f , str_type ) : <EOL> if filename is None : <EOL> filename = f <EOL> f = file ( f , opts ) <EOL> want_close = True <EOL> else : <EOL> if filename is None : <EOL> filename = '<STR_LIT>' <EOL> want_close = False <EOL> try : <EOL> z = from_text ( f , origin , rdclass , relativize , zone_factory , <EOL> filename , allow_include , check_origin ) <EOL> finally : <EOL> if want_close : <EOL> f . close ( ) <EOL> return z <EOL> def from_xfr ( xfr , zone_factory = Zone , relativize = True ) : <EOL> """<STR_LIT>""" <EOL> z = None <EOL> for r in xfr : <EOL> if z is None : <EOL> if relativize : <EOL> origin = r . origin <EOL> else : <EOL> origin = r . answer [ <NUM_LIT:0> ] . name <EOL> rdclass = r . answer [ <NUM_LIT:0> ] . rdclass <EOL> z = zone_factory ( origin , rdclass , relativize = relativize ) <EOL> for rrset in r . answer : <EOL> znode = z . nodes . get ( rrset . name ) <EOL> if not znode : <EOL> znode = z . node_factory ( ) <EOL> z . nodes [ rrset . name ] = znode <EOL> zrds = znode . find_rdataset ( rrset . rdclass , rrset . rdtype , <EOL> rrset . covers , True ) <EOL> zrds . update_ttl ( rrset . ttl ) <EOL> for rd in rrset : <EOL> rd . choose_relativity ( z . origin , relativize ) <EOL> zrds . add ( rd ) <EOL> z . check_origin ( ) <EOL> return z </s>
