93 lines
3.3 KiB
Python
93 lines
3.3 KiB
Python
# hobo - portal to configure and deploy applications
|
|
# Copyright (C) 2015-2016 Entr'ouvert
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it
|
|
# under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import smtplib
|
|
import socket
|
|
import urllib.parse
|
|
|
|
import dns.resolver
|
|
from django.conf import settings
|
|
from django.core.exceptions import ValidationError
|
|
from django.utils.encoding import force_bytes
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from hobo.environment.utils import get_operational_services
|
|
|
|
|
|
def validate_email_address(value):
|
|
if not settings.HOBO_VALIDATE_EMAIL_WITH_SMTP:
|
|
return
|
|
email_domain = value.split('@')[-1]
|
|
try:
|
|
mx_server = sorted(dns.resolver.query(email_domain, 'MX'), key=lambda rdata: rdata.preference)[
|
|
0
|
|
].exchange.to_text()
|
|
except dns.exception.DNSException as e:
|
|
raise ValidationError(_('Error: %s') % str(e))
|
|
try:
|
|
smtp = smtplib.SMTP(host=mx_server, timeout=10)
|
|
try:
|
|
status, msg = smtp.helo()
|
|
if status != 250:
|
|
raise OSError(msg)
|
|
smtp.mail('')
|
|
status, msg = smtp.rcpt(value)
|
|
smtp.quit()
|
|
finally:
|
|
smtp.close()
|
|
except OSError as e:
|
|
raise ValidationError(
|
|
_('Error while connecting to %(server)s: %(msg)s') % {'server': mx_server, 'msg': e}
|
|
)
|
|
if status // 100 == 5:
|
|
raise ValidationError(_('Email address not found on %s') % mx_server)
|
|
|
|
|
|
def validate_email_spf(value, strict=False):
|
|
allowed_records = settings.ALLOWED_SPF_RECORDS
|
|
if not allowed_records:
|
|
return
|
|
email_domain = value.split('@')[-1]
|
|
txt_records = []
|
|
try:
|
|
for txt_record in dns.resolver.query(email_domain, 'TXT'):
|
|
txt_records.extend(txt_record.strings)
|
|
except dns.exception.DNSException:
|
|
pass
|
|
spf_records = [x for x in txt_records if x.startswith(b'v=spf1 ')]
|
|
if not strict and not spf_records:
|
|
return
|
|
allowed_records = [force_bytes(x) for x in allowed_records]
|
|
for spf_record in spf_records:
|
|
if b'+all' in spf_record:
|
|
return
|
|
for allowed_record in allowed_records:
|
|
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
|
|
for service in get_operational_services():
|
|
fqdn = urllib.parse.urlparse(service.base_url).netloc.split(':')[0]
|
|
if fqdn == domain or fqdn == 'www.' + domain:
|
|
return
|
|
raise ValidationError(_('Domain %s is not allowed') % domain)
|