<s> __all__ = [ <EOL> '<STR_LIT>' <EOL> ] <EOL> import copy <EOL> from libcloud . common . base import JsonResponse , ConnectionUserAndKey <EOL> from libcloud . common . types import InvalidCredsError , LibcloudError <EOL> from libcloud . utils . py3 import httplib <EOL> from libcloud . dns . base import DNSDriver , Zone , Record <EOL> from libcloud . dns . types import Provider , RecordType <EOL> from libcloud . dns . types import ZoneDoesNotExistError , RecordDoesNotExistError <EOL> API_URL = '<STR_LIT>' <EOL> API_HOST = '<STR_LIT>' <EOL> API_PATH = '<STR_LIT>' <EOL> ZONE_EXTRA_ATTRIBUTES = [ <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> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> RECORD_EXTRA_ATTRIBUTES = [ <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> class CloudFlareDNSResponse ( JsonResponse ) : <EOL> def success ( self ) : <EOL> return self . status in [ httplib . OK , httplib . CREATED , httplib . ACCEPTED ] <EOL> def parse_body ( self ) : <EOL> body = super ( CloudFlareDNSResponse , self ) . parse_body ( ) <EOL> body = body or { } <EOL> result = body . get ( '<STR_LIT:result>' , None ) <EOL> error_code = body . get ( '<STR_LIT>' , None ) <EOL> msg = body . get ( '<STR_LIT>' , None ) <EOL> is_error_result = result == '<STR_LIT:error>' <EOL> context = self . connection . context or { } <EOL> context_record_id = context . get ( '<STR_LIT>' , None ) <EOL> context_zone_domain = context . get ( '<STR_LIT>' , None ) <EOL> if ( is_error_result and '<STR_LIT>' in msg . lower ( ) and <EOL> context_record_id ) : <EOL> raise RecordDoesNotExistError ( value = msg , <EOL> driver = self . connection . driver , <EOL> record_id = context_record_id ) <EOL> elif ( is_error_result and '<STR_LIT>' in msg . lower ( ) and <EOL> context_zone_domain ) : <EOL> raise ZoneDoesNotExistError ( value = msg , <EOL> driver = self . connection . driver , <EOL> zone_id = context_zone_domain ) <EOL> if error_code == '<STR_LIT>' : <EOL> raise InvalidCredsError ( msg ) <EOL> elif result == '<STR_LIT:error>' or error_code is not None : <EOL> msg = '<STR_LIT>' % ( self . body ) <EOL> raise LibcloudError ( value = msg , driver = self . connection . driver ) <EOL> return body <EOL> class CloudFlareDNSConnection ( ConnectionUserAndKey ) : <EOL> host = API_HOST <EOL> secure = True <EOL> responseCls = CloudFlareDNSResponse <EOL> def request ( self , action , params = None , data = None , headers = None , <EOL> method = '<STR_LIT:GET>' ) : <EOL> params = params or { } <EOL> data = data or { } <EOL> base_params = { <EOL> '<STR_LIT:email>' : self . user_id , <EOL> '<STR_LIT>' : self . key , <EOL> '<STR_LIT:a>' : action <EOL> } <EOL> params = copy . deepcopy ( params ) <EOL> params . update ( base_params ) <EOL> return super ( CloudFlareDNSConnection , self ) . request ( action = API_PATH , <EOL> params = params , <EOL> data = None , <EOL> method = method , <EOL> headers = headers ) <EOL> class CloudFlareDNSDriver ( DNSDriver ) : <EOL> type = Provider . CLOUDFLARE <EOL> name = '<STR_LIT>' <EOL> website = '<STR_LIT>' <EOL> connectionCls = CloudFlareDNSConnection <EOL> RECORD_TYPE_MAP = { <EOL> RecordType . A : '<STR_LIT:A>' , <EOL> RecordType . AAAA : '<STR_LIT>' , <EOL> RecordType . CNAME : '<STR_LIT>' , <EOL> RecordType . MX : '<STR_LIT>' , <EOL> RecordType . TXT : '<STR_LIT>' , <EOL> RecordType . SPF : '<STR_LIT>' , <EOL> RecordType . NS : '<STR_LIT>' , <EOL> RecordType . SRV : '<STR_LIT>' , <EOL> RecordType . URL : '<STR_LIT>' <EOL> } <EOL> def iterate_zones ( self ) : <EOL> result = self . connection . request ( action = '<STR_LIT>' ) . object <EOL> zones = self . _to_zones ( data = result [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> return zones <EOL> def iterate_records ( self , zone ) : <EOL> params = { '<STR_LIT:z>' : zone . domain } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> data = resp . object [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> records = self . _to_records ( zone = zone , data = data ) <EOL> return records <EOL> def get_zone ( self , zone_id ) : <EOL> zones = self . list_zones ( ) <EOL> try : <EOL> zone = [ z for z in zones if z . id == zone_id ] [ <NUM_LIT:0> ] <EOL> except IndexError : <EOL> raise ZoneDoesNotExistError ( value = '<STR_LIT>' , driver = self , zone_id = zone_id ) <EOL> return zone <EOL> def create_record ( self , name , zone , type , data , extra = None ) : <EOL> extra = extra or { } <EOL> params = { '<STR_LIT:name>' : name , '<STR_LIT:z>' : zone . domain , '<STR_LIT:type>' : type , <EOL> '<STR_LIT:content>' : data } <EOL> params [ '<STR_LIT>' ] = extra . get ( '<STR_LIT>' , <NUM_LIT> ) <EOL> if '<STR_LIT>' in extra : <EOL> params [ '<STR_LIT>' ] = extra [ '<STR_LIT>' ] <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> item = resp . object [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> record = self . _to_record ( zone = zone , item = item ) <EOL> return record <EOL> def update_record ( self , record , name = None , type = None , data = None , <EOL> extra = None ) : <EOL> extra = extra or { } <EOL> params = { '<STR_LIT:z>' : record . zone . domain , '<STR_LIT:id>' : record . id } <EOL> params [ '<STR_LIT:name>' ] = name or record . name <EOL> params [ '<STR_LIT:type>' ] = type or record . type <EOL> params [ '<STR_LIT:content>' ] = data or record . data <EOL> params [ '<STR_LIT>' ] = extra . get ( '<STR_LIT>' , None ) or record . extra [ '<STR_LIT>' ] <EOL> self . connection . set_context ( { '<STR_LIT>' : record . zone . domain } ) <EOL> self . connection . set_context ( { '<STR_LIT>' : record . id } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> item = resp . object [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> record = self . _to_record ( zone = record . zone , item = item ) <EOL> return record <EOL> def delete_record ( self , record ) : <EOL> params = { '<STR_LIT:z>' : record . zone . domain , '<STR_LIT:id>' : record . id } <EOL> self . connection . set_context ( { '<STR_LIT>' : record . zone . domain } ) <EOL> self . connection . set_context ( { '<STR_LIT>' : record . id } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_get_zone_stats ( self , zone , interval = <NUM_LIT:30> ) : <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT>' : interval } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object [ '<STR_LIT>' ] [ '<STR_LIT:result>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> return result <EOL> def ex_zone_check ( self , zones ) : <EOL> zone_domains = [ zone . domain for zone in zones ] <EOL> zone_domains = '<STR_LIT:U+002C>' . join ( zone_domains ) <EOL> params = { '<STR_LIT>' : zone_domains } <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> return result <EOL> def ex_get_ip_threat_score ( self , ip ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : ip } <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object [ '<STR_LIT>' ] <EOL> return result <EOL> def ex_get_zone_settings ( self , zone ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object [ '<STR_LIT>' ] [ '<STR_LIT:result>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> return result <EOL> def ex_set_zone_security_level ( self , zone , level ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:v>' : level } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_set_zone_cache_level ( self , zone , level ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:v>' : level } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_enable_development_mode ( self , zone ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:v>' : <NUM_LIT:1> } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_disable_development_mode ( self , zone ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:v>' : <NUM_LIT:0> } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_purge_cached_files ( self , zone ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:v>' : <NUM_LIT:1> } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_purge_cached_file ( self , zone , url ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:url>' : url } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_whitelist_ip ( self , zone , ip ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:key>' : ip } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_blacklist_ip ( self , zone , ip ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:key>' : ip } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_unlist_ip ( self , zone , ip ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:key>' : ip } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_enable_ipv6_support ( self , zone ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:v>' : <NUM_LIT:3> } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def ex_disable_ipv6_support ( self , zone ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:z>' : zone . domain , '<STR_LIT:v>' : <NUM_LIT:0> } <EOL> self . connection . set_context ( { '<STR_LIT>' : zone . domain } ) <EOL> resp = self . connection . request ( action = '<STR_LIT>' , params = params ) <EOL> result = resp . object <EOL> return result . get ( '<STR_LIT:result>' , None ) == '<STR_LIT:success>' <EOL> def _to_zones ( self , data ) : <EOL> zones = [ ] <EOL> for item in data : <EOL> zone = self . _to_zone ( item = item ) <EOL> zones . append ( zone ) <EOL> return zones <EOL> def _to_zone ( self , item ) : <EOL> type = '<STR_LIT>' <EOL> extra = { } <EOL> extra [ '<STR_LIT>' ] = item . get ( '<STR_LIT>' , { } ) <EOL> extra [ '<STR_LIT>' ] = item . get ( '<STR_LIT>' , { } ) <EOL> extra [ '<STR_LIT>' ] = item . get ( '<STR_LIT>' , { } ) <EOL> for attribute in ZONE_EXTRA_ATTRIBUTES : <EOL> value = item . get ( attribute , None ) <EOL> extra [ attribute ] = value <EOL> zone = Zone ( id = str ( item [ '<STR_LIT>' ] ) , domain = item [ '<STR_LIT>' ] , <EOL> type = type , ttl = None , driver = self , extra = extra ) <EOL> return zone <EOL> def _to_records ( self , zone , data ) : <EOL> records = [ ] <EOL> for item in data : <EOL> record = self . _to_record ( zone = zone , item = item ) <EOL> records . append ( record ) <EOL> return records <EOL> def _to_record ( self , zone , item ) : <EOL> name = self . _get_record_name ( item = item ) <EOL> type = item [ '<STR_LIT:type>' ] <EOL> data = item [ '<STR_LIT:content>' ] <EOL> if item . get ( '<STR_LIT>' , None ) : <EOL> ttl = int ( item [ '<STR_LIT>' ] ) <EOL> else : <EOL> ttl = None <EOL> extra = { } <EOL> extra [ '<STR_LIT>' ] = ttl <EOL> extra [ '<STR_LIT>' ] = item . get ( '<STR_LIT>' , { } ) <EOL> for attribute in RECORD_EXTRA_ATTRIBUTES : <EOL> value = item . get ( attribute , None ) <EOL> extra [ attribute ] = value <EOL> record = Record ( id = str ( item [ '<STR_LIT>' ] ) , name = name , type = type , <EOL> data = data , zone = zone , driver = self , ttl = ttl , <EOL> extra = extra ) <EOL> return record <EOL> def _get_record_name ( self , item ) : <EOL> name = item [ '<STR_LIT:name>' ] . replace ( '<STR_LIT:.>' + item [ '<STR_LIT>' ] , '<STR_LIT>' ) or None <EOL> if name : <EOL> name = name . replace ( item [ '<STR_LIT>' ] , '<STR_LIT>' ) or None <EOL> return name </s>
