Server IP : 85.214.239.14 / Your IP : 18.117.94.221 Web Server : Apache/2.4.62 (Debian) System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.18 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : OFF | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /proc/3/root/lib/python3/dist-packages/libcloud/dns/drivers/ |
Upload File : |
# Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License.You may obtain a copy of the License at # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. __all__ = [ 'GoDaddyDNSDriver' ] try: import simplejson as json except Exception: import json from libcloud.common.base import ConnectionKey, JsonResponse from libcloud.common.types import LibcloudError from libcloud.utils.py3 import httplib from libcloud.dns.types import Provider, RecordType, RecordDoesNotExistError from libcloud.dns.base import DNSDriver, Zone, Record API_HOST = 'api.godaddy.com' VALID_RECORD_EXTRA_PARAMS = ['prio', 'ttl'] class GoDaddyDNSException(LibcloudError): def __init__(self, code, message): self.code = code self.message = message self.args = (code, message) def __str__(self): return self.__repr__() def __repr__(self): return ('<GoDaddyDNSException in %s: %s>' % (self.code, self.message)) class GoDaddyDNSResponse(JsonResponse): valid_response_codes = [httplib.OK, httplib.ACCEPTED, httplib.CREATED, httplib.NO_CONTENT] def parse_body(self): if not self.body: return None # json.loads doesn't like the regex expressions used in godaddy schema self.body = self.body.replace('\\.', '\\\\.') data = json.loads(self.body) return data def parse_error(self): data = self.parse_body() raise GoDaddyDNSException(code=data['code'], message=data['message']) def success(self): return self.status in self.valid_response_codes class GoDaddyDNSConnection(ConnectionKey): responseCls = GoDaddyDNSResponse host = API_HOST allow_insecure = False def __init__(self, key, secret, secure=True, shopper_id=None, host=None, port=None, url=None, timeout=None, proxy_url=None, backoff=None, retry_delay=None): super(GoDaddyDNSConnection, self).__init__( key, secure=secure, host=host, port=port, url=url, timeout=timeout, proxy_url=proxy_url, backoff=backoff, retry_delay=retry_delay) self.key = key self.secret = secret self.shopper_id = shopper_id def add_default_headers(self, headers): if self.shopper_id is not None: headers['X-Shopper-Id'] = self.shopper_id headers['Content-type'] = 'application/json' headers['Authorization'] = "sso-key %s:%s" % \ (self.key, self.secret) return headers class GoDaddyDNSDriver(DNSDriver): """ A driver for GoDaddy DNS. This is for customers of GoDaddy who wish to purchase, update existing domains and manage records for DNS zones owned by GoDaddy NS servers. """ type = Provider.GODADDY name = 'GoDaddy DNS' website = 'https://www.godaddy.com/' connectionCls = GoDaddyDNSConnection RECORD_TYPE_MAP = { RecordType.A: 'A', RecordType.AAAA: 'AAAA', RecordType.CNAME: 'CNAME', RecordType.MX: 'MX', RecordType.NS: 'SPF', RecordType.SRV: 'SRV', RecordType.TXT: 'TXT', } def __init__(self, shopper_id, key, secret, secure=True, host=None, port=None): """ Instantiate a new `GoDaddyDNSDriver` :param shopper_id: Your customer ID or shopper ID with GoDaddy :type shopper_id: ``str`` :param key: Your access key from developer.godaddy.com :type key: ``str`` :param secret: Your access key secret :type secret: ``str`` """ self.shopper_id = shopper_id super(GoDaddyDNSDriver, self).__init__(key=key, secret=secret, secure=secure, host=host, port=port, shopper_id=str(shopper_id)) def list_zones(self): """ Return a list of zones (purchased domains) :return: ``list`` of :class:`Zone` """ result = self.connection.request( '/v1/domains/').object zones = self._to_zones(result) return zones def list_records(self, zone): """ Return a list of records for the provided zone. :param zone: Zone to list records for. :type zone: :class:`Zone` :return: ``list`` of :class:`Record` """ result = self.connection.request( '/v1/domains/%s/records' % (zone.domain)).object records = self._to_records(items=result, zone=zone) return records def create_record(self, name, zone, type, data, extra=None): """ Create a new record. :param name: Record name without the domain name (e.g. www). Note: If you want to create a record for a base domain name, you should specify empty string ('') for this argument. :type name: ``str`` :param zone: Zone where the requested record is created. :type zone: :class:`Zone` :param type: DNS record type (A, AAAA, ...). :type type: :class:`RecordType` :param data: Data for the record (depends on the record type). :type data: ``str`` :param extra: Extra attributes (driver specific). (optional) :type extra: ``dict`` :rtype: :class:`Record` """ new_record = self._format_record(name, type, data, extra) self.connection.request( '/v1/domains/%s/records' % (zone.domain), method='PATCH', data=json.dumps([new_record])) id = self._get_id_of_record(name, type) return Record( id=id, name=name, type=type, data=data, zone=zone, driver=self, ttl=new_record['ttl'], extra=extra) def update_record(self, record, name, type, data, extra=None): """ Update an existing record. :param record: Record to update. :type record: :class:`Record` :param name: Record name without the domain name (e.g. www). Note: If you want to create a record for a base domain name, you should specify empty string ('') for this argument. :type name: ``str`` :param type: DNS record type (A, AAAA, ...). :type type: :class:`RecordType` :param data: Data for the record (depends on the record type). :type data: ``str`` :param extra: (optional) Extra attributes (driver specific). :type extra: ``dict`` :rtype: :class:`Record` """ new_record = self._format_record(name, type, data, extra) self.connection.request( '/v1/domains/%s/records/%s/%s' % (record.zone.domain, record.type, record.name), method='PUT', data=json.dumps([new_record])) id = self._get_id_of_record(name, type) return Record( id=id, name=name, type=type, data=data, zone=record.zone, driver=self, ttl=new_record['ttl'], extra=extra) def get_record(self, zone_id, record_id): """ Return a Record instance. :param zone_id: ID of the required zone :type zone_id: ``str`` :param record_id: ID of the required record :type record_id: ``str`` :rtype: :class:`Record` """ parts = record_id.split(':') result = self.connection.request( '/v1/domains/%s/records/%s/%s' % ( zone_id, parts[1], parts[0])).object if len(result) == 0: raise RecordDoesNotExistError(record_id, driver=self, record_id=record_id) return self._to_record(result[0], self.get_zone(zone_id)) def get_zone(self, zone_id): """ Get a zone (by domain) :param zone_id: The domain, not the ID :type zone_id: ``str`` :rtype: :class:`Zone` """ result = self.connection.request( '/v1/domains/%s/' % zone_id).object zone = self._to_zone(result) return zone def delete_zone(self, zone): """ Delete a zone. Note: This will CANCEL a purchased domain :param zone: Zone to delete. :type zone: :class:`Zone` :rtype: ``bool`` """ self.connection.request( '/v1/domains/%s' % (zone.domain), method='DELETE') # no error means ok return True def ex_check_availability(self, domain, for_transfer=False): """ Check the availability of the domain :param domain: the domain name e.g. wazzlewobbleflooble.com :type domain: ``str`` :param for_transfer: Check if domain is available for transfer :type for_transfer: ``bool`` :rtype: `list` of :class:`GoDaddyAvailability` """ result = self.connection.request( '/v1/domains/available', method='GET', params={ 'domain': domain, 'forTransfer': str(for_transfer) } ).object return GoDaddyAvailability( domain=result['domain'], available=result['available'], price=result['price'], currency=result['currency'], period=result['period'] ) def ex_list_tlds(self): """ List available TLDs for sale :rtype: ``list`` of :class:`GoDaddyTLD` """ result = self.connection.request( '/v1/domains/tlds', method='GET' ).object return self._to_tlds(result) def ex_get_purchase_schema(self, tld): """ Get the schema that needs completing to purchase a new domain Use this in conjunction with ex_purchase_domain :param tld: The top level domain e.g com, eu, uk :type tld: ``str`` :rtype: `dict` the JSON Schema """ result = self.connection.request( '/v1/domains/purchase/schema/%s' % tld, method='GET' ).object return result def ex_get_agreements(self, tld, privacy=True): """ Get the legal agreements for a tld Use this in conjunction with ex_purchase_domain :param tld: The top level domain e.g com, eu, uk :type tld: ``str`` :rtype: `dict` the JSON Schema """ result = self.connection.request( '/v1/domains/agreements', params={ 'tlds': tld, 'privacy': str(privacy) }, method='GET' ).object agreements = [] for item in result: agreements.append( GoDaddyLegalAgreement( agreement_key=item['agreementKey'], title=item['title'], url=item['url'], content=item['content'])) return agreements def ex_purchase_domain(self, purchase_request): """ Purchase a domain with GoDaddy :param purchase_request: The completed document from ex_get_purchase_schema :type purchase_request: ``dict`` :rtype: :class:`GoDaddyDomainPurchaseResponse` Your order """ result = self.connection.request( '/v1/domains/purchase', data=purchase_request, method='POST' ).object return GoDaddyDomainPurchaseResponse( order_id=result['orderId'], item_count=result['itemCount'], total=result['total'], currency=result['currency'] ) def _format_record(self, name, type, data, extra): if extra is None: extra = {} new_record = {} if type == RecordType.SRV: new_record = { 'type': type, 'name': name, 'data': data, 'priority': 1, 'ttl': extra.get('ttl', 5), 'service': extra.get('service', ''), 'protocol': extra.get('protocol', ''), 'port': extra.get('port', ''), 'weight': extra.get('weight', '1') } else: new_record = { 'type': type, 'name': name, 'data': data, 'ttl': extra.get('ttl', 5) } if type == RecordType.MX: new_record['priority'] = 1 return new_record def _to_zones(self, items): zones = [] for item in items: zones.append(self._to_zone(item)) return zones def _to_zone(self, item): extra = {"expires": item['expires']} zone = Zone(id=item['domainId'], domain=item['domain'], type='master', ttl=None, driver=self, extra=extra) return zone def _to_records(self, items, zone=None): records = [] for item in items: records.append(self._to_record(item=item, zone=zone)) return records def _to_record(self, item, zone=None): ttl = item['ttl'] type = self._string_to_record_type(item['type']) name = item['name'] id = self._get_id_of_record(name, type) record = Record(id=id, name=name, type=type, data=item['data'], zone=zone, driver=self, ttl=ttl) return record def _to_tlds(self, items): tlds = [] for item in items: tlds.append(self._to_tld(item)) return tlds def _to_tld(self, item): return GoDaddyTLD( name=item['name'], tld_type=item['type'] ) def _get_id_of_record(self, name, type): return '%s:%s' % (name, type) def _ex_connection_class_kwargs(self): return {'shopper_id': self.shopper_id} class GoDaddyAvailability(object): def __init__(self, domain, available, price, currency, period): self.domain = domain self.available = bool(available) # currency comes in micro-units, convert to dollars. self.price = float(price) / 1000000 self.currency = currency self.period = int(period) class GoDaddyTLD(object): def __init__(self, name, tld_type): self.name = name self.type = tld_type class GoDaddyDomainPurchaseResponse(object): def __init__(self, order_id, item_count, total, currency): self.order_id = order_id self.item_count = item_count self.total = total self.current = currency class GoDaddyLegalAgreement(object): def __init__(self, agreement_key, title, url, content): self.agreement_key = agreement_key self.title = title self.url = url self.content = content