Merge pull request #2827 from cisagov/ms/2307-send-notification-emails

#2307: send notification emails on changes to domain - [MS]
This commit is contained in:
Matt-Spence 2024-10-21 12:02:18 -05:00 committed by GitHub
commit 6ebafec72a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 601 additions and 28 deletions

View file

@ -25,7 +25,7 @@ services:
# Run Django in debug mode on local
- DJANGO_DEBUG=True
# Set DJANGO_LOG_LEVEL in env
- DJANGO_LOG_LEVEL
- DJANGO_LOG_LEVEL=DEBUG
# Run Django without production flags
- IS_PRODUCTION=False
# Tell Django where it is being hosted

View file

@ -476,8 +476,10 @@ class JsonServerFormatter(ServerFormatter):
def format(self, record):
formatted_record = super().format(record)
if not hasattr(record, "server_time"):
record.server_time = self.formatTime(record, self.datefmt)
log_entry = {"server_time": record.server_time, "level": record.levelname, "message": formatted_record}
return json.dumps(log_entry)

View file

@ -831,7 +831,6 @@ class DomainRequest(TimeStampedModel):
if custom_email_content:
context["custom_email_content"] = custom_email_content
send_templated_email(
email_template,
email_template_subject,
@ -877,7 +876,6 @@ class DomainRequest(TimeStampedModel):
DraftDomain = apps.get_model("registrar.DraftDomain")
if not DraftDomain.string_could_be_domain(self.requested_domain.name):
raise ValueError("Requested domain is not a valid domain name.")
# if the domain has not been submitted before this must be the first time
if not self.first_submitted_date:
self.first_submitted_date = timezone.now().date()

View file

@ -8,7 +8,7 @@
<p>
Domain managers can update all information related to a domain within the
.gov registrar, including including security email and DNS name servers.
.gov registrar, including security email and DNS name servers.
</p>
<ul class="usa-list">

View file

@ -0,0 +1,31 @@
{% autoescape off %}{# In a text file, we don't want to have HTML entities escaped #}
Hi,
An update was made to a domain you manage.
DOMAIN: {{domain}}
UPDATED BY: {{user}}
UPDATED ON: {{date}}
INFORMATION UPDATED: {{changes}}
You can view this update in the .gov registrar <https://manage.get.gov/>.
Get help with managing your .gov domain <https://get.gov/help/domain-management/>.
----------------------------------------------------------------
WHY DID YOU RECEIVE THIS EMAIL?
Youre listed as a domain manager for {{domain}}, so youll receive a notification whenever changes are made to that domain.
If you have questions or concerns, reach out to the person who made the change or reply to this email.
THANK YOU
.Gov helps the public identify official, trusted information. Thank you for using a .gov domain.
----------------------------------------------------------------
The .gov team
Contact us <https://get.gov/contact/>
Learn about .gov <https://get.gov>
The .gov registry is a part of the Cybersecurity and Infrastructure Security Agency (CISA) <https://cisa.gov/>
{% endautoescape %}

View file

@ -0,0 +1 @@
An update was made to {{domain}}

View file

@ -61,11 +61,58 @@ class TestEmails(TestCase):
# Assert that an email wasn't sent
self.assertFalse(self.mock_client.send_email.called)
@boto3_mocking.patching
def test_email_with_cc(self):
"""Test sending email with cc works"""
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
send_templated_email(
"emails/update_to_approved_domain.txt",
"emails/update_to_approved_domain_subject.txt",
"doesnotexist@igorville.com",
context={"domain": "test", "user": "test", "date": 1, "changes": "test"},
bcc_address=None,
cc_addresses=["testy2@town.com", "mayor@igorville.gov"],
)
# check that an email was sent
self.assertTrue(self.mock_client.send_email.called)
# check the call sequence for the email
args, kwargs = self.mock_client.send_email.call_args
self.assertIn("Destination", kwargs)
self.assertIn("CcAddresses", kwargs["Destination"])
self.assertEqual(["testy2@town.com", "mayor@igorville.gov"], kwargs["Destination"]["CcAddresses"])
@boto3_mocking.patching
@override_settings(IS_PRODUCTION=True)
def test_email_with_cc_in_prod(self):
"""Test sending email with cc works in prod"""
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
send_templated_email(
"emails/update_to_approved_domain.txt",
"emails/update_to_approved_domain_subject.txt",
"doesnotexist@igorville.com",
context={"domain": "test", "user": "test", "date": 1, "changes": "test"},
bcc_address=None,
cc_addresses=["testy2@town.com", "mayor@igorville.gov"],
)
# check that an email was sent
self.assertTrue(self.mock_client.send_email.called)
# check the call sequence for the email
args, kwargs = self.mock_client.send_email.call_args
self.assertIn("Destination", kwargs)
self.assertIn("CcAddresses", kwargs["Destination"])
self.assertEqual(["testy2@town.com", "mayor@igorville.gov"], kwargs["Destination"]["CcAddresses"])
@boto3_mocking.patching
@less_console_noise_decorator
def test_submission_confirmation(self):
"""Submission confirmation email works."""
domain_request = completed_domain_request()
domain_request = completed_domain_request(user=User.objects.create(username="test", email="testy@town.com"))
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
@ -102,7 +149,9 @@ class TestEmails(TestCase):
@less_console_noise_decorator
def test_submission_confirmation_no_current_website_spacing(self):
"""Test line spacing without current_website."""
domain_request = completed_domain_request(has_current_website=False)
domain_request = completed_domain_request(
has_current_website=False, user=User.objects.create(username="test", email="testy@town.com")
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
_, kwargs = self.mock_client.send_email.call_args
@ -115,7 +164,9 @@ class TestEmails(TestCase):
@less_console_noise_decorator
def test_submission_confirmation_current_website_spacing(self):
"""Test line spacing with current_website."""
domain_request = completed_domain_request(has_current_website=True)
domain_request = completed_domain_request(
has_current_website=True, user=User.objects.create(username="test", email="testy@town.com")
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
_, kwargs = self.mock_client.send_email.call_args
@ -132,7 +183,11 @@ class TestEmails(TestCase):
# Create fake creator
_creator = User.objects.create(
username="MrMeoward", first_name="Meoward", last_name="Jones", phone="(888) 888 8888"
username="MrMeoward",
first_name="Meoward",
last_name="Jones",
phone="(888) 888 8888",
email="testy@town.com",
)
# Create a fake domain request
@ -149,7 +204,9 @@ class TestEmails(TestCase):
@less_console_noise_decorator
def test_submission_confirmation_no_other_contacts_spacing(self):
"""Test line spacing without other contacts."""
domain_request = completed_domain_request(has_other_contacts=False)
domain_request = completed_domain_request(
has_other_contacts=False, user=User.objects.create(username="test", email="testy@town.com")
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
_, kwargs = self.mock_client.send_email.call_args
@ -161,7 +218,9 @@ class TestEmails(TestCase):
@less_console_noise_decorator
def test_submission_confirmation_alternative_govdomain_spacing(self):
"""Test line spacing with alternative .gov domain."""
domain_request = completed_domain_request(has_alternative_gov_domain=True)
domain_request = completed_domain_request(
has_alternative_gov_domain=True, user=User.objects.create(username="test", email="testy@town.com")
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
_, kwargs = self.mock_client.send_email.call_args
@ -174,7 +233,9 @@ class TestEmails(TestCase):
@less_console_noise_decorator
def test_submission_confirmation_no_alternative_govdomain_spacing(self):
"""Test line spacing without alternative .gov domain."""
domain_request = completed_domain_request(has_alternative_gov_domain=False)
domain_request = completed_domain_request(
has_alternative_gov_domain=False, user=User.objects.create(username="test", email="testy@town.com")
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
_, kwargs = self.mock_client.send_email.call_args
@ -187,7 +248,9 @@ class TestEmails(TestCase):
@less_console_noise_decorator
def test_submission_confirmation_about_your_organization_spacing(self):
"""Test line spacing with about your organization."""
domain_request = completed_domain_request(has_about_your_organization=True)
domain_request = completed_domain_request(
has_about_your_organization=True, user=User.objects.create(username="test", email="testy@town.com")
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
_, kwargs = self.mock_client.send_email.call_args
@ -200,7 +263,9 @@ class TestEmails(TestCase):
@less_console_noise_decorator
def test_submission_confirmation_no_about_your_organization_spacing(self):
"""Test line spacing without about your organization."""
domain_request = completed_domain_request(has_about_your_organization=False)
domain_request = completed_domain_request(
has_about_your_organization=False, user=User.objects.create(username="test", email="testy@town.com")
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
_, kwargs = self.mock_client.send_email.call_args
@ -213,7 +278,9 @@ class TestEmails(TestCase):
@less_console_noise_decorator
def test_submission_confirmation_anything_else_spacing(self):
"""Test line spacing with anything else."""
domain_request = completed_domain_request(has_anything_else=True)
domain_request = completed_domain_request(
has_anything_else=True, user=User.objects.create(username="test", email="testy@town.com")
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
_, kwargs = self.mock_client.send_email.call_args
@ -225,7 +292,9 @@ class TestEmails(TestCase):
@less_console_noise_decorator
def test_submission_confirmation_no_anything_else_spacing(self):
"""Test line spacing without anything else."""
domain_request = completed_domain_request(has_anything_else=False)
domain_request = completed_domain_request(
has_anything_else=False, user=User.objects.create(username="test", email="testy@town.com")
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
domain_request.submit()
_, kwargs = self.mock_client.send_email.call_args

View file

@ -305,7 +305,7 @@ class TestDomainRequest(TestCase):
@less_console_noise_decorator
def test_submit_from_withdrawn_sends_email(self):
msg = "Create a withdrawn domain request and submit it and see if email was sent."
user, _ = User.objects.get_or_create(username="testy")
user, _ = User.objects.get_or_create(username="testy", email="testy@town.com")
domain_request = completed_domain_request(status=DomainRequest.DomainRequestStatus.WITHDRAWN, user=user)
self.check_email_sent(domain_request, msg, "submit", 1, expected_content="Hi", expected_email=user.email)
@ -324,14 +324,14 @@ class TestDomainRequest(TestCase):
@less_console_noise_decorator
def test_approve_sends_email(self):
msg = "Create a domain request and approve it and see if email was sent."
user, _ = User.objects.get_or_create(username="testy")
user, _ = User.objects.get_or_create(username="testy", email="testy@town.com")
domain_request = completed_domain_request(status=DomainRequest.DomainRequestStatus.IN_REVIEW, user=user)
self.check_email_sent(domain_request, msg, "approve", 1, expected_content="approved", expected_email=user.email)
@less_console_noise_decorator
def test_withdraw_sends_email(self):
msg = "Create a domain request and withdraw it and see if email was sent."
user, _ = User.objects.get_or_create(username="testy")
user, _ = User.objects.get_or_create(username="testy", email="testy@town.com")
domain_request = completed_domain_request(status=DomainRequest.DomainRequestStatus.IN_REVIEW, user=user)
self.check_email_sent(
domain_request, msg, "withdraw", 1, expected_content="withdrawn", expected_email=user.email
@ -339,7 +339,7 @@ class TestDomainRequest(TestCase):
def test_reject_sends_email(self):
"Create a domain request and reject it and see if email was sent."
user, _ = User.objects.get_or_create(username="testy")
user, _ = User.objects.get_or_create(username="testy", email="testy@town.com")
domain_request = completed_domain_request(status=DomainRequest.DomainRequestStatus.APPROVED, user=user)
expected_email = user.email
email_allowed, _ = AllowedEmail.objects.get_or_create(email=expected_email)

View file

@ -65,6 +65,10 @@ class TestWithDomainPermissions(TestWithUser):
datetime.combine(date.today() + timedelta(days=1), datetime.min.time())
),
)
self.domain_dns_needed, _ = Domain.objects.get_or_create(
name="dns-needed.gov",
state=Domain.State.DNS_NEEDED,
)
self.domain_deleted, _ = Domain.objects.get_or_create(
name="deleted.gov",
state=Domain.State.DELETED,
@ -91,6 +95,7 @@ class TestWithDomainPermissions(TestWithUser):
DomainInformation.objects.get_or_create(creator=self.user, domain=self.domain_just_nameserver)
DomainInformation.objects.get_or_create(creator=self.user, domain=self.domain_on_hold)
DomainInformation.objects.get_or_create(creator=self.user, domain=self.domain_deleted)
DomainInformation.objects.get_or_create(creator=self.user, domain=self.domain_dns_needed)
self.role, _ = UserDomainRole.objects.get_or_create(
user=self.user, domain=self.domain, role=UserDomainRole.Roles.MANAGER
@ -99,6 +104,9 @@ class TestWithDomainPermissions(TestWithUser):
UserDomainRole.objects.get_or_create(
user=self.user, domain=self.domain_dsdata, role=UserDomainRole.Roles.MANAGER
)
UserDomainRole.objects.get_or_create(
user=self.user, domain=self.domain_dns_needed, role=UserDomainRole.Roles.MANAGER
)
UserDomainRole.objects.get_or_create(
user=self.user,
domain=self.domain_multdsdata,
@ -236,6 +244,7 @@ class TestDomainDetail(TestDomainOverview):
# At the time of this test's writing, there are 6 UNKNOWN domains inherited
# from constructors. Let's reset.
with less_console_noise():
PublicContact.objects.all().delete()
Domain.objects.all().delete()
UserDomainRole.objects.all().delete()
@ -1967,3 +1976,292 @@ class TestDomainDNSSEC(TestDomainOverview):
self.assertContains(
result, str(DsDataError(code=DsDataErrorCodes.INVALID_DIGEST_SHA256)), count=2, status_code=200
)
class TestDomainChangeNotifications(TestDomainOverview):
"""Test email notifications on updates to domain information"""
@classmethod
def setUpClass(cls):
super().setUpClass()
allowed_emails = [
AllowedEmail(email="info@example.com"),
AllowedEmail(email="doesnotexist@igorville.com"),
]
AllowedEmail.objects.bulk_create(allowed_emails)
def setUp(self):
super().setUp()
self.mock_client_class = MagicMock()
self.mock_client = self.mock_client_class.return_value
@classmethod
def tearDownClass(cls):
super().tearDownClass()
AllowedEmail.objects.all().delete()
@boto3_mocking.patching
@less_console_noise_decorator
def test_notification_on_org_name_change(self):
"""Test that an email is sent when the organization name is changed."""
# We may end up sending emails on org name changes later, but it will be addressed
# in the portfolio itself, rather than the individual domain.
self.domain_information.organization_name = "Town of Igorville"
self.domain_information.address_line1 = "123 Main St"
self.domain_information.city = "Igorville"
self.domain_information.state_territory = "IL"
self.domain_information.zipcode = "62052"
self.domain_information.save()
org_name_page = self.app.get(reverse("domain-org-name-address", kwargs={"pk": self.domain.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
org_name_page.form["organization_name"] = "Not igorville"
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
org_name_page.form.submit()
# Check that an email was sent
self.assertTrue(self.mock_client.send_email.called)
# Check email content
# check the call sequence for the email
_, kwargs = self.mock_client.send_email.call_args
self.assertIn("Content", kwargs)
self.assertIn("Simple", kwargs["Content"])
self.assertIn("Subject", kwargs["Content"]["Simple"])
self.assertIn("Body", kwargs["Content"]["Simple"])
body = kwargs["Content"]["Simple"]["Body"]["Text"]["Data"]
self.assertIn("DOMAIN: igorville.gov", body)
self.assertIn("UPDATED BY: First Last info@example.com", body)
self.assertIn("INFORMATION UPDATED: Organization details", body)
@boto3_mocking.patching
@less_console_noise_decorator
def test_no_notification_on_org_name_change_with_portfolio(self):
"""Test that an email is not sent on org name change when the domain is in a portfolio"""
portfolio, _ = Portfolio.objects.get_or_create(organization_name="Test org", creator=self.user)
self.domain_information.organization_name = "Town of Igorville"
self.domain_information.address_line1 = "123 Main St"
self.domain_information.city = "Igorville"
self.domain_information.state_territory = "IL"
self.domain_information.zipcode = "62052"
self.domain_information.portfolio = portfolio
self.domain_information.save()
org_name_page = self.app.get(reverse("domain-org-name-address", kwargs={"pk": self.domain.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
org_name_page.form["organization_name"] = "Not igorville"
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
org_name_page.form.submit()
# Check that an email was not sent
self.assertFalse(self.mock_client.send_email.called)
@boto3_mocking.patching
@less_console_noise_decorator
def test_no_notification_on_change_by_analyst(self):
"""Test that an email is not sent on org name change when the domain is in a portfolio"""
portfolio, _ = Portfolio.objects.get_or_create(organization_name="Test org", creator=self.user)
self.domain_information.organization_name = "Town of Igorville"
self.domain_information.address_line1 = "123 Main St"
self.domain_information.city = "Igorville"
self.domain_information.state_territory = "IL"
self.domain_information.zipcode = "62052"
self.domain_information.portfolio = portfolio
self.domain_information.save()
org_name_page = self.app.get(reverse("domain-org-name-address", kwargs={"pk": self.domain.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
session = self.app.session
session["analyst_action"] = "foo"
session["analyst_action_location"] = self.domain.id
session.save()
org_name_page.form["organization_name"] = "Not igorville"
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
org_name_page.form.submit()
# Check that an email was not sent
self.assertFalse(self.mock_client.send_email.called)
@boto3_mocking.patching
@less_console_noise_decorator
def test_notification_on_security_email_change(self):
"""Test that an email is sent when the security email is changed."""
security_email_page = self.app.get(reverse("domain-security-email", kwargs={"pk": self.domain.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
security_email_page.form["security_email"] = "new_security@example.com"
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
security_email_page.form.submit()
self.assertTrue(self.mock_client.send_email.called)
_, kwargs = self.mock_client.send_email.call_args
body = kwargs["Content"]["Simple"]["Body"]["Text"]["Data"]
self.assertIn("DOMAIN: igorville.gov", body)
self.assertIn("UPDATED BY: First Last info@example.com", body)
self.assertIn("INFORMATION UPDATED: Security email", body)
@boto3_mocking.patching
@less_console_noise_decorator
def test_notification_on_dnssec_enable(self):
"""Test that an email is sent when DNSSEC is enabled."""
page = self.client.get(reverse("domain-dns-dnssec", kwargs={"pk": self.domain_multdsdata.id}))
self.assertContains(page, "Disable DNSSEC")
# Prepare the data for the POST request
post_data = {
"disable_dnssec": "Disable DNSSEC",
}
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
updated_page = self.client.post(
reverse("domain-dns-dnssec", kwargs={"pk": self.domain.id}),
post_data,
follow=True,
)
self.assertEqual(updated_page.status_code, 200)
self.assertContains(updated_page, "Enable DNSSEC")
self.assertTrue(self.mock_client.send_email.called)
_, kwargs = self.mock_client.send_email.call_args
body = kwargs["Content"]["Simple"]["Body"]["Text"]["Data"]
self.assertIn("DOMAIN: igorville.gov", body)
self.assertIn("UPDATED BY: First Last info@example.com", body)
self.assertIn("INFORMATION UPDATED: DNSSEC / DS Data", body)
@boto3_mocking.patching
@less_console_noise_decorator
def test_notification_on_ds_data_change(self):
"""Test that an email is sent when DS data is changed."""
ds_data_page = self.app.get(reverse("domain-dns-dnssec-dsdata", kwargs={"pk": self.domain.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
# Add DS data
ds_data_page.forms[0]["form-0-key_tag"] = "12345"
ds_data_page.forms[0]["form-0-algorithm"] = "13"
ds_data_page.forms[0]["form-0-digest_type"] = "2"
ds_data_page.forms[0]["form-0-digest"] = "1234567890ABCDEF1234567890ABCDEF1234567890ABCDEF1234567890ABCDEF"
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
ds_data_page.forms[0].submit()
# check that the email was sent
self.assertTrue(self.mock_client.send_email.called)
# check some stuff about the email
_, kwargs = self.mock_client.send_email.call_args
body = kwargs["Content"]["Simple"]["Body"]["Text"]["Data"]
self.assertIn("DOMAIN: igorville.gov", body)
self.assertIn("UPDATED BY: First Last info@example.com", body)
self.assertIn("INFORMATION UPDATED: DNSSEC / DS Data", body)
@boto3_mocking.patching
@less_console_noise_decorator
def test_notification_on_senior_official_change(self):
"""Test that an email is sent when the senior official information is changed."""
self.domain_information.senior_official = Contact.objects.create(
first_name="Old", last_name="Official", title="Manager", email="old_official@example.com"
)
self.domain_information.save()
senior_official_page = self.app.get(reverse("domain-senior-official", kwargs={"pk": self.domain.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
senior_official_page.form["first_name"] = "New"
senior_official_page.form["last_name"] = "Official"
senior_official_page.form["title"] = "Director"
senior_official_page.form["email"] = "new_official@example.com"
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
senior_official_page.form.submit()
self.assertTrue(self.mock_client.send_email.called)
_, kwargs = self.mock_client.send_email.call_args
body = kwargs["Content"]["Simple"]["Body"]["Text"]["Data"]
self.assertIn("DOMAIN: igorville.gov", body)
self.assertIn("UPDATED BY: First Last info@example.com", body)
self.assertIn("INFORMATION UPDATED: Senior official", body)
@boto3_mocking.patching
@less_console_noise_decorator
def test_no_notification_on_senior_official_when_portfolio(self):
"""Test that an email is not sent when the senior official information is changed
and the domain is in a portfolio."""
self.domain_information.senior_official = Contact.objects.create(
first_name="Old", last_name="Official", title="Manager", email="old_official@example.com"
)
portfolio, _ = Portfolio.objects.get_or_create(
organization_name="portfolio",
creator=self.user,
)
self.domain_information.portfolio = portfolio
self.domain_information.save()
senior_official_page = self.app.get(reverse("domain-senior-official", kwargs={"pk": self.domain.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
senior_official_page.form["first_name"] = "New"
senior_official_page.form["last_name"] = "Official"
senior_official_page.form["title"] = "Director"
senior_official_page.form["email"] = "new_official@example.com"
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
senior_official_page.form.submit()
self.assertFalse(self.mock_client.send_email.called)
@boto3_mocking.patching
@less_console_noise_decorator
def test_no_notification_when_dns_needed(self):
"""Test that an email is not sent when nameservers are changed while the state is DNS_NEEDED."""
nameservers_page = self.app.get(reverse("domain-dns-nameservers", kwargs={"pk": self.domain_dns_needed.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
# add nameservers
nameservers_page.form["form-0-server"] = "ns1-new.dns-needed.gov"
nameservers_page.form["form-0-ip"] = "192.168.1.1"
nameservers_page.form["form-1-server"] = "ns2-new.dns-needed.gov"
nameservers_page.form["form-1-ip"] = "192.168.1.2"
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client_class):
nameservers_page.form.submit()
# Check that an email was not sent
self.assertFalse(self.mock_client.send_email.called)

View file

@ -22,30 +22,47 @@ class EmailSendingError(RuntimeError):
pass
def send_templated_email(
def send_templated_email( # noqa
template_name: str,
subject_template_name: str,
to_address: str,
bcc_address="",
to_address: str = "",
bcc_address: str = "",
context={},
attachment_file=None,
wrap_email=False,
cc_addresses: list[str] = [],
):
"""Send an email built from a template to one email address.
"""Send an email built from a template.
to_address and bcc_address currently only support single addresses.
cc_address is a list and can contain many addresses. Emails not in the
whitelist (if applicable) will be filtered out before sending.
template_name and subject_template_name are relative to the same template
context as Django's HTML templates. context gives additional information
that the template may use.
Raises EmailSendingError if SES client could not be accessed
Raises EmailSendingError if:
SES client could not be accessed
No valid recipient addresses are provided
"""
# by default assume we can send to all addresses (prod has no whitelist)
sendable_cc_addresses = cc_addresses
if not settings.IS_PRODUCTION: # type: ignore
# Split into a function: C901 'send_templated_email' is too complex.
# Raises an error if we cannot send an email (due to restrictions).
# Does nothing otherwise.
_can_send_email(to_address, bcc_address)
# if we're not in prod, we need to check the whitelist for CC'ed addresses
sendable_cc_addresses, blocked_cc_addresses = get_sendable_addresses(cc_addresses)
if blocked_cc_addresses:
logger.warning("Some CC'ed addresses were removed: %s.", blocked_cc_addresses)
template = get_template(template_name)
email_body = template.render(context=context)
@ -64,14 +81,23 @@ def send_templated_email(
aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
config=settings.BOTO_CONFIG,
)
logger.info(f"An email was sent! Template name: {template_name} to {to_address}")
logger.info(f"Connected to SES client! Template name: {template_name} to {to_address}")
except Exception as exc:
logger.debug("E-mail unable to send! Could not access the SES client.")
raise EmailSendingError("Could not access the SES client.") from exc
destination = {"ToAddresses": [to_address]}
destination = {}
if to_address:
destination["ToAddresses"] = [to_address]
if bcc_address:
destination["BccAddresses"] = [bcc_address]
if cc_addresses:
destination["CcAddresses"] = sendable_cc_addresses
# make sure we don't try and send an email to nowhere
if not destination:
message = "Email unable to send, no valid recipients provided."
raise EmailSendingError(message)
try:
if not attachment_file:
@ -90,6 +116,7 @@ def send_templated_email(
},
},
)
logger.info("Email sent to [%s], bcc [%s], cc %s", to_address, bcc_address, sendable_cc_addresses)
else:
ses_client = boto3.client(
"ses",
@ -101,6 +128,10 @@ def send_templated_email(
send_email_with_attachment(
settings.DEFAULT_FROM_EMAIL, to_address, subject, email_body, attachment_file, ses_client
)
logger.info(
"Email with attachment sent to [%s], bcc [%s], cc %s", to_address, bcc_address, sendable_cc_addresses
)
except Exception as exc:
raise EmailSendingError("Could not send SES email.") from exc
@ -125,6 +156,33 @@ def _can_send_email(to_address, bcc_address):
raise EmailSendingError(message.format(bcc_address))
def get_sendable_addresses(addresses: list[str]) -> tuple[list[str], list[str]]:
"""Checks whether a list of addresses can be sent to.
Returns: a lists of all provided addresses that are ok to send to and a list of addresses that were blocked.
Paramaters:
addresses: a list of strings representing all addresses to be checked.
"""
if flag_is_active(None, "disable_email_sending"): # type: ignore
message = "Could not send email. Email sending is disabled due to flag 'disable_email_sending'."
logger.warning(message)
return ([], [])
else:
AllowedEmail = apps.get_model("registrar", "AllowedEmail")
allowed_emails = []
blocked_emails = []
for address in addresses:
if AllowedEmail.is_allowed_email(address):
allowed_emails.append(address)
else:
blocked_emails.append(address)
return allowed_emails, blocked_emails
def wrap_text_and_preserve_paragraphs(text, width):
"""
Wraps text to `width` preserving newlines; splits on '\n', wraps segments, rejoins with '\n'.

View file

@ -5,6 +5,7 @@ authorized users can see information on a domain, every view here should
inherit from `DomainPermissionView` (or DomainInvitationPermissionDeleteView).
"""
from datetime import date
import logging
from django.contrib import messages
@ -152,6 +153,103 @@ class DomainFormBaseView(DomainBaseView, FormMixin):
return current_domain_info
def send_update_notification(self, form, force_send=False):
"""Send a notification to all domain managers that an update has occured
for a single domain. Uses update_to_approved_domain.txt template.
If there are no changes to the form, emails will NOT be sent unless force_send
is set to True.
"""
# send notification email for changes to any of these forms
form_label_dict = {
DomainSecurityEmailForm: "Security email",
DomainDnssecForm: "DNSSEC / DS Data",
DomainDsdataFormset: "DNSSEC / DS Data",
DomainOrgNameAddressForm: "Organization details",
SeniorOfficialContactForm: "Senior official",
NameserverFormset: "Name servers",
}
# forms of these types should not send notifications if they're part of a portfolio/Organization
check_for_portfolio = {
DomainOrgNameAddressForm,
SeniorOfficialContactForm,
}
is_analyst_action = "analyst_action" in self.session and "analyst_action_location" in self.session
should_notify = False
if form.__class__ in form_label_dict:
if is_analyst_action:
logger.debug("No notification sent: Action was conducted by an analyst")
else:
# these types of forms can cause notifications
should_notify = True
if form.__class__ in check_for_portfolio:
# some forms shouldn't cause notifications if they are in a portfolio
info = self.get_domain_info_from_domain()
if not info or info.portfolio:
logger.debug("No notification sent: Domain is part of a portfolio")
should_notify = False
else:
# don't notify for any other types of forms
should_notify = False
if should_notify and (form.has_changed() or force_send):
context = {
"domain": self.object.name,
"user": self.request.user,
"date": date.today(),
"changes": form_label_dict[form.__class__],
}
self.email_domain_managers(
self.object,
"emails/update_to_approved_domain.txt",
"emails/update_to_approved_domain_subject.txt",
context,
)
else:
logger.info(f"No notification sent for {form.__class__}.")
def email_domain_managers(self, domain: Domain, template: str, subject_template: str, context={}):
"""Send a single email built from a template to all managers for a given domain.
template_name and subject_template_name are relative to the same template
context as Django's HTML templates. context gives additional information
that the template may use.
context is a dictionary containing any information needed to fill in values
in the provided template, exactly the same as with send_templated_email.
Will log a warning if the email fails to send for any reason, but will not raise an error.
"""
manager_pks = UserDomainRole.objects.filter(domain=domain.pk, role=UserDomainRole.Roles.MANAGER).values_list(
"user", flat=True
)
emails = list(User.objects.filter(pk__in=manager_pks).values_list("email", flat=True))
try:
# Remove the current user so they aren't CC'ed, since they will be the "to_address"
emails.remove(self.request.user.email) # type: ignore
except ValueError:
pass
try:
send_templated_email(
template,
subject_template,
to_address=self.request.user.email, # type: ignore
context=context,
cc_addresses=emails,
)
except EmailSendingError:
logger.warning(
"Could not sent notification email to %s for domain %s",
emails,
domain.name,
exc_info=True,
)
class DomainView(DomainBaseView):
"""Domain detail overview page."""
@ -227,6 +325,8 @@ class DomainOrgNameAddressView(DomainFormBaseView):
def form_valid(self, form):
"""The form is valid, save the organization name and mailing address."""
self.send_update_notification(form)
form.save()
messages.success(self.request, "The organization information for this domain has been updated.")
@ -330,6 +430,8 @@ class DomainSeniorOfficialView(DomainFormBaseView):
form.set_domain_info(self.object.domain_info)
form.save()
self.send_update_notification(form)
messages.success(self.request, "The senior official for this domain has been updated.")
# superclass has the redirect
@ -408,19 +510,25 @@ class DomainNameserversView(DomainFormBaseView):
self._get_domain(request)
formset = self.get_form()
logger.debug("got formet")
if "btn-cancel-click" in request.POST:
url = self.get_success_url()
return HttpResponseRedirect(url)
if formset.is_valid():
logger.debug("formset is valid")
return self.form_valid(formset)
else:
logger.debug("formset is invalid")
logger.debug(formset.errors)
return self.form_invalid(formset)
def form_valid(self, formset):
"""The formset is valid, perform something with it."""
self.request.session["nameservers_form_domain"] = self.object
initial_state = self.object.state
# Set the nameservers from the formset
nameservers = []
@ -442,7 +550,6 @@ class DomainNameserversView(DomainFormBaseView):
except KeyError:
# no server information in this field, skip it
pass
try:
self.object.nameservers = nameservers
except NameserverError as Err:
@ -462,6 +569,8 @@ class DomainNameserversView(DomainFormBaseView):
messages.error(self.request, NameserverError(code=nsErrorCodes.BAD_DATA))
logger.error(f"Registry error: {Err}")
else:
if initial_state == Domain.State.READY:
self.send_update_notification(formset)
messages.success(
self.request,
"The name servers for this domain have been updated. "
@ -514,7 +623,8 @@ class DomainDNSSECView(DomainFormBaseView):
errmsg = "Error removing existing DNSSEC record(s)."
logger.error(errmsg + ": " + err)
messages.error(self.request, errmsg)
else:
self.send_update_notification(form, force_send=True)
return self.form_valid(form)
@ -638,6 +748,8 @@ class DomainDsDataView(DomainFormBaseView):
logger.error(f"Registry error: {err}")
return self.form_invalid(formset)
else:
self.send_update_notification(formset)
messages.success(self.request, "The DS data records for this domain have been updated.")
# superclass has the redirect
return super().form_valid(formset)
@ -704,8 +816,12 @@ class DomainSecurityEmailView(DomainFormBaseView):
messages.error(self.request, SecurityEmailError(code=SecurityEmailErrorCodes.BAD_DATA))
logger.error(f"Generic registry error: {Err}")
else:
self.send_update_notification(form)
messages.success(self.request, "The security email for this domain has been updated.")
# superclass has the redirect
return super().form_valid(form)
# superclass has the redirect
return redirect(self.get_success_url())