Add support for Porkbun (closes #667)

CODEOWNERS

@@ -68,6 +68,7 @@ lexicon/providers/online.py         @kapouer
 lexicon/providers/ovh.py            @adferrand
 lexicon/providers/plesk.py          @ctron
 lexicon/providers/pointhq.py        @analogj
+lexicon/providers/porkbun.py        @guidopetri
 lexicon/providers/powerdns.py       @insertjokehere @splashx
 lexicon/providers/rackspace.py      @rmarscher @mattgauf
 lexicon/providers/rage4.py          @analogj

lexicon/providers/porkbun.py

@@ -0,0 +1,171 @@
+import logging
+import requests
+from lexicon.exceptions import AuthenticationError
+from lexicon.providers.base import Provider as BaseProvider
+import json
+
+LOGGER = logging.getLogger(__name__)
+
+NAMESERVER_DOMAINS = ["porkbun.com"]
+
+
+def provider_parser(subparser):
+    """Return the parser for this provider"""
+    subparser.description = """
+        To authenticate with Porkbun, you need both an API key and a
+        secret API key. These can be created at porkbun.com/account/api .
+    """
+
+    subparser.add_argument("--auth-key", help="specify API key for authentication")
+    subparser.add_argument(
+        "--auth-secret", help="specify secret API key for authentication"
+    )
+
+
+class Provider(BaseProvider):
+    """Provider class for Porkbun"""
+
+    def __init__(self, config):
+        super(Provider, self).__init__(config)
+        self.api_endpoint = "https://porkbun.com/api/json/v3"
+        self._api_key = self._get_provider_option("auth_key")
+        self._secret_api_key = self._get_provider_option("auth_secret")
+        self._auth_data = {
+            "apikey": self._api_key,
+            "secretapikey": self._secret_api_key,
+        }
+
+        self.domain = self._get_lexicon_option("domain")
+
+    def _authenticate(self):
+        # more of a test that the authentication works
+        response = self._post("/ping")
+
+        if response["status"] != "SUCCESS":
+            raise AuthenticationError("Incorrect API keys")
+        self.domain_id = self.domain
+        self._list_records()
+
+    def _create_record(self, rtype, name, content):
+        active_records = self._list_records(rtype, name, content)
+        # if the record already exists: early exit, return success
+        if active_records:
+            LOGGER.debug("create_record: record already exists")
+            return True
+
+        data = {
+            "type": rtype,
+            "content": content,
+            "name": self._relative_name(name),
+        }
+
+        if self._get_lexicon_option("priority"):
+            data["prio"] = self._get_lexicon_option("priority")
+
+        if self._get_lexicon_option("ttl"):
+            data["ttl"] = self._get_lexicon_option("ttl")
+
+        response = self._post(f"/dns/create/{self.domain}", data)
+
+        LOGGER.debug(f"create_record: {response}")
+        return response["status"] == "SUCCESS"
+
+    def _list_records(self, rtype=None, name=None, content=None):
+        # porkbun has some weird behavior on the retrieveByNameType endpoint
+        # related to how it handles subdomains.
+        # so we ignore it and filter locally instead
+        records = self._post(f"/dns/retrieve/{self.domain}")
+
+        if records["status"] != "SUCCESS":
+            raise requests.exceptions.HTTPError(records)
+
+        records = records["records"]
+
+        records = self._format_records(records)
+
+        # filter for content if it was provided
+        if content is not None:
+            records = [x for x in records if x["content"] == content]
+
+        # filter for name if it was provided
+        if name is not None:
+            records = [x for x in records if x["name"] == self._full_name(name)]
+
+        # filter for rtype if it was provided
+        if rtype is not None:
+            records = [x for x in records if x["type"] == rtype]
+
+        LOGGER.debug(f"list_records: {records}")
+        LOGGER.debug(f"Number of records retrieved: {len(records)}")
+        return records
+
+    def _update_record(self, identifier=None, rtype=None, name=None, content=None):
+        if identifier is None:
+            records = self._list_records(rtype, name)
+            if len(records) == 1:
+                identifier = records[0]["id"]
+            elif len(records) == 0:
+                raise Exception(
+                    "No records found matching type and name - won't update"
+                )
+            else:
+                raise Exception(
+                    "Multiple records found matching type and name - won't update"
+                )
+
+        endpoint = f"/dns/edit/{self.domain}/{identifier}"
+
+        data = {"name": self._relative_name(name), "type": rtype, "content": content}
+
+        # if set to 0, then this will automatically get set to 300
+        if self._get_lexicon_option("ttl"):
+            data["ttl"] = self._get_lexicon_option("ttl")
+
+        if self._get_lexicon_option("priority"):
+            data["prio"] = self._get_lexicon_option("priority")
+
+        result = self._post(endpoint, data)
+
+        LOGGER.debug(f"update_record: {result}")
+        return result["status"] == "SUCCESS"
+
+    def _delete_record(self, identifier=None, rtype=None, name=None, content=None):
+        if identifier is None:
+            records = self._list_records(rtype, name, content)
+            delete_record_ids = [record["id"] for record in records]
+        else:
+            delete_record_ids = [identifier]
+
+        LOGGER.debug(f"deleting records: {delete_record_ids}")
+
+        for record_id in delete_record_ids:
+            self._post(f"/dns/delete/{self.domain}/{record_id}")
+
+        LOGGER.debug("delete_record: success")
+        return True
+
+    def _request(self, action="GET", url="/", data=None, query_params=None):
+        if data is None:
+            data = {}
+        if query_params is None:
+            query_params = {}
+        headers = {"Content-Type": "application/json"}
+
+        response = requests.request(
+            action,
+            self.api_endpoint + url,
+            params=query_params,
+            data=json.dumps({**data, **self._auth_data}),
+            headers=headers,
+        )
+
+        response.raise_for_status()
+        return response.json()
+
+    def _format_records(self, records):
+        for record in records:
+            record["name"] = self._full_name(record["name"])
+            if "prio" in record:
+                record["options"] = {"mx": {"priority": record["prio"]}}
+                del record["prio"]
+        return records

lexicon/tests/providers/test_porkbun.py

@@ -0,0 +1,27 @@
+# Test for one implementation of the interface
+from lexicon.tests.providers.integration_tests import IntegrationTestsV2
+from unittest import TestCase
+
+
+# Hook into testing framework by inheriting unittest.TestCase and reuse
+# the tests which *each and every* implementation of the interface must
+# pass, by inheritance from integration_tests.IntegrationTests
+class PorkbunProviderTests(TestCase, IntegrationTestsV2):
+    """Integration tests for Porkbun provider"""
+
+    provider_name = "porkbun"
+    domain = "example.xyz"
+
+    def _filter_post_data_parameters(self):
+        return ["login_token", "apikey", "secretapikey"]
+
+    def _filter_headers(self):
+        return ["Authorization"]
+
+    def _filter_query_parameters(self):
+        return ["secret_key"]
+
+    def _filter_response(self, response):
+        """See `IntegrationTests._filter_response` for more information on how
+        to filter the provider response."""
+        return response

tests/fixtures/cassettes/porkbun/IntegrationTests/test_provider_authenticate.yaml

@@ -0,0 +1,120 @@
+interactions:
+- request:
+    body: '{}'
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '170'
+      Content-Type:
+      - application/json
+      User-Agent:
+      - python-requests/2.27.1
+    method: POST
+    uri: https://porkbun.com/api/json/v3/ping
+  response:
+    body:
+      string: '{"status":"SUCCESS","yourIp":"8.8.8.8"}'
+    headers:
+      Cache-Control:
+      - no-store, no-cache, must-revalidate
+      Connection:
+      - keep-alive
+      Content-Language:
+      - en-US, en
+      Content-Type:
+      - application/json
+      Date:
+      - Sun, 12 Jun 2022 01:15:17 GMT
+      Expires:
+      - Thu, 19 Nov 1981 08:52:00 GMT
+      Pragma:
+      - no-cache
+      Referrer-Policy:
+      - origin
+      Server:
+      - openresty
+      Set-Cookie:
+      - AWSALB=B7r1kQtkaZNk+JDxYzONZVyzGKg4pGhlNH6CMAYGQar54tjSRckgI1w8XYfBNr+P8pQOeuocEM4NN/cGs1QVM3tIijqYOMacSZrgIoR98IzEPPQRyNYrmDDYYzbh;
+        Expires=Sun, 19 Jun 2022 01:15:17 GMT; Path=/
+      - AWSALBCORS=B7r1kQtkaZNk+JDxYzONZVyzGKg4pGhlNH6CMAYGQar54tjSRckgI1w8XYfBNr+P8pQOeuocEM4NN/cGs1QVM3tIijqYOMacSZrgIoR98IzEPPQRyNYrmDDYYzbh;
+        Expires=Sun, 19 Jun 2022 01:15:17 GMT; Path=/; SameSite=None; Secure
+      - BUNSESSION2=gDtESXOxSvvBcZ%2CbRXLGtvML1vO784tXkYlfe9aP6M9JriGM; path=/; secure;
+        HttpOnly; SameSite=Lax
+      Strict-Transport-Security:
+      - max-age=63072000; includeSubDomains; preload
+      Transfer-Encoding:
+      - chunked
+      X-Content-Type-Options:
+      - nosniff
+      X-Frame-Options:
+      - sameorigin
+      X-XSS-Protection:
+      - 1; mode=block
+    status:
+      code: 200
+      message: OK
+- request:
+    body: '{}'
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '170'
+      Content-Type:
+      - application/json
+      User-Agent:
+      - python-requests/2.27.1
+    method: POST
+    uri: https://porkbun.com/api/json/v3/dns/retrieve/example.xyz
+  response:
+    body:
+      string: '{"status":"SUCCESS","cloudflare":"disabled","records":[{"id":"236096065","name":"example.xyz","type":"A","content":"8.8.8.8","ttl":"600","prio":"0"},{"id":"236238296","name":"www.example.xyz","type":"A","content":"8.8.8.8","ttl":"600","prio":"0"},{"id":"230514990","name":"example.xyz","type":"ALIAS","content":"pixie.porkbun.com","ttl":"600","prio":null},{"id":"230514991","name":"www.example.xyz","type":"ALIAS","content":"pixie.porkbun.com","ttl":"600","prio":null},{"id":"230514992","name":"*.example.xyz","type":"CNAME","content":"pixie.porkbun.com","ttl":"600","prio":null},{"id":"230514985","name":"example.xyz","type":"MX","content":"fwd1.porkbun.com","ttl":"600","prio":"10"},{"id":"230514986","name":"example.xyz","type":"MX","content":"fwd2.porkbun.com","ttl":"600","prio":"20"}]}'
+    headers:
+      Cache-Control:
+      - no-store, no-cache, must-revalidate
+      Connection:
+      - keep-alive
+      Content-Language:
+      - en-US, en
+      Content-Type:
+      - application/json
+      Date:
+      - Sun, 12 Jun 2022 01:15:18 GMT
+      Expires:
+      - Thu, 19 Nov 1981 08:52:00 GMT
+      Pragma:
+      - no-cache
+      Referrer-Policy:
+      - origin
+      Server:
+      - openresty
+      Set-Cookie:
+      - AWSALB=okCtZrqDYouwVsj+B133KA4OZhdpVtmVMl4mLOpoW2EUf7qR4eQgRuwZUSKZcEWbvYNSLUGMpFzdqEDjnv/mmdpIisHZ7U47tfG1sphcuZBYiXhyPWG7K6qdJvQe;
+        Expires=Sun, 19 Jun 2022 01:15:17 GMT; Path=/
+      - AWSALBCORS=okCtZrqDYouwVsj+B133KA4OZhdpVtmVMl4mLOpoW2EUf7qR4eQgRuwZUSKZcEWbvYNSLUGMpFzdqEDjnv/mmdpIisHZ7U47tfG1sphcuZBYiXhyPWG7K6qdJvQe;
+        Expires=Sun, 19 Jun 2022 01:15:17 GMT; Path=/; SameSite=None; Secure
+      - BUNSESSION2=E-5gLYTfTi3zrhWyRGTmtE%2CsltPckNlkkX4To7Igmsu0y1p4; path=/; secure;
+        HttpOnly; SameSite=Lax
+      Strict-Transport-Security:
+      - max-age=63072000; includeSubDomains; preload
+      Transfer-Encoding:
+      - chunked
+      X-Content-Type-Options:
+      - nosniff
+      X-Frame-Options:
+      - sameorigin
+      X-XSS-Protection:
+      - 1; mode=block
+    status:
+      code: 200
+      message: OK
+version: 1