From f940757639a9f535beeb7c6a5875797971a787b3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Thomas=20NO=C3=8BL?= <tnoel@entrouvert.com>
Date: Mon, 12 Dec 2022 10:54:57 +0100
Subject: [PATCH] emails: restrict domains for default_from_email (#72173)

---
 hobo/emails/forms.py      |  3 ++-
 hobo/emails/validators.py | 18 +++++++++++++
 hobo/settings.py          |  4 +++
 tests/test_emails.py      | 57 +++++++++++++++++++++++++++++++++++++--
 4 files changed, 79 insertions(+), 3 deletions(-)

diff --git a/hobo/emails/forms.py b/hobo/emails/forms.py
index 4c838a4..6884a91 100644
--- a/hobo/emails/forms.py
+++ b/hobo/emails/forms.py
@@ -17,12 +17,13 @@ from django import forms
 from django.core.validators import validate_email
 from django.utils.translation import ugettext_lazy as _
 
-from hobo.emails.validators import validate_email_address, validate_email_spf
+from hobo.emails.validators import validate_email_address, validate_email_spf, validate_email_domain
 
 
 class ValidEmailField(forms.EmailField):
     def validate(self, value):
         validate_email(value)
+        validate_email_domain(value)
         validate_email_address(value)
         validate_email_spf(value)
 
diff --git a/hobo/emails/validators.py b/hobo/emails/validators.py
index d2ec203..f9c3e5a 100644
--- a/hobo/emails/validators.py
+++ b/hobo/emails/validators.py
@@ -16,6 +16,7 @@
 
 import smtplib
 import socket
+import urllib.parse
 
 import dns.resolver
 from django.conf import settings
@@ -75,3 +76,20 @@ def validate_email_spf(value, strict=False):
             if allowed_record in spf_record:
                 return
     raise ValidationError(_('No suitable SPF record found for %s') % email_domain)
+
+
+def validate_email_domain(value):
+    domain = value.split('@')[-1].strip().strip('.')
+    domains = settings.EMAIL_FROM_ALLOWED_DOMAINS
+    if '*' in domains or domain in domains:
+        return
+    known_services = getattr(settings, 'KNOWN_SERVICES', {})
+    for app in known_services.values():
+        for instance in app.values():
+            url = instance.get('url')
+            if not url:
+                continue
+            fqdn = urllib.parse.urlparse(url).netloc.split(':')[0]
+            if fqdn == domain or fqdn == 'www.' + domain:
+                return
+    raise ValidationError(_('Domain %s is not allowed') % domain)
diff --git a/hobo/settings.py b/hobo/settings.py
index 12fa3ea..749e1a9 100644
--- a/hobo/settings.py
+++ b/hobo/settings.py
@@ -29,6 +29,10 @@ ALLOWED_HOSTS = []
 
 HOBO_VALIDATE_EMAIL_WITH_SMTP = True
 ALLOWED_SPF_RECORDS = []
+# EMAIL_FROM_ALLOWED_DOMAINS: list of allowed domains for default_from_email.
+# Use ['*'] to allow all domains.
+# Note: all KNOWN_SERVICES url domains are always allowed
+EMAIL_FROM_ALLOWED_DOMAINS = []
 
 # Application definition
 
diff --git a/tests/test_emails.py b/tests/test_emails.py
index 47f601e..6887176 100644
--- a/tests/test_emails.py
+++ b/tests/test_emails.py
@@ -137,14 +137,16 @@ def test_invalid_address(client, admin_user):
     assert 'Enter a valid email address' in force_text(response.content)
 
 
-def test_unkown_address(client, admin_user, dns_resolver, smtp_server):
+def test_unkown_address(client, admin_user, dns_resolver, smtp_server, settings):
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = ['*']
     client.post('/login/', {'username': 'admin', 'password': 'password'})
     response = client.post('/emails/', {'default_from_email': 'john.doe@unknown.com'})
     assert response.status_code == 200
     assert 'Email address not found' in force_text(response.content)
 
 
-def test_kown_address_nospf(client, admin_user, dns_resolver, smtp_server):
+def test_kown_address_nospf(client, admin_user, dns_resolver, smtp_server, settings):
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = ['*']
     client.post('/login/', {'username': 'admin', 'password': 'password'})
     response = client.post('/emails/', {'default_from_email': 'john.doe@example.com'}, follow=True)
     assert response.status_code == 200
@@ -154,6 +156,7 @@ def test_kown_address_nospf(client, admin_user, dns_resolver, smtp_server):
 
 
 def test_spf_allow_all_mail(client, admin_user, dns_resolver, smtp_server, settings):
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = ['*']
     client.post('/login/', {'username': 'admin', 'password': 'password'})
     response = client.post(
         '/emails/', {'default_from_email': 'john.doe@example-spf-allow-all.com'}, follow=True
@@ -165,6 +168,7 @@ def test_spf_allow_all_mail(client, admin_user, dns_resolver, smtp_server, setti
 
 
 def test_invalid_spf(client, admin_user, dns_resolver, smtp_server, settings):
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = ['*']
     settings.ALLOWED_SPF_RECORDS = ['include:example.com']
     client.post('/login/', {'username': 'admin', 'password': 'password'})
     response = client.post('/emails/', {'default_from_email': 'john.doe@example-invalid-spf.com'})
@@ -173,6 +177,7 @@ def test_invalid_spf(client, admin_user, dns_resolver, smtp_server, settings):
 
 
 def test_strict_nospf(client, admin_user, dns_resolver, smtp_server, monkeypatch, settings):
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = ['*']
     settings.ALLOWED_SPF_RECORDS = ['include:allowed_mx.com']
     monkeypatch.setattr('hobo.emails.validators.validate_email_spf.__defaults__', (True,))
     client.post('/login/', {'username': 'admin', 'password': 'password'})
@@ -182,6 +187,7 @@ def test_strict_nospf(client, admin_user, dns_resolver, smtp_server, monkeypatch
 
 
 def test_valid_spf(client, admin_user, dns_resolver, smtp_server, settings):
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = ['*']
     settings.ALLOWED_SPF_RECORDS = ['include:allowed_mx.com']
     client.post('/login/', {'username': 'admin', 'password': 'password'})
     response = client.post('/emails/', {'default_from_email': 'john.doe@example-spf.com'}, follow=True)
@@ -192,6 +198,7 @@ def test_valid_spf(client, admin_user, dns_resolver, smtp_server, settings):
 
 
 def test_no_spf_validation(client, admin_user, dns_resolver, smtp_server, settings):
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = ['*']
     settings.ALLOWED_SPF_RECORDS = []
     client.post('/login/', {'username': 'admin', 'password': 'password'})
     response = client.post(
@@ -203,7 +210,53 @@ def test_no_spf_validation(client, admin_user, dns_resolver, smtp_server, settin
     )
 
 
+def test_sender_allowed_domains(client, admin_user, dns_resolver, smtp_server, settings):
+    settings.HOBO_VALIDATE_EMAIL_WITH_SMTP = False
+
+    client.post('/login/', {'username': 'admin', 'password': 'password'})
+    response = client.post('/emails/', {'default_from_email': 'john.doe@example.com'}, follow=True)
+    assert response.status_code == 200
+    assert 'Domain example.com is not allowed' in force_text(response.content)
+    assert 'Emails settings have been updated.' not in force_text(response.content)
+
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = ['example.com', 'foo.bar']
+    response = client.post('/emails/', {'default_from_email': 'john.doe@example.com'}, follow=True)
+    assert response.status_code == 200
+    assert 'Emails settings have been updated.' in force_text(response.content)
+    assert 'Domain example.com is not allowed' not in force_text(response.content)
+
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = []
+    settings.KNOWN_SERVICES = {
+        'combo': {
+            'portal': {
+                'title': 'Portal',
+                'url': 'https://example.org/test',
+            },
+            'other': {
+                'title': 'Empty',
+            },
+        }
+    }
+    response = client.post('/emails/', {'default_from_email': 'john.doe@example.org'}, follow=True)
+    assert response.status_code == 200
+    assert 'Emails settings have been updated.' in force_text(response.content)
+
+    response = client.post('/emails/', {'default_from_email': 'john.doe@example.com'}, follow=True)
+    assert response.status_code == 200
+    assert 'Domain example.com is not allowed' in force_text(response.content)
+
+    response = client.post('/emails/', {'default_from_email': 'john.doe@brother.example.org'}, follow=True)
+    assert response.status_code == 200
+    assert 'Domain brother.example.org is not allowed' in force_text(response.content)
+
+    settings.KNOWN_SERVICES['combo']['portal']['url'] = 'https://www.example.com'
+    response = client.post('/emails/', {'default_from_email': 'john.doe@example.com'}, follow=True)
+    assert response.status_code == 200
+    assert 'Emails settings have been updated.' in force_text(response.content)
+
+
 def test_emails_view(app, admin_user, dns_resolver, smtp_server, settings):
+    settings.EMAIL_FROM_ALLOWED_DOMAINS = ['*']
     settings.ALLOWED_SPF_RECORDS = ['include:allowed_mx.com']
     app = login(app)
     resp = app.get('/emails/')
-- 
2.35.1

