Merge remote-tracking branch 'origin/main' into rjm/1027-groups-permissions-logging

This commit is contained in:
Rachid Mrad 2023-10-03 18:46:22 -04:00
commit 3ea1f5aa77
No known key found for this signature in database
GPG key ID: EF38E4CEC4A8F3CF
16 changed files with 952 additions and 141 deletions

View file

@ -19,6 +19,7 @@ jobs:
|| startsWith(github.head_ref, 'rh/')
|| startsWith(github.head_ref, 'nl/')
|| startsWith(github.head_ref, 'dk/')
|| startsWith(github.head_ref, 'es/')
outputs:
environment: ${{ steps.var.outputs.environment}}
runs-on: "ubuntu-latest"

View file

@ -15,6 +15,7 @@ on:
options:
- stable
- staging
- es
- nl
- rh
- za

View file

@ -16,6 +16,7 @@ on:
options:
- stable
- staging
- es
- nl
- rh
- za

View file

@ -0,0 +1,30 @@
---
applications:
- name: getgov-es
buildpacks:
- python_buildpack
path: ../../src
instances: 1
memory: 512M
stack: cflinuxfs4
timeout: 180
command: ./run.sh
health-check-type: http
health-check-http-endpoint: /health
health-check-invocation-timeout: 40
env:
# Send stdout and stderr straight to the terminal without buffering
PYTHONUNBUFFERED: yup
# Tell Django where to find its configuration
DJANGO_SETTINGS_MODULE: registrar.config.settings
# Tell Django where it is being hosted
DJANGO_BASE_URL: https://getgov-es.app.cloud.gov
# Tell Django how much stuff to log
DJANGO_LOG_LEVEL: INFO
# default public site location
GETGOV_PUBLIC_SITE_URL: https://beta.get.gov
routes:
- route: getgov-es.app.cloud.gov
services:
- getgov-credentials
- getgov-es-database

2
src/Pipfile.lock generated
View file

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

View file

@ -39,7 +39,7 @@ class AvailableViewTest(TestCase):
self.assertIn("gsa.gov", domains)
# entries are all lowercase so GSA.GOV is not in the set
self.assertNotIn("GSA.GOV", domains)
self.assertNotIn("igorville.gov", domains)
self.assertNotIn("igorvilleremixed.gov", domains)
# all the entries have dots
self.assertNotIn("gsa", domains)
@ -48,7 +48,7 @@ class AvailableViewTest(TestCase):
# input is lowercased so GSA.GOV should be found
self.assertTrue(in_domains("GSA.GOV"))
# This domain should not have been registered
self.assertFalse(in_domains("igorville.gov"))
self.assertFalse(in_domains("igorvilleremixed.gov"))
def test_in_domains_dotgov(self):
"""Domain searches work without trailing .gov"""
@ -56,7 +56,7 @@ class AvailableViewTest(TestCase):
# input is lowercased so GSA.GOV should be found
self.assertTrue(in_domains("GSA"))
# This domain should not have been registered
self.assertFalse(in_domains("igorville"))
self.assertFalse(in_domains("igorvilleremixed"))
def test_not_available_domain(self):
"""gsa.gov is not available"""
@ -66,17 +66,17 @@ class AvailableViewTest(TestCase):
self.assertFalse(json.loads(response.content)["available"])
def test_available_domain(self):
"""igorville.gov is still available"""
request = self.factory.get(API_BASE_PATH + "igorville.gov")
"""igorvilleremixed.gov is still available"""
request = self.factory.get(API_BASE_PATH + "igorvilleremixed.gov")
request.user = self.user
response = available(request, domain="igorville.gov")
response = available(request, domain="igorvilleremixed.gov")
self.assertTrue(json.loads(response.content)["available"])
def test_available_domain_dotgov(self):
"""igorville.gov is still available even without the .gov suffix"""
request = self.factory.get(API_BASE_PATH + "igorville")
"""igorvilleremixed.gov is still available even without the .gov suffix"""
request = self.factory.get(API_BASE_PATH + "igorvilleremixed")
request.user = self.user
response = available(request, domain="igorville")
response = available(request, domain="igorvilleremixed")
self.assertTrue(json.loads(response.content)["available"])
def test_error_handling(self):

