We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Plesk provider assumes `site_name` equals `domain` (which is not always true)
The plesk provider assumes that the `site_name` is always the same as the `domain`: https://github.com/AnalogJ/lexicon/blob/master/lexicon/providers/plesk.py#L52

That creates problems if one wants to create a record for a subdomain (e.g. `test.example.com`) for which there is no matching plesk site. It could be the case that the record for `test.example.com` (and even all records `*.test.example.com`) are part of the site `example.com`.

Example (source for the certbot plugin [here](https://gitlab.com/spike77453/certbot-dns-plesk/)):
```
# certbot certonly --authenticator dns-plesk --dns-plesk-credentials plesk.ini -d test.example.com
Saving debug log to /var/log/letsencrypt/letsencrypt.log
Plugins selected: Authenticator dns-plesk, Installer None
Simulating a certificate request for test.example.com
Performing the following challenges:
dns-01 challenge for test.example.com
Cleaning up challenges
Unexpected error determining zone identifier for test.example.com: API returned error: 1013 (Site does not exist)
```

One could fix this by doing something similar to the `Client` class and strip the subdomain part https://github.com/AnalogJ/lexicon/blob/master/lexicon/client.py#L36. The `Provider` constructor could do something like:

```
        try:
            domain_extractor = tldextract.TLDExtract(
                cache_dir=_get_tldextract_cache_path(), include_psl_private_domains=True
            )
        except TypeError:
            domain_extractor = tldextract.TLDExtract(
                cache_file=_get_tldextract_cache_path(), include_psl_private_domains=True  # type: ignore
            )
        domain_parts = domain_extractor(self.domain)

        #self.site_name = self.domain
        self.site_name = f"{domain_parts.domain}.{domain_parts.suffix}"
        assert self.site_name is not None
```

And then instead of stripping the domain from `host`, one would strip the `site_name` here: https://github.com/AnalogJ/lexicon/blob/master/lexicon/providers/plesk.py#L179  
That however doesn't work any more if there actually is a site for a particular subdomain.

How about adding an optional `site-id` provider option so the user can specify the site manually in the configuration file?


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `lexicon/providers/plesk.py`
Content:
```
1 """
2 Lexicon Plesk Provider
3 
4 Author: Jens Reimann, 2018
5 
6 API Docs: https://docs.plesk.com/en-US/onyx/api-rpc
7 """
8 import logging
9 from collections import defaultdict
10 from typing import Dict, List, Optional
11 from xml.etree import cElementTree
12 from xml.etree.ElementTree import Element
13 
14 import requests
15 
16 from lexicon.providers.base import Provider as BaseProvider
17 
18 LOGGER = logging.getLogger(__name__)
19 
20 PLEX_URL_SUFFIX = "/enterprise/control/agent.php"
21 
22 NAMESERVER_DOMAINS: List[str] = []
23 
24 
25 def provider_parser(subparser):
26     """Configure provider parser for Plesk"""
27     subparser.add_argument(
28         "--auth-username", help="specify username for authentication"
29     )
30     subparser.add_argument(
31         "--auth-password", help="specify password for authentication"
32     )
33     subparser.add_argument(
34         "--plesk-server", help="specify URL to the Plesk Web UI, including the port"
35     )
36 
37 
38 class Provider(BaseProvider):
39     """Provider class for Plesk"""
40 
41     def __init__(self, config):
42         super(Provider, self).__init__(config)
43 
44         self.api_endpoint = self._get_provider_option("plesk_server")
45 
46         if self.api_endpoint.endswith("/"):
47             self.api_endpoint = self.api_endpoint[:-1]
48 
49         if not self.api_endpoint.endswith(PLEX_URL_SUFFIX):
50             self.api_endpoint += PLEX_URL_SUFFIX
51 
52         self.site_name = self.domain
53         assert self.site_name is not None
54 
55         self.domain_id = None
56 
57         self.username = self._get_provider_option("auth_username")
58         assert self.username is not None
59 
60         self.password = self._get_provider_option("auth_password")
61         assert self.password is not None
62 
63     def __simple_request(self, rtype, operation, req):
64 
65         response = self.__plesk_request({rtype: {operation: req}})[rtype][operation]
66 
67         result = response["result"]
68 
69         if isinstance(result, list):
70             for record in result:
71                 if record["status"] == "error":
72                     raise Exception(
73                         f"API returned at least one error: {record['errtext']}"
74                     )
75         elif response["result"]["status"] == "error":
76             errcode = response["result"]["errcode"]
77             errtext = response["result"]["errtext"]
78             raise Exception(f"API returned error: {errcode} ({errtext})")
79 
80         return response
81 
82     def __plesk_request(self, request):
83 
84         headers = {
85             "Content-type": "text/xml",
86             "HTTP_PRETTY_PRINT": "TRUE",
87             "HTTP_AUTH_LOGIN": self.username,
88             "HTTP_AUTH_PASSWD": self.password,
89         }
90 
91         xml = f"""\
92 <?xml version="1.0" encoding="utf-8"?>
93 {cElementTree.tostring(_dict_to_etree({"packet": request}), encoding="unicode")}\
94 """
95 
96         LOGGER.debug("Request: %s", xml)
97 
98         response = requests.post(
99             self.api_endpoint,
100             headers=headers,
101             data=xml,
102             auth=(self.username, self.password),
103         )
104 
105         data = response.text
106 
107         LOGGER.debug("Response: %s", data)
108         result = _etree_to_dict(cElementTree.XML(data))
109         return result["packet"]
110 
111     def __find_site(self):
112         return self.__simple_request(
113             "site", "get", {"filter": {"name": self.site_name, "dataset": {}}}
114         )["result"]["id"]
115 
116     def _authenticate(self):
117         self.domain_id = self.__find_site()
118 
119         if self.domain_id is None:
120             raise Exception("Domain not found")
121 
122     def _create_record(self, rtype, name, content):
123         return self.__create_entry(rtype, name, content, None)
124 
125     def _list_records(self, rtype=None, name=None, content=None):
126         entries = self.__find_dns_entries(rtype, name, content)
127         LOGGER.debug("list_records: %s", entries)
128         return entries
129 
130     def _update_record(self, identifier, rtype=None, name=None, content=None):
131         if identifier is None:
132             entries = self.__find_dns_entries(rtype, name, None)
133             LOGGER.debug("Entries found: %s", entries)
134 
135             if not entries:
136                 raise Exception("No entry found for updating")
137 
138             identifier = entries[0]["id"]
139             entry = self.__get_dns_entry(identifier)
140 
141             ids = []
142             for an_entry in entries:
143                 ids.append(an_entry["id"])
144 
145             self.__delete_dns_records_by_id(ids)
146 
147         else:
148 
149             entry = self.__get_dns_entry(identifier)
150             self.__delete_dns_records_by_id([identifier])
151 
152         assert entry is not None
153 
154         LOGGER.debug("Updating: %s", entry)
155 
156         if rtype:
157             entry["type"] = rtype
158         if name:
159             entry["host"] = name
160         if content:
161             entry["value"] = content
162 
163         return self.__create_entry(
164             entry["type"], entry["host"], entry["value"], entry["opt"]
165         )
166 
167     def __create_entry(self, rtype, host, value, opt):
168         entries = self.__find_dns_entries(rtype, self._fqdn_name(host), value)
169 
170         if entries:
171             return True  # already exists
172 
173         self.__simple_request(
174             "dns",
175             "add_rec",
176             {
177                 "site-id": self.domain_id,
178                 "type": rtype,
179                 "host": self._relative_name(host),
180                 "value": value,
181                 "opt": opt,
182             },
183         )
184 
185         return True
186 
187     def _delete_record(self, identifier=None, rtype=None, name=None, content=None):
188         if identifier:
189             self.__delete_dns_records_by_id([identifier])
190             return True
191         entries = self.__find_dns_entries(rtype, self._fqdn_name(name), content)
192         ids = []
193 
194         for entry in entries:
195             ids.append(entry["id"])
196 
197         self.__delete_dns_records_by_id(ids)
198         return bool(ids)
199 
200     def __get_dns_entry(self, identifier):
201         return self.__simple_request("dns", "get_rec", {"filter": {"id": identifier}})[
202             "result"
203         ]["data"]
204 
205     def __find_dns_entries(self, rtype=None, host=None, value=None):
206         LOGGER.debug("Searching for: %s, %s, %s", rtype, host, value)
207 
208         if value and rtype and rtype in ["CNAME"]:
209             LOGGER.debug("CNAME transformation")
210             value = value.rstrip(".") + "."
211 
212         if host:
213             host = self._fqdn_name(host)
214 
215         result = self.__simple_request(
216             "dns", "get_rec", {"filter": {"site-id": self.domain_id}}
217         )
218 
219         entries = []
220 
221         for record in result["result"]:
222 
223             LOGGER.debug("Record: %s", record)
224 
225             if (rtype is not None) and (record["data"]["type"] != rtype):
226                 LOGGER.debug(
227                     "\tType doesn't match - expected: '%s', found: '%s'",
228                     rtype,
229                     record["data"]["type"],
230                 )
231                 continue
232 
233             if (host is not None) and (record["data"]["host"] != host):
234                 LOGGER.debug(
235                     "\tHost doesn't match - expected: '%s', found: '%s'",
236                     host,
237                     record["data"]["host"],
238                 )
239                 continue
240 
241             if (value is not None) and (record["data"]["value"] != value):
242                 LOGGER.debug(
243                     "\tValue doesn't match - expected: '%s', found: '%s'",
244                     value,
245                     record["data"]["value"],
246                 )
247                 continue
248 
249             entry = {
250                 "id": record["id"],
251                 "type": record["data"]["type"],
252                 "name": self._full_name(record["data"]["host"]),
253                 "ttl": None,
254                 "options": {},
255             }
256 
257             if record["data"]["type"] in ["CNAME"]:
258                 entry["content"] = record["data"]["value"].rstrip(".")
259             else:
260                 entry["content"] = record["data"]["value"]
261 
262             if record["data"]["type"] == "MX":
263                 entry["options"]["mx"] = {"priority": int(record["data"]["opt"])}
264 
265             entries.append(entry)
266 
267         return entries
268 
269     def __delete_dns_records_by_id(self, ids):
270         if not ids:
271             return
272 
273         req = []
274         for i in ids:
275             req.append({"del_rec": {"filter": {"id": i}}})
276 
277         self.__plesk_request({"dns": req})
278 
279     def _request(self, action="GET", url="/", data=None, query_params=None):
280         # Helper _request is not used for Plesk provider
281         pass
282 
283 
284 def _etree_to_dict(t: Element) -> Optional[Dict]:
285     d: Optional[Dict] = {t.tag: {} if t.attrib else None}
286     children = list(t)
287     if children:
288         dd = defaultdict(list)
289         for dc in map(_etree_to_dict, children):
290             if dc:
291                 for k, v in dc.items():
292                     dd[k].append(v)
293         d = {t.tag: {k: v[0] if len(v) == 1 else v for k, v in dd.items()}}
294     if t.attrib and d:
295         d[t.tag].update(("@" + k, v) for k, v in t.attrib.items())
296     if t.text and d:
297         text = t.text.strip()
298         if children or t.attrib:
299             if text:
300                 d[t.tag]["#text"] = text
301         else:
302             d[t.tag] = text
303     return d
304 
305 
306 def _dict_to_etree(d: Dict) -> Element:
307     def _to_etree(d1, root):
308         if not d1:
309             pass
310         elif isinstance(d1, str):
311             root.text = d1
312         elif isinstance(d1, dict):
313             for k, v in d1.items():
314                 assert isinstance(k, str)
315                 if k.startswith("#"):
316                     assert k == "#text" and isinstance(v, str)
317                     root.text = v
318                 elif k.startswith("@"):
319                     assert isinstance(v, str)
320                     root.set(k[1:], v)
321                 elif isinstance(v, list):
322                     for e in v:
323                         _to_etree(e, cElementTree.SubElement(root, k))
324                 else:
325                     _to_etree(v, cElementTree.SubElement(root, k))
326         else:
327             raise TypeError("invalid type: " + str(type(d1)))
328 
329     assert isinstance(d, dict) and len(d) == 1
330     tag, body = next(iter(d.items()))
331     node = cElementTree.Element(tag)
332     _to_etree(body, node)
333     return node
334
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/lexicon/providers/plesk.py b/lexicon/providers/plesk.py
--- a/lexicon/providers/plesk.py
+++ b/lexicon/providers/plesk.py
@@ -49,9 +49,6 @@
         if not self.api_endpoint.endswith(PLEX_URL_SUFFIX):
             self.api_endpoint += PLEX_URL_SUFFIX
 
-        self.site_name = self.domain
-        assert self.site_name is not None
-
         self.domain_id = None
 
         self.username = self._get_provider_option("auth_username")
@@ -110,7 +107,7 @@
 
     def __find_site(self):
         return self.__simple_request(
-            "site", "get", {"filter": {"name": self.site_name, "dataset": {}}}
+            "site", "get", {"filter": {"name": self.domain, "dataset": {}}}
         )["result"]["id"]
 
     def _authenticate(self):

{"golden_diff": "diff --git a/lexicon/providers/plesk.py b/lexicon/providers/plesk.py\n--- a/lexicon/providers/plesk.py\n+++ b/lexicon/providers/plesk.py\n@@ -49,9 +49,6 @@\n         if not self.api_endpoint.endswith(PLEX_URL_SUFFIX):\n             self.api_endpoint += PLEX_URL_SUFFIX\n \n-        self.site_name = self.domain\n-        assert self.site_name is not None\n-\n         self.domain_id = None\n \n         self.username = self._get_provider_option(\"auth_username\")\n@@ -110,7 +107,7 @@\n \n     def __find_site(self):\n         return self.__simple_request(\n-            \"site\", \"get\", {\"filter\": {\"name\": self.site_name, \"dataset\": {}}}\n+            \"site\", \"get\", {\"filter\": {\"name\": self.domain, \"dataset\": {}}}\n         )[\"result\"][\"id\"]\n \n     def _authenticate(self):\n", "issue": "Plesk provider assumes `site_name` equals `domain` (which is not always true)\nThe plesk provider assumes that the `site_name` is always the same as the `domain`: https://github.com/AnalogJ/lexicon/blob/master/lexicon/providers/plesk.py#L52\r\n\r\nThat creates problems if one wants to create a record for a subdomain (e.g. `test.example.com`) for which there is no matching plesk site. It could be the case that the record for `test.example.com` (and even all records `*.test.example.com`) are part of the site `example.com`.\r\n\r\nExample (source for the certbot plugin [here](https://gitlab.com/spike77453/certbot-dns-plesk/)):\r\n```\r\n# certbot certonly --authenticator dns-plesk --dns-plesk-credentials plesk.ini -d test.example.com\r\nSaving debug log to /var/log/letsencrypt/letsencrypt.log\r\nPlugins selected: Authenticator dns-plesk, Installer None\r\nSimulating a certificate request for test.example.com\r\nPerforming the following challenges:\r\ndns-01 challenge for test.example.com\r\nCleaning up challenges\r\nUnexpected error determining zone identifier for test.example.com: API returned error: 1013 (Site does not exist)\r\n```\r\n\r\nOne could fix this by doing something similar to the `Client` class and strip the subdomain part https://github.com/AnalogJ/lexicon/blob/master/lexicon/client.py#L36. The `Provider` constructor could do something like:\r\n\r\n```\r\n        try:\r\n            domain_extractor = tldextract.TLDExtract(\r\n                cache_dir=_get_tldextract_cache_path(), include_psl_private_domains=True\r\n            )\r\n        except TypeError:\r\n            domain_extractor = tldextract.TLDExtract(\r\n                cache_file=_get_tldextract_cache_path(), include_psl_private_domains=True  # type: ignore\r\n            )\r\n        domain_parts = domain_extractor(self.domain)\r\n\r\n        #self.site_name = self.domain\r\n        self.site_name = f\"{domain_parts.domain}.{domain_parts.suffix}\"\r\n        assert self.site_name is not None\r\n```\r\n\r\nAnd then instead of stripping the domain from `host`, one would strip the `site_name` here: https://github.com/AnalogJ/lexicon/blob/master/lexicon/providers/plesk.py#L179  \r\nThat however doesn't work any more if there actually is a site for a particular subdomain.\r\n\r\nHow about adding an optional `site-id` provider option so the user can specify the site manually in the configuration file?\r\n\n", "before_files": [{"content": "\"\"\"\nLexicon Plesk Provider\n\nAuthor: Jens Reimann, 2018\n\nAPI Docs: https://docs.plesk.com/en-US/onyx/api-rpc\n\"\"\"\nimport logging\nfrom collections import defaultdict\nfrom typing import Dict, List, Optional\nfrom xml.etree import cElementTree\nfrom xml.etree.ElementTree import Element\n\nimport requests\n\nfrom lexicon.providers.base import Provider as BaseProvider\n\nLOGGER = logging.getLogger(__name__)\n\nPLEX_URL_SUFFIX = \"/enterprise/control/agent.php\"\n\nNAMESERVER_DOMAINS: List[str] = []\n\n\ndef provider_parser(subparser):\n    \"\"\"Configure provider parser for Plesk\"\"\"\n    subparser.add_argument(\n        \"--auth-username\", help=\"specify username for authentication\"\n    )\n    subparser.add_argument(\n        \"--auth-password\", help=\"specify password for authentication\"\n    )\n    subparser.add_argument(\n        \"--plesk-server\", help=\"specify URL to the Plesk Web UI, including the port\"\n    )\n\n\nclass Provider(BaseProvider):\n    \"\"\"Provider class for Plesk\"\"\"\n\n    def __init__(self, config):\n        super(Provider, self).__init__(config)\n\n        self.api_endpoint = self._get_provider_option(\"plesk_server\")\n\n        if self.api_endpoint.endswith(\"/\"):\n            self.api_endpoint = self.api_endpoint[:-1]\n\n        if not self.api_endpoint.endswith(PLEX_URL_SUFFIX):\n            self.api_endpoint += PLEX_URL_SUFFIX\n\n        self.site_name = self.domain\n        assert self.site_name is not None\n\n        self.domain_id = None\n\n        self.username = self._get_provider_option(\"auth_username\")\n        assert self.username is not None\n\n        self.password = self._get_provider_option(\"auth_password\")\n        assert self.password is not None\n\n    def __simple_request(self, rtype, operation, req):\n\n        response = self.__plesk_request({rtype: {operation: req}})[rtype][operation]\n\n        result = response[\"result\"]\n\n        if isinstance(result, list):\n            for record in result:\n                if record[\"status\"] == \"error\":\n                    raise Exception(\n                        f\"API returned at least one error: {record['errtext']}\"\n                    )\n        elif response[\"result\"][\"status\"] == \"error\":\n            errcode = response[\"result\"][\"errcode\"]\n            errtext = response[\"result\"][\"errtext\"]\n            raise Exception(f\"API returned error: {errcode} ({errtext})\")\n\n        return response\n\n    def __plesk_request(self, request):\n\n        headers = {\n            \"Content-type\": \"text/xml\",\n            \"HTTP_PRETTY_PRINT\": \"TRUE\",\n            \"HTTP_AUTH_LOGIN\": self.username,\n            \"HTTP_AUTH_PASSWD\": self.password,\n        }\n\n        xml = f\"\"\"\\\n<?xml version=\"1.0\" encoding=\"utf-8\"?>\n{cElementTree.tostring(_dict_to_etree({\"packet\": request}), encoding=\"unicode\")}\\\n\"\"\"\n\n        LOGGER.debug(\"Request: %s\", xml)\n\n        response = requests.post(\n            self.api_endpoint,\n            headers=headers,\n            data=xml,\n            auth=(self.username, self.password),\n        )\n\n        data = response.text\n\n        LOGGER.debug(\"Response: %s\", data)\n        result = _etree_to_dict(cElementTree.XML(data))\n        return result[\"packet\"]\n\n    def __find_site(self):\n        return self.__simple_request(\n            \"site\", \"get\", {\"filter\": {\"name\": self.site_name, \"dataset\": {}}}\n        )[\"result\"][\"id\"]\n\n    def _authenticate(self):\n        self.domain_id = self.__find_site()\n\n        if self.domain_id is None:\n            raise Exception(\"Domain not found\")\n\n    def _create_record(self, rtype, name, content):\n        return self.__create_entry(rtype, name, content, None)\n\n    def _list_records(self, rtype=None, name=None, content=None):\n        entries = self.__find_dns_entries(rtype, name, content)\n        LOGGER.debug(\"list_records: %s\", entries)\n        return entries\n\n    def _update_record(self, identifier, rtype=None, name=None, content=None):\n        if identifier is None:\n            entries = self.__find_dns_entries(rtype, name, None)\n            LOGGER.debug(\"Entries found: %s\", entries)\n\n            if not entries:\n                raise Exception(\"No entry found for updating\")\n\n            identifier = entries[0][\"id\"]\n            entry = self.__get_dns_entry(identifier)\n\n            ids = []\n            for an_entry in entries:\n                ids.append(an_entry[\"id\"])\n\n            self.__delete_dns_records_by_id(ids)\n\n        else:\n\n            entry = self.__get_dns_entry(identifier)\n            self.__delete_dns_records_by_id([identifier])\n\n        assert entry is not None\n\n        LOGGER.debug(\"Updating: %s\", entry)\n\n        if rtype:\n            entry[\"type\"] = rtype\n        if name:\n            entry[\"host\"] = name\n        if content:\n            entry[\"value\"] = content\n\n        return self.__create_entry(\n            entry[\"type\"], entry[\"host\"], entry[\"value\"], entry[\"opt\"]\n        )\n\n    def __create_entry(self, rtype, host, value, opt):\n        entries = self.__find_dns_entries(rtype, self._fqdn_name(host), value)\n\n        if entries:\n            return True  # already exists\n\n        self.__simple_request(\n            \"dns\",\n            \"add_rec\",\n            {\n                \"site-id\": self.domain_id,\n                \"type\": rtype,\n                \"host\": self._relative_name(host),\n                \"value\": value,\n                \"opt\": opt,\n            },\n        )\n\n        return True\n\n    def _delete_record(self, identifier=None, rtype=None, name=None, content=None):\n        if identifier:\n            self.__delete_dns_records_by_id([identifier])\n            return True\n        entries = self.__find_dns_entries(rtype, self._fqdn_name(name), content)\n        ids = []\n\n        for entry in entries:\n            ids.append(entry[\"id\"])\n\n        self.__delete_dns_records_by_id(ids)\n        return bool(ids)\n\n    def __get_dns_entry(self, identifier):\n        return self.__simple_request(\"dns\", \"get_rec\", {\"filter\": {\"id\": identifier}})[\n            \"result\"\n        ][\"data\"]\n\n    def __find_dns_entries(self, rtype=None, host=None, value=None):\n        LOGGER.debug(\"Searching for: %s, %s, %s\", rtype, host, value)\n\n        if value and rtype and rtype in [\"CNAME\"]:\n            LOGGER.debug(\"CNAME transformation\")\n            value = value.rstrip(\".\") + \".\"\n\n        if host:\n            host = self._fqdn_name(host)\n\n        result = self.__simple_request(\n            \"dns\", \"get_rec\", {\"filter\": {\"site-id\": self.domain_id}}\n        )\n\n        entries = []\n\n        for record in result[\"result\"]:\n\n            LOGGER.debug(\"Record: %s\", record)\n\n            if (rtype is not None) and (record[\"data\"][\"type\"] != rtype):\n                LOGGER.debug(\n                    \"\\tType doesn't match - expected: '%s', found: '%s'\",\n                    rtype,\n                    record[\"data\"][\"type\"],\n                )\n                continue\n\n            if (host is not None) and (record[\"data\"][\"host\"] != host):\n                LOGGER.debug(\n                    \"\\tHost doesn't match - expected: '%s', found: '%s'\",\n                    host,\n                    record[\"data\"][\"host\"],\n                )\n                continue\n\n            if (value is not None) and (record[\"data\"][\"value\"] != value):\n                LOGGER.debug(\n                    \"\\tValue doesn't match - expected: '%s', found: '%s'\",\n                    value,\n                    record[\"data\"][\"value\"],\n                )\n                continue\n\n            entry = {\n                \"id\": record[\"id\"],\n                \"type\": record[\"data\"][\"type\"],\n                \"name\": self._full_name(record[\"data\"][\"host\"]),\n                \"ttl\": None,\n                \"options\": {},\n            }\n\n            if record[\"data\"][\"type\"] in [\"CNAME\"]:\n                entry[\"content\"] = record[\"data\"][\"value\"].rstrip(\".\")\n            else:\n                entry[\"content\"] = record[\"data\"][\"value\"]\n\n            if record[\"data\"][\"type\"] == \"MX\":\n                entry[\"options\"][\"mx\"] = {\"priority\": int(record[\"data\"][\"opt\"])}\n\n            entries.append(entry)\n\n        return entries\n\n    def __delete_dns_records_by_id(self, ids):\n        if not ids:\n            return\n\n        req = []\n        for i in ids:\n            req.append({\"del_rec\": {\"filter\": {\"id\": i}}})\n\n        self.__plesk_request({\"dns\": req})\n\n    def _request(self, action=\"GET\", url=\"/\", data=None, query_params=None):\n        # Helper _request is not used for Plesk provider\n        pass\n\n\ndef _etree_to_dict(t: Element) -> Optional[Dict]:\n    d: Optional[Dict] = {t.tag: {} if t.attrib else None}\n    children = list(t)\n    if children:\n        dd = defaultdict(list)\n        for dc in map(_etree_to_dict, children):\n            if dc:\n                for k, v in dc.items():\n                    dd[k].append(v)\n        d = {t.tag: {k: v[0] if len(v) == 1 else v for k, v in dd.items()}}\n    if t.attrib and d:\n        d[t.tag].update((\"@\" + k, v) for k, v in t.attrib.items())\n    if t.text and d:\n        text = t.text.strip()\n        if children or t.attrib:\n            if text:\n                d[t.tag][\"#text\"] = text\n        else:\n            d[t.tag] = text\n    return d\n\n\ndef _dict_to_etree(d: Dict) -> Element:\n    def _to_etree(d1, root):\n        if not d1:\n            pass\n        elif isinstance(d1, str):\n            root.text = d1\n        elif isinstance(d1, dict):\n            for k, v in d1.items():\n                assert isinstance(k, str)\n                if k.startswith(\"#\"):\n                    assert k == \"#text\" and isinstance(v, str)\n                    root.text = v\n                elif k.startswith(\"@\"):\n                    assert isinstance(v, str)\n                    root.set(k[1:], v)\n                elif isinstance(v, list):\n                    for e in v:\n                        _to_etree(e, cElementTree.SubElement(root, k))\n                else:\n                    _to_etree(v, cElementTree.SubElement(root, k))\n        else:\n            raise TypeError(\"invalid type: \" + str(type(d1)))\n\n    assert isinstance(d, dict) and len(d) == 1\n    tag, body = next(iter(d.items()))\n    node = cElementTree.Element(tag)\n    _to_etree(body, node)\n    return node\n", "path": "lexicon/providers/plesk.py"}], "after_files": [{"content": "\"\"\"\nLexicon Plesk Provider\n\nAuthor: Jens Reimann, 2018\n\nAPI Docs: https://docs.plesk.com/en-US/onyx/api-rpc\n\"\"\"\nimport logging\nfrom collections import defaultdict\nfrom typing import Dict, List, Optional\nfrom xml.etree import cElementTree\nfrom xml.etree.ElementTree import Element\n\nimport requests\n\nfrom lexicon.providers.base import Provider as BaseProvider\n\nLOGGER = logging.getLogger(__name__)\n\nPLEX_URL_SUFFIX = \"/enterprise/control/agent.php\"\n\nNAMESERVER_DOMAINS: List[str] = []\n\n\ndef provider_parser(subparser):\n    \"\"\"Configure provider parser for Plesk\"\"\"\n    subparser.add_argument(\n        \"--auth-username\", help=\"specify username for authentication\"\n    )\n    subparser.add_argument(\n        \"--auth-password\", help=\"specify password for authentication\"\n    )\n    subparser.add_argument(\n        \"--plesk-server\", help=\"specify URL to the Plesk Web UI, including the port\"\n    )\n\n\nclass Provider(BaseProvider):\n    \"\"\"Provider class for Plesk\"\"\"\n\n    def __init__(self, config):\n        super(Provider, self).__init__(config)\n\n        self.api_endpoint = self._get_provider_option(\"plesk_server\")\n\n        if self.api_endpoint.endswith(\"/\"):\n            self.api_endpoint = self.api_endpoint[:-1]\n\n        if not self.api_endpoint.endswith(PLEX_URL_SUFFIX):\n            self.api_endpoint += PLEX_URL_SUFFIX\n\n        self.domain_id = None\n\n        self.username = self._get_provider_option(\"auth_username\")\n        assert self.username is not None\n\n        self.password = self._get_provider_option(\"auth_password\")\n        assert self.password is not None\n\n    def __simple_request(self, rtype, operation, req):\n\n        response = self.__plesk_request({rtype: {operation: req}})[rtype][operation]\n\n        result = response[\"result\"]\n\n        if isinstance(result, list):\n            for record in result:\n                if record[\"status\"] == \"error\":\n                    raise Exception(\n                        f\"API returned at least one error: {record['errtext']}\"\n                    )\n        elif response[\"result\"][\"status\"] == \"error\":\n            errcode = response[\"result\"][\"errcode\"]\n            errtext = response[\"result\"][\"errtext\"]\n            raise Exception(f\"API returned error: {errcode} ({errtext})\")\n\n        return response\n\n    def __plesk_request(self, request):\n\n        headers = {\n            \"Content-type\": \"text/xml\",\n            \"HTTP_PRETTY_PRINT\": \"TRUE\",\n            \"HTTP_AUTH_LOGIN\": self.username,\n            \"HTTP_AUTH_PASSWD\": self.password,\n        }\n\n        xml = f\"\"\"\\\n<?xml version=\"1.0\" encoding=\"utf-8\"?>\n{cElementTree.tostring(_dict_to_etree({\"packet\": request}), encoding=\"unicode\")}\\\n\"\"\"\n\n        LOGGER.debug(\"Request: %s\", xml)\n\n        response = requests.post(\n            self.api_endpoint,\n            headers=headers,\n            data=xml,\n            auth=(self.username, self.password),\n        )\n\n        data = response.text\n\n        LOGGER.debug(\"Response: %s\", data)\n        result = _etree_to_dict(cElementTree.XML(data))\n        return result[\"packet\"]\n\n    def __find_site(self):\n        return self.__simple_request(\n            \"site\", \"get\", {\"filter\": {\"name\": self.domain, \"dataset\": {}}}\n        )[\"result\"][\"id\"]\n\n    def _authenticate(self):\n        self.domain_id = self.__find_site()\n\n        if self.domain_id is None:\n            raise Exception(\"Domain not found\")\n\n    def _create_record(self, rtype, name, content):\n        return self.__create_entry(rtype, name, content, None)\n\n    def _list_records(self, rtype=None, name=None, content=None):\n        entries = self.__find_dns_entries(rtype, name, content)\n        LOGGER.debug(\"list_records: %s\", entries)\n        return entries\n\n    def _update_record(self, identifier, rtype=None, name=None, content=None):\n        if identifier is None:\n            entries = self.__find_dns_entries(rtype, name, None)\n            LOGGER.debug(\"Entries found: %s\", entries)\n\n            if not entries:\n                raise Exception(\"No entry found for updating\")\n\n            identifier = entries[0][\"id\"]\n            entry = self.__get_dns_entry(identifier)\n\n            ids = []\n            for an_entry in entries:\n                ids.append(an_entry[\"id\"])\n\n            self.__delete_dns_records_by_id(ids)\n\n        else:\n\n            entry = self.__get_dns_entry(identifier)\n            self.__delete_dns_records_by_id([identifier])\n\n        assert entry is not None\n\n        LOGGER.debug(\"Updating: %s\", entry)\n\n        if rtype:\n            entry[\"type\"] = rtype\n        if name:\n            entry[\"host\"] = name\n        if content:\n            entry[\"value\"] = content\n\n        return self.__create_entry(\n            entry[\"type\"], entry[\"host\"], entry[\"value\"], entry[\"opt\"]\n        )\n\n    def __create_entry(self, rtype, host, value, opt):\n        entries = self.__find_dns_entries(rtype, self._fqdn_name(host), value)\n\n        if entries:\n            return True  # already exists\n\n        self.__simple_request(\n            \"dns\",\n            \"add_rec\",\n            {\n                \"site-id\": self.domain_id,\n                \"type\": rtype,\n                \"host\": self._relative_name(host),\n                \"value\": value,\n                \"opt\": opt,\n            },\n        )\n\n        return True\n\n    def _delete_record(self, identifier=None, rtype=None, name=None, content=None):\n        if identifier:\n            self.__delete_dns_records_by_id([identifier])\n            return True\n        entries = self.__find_dns_entries(rtype, self._fqdn_name(name), content)\n        ids = []\n\n        for entry in entries:\n            ids.append(entry[\"id\"])\n\n        self.__delete_dns_records_by_id(ids)\n        return bool(ids)\n\n    def __get_dns_entry(self, identifier):\n        return self.__simple_request(\"dns\", \"get_rec\", {\"filter\": {\"id\": identifier}})[\n            \"result\"\n        ][\"data\"]\n\n    def __find_dns_entries(self, rtype=None, host=None, value=None):\n        LOGGER.debug(\"Searching for: %s, %s, %s\", rtype, host, value)\n\n        if value and rtype and rtype in [\"CNAME\"]:\n            LOGGER.debug(\"CNAME transformation\")\n            value = value.rstrip(\".\") + \".\"\n\n        if host:\n            host = self._fqdn_name(host)\n\n        result = self.__simple_request(\n            \"dns\", \"get_rec\", {\"filter\": {\"site-id\": self.domain_id}}\n        )\n\n        entries = []\n\n        for record in result[\"result\"]:\n\n            LOGGER.debug(\"Record: %s\", record)\n\n            if (rtype is not None) and (record[\"data\"][\"type\"] != rtype):\n                LOGGER.debug(\n                    \"\\tType doesn't match - expected: '%s', found: '%s'\",\n                    rtype,\n                    record[\"data\"][\"type\"],\n                )\n                continue\n\n            if (host is not None) and (record[\"data\"][\"host\"] != host):\n                LOGGER.debug(\n                    \"\\tHost doesn't match - expected: '%s', found: '%s'\",\n                    host,\n                    record[\"data\"][\"host\"],\n                )\n                continue\n\n            if (value is not None) and (record[\"data\"][\"value\"] != value):\n                LOGGER.debug(\n                    \"\\tValue doesn't match - expected: '%s', found: '%s'\",\n                    value,\n                    record[\"data\"][\"value\"],\n                )\n                continue\n\n            entry = {\n                \"id\": record[\"id\"],\n                \"type\": record[\"data\"][\"type\"],\n                \"name\": self._full_name(record[\"data\"][\"host\"]),\n                \"ttl\": None,\n                \"options\": {},\n            }\n\n            if record[\"data\"][\"type\"] in [\"CNAME\"]:\n                entry[\"content\"] = record[\"data\"][\"value\"].rstrip(\".\")\n            else:\n                entry[\"content\"] = record[\"data\"][\"value\"]\n\n            if record[\"data\"][\"type\"] == \"MX\":\n                entry[\"options\"][\"mx\"] = {\"priority\": int(record[\"data\"][\"opt\"])}\n\n            entries.append(entry)\n\n        return entries\n\n    def __delete_dns_records_by_id(self, ids):\n        if not ids:\n            return\n\n        req = []\n        for i in ids:\n            req.append({\"del_rec\": {\"filter\": {\"id\": i}}})\n\n        self.__plesk_request({\"dns\": req})\n\n    def _request(self, action=\"GET\", url=\"/\", data=None, query_params=None):\n        # Helper _request is not used for Plesk provider\n        pass\n\n\ndef _etree_to_dict(t: Element) -> Optional[Dict]:\n    d: Optional[Dict] = {t.tag: {} if t.attrib else None}\n    children = list(t)\n    if children:\n        dd = defaultdict(list)\n        for dc in map(_etree_to_dict, children):\n            if dc:\n                for k, v in dc.items():\n                    dd[k].append(v)\n        d = {t.tag: {k: v[0] if len(v) == 1 else v for k, v in dd.items()}}\n    if t.attrib and d:\n        d[t.tag].update((\"@\" + k, v) for k, v in t.attrib.items())\n    if t.text and d:\n        text = t.text.strip()\n        if children or t.attrib:\n            if text:\n                d[t.tag][\"#text\"] = text\n        else:\n            d[t.tag] = text\n    return d\n\n\ndef _dict_to_etree(d: Dict) -> Element:\n    def _to_etree(d1, root):\n        if not d1:\n            pass\n        elif isinstance(d1, str):\n            root.text = d1\n        elif isinstance(d1, dict):\n            for k, v in d1.items():\n                assert isinstance(k, str)\n                if k.startswith(\"#\"):\n                    assert k == \"#text\" and isinstance(v, str)\n                    root.text = v\n                elif k.startswith(\"@\"):\n                    assert isinstance(v, str)\n                    root.set(k[1:], v)\n                elif isinstance(v, list):\n                    for e in v:\n                        _to_etree(e, cElementTree.SubElement(root, k))\n                else:\n                    _to_etree(v, cElementTree.SubElement(root, k))\n        else:\n            raise TypeError(\"invalid type: \" + str(type(d1)))\n\n    assert isinstance(d, dict) and len(d) == 1\n    tag, body = next(iter(d.items()))\n    node = cElementTree.Element(tag)\n    _to_etree(body, node)\n    return node\n", "path": "lexicon/providers/plesk.py"}]}