path: root/src/mailman/rules/tests/test_dmarc.py

# Copyright (C) 2016 by the Free Software Foundation, Inc.
#
# This file is part of GNU Mailman.
#
# GNU Mailman 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 3 of the License, or (at your option)
# any later version.
#
# GNU Mailman 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
# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.

"""Tests and mocks for DMARC rule."""

import os
import threading

from contextlib import ExitStack
from datetime import timedelta
from dns.exception import DNSException
from dns.rdatatype import CNAME, TXT
from dns.resolver import NXDOMAIN, NoAnswer
from http.server import BaseHTTPRequestHandler, HTTPServer
from lazr.config import as_timedelta
from mailman.app.lifecycle import create_list
from mailman.config import config
from mailman.interfaces.mailinglist import DMARCMitigateAction
from mailman.rules import dmarc
from mailman.testing.helpers import (
    LogFileMark, configuration, specialized_message_from_string as mfs,
    wait_for_webservice)
from mailman.testing.layers import ConfigLayer
from mailman.utilities.datetime import now
from pkg_resources import resource_filename
from public import public
from unittest import TestCase
from unittest.mock import patch


@public
def get_dns_resolver(
        rtype=TXT,
        rdata=b'v=DMARC1; p=reject;',
        rmult=False,
        cmult=False,
        cloop=False,
        cmiss=False):
    """Create a dns.resolver.Resolver mock.

    This is used to return a predictable response to a _dmarc query.  It
    returns p=reject for the example.biz domain and raises an exception for
    other examples.

    It only implements those classes and attributes used by the dmarc rule.
    """
    class Name:
        # Mock answer.name.
        def __init__(self, name='_dmarc.example.biz.'):
            self.name = name

        def to_text(self):
            return self.name

    class Item:
        # Mock answer.items.
        def __init__(self, rdata=rdata, cname='_dmarc.example.com.'):
            self.strings = [rdata]
            # For CNAMEs.
            self.target = Name(cname)

    class Ans_e:
        # Mock answer element.
        def __init__(
                self,
                rtype=rtype,
                rdata=rdata,
                cname='_dmarc.example.com.',
                name='_dmarc.example.biz.'):
            self.rdtype = rtype
            self.items = [Item(rdata, cname)]
            self.name = Name(name)

    class Answer:
        # Mock answer.
        def __init__(self):
            if cloop:
                self.answer = [
                    Ans_e(
                        rtype=CNAME,
                        name='_dmarc.example.biz.',
                        cname='_dmarc.example.org.'
                        ),
                    Ans_e(
                        rtype=CNAME,
                        name='_dmarc.example.org.',
                        cname='_dmarc.example.biz.'
                        ),
                    Ans_e(
                        rtype=TXT,
                        name='_dmarc.example.org.',
                        rdata=b'v=DMARC1; p=reject;'
                        ),
                    ]
            elif cmult:
                self.answer = [
                    Ans_e(
                        rtype=CNAME,
                        name='_dmarc.example.biz.',
                        cname='_dmarc.example.net.'
                        ),
                    Ans_e(
                        rtype=CNAME,
                        name='_dmarc.example.net.',
                        cname='_dmarc.example.com.'
                        ),
                    Ans_e(
                        rtype=TXT,
                        name='_dmarc.example.com.',
                        rdata=b'v=DMARC1; p=quarantine;'
                        ),
                    ]
            elif cmiss:
                self.answer = [
                    Ans_e(
                        rtype=CNAME,
                        name='_dmarc.example.biz.',
                        cname='_dmarc.example.net.'
                        ),
                    Ans_e(
                        rtype=TXT,
                        name='_dmarc.example.biz.',
                        rdata=b'v=DMARC1; p=reject;'
                        ),
                    ]
            elif rmult:
                self.answer = [Ans_e(), Ans_e(rdata=b'v=DMARC1; p=none;')]
            else:
                self.answer = [Ans_e()]

    class Resolver:
        # Mock dns.resolver.Resolver class.
        def query(self, domain, data_type):
            if data_type != TXT:
                raise NoAnswer
            dparts = domain.split('.')
            if len(dparts) < 3:
                raise NXDOMAIN
            if len(dparts) > 3:
                raise NoAnswer
            if dparts[0] != '_dmarc':
                raise NoAnswer
            if dparts[2] == 'info':
                raise DNSException('no internet')
            if dparts[1] != 'example' or dparts[2] != 'biz':
                raise NXDOMAIN
            self.response = Answer()
            return self
    patcher = patch('dns.resolver.Resolver', Resolver)
    return patcher


