import os
import re

import pytest

import testinfra.utils.ansible_runner


testinfra_hosts = testinfra.utils.ansible_runner.AnsibleRunner(
    os.environ['MOLECULE_INVENTORY_FILE']).get_hosts('client-relay-forbidden')


def test_open_relay(host):
    """
    Tests if mail server behaves as open relay.
    """

    no_recipients_accepted = 24

    send = host.run('swaks --suppress-data --to root@client1 --server parameters-mandatory')
    assert send.rc == no_recipients_accepted
    assert "Relay access denied" in send.stdout

    send = host.run('swaks --suppress-data --to root@client1 --server parameters-optional')
    assert send.rc == no_recipients_accepted
    assert "Relay access denied" in send.stdout

    send = host.run('swaks --port 27 --suppress-data --to root@client1 --server parameters-mandatory')
    assert send.rc == no_recipients_accepted
    assert "Relay access denied" in send.stdout

    send = host.run('swaks --port 27 --suppress-data --to root@client1 --server parameters-optional')
    assert send.rc == no_recipients_accepted
    assert "Relay access denied" in send.stdout


def test_mail_delivery(host):
    """
    Tests if mails can be delivered to valid accounts. Has to be run on client
    with no unauthenticated relay permissions.
    """

    no_recipients_accepted = 24

    # Valid accounts.
    send = host.run('swaks --suppress-data --to john.doe@domain1 --server parameters-mandatory')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks --suppress-data --to john.doe@domain1 --server parameters-optional')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks --suppress-data --to jane.doe@domain2 --server parameters-mandatory')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks --suppress-data --to jane.doe@domain2 --server parameters-optional')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    # Invalid accounts.
    send = host.run('swaks --suppress-data --to john.doe@domain2 --server parameters-mandatory')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout

    send = host.run('swaks --suppress-data --to john.doe@domain2 --server parameters-optional')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout

    send = host.run('swaks --suppress-data --to jane.doe@domain1 --server parameters-mandatory')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout

    send = host.run('swaks --suppress-data --to jane.doe@domain1 --server parameters-optional')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout

    # Test for valid mail address that's not allowed by LDAP group membership.
    send = host.run('swaks --suppress-data --to nomail@domain1 --server parameters-mandatory')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout

    send = host.run('swaks --suppress-data --to nomail@domain1 --server parameters-optional')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout

    # Valid aliases.
    send = host.run('swaks --suppress-data --to postmaster@domain1 --server parameters-mandatory')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks --suppress-data --to postmaster@domain1 --server parameters-optional')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks --suppress-data --to webmaster@domain2 --server parameters-mandatory')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks --suppress-data --to webmaster@domain2 --server parameters-optional')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    # Invalid aliases.
    send = host.run('swaks --suppress-data --to postmaster@domain2 --server parameters-mandatory')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout

    send = host.run('swaks --suppress-data --to postmaster@domain2 --server parameters-optional')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout

    send = host.run('swaks --suppress-data --to webmaster@domain1 --server parameters-mandatory')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout

    send = host.run('swaks --suppress-data --to webmaster@domain1 --server parameters-optional')
    assert send.rc == no_recipients_accepted
    assert "Recipient address rejected: User unknown in virtual mailbox table" in send.stdout


def test_smtp_authentication(host):
    """
    Tests if SMTP authentication works via TLS and allows sending mails to
    anywhere.
    """

    send = host.run('swaks -tls --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-mandatory')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks -tls --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-optional')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout


def test_smtp_authentication_with_alias_sender(host):
    """
    Tests if SMTP authentication works via TLS and allows sending mails to
    anywhere while using sender alias.
    """

    send = host.run('swaks -tls --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from postmaster@domain1 --to root@client1 --server parameters-mandatory')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks -tls --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from postmaster@domain1 --to root@client1 --server parameters-optional')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout


def test_smtp_authentication_requires_tls(host):
    """
    Tests if SMTP authentication requires TLS.
    """

    auth_error = 28

    send = host.run('swaks --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-mandatory')
    assert send.rc == auth_error
    assert "Host did not advertise authentication" in send.stderr

    send = host.run('swaks --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-optional')
    assert send.rc == auth_error
    assert "Host did not advertise authentication" in send.stderr


def test_smtp_authentication_requires_submission_port(host):
    """
    Tests if SMTP authentication cannot be done on regular SMTP port.
    """

    auth_error = 28

    send = host.run('swaks --port 25 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-mandatory')
    assert send.rc == auth_error
    assert "Host did not advertise authentication" in send.stderr

    send = host.run('swaks -tls --port 25 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-mandatory')
    assert send.rc == auth_error
    assert "Host did not advertise authentication" in send.stderr

    send = host.run('swaks --port 25 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-optional')
    assert send.rc == auth_error
    assert "Host did not advertise authentication" in send.stderr

    send = host.run('swaks -tls --port 25 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-optional')
    assert send.rc == auth_error
    assert "Host did not advertise authentication" in send.stderr


def test_dovecot_inbox_separator(host):
    """
    Tests if inbox separator has been configured correctly.
    """

    pattern_slash_separator = re.compile(r'WARNING:imap-cli:Ignoring "LIST" response part : \([^)]*\) "/" INBOX')
    pattern_dot_separator = re.compile(r'WARNING:imap-cli:Ignoring "LIST" response part : \([^)]*\) "\." INBOX')

    status = host.run("imapcli status -c ~/imapcli-parameters-mandatory-john_doe.conf")
    assert pattern_slash_separator.search(status.stdout) is not None

    status = host.run("imapcli status -c ~/imapcli-parameters-optional-john_doe.conf")
    assert pattern_dot_separator.search(status.stdout) is not None


@pytest.mark.parametrize("server", [
    "parameters-mandatory",
    "parameters-optional"
])
def test_imap_authentication_requires_tls(host, server):
    """
    Tests if IMAP authentication requires TLS.
    """

    capabilities_command = "a0001 CAPABILITY\na0002 LOGOUT"

    # No TLS.
    command = host.run("echo %s | nc %s 143", capabilities_command, server)
    assert command.rc == 0
    assert "LOGINDISABLED" in command.stdout

    # STARTTLS.
    command = host.run("echo %s | openssl s_client -quiet -connect %s:143 -starttls imap", capabilities_command, server)
    assert command.rc == 0
    assert "LOGINDISABLED" not in command.stdout

    # TLS.
    command = host.run("echo %s | openssl s_client -quiet -connect %s:993", capabilities_command, server)
    assert command.rc == 0
    assert "LOGINDISABLED" not in command.stdout


def test_sieve_authentication_requires_tls(host):
    """
    Tests if SIEVE authentication requires TLS.
    """

    # No TLS.
    command = host.run("echo 'LOGOUT' | nc parameters-mandatory 4190")
    assert command.rc == 0
    assert "PLAIN LOGIN" not in command.stdout

    command = host.run("echo 'LOGOUT' | nc parameters-optional 4190")
    assert command.rc == 0
    assert "PLAIN LOGIN" not in command.stdout

    # STARTTLS
    # @TODO: In case of failed login (authentication rejected),
    #        sieve-connect will return error code 255. However,
    #        internally Testinfra treats error code as if it is a
    #        signal from the SSH (over which the command is being run)
    #        that the SSH itself has failed, and throws a runtime
    #        exception. The use of || //bin/false is a workaround for
    #        now. It would be a good idea to get in contact with
    #        Testinfra maintainer and raise an issue around
    #        this. There are more similar uses of sieve-connect in the
    #        test files, don't forget to update those as well if some
    #        kind of better solution pops-up.
    command = host.run("echo 'johnpassword' | sieve-connect -u john.doe@domain1 --password 0 --server parameters-mandatory --port 4190 --list || /bin/false")
    assert command.rc == 0

    command = host.run("echo 'johnpassword' | sieve-connect -u john.doe@domain1 --password 0 --server parameters-optional --port 4190 --list || /bin/false")
    assert command.rc == 0


@pytest.mark.parametrize("server", [
    "parameters-mandatory",
    "parameters-optional"
])
@pytest.mark.parametrize("port", [
    25,
    26,
    27,
    587,
    143,
    993,
    4190
])
def test_connectivity(host, server, port):
    """
    Tests connectivity to the mail server (ports that should be reachable).
    """

    with host.sudo():
        ping = host.run('hping3 -S -p %s -c 1 %s', str(port), server)
        assert ping.rc == 0


def test_port_forwarding(host):
    """
    Tests if port forwarding is set-up correctly for additional SMTP and
    submission ports.
    """

    # Regular SMTP.
    send = host.run('swaks -tls --port 27 --to john.doe@domain1 --server parameters-mandatory')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks -tls --port 27 --to john.doe@domain1 --server parameters-optional')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    # Submission port.
    send = host.run('swaks -tls --port 26 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-mandatory')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout

    send = host.run('swaks -tls --port 26 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from john.doe@domain1 --to root@client1 --server parameters-optional')
    assert send.rc == 0
    assert "Ok: queued as" in send.stdout


def test_dovecot_sieve(host):
    """
    Tests if Sieve service is available.
    """

    # Test valid users.
    command = host.run('echo johnpassword | sieve-connect --list -s parameters-mandatory -p 4190 -u john.doe@domain1 --password 0 || /bin/false')
    assert command.rc == 0

    command = host.run('echo janepassword | sieve-connect --list -s parameters-optional -p 4190 -u jane.doe@domain2 --password 0 || /bin/false')
    assert command.rc == 0

    # Test invalid users.
    command = host.run('echo johnpassword | sieve-connect --list -s parameters-mandatory -p 4190 -u john.doe@domain2 --password 0 || /bin/false')
    assert command.rc != 0
    assert "Authentication refused by server" in command.stderr

    command = host.run('echo janepassword | sieve-connect --list -s parameters-optional -p 4190 -u jane.doe@domain1 --password 0 || /bin/false')
    assert command.rc != 0
    assert "Authentication refused by server" in command.stderr


def test_smtp_sender_forging(host):
    """
    Tests if SMTP sender forging is possible.
    """

    send = host.run('swaks -tls --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from jane.doe@domain2 --to root@client1 --server parameters-mandatory')
    assert send.rc == 24
    assert "Sender address rejected: not owned by user john.doe@domain1" in send.stdout

    send = host.run('swaks -tls --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from jane.doe@domain2 --to root@client1 --server parameters-optional')
    assert send.rc == 24
    assert "Sender address rejected: not owned by user john.doe@domain1" in send.stdout

    send = host.run('swaks -tls --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from webmaster@domain2 --to root@client1 --server parameters-mandatory')
    assert send.rc == 24
    assert "Sender address rejected: not owned by user john.doe@domain1" in send.stdout

    send = host.run('swaks -tls --port 587 --auth-user john.doe@domain1 --auth-password johnpassword '
                    '--from webmaster@domain2 --to root@client1 --server parameters-optional')
    assert send.rc == 24
    assert "Sender address rejected: not owned by user john.doe@domain1" in send.stdout