import os
import re
import time
import uuid

import defusedxml.ElementTree as ElementTree

import pytest

import testinfra.utils.ansible_runner


testinfra_hosts = testinfra.utils.ansible_runner.AnsibleRunner(
    os.environ['MOLECULE_INVENTORY_FILE']).get_hosts('parameters-optional')


def test_mailname_file_content(host):
    """
    Tests the system mail name file content.
    """

    mailname = host.file('/etc/mailname')
    hostname = host.run('hostname').stdout.strip()

    assert mailname.content_string == hostname


def test_postfix_main_cf_file_content(host):
    """
    Tests if the Postfix main configuration file content is correct.
    """

    distribution_release = host.ansible("setup")["ansible_facts"]["ansible_distribution_release"]
    host_variables = host.ansible.get_variables()
    allow_relay_from_ip = host_variables["release_based_smtp_allow_relay_from"][distribution_release]

    hostname = host.run('hostname').stdout.strip()

    config = host.file('/etc/postfix/main.cf')
    config_lines = config.content_string.split("\n")

    assert "myhostname = %s" % hostname in config_lines
    assert "mydestination = %s, %s, localhost.localdomain, localhost" % (hostname, hostname) in config_lines
    assert "mynetworks = 127.0.0.0/8 [::ffff:127.0.0.0]/104 [::1]/128 %s" % allow_relay_from_ip in config_lines
    assert "smtpd_tls_cert_file = /etc/ssl/certs/%s_smtp.pem" % hostname in config_lines
    assert "smtpd_tls_key_file = /etc/ssl/private/%s_smtp.key" % hostname in config_lines
    assert "  reject_rbl_client bl.spamcop.net" in config_lines
    assert "  reject_rbl_client zen.spamhaus.org" in config_lines
    assert "smtp_host_lookup = dns, native" in config_lines


def test_local_aliases(host):
    """
    Tests if local aliases are configured correctly.
    """

    message_id = "%s@localhost" % str(uuid.uuid4())

    send = host.run('swaks --header %s --suppress-data --to root@localhost', "Message-Id: <%s>" % message_id)
    time.sleep(1)
    assert send.rc == 0

    with host.sudo():
        mail_log = host.file('/var/log/mail.log')
        pattern = r"dovecot: lda\(john.doe@domain1\)<\d+><.+?>: msgid=<%s>: saved mail to INBOX" % message_id
        assert re.search(pattern, mail_log.content_string) is not None


def test_dovecot_mailbox_directories(host):
    """
    Tests if mailbox directories are created correctly.
    """

    # Deliver two mails in order to make sure the directory structure is
    # created.
    send = host.run('swaks --suppress-data --to john.doe@domain1 --server localhost')
    assert send.rc == 0
    send = host.run('swaks --suppress-data --to jane.doe@domain2 --server localhost')
    assert send.rc == 0

    with host.sudo():

        for directory_path in ["/var/virtmail/domain1",
                               "/var/virtmail/domain1/john.doe",
                               "/var/virtmail/domain1/john.doe/Maildir",
                               "/var/virtmail/domain2",
                               "/var/virtmail/domain2/jane.doe",
                               "/var/virtmail/domain2/jane.doe/Maildir"]:

            directory = host.file(directory_path)

            assert directory.is_directory
            assert directory.user == "virtmail"
            assert directory.group == "virtmail"
            assert directory.mode == 0o700


def test_mail_owner(host):
    """
    Tests creation of mail owner group and user.
    """

    group = host.group("virtmail")
    assert group.exists
    assert group.gid == 5000

    user = host.user("virtmail")
    assert user.exists
    assert user.uid == 5000
    assert user.home == "/var/virtmail"
    assert user.group == "virtmail"
    assert user.groups == ["virtmail"]


@pytest.mark.parametrize("port", [
    143,
    993,
    587,
])
def test_imap_and_smtp_submission_tls_version_and_ciphers(host, port):
    """
    Tests if the correct TLS version and ciphers have been enabled for
    IMAP and SMTP submission.
    """

    distribution_release = host.ansible("setup")["ansible_facts"]["ansible_distribution_release"]

    if distribution_release == "bullseye":
        expected_tls_versions = ["TLSv1.1", "TLSv1.2"]
        expected_tls_ciphers = [
            "TLS_DHE_RSA_WITH_AES_128_CBC_SHA256",
            "TLS_DHE_RSA_WITH_AES_128_GCM_SHA256",
            "TLS_DHE_RSA_WITH_AES_256_CBC_SHA256",
            "TLS_DHE_RSA_WITH_AES_256_GCM_SHA384",
            "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA",
            "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256",
            "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
            "TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384",
            "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384",
        ]
    else:
        expected_tls_versions = ["TLSv1.1", "TLSv1.2", "TLSv1.3"]
        expected_tls_ciphers = [
            "TLS_AKE_WITH_AES_128_GCM_SHA256",
            "TLS_AKE_WITH_AES_256_GCM_SHA384",
            "TLS_AKE_WITH_CHACHA20_POLY1305_SHA256",
            "TLS_DHE_RSA_WITH_AES_128_CBC_SHA256",
            "TLS_DHE_RSA_WITH_AES_128_GCM_SHA256",
            "TLS_DHE_RSA_WITH_AES_256_CBC_SHA256",
            "TLS_DHE_RSA_WITH_AES_256_GCM_SHA384",
            "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA",
            "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256",
            "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
            "TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384",
            "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384",
        ]

    # Run the nmap scanner against the server, and fetch the results.
    nmap = host.run("nmap -sV --script ssl-enum-ciphers -p %s localhost -oX /tmp/report.xml", str(port))
    assert nmap.rc == 0
    report_content = host.file('/tmp/report.xml').content_string

    report_root = ElementTree.fromstring(report_content)

    tls_versions = []
    tls_ciphers = set()

    for child in report_root.findall("./host/ports/port/script/table"):
        tls_versions.append(child.attrib['key'])

    for child in report_root.findall(".//table[@key='ciphers']/table/elem[@key='name']"):
        tls_ciphers.add(child.text)

    tls_versions.sort()
    tls_ciphers = sorted(list(tls_ciphers))

    assert tls_versions == expected_tls_versions
    assert tls_ciphers == expected_tls_ciphers


def test_dovecot_postmaster(host):
    """
    Tests if Dovecot postmaster has been correctly configured.
    """

    with host.sudo():

        config = host.run("doveadm config")

        assert config.rc == 0
        assert "  postmaster_address = webmaster@parameters-optional" in config.stdout


def test_imap_max_user_connections_per_ip(host):
    """
    Tests if Dovecot per-user connection limit has been set-up correctly.
    """

    with host.sudo():

        config = host.run("doveadm config")

        assert config.rc == 0
        assert "  mail_max_userip_connections = 2" in config.stdout


def test_sieve_tls_configuration(host):
    """
    Tests TLS configuration for SIEVE in Dovecot
    """

    # @TODO: Currently not possible to test since nmap and openssl
    # s_client do not support STARTTLS for Sieve.
    pass


def test_mail_message_size_limit(host):
    """
    Tests if the mail message size limit advertised by the SMTP server
    is correct.
    """

    capabilities = host.run("(echo 'ehlo localhost' && sleep 2) | telnet localhost 25")
    begin = capabilities.stdout.find("250-SIZE")
    end = capabilities.stdout.find("\n", begin)
    mail_message_size_limit = capabilities.stdout[begin:end]

    assert mail_message_size_limit == "250-SIZE 20480001"


def test_smtp_additional_configuration_present_in_file(host):
    """
    Tests if additional SMTP server configuration has been applied
    against the configuration file.
    """

    config = host.file("/etc/postfix/main.cf")

    assert "mail_name = MySMTP" in config.content_string
    assert "smtp_skip_5xx_greeting = no" in config.content_string


def test_smtp_additional_configuration_active(host):
    """
    Tests if additional SMTP server configuration has been applied
    against the running server.
    """

    command = host.run('swaks --quit-after BANNER --to root@localhost')

    assert "ESMTP MySMTP (Debian/GNU)" in command.stdout