@public
def use_test_organizational_data():
    # Point the organizational URL to our test data.
    path = resource_filename('mailman.rules.tests.data', 'org_domain.txt')
    url = 'file:///{}'.format(path)
    return configuration('dmarc', org_domain_data_url=url)


class TestDMARCRules(TestCase):
    """Test organizational domain determination."""

    layer = ConfigLayer

    def setUp(self):
        self.resources = ExitStack()
        self.addCleanup(self.resources.close)
        # Make sure every test has a clean cache.
        self.cache = {}
        self.resources.enter_context(
            patch('mailman.rules.dmarc.suffix_cache', self.cache))
        self.resources.enter_context(use_test_organizational_data())

    def test_no_data_for_domain(self):
        self.assertEqual(
            dmarc.get_organizational_domain('sub.dom.example.nxtld'),
            'example.nxtld')

    def test_domain_with_wild_card(self):
        self.assertEqual(
            dmarc.get_organizational_domain('ssub.sub.foo.kobe.jp'),
            'sub.foo.kobe.jp')

    def test_exception_to_wild_card(self):
        self.assertEqual(
            dmarc.get_organizational_domain('ssub.sub.city.kobe.jp'),
            'city.kobe.jp')

    def test_no_at_sign_in_from_address(self):
        # If there's no @ sign in the From: address, the rule can't hit.
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne
To: ant@example.com

""")
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver():
            self.assertFalse(rule.check(mlist, msg, {}))

    def test_dmarc_dns_exception(self):
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne@example.info
To: ant@example.com

""")
        mark = LogFileMark('mailman.error')
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver():
            self.assertFalse(rule.check(mlist, msg, {}))
        line = mark.readline()
        self.assertEqual(
            line[-144:],
            'DNSException: Unable to query DMARC policy for '
            'anne@example.info (_dmarc.example.info). '
            'Abstract base class shared by all dnspython exceptions.\n')

    def test_cname_wrong_txt_name(self):
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne@example.biz
To: ant@example.com

""")
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver(cmiss=True):
            self.assertFalse(rule.check(mlist, msg, {}))

    def test_domain_with_subdomain_policy(self):
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne@example.biz
To: ant@example.com

""")
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver(rdata=b'v=DMARC1; sp=quarantine;'):
            self.assertFalse(rule.check(mlist, msg, {}))

    def test_org_domain_with_subdomain_policy(self):
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne@sub.domain.example.biz
To: ant@example.com

""")
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver(rdata=b'v=DMARC1; sp=quarantine;'):
            self.assertTrue(rule.check(mlist, msg, {}))

    def test_wrong_dmarc_version(self):
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne@example.biz
To: ant@example.com

""")
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver(rdata=b'v=DMARC01; p=reject;'):
            self.assertFalse(rule.check(mlist, msg, {}))

    def test_multiple_records(self):
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne@example.biz
To: ant@example.com

""")
        mark = LogFileMark('mailman.error')
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver(rmult=True):
            self.assertTrue(rule.check(mlist, msg, {}))
        line = mark.readline()
        self.assertEqual(
            line[-85:],
            'RRset of TXT records for _dmarc.example.biz has 2 '
            'v=DMARC1 entries; testing them all\n')

    def test_multiple_cnames(self):
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne@example.biz
To: ant@example.com

""")
        mark = LogFileMark('mailman.vette')
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver(cmult=True):
            self.assertTrue(rule.check(mlist, msg, {}))
        line = mark.readline()
        self.assertEqual(
            line[-128:],
            'ant: DMARC lookup for anne@example.biz (_dmarc.example.biz) '
            'found p=quarantine in _dmarc.example.com. = v=DMARC1; '
            'p=quarantine;\n')

    def test_looping_cnames(self):
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne@example.biz
To: ant@example.com

""")
        mark = LogFileMark('mailman.vette')
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver(cloop=True):
            self.assertTrue(rule.check(mlist, msg, {}))
        line = mark.readline()
        self.assertEqual(
            line[-120:],
            'ant: DMARC lookup for anne@example.biz (_dmarc.example.biz) '
            'found p=reject in _dmarc.example.org. = v=DMARC1; p=reject;\n')

    def test_no_policy(self):
        mlist = create_list('ant@example.com')
        # Use action reject.  The rule only hits on reject and discard.
        mlist.dmarc_mitigate_action = DMARCMitigateAction.reject
        msg = mfs("""\
From: anne@example.biz
To: ant@example.com

""")
        rule = dmarc.DMARCMitigation()
        with get_dns_resolver(rdata=b'v=DMARC1; pct=100;'):
            self.assertFalse(rule.check(mlist, msg, {}))

    def test_parser(self):
        data_file = resource_filename(
            'mailman.rules.tests.data', 'org_domain.txt')
        dmarc.parse_suffix_list(data_file)
        # There is no entry for example.biz because that line starts with
        # whitespace.
        self.assertNotIn('biz.example', self.cache)
        # The file had !city.kobe.jp so the flag says there's an exception.
        self.assertTrue(self.cache['jp.kobe.city'])
        # The file had *.kobe.jp so there's no exception.
        self.assertFalse(self.cache['jp.kobe.*'])


# New in Python 3.5.
try:
    from http import HTTPStatus
except ImportError:
    class HTTPStatus:
        FORBIDDEN = 403
        NOT_FOUND = 404
        OK = 200


# We need a web server to vend non-mailman: urls.
class TestableHandler(BaseHTTPRequestHandler):
    # Be quiet.
    def log_request(*args, **kws):
        pass

    log_error = log_request

    def do_GET(self):
        if self.path == '/public_suffix_list.dat':
            self.send_response(HTTPStatus.OK)
            self.send_header('Content-Type', 'UTF-8')
            self.end_headers()
            self.wfile.write(b'abc')
        else:
            self.send_error(HTTPStatus.NOT_FOUND)


class HTTPLayer(ConfigLayer):
    httpd = None

    @classmethod
    def setUp(cls):
        assert cls.httpd is None, 'Layer already set up'
        cls.httpd = HTTPServer(('localhost', 8180), TestableHandler)
        cls._thread = threading.Thread(target=cls.httpd.serve_forever)
        cls._thread.daemon = True
        cls._thread.start()
        wait_for_webservice('localhost', 8180)

    @classmethod
    def tearDown(cls):
        assert cls.httpd is not None, 'Layer not set up'
        cls.httpd.shutdown()
        cls.httpd.server_close()
        cls._thread.join()


class TestSuffixList(TestCase):
    layer = HTTPLayer

    def test_cached_copy_is_good(self):
        cache_path = os.path.join(config.VAR_DIR, dmarc.LOCAL_FILE_NAME)
        with open(cache_path, 'w', encoding='utf-8') as fp:
            print('xyz', end='', file=fp)
        # The cache expires a day from now.
        expires = (now() + timedelta(days=1)).timestamp()
        os.utime(cache_path, (expires, expires))
        new_path = dmarc.ensure_current_suffix_list()
        self.assertEqual(cache_path, new_path)
        with open(cache_path, 'r', encoding='utf-8') as fp:
            contents = fp.read()
        self.assertEqual(contents, 'xyz')
        self.assertEqual(os.stat(new_path).st_mtime, expires)

    @configuration(
        'dmarc',
        org_domain_data_url='http://localhost:8180/public_suffix_list.dat')
    def test_cached_copy_is_expired(self):
        cache_path = os.path.join(config.VAR_DIR, dmarc.LOCAL_FILE_NAME)
        with open(cache_path, 'w', encoding='utf-8') as fp:
            print('xyz', end='', file=fp)
        # Expire the cache file.  That way the current cached file will be
        # invalid and a new one will be downloaded.
        expires = (now() - timedelta(days=1)).timestamp()
        os.utime(cache_path, (expires, expires))
        new_path = dmarc.ensure_current_suffix_list()
        self.assertEqual(cache_path, new_path)
        with open(cache_path, 'r', encoding='utf-8') as fp:
            contents = fp.read()
        self.assertEqual(contents, 'abc')
        self.assertEqual(
            os.stat(new_path).st_mtime,
            (now() + as_timedelta(config.dmarc.cache_lifetime)).timestamp())

    @configuration(
        'dmarc',
        org_domain_data_url='http://localhost:8180/public_suffix_list.dat')
    def test_cached_copy_is_missing(self):
        cache_path = os.path.join(config.VAR_DIR, dmarc.LOCAL_FILE_NAME)
        self.assertFalse(os.path.exists(cache_path))
        new_path = dmarc.ensure_current_suffix_list()
        self.assertEqual(cache_path, new_path)
        with open(cache_path, 'r', encoding='utf-8') as fp:
            contents = fp.read()
        self.assertEqual(contents, 'abc')
        self.assertEqual(
            os.stat(new_path).st_mtime,
            (now() + as_timedelta(config.dmarc.cache_lifetime)).timestamp())

    @configuration(
        'dmarc',
        org_domain_data_url='http://localhost:8180/public_suffix_list.err')
    def test_cached_copy_is_missing_download_404s(self):
        # There's no cached file and we'll get a 404 with the .err file so
        # we'll have to fall back to our internal copy.
        cache_path = os.path.join(config.VAR_DIR, dmarc.LOCAL_FILE_NAME)
        self.assertFalse(os.path.exists(cache_path))
        new_path = dmarc.ensure_current_suffix_list()
        self.assertEqual(cache_path, new_path)
        with open(cache_path, 'r', encoding='utf-8') as fp:
            contents = fp.read()
        # The contents is *not* equal to our dummy test data, but don't tie it
        # too closely to the in-tree file contents since that might change
        # when and if we update that.
        self.assertNotEqual(contents, 'abc')
        self.assertEqual(
            os.stat(new_path).st_mtime,
            (now() + as_timedelta(config.dmarc.cache_lifetime)).timestamp())

    @configuration(
        'dmarc',
        org_domain_data_url='http://localhost:8180/public_suffix_list.err')
    def test_cached_copy_is_expired_download_404s(self):
        # Because the cached copy is out of date, we try to download the new
        # version.  But that 404s so we end up continuing to use the cached
        # copy.
        cache_path = os.path.join(config.VAR_DIR, dmarc.LOCAL_FILE_NAME)
        with open(cache_path, 'w', encoding='utf-8') as fp:
            print('xyz', end='', file=fp)
        # Expire the cache file.  That way the current cached file will be
        # invalid and a new one will be downloaded.
        expires = (now() - timedelta(days=1)).timestamp()
        os.utime(cache_path, (expires, expires))
        new_path = dmarc.ensure_current_suffix_list()
        self.assertEqual(cache_path, new_path)
        with open(cache_path, 'r', encoding='utf-8') as fp:
            contents = fp.read()
        # The contents are from the cached file.
        self.assertEqual(contents, 'xyz')
        # The cached file timestamp doesn't change.
        self.assertEqual(os.stat(new_path).st_mtime, expires)