zydronium/manage.get.gov
1
0
Fork 0
mirror of https://github.com/cisagov/manage.get.gov.git synced 2025-05-16 17:47:02 +02:00
Code Issues Projects Releases Packages Wiki Activity

ran black formatting

Browse source
This commit is contained in:
Alysia Broddrick 2023-09-08 19:07:59 -07:00
parent c9cadd3401
commit e71b5b0bd4
No known key found for this signature in database
GPG key ID: 03917052CD0F06B7
7 changed files with 559 additions and 358 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/epplibwrapper/client.py
View file

@ -83,7 +83,7 @@ class EPPLibWrapper:
logger.warning(message, cmd_type, exc_info=True)
raise RegistryError(message) from err
except Exception as err:
message = '%s failed to execute due to an unknown error.' % err
message = "%s failed to execute due to an unknown error." % err
logger.warning(message, cmd_type, exc_info=True)
raise RegistryError(message) from err
else:

44
src/registrar/admin.py
View file

@ -147,9 +147,9 @@ class DomainAdmin(ListHeaderAdmin):
def response_change(self, request, obj):
print(request.POST)
ACTION_BUTTON = "_place_client_hold"
GET_SECURITY_EMAIL="_get_security_email"
SET_SECURITY_CONTACT="_set_security_contact"
MAKE_DOMAIN="_make_domain_in_registry"
GET_SECURITY_EMAIL = "_get_security_email"
SET_SECURITY_CONTACT = "_set_security_contact"
MAKE_DOMAIN = "_make_domain_in_registry"
logger.info("in response")
if ACTION_BUTTON in request.POST:
logger.info("in action button")
@ -171,37 +171,29 @@ class DomainAdmin(ListHeaderAdmin):
if GET_SECURITY_EMAIL in request.POST:
try:
security_email=obj.get_security_email()
security_email = obj.get_security_email()
except Exception as err:
self.message_user(request, err, messages.ERROR)
else:
self.message_user(request,
(
"The security email is %"
". Thanks!"
)
% security_email,
self.message_user(
request,
("The security email is %" ". Thanks!") % security_email,
)
return HttpResponseRedirect(".")
if SET_SECURITY_CONTACT in request.POST:
try:
security_contact = obj.get_default_security_contact()
security_contact.email="ab@test.gov"
security_contact = obj.get_default_security_contact()
security_contact.email = "ab@test.gov"
obj.security_contact=security_contact
obj.security_contact = security_contact
except Exception as err:
self.message_user(request, err, messages.ERROR)
else:
self.message_user(request,
(
"The security email is %"
". Thanks!"
)
% security_email,
self.message_user(
request,
("The security email is %" ". Thanks!") % security_email,
)
print("above make domain")
@ -213,15 +205,13 @@ class DomainAdmin(ListHeaderAdmin):
except Exception as err:
self.message_user(request, err, messages.ERROR)
else:
self.message_user(request,
(
"Domain created with %"
". Thanks!"
)
% obj.name,
self.message_user(
request,
("Domain created with %" ". Thanks!") % obj.name,
)
return HttpResponseRedirect(".")
return super().response_change(request, obj)
# def response_change(self, request, obj):
# ACTION_BUTTON = "_get_security_email"

541
src/registrar/models/domain.py
View file

@ -19,6 +19,7 @@ from .utility.domain_helper import DomainHelper
from .utility.time_stamped_model import TimeStampedModel
from .public_contact import PublicContact
logger = logging.getLogger(__name__)
@ -103,25 +104,24 @@ class Domain(TimeStampedModel, DomainHelper):
class State(models.TextChoices):
"""These capture (some of) the states a domain object can be in."""
# the state is indeterminate
UNKNOWN = "unknown"
#The domain object exists in the registry but nameservers don't exist for it yet
PENDING_CREATE="pending create"
# The domain object exists in the registry but nameservers don't exist for it yet
PENDING_CREATE = "pending create"
# Domain has had nameservers set, may or may not be active
CREATED = "created"
#Registrar manually changed state to client hold
CLIENT_HOLD ="client hold"
# Registrar manually changed state to client hold
CLIENT_HOLD = "client hold"
#Registry
# Registry
SERVER_HOLD = "server hold"
# previously existed but has been deleted from the registry
DELETED = "deleted"
class Cache(property):
"""
Python descriptor to turn class methods into properties.
@ -228,24 +228,30 @@ class Domain(TimeStampedModel, DomainHelper):
while non-subordinate hosts MUST NOT.
"""
# TODO: call EPP to get this info instead of returning fake data.
#MISSING FROM DISPLAY
# MISSING FROM DISPLAY
return [
("ns1.example.com",),
("ns2.example.com",),
("ns3.example.com",),
]
def _check_host(self,hostnames:list[str]):
""" check if host is available, True if available
def _check_host(self, hostnames: list[str]):
"""check if host is available, True if available
returns boolean"""
checkCommand=commands.CheckHost(hostnames)
checkCommand = commands.CheckHost(hostnames)
try:
response=registry.send(checkCommand,cleaned=True)
response = registry.send(checkCommand, cleaned=True)
return response.res_data[0].avail
except RegistryError as err:
logger.warning("Couldn't check hosts %. Errorcode was %s, error was %s"%(hostnames),err.code, err)
logger.warning(
"Couldn't check hosts %. Errorcode was %s, error was %s" % (hostnames),
err.code,
err,
)
return False
def _create_host(self, host,addrs):
def _create_host(self, host, addrs):
"""Call _check_host first before using this function,
This creates the host object in the registry
doesn't add the created host to the domain
@ -253,17 +259,17 @@ class Domain(TimeStampedModel, DomainHelper):
logger.info("_create_host()->addresses is NONE")
if not addrs is None:
logger.info("addresses is not None %s"%addrs)
addresses=[epp.Ip(addr=addr) for addr in addrs]
logger.info("addresses is not None %s" % addrs)
addresses = [epp.Ip(addr=addr) for addr in addrs]
request = commands.CreateHost(name=host, addrs=addresses)
else:
logger.info("_create_host()-> address IS None")
request = commands.CreateHost(name=host)
#[epp.Ip(addr="127.0.0.1"), epp.Ip(addr="0:0:0:0:0:0:0:1", ip="v6")]
# [epp.Ip(addr="127.0.0.1"), epp.Ip(addr="0:0:0:0:0:0:0:1", ip="v6")]
try:
logger.info("_create_host()-> sending req as %s"%request)
response=registry.send(request, cleaned=True)
logger.info("_create_host()-> sending req as %s" % request)
response = registry.send(request, cleaned=True)
return response.code
except RegistryError as e:
logger.error("Error _create_host, code was %s error was %s" % (e.code, e))
@ -276,35 +282,43 @@ class Domain(TimeStampedModel, DomainHelper):
example: [(ns1.okay.gov, 127.0.0.1, others ips)]"""
# TODO: call EPP to set this info.
# if two nameservers change state to created, don't do it automatically
hostSuccessCount=0
if len(hosts)>13:
raise ValueError("Too many hosts provided, you may not have more than 13 nameservers.")
hostSuccessCount = 0
if len(hosts) > 13:
raise ValueError(
"Too many hosts provided, you may not have more than 13 nameservers."
)
logger.info("hosts will follow")
logger.info(hosts)
for hostTuple in hosts:
print("hostTuple is %s"% str(hostTuple))
host=hostTuple[0]
addrs=None
if len(hostTuple)>1:
addrs=hostTuple[1:]
avail=self._check_host([host])
print("hostTuple is %s" % str(hostTuple))
host = hostTuple[0]
addrs = None
if len(hostTuple) > 1:
addrs = hostTuple[1:]
avail = self._check_host([host])
if avail:
createdCode=self._create_host(host=host, addrs=addrs)
if createdCode==ErrorCode.OBJECT_EXISTS:
hostSuccessCount+=1
#update the object instead
elif createdCode==ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY:
#add host to domain
request = commands.UpdateDomain(name=self.name, add=[epp.HostObjSet([host])])
createdCode = self._create_host(host=host, addrs=addrs)
if createdCode == ErrorCode.OBJECT_EXISTS:
hostSuccessCount += 1
# update the object instead
elif createdCode == ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY:
# add host to domain
request = commands.UpdateDomain(
name=self.name, add=[epp.HostObjSet([host])]
)
try:
registry.send(request, cleaned=True)
hostSuccessCount+=1
hostSuccessCount += 1
except RegistryError as e:
logger.error("Error adding nameserver, code was %s error was %s" % (e.code, e))
logger.error(
"Error adding nameserver, code was %s error was %s"
% (e.code, e)
)
if self.state==self.State.PENDING_CREATE and hostSuccessCount>=2:
if self.state == self.State.PENDING_CREATE and hostSuccessCount >= 2:
self.created()
self.save()
##TODO - handle removed nameservers here will need to change the state go back to pending_create
@Cache
@ -318,7 +332,7 @@ class Domain(TimeStampedModel, DomainHelper):
# a dataclass property `state`, not to be confused with the `state` field here
if not "statuses" in self._cache:
self._fetch_cache()
if not "statuses"in self._cache:
if not "statuses" in self._cache:
raise Exception("Can't retreive status from domain info")
else:
return self._cache["statuses"]
@ -329,20 +343,21 @@ class Domain(TimeStampedModel, DomainHelper):
# can be combined; check that here and raise errors for invalid combinations -
# some statuses cannot be set by the client at all
raise NotImplementedError()
# ### implement get status which checks the status of the domain object on error it logs but goes with whatever the status is
# def get_status(self):
# try:
# DomainInfoReq
# response=send
# response.statuses
# for status in status:
# if status==serverhold and self.state!=serverhld
# transition to serverhold
# if status ==client & self.state!=clientHold:
# transition to clienthold
# except:
# logger
# return self.state
# ### implement get status which checks the status of the domain object on error it logs but goes with whatever the status is
# def get_status(self):
# try:
# DomainInfoReq
# response=send
# response.statuses
# for status in status:
# if status==serverhold and self.state!=serverhld
# transition to serverhold
# if status ==client & self.state!=clientHold:
# transition to clienthold
# except:
# logger
# return self.state
@Cache
def registrant_contact(self) -> PublicContact:
"""Get or set the registrant for this domain."""
@ -353,8 +368,9 @@ class Domain(TimeStampedModel, DomainHelper):
"""Registrant is set when a domain is created, so follow on additions will update the current registrant"""
###incorrect should update an existing registrant
logger.info("making registrant contact")
self._set_singleton_contact(contact=contact, expectedType=contact.ContactTypeChoices.REGISTRANT)
self._set_singleton_contact(
contact=contact, expectedType=contact.ContactTypeChoices.REGISTRANT
)
@Cache
def administrative_contact(self) -> PublicContact:
@ -368,75 +384,88 @@ class Domain(TimeStampedModel, DomainHelper):
# type options are[admin, billing, tech, security]
# use admin as type parameter for this contact
logger.info("making admin contact")
if contact.contact_type!=contact.ContactTypeChoices.ADMINISTRATIVE:
raise ValueError("Cannot set a registrant contact with a different contact type")
if contact.contact_type != contact.ContactTypeChoices.ADMINISTRATIVE:
raise ValueError(
"Cannot set a registrant contact with a different contact type"
)
logger.info("administrative_contact()-> update domain with admin contact")
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):
def _update_epp_contact(self, contact: PublicContact):
"""Sends UpdateContact to update the actual contact object, domain object remains unaffected
should be used when changing email address or other contact infor on an existing domain"""
updateContact=commands.UpdateContact(id=contact.registry_id, postal_info=self._make_epp_contact_postal_info(contact=contact),
should be used when changing email address or other contact infor on an existing domain
"""
updateContact = commands.UpdateContact(
id=contact.registry_id,
postal_info=self._make_epp_contact_postal_info(contact=contact),
email=contact.email,
voice=contact.voice,
fax=contact.fax)
fax=contact.fax,
)
try:
registry.send(updateContact, cleaned=True)
except RegistryError as e:
logger.error("Error updating contact, code was %s error was %s" % (e.code, e))
#add more error handling here
#ticket for error handling in epp
logger.error(
"Error updating contact, code was %s error was %s" % (e.code, e)
)
# add more error handling here
# ticket for error handling in epp
def _update_domain_with_contact(self, contact:PublicContact,rem=False):
def _update_domain_with_contact(self, contact: PublicContact, rem=False):
logger.info("received type %s " % contact.contact_type)
domainContact=epp.DomainContact(contact=contact.registry_id,type=contact.contact_type)
domainContact = epp.DomainContact(
contact=contact.registry_id, type=contact.contact_type
)
updateDomain=commands.UpdateDomain(name=self.name, add=[domainContact] )
updateDomain = commands.UpdateDomain(name=self.name, add=[domainContact])
if rem:
updateDomain=commands.UpdateDomain(name=self.name, rem=[domainContact] )
updateDomain = commands.UpdateDomain(name=self.name, rem=[domainContact])
logger.info("Send updated")
try:
registry.send(updateDomain, cleaned=True)
except RegistryError as e:
logger.error("Error changing contact on a domain. Error code is %s error was %s" % (e.code, e))
action="add"
logger.error(
"Error changing contact on a domain. Error code is %s error was %s"
% (e.code, e)
)
action = "add"
if rem:
action="remove"
raise Exception("Can't %s the contact of type %s"%( action, contact.contact_type))
action = "remove"
raise Exception(
"Can't %s the contact of type %s" % (action, contact.contact_type)
)
@Cache
def security_contact(self) -> PublicContact:
"""Get or set the security contact for this domain."""
#get the contacts: call _get_property(contacts=True)
#if contacts exist and security contact is in the contact list
#return that contact
#else call the setter
# get the contacts: call _get_property(contacts=True)
# if contacts exist and security contact is in the contact list
# return that contact
# else call the setter
# send the public default contact
try:
contacts=self._get_property("contacts")
contacts = self._get_property("contacts")
except KeyError as err:
logger.info("Found a key error in security_contact get")
## send public contact to the thingy
##TODO - change to get or create in db?
default= self.get_default_security_contact()
default = self.get_default_security_contact()
# self._cache["contacts"]=[]
# self._cache["contacts"].append({"type":"security", "contact":default})
self.security_contact=default
self.security_contact = default
return default
except Exception as e:
logger.error("found an error ")
@ -444,97 +473,129 @@ class Domain(TimeStampedModel, DomainHelper):
else:
logger.info("Showing contacts")
for contact in contacts:
if isinstance(contact, dict) and "type" in contact.keys() and \
"contact" in contact.keys() and contact["type"]=="security":
if (
isinstance(contact, dict)
and "type" in contact.keys()
and "contact" in contact.keys()
and contact["type"] == "security"
):
return contact["contact"]
##TODO -get the security contact, requires changing the implemenation below and the parser from epplib
#request=InfoContact(securityID)
#contactInfo=...send(request)
#convert info to a PublicContact
#return the info in Public conta
#TODO - below line never executes with current logic
# request=InfoContact(securityID)
# contactInfo=...send(request)
# convert info to a PublicContact
# return the info in Public conta
# TODO - below line never executes with current logic
return self.get_default_security_contact()
def _add_registrant_to_existing_domain(self, contact: PublicContact):
self._update_epp_contact(contact=contact)
self._update_epp_contact(contact=contact)
updateDomain=commands.UpdateDomain(name=self.name, registrant=contact.registry_id )
try:
registry.send(updateDomain, cleaned=True)
except RegistryError as e:
logger.error("Error changing to new registrant error code is %s, error is %s" % (e.code, e))
#TODO-error handling better here?
updateDomain = commands.UpdateDomain(
name=self.name, registrant=contact.registry_id
)
try:
registry.send(updateDomain, cleaned=True)
except RegistryError as e:
logger.error(
"Error changing to new registrant error code is %s, error is %s"
% (e.code, e)
)
# TODO-error handling better here?
def _set_singleton_contact(self, contact: PublicContact, expectedType:str):
def _set_singleton_contact(self, contact: PublicContact, expectedType: str):
""""""
logger.info("_set_singleton_contact()-> contactype type being set: %s expected type is: %s"%(contact, expectedType))
if expectedType!=contact.contact_type:
raise ValueError("Cannot set a contact with a different contact type, expected type was %s"% expectedType)
logger.info(
"_set_singleton_contact()-> contactype type being set: %s expected type is: %s"
% (contact, expectedType)
)
if expectedType != contact.contact_type:
raise ValueError(
"Cannot set a contact with a different contact type, expected type was %s"
% expectedType
)
isRegistrant=contact.contact_type==contact.ContactTypeChoices.REGISTRANT
isRegistrant = contact.contact_type == contact.ContactTypeChoices.REGISTRANT
isEmptySecurity = (
contact.contact_type == contact.ContactTypeChoices.SECURITY
and contact.email == ""
)
# get publicContact objects that have the matching domain and type but a different id, should be only one
hasOtherContact = (
PublicContact.objects.exclude(registry_id=contact.registry_id)
.filter(domain=self, contact_type=contact.contact_type)
.exists()
)
logger.info("has other contact %s" % hasOtherContact)
domainContactExists = PublicContact.objects.filter(registry_id=contact.registry_id).exists()
contactIsAlreadyOnDomain = PublicContact.objects.filter(domain=self,registry_id=contact.registry_id,contact_type=contact.contact_type ).exists()
contactOfTypeExists = PublicContact.objects.filter(domain=self,contact_type=contact.contact_type ).exists()
#get publicContact objects that have the matching domain and type but a different id, should be only one
hasOtherContact = PublicContact.objects.exclude(registry_id=contact.registry_id).filter(domain=self,contact_type=contact.contact_type ).exists()
logger.info("has other contact %s"%hasOtherContact)
##if no record exists with this contact type
logger.info(
"_set_singleton_contact()-> adding contact that shouldn't exist already"
)
# make contact in registry, duplicate and errors handled there
errorCode = self._make_contact_in_registry(contact)
logger.info("_set_singleton_contact()-> adding contact that shouldn't exist already")
#make contact in registry, duplicate and errors handled there
errorCode= self._make_contact_in_registry(contact)
# if contact.contact_type==contact.ContactTypeChoices.REGISTRANT:
# logger.info("_set_singleton_contact()-> creating the registrant")
# if contact.contact_type==contact.ContactTypeChoices.REGISTRANT:
# logger.info("_set_singleton_contact()-> creating the registrant")
# self._make_contact_in_registry(contact)
# else:
# logger.info("_set_singleton_contact()-> updating domain with the new contact")
# self._make_contact_in_registry(contact)
# else:
# logger.info("_set_singleton_contact()-> updating domain with the new contact")
# self._update_domain_with_contact(contact, rem=False)
# self._update_domain_with_contact(contact, rem=False)
# contact is already added to the domain, but something has changed on it
#contact is already added to the domain, but something has changed on it
#TODO - check here if contact already exists on domain in registry
#if domain has registrant and type is registrant this will be true,
#if type is anything else it should be in the contact list
alreadyExistsInRegistry=errorCode==ErrorCode.OBJECT_EXISTS
#if an error occured besides duplication, stop
if not alreadyExistsInRegistry and errorCode!= ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY:
# TODO - check here if contact already exists on domain in registry
# if domain has registrant and type is registrant this will be true,
# if type is anything else it should be in the contact list
alreadyExistsInRegistry = errorCode == ErrorCode.OBJECT_EXISTS
# if an error occured besides duplication, stop
if (
not alreadyExistsInRegistry
and errorCode != ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY
):
raise Exception("Unable to add contact to registry")
#contact doesn't exist on the domain yet
# contact doesn't exist on the domain yet
logger.info("_set_singleton_contact()-> contact has been added to the registry")
#if has conflicting contacts in our db remove them
# if has conflicting contacts in our db remove them
if hasOtherContact:
logger.info("_set_singleton_contact()-> updating domain by removing old contact and adding new one")
existing_contact=PublicContact.objects.exclude(registry_id=contact.registry_id).filter(domain=self,contact_type=contact.contact_type ).get()
logger.info(
"_set_singleton_contact()-> updating domain by removing old contact and adding new one"
)
existing_contact = (
PublicContact.objects.exclude(registry_id=contact.registry_id)
.filter(domain=self, contact_type=contact.contact_type)
.get()
)
if isRegistrant:
#send update domain only for registant contacts
# send update domain only for registant contacts
existing_contact.delete()
self._add_registrant_to_existing_domain(contact)
else:
#remove the old contact and add a new one
# remove the old contact and add a new one
try:
self._update_domain_with_contact(contact=existing_contact, rem=True)
existing_contact.delete()
except Exception as err:
logger.error("Raising error after removing and adding a new contact")
raise(err)
#if just added to registry and not a registrant add contact to domain
if not alreadyExistsInRegistry and not isRegistrant:
self._update_domain_with_contact(contact=contact, rem=False)
#if already exists just update
elif alreadyExistsInRegistry:
self._update_epp_contact(contact=contact)
logger.error(
"Raising error after removing and adding a new contact"
)
raise (err)
# if just added to registry and not a registrant add contact to domain
if not isEmptySecurity:
if not alreadyExistsInRegistry and not isRegistrant:
print("UPDATING domain with the contact")
self._update_domain_with_contact(contact=contact, rem=False)
# if already exists just update
elif alreadyExistsInRegistry:
print("updating the contact itself")
self._update_epp_contact(contact=contact)
@security_contact.setter # type: ignore
def security_contact(self, contact: PublicContact):
@ -544,7 +605,9 @@ class Domain(TimeStampedModel, DomainHelper):
and should have the security email from DomainApplication"""
logger.info("making security contact in registry")
self._set_singleton_contact(contact, expectedType=contact.ContactTypeChoices.SECURITY)
self._set_singleton_contact(
contact, expectedType=contact.ContactTypeChoices.SECURITY
)
@Cache
def technical_contact(self) -> PublicContact:
@ -554,7 +617,9 @@ class Domain(TimeStampedModel, DomainHelper):
@technical_contact.setter # type: ignore
def technical_contact(self, contact: PublicContact):
logger.info("making technical contact")
self._set_singleton_contact(contact, expectedType=contact.ContactTypeChoices.TECHNICAL)
self._set_singleton_contact(
contact, expectedType=contact.ContactTypeChoices.TECHNICAL
)
def is_active(self) -> bool:
"""Currently just returns if the state is created, because then it should be live, theoretically.
@ -562,7 +627,7 @@ class Domain(TimeStampedModel, DomainHelper):
Is the domain live on the inter webs?
could be replaced with request to see if ok status is set
"""
return self.state==self.State.CREATED
return self.state == self.State.CREATED
def transfer(self):
"""Going somewhere. Not implemented."""
@ -578,7 +643,7 @@ class Domain(TimeStampedModel, DomainHelper):
def get_security_email(self):
logger.info("get_security_email-> getting the contact ")
secContact=self.security_contact
secContact = self.security_contact
return secContact.email
def remove_client_hold(self):
@ -644,17 +709,19 @@ class Domain(TimeStampedModel, DomainHelper):
def _get_or_create_domain(self):
"""Try to fetch info about this domain. Create it if it does not exist."""
already_tried_to_create = False
count=0
while not already_tried_to_create and count<3:
count = 0
while not already_tried_to_create and count < 3:
try:
logger.info("_get_or_create_domain()-> getting info on the domain, should hit an error")
logger.info(
"_get_or_create_domain()-> getting info on the domain, should hit an error"
)
req = commands.InfoDomain(name=self.name)
domainInfo= registry.send(req, cleaned=True).res_data[0]
already_tried_to_create = True
domainInfo = registry.send(req, cleaned=True).res_data[0]
already_tried_to_create = True
return domainInfo
except RegistryError as e:
count+=1
count += 1
if already_tried_to_create:
logger.error("Already tried to create")
@ -665,6 +732,7 @@ class Domain(TimeStampedModel, DomainHelper):
# avoid infinite loop
already_tried_to_create = True
self.pendingCreate()
self.save()
else:
logger.error(e)
logger.error(e.code)
@ -674,103 +742,126 @@ class Domain(TimeStampedModel, DomainHelper):
def pendingCreate(self):
logger.info("In make domain in registry ")
registrant = PublicContact.get_default_registrant()
registrant.domain=self
registrant.save() ##calls the registrant_contact.setter
registrant.domain = self
registrant.save() ##calls the registrant_contact.setter
logger.info("registrant is %s" % registrant)
#TODO-notes no chg item for registrant in the epplib should
security_contact=self.get_default_security_contact()
# TODO-notes no chg item for registrant in the epplib should
req = commands.CreateDomain(
name=self.name,
registrant=registrant.registry_id,
auth_info=epp.DomainAuthInfo(
pw="2fooBAR123fooBaz"
), # not a password
auth_info=epp.DomainAuthInfo(pw="2fooBAR123fooBaz"), # not a password
)
logger.info("_get_or_create_domain()-> about to send domain request")
logger.info(req)
try:
response=registry.send(req, cleaned=True)
response = registry.send(req, cleaned=True)
logger.info(response)
except RegistryError as err:
if err.code!=ErrorCode.OBJECT_EXISTS:
if err.code != ErrorCode.OBJECT_EXISTS:
raise err
logger.info("_get_or_create_domain()-> registry received create for "+self.name)
print("making all defaults")
self.addAllDefaults()
logger.info(
"_get_or_create_domain()-> registry received create for " + self.name
)
def addAllDefaults(self):
security_contact = self.get_default_security_contact()
security_contact.domain = self
technical_contact = PublicContact.get_default_technical()
technical_contact.domain = self
administrative_contact = PublicContact.get_default_administrative()
administrative_contact.domain = self
technical_contact.save()
administrative_contact.save()
security_contact.save()
self.save()
print("security contact")
print(security_contact)
def testSettingOtherContacts(self):
##delete this funciton
logger.info("testSettingAllContacts")
technical_contact=PublicContact.get_default_technical()
technical_contact.domain=self
administrative_contact=PublicContact.get_default_administrative()
administrative_contact.domain=self
technical_contact = PublicContact.get_default_technical()
technical_contact.domain = self
administrative_contact = PublicContact.get_default_administrative()
administrative_contact.domain = self
# security_contact.save()
technical_contact.save()
administrative_contact.save()
@transition(field="state", source=State.PENDING_CREATE, target=State.CLIENT_HOLD)
def clientHold(self):
##TODO - check to see if client hold is allowed should happen outside of this function
#(check prohibited statuses)
# (check prohibited statuses)
logger.info("clientHold()-> inside clientHold")
pass
#TODO -send clientHold here
# TODO -send clientHold here
@transition(field="state", source=State.CLIENT_HOLD, target=State.DELETED)
def deleted(self):
logger.info("pendingCreate()-> inside pending create")
pass
#TODO - send delete here
@transition(field="state", source=[State.PENDING_CREATE, State.SERVER_HOLD, State.CLIENT_HOLD], target=State.CREATED)
# TODO - send delete here
@transition(
field="state",
source=[State.PENDING_CREATE, State.SERVER_HOLD, State.CLIENT_HOLD],
target=State.CREATED,
)
def created(self):
logger.info("created()-> inside setting create")
#TODO - do anything else here?
def _disclose_fields(self,isSecurity=False):
# TODO - do anything else here?
def _disclose_fields(self, contact: PublicContact):
"""creates a disclose object that can be added to a contact Create using
.disclose= <this function> on the command before sending.
if item is security email then make sure email is visable"""
.disclose= <this function> on the command before sending.
if item is security email then make sure email is visable"""
isSecurity = contact.contact_type == contact.ContactTypeChoices.SECURITY
DF = epp.DiscloseField
fields={DF.FAX, DF.VOICE, DF.ADDR}
if not isSecurity:
fields = {DF.FAX, DF.VOICE, DF.ADDR}
if not isSecurity or (
isSecurity and contact.email == PublicContact.get_default_security().email
):
fields.add(DF.EMAIL)
return epp.Disclose(
flag=False,
fields={DF.FAX, DF.VOICE, DF.ADDR},
types={DF.ADDR: "loc"},
)
def _make_epp_contact_postal_info(self, contact:PublicContact):
flag=False,
fields={DF.FAX, DF.VOICE, DF.ADDR},
types={DF.ADDR: "loc"},
)
def _make_epp_contact_postal_info(self, contact: PublicContact):
return epp.PostalInfo( # type: ignore
name=contact.name,
addr=epp.ContactAddr(
street=[
getattr(contact, street)
for street in ["street1", "street2", "street3"]
if hasattr(contact, street)
],
city=contact.city,
pc=contact.pc,
cc=contact.cc,
sp=contact.sp,
),
org=contact.org,
type="loc",
)
name=contact.name,
addr=epp.ContactAddr(
street=[
getattr(contact, street)
for street in ["street1", "street2", "street3"]
if hasattr(contact, street)
],
city=contact.city,
pc=contact.pc,
cc=contact.cc,
sp=contact.sp,
),
org=contact.org,
type="loc",
)
def _make_contact_in_registry(self, contact: PublicContact):
"""Create the contact in the registry, ignore duplicate contact errors
returns int corresponding to ErrorCode values"""
logger.info(contact)
logger.info(contact.registry_id)
print("***CREATING THE CCONTACT")
create = commands.CreateContact(
id=contact.registry_id,
postal_info=self._make_epp_contact_postal_info(contact=contact),
@ -779,22 +870,32 @@ class Domain(TimeStampedModel, DomainHelper):
fax=contact.fax,
auth_info=epp.ContactAuthInfo(pw="2fooBAR123fooBaz"),
)
# security contacts should only show email addresses, for now
create.disclose=self._disclose_fields(isSecurity=contact.contact_type==contact.ContactTypeChoices.SECURITY)
# security contacts should only show email addresses, for now
create.disclose = self._disclose_fields(contact=contact)
try:
logger.info("sending contact")
registry.send(create, cleaned=True)
return ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY
except RegistryError as err:
#don't throw an error if it is just saying this is a duplicate contact
if err.code!=ErrorCode.OBJECT_EXISTS:
logger.error("Registry threw error for contact id %s contact type is %s, error code is\n %s full error is %s",contact.registry_id, contact.contact_type, err.code, err)
#TODO - Error handling here
# don't throw an error if it is just saying this is a duplicate contact
if err.code != ErrorCode.OBJECT_EXISTS:
logger.error(
"Registry threw error for contact id %s contact type is %s, error code is\n %s full error is %s",
contact.registry_id,
contact.contact_type,
err.code,
err,
)
# TODO - Error handling here
else:
logger.warning("Registrar tried to create duplicate contact for id %s",contact.registry_id)
logger.warning(
"Registrar tried to create duplicate contact for id %s",
contact.registry_id,
)
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]
@ -806,14 +907,21 @@ class Domain(TimeStampedModel, DomainHelper):
return self._request_contact_info(contact)
except RegistryError as e:
if e.code == ErrorCode.OBJECT_DOES_NOT_EXIST:
logger.info("_get_or_create_contact()-> contact doesn't exist so making it")
contact.domain=self
contact.save()#this will call the function based on type of contact
logger.info(
"_get_or_create_contact()-> contact doesn't exist so making it"
)
contact.domain = self
contact.save() # this will call the function based on type of contact
return self._request_contact_info(contact=contact)
else:
logger.error("Registry threw error for contact id %s contact type is %s, error code is\n %s full error is %s",contact.registry_id, contact.contact_type, err.code, err)
logger.error(
"Registry threw error for contact id %s contact type is %s, error code is\n %s full error is %s",
contact.registry_id,
contact.contact_type,
err.code,
err,
)
raise e
@ -846,12 +954,12 @@ class Domain(TimeStampedModel, DomainHelper):
# 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 ...}
logger.info("_fetch_cache()-> cleaned is "+str(cleaned))
logger.info("_fetch_cache()-> cleaned is " + str(cleaned))
# get contact info, if there are any
if (
# fetch_contacts and
"_contacts" in cleaned
"_contacts" in cleaned
and isinstance(cleaned["_contacts"], list)
and len(cleaned["_contacts"])
):
@ -884,12 +992,15 @@ class Domain(TimeStampedModel, DomainHelper):
cleaned["contacts"].append(
{k: v for k, v in contact.items() if v is not ...}
)
logger.info("_fetch_cache()-> after getting contacts cleaned is "+str(cleaned))
logger.info(
"_fetch_cache()-> after getting contacts cleaned is "
+ str(cleaned)
)
# get nameserver info, if there are any
if (
# fetch_hosts and
"_hosts" in cleaned
"_hosts" in cleaned
and isinstance(cleaned["_hosts"], list)
and len(cleaned["_hosts"])
):

2
src/registrar/models/domain_information.py
View file

@ -13,7 +13,7 @@ logger = logging.getLogger(__name__)
class DomainInformation(TimeStampedModel):
"""A registrant's domain information for that domain, exported from
DomainApplication. We use these field from DomainApplication with few exceptation
DomainApplication. We use these field from DomainAchpplication with few exceptation
which are 'removed' via pop at the bottom of this file. Most of design for domain
management's user information are based on application, but we cannot change
the application once approved, so copying them that way we can make changes

2
src/registrar/models/public_contact.py
View file

@ -149,4 +149,4 @@ class PublicContact(TimeStampedModel):
)
def __str__(self):
return f"{self.name} <{self.email}> id: {self.registry_id}"
return f"{self.name} <{self.email}> id: {self.registry_id} type: {self.contact_type}"

220
src/registrar/tests/test_models_domain.py
View file

@ -10,13 +10,14 @@ import datetime
from registrar.models import Domain # add in DomainApplication, User,
from unittest import skip
from epplibwrapper import commands,common
from epplibwrapper import commands, common
from registrar.models.domain_application import DomainApplication
from registrar.models.domain_information import DomainInformation
from registrar.models.draft_domain import DraftDomain
from registrar.models.public_contact import PublicContact
from registrar.models.user import User
class MockEppLib(TestCase):
class fakedEppObject(object):
""""""
@ -33,7 +34,7 @@ class MockEppLib(TestCase):
contacts=["123"],
hosts=["fake.host.com"],
)
infoDomainNoContact= fakedEppObject(
infoDomainNoContact = fakedEppObject(
"security",
cr_date=datetime.datetime(2023, 5, 25, 19, 45, 35),
contacts=[],
@ -49,7 +50,7 @@ class MockEppLib(TestCase):
def mockSend(self, _request, cleaned):
""""""
if isinstance(_request, commands.InfoDomain):
if getattr(_request,"name",None)=="security.gov":
if getattr(_request, "name", None) == "security.gov":
return MagicMock(res_data=[self.infoDomainNoContact])
return MagicMock(res_data=[self.mockDataInfoDomain])
elif isinstance(_request, commands.InfoContact):
@ -66,9 +67,8 @@ class MockEppLib(TestCase):
def tearDown(self):
self.mockSendPatch.stop()
class TestDomainCache(MockEppLib):
# def setUp(self):
# #call setup from the mock epplib
# super().setUp()
@ -142,6 +142,7 @@ class TestDomainCache(MockEppLib):
self.assertEqual(domain._cache["hosts"], [expectedHostsDict])
##IS THERE AN ERROR HERE???,
class TestDomainCreation(TestCase):
"""Rule: An approved domain application must result in a domain"""
@ -159,7 +160,7 @@ class TestDomainCreation(TestCase):
But a domain object does not exist in the registry
"""
patcher = patch("registrar.models.domain.Domain._get_or_create_domain")
mocked_domain_creation=patcher.start()
mocked_domain_creation = patcher.start()
draft_domain, _ = DraftDomain.objects.get_or_create(name="igorville.gov")
user, _ = User.objects.get_or_create()
application = DomainApplication.objects.create(
@ -167,14 +168,13 @@ class TestDomainCreation(TestCase):
)
# skip using the submit method
application.status = DomainApplication.SUBMITTED
#transition to approve state
# transition to approve state
application.approve()
# should hav information present for this domain
domain = Domain.objects.get(name="igorville.gov")
self.assertTrue(domain)
mocked_domain_creation.assert_not_called()
@skip("not implemented yet")
def test_accessing_domain_properties_creates_domain_in_registry(self):
"""
@ -216,6 +216,7 @@ class TestDomainCreation(TestCase):
Domain.objects.delete()
# User.objects.delete()
class TestRegistrantContacts(MockEppLib):
"""Rule: Registrants may modify their WHOIS data"""
@ -226,17 +227,20 @@ class TestRegistrantContacts(MockEppLib):
And the registrant is the admin on a domain
"""
super().setUp()
#mock create contact email extension
self.contactMailingAddressPatch = patch("registrar.models.domain.commands.command_extensions.CreateContactMailingAddressExtension")
self.mockCreateContactExtension=self.contactMailingAddressPatch.start()
# mock create contact email extension
self.contactMailingAddressPatch = patch(
"registrar.models.domain.commands.command_extensions.CreateContactMailingAddressExtension"
)
self.mockCreateContactExtension = self.contactMailingAddressPatch.start()
#mock create contact
self.createContactPatch = patch("registrar.models.domain.commands.CreateContact")
self.mockCreateContact=self.createContactPatch.start()
#mock the sending
# mock create contact
self.createContactPatch = patch(
"registrar.models.domain.commands.CreateContact"
)
self.mockCreateContact = self.createContactPatch.start()
# mock the sending
self.domain, _ = Domain.objects.get_or_create(name="security.gov")
self.domain,_ = Domain.objects.get_or_create(name="security.gov")
# draft_domain, _ = DraftDomain.objects.get_or_create(name="igorville.gov")
# user, _ = User.objects.get_or_create()
@ -244,14 +248,13 @@ class TestRegistrantContacts(MockEppLib):
# creator=user, requested_domain=draft_domain
# )
# self.application.status = DomainApplication.SUBMITTED
#transition to approve state
# transition to approve state
def tearDown(self):
super().tearDown()
# self.contactMailingAddressPatch.stop()
# self.createContactPatch.stop()
# @skip("source code not implemented")
def test_no_security_email(self):
"""
Scenario: Registrant has not added a security contact email
@ -260,31 +263,85 @@ class TestRegistrantContacts(MockEppLib):
Then the domain has a valid security contact with CISA defaults
And disclose flags are set to keep the email address hidden
"""
print(self.domain)
#get security contact
expectedSecContact=PublicContact.get_default_security()
expectedSecContact.domain=self.domain
receivedSecContact=self.domain.security_contact
# making a domain should make it domain
print(self.domain)
expectedSecContact = PublicContact.get_default_security()
expectedSecContact.domain = self.domain
self.domain.pendingCreate()
DF = common.DiscloseField
di = common.Disclose(flag=False, fields={DF.FAX, DF.VOICE, DF.ADDR}, types={DF.ADDR: "loc"})
di = common.Disclose(
flag=False,
fields={DF.FAX, DF.VOICE, DF.ADDR, DF.EMAIL},
types={DF.ADDR: "loc"},
)
#check docs here looks like we may have more than one address field but
addr = common.ContactAddr(street=[expectedSecContact.street1,expectedSecContact.street2,expectedSecContact.street3] , city=expectedSecContact.city, pc=expectedSecContact.pc, cc=expectedSecContact.cc, sp=expectedSecContact.sp)
pi = common.PostalInfo(name=expectedSecContact.name, addr=addr, org=expectedSecContact.org, type="loc")
ai = common.ContactAuthInfo(pw='feedabee')
expectedCreateCommand=commands.CreateContact(id=expectedSecContact.registry_id, postal_info=pi, email=expectedSecContact.email, voice=expectedSecContact.voice, fax=expectedSecContact.fax, auth_info=ai, disclose=di, vat=None, ident=None, notify_email=None)
expectedUpdateDomain =commands.UpdateDomain(name=self.domain.name, add=[common.DomainContact(contact=expectedSecContact.registry_id, type="security")])
#check that send has triggered the create command
# check docs here looks like we may have more than one address field but
addr = common.ContactAddr(
street=[
expectedSecContact.street1,
expectedSecContact.street2,
expectedSecContact.street3,
],
city=expectedSecContact.city,
pc=expectedSecContact.pc,
cc=expectedSecContact.cc,
sp=expectedSecContact.sp,
)
pi = common.PostalInfo(
name=expectedSecContact.name,
addr=addr,
org=expectedSecContact.org,
type="loc",
)
ai = common.ContactAuthInfo(pw="feedabee")
expectedCreateCommand = commands.CreateContact(
id=expectedSecContact.registry_id,
postal_info=pi,
email=expectedSecContact.email,
voice=expectedSecContact.voice,
fax=expectedSecContact.fax,
auth_info=ai,
disclose=di,
vat=None,
ident=None,
notify_email=None,
)
expectedUpdateDomain = commands.UpdateDomain(
name=self.domain.name,
add=[
common.DomainContact(
contact=expectedSecContact.registry_id, type="security"
)
],
)
# check that send has triggered the create command
# print(expectedCreateCommand)
print(self.mockedSendFunction.call_count)
print(
PublicContact.objects.filter(
domain=self.domain,
contact_type=PublicContact.ContactTypeChoices.SECURITY,
)
)
# assert( self.mockedSendFunction.call_count
assert PublicContact.objects.filter(domain=self.domain).count() == 4
assert (
PublicContact.objects.get(
domain=self.domain,
contact_type=PublicContact.ContactTypeChoices.SECURITY,
).email
== expectedSecContact.email
)
# assert()
# self.mockedSendFunction.assert_any_call(expectedCreateCommand,True)
# self.mockedSendFunction.assert_any_call(expectedUpdateDomain, True)
# check that the security contact sent is the same as the one recieved
self.mockedSendFunction.assert_any_call(expectedCreateCommand,True)
self.mockedSendFunction.assert_any_call(expectedUpdateDomain, True)
#check that the security contact sent is the same as the one recieved
self.assertEqual(receivedSecContact,expectedSecContact)
@skip("not implemented yet")
# @skip("not implemented yet")
def test_user_adds_security_email(self):
"""
Scenario: Registrant adds a security contact email
@ -294,30 +351,73 @@ class TestRegistrantContacts(MockEppLib):
And Domain sends `commands.UpdateDomain` to the registry with the newly
created contact of type 'security'
"""
#make a security contact that is a PublicContact
expectedSecContact=PublicContact.get_default_security()
expectedSecContact.domain=self.domain
expectedSecContact.email="newEmail@fake.com"
expectedSecContact.registry_id="456"
expectedSecContact.name="Fakey McPhakerson"
self.domain.security_contact=expectedSecContact
# make a security contact that is a PublicContact
expectedSecContact = PublicContact.get_default_security()
expectedSecContact.domain = self.domain
expectedSecContact.email = "newEmail@fake.com"
expectedSecContact.registry_id = "456"
expectedSecContact.name = "Fakey McPhakerson"
#check create contact sent with email
# calls the security contact setter as if you did
# self.domain.security_contact=expectedSecContact
expectedSecContact.save()
# check create contact sent with email
DF = common.DiscloseField
di = common.Disclose(flag=False, fields={DF.FAX, DF.VOICE, DF.ADDR}, types={DF.ADDR: "loc"})
di = common.Disclose(
flag=False, fields={DF.FAX, DF.VOICE, DF.ADDR}, types={DF.ADDR: "loc"}
)
addr = common.ContactAddr(street=[expectedSecContact.street1,expectedSecContact.street2,expectedSecContact.street3] , city=expectedSecContact.city, pc=expectedSecContact.pc, cc=expectedSecContact.cc, sp=expectedSecContact.sp)
pi = common.PostalInfo(name=expectedSecContact.name, addr=addr, org=expectedSecContact.org, type="loc")
ai = common.ContactAuthInfo(pw='feedabee')
addr = common.ContactAddr(
street=[
expectedSecContact.street1,
expectedSecContact.street2,
expectedSecContact.street3,
],
city=expectedSecContact.city,
pc=expectedSecContact.pc,
cc=expectedSecContact.cc,
sp=expectedSecContact.sp,
)
pi = common.PostalInfo(
name=expectedSecContact.name,
addr=addr,
org=expectedSecContact.org,
type="loc",
)
ai = common.ContactAuthInfo(pw="feedabee")
expectedCreateCommand=commands.CreateContact(id=expectedSecContact.registry_id, postal_info=pi, email=expectedSecContact.email, voice=expectedSecContact.voice, fax=expectedSecContact.fax, auth_info=ai, disclose=di, vat=None, ident=None, notify_email=None)
expectedUpdateDomain =commands.UpdateDomain(name=self.domain.name, add=[common.DomainContact(contact=expectedSecContact.registry_id, type="security")])
expectedCreateCommand = commands.CreateContact(
id=expectedSecContact.registry_id,
postal_info=pi,
email=expectedSecContact.email,
voice=expectedSecContact.voice,
fax=expectedSecContact.fax,
auth_info=ai,
disclose=di,
vat=None,
ident=None,
notify_email=None,
)
expectedUpdateDomain = commands.UpdateDomain(
name=self.domain.name,
add=[
common.DomainContact(
contact=expectedSecContact.registry_id, type="security"
)
],
)
#check that send has triggered the create command for the contact
self.mockedSendFunction.assert_any_call(expectedCreateCommand, True)
##check domain contact was updated
self.mockedSendFunction.assert_any_call(expectedUpdateDomain, True)
# check that send has triggered the create command for the contact
print("finishing")
print(PublicContact.objects.filter(domain=self.domain))
receivedSecurityContact = PublicContact.objects.get(
domain=self.domain, contact_type=PublicContact.ContactTypeChoices.SECURITY
)
print(self.mockedSendFunction.call_count)
assert self.mockedSendFunction.call_count == 2
assert receivedSecurityContact == expectedSecContact
@skip("not implemented yet")
def test_security_email_is_idempotent(self):
@ -330,10 +430,10 @@ class TestRegistrantContacts(MockEppLib):
# 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
#will send epplibwrapper.errors.RegistryError with code 2302 for a duplicate contact
# will send epplibwrapper.errors.RegistryError with code 2302 for a duplicate contact
#set the smae fake contact to the email
#show no errors
# set the smae fake contact to the email
# show no errors
raise
@skip("not implemented yet")

2
src/registrar/views/domain.py
View file

@ -270,7 +270,7 @@ class DomainSecurityEmailView(DomainPermissionView, FormMixin):
contact.save()
##update security email here
#call the setter
# call the setter
messages.success(
self.request, "The security email for this domain have been updated."
)