From 6cc8f31f847bd7cc5df4491c454cf688f26b9e38 Mon Sep 17 00:00:00 2001 From: Adrien Ferrand Date: Mon, 14 Feb 2022 23:37:31 +0100 Subject: [PATCH] Version 3.9.4 --- CHANGELOG.md | 2 + lexicon/providers/dnspod.py | 4 +- lexicon/providers/dreamhost.py | 9 ++- lexicon/providers/webgo.py | 71 ++++++++++++++------ lexicon/tests/providers/test_azure.py | 4 +- lexicon/tests/providers/test_euserv.py | 4 +- lexicon/tests/providers/test_mythicbeasts.py | 2 +- lexicon/tests/providers/test_oci.py | 4 +- lexicon/tests/providers/test_transip.py | 4 +- lexicon/tests/providers/test_webgo.py | 14 +++- pyproject.toml | 2 +- 11 files changed, 84 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 125e9fc51..0a68e0a05 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,8 @@ # Changelog ## master - CURRENT + +## 3.9.4 - 14/02/2022 ### Added * Add `webgo` provider (#1102) diff --git a/lexicon/providers/dnspod.py b/lexicon/providers/dnspod.py index 7bca0a77b..3690988cd 100644 --- a/lexicon/providers/dnspod.py +++ b/lexicon/providers/dnspod.py @@ -40,7 +40,7 @@ def _create_record(self, rtype, name, content): "domain_id": self.domain_id, "sub_domain": self._relative_name(name), "record_type": rtype, - "record_line": u"\u9ED8\u8BA4", + "record_line": "\u9ED8\u8BA4", "value": content, } if self._get_lexicon_option("ttl"): @@ -93,7 +93,7 @@ def _update_record(self, identifier, rtype=None, name=None, content=None): "record_id": identifier, "sub_domain": self._relative_name(name), "record_type": rtype, - "record_line": u"\u9ED8\u8BA4", + "record_line": "\u9ED8\u8BA4", "value": content, } if self._get_lexicon_option("ttl"): diff --git a/lexicon/providers/dreamhost.py b/lexicon/providers/dreamhost.py index f096693fb..0ccb6c4c0 100644 --- a/lexicon/providers/dreamhost.py +++ b/lexicon/providers/dreamhost.py @@ -107,7 +107,14 @@ def _authenticate(self): for record in data: if record.get("record", "") == self.domain and record.get("type", "") in [ - "A", "AAAA", "CNAME", "MX", "NS", "SOA", "TXT", "SRV" + "A", + "AAAA", + "CNAME", + "MX", + "NS", + "SOA", + "TXT", + "SRV", ]: self.domain_id = self.domain break diff --git a/lexicon/providers/webgo.py b/lexicon/providers/webgo.py index 0e5960c3d..e262fa664 100644 --- a/lexicon/providers/webgo.py +++ b/lexicon/providers/webgo.py @@ -1,7 +1,6 @@ """Module provider for Webgo""" import logging - from bs4 import BeautifulSoup # type: ignore from requests import Session @@ -44,8 +43,10 @@ def _authenticate(self): login_response = self.session.post( "https://login.webgo.de/login", data={ - "data[User][username]": self._get_provider_option("auth_username") or "", - "data[User][password]": self._get_provider_option("auth_password") or "", + "data[User][username]": self._get_provider_option("auth_username") + or "", + "data[User][password]": self._get_provider_option("auth_password") + or "", }, ) @@ -61,13 +62,15 @@ def _authenticate(self): html = BeautifulSoup(zones_response.content, "html.parser") domain_table = html.find("table", {"class": "alltable"}) - rows = domain_table.find_all('tr') + rows = domain_table.find_all("tr") dns_link = None for row in rows[1:]: - domain = row.findAll('td')[1].renderContents().decode() + domain = row.findAll("td")[1].renderContents().decode() if domain == self.domain: - dns_link = row.findAll('td')[5] - dns_link = dns_link.find("a", {"class": "domainButton fcon-sliders"}).get('href') + dns_link = row.findAll("td")[5] + dns_link = dns_link.find( + "a", {"class": "domainButton fcon-sliders"} + ).get("href") # If the Domain couldn't be found, error, otherwise, return the value of the tag if dns_link is None: @@ -84,7 +87,10 @@ def _create_record(self, rtype, name, content): # Pull a list of records and check for ours if name: if name == self.domain: - LOGGER.warning("Unable to create record because your main domain %s can't be re-created", self.domain) + LOGGER.warning( + "Unable to create record because your main domain %s can't be re-created", + self.domain, + ) return False name = self._relative_name(name) if rtype == "CNAME" and not content.endswith("."): @@ -100,7 +106,8 @@ def _create_record(self, rtype, name, content): "data[DnsSetting][pref-mx]": "0", "data[DnsSetting][value]": content, "data[DnsSetting][action]": "newsub", - "data[DnsSetting][domain_id]": self.domain_id, } + "data[DnsSetting][domain_id]": self.domain_id, + } ttl = self._get_lexicon_option("ttl") if ttl: if ttl <= 0: @@ -114,8 +121,12 @@ def _create_record(self, rtype, name, content): else: data["data[DnsSetting][pref-mx]"] = str(prio) - self.session.post("https://login.webgo.de/dns_settings/domainDnsEditForm", data=data) - self.session.get(f"https://login.webgo.de/dnsSettings/domainDnsDo/{self.domain_id}/ok") + self.session.post( + "https://login.webgo.de/dns_settings/domainDnsEditForm", data=data + ) + self.session.get( + f"https://login.webgo.de/dnsSettings/domainDnsDo/{self.domain_id}/ok" + ) # Pull a list of records and check for ours records = self._list_records(name=name) if len(records) >= 1: @@ -155,7 +166,7 @@ def _list_records_internal( rec = {} mainip = html.find("span", {"class": "mainIp"}) mainip_record = mainip.find_next("span").text - dns_link = mainip.find_next("a").get('href') + dns_link = mainip.find_next("a").get("href") rec["name"] = self.domain rec["ttl"] = "3600" rec["type"] = "A" @@ -175,7 +186,9 @@ def _list_records_internal( rec["prio"] = tds[3].string rec["content"] = tds[4].string dns_link = tds[5] - dns_link = dns_link.find("a", {"class": "domainButton fcon-edit"}).get('href') + dns_link = dns_link.find("a", {"class": "domainButton fcon-edit"}).get( + "href" + ) rec["id"] = dns_link.rsplit("/", 2)[1] if rec["content"].startswith('"'): rec = self._clean_TXT_record(rec) @@ -184,7 +197,9 @@ def _list_records_internal( records = new_records if identifier: LOGGER.debug("Filtering %d records by id: %s", len(records), identifier) - records = [record for record in records if str(record["id"]) == str(identifier)] + records = [ + record for record in records if str(record["id"]) == str(identifier) + ] if rtype: LOGGER.debug("Filtering %d records by rtype: %s", len(records), rtype) records = [record for record in records if record["type"] == rtype] @@ -220,7 +235,8 @@ def _update_record(self, identifier=None, rtype=None, name=None, content=None): maindata = { "data[DnsSetting][value]": content, "data[DnsSetting][action]": "main", - "data[DnsSetting][domain_id]": record["id"], } + "data[DnsSetting][domain_id]": record["id"], + } # Update every Subrecord else: # Delete record if it exists @@ -242,9 +258,15 @@ def _update_record(self, identifier=None, rtype=None, name=None, content=None): maindata = { "data[DnsSetting][value]": content, "data[DnsSetting][action]": "main", - "data[DnsSetting][domain_id]": record["id"], } - self.session.post("https://login.webgo.de/dns_settings/domainDnsEditForm", data=maindata) - self.session.get(f"https://login.webgo.de/dnsSettings/domainDnsDo/{self.domain_id}/ok") + "data[DnsSetting][domain_id]": record["id"], + } + self.session.post( + "https://login.webgo.de/dns_settings/domainDnsEditForm", + data=maindata, + ) + self.session.get( + f"https://login.webgo.de/dnsSettings/domainDnsDo/{self.domain_id}/ok" + ) LOGGER.debug("Updated Main Domain %s", records[0]["name"]) return True @@ -254,14 +276,21 @@ def _delete_record(self, identifier=None, rtype=None, name=None, content=None): delete_record_ids = [] records = self._list_records_internal(rtype, name, content, identifier) if "main" in [record["option"] for record in records]: - LOGGER.warning("Unable to delete records because your main domain %s can't be deleted", self.domain) + LOGGER.warning( + "Unable to delete records because your main domain %s can't be deleted", + self.domain, + ) return False delete_record_ids = [record["id"] for record in records] LOGGER.debug("Record IDs to delete: %s", delete_record_ids) for rec_id in delete_record_ids: - response = self.session.get(f"https://login.webgo.de/dnsSettings/domainDnsDo/{rec_id}/delete") + response = self.session.get( + f"https://login.webgo.de/dnsSettings/domainDnsDo/{rec_id}/delete" + ) if response.status_code == 200: - self.session.get(f"https://login.webgo.de/dnsSettings/domainDnsDo/{self.domain_id}/ok") + self.session.get( + f"https://login.webgo.de/dnsSettings/domainDnsDo/{self.domain_id}/ok" + ) else: LOGGER.warning("Unable to delete record %s", rec_id) return False diff --git a/lexicon/tests/providers/test_azure.py b/lexicon/tests/providers/test_azure.py index 0cd0ed6a1..0011fcbfa 100644 --- a/lexicon/tests/providers/test_azure.py +++ b/lexicon/tests/providers/test_azure.py @@ -38,12 +38,12 @@ def _filter_request(self, request): def _filter_response(self, response): # Hide access_token value in oauth token responses response["body"]["string"] = re.sub( - br'"access_token":"[\w.-]+"', + rb'"access_token":"[\w.-]+"', b'"access_token":"TOKEN"', response["body"]["string"], ) response["body"]["string"] = re.sub( - br"\\/subscriptions\\/[\w-]+\\/", + rb"\\/subscriptions\\/[\w-]+\\/", b"\\/subscriptions\\/SUBSCRIPTION_ID\\/", response["body"]["string"], ) diff --git a/lexicon/tests/providers/test_euserv.py b/lexicon/tests/providers/test_euserv.py index 9c827295e..42de8222c 100644 --- a/lexicon/tests/providers/test_euserv.py +++ b/lexicon/tests/providers/test_euserv.py @@ -43,13 +43,13 @@ def _filter_response(self, response): if "string" in response["body"]: # Replace session and order id with placeholders response["body"]["string"] = re.sub( - br'"sess_id":{"value":"[\w.-]+"', + rb'"sess_id":{"value":"[\w.-]+"', b'"sess_id":{"value":"SESSION_ID"', response["body"]["string"], ) response["body"]["string"] = re.sub( - br'"ord_no":{"value":"[\w.-]+"', + rb'"ord_no":{"value":"[\w.-]+"', b'"ord_no":{"value":"ORDER_ID"', response["body"]["string"], ) diff --git a/lexicon/tests/providers/test_mythicbeasts.py b/lexicon/tests/providers/test_mythicbeasts.py index d1dc50018..b486f0abd 100644 --- a/lexicon/tests/providers/test_mythicbeasts.py +++ b/lexicon/tests/providers/test_mythicbeasts.py @@ -26,7 +26,7 @@ def _filter_response(self, response): if "string" in response["body"]: response["body"]["string"] = re.sub( - br"\"access_token\":\"[\w-]+\"", + rb"\"access_token\":\"[\w-]+\"", b'"access_token": "DUMMY_TOKEN"', response["body"]["string"], ) diff --git a/lexicon/tests/providers/test_oci.py b/lexicon/tests/providers/test_oci.py index 24d8e2c5c..eafe33f42 100644 --- a/lexicon/tests/providers/test_oci.py +++ b/lexicon/tests/providers/test_oci.py @@ -59,12 +59,12 @@ def _filter_headers(self): def _filter_response(self, response): response["body"]["string"] = re.sub( - br'"compartmentId":"[\w.-]+"', + rb'"compartmentId":"[\w.-]+"', b'"compartmentId":"OCI-COMPARTMENT-ID"', response["body"]["string"], ) response["body"]["string"] = re.sub( - br'"id":"[\w.-]+"', + rb'"id":"[\w.-]+"', b'"id":"DNS-ZONE-ID"', response["body"]["string"], ) diff --git a/lexicon/tests/providers/test_transip.py b/lexicon/tests/providers/test_transip.py index 4743f9c5f..6b3b54044 100644 --- a/lexicon/tests/providers/test_transip.py +++ b/lexicon/tests/providers/test_transip.py @@ -75,12 +75,12 @@ def _filter_post_data_parameters(self): def _filter_response(self, response): response["body"]["string"] = re.sub( - br'"token":"[\w.-]+"', + rb'"token":"[\w.-]+"', b'"token":"TOKEN"', response["body"]["string"], ) response["body"]["string"] = re.sub( - br'"authCode":"[\w.-]+"', + rb'"authCode":"[\w.-]+"', b'"authCode":"AUTH_CODE"', response["body"]["string"], ) diff --git a/lexicon/tests/providers/test_webgo.py b/lexicon/tests/providers/test_webgo.py index 8630160b2..53d9811a3 100644 --- a/lexicon/tests/providers/test_webgo.py +++ b/lexicon/tests/providers/test_webgo.py @@ -28,8 +28,18 @@ def _filter_response(self, response): # Filter out all Customer/Service IDs from Response body = re.sub(r"\b(16)([0-9]{3})\b", "XXXXX", body) # Filter out Clearname from Response - body = re.sub(r"
.*?<\/div>", "
John Doe
", body) + body = re.sub( + r"
.*?<\/div>", + '
John Doe
', + body, + ) # Filter out all Domains not tested - body = re.sub(r'<\s*td[^>]*>(?!(' + re.escape(self.domain) + r'))(.[A-Za-z0-9]*\.[a-z]{2,3})<\s*/\s*td>', 'filtereddomain.de', body) + body = re.sub( + r"<\s*td[^>]*>(?!(" + + re.escape(self.domain) + + r"))(.[A-Za-z0-9]*\.[a-z]{2,3})<\s*/\s*td>", + "filtereddomain.de", + body, + ) response["body"]["string"] = body.encode("utf-8") return response diff --git a/pyproject.toml b/pyproject.toml index 28062613f..b57d13464 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api" [tool.poetry] name = "dns-lexicon" -version = "3.9.3" +version = "3.9.4" description = "Manipulate DNS records on various DNS providers in a standardized/agnostic way" license = "MIT" keywords = [