# Copyright (C) 1998-2018 by the Free Software Foundation, Inc.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
# USA.
"""Do more detailed spam detection.
This module hard codes site wide spam detection. By hacking the
KNOWN_SPAMMERS variable, you can set up more regular expression matches
against message headers. If spam is detected the message is discarded
immediately.
TBD: This needs to be made more configurable and robust.
"""
import re
from unicodedata import normalize
from email.Errors import HeaderParseError
from email.Header import decode_header
from email.Utils import parseaddr
from Mailman import mm_cfg
from Mailman import Errors
from Mailman import i18n
from Mailman import Utils
from Mailman.Handlers.Hold import hold_for_approval
from Mailman.Logging.Syslog import syslog
try:
True, False
except NameError:
True = 1
False = 0
# First, play footsie with _ so that the following are marked as translated,
# but aren't actually translated until we need the text later on.
def _(s):
return s
class SpamDetected(Errors.DiscardMessage):
"""The message contains known spam"""
class HeaderMatchHold(Errors.HoldMessage):
def __init__(self, pattern):
self.__pattern = pattern
def reason_notice(self):
pattern = self.__pattern
return _('Header matched regexp: %(pattern)s')
# And reset the translator
_ = i18n._
def getDecodedHeaders(msg, cset='utf-8'):
"""Returns a unicode containing all the headers of msg, unfolded and
RFC 2047 decoded, normalized and separated by new lines.
"""
headers = u''
for h, v in msg.items():
uvalue = u''
try:
v = decode_header(re.sub('\n\s', ' ', v))
except HeaderParseError:
v = [(v, 'us-ascii')]
for frag, cs in v:
if not cs:
cs = 'us-ascii'
try:
uvalue += unicode(frag, cs, 'replace')
except LookupError:
# The encoding charset is unknown. At this point, frag
# has been QP or base64 decoded into a byte string whose
# charset we don't know how to handle. We will try to
# unicode it as iso-8859-1 which may result in a garbled
# mess, but we have to do something.
uvalue += unicode(frag, 'iso-8859-1', 'replace')
uhdr = h.decode('us-ascii', 'replace')
headers += u'%s: %s\n' % (h, normalize(mm_cfg.NORMALIZE_FORM, uvalue))
return headers
def process(mlist, msg, msgdata):
# Before anything else, check DMARC if necessary. We do this as early
# as possible so reject/discard actions trump other holds/approvals and
# wrap/munge actions get flagged even for approved messages.
# But not for owner mail which should not be subject to DMARC reject or
# discard actions.
if not msgdata.get('toowner'):
msgdata['from_is_list'] = 0
dn, addr = parseaddr(msg.get('from'))
if addr and mlist.dmarc_moderation_action > 0:
if (mlist.GetPattern(addr, mlist.dmarc_moderation_addresses) or
Utils.IsDMARCProhibited(mlist, addr)):
# Note that for dmarc_moderation_action, 0 = Accept,
# 1 = Munge, 2 = Wrap, 3 = Reject, 4 = Discard
if mlist.dmarc_moderation_action == 1:
msgdata['from_is_list'] = 1
elif mlist.dmarc_moderation_action == 2:
msgdata['from_is_list'] = 2
elif mlist.dmarc_moderation_action == 3:
# Reject
text = mlist.dmarc_moderation_notice
if text:
text = Utils.wrap(text)
else:
listowner = mlist.GetOwnerEmail()
text = Utils.wrap(_(
"""You are not allowed to post to this mailing list From: a domain which
publishes a DMARC policy of reject or quarantine, and your message has been
automatically rejected. If you think that your messages are being rejected in
error, contact the mailing list owner at %(listowner)s."""))
raise Errors.RejectMessage, text
elif mlist.dmarc_moderation_action == 4:
raise Errors.DiscardMessage
# Get member address if any.
for sender in msg.get_senders():
if mlist.isMember(sender):
break
else:
sender = msg.get_sender()
if (mlist.member_verbosity_threshold > 0 and
Utils.IsVerboseMember(mlist, sender)
):
mlist.setMemberOption(sender, mm_cfg.Moderate, 1)
syslog('vette',
'%s: Automatically Moderated %s for verbose postings.',
mlist.real_name, sender)
if msgdata.get('approved'):
return
# First do site hard coded header spam checks
for header, regex in mm_cfg.KNOWN_SPAMMERS:
cre = re.compile(regex, re.IGNORECASE)
for value in msg.get_all(header, []):
mo = cre.search(value)
if mo:
# we've detected spam, so throw the message away
raise SpamDetected
# Now do header_filter_rules
# TK: Collect headers in sub-parts because attachment filename
# extension may be a clue to possible virus/spam.
headers = u''
# Get the character set of the lists preferred language for headers
lcset = Utils.GetCharSet(mlist.preferred_language)
for p in msg.walk():
headers += getDecodedHeaders(p, lcset)
for patterns, action, empty in mlist.header_filter_rules:
if action == mm_cfg.DEFER:
continue
for pattern in patterns.splitlines():
if pattern.startswith('#'):
continue
# ignore 'empty' patterns
if not pattern.strip():
continue
pattern = Utils.xml_to_unicode(pattern, lcset)
pattern = normalize(mm_cfg.NORMALIZE_FORM, pattern)
try:
mo = re.search(pattern,
headers,
re.IGNORECASE|re.MULTILINE|re.UNICODE)
except (re.error, TypeError):
syslog('error',
'ignoring header_filter_rules invalid pattern: %s',
pattern)
if mo:
if action == mm_cfg.DISCARD:
raise Errors.DiscardMessage
if action == mm_cfg.REJECT:
if msgdata.get('toowner'):
# Don't send rejection notice if addressed to '-owner'
# because it may trigger a loop of notices if the
# sender address is forged. We just discard it here.
raise Errors.DiscardMessage
raise Errors.RejectMessage(
_('Message rejected by filter rule match'))
if action == mm_cfg.HOLD:
if msgdata.get('toowner'):
# Don't hold '-owner' addressed message. We just
# pass it here but list-owner can set this to be
# discarded on the GUI if he wants.
return
hold_for_approval(
mlist, msg, msgdata, HeaderMatchHold(pattern))
if action == mm_cfg.ACCEPT:
return
