| # 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. |
| |
| import re |
| from typing import Dict, List |
| from xml.etree import ElementTree as ET # noqa |
| |
| from libcloud.common.base import XmlResponse, ConnectionUserAndKey |
| |
| # API HOST to connect |
| API_HOST = "durabledns.com" |
| |
| |
| def _schema_builder(urn_nid, method, attributes): |
| """ |
| Return a xml schema used to do an API request. |
| |
| :param urn_nid: API urn namespace id. |
| :type urn_nid: type: ``str`` |
| |
| :param method: API method. |
| :type method: type: ``str`` |
| |
| :param attributes: List of attributes to include. |
| :type attributes: ``list`` of ``str`` |
| |
| rtype: :class:`Element` |
| """ |
| soap = ET.Element("soap:Body", {"xmlns:m": "https://durabledns.com/services/dns/%s" % method}) |
| urn = ET.SubElement(soap, "urn:{}:{}".format(urn_nid, method)) |
| # Attributes specification |
| for attribute in attributes: |
| ET.SubElement(urn, "urn:{}:{}".format(urn_nid, attribute)) |
| return soap |
| |
| |
| SCHEMA_BUILDER_MAP = { |
| "list_zones": { |
| "urn_nid": "listZoneswsdl", |
| "method": "listZones", |
| "attributes": ["apiuser", "apikey"], |
| }, |
| "list_records": { |
| "urn_nid": "listRecordswsdl", |
| "method": "listRecords", |
| "attributes": ["apiuser", "apikey", "zonename"], |
| }, |
| "get_zone": { |
| "urn_nid": "getZonewsdl", |
| "method": "getZone", |
| "attributes": ["apiuser", "apikey", "zonename"], |
| }, |
| "get_record": { |
| "urn_nid": "getRecordwsdl", |
| "method": "getRecord", |
| "attributes": ["apiuser", "apikey", "zonename", "recordid"], |
| }, |
| "create_zone": { |
| "urn_nid": "createZonewsdl", |
| "method": "createZone", |
| "attributes": [ |
| "apiuser", |
| "apikey", |
| "zonename", |
| "ns", |
| "mbox", |
| "refresh", |
| "retry", |
| "expire", |
| "minimum", |
| "ttl", |
| "xfer", |
| "update_acl", |
| ], |
| }, |
| "create_record": { |
| "urn_nid": "createRecordwsdl", |
| "method": "createRecord", |
| "attributes": [ |
| "apiuser", |
| "apikey", |
| "zonename", |
| "name", |
| "type", |
| "data", |
| "aux", |
| "ttl", |
| "ddns_enabled", |
| ], |
| }, |
| "update_zone": { |
| "urn_nid": "updateZonewsdl", |
| "method": "updateZone", |
| "attributes": [ |
| "apiuser", |
| "apikey", |
| "zonename", |
| "ns", |
| "mbox", |
| "refresh", |
| "retry", |
| "expire", |
| "minimum", |
| "ttl", |
| "xfer", |
| "update_acl", |
| ], |
| }, |
| "update_record": { |
| "urn_nid": "updateRecordwsdl", |
| "method": "updateRecord", |
| "attributes": [ |
| "apiuser", |
| "apikey", |
| "zonename", |
| "id", |
| "name", |
| "aux", |
| "data", |
| "ttl", |
| "ddns_enabled", |
| ], |
| }, |
| "delete_zone": { |
| "urn_nid": "deleteZonewsdl", |
| "method": "deleteZone", |
| "attributes": ["apiuser", "apikey", "zonename"], |
| }, |
| "delete_record": { |
| "urn_nid": "deleteRecordwsdl", |
| "method": "deleteRecord", |
| "attributes": ["apiuser", "apikey", "zonename", "id"], |
| }, |
| } |
| |
| |
| class DurableDNSException(Exception): |
| def __init__(self, code, message): |
| self.code = code |
| self.message = message |
| self.args = (code, message) |
| |
| def __str__(self): |
| return "{} {}".format(self.code, self.message) |
| |
| def __repr__(self): |
| return "DurableDNSException {} {}".format(self.code, self.message) |
| |
| |
| class DurableResponse(XmlResponse): |
| |
| errors = [] # type: List[Dict] |
| objects = [] # type: List[Dict] |
| |
| def __init__(self, response, connection): |
| super().__init__(response=response, connection=connection) |
| |
| self.objects, self.errors = self.parse_body_and_error() |
| if self.errors: |
| raise self._make_excp(self.errors[0]) |
| |
| def parse_body_and_error(self): |
| """ |
| Used to parse body from httplib.HttpResponse object. |
| """ |
| objects = [] |
| errors = [] |
| error_dict = {} |
| extra = {} |
| zone_dict = {} |
| record_dict = {} |
| xml_obj = self.parse_body() |
| |
| # pylint: disable=no-member |
| envelop_body = list(xml_obj)[0] |
| method_resp = list(envelop_body)[0] |
| # parse the xml_obj |
| # handle errors |
| if "Fault" in method_resp.tag: |
| fault = [fault for fault in list(method_resp) if fault.tag == "faultstring"][0] |
| error_dict["ERRORMESSAGE"] = fault.text.strip() |
| error_dict["ERRORCODE"] = self.status |
| errors.append(error_dict) |
| |
| # parsing response from listZonesResponse |
| if "listZonesResponse" in method_resp.tag: |
| answer = list(method_resp)[0] |
| for element in answer: |
| zone_dict["id"] = list(element)[0].text |
| objects.append(zone_dict) |
| # reset the zone_dict |
| zone_dict = {} |
| # parse response from listRecordsResponse |
| if "listRecordsResponse" in method_resp.tag: |
| answer = list(method_resp)[0] |
| for element in answer: |
| for child in list(element): |
| if child.tag == "id": |
| record_dict["id"] = child.text.strip() |
| objects.append(record_dict) |
| # reset the record_dict for later usage |
| record_dict = {} |
| # parse response from getZoneResponse |
| if "getZoneResponse" in method_resp.tag: |
| for child in list(method_resp): |
| if child.tag == "origin": |
| zone_dict["id"] = child.text.strip() |
| zone_dict["domain"] = child.text.strip() |
| elif child.tag == "ttl": |
| zone_dict["ttl"] = int(child.text.strip()) |
| elif child.tag == "retry": |
| extra["retry"] = int(child.text.strip()) |
| elif child.tag == "expire": |
| extra["expire"] = int(child.text.strip()) |
| elif child.tag == "minimum": |
| extra["minimum"] = int(child.text.strip()) |
| else: |
| if child.text: |
| extra[child.tag] = child.text.strip() |
| else: |
| extra[child.tag] = "" |
| zone_dict["extra"] = extra |
| objects.append(zone_dict) |
| # parse response from getRecordResponse |
| if "getRecordResponse" in method_resp.tag: |
| answer = list(method_resp)[0] |
| for child in list(method_resp): |
| if child.tag == "id" and child.text: |
| record_dict["id"] = child.text.strip() |
| elif child.tag == "name" and child.text: |
| record_dict["name"] = child.text.strip() |
| elif child.tag == "type" and child.text: |
| record_dict["type"] = child.text.strip() |
| elif child.tag == "data" and child.text: |
| record_dict["data"] = child.text.strip() |
| elif child.tag == "aux" and child.text: |
| record_dict["aux"] = child.text.strip() |
| elif child.tag == "ttl" and child.text: |
| record_dict["ttl"] = child.text.strip() |
| if not record_dict: |
| error_dict["ERRORMESSAGE"] = "Record does not exist" |
| error_dict["ERRORCODE"] = 404 |
| errors.append(error_dict) |
| objects.append(record_dict) |
| record_dict = {} |
| if "createZoneResponse" in method_resp.tag: |
| answer = list(method_resp)[0] |
| if answer.tag == "return" and answer.text: |
| record_dict["id"] = answer.text.strip() |
| objects.append(record_dict) |
| # catch Record does not exists error when deleting record |
| if "deleteRecordResponse" in method_resp.tag: |
| answer = list(method_resp)[0] |
| if "Record does not exists" in answer.text.strip(): |
| errors.append({"ERRORMESSAGE": answer.text.strip(), "ERRORCODE": self.status}) |
| # parse response in createRecordResponse |
| if "createRecordResponse" in method_resp.tag: |
| answer = list(method_resp)[0] |
| record_dict["id"] = answer.text.strip() |
| objects.append(record_dict) |
| record_dict = {} |
| |
| return (objects, errors) |
| |
| def parse_body(self): |
| # A problem arise in the api response because there are undeclared |
| # xml namespaces. In order to fix that at the moment, we use the |
| # _fix_response method to clean up since we won't always have lxml |
| # library. |
| self._fix_response() |
| body = super().parse_body() |
| return body |
| |
| def success(self): |
| """ |
| Used to determine if the request was successful. |
| """ |
| return len(self.errors) == 0 |
| |
| def _make_excp(self, error): |
| return DurableDNSException(error["ERRORCODE"], error["ERRORMESSAGE"]) |
| |
| def _fix_response(self): |
| items = re.findall('<ns1:.+ xmlns:ns1="">', self.body, flags=0) |
| for item in items: |
| parts = item.split(" ") |
| prefix = parts[0].replace("<", "").split(":")[1] |
| new_item = "<" + prefix + ">" |
| close_tag = "</" + parts[0].replace("<", "") + ">" |
| new_close_tag = "</" + prefix + ">" |
| self.body = self.body.replace(item, new_item) |
| self.body = self.body.replace(close_tag, new_close_tag) |
| |
| |
| class DurableConnection(ConnectionUserAndKey): |
| host = API_HOST |
| responseCls = DurableResponse |
| |
| def add_default_params(self, params): |
| params["user_id"] = self.user_id |
| params["key"] = self.key |
| return params |
| |
| def add_default_headers(self, headers): |
| headers["Content-Type"] = "text/xml" |
| headers["Content-Encoding"] = "gzip; charset=ISO-8859-1" |
| return headers |
