<s> import netaddr <EOL> from designateclient import exceptions as d_exc <EOL> from designateclient . v2 import client as d_client <EOL> from keystoneclient . auth . identity . generic import password <EOL> from keystoneclient . auth import token_endpoint <EOL> from keystoneclient import session <EOL> from oslo_config import cfg <EOL> from oslo_log import log <EOL> from neutron . _i18n import _ <EOL> from neutron . extensions import dns <EOL> from neutron . services . externaldns import driver <EOL> IPV4_PTR_ZONE_PREFIX_MIN_SIZE = <NUM_LIT:8> <EOL> IPV4_PTR_ZONE_PREFIX_MAX_SIZE = <NUM_LIT> <EOL> IPV6_PTR_ZONE_PREFIX_MIN_SIZE = <NUM_LIT:4> <EOL> IPV6_PTR_ZONE_PREFIX_MAX_SIZE = <NUM_LIT> <EOL> LOG = log . getLogger ( __name__ ) <EOL> _SESSION = None <EOL> designate_opts = [ <EOL> cfg . StrOpt ( '<STR_LIT:url>' , <EOL> help = _ ( '<STR_LIT>' ) ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> help = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> help = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> secret = True ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> help = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> help = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> help = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) , <EOL> cfg . BoolOpt ( '<STR_LIT>' , default = True , <EOL> help = _ ( '<STR_LIT>' ) ) , <EOL> cfg . IntOpt ( '<STR_LIT>' , default = <NUM_LIT> , <EOL> help = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) ) , <EOL> cfg . IntOpt ( '<STR_LIT>' , default = <NUM_LIT> , <EOL> help = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , default = '<STR_LIT>' , <EOL> help = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) ) , <EOL> ] <EOL> DESIGNATE_GROUP = '<STR_LIT>' <EOL> CONF = cfg . CONF <EOL> CONF . register_opts ( designate_opts , DESIGNATE_GROUP ) <EOL> def get_clients ( context ) : <EOL> global _SESSION <EOL> if not _SESSION : <EOL> _SESSION = session . Session ( ) <EOL> auth = token_endpoint . Token ( CONF . designate . url , context . auth_token ) <EOL> client = d_client . Client ( session = _SESSION , auth = auth ) <EOL> admin_auth = password . Password ( <EOL> auth_url = CONF . designate . admin_auth_url , <EOL> username = CONF . designate . admin_username , <EOL> password = CONF . designate . admin_password , <EOL> tenant_name = CONF . designate . admin_tenant_name , <EOL> tenant_id = CONF . designate . admin_tenant_id ) <EOL> admin_client = d_client . Client ( session = _SESSION , auth = admin_auth ) <EOL> return client , admin_client <EOL> class Designate ( driver . ExternalDNSService ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> ipv4_ptr_zone_size = CONF . designate . ipv4_ptr_zone_prefix_size <EOL> ipv6_ptr_zone_size = CONF . designate . ipv6_ptr_zone_prefix_size <EOL> if ( ipv4_ptr_zone_size < IPV4_PTR_ZONE_PREFIX_MIN_SIZE or <EOL> ipv4_ptr_zone_size > IPV4_PTR_ZONE_PREFIX_MAX_SIZE or <EOL> ( ipv4_ptr_zone_size % <NUM_LIT:8> ) != <NUM_LIT:0> ) : <EOL> raise dns . InvalidPTRZoneConfiguration ( <EOL> parameter = '<STR_LIT>' , number = '<STR_LIT>' , <EOL> maximum = str ( IPV4_PTR_ZONE_PREFIX_MAX_SIZE ) , <EOL> minimum = str ( IPV4_PTR_ZONE_PREFIX_MIN_SIZE ) ) <EOL> if ( ipv6_ptr_zone_size < IPV6_PTR_ZONE_PREFIX_MIN_SIZE or <EOL> ipv6_ptr_zone_size > IPV6_PTR_ZONE_PREFIX_MAX_SIZE or <EOL> ( ipv6_ptr_zone_size % <NUM_LIT:4> ) != <NUM_LIT:0> ) : <EOL> raise dns . InvalidPTRZoneConfiguration ( <EOL> parameter = '<STR_LIT>' , number = '<STR_LIT:4>' , <EOL> maximum = str ( IPV6_PTR_ZONE_PREFIX_MAX_SIZE ) , <EOL> minimum = str ( IPV6_PTR_ZONE_PREFIX_MIN_SIZE ) ) <EOL> def create_record_set ( self , context , dns_domain , dns_name , records ) : <EOL> designate , designate_admin = get_clients ( context ) <EOL> v4 , v6 = self . _classify_records ( records ) <EOL> try : <EOL> if v4 : <EOL> designate . recordsets . create ( dns_domain , dns_name , '<STR_LIT:A>' , v4 ) <EOL> if v6 : <EOL> designate . recordsets . create ( dns_domain , dns_name , '<STR_LIT>' , v6 ) <EOL> except d_exc . NotFound : <EOL> raise dns . DNSDomainNotFound ( dns_domain = dns_domain ) <EOL> except d_exc . Conflict : <EOL> raise dns . DuplicateRecordSet ( dns_name = dns_name ) <EOL> if not CONF . designate . allow_reverse_dns_lookup : <EOL> return <EOL> recordset_name = '<STR_LIT>' % ( dns_name , dns_domain ) <EOL> ptr_zone_email = '<STR_LIT>' % dns_domain [ : - <NUM_LIT:1> ] <EOL> if CONF . designate . ptr_zone_email : <EOL> ptr_zone_email = CONF . designate . ptr_zone_email <EOL> for record in records : <EOL> in_addr_name = netaddr . IPAddress ( record ) . reverse_dns <EOL> in_addr_zone_name = self . _get_in_addr_zone_name ( in_addr_name ) <EOL> in_addr_zone_description = ( <EOL> '<STR_LIT>' % <EOL> '<STR_LIT:.>' . join ( in_addr_name . split ( '<STR_LIT:.>' ) [ - <NUM_LIT:3> : ] ) ) <EOL> try : <EOL> designate_admin . recordsets . create ( in_addr_zone_name , <EOL> in_addr_name , '<STR_LIT>' , <EOL> [ recordset_name ] ) <EOL> except d_exc . NotFound : <EOL> designate_admin . zones . create ( <EOL> in_addr_zone_name , email = ptr_zone_email , <EOL> description = in_addr_zone_description ) <EOL> designate_admin . recordsets . create ( in_addr_zone_name , <EOL> in_addr_name , '<STR_LIT>' , <EOL> [ recordset_name ] ) <EOL> def _classify_records ( self , records ) : <EOL> v4 = [ ] <EOL> v6 = [ ] <EOL> for record in records : <EOL> if netaddr . IPAddress ( record ) . version == <NUM_LIT:4> : <EOL> v4 . append ( record ) <EOL> else : <EOL> v6 . append ( record ) <EOL> return v4 , v6 <EOL> def _get_in_addr_zone_name ( self , in_addr_name ) : <EOL> units = self . _get_bytes_or_nybles_to_skip ( in_addr_name ) <EOL> return '<STR_LIT:.>' . join ( in_addr_name . split ( '<STR_LIT:.>' ) [ units : ] ) <EOL> def _get_bytes_or_nybles_to_skip ( self , in_addr_name ) : <EOL> if '<STR_LIT>' in in_addr_name : <EOL> return int ( ( <NUM_LIT:32> - CONF . designate . ipv4_ptr_zone_prefix_size ) / <NUM_LIT:8> ) <EOL> return int ( ( <NUM_LIT> - CONF . designate . ipv6_ptr_zone_prefix_size ) / <NUM_LIT:4> ) <EOL> def delete_record_set ( self , context , dns_domain , dns_name , records ) : <EOL> designate , designate_admin = get_clients ( context ) <EOL> ids_to_delete = self . _get_ids_ips_to_delete ( <EOL> dns_domain , '<STR_LIT>' % ( dns_name , dns_domain ) , records , designate ) <EOL> for _id in ids_to_delete : <EOL> designate . recordsets . delete ( dns_domain , _id ) <EOL> if not CONF . designate . allow_reverse_dns_lookup : <EOL> return <EOL> for record in records : <EOL> in_addr_name = netaddr . IPAddress ( record ) . reverse_dns <EOL> in_addr_zone_name = self . _get_in_addr_zone_name ( in_addr_name ) <EOL> designate_admin . recordsets . delete ( in_addr_zone_name , in_addr_name ) <EOL> def _get_ids_ips_to_delete ( self , dns_domain , name , records , <EOL> designate_client ) : <EOL> try : <EOL> recordsets = designate_client . recordsets . list ( <EOL> dns_domain , criterion = { "<STR_LIT:name>" : "<STR_LIT:%s>" % name } ) <EOL> except d_exc . NotFound : <EOL> raise dns . DNSDomainNotFound ( dns_domain = dns_domain ) <EOL> ids = [ rec [ '<STR_LIT:id>' ] for rec in recordsets ] <EOL> ips = [ ip for rec in recordsets for ip in rec [ '<STR_LIT>' ] ] <EOL> if set ( ips ) != set ( records ) : <EOL> raise dns . DuplicateRecordSet ( dns_name = name ) <EOL> return ids </s>