View file

@ -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,8 @@ 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
@ -52,6 +56,8 @@ __all__ = [
"CLIENT",
"commands",
"common",
"extensions",
"responses",
"ErrorCode",
"RegistryError",
]

View file

@ -7,10 +7,13 @@ from django.contrib.auth.models import Group
from django.contrib.contenttypes.models import ContentType
from django.http.response import HttpResponseRedirect
from django.urls import reverse
from epplibwrapper.errors import ErrorCode, RegistryError
from registrar.models.domain import Domain
from registrar.models.utility.admin_sort_fields import AdminSortFields
from . import models
from auditlog.models import LogEntry # type: ignore
from auditlog.admin import LogEntryAdmin # type: ignore
from django_fsm import TransitionNotAllowed # type: ignore
logger = logging.getLogger(__name__)
@ -736,16 +739,61 @@ class DomainAdmin(ListHeaderAdmin):
return super().response_change(request, obj)
def do_delete_domain(self, request, obj):
if not isinstance(obj, Domain):
# Could be problematic if the type is similar,
# but not the same (same field/func names).
# We do not want to accidentally delete records.
self.message_user(request, "Object is not of type Domain", messages.ERROR)
return
try:
obj.deleted()
obj.deletedInEpp()
obj.save()
except Exception as err:
self.message_user(request, err, messages.ERROR)
except RegistryError as err:
# Using variables to get past the linter
message1 = f"Cannot delete Domain when in state {obj.state}"
message2 = "This subdomain is being used as a hostname on another domain"
# Human-readable mappings of ErrorCodes. Can be expanded.
error_messages = {
# noqa on these items as black wants to reformat to an invalid length
ErrorCode.OBJECT_STATUS_PROHIBITS_OPERATION: message1,
ErrorCode.OBJECT_ASSOCIATION_PROHIBITS_OPERATION: message2,
}
message = "Cannot connect to the registry"
if not err.is_connection_error():
# If nothing is found, will default to returned err
message = error_messages.get(err.code, err)
self.message_user(
request, f"Error deleting this Domain: {message}", messages.ERROR
)
except TransitionNotAllowed:
if obj.state == Domain.State.DELETED:
self.message_user(
request,
"This domain is already deleted",
messages.INFO,
)
else:
self.message_user(
request,
("Domain %s Should now be deleted " ". Thanks!") % obj.name,
"Error deleting this Domain: "
f"Can't switch from state '{obj.state}' to 'deleted'"
", must be either 'dns_needed' or 'on_hold'",
messages.ERROR,
)
except Exception:
self.message_user(
request,
"Could not delete: An unspecified error occured",
messages.ERROR,
)
else:
self.message_user(
request,
("Domain %s has been deleted. Thanks!") % obj.name,
)
return HttpResponseRedirect(".")
def do_get_status(self, request, obj):

View file

@ -570,6 +570,7 @@ SECURE_SSL_REDIRECT = True
ALLOWED_HOSTS = [
"getgov-stable.app.cloud.gov",
"getgov-staging.app.cloud.gov",
"getgov-es.app.cloud.gov",
"getgov-nl.app.cloud.gov",
"getgov-rh.app.cloud.gov",
"getgov-za.app.cloud.gov",

View file

@ -2,7 +2,7 @@ import logging
from datetime import date
from string import digits
from django_fsm import FSMField, transition # type: ignore
from django_fsm import FSMField, transition, TransitionNotAllowed # type: ignore
from django.db import models
@ -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
@ -609,11 +631,6 @@ class Domain(TimeStampedModel, DomainHelper):
"""
return self.state == self.State.READY
def delete_request(self):
"""Delete from host. Possibly a duplicate of _delete_host?"""
# TODO fix in ticket #901
pass
def transfer(self):
"""Going somewhere. Not implemented."""
raise NotImplementedError()
@ -658,7 +675,7 @@ class Domain(TimeStampedModel, DomainHelper):
"""This domain should be deleted from the registry
may raises RegistryError, should be caught or handled correctly by caller"""
request = commands.DeleteDomain(name=self.name)
registry.send(request)
registry.send(request, cleaned=True)
def __str__(self) -> str:
return self.name
@ -725,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
@ -804,16 +821,32 @@ class Domain(TimeStampedModel, DomainHelper):
self._remove_client_hold()
# TODO -on the client hold ticket any additional error handling here
@transition(field="state", source=State.ON_HOLD, target=State.DELETED)
def deleted(self):
"""domain is deleted in epp but is saved in our database"""
# TODO Domains may not be deleted if:
# a child host is being used by
# another .gov domains. The host must be first removed
# and/or renamed before the parent domain may be deleted.
logger.info("pendingCreate()-> inside pending create")
@transition(
field="state", source=[State.ON_HOLD, State.DNS_NEEDED], target=State.DELETED
)
def deletedInEpp(self):
"""Domain is deleted in epp but is saved in our database.
Error handling should be provided by the caller."""
# While we want to log errors, we want to preserve
# that information when this function is called.
# Human-readable errors are introduced at the admin.py level,
# as doing everything here would reduce reliablity.
try:
logger.info("deletedInEpp()-> inside _delete_domain")
self._delete_domain()
# TODO - delete ticket any additional error handling here
except RegistryError as err:
logger.error(f"Could not delete domain. Registry returned error: {err}")
raise err
except TransitionNotAllowed as err:
logger.error("Could not delete domain. FSM failure: {err}")
raise err
except Exception as err:
logger.error(
f"Could not delete domain. An unspecified error occured: {err}"
)
raise err
else:
self._invalidate_cache()
@transition(
field="state",
@ -952,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 = {
@ -974,27 +1008,59 @@ class Domain(TimeStampedModel, DomainHelper):
# statuses can just be a list no need to keep the epp object
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")
# get contact info, if there are any
if (
# fetch_contacts and
"_contacts" in cleaned
fetch_contacts
and "_contacts" in cleaned
and isinstance(cleaned["_contacts"], list)
and len(cleaned["_contacts"])
):
cleaned["contacts"] = []
for domainContact in cleaned["_contacts"]:
# we do not use _get_or_create_* because we expect the object we
# just asked the registry for still exists --
# if not, that's a problem
cleaned["contacts"] = self._fetch_contacts(cleaned["_contacts"])
# We're only getting contacts, so retain the old
# hosts that existed in cache (if they existed)
# and pass them along.
if old_cache_hosts is not None:
cleaned["hosts"] = old_cache_hosts
# TODO- discuss-should we check if contact is in public contacts
# and add it if not- this is really to keep in mine the transisiton
# get nameserver info, if there are any
if (
fetch_hosts
and "_hosts" in cleaned
and isinstance(cleaned["_hosts"], list)
and len(cleaned["_hosts"])
):
cleaned["hosts"] = self._fetch_hosts(cleaned["_hosts"])
# We're only getting hosts, so retain the old
# contacts that existed in cache (if they existed)
# and pass them along.
if old_cache_contacts is not None:
cleaned["contacts"] = old_cache_contacts
# replace the prior cache with new data
self._cache = cleaned
except RegistryError as e:
logger.error(e)
def _fetch_contacts(self, contact_data):
"""Fetch contact info."""
contacts = []
for domainContact in contact_data:
req = commands.InfoContact(id=domainContact.contact)
data = registry.send(req, cleaned=True).res_data[0]
# extract properties from response
# (Ellipsis is used to mean "null")
# convert this to use PublicContactInstead
contact = {
"id": domainContact.contact,
"type": domainContact.type,
@ -1009,29 +1075,15 @@ class Domain(TimeStampedModel, DomainHelper):
"up_date": getattr(data, "up_date", ...),
"voice": getattr(data, "voice", ...),
}
contacts.append({k: v for k, v in contact.items() if v is not ...})
return contacts
cleaned["contacts"].append(
{k: v for k, v in contact.items() if v is not ...}
)
# get nameserver info, if there are any
if (
# fetch_hosts and
"_hosts" in cleaned
and isinstance(cleaned["_hosts"], list)
and len(cleaned["_hosts"])
):
# TODO- add elif in cache set it to be the old cache value
# no point in removing
cleaned["hosts"] = []
for name in cleaned["_hosts"]:
# we do not use _get_or_create_* because we expect the object we
# just asked the registry for still exists --
# if not, that's a problem
def _fetch_hosts(self, host_data):
"""Fetch host info."""
hosts = []
for name in host_data:
req = commands.InfoHost(name=name)
data = registry.send(req, cleaned=True).res_data[0]
# extract properties from response
# (Ellipsis is used to mean "null")
host = {
"name": name,
"addrs": getattr(data, "addrs", ...),
@ -1040,15 +1092,8 @@ class Domain(TimeStampedModel, DomainHelper):
"tr_date": getattr(data, "tr_date", ...),
"up_date": getattr(data, "up_date", ...),
}
cleaned["hosts"].append(
{k: v for k, v in host.items() if v is not ...}
)
# replace the prior cache with new data
self._cache = cleaned
except RegistryError as e:
logger.error(e)
hosts.append({k: v for k, v in host.items() if v is not ...})
return hosts
def _invalidate_cache(self):
"""Remove cache data when updates are made."""

View file

@ -6,6 +6,7 @@ import logging
from django.apps import apps
from django.db import models
from django_fsm import FSMField, transition # type: ignore
from registrar.models.domain import Domain
from .utility.time_stamped_model import TimeStampedModel
from ..utility.email import send_templated_email, EmailSendingError
@ -610,9 +611,11 @@ class DomainApplication(TimeStampedModel):
As side effects this will delete the domain and domain_information
(will cascade), and send an email notification."""
if self.status == self.APPROVED:
self.approved_domain.delete_request()
domain_state = self.approved_domain.state
# Only reject if it exists on EPP
if domain_state != Domain.State.UNKNOWN:
self.approved_domain.deletedInEpp()
self.approved_domain.delete()
self.approved_domain = None
@ -638,7 +641,10 @@ class DomainApplication(TimeStampedModel):
and domain_information (will cascade) when they exist."""
if self.status == self.APPROVED:
self.approved_domain.delete_request()
domain_state = self.approved_domain.state
# Only reject if it exists on EPP
if domain_state != Domain.State.UNKNOWN:
self.approved_domain.deletedInEpp()
self.approved_domain.delete()
self.approved_domain = None

View file

@ -15,7 +15,9 @@
{% endif %}
<input id="manageDomainSubmitButton" type="submit" value="Manage Domain" name="_edit_domain">
<input type="submit" value="get status" name="_get_status">
<input type="submit" value="EPP Delete Domain" name="_delete_domain">
{% if original.state != original.State.DELETED %}
<input type="submit" value="Delete Domain in Registry" name="_delete_domain">
{% endif %}
</div>
{{ block.super }}
{% endblock %}

View file

@ -619,6 +619,16 @@ class MockEppLib(TestCase):
# use this for when a contact is being updated
# sets the second send() to fail
raise RegistryError(code=ErrorCode.OBJECT_EXISTS)
elif (
isinstance(_request, commands.DeleteDomain)
and getattr(_request, "name", None) == "failDelete.gov"
):
name = getattr(_request, "name", None)
fake_nameserver = "ns1.failDelete.gov"
if name in fake_nameserver:
raise RegistryError(
code=ErrorCode.OBJECT_ASSOCIATION_PROHIBITS_OPERATION
)
return MagicMock(res_data=[self.mockDataInfoHosts])
def setUp(self):

View file

@ -49,6 +49,7 @@ class TestDomainAdmin(MockEppLib):
self.client = Client(HTTP_HOST="localhost:8080")
self.superuser = create_superuser()
self.staffuser = create_user()
self.factory = RequestFactory()
super().setUp()
@skip("Why did this test stop working, and is is a good test")
@ -88,6 +89,155 @@ class TestDomainAdmin(MockEppLib):
self.assertContains(response, "Place hold")
self.assertNotContains(response, "Remove hold")
def test_deletion_is_successful(self):
"""
Scenario: Domain deletion is unsuccessful
When the domain is deleted
Then a user-friendly success message is returned for displaying on the web
And `state` is et to `DELETED`
"""
domain = create_ready_domain()
# Put in client hold
domain.place_client_hold()
p = "userpass"
self.client.login(username="staffuser", password=p)
# Ensure everything is displaying correctly
response = self.client.get(
"/admin/registrar/domain/{}/change/".format(domain.pk),
follow=True,
)
self.assertEqual(response.status_code, 200)
self.assertContains(response, domain.name)
self.assertContains(response, "Delete Domain in Registry")
# Test the info dialog
request = self.factory.post(
"/admin/registrar/domain/{}/change/".format(domain.pk),
{"_delete_domain": "Delete Domain in Registry", "name": domain.name},
follow=True,
)
request.user = self.client
with patch("django.contrib.messages.add_message") as mock_add_message:
self.admin.do_delete_domain(request, domain)
mock_add_message.assert_called_once_with(
request,
messages.INFO,
"Domain city.gov has been deleted. Thanks!",
extra_tags="",
fail_silently=False,
)
self.assertEqual(domain.state, Domain.State.DELETED)
def test_deletion_ready_fsm_failure(self):
"""
Scenario: Domain deletion is unsuccessful
When an error is returned from epplibwrapper
Then a user-friendly error message is returned for displaying on the web
And `state` is not set to `DELETED`
"""
domain = create_ready_domain()
p = "userpass"
self.client.login(username="staffuser", password=p)
# Ensure everything is displaying correctly
response = self.client.get(
"/admin/registrar/domain/{}/change/".format(domain.pk),
follow=True,
)
self.assertEqual(response.status_code, 200)
self.assertContains(response, domain.name)
self.assertContains(response, "Delete Domain in Registry")
# Test the error
request = self.factory.post(
"/admin/registrar/domain/{}/change/".format(domain.pk),
{"_delete_domain": "Delete Domain in Registry", "name": domain.name},
follow=True,
)
request.user = self.client
with patch("django.contrib.messages.add_message") as mock_add_message:
self.admin.do_delete_domain(request, domain)
mock_add_message.assert_called_once_with(
request,
messages.ERROR,
"Error deleting this Domain: "
"Can't switch from state 'ready' to 'deleted'"
", must be either 'dns_needed' or 'on_hold'",
extra_tags="",
fail_silently=False,
)
self.assertEqual(domain.state, Domain.State.READY)
def test_analyst_deletes_domain_idempotent(self):
"""
Scenario: Analyst tries to delete an already deleted domain
Given `state` is already `DELETED`
When `domain.deletedInEpp()` is called
Then `commands.DeleteDomain` is sent to the registry
And Domain returns normally without an error dialog
"""
domain = create_ready_domain()
# Put in client hold
domain.place_client_hold()
p = "userpass"
self.client.login(username="staffuser", password=p)
# Ensure everything is displaying correctly
response = self.client.get(
"/admin/registrar/domain/{}/change/".format(domain.pk),
follow=True,
)
self.assertEqual(response.status_code, 200)
self.assertContains(response, domain.name)
self.assertContains(response, "Delete Domain in Registry")
# Test the info dialog
request = self.factory.post(
"/admin/registrar/domain/{}/change/".format(domain.pk),
{"_delete_domain": "Delete Domain in Registry", "name": domain.name},
follow=True,
)
request.user = self.client
# Delete it once
with patch("django.contrib.messages.add_message") as mock_add_message:
self.admin.do_delete_domain(request, domain)
mock_add_message.assert_called_once_with(
request,
messages.INFO,
"Domain city.gov has been deleted. Thanks!",
extra_tags="",
fail_silently=False,
)
self.assertEqual(domain.state, Domain.State.DELETED)
# Try to delete it again
# Test the info dialog
request = self.factory.post(
"/admin/registrar/domain/{}/change/".format(domain.pk),
{"_delete_domain": "Delete Domain in Registry", "name": domain.name},
follow=True,
)
request.user = self.client
with patch("django.contrib.messages.add_message") as mock_add_message:
self.admin.do_delete_domain(request, domain)
mock_add_message.assert_called_once_with(
request,
messages.INFO,
"This domain is already deleted",
extra_tags="",
fail_silently=False,
)
self.assertEqual(domain.state, Domain.State.DELETED)
@skip("Waiting on epp lib to implement")
def test_place_and_remove_hold_epp(self):
raise
@ -139,8 +289,9 @@ class TestDomainApplicationAdminForm(TestCase):
)
class TestDomainApplicationAdmin(TestCase):
class TestDomainApplicationAdmin(MockEppLib):
def setUp(self):
super().setUp()
self.site = AdminSite()
self.factory = RequestFactory()
self.admin = DomainApplicationAdmin(
@ -691,6 +842,7 @@ class TestDomainApplicationAdmin(TestCase):
domain_information.refresh_from_db()
def tearDown(self):
super().tearDown()
Domain.objects.all().delete()
DomainInformation.objects.all().delete()
DomainApplication.objects.all().delete()

View file

@ -3,9 +3,10 @@ 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 patch, call
from unittest.mock import MagicMock, patch, call
import datetime
from registrar.models import Domain
@ -16,10 +17,12 @@ from registrar.models.draft_domain import DraftDomain
from registrar.models.public_contact import PublicContact
from registrar.models.user import User
from .common import MockEppLib
from django_fsm import TransitionNotAllowed # type: ignore
from epplibwrapper import (
commands,
common,
extensions,
responses,
RegistryError,
ErrorCode,
)
@ -51,8 +54,6 @@ class TestDomainCache(MockEppLib):
commands.InfoDomain(name="igorville.gov", auth_info=None),
cleaned=True,
),
call(commands.InfoContact(id="123", auth_info=None), cleaned=True),
call(commands.InfoHost(name="fake.host.com"), cleaned=True),
],
any_order=False, # Ensure calls are in the specified order
)
@ -74,8 +75,6 @@ class TestDomainCache(MockEppLib):
call(
commands.InfoDomain(name="igorville.gov", auth_info=None), cleaned=True
),
call(commands.InfoContact(id="123", auth_info=None), cleaned=True),
call(commands.InfoHost(name="fake.host.com"), cleaned=True),
]
self.mockedSendFunction.assert_has_calls(expectedCalls)
@ -110,6 +109,19 @@ class TestDomainCache(MockEppLib):
# get and check hosts is set correctly
domain._get_property("hosts")
self.assertEqual(domain._cache["hosts"], [expectedHostsDict])
self.assertEqual(domain._cache["contacts"], [expectedContactsDict])
# invalidate cache
domain._cache = {}
# get host
domain._get_property("hosts")
self.assertEqual(domain._cache["hosts"], [expectedHostsDict])
# get contacts
domain._get_property("contacts")
self.assertEqual(domain._cache["hosts"], [expectedHostsDict])
self.assertEqual(domain._cache["contacts"], [expectedContactsDict])
def tearDown(self) -> None:
Domain.objects.all().delete()
@ -170,8 +182,6 @@ class TestDomainCreation(MockEppLib):
commands.InfoDomain(name="beef-tongue.gov", auth_info=None),
cleaned=True,
),
call(commands.InfoContact(id="123", auth_info=None), cleaned=True),
call(commands.InfoHost(name="fake.host.com"), cleaned=True),
],
any_order=False, # Ensure calls are in the specified order
)
@ -221,8 +231,6 @@ class TestDomainStatuses(MockEppLib):
commands.InfoDomain(name="chicken-liver.gov", auth_info=None),
cleaned=True,
),
call(commands.InfoContact(id="123", auth_info=None), cleaned=True),
call(commands.InfoHost(name="fake.host.com"), cleaned=True),
],
any_order=False, # Ensure calls are in the specified order
)
@ -263,6 +271,114 @@ class TestDomainStatuses(MockEppLib):
super().tearDown()
class TestDomainAvailable(MockEppLib):
"""Test Domain.available"""
# No SetUp or tearDown necessary for these tests
def test_domain_available(self):
"""
Scenario: Testing whether an available domain is available
Should return True
Mock response to mimic EPP Response
Validate CheckDomain command is called
Validate response given mock
"""
def side_effect(_request, cleaned):
return MagicMock(
res_data=[
responses.check.CheckDomainResultData(
name="available.gov", avail=True, reason=None
)
],
)
patcher = patch("registrar.models.domain.registry.send")
mocked_send = patcher.start()
mocked_send.side_effect = side_effect
available = Domain.available("available.gov")
mocked_send.assert_has_calls(
[
call(
commands.CheckDomain(
["available.gov"],
),
cleaned=True,
)
]
)
self.assertTrue(available)
patcher.stop()
def test_domain_unavailable(self):
"""
Scenario: Testing whether an unavailable domain is available
Should return False
Mock response to mimic EPP Response
Validate CheckDomain command is called
Validate response given mock
"""
def side_effect(_request, cleaned):
return MagicMock(
res_data=[
responses.check.CheckDomainResultData(
name="unavailable.gov", avail=False, reason="In Use"
)
],
)
patcher = patch("registrar.models.domain.registry.send")
mocked_send = patcher.start()
mocked_send.side_effect = side_effect
available = Domain.available("unavailable.gov")
mocked_send.assert_has_calls(
[
call(
commands.CheckDomain(
["unavailable.gov"],
),
cleaned=True,
)
]
)
self.assertFalse(available)
patcher.stop()
def test_domain_available_with_value_error(self):
"""
Scenario: Testing whether an invalid domain is available
Should throw ValueError
Validate ValueError is raised
"""
with self.assertRaises(ValueError):
Domain.available("invalid-string")
def test_domain_available_unsuccessful(self):
"""
Scenario: Testing behavior when registry raises a RegistryError
Validate RegistryError is raised
"""
def side_effect(_request, cleaned):
raise RegistryError(code=ErrorCode.COMMAND_SYNTAX_ERROR)
patcher = patch("registrar.models.domain.registry.send")
mocked_send = patcher.start()
mocked_send.side_effect = side_effect
with self.assertRaises(RegistryError):
Domain.available("raises-error.gov")
patcher.stop()
class TestRegistrantContacts(MockEppLib):
"""Rule: Registrants may modify their WHOIS data"""
@ -664,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
And the registrant is the admin on a domain
Given the analyst is logged in
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 test_user_adds_dns_data(self):
def tearDown(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):
@ -940,7 +1384,7 @@ class TestAnalystLock(TestCase):
raise
class TestAnalystDelete(TestCase):
class TestAnalystDelete(MockEppLib):
"""Rule: Analysts may delete a domain"""
def setUp(self):
@ -949,35 +1393,99 @@ class TestAnalystDelete(TestCase):
Given the analyst is logged in
And a domain exists in the registry
"""
pass
super().setUp()
self.domain, _ = Domain.objects.get_or_create(
name="fake.gov", state=Domain.State.READY
)
self.domain_on_hold, _ = Domain.objects.get_or_create(
name="fake-on-hold.gov", state=Domain.State.ON_HOLD
)
def tearDown(self):
Domain.objects.all().delete()
super().tearDown()
@skip("not implemented yet")
def test_analyst_deletes_domain(self):
"""
Scenario: Analyst permanently deletes a domain
When `domain.delete()` is called
When `domain.deletedInEpp()` is called
Then `commands.DeleteDomain` is sent to the registry
And `state` is set to `DELETED`
"""
raise
# Put the domain in client hold
self.domain.place_client_hold()
# Delete it...
self.domain.deletedInEpp()
self.mockedSendFunction.assert_has_calls(
[
call(
commands.DeleteDomain(name="fake.gov"),
cleaned=True,
)
]
)
@skip("not implemented yet")
def test_analyst_deletes_domain_idempotent(self):
"""
Scenario: Analyst tries to delete an already deleted domain
Given `state` is already `DELETED`
When `domain.delete()` is called
Then `commands.DeleteDomain` is sent to the registry
And Domain returns normally (without error)
"""
raise
# Domain itself should not be deleted
self.assertNotEqual(self.domain, None)
# Domain should have the right state
self.assertEqual(self.domain.state, Domain.State.DELETED)
# Cache should be invalidated
self.assertEqual(self.domain._cache, {})
@skip("not implemented yet")
def test_deletion_is_unsuccessful(self):
"""
Scenario: Domain deletion is unsuccessful
When an error is returned from epplibwrapper
Then a user-friendly error message is returned for displaying on the web
When a subdomain exists
Then a client error is returned of code 2305
And `state` is not set to `DELETED`
"""
raise
# Desired domain
domain, _ = Domain.objects.get_or_create(
name="failDelete.gov", state=Domain.State.ON_HOLD
)
# Put the domain in client hold
domain.place_client_hold()
# Delete it
with self.assertRaises(RegistryError) as err:
domain.deletedInEpp()
self.assertTrue(
err.is_client_error()
and err.code == ErrorCode.OBJECT_ASSOCIATION_PROHIBITS_OPERATION
)
self.mockedSendFunction.assert_has_calls(
[
call(
commands.DeleteDomain(name="failDelete.gov"),
cleaned=True,
)
]
)
# Domain itself should not be deleted
self.assertNotEqual(domain, None)
# State should not have changed
self.assertEqual(domain.state, Domain.State.ON_HOLD)
def test_deletion_ready_fsm_failure(self):
"""
Scenario: Domain deletion is unsuccessful due to FSM rules
Given state is 'ready'
When `domain.deletedInEpp()` is called
and domain is of `state` is `READY`
Then an FSM error is returned
And `state` is not set to `DELETED`
"""
self.assertEqual(self.domain.state, Domain.State.READY)
with self.assertRaises(TransitionNotAllowed) as err:
self.domain.deletedInEpp()
self.assertTrue(
err.is_client_error()
and err.code == ErrorCode.OBJECT_STATUS_PROHIBITS_OPERATION
)
# Domain should not be deleted
self.assertNotEqual(self.domain, None)
# Domain should have the right state
self.assertEqual(self.domain.state, Domain.State.READY)

View file

@ -7,7 +7,7 @@ certifi==2023.7.22 ; python_version >= '3.6'
cfenv==0.5.3
cffi==1.15.1
charset-normalizer==3.1.0 ; python_full_version >= '3.7.0'
cryptography==41.0.3 ; python_version >= '3.7'
cryptography==41.0.4 ; python_version >= '3.7'
defusedxml==0.7.1 ; python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'
dj-database-url==2.0.0
dj-email-url==1.0.6
@ -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