Merge pull request #1093 from cisagov/dk/902-epp-dnssec

902 - EPPLIB DNS SEC Get and Set DSData and KeyData

src/Pipfile.lock

@@ -353,7 +353,7 @@
         },
         "fred-epplib": {
             "git": "https://github.com/cisagov/epplib.git",
-            "ref": "f818cbf0b069a12f03e1d72e4b9f4900924b832d"
+            "ref": "d56d183f1664f34c40ca9716a3a9a345f0ef561c"
         },
         "furl": {
             "hashes": [

src/epplibwrapper/__init__.py

@@ -11,6 +11,7 @@ logger = logging.getLogger(__name__)
 
 NAMESPACE = SimpleNamespace(
     EPP="urn:ietf:params:xml:ns:epp-1.0",
+    SEC_DNS="urn:ietf:params:xml:ns:secDNS-1.1",
     XSI="http://www.w3.org/2001/XMLSchema-instance",
     FRED="noop",
     NIC_CONTACT="urn:ietf:params:xml:ns:contact-1.0",
@@ -25,6 +26,7 @@ NAMESPACE = SimpleNamespace(
 SCHEMA_LOCATION = SimpleNamespace(
     XSI="urn:ietf:params:xml:ns:epp-1.0 epp-1.0.xsd",
     FRED="noop fred-1.5.0.xsd",
+    SEC_DNS="urn:ietf:params:xml:ns:secDNS-1.1 secDNS-1.1.xsd",
     NIC_CONTACT="urn:ietf:params:xml:ns:contact-1.0 contact-1.0.xsd",
     NIC_DOMAIN="urn:ietf:params:xml:ns:domain-1.0 domain-1.0.xsd",
     NIC_ENUMVAL="noop enumval-1.2.0.xsd",
@@ -45,6 +47,7 @@ try:
     from .client import CLIENT, commands
     from .errors import RegistryError, ErrorCode
     from epplib.models import common
+    from epplib.responses import extensions
     from epplib import responses
 except ImportError:
     pass
@@ -53,6 +56,7 @@ __all__ = [
     "CLIENT",
     "commands",
     "common",
+    "extensions",
     "responses",
     "ErrorCode",
     "RegistryError",

src/registrar/models/domain.py

@@ -10,6 +10,7 @@ from epplibwrapper import (
     CLIENT as registry,
     commands,
     common as epp,
+    extensions,
     RegistryError,
     ErrorCode,
 )
@@ -279,6 +280,27 @@ class Domain(TimeStampedModel, DomainHelper):
             logger.error("Error _create_host, code was %s error was %s" % (e.code, e))
             return e.code
 
+    @Cache
+    def dnssecdata(self) -> extensions.DNSSECExtension:
+        return self._get_property("dnssecdata")
+
+    @dnssecdata.setter  # type: ignore
+    def dnssecdata(self, _dnssecdata: extensions.DNSSECExtension):
+        updateParams = {
+            "maxSigLife": _dnssecdata.get("maxSigLife", None),
+            "dsData": _dnssecdata.get("dsData", None),
+            "keyData": _dnssecdata.get("keyData", None),
+            "remAllDsKeyData": True,
+        }
+        request = commands.UpdateDomain(name=self.name)
+        extension = commands.UpdateDomainDNSSECExtension(**updateParams)
+        request.add_extension(extension)
+        try:
+            registry.send(request, cleaned=True)
+        except RegistryError as e:
+            logger.error("Error adding DNSSEC, code was %s error was %s" % (e.code, e))
+            raise e
+
     @nameservers.setter  # type: ignore
     def nameservers(self, hosts: list[tuple[str]]):
         """host should be a tuple of type str, str,... where the elements are
@@ -720,9 +742,9 @@ class Domain(TimeStampedModel, DomainHelper):
             try:
                 logger.info("Getting domain info from epp")
                 req = commands.InfoDomain(name=self.name)
-                domainInfo = registry.send(req, cleaned=True).res_data[0]
+                domainInfoResponse = registry.send(req, cleaned=True)
                 exitEarly = True
-                return domainInfo
+                return domainInfoResponse
             except RegistryError as e:
                 count += 1
 
@@ -963,7 +985,8 @@ class Domain(TimeStampedModel, DomainHelper):
         """Contact registry for info about a domain."""
         try:
             # get info from registry
-            data = self._get_or_create_domain()
+            dataResponse = self._get_or_create_domain()
+            data = dataResponse.res_data[0]
             # extract properties from response
             # (Ellipsis is used to mean "null")
             cache = {
@@ -986,6 +1009,14 @@ class Domain(TimeStampedModel, DomainHelper):
             if "statuses" in cleaned.keys():
                 cleaned["statuses"] = [status.state for status in cleaned["statuses"]]
 
+            # get extensions info, if there is any
+            # DNSSECExtension is one possible extension, make sure to handle
+            # only DNSSECExtension and not other type extensions
+            returned_extensions = dataResponse.extensions
+            cleaned["dnssecdata"] = None
+            for extension in returned_extensions:
+                if isinstance(extension, extensions.DNSSECExtension):
+                    cleaned["dnssecdata"] = extension
             # Capture and store old hosts and contacts from cache if they exist
             old_cache_hosts = self._cache.get("hosts")
             old_cache_contacts = self._cache.get("contacts")

src/registrar/tests/test_models_domain.py

@@ -3,6 +3,7 @@ Feature being tested: Registry Integration
 
 This file tests the various ways in which the registrar interacts with the registry.
 """
+from typing import Mapping, Any
 from django.test import TestCase
 from django.db.utils import IntegrityError
 from unittest.mock import MagicMock, patch, call
@@ -20,6 +21,7 @@ from django_fsm import TransitionNotAllowed  # type: ignore
 from epplibwrapper import (
     commands,
     common,
+    extensions,
     responses,
     RegistryError,
     ErrorCode,
@@ -778,44 +780,372 @@ class TestRegistrantNameservers(TestCase):
         raise
 
 
-class TestRegistrantDNSSEC(TestCase):
+class TestRegistrantDNSSEC(MockEppLib):
     """Rule: Registrants may modify their secure DNS data"""
 
+    # helper function to create UpdateDomainDNSSECExtention object for verification
+    def createUpdateExtension(self, dnssecdata: extensions.DNSSECExtension):
+        return commands.UpdateDomainDNSSECExtension(
+            maxSigLife=dnssecdata.maxSigLife,
+            dsData=dnssecdata.dsData,
+            keyData=dnssecdata.keyData,
+            remDsData=None,
+            remKeyData=None,
+            remAllDsKeyData=True,
+        )
+
     def setUp(self):
         """
         Background:
-            Given the registrant is logged in
+            Given the analyst is logged in
-            And the registrant is the admin on a domain
+            And a domain exists in the registry
         """
-        pass
+        super().setUp()
+        # for the tests, need a domain in the unknown state
+        self.domain, _ = Domain.objects.get_or_create(name="fake.gov")
+        self.addDsData1 = {
+            "keyTag": 1234,
+            "alg": 3,
+            "digestType": 1,
+            "digest": "ec0bdd990b39feead889f0ba613db4adec0bdd99",
+        }
+        self.addDsData2 = {
+            "keyTag": 2345,
+            "alg": 3,
+            "digestType": 1,
+            "digest": "ec0bdd990b39feead889f0ba613db4adecb4adec",
+        }
+        self.keyDataDict = {
+            "flags": 257,
+            "protocol": 3,
+            "alg": 1,
+            "pubKey": "AQPJ////4Q==",
+        }
+        self.dnssecExtensionWithDsData: Mapping[str, Any] = {
+            "dsData": [common.DSData(**self.addDsData1)]
+        }
+        self.dnssecExtensionWithMultDsData: Mapping[str, Any] = {
+            "dsData": [
+                common.DSData(**self.addDsData1),
+                common.DSData(**self.addDsData2),
+            ],
+        }
+        self.dnssecExtensionWithKeyData: Mapping[str, Any] = {
+            "maxSigLife": 3215,
+            "keyData": [common.DNSSECKeyData(**self.keyDataDict)],
+        }
 
-    @skip("not implemented yet")
+    def tearDown(self):
-    def test_user_adds_dns_data(self):
+        Domain.objects.all().delete()
+        super().tearDown()
+
+    def test_user_adds_dnssec_data(self):
         """
-        Scenario: Registrant adds DNS data
+        Scenario: Registrant adds DNSSEC data.
+        Verify that both the setter and getter are functioning properly
+
+        This test verifies:
+        1 - setter calls UpdateDomain command
+        2 - setter adds the UpdateDNSSECExtension extension to the command
+        3 - setter causes the getter to call info domain on next get from cache
+        4 - getter properly parses dnssecdata from InfoDomain response and sets to cache
 
         """
-        raise
 
-    @skip("not implemented yet")
+        # make sure to stop any other patcher so there are no conflicts
+        self.mockSendPatch.stop()
+
+        def side_effect(_request, cleaned):
+            return MagicMock(
+                res_data=[self.mockDataInfoDomain],
+                extensions=[
+                    extensions.DNSSECExtension(**self.dnssecExtensionWithDsData)
+                ],
+            )
+
+        patcher = patch("registrar.models.domain.registry.send")
+        mocked_send = patcher.start()
+        mocked_send.side_effect = side_effect
+
+        self.domain.dnssecdata = self.dnssecExtensionWithDsData
+        # get the DNS SEC extension added to the UpdateDomain command and
+        # verify that it is properly sent
+        # args[0] is the _request sent to registry
+        args, _ = mocked_send.call_args
+        # assert that the extension matches
+        self.assertEquals(
+            args[0].extensions[0],
+            self.createUpdateExtension(
+                extensions.DNSSECExtension(**self.dnssecExtensionWithDsData)
+            ),
+        )
+        # test that the dnssecdata getter is functioning properly
+        dnssecdata_get = self.domain.dnssecdata
+        mocked_send.assert_has_calls(
+            [
+                call(
+                    commands.UpdateDomain(
+                        name="fake.gov",
+                        nsset=None,
+                        keyset=None,
+                        registrant=None,
+                        auth_info=None,
+                    ),
+                    cleaned=True,
+                ),
+                call(
+                    commands.InfoDomain(
+                        name="fake.gov",
+                    ),
+                    cleaned=True,
+                ),
+            ]
+        )
+
+        self.assertEquals(
+            dnssecdata_get.dsData, self.dnssecExtensionWithDsData["dsData"]
+        )
+
+        patcher.stop()
+
     def test_dnssec_is_idempotent(self):
         """
         Scenario: Registrant adds DNS data twice, due to a UI glitch
 
-        """
         # implementation note: this requires seeing what happens when these are actually
         # sent like this, and then implementing appropriate mocks for any errors the
         # registry normally sends in this case
-        raise
 
-    @skip("not implemented yet")
+        This test verifies:
+        1 - UpdateDomain command called twice
+        2 - setter causes the getter to call info domain on next get from cache
+        3 - getter properly parses dnssecdata from InfoDomain response and sets to cache
+
+        """
+
+        # make sure to stop any other patcher so there are no conflicts
+        self.mockSendPatch.stop()
+
+        def side_effect(_request, cleaned):
+            return MagicMock(
+                res_data=[self.mockDataInfoDomain],
+                extensions=[
+                    extensions.DNSSECExtension(**self.dnssecExtensionWithDsData)
+                ],
+            )
+
+        patcher = patch("registrar.models.domain.registry.send")
+        mocked_send = patcher.start()
+        mocked_send.side_effect = side_effect
+
+        # set the dnssecdata once
+        self.domain.dnssecdata = self.dnssecExtensionWithDsData
+        # set the dnssecdata again
+        self.domain.dnssecdata = self.dnssecExtensionWithDsData
+        # test that the dnssecdata getter is functioning properly
+        dnssecdata_get = self.domain.dnssecdata
+        mocked_send.assert_has_calls(
+            [
+                call(
+                    commands.UpdateDomain(
+                        name="fake.gov",
+                        nsset=None,
+                        keyset=None,
+                        registrant=None,
+                        auth_info=None,
+                    ),
+                    cleaned=True,
+                ),
+                call(
+                    commands.UpdateDomain(
+                        name="fake.gov",
+                        nsset=None,
+                        keyset=None,
+                        registrant=None,
+                        auth_info=None,
+                    ),
+                    cleaned=True,
+                ),
+                call(
+                    commands.InfoDomain(
+                        name="fake.gov",
+                    ),
+                    cleaned=True,
+                ),
+            ]
+        )
+
+        self.assertEquals(
+            dnssecdata_get.dsData, self.dnssecExtensionWithDsData["dsData"]
+        )
+
+        patcher.stop()
+
+    def test_user_adds_dnssec_data_multiple_dsdata(self):
+        """
+        Scenario: Registrant adds DNSSEC data with multiple DSData.
+        Verify that both the setter and getter are functioning properly
+
+        This test verifies:
+        1 - setter calls UpdateDomain command
+        2 - setter adds the UpdateDNSSECExtension extension to the command
+        3 - setter causes the getter to call info domain on next get from cache
+        4 - getter properly parses dnssecdata from InfoDomain response and sets to cache
+
+        """
+
+        # make sure to stop any other patcher so there are no conflicts
+        self.mockSendPatch.stop()
+
+        def side_effect(_request, cleaned):
+            return MagicMock(
+                res_data=[self.mockDataInfoDomain],
+                extensions=[
+                    extensions.DNSSECExtension(**self.dnssecExtensionWithMultDsData)
+                ],
+            )
+
+        patcher = patch("registrar.models.domain.registry.send")
+        mocked_send = patcher.start()
+        mocked_send.side_effect = side_effect
+
+        self.domain.dnssecdata = self.dnssecExtensionWithMultDsData
+        # get the DNS SEC extension added to the UpdateDomain command
+        # and verify that it is properly sent
+        # args[0] is the _request sent to registry
+        args, _ = mocked_send.call_args
+        # assert that the extension matches
+        self.assertEquals(
+            args[0].extensions[0],
+            self.createUpdateExtension(
+                extensions.DNSSECExtension(**self.dnssecExtensionWithMultDsData)
+            ),
+        )
+        # test that the dnssecdata getter is functioning properly
+        dnssecdata_get = self.domain.dnssecdata
+        mocked_send.assert_has_calls(
+            [
+                call(
+                    commands.UpdateDomain(
+                        name="fake.gov",
+                        nsset=None,
+                        keyset=None,
+                        registrant=None,
+                        auth_info=None,
+                    ),
+                    cleaned=True,
+                ),
+                call(
+                    commands.InfoDomain(
+                        name="fake.gov",
+                    ),
+                    cleaned=True,
+                ),
+            ]
+        )
+
+        self.assertEquals(
+            dnssecdata_get.dsData, self.dnssecExtensionWithMultDsData["dsData"]
+        )
+
+        patcher.stop()
+
+    def test_user_adds_dnssec_keydata(self):
+        """
+        Scenario: Registrant adds DNSSEC data.
+        Verify that both the setter and getter are functioning properly
+
+        This test verifies:
+        1 - setter calls UpdateDomain command
+        2 - setter adds the UpdateDNSSECExtension extension to the command
+        3 - setter causes the getter to call info domain on next get from cache
+        4 - getter properly parses dnssecdata from InfoDomain response and sets to cache
+
+        """
+
+        # make sure to stop any other patcher so there are no conflicts
+        self.mockSendPatch.stop()
+
+        def side_effect(_request, cleaned):
+            return MagicMock(
+                res_data=[self.mockDataInfoDomain],
+                extensions=[
+                    extensions.DNSSECExtension(**self.dnssecExtensionWithKeyData)
+                ],
+            )
+
+        patcher = patch("registrar.models.domain.registry.send")
+        mocked_send = patcher.start()
+        mocked_send.side_effect = side_effect
+
+        self.domain.dnssecdata = self.dnssecExtensionWithKeyData
+        # get the DNS SEC extension added to the UpdateDomain command
+        # and verify that it is properly sent
+        # args[0] is the _request sent to registry
+        args, _ = mocked_send.call_args
+        # assert that the extension matches
+        self.assertEquals(
+            args[0].extensions[0],
+            self.createUpdateExtension(
+                extensions.DNSSECExtension(**self.dnssecExtensionWithKeyData)
+            ),
+        )
+        # test that the dnssecdata getter is functioning properly
+        dnssecdata_get = self.domain.dnssecdata
+        mocked_send.assert_has_calls(
+            [
+                call(
+                    commands.UpdateDomain(
+                        name="fake.gov",
+                        nsset=None,
+                        keyset=None,
+                        registrant=None,
+                        auth_info=None,
+                    ),
+                    cleaned=True,
+                ),
+                call(
+                    commands.InfoDomain(
+                        name="fake.gov",
+                    ),
+                    cleaned=True,
+                ),
+            ]
+        )
+
+        self.assertEquals(
+            dnssecdata_get.keyData, self.dnssecExtensionWithKeyData["keyData"]
+        )
+
+        patcher.stop()
+
     def test_update_is_unsuccessful(self):
         """
         Scenario: An update to the dns data is unsuccessful
             When an error is returned from epplibwrapper
             Then a user-friendly error message is returned for displaying on the web
         """
-        raise
+
+        # make sure to stop any other patcher so there are no conflicts
+        self.mockSendPatch.stop()
+
+        def side_effect(_request, cleaned):
+            raise RegistryError(code=ErrorCode.PARAMETER_VALUE_RANGE_ERROR)
+
+        patcher = patch("registrar.models.domain.registry.send")
+        mocked_send = patcher.start()
+        mocked_send.side_effect = side_effect
+
+        # if RegistryError is raised, view formats user-friendly
+        # error message if error is_client_error, is_session_error, or
+        # is_server_error; so test for those conditions
+        with self.assertRaises(RegistryError) as err:
+            self.domain.dnssecdata = self.dnssecExtensionWithDsData
+            self.assertTrue(
+                err.is_client_error() or err.is_session_error() or err.is_server_error()
+            )
+
+        patcher.stop()
 
 
 class TestAnalystClientHold(MockEppLib):

src/requirements.txt

@@ -22,7 +22,7 @@ django-phonenumber-field[phonenumberslite]==7.1.0
 django-widget-tweaks==1.4.12
 environs[django]==9.5.0
 faker==18.10.0
-git+https://github.com/cisagov/epplib.git@f818cbf0b069a12f03e1d72e4b9f4900924b832d#egg=fred-epplib
+git+https://github.com/cisagov/epplib.git@d56d183f1664f34c40ca9716a3a9a345f0ef561c#egg=fred-epplib
 furl==2.1.3
 future==0.18.3 ; python_version >= '2.6' and python_version not in '3.0, 3.1, 3.2, 3.3'
 gunicorn==20.1.0