mirror of
https://github.com/cisagov/manage.get.gov.git
synced 2025-07-22 18:56:15 +02:00
Merge branch 'main' into dk/1122-dnssec-rewrite
This commit is contained in:
David Kennedy
commit
096c2bec8d
32 changed files with 1396 additions and 472 deletions
|
|
@ -1,5 +1,5 @@
|
|||
from itertools import zip_longest
|
||||
import logging
|
||||
|
||||
from datetime import date
|
||||
from string import digits
|
||||
from django_fsm import FSMField, transition, TransitionNotAllowed # type: ignore
|
||||
|
|
@ -11,9 +11,11 @@ from epplibwrapper import (
|
|||
commands,
|
||||
common as epp,
|
||||
extensions,
|
||||
info as eppInfo,
|
||||
RegistryError,
|
||||
ErrorCode,
|
||||
)
|
||||
from registrar.models.utility.contact_error import ContactError
|
||||
|
||||
from .utility.domain_field import DomainField
|
||||
from .utility.domain_helper import DomainHelper
|
||||
|
|
@ -380,9 +382,9 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
raise NotImplementedError()
|
||||
|
||||
@Cache
|
||||
def registrant_contact(self) -> PublicContact:
|
||||
"""Get or set the registrant for this domain."""
|
||||
raise NotImplementedError()
|
||||
def registrant_contact(self) -> PublicContact | None:
|
||||
registrant = PublicContact.ContactTypeChoices.REGISTRANT
|
||||
return self.generic_contact_getter(registrant)
|
||||
|
||||
@registrant_contact.setter # type: ignore
|
||||
def registrant_contact(self, contact: PublicContact):
|
||||
|
|
@ -395,9 +397,10 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
)
|
||||
|
||||
@Cache
|
||||
def administrative_contact(self) -> PublicContact:
|
||||
"""Get or set the admin contact for this domain."""
|
||||
raise NotImplementedError()
|
||||
def administrative_contact(self) -> PublicContact | None:
|
||||
"""Get the admin contact for this domain."""
|
||||
admin = PublicContact.ContactTypeChoices.ADMINISTRATIVE
|
||||
return self.generic_contact_getter(admin)
|
||||
|
||||
@administrative_contact.setter # type: ignore
|
||||
def administrative_contact(self, contact: PublicContact):
|
||||
|
|
@ -409,12 +412,6 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
self._make_contact_in_registry(contact=contact)
|
||||
self._update_domain_with_contact(contact, rem=False)
|
||||
|
||||
def get_default_security_contact(self):
|
||||
logger.info("getting default sec contact")
|
||||
contact = PublicContact.get_default_security()
|
||||
contact.domain = self
|
||||
return contact
|
||||
|
||||
def _update_epp_contact(self, contact: PublicContact):
|
||||
"""Sends UpdateContact to update the actual contact object,
|
||||
domain object remains unaffected
|
||||
|
|
@ -468,26 +465,10 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
)
|
||||
|
||||
@Cache
|
||||
def security_contact(self) -> PublicContact:
|
||||
def security_contact(self) -> PublicContact | None:
|
||||
"""Get or set the security contact for this domain."""
|
||||
try:
|
||||
contacts = self._get_property("contacts")
|
||||
for contact in contacts:
|
||||
if (
|
||||
"type" in contact.keys()
|
||||
and contact["type"] == PublicContact.ContactTypeChoices.SECURITY
|
||||
):
|
||||
tempContact = self.get_default_security_contact()
|
||||
tempContact.email = contact["email"]
|
||||
return tempContact
|
||||
|
||||
except Exception as err: # use better error handling
|
||||
logger.info("Couldn't get contact %s" % err)
|
||||
|
||||
# TODO - remove this ideally it should return None,
|
||||
# but error handling needs to be
|
||||
# added on the security email page so that it can handle it being none
|
||||
return self.get_default_security_contact()
|
||||
security = PublicContact.ContactTypeChoices.SECURITY
|
||||
return self.generic_contact_getter(security)
|
||||
|
||||
def _add_registrant_to_existing_domain(self, contact: PublicContact):
|
||||
"""Used to change the registrant contact on an existing domain"""
|
||||
|
|
@ -561,6 +542,7 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
.filter(domain=self, contact_type=contact.contact_type)
|
||||
.get()
|
||||
)
|
||||
|
||||
if isRegistrant:
|
||||
# send update domain only for registant contacts
|
||||
existing_contact.delete()
|
||||
|
|
@ -617,9 +599,10 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
)
|
||||
|
||||
@Cache
|
||||
def technical_contact(self) -> PublicContact:
|
||||
def technical_contact(self) -> PublicContact | None:
|
||||
"""Get or set the tech contact for this domain."""
|
||||
raise NotImplementedError()
|
||||
tech = PublicContact.ContactTypeChoices.TECHNICAL
|
||||
return self.generic_contact_getter(tech)
|
||||
|
||||
@technical_contact.setter # type: ignore
|
||||
def technical_contact(self, contact: PublicContact):
|
||||
|
|
@ -702,20 +685,230 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
help_text="Very basic info about the lifecycle of this domain object",
|
||||
)
|
||||
|
||||
dnssec_enabled = models.BooleanField(
|
||||
default=False,
|
||||
help_text="Boolean indicating if dnssec is enabled",
|
||||
)
|
||||
|
||||
dnssec_ds_confirmed = models.BooleanField(
|
||||
default=False,
|
||||
help_text="Boolean indicating if DS record adding is confirmed",
|
||||
)
|
||||
|
||||
dnssec_key_confirmed = models.BooleanField(
|
||||
default=False,
|
||||
help_text="Boolean indicating if Key record adding is confirmed",
|
||||
)
|
||||
def isActive(self):
|
||||
return self.state == Domain.State.CREATED
|
||||
|
||||
def map_epp_contact_to_public_contact(
|
||||
self, contact: eppInfo.InfoContactResultData, contact_id, contact_type
|
||||
):
|
||||
"""Maps the Epp contact representation to a PublicContact object.
|
||||
|
||||
contact -> eppInfo.InfoContactResultData: The converted contact object
|
||||
|
||||
contact_id -> str: The given registry_id of the object (i.e "cheese@cia.gov")
|
||||
|
||||
contact_type -> str: The given contact type, (i.e. "tech" or "registrant")
|
||||
"""
|
||||
|
||||
if contact is None:
|
||||
return None
|
||||
|
||||
if contact_type is None:
|
||||
raise ContactError("contact_type is None")
|
||||
|
||||
if contact_id is None:
|
||||
raise ContactError("contact_id is None")
|
||||
|
||||
# Since contact_id is registry_id,
|
||||
# check that its the right length
|
||||
contact_id_length = len(contact_id)
|
||||
if (
|
||||
contact_id_length > PublicContact.get_max_id_length()
|
||||
or contact_id_length < 1
|
||||
):
|
||||
raise ContactError(
|
||||
"contact_id is of invalid length. "
|
||||
"Cannot exceed 16 characters, "
|
||||
f"got {contact_id} with a length of {contact_id_length}"
|
||||
)
|
||||
|
||||
if not isinstance(contact, eppInfo.InfoContactResultData):
|
||||
raise ContactError("Contact must be of type InfoContactResultData")
|
||||
|
||||
auth_info = contact.auth_info
|
||||
postal_info = contact.postal_info
|
||||
addr = postal_info.addr
|
||||
streets = None
|
||||
if addr is not None:
|
||||
streets = addr.street
|
||||
streets_kwargs = self._convert_streets_to_dict(streets)
|
||||
desired_contact = PublicContact(
|
||||
domain=self,
|
||||
contact_type=contact_type,
|
||||
registry_id=contact_id,
|
||||
email=contact.email or "",
|
||||
voice=contact.voice or "",
|
||||
fax=contact.fax,
|
||||
name=postal_info.name or "",
|
||||
org=postal_info.org,
|
||||
# For linter - default to "" instead of None
|
||||
pw=getattr(auth_info, "pw", ""),
|
||||
city=getattr(addr, "city", ""),
|
||||
pc=getattr(addr, "pc", ""),
|
||||
cc=getattr(addr, "cc", ""),
|
||||
sp=getattr(addr, "sp", ""),
|
||||
**streets_kwargs,
|
||||
) # type: ignore
|
||||
|
||||
return desired_contact
|
||||
|
||||
def _convert_streets_to_dict(self, streets):
|
||||
"""
|
||||
Converts EPPLibs street representation
|
||||
to PublicContacts.
|
||||
|
||||
Args:
|
||||
streets (Sequence[str]): Streets from EPPLib.
|
||||
|
||||
Returns:
|
||||
dict: {
|
||||
"street1": str or "",
|
||||
|
||||
"street2": str or None,
|
||||
|
||||
"street3": str or None,
|
||||
}
|
||||
|
||||
EPPLib returns 'street' as an sequence of strings.
|
||||
Meanwhile, PublicContact has this split into three
|
||||
seperate properties: street1, street2, street3.
|
||||
|
||||
Handles this disparity.
|
||||
"""
|
||||
# 'zips' two lists together.
|
||||
# For instance, (('street1', 'some_value_here'),
|
||||
# ('street2', 'some_value_here'))
|
||||
# Dict then converts this to a useable kwarg which we can pass in
|
||||
street_dict = dict(
|
||||
zip_longest(
|
||||
["street1", "street2", "street3"],
|
||||
streets if streets is not None else [""],
|
||||
fillvalue=None,
|
||||
)
|
||||
)
|
||||
return street_dict
|
||||
|
||||
def _request_contact_info(self, contact: PublicContact):
|
||||
try:
|
||||
req = commands.InfoContact(id=contact.registry_id)
|
||||
return registry.send(req, cleaned=True).res_data[0]
|
||||
except RegistryError as error:
|
||||
logger.error(
|
||||
"Registry threw error for contact id %s contact type is %s, error code is\n %s full error is %s", # noqa
|
||||
contact.registry_id,
|
||||
contact.contact_type,
|
||||
error.code,
|
||||
error,
|
||||
)
|
||||
raise error
|
||||
|
||||
def generic_contact_getter(
|
||||
self, contact_type_choice: PublicContact.ContactTypeChoices
|
||||
) -> PublicContact | None:
|
||||
"""Retrieves the desired PublicContact from the registry.
|
||||
This abstracts the caching and EPP retrieval for
|
||||
all contact items and thus may result in EPP calls being sent.
|
||||
|
||||
contact_type_choice is a literal in PublicContact.ContactTypeChoices,
|
||||
for instance: PublicContact.ContactTypeChoices.SECURITY.
|
||||
|
||||
If you wanted to setup getter logic for Security, you would call:
|
||||
cache_contact_helper(PublicContact.ContactTypeChoices.SECURITY),
|
||||
or cache_contact_helper("security").
|
||||
|
||||
"""
|
||||
# registrant_contact(s) are an edge case. They exist on
|
||||
# the "registrant" property as opposed to contacts.
|
||||
desired_property = "contacts"
|
||||
if contact_type_choice == PublicContact.ContactTypeChoices.REGISTRANT:
|
||||
desired_property = "registrant"
|
||||
|
||||
try:
|
||||
# Grab from cache
|
||||
contacts = self._get_property(desired_property)
|
||||
except KeyError as error:
|
||||
logger.error(f"Could not find {contact_type_choice}: {error}")
|
||||
return None
|
||||
else:
|
||||
cached_contact = self.get_contact_in_keys(contacts, contact_type_choice)
|
||||
if cached_contact is None:
|
||||
# TODO - #1103
|
||||
raise ContactError("No contact was found in cache or the registry")
|
||||
|
||||
return cached_contact
|
||||
|
||||
def get_default_security_contact(self):
|
||||
"""Gets the default security contact."""
|
||||
contact = PublicContact.get_default_security()
|
||||
contact.domain = self
|
||||
return contact
|
||||
|
||||
def get_default_administrative_contact(self):
|
||||
"""Gets the default administrative contact."""
|
||||
contact = PublicContact.get_default_administrative()
|
||||
contact.domain = self
|
||||
return contact
|
||||
|
||||
def get_default_technical_contact(self):
|
||||
"""Gets the default technical contact."""
|
||||
contact = PublicContact.get_default_technical()
|
||||
contact.domain = self
|
||||
return contact
|
||||
|
||||
def get_default_registrant_contact(self):
|
||||
"""Gets the default registrant contact."""
|
||||
contact = PublicContact.get_default_registrant()
|
||||
contact.domain = self
|
||||
return contact
|
||||
|
||||
def get_contact_in_keys(self, contacts, contact_type):
|
||||
"""Gets a contact object.
|
||||
|
||||
Args:
|
||||
contacts ([PublicContact]): List of PublicContacts
|
||||
contact_type (literal): Which PublicContact to get
|
||||
Returns:
|
||||
PublicContact | None
|
||||
"""
|
||||
# Registrant doesn't exist as an array, and is of
|
||||
# a special data type, so we need to handle that.
|
||||
if contact_type == PublicContact.ContactTypeChoices.REGISTRANT:
|
||||
desired_contact = None
|
||||
if isinstance(contacts, str):
|
||||
desired_contact = self._registrant_to_public_contact(
|
||||
self._cache["registrant"]
|
||||
)
|
||||
# Set the cache with the updated object
|
||||
# for performance reasons.
|
||||
if "registrant" in self._cache:
|
||||
self._cache["registrant"] = desired_contact
|
||||
elif isinstance(contacts, PublicContact):
|
||||
desired_contact = contacts
|
||||
|
||||
return self._handle_registrant_contact(desired_contact)
|
||||
|
||||
_registry_id: str
|
||||
if contact_type in contacts:
|
||||
_registry_id = contacts.get(contact_type)
|
||||
|
||||
desired = PublicContact.objects.filter(
|
||||
registry_id=_registry_id, domain=self, contact_type=contact_type
|
||||
)
|
||||
|
||||
if desired.count() == 1:
|
||||
return desired.get()
|
||||
|
||||
logger.info(f"Requested contact {_registry_id} does not exist in cache.")
|
||||
return None
|
||||
|
||||
def _handle_registrant_contact(self, contact):
|
||||
if (
|
||||
contact.contact_type is not None
|
||||
and contact.contact_type == PublicContact.ContactTypeChoices.REGISTRANT
|
||||
):
|
||||
return contact
|
||||
else:
|
||||
raise ValueError("Invalid contact object for registrant_contact")
|
||||
|
||||
# ForeignKey on UserDomainRole creates a "permissions" member for
|
||||
# all of the user-roles that are in place for this domain
|
||||
|
|
@ -815,12 +1008,10 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
security_contact = self.get_default_security_contact()
|
||||
security_contact.save()
|
||||
|
||||
technical_contact = PublicContact.get_default_technical()
|
||||
technical_contact.domain = self
|
||||
technical_contact = self.get_default_technical_contact()
|
||||
technical_contact.save()
|
||||
|
||||
administrative_contact = PublicContact.get_default_administrative()
|
||||
administrative_contact.domain = self
|
||||
administrative_contact = self.get_default_administrative_contact()
|
||||
administrative_contact.save()
|
||||
|
||||
@transition(
|
||||
|
|
@ -964,16 +1155,34 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
)
|
||||
return err.code
|
||||
|
||||
def _request_contact_info(self, contact: PublicContact):
|
||||
req = commands.InfoContact(id=contact.registry_id)
|
||||
return registry.send(req, cleaned=True).res_data[0]
|
||||
def _fetch_contacts(self, contact_data):
|
||||
"""Fetch contact info."""
|
||||
choices = PublicContact.ContactTypeChoices
|
||||
# We expect that all these fields get populated,
|
||||
# so we can create these early, rather than waiting.
|
||||
contacts_dict = {
|
||||
choices.ADMINISTRATIVE: None,
|
||||
choices.SECURITY: None,
|
||||
choices.TECHNICAL: None,
|
||||
}
|
||||
for domainContact in contact_data:
|
||||
req = commands.InfoContact(id=domainContact.contact)
|
||||
data = registry.send(req, cleaned=True).res_data[0]
|
||||
|
||||
# Map the object we recieved from EPP to a PublicContact
|
||||
mapped_object = self.map_epp_contact_to_public_contact(
|
||||
data, domainContact.contact, domainContact.type
|
||||
)
|
||||
|
||||
# Find/create it in the DB
|
||||
in_db = self._get_or_create_public_contact(mapped_object)
|
||||
contacts_dict[in_db.contact_type] = in_db.registry_id
|
||||
return contacts_dict
|
||||
|
||||
def _get_or_create_contact(self, contact: PublicContact):
|
||||
"""Try to fetch info about a contact. Create it if it does not exist."""
|
||||
|
||||
try:
|
||||
return self._request_contact_info(contact)
|
||||
|
||||
except RegistryError as e:
|
||||
if e.code == ErrorCode.OBJECT_DOES_NOT_EXIST:
|
||||
logger.info(
|
||||
|
|
@ -996,6 +1205,23 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
|
||||
raise e
|
||||
|
||||
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]
|
||||
host = {
|
||||
"name": name,
|
||||
"addrs": getattr(data, "addrs", ...),
|
||||
"cr_date": getattr(data, "cr_date", ...),
|
||||
"statuses": getattr(data, "statuses", ...),
|
||||
"tr_date": getattr(data, "tr_date", ...),
|
||||
"up_date": getattr(data, "up_date", ...),
|
||||
}
|
||||
hosts.append({k: v for k, v in host.items() if v is not ...})
|
||||
return hosts
|
||||
|
||||
def _update_or_create_host(self, host):
|
||||
raise NotImplementedError()
|
||||
|
||||
|
|
@ -1022,12 +1248,11 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
"tr_date": getattr(data, "tr_date", ...),
|
||||
"up_date": getattr(data, "up_date", ...),
|
||||
}
|
||||
|
||||
# remove null properties (to distinguish between "a value of None" and null)
|
||||
cleaned = {k: v for k, v in cache.items() if v is not ...}
|
||||
|
||||
# statuses can just be a list no need to keep the epp object
|
||||
if "statuses" in cleaned.keys():
|
||||
if "statuses" in cleaned:
|
||||
cleaned["statuses"] = [status.state for status in cleaned["statuses"]]
|
||||
|
||||
# get extensions info, if there is any
|
||||
|
|
@ -1047,7 +1272,7 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
fetch_contacts
|
||||
and "_contacts" in cleaned
|
||||
and isinstance(cleaned["_contacts"], list)
|
||||
and len(cleaned["_contacts"])
|
||||
and len(cleaned["_contacts"]) > 0
|
||||
):
|
||||
cleaned["contacts"] = self._fetch_contacts(cleaned["_contacts"])
|
||||
# We're only getting contacts, so retain the old
|
||||
|
|
@ -1069,52 +1294,69 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
# 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]
|
||||
contact = {
|
||||
"id": domainContact.contact,
|
||||
"type": domainContact.type,
|
||||
"auth_info": getattr(data, "auth_info", ...),
|
||||
"cr_date": getattr(data, "cr_date", ...),
|
||||
"disclose": getattr(data, "disclose", ...),
|
||||
"email": getattr(data, "email", ...),
|
||||
"fax": getattr(data, "fax", ...),
|
||||
"postal_info": getattr(data, "postal_info", ...),
|
||||
"statuses": getattr(data, "statuses", ...),
|
||||
"tr_date": getattr(data, "tr_date", ...),
|
||||
"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
|
||||
def _get_or_create_public_contact(self, public_contact: PublicContact):
|
||||
"""Tries to find a PublicContact object in our DB.
|
||||
If it can't, it'll create it. Returns PublicContact"""
|
||||
db_contact = PublicContact.objects.filter(
|
||||
registry_id=public_contact.registry_id,
|
||||
contact_type=public_contact.contact_type,
|
||||
domain=self,
|
||||
)
|
||||
|
||||
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]
|
||||
host = {
|
||||
"name": name,
|
||||
"addrs": getattr(data, "addrs", ...),
|
||||
"cr_date": getattr(data, "cr_date", ...),
|
||||
"statuses": getattr(data, "statuses", ...),
|
||||
"tr_date": getattr(data, "tr_date", ...),
|
||||
"up_date": getattr(data, "up_date", ...),
|
||||
}
|
||||
hosts.append({k: v for k, v in host.items() if v is not ...})
|
||||
return hosts
|
||||
# Raise an error if we find duplicates.
|
||||
# This should not occur
|
||||
if db_contact.count() > 1:
|
||||
raise Exception(
|
||||
f"Multiple contacts found for {public_contact.contact_type}"
|
||||
)
|
||||
|
||||
# Save to DB if it doesn't exist already.
|
||||
if db_contact.count() == 0:
|
||||
# Doesn't run custom save logic, just saves to DB
|
||||
public_contact.save(skip_epp_save=True)
|
||||
logger.info(f"Created a new PublicContact: {public_contact}")
|
||||
# Append the item we just created
|
||||
return public_contact
|
||||
|
||||
existing_contact = db_contact.get()
|
||||
|
||||
# Does the item we're grabbing match
|
||||
# what we have in our DB?
|
||||
if (
|
||||
existing_contact.email != public_contact.email
|
||||
or existing_contact.registry_id != public_contact.registry_id
|
||||
):
|
||||
existing_contact.delete()
|
||||
public_contact.save()
|
||||
logger.warning("Requested PublicContact is out of sync " "with DB.")
|
||||
return public_contact
|
||||
# If it already exists, we can
|
||||
# assume that the DB instance was updated
|
||||
# during set, so we should just use that.
|
||||
return existing_contact
|
||||
|
||||
def _registrant_to_public_contact(self, registry_id: str):
|
||||
"""EPPLib returns the registrant as a string,
|
||||
which is the registrants associated registry_id. This function is used to
|
||||
convert that id to a useable object by calling commands.InfoContact
|
||||
on that ID, then mapping that object to type PublicContact."""
|
||||
contact = PublicContact(
|
||||
registry_id=registry_id,
|
||||
contact_type=PublicContact.ContactTypeChoices.REGISTRANT,
|
||||
)
|
||||
# Grabs the expanded contact
|
||||
full_object = self._request_contact_info(contact)
|
||||
# Maps it to type PublicContact
|
||||
mapped_object = self.map_epp_contact_to_public_contact(
|
||||
full_object, contact.registry_id, contact.contact_type
|
||||
)
|
||||
return self._get_or_create_public_contact(mapped_object)
|
||||
|
||||
def _invalidate_cache(self):
|
||||
"""Remove cache data when updates are made."""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue