Merge branch 'main' into dk/1359-security-email

This commit is contained in:
David Kennedy 2023-12-29 17:09:04 -05:00
commit af4e821d21
No known key found for this signature in database
GPG key ID: 6528A5386E66B96B
36 changed files with 1606 additions and 654 deletions

View file

@ -203,7 +203,7 @@ class Domain(TimeStampedModel, DomainHelper):
return self._get_property("cr_date")
@creation_date.setter # type: ignore
def creation_date(self, ex_date: date):
def creation_date(self, cr_date: date):
"""
Direct setting of the creation date in the registry is not implemented.
@ -733,8 +733,10 @@ class Domain(TimeStampedModel, DomainHelper):
email=contact.email,
voice=contact.voice,
fax=contact.fax,
auth_info=epp.ContactAuthInfo(pw="2fooBAR123fooBaz"),
) # type: ignore
updateContact.disclose = self._disclose_fields(contact=contact) # type: ignore
try:
registry.send(updateContact, cleaned=True)
except RegistryError as e:
@ -1620,7 +1622,8 @@ class Domain(TimeStampedModel, DomainHelper):
cache = self._extract_data_from_response(data_response)
cleaned = self._clean_cache(cache, data_response)
self._update_hosts_and_contacts(cleaned, fetch_hosts, fetch_contacts)
self._update_hosts_and_ips_in_db(cleaned, fetch_hosts)
if fetch_hosts:
self._update_hosts_and_ips_in_db(cleaned)
self._update_security_contact_in_db(cleaned, fetch_contacts)
self._update_dates(cleaned)
@ -1668,7 +1671,11 @@ class Domain(TimeStampedModel, DomainHelper):
return dnssec_data
def _update_hosts_and_contacts(self, cleaned, fetch_hosts, fetch_contacts):
"""Capture and cache old hosts and contacts from cache if they don't exist in cleaned"""
"""
Update hosts and contacts if fetch_hosts and/or fetch_contacts.
Additionally, capture and cache old hosts and contacts from cache if they
don't exist in cleaned
"""
old_cache_hosts = self._cache.get("hosts")
old_cache_contacts = self._cache.get("contacts")
@ -1683,50 +1690,49 @@ class Domain(TimeStampedModel, DomainHelper):
if old_cache_contacts is not None:
cleaned["contacts"] = old_cache_contacts
def _update_hosts_and_ips_in_db(self, cleaned, fetch_hosts):
def _update_hosts_and_ips_in_db(self, cleaned):
"""Update hosts and host_ips in database if retrieved from registry.
Only called when fetch_hosts is True.
Parameters:
self: the domain to be updated with hosts and ips from cleaned
cleaned: dict containing hosts. Hosts are provided as a list of dicts, e.g.
[{"name": "ns1.example.com",}, {"name": "ns1.example.gov"}, "addrs": ["0.0.0.0"])]
fetch_hosts: boolean indicating whether or not fetch_hosts was called
"""
if fetch_hosts:
cleaned_hosts = cleaned["hosts"]
# Get all existing hosts from the database for this domain
existing_hosts_in_db = Host.objects.filter(domain=self)
# Identify hosts to delete
cleaned_host_names = set(cleaned_host["name"] for cleaned_host in cleaned_hosts)
hosts_to_delete_from_db = [
existing_host for existing_host in existing_hosts_in_db if existing_host.name not in cleaned_host_names
]
# Delete hosts and their associated HostIP instances
for host_to_delete in hosts_to_delete_from_db:
# Delete associated HostIP instances
HostIP.objects.filter(host=host_to_delete).delete()
# Delete the host itself
host_to_delete.delete()
# Update or create Hosts and HostIPs
for cleaned_host in cleaned_hosts:
# Check if the cleaned_host already exists
host_in_db, host_created = Host.objects.get_or_create(domain=self, name=cleaned_host["name"])
# Get cleaned list of ips for update
cleaned_ips = cleaned_host["addrs"]
if not host_created:
# Get all existing ips from the database for this host
existing_ips_in_db = HostIP.objects.filter(host=host_in_db)
# Identify IPs to delete
ips_to_delete_from_db = [
existing_ip for existing_ip in existing_ips_in_db if existing_ip.address not in cleaned_ips
]
# Delete IPs
for ip_to_delete in ips_to_delete_from_db:
# Delete the ip
ip_to_delete.delete()
# Update or create HostIP instances
for ip_address in cleaned_ips:
HostIP.objects.get_or_create(address=ip_address, host=host_in_db)
cleaned_hosts = cleaned["hosts"]
# Get all existing hosts from the database for this domain
existing_hosts_in_db = Host.objects.filter(domain=self)
# Identify hosts to delete
cleaned_host_names = set(cleaned_host["name"] for cleaned_host in cleaned_hosts)
hosts_to_delete_from_db = [
existing_host for existing_host in existing_hosts_in_db if existing_host.name not in cleaned_host_names
]
# Delete hosts and their associated HostIP instances
for host_to_delete in hosts_to_delete_from_db:
# Delete associated HostIP instances
HostIP.objects.filter(host=host_to_delete).delete()
# Delete the host itself
host_to_delete.delete()
# Update or create Hosts and HostIPs
for cleaned_host in cleaned_hosts:
# Check if the cleaned_host already exists
host_in_db, host_created = Host.objects.get_or_create(domain=self, name=cleaned_host["name"])
# Get cleaned list of ips for update
cleaned_ips = cleaned_host["addrs"]
if not host_created:
# Get all existing ips from the database for this host
existing_ips_in_db = HostIP.objects.filter(host=host_in_db)
# Identify IPs to delete
ips_to_delete_from_db = [
existing_ip for existing_ip in existing_ips_in_db if existing_ip.address not in cleaned_ips
]
# Delete IPs
for ip_to_delete in ips_to_delete_from_db:
# Delete the ip
ip_to_delete.delete()
# Update or create HostIP instances
for ip_address in cleaned_ips:
HostIP.objects.get_or_create(address=ip_address, host=host_in_db)
def _update_security_contact_in_db(self, cleaned, fetch_contacts):
"""Update security contact registry id in database if retrieved from registry.