From 2f093c5db446688e102140859cf9052fea98fc32 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 | 20 ++++++++++++++++
 hobo/settings.py          |  4 ++++
 tests/test_emails.py      | 49 ++++++++++++++++++++++++++++++++++++++-
 4 files changed, 74 insertions(+), 2 deletions(-)

diff --git a/hobo/emails/forms.py b/hobo/emails/forms.py
index 4c838a4..5e0bdf2 100644
--- a/hobo/emails/forms.py
+++ b/hobo/emails/forms.py
@@ -17,7 +17,7 @@ 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):
@@ -25,6 +25,7 @@ class ValidEmailField(forms.EmailField):
         validate_email(value)
         validate_email_address(value)
         validate_email_spf(value)
+        validate_email_domain(value)
 
 
 class EmailsForm(forms.Form):
diff --git a/hobo/emails/validators.py b/hobo/emails/validators.py
index d2ec203..87f697c 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,22 @@ 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_ACCEPTED_DOMAINS
+    if '*' in domains or domain in domains:
+        return
+    if '.' not in domain:
+        raise ValidationError(_('Domain %s is not accepted') % email_domain)
+    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:
+                return
+    raise ValidationError(_('Domain %s is not accepted') % domain)
diff --git a/hobo/settings.py b/hobo/settings.py
index 12fa3ea..1b3fd29 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_SENDER_ACCEPTED_DOMAINS: list of accepted domains for
+# default_from_email. It can contains '*' to accept all domains.
+# Note: all KNOWN_SERVICES url domains are always accepter
+EMAIL_FROM_ACCEPTED_DOMAINS = []
 
 # Application definition
 
diff --git a/tests/test_emails.py b/tests/test_emails.py
index 47f601e..b759be9 100644
--- a/tests/test_emails.py
+++ b/tests/test_emails.py
@@ -144,7 +144,8 @@ def test_unkown_address(client, admin_user, dns_resolver, smtp_server):
     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_ACCEPTED_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 +155,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_ACCEPTED_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 +167,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_ACCEPTED_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 +176,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_ACCEPTED_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 +186,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_ACCEPTED_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 +197,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_ACCEPTED_DOMAINS = ['*']
     settings.ALLOWED_SPF_RECORDS = []
     client.post('/login/', {'username': 'admin', 'password': 'password'})
     response = client.post(
@@ -203,7 +209,48 @@ def test_no_spf_validation(client, admin_user, dns_resolver, smtp_server, settin
     )
 
 
+def test_sender_accepted_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 accepted' in force_text(response.content)
+    assert 'Emails settings have been updated.' not in force_text(response.content)
+
+    settings.EMAIL_FROM_ACCEPTED_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 accepted' not in force_text(response.content)
+
+    settings.EMAIL_FROM_ACCEPTED_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 accepted' 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 accepted' in force_text(response.content)
+
+
 def test_emails_view(app, admin_user, dns_resolver, smtp_server, settings):
+    settings.EMAIL_FROM_ACCEPTED_DOMAINS = ['*']
     settings.ALLOWED_SPF_RECORDS = ['include:allowed_mx.com']
     app = login(app)
     resp = app.get('/emails/')
-- 
2.35.1

