import socket

import paramiko

import testinfra.utils.ansible_runner

testinfra_hosts = testinfra.utils.ansible_runner.AnsibleRunner(

    '.molecule/ansible_inventory').get_hosts('all')

def test_installed_software(Package):

    """

    Tests if the required packages have been installed.

    """

    assert Package('duplicity').is_installed

    assert Package('duply').is_installed

def test_backup_directory(File, Sudo):

    """

    Tests if the backup directory has been set-up correctly.

    """

    with Sudo():

        backup_directory = File('/srv/backups')

        assert backup_directory.is_directory

        assert backup_directory.user == 'root'

        assert backup_directory.group == 'root'

        assert backup_directory.mode == 0o751

def test_regular_ssh_server_configuration(File, Sudo):

    """

    Tests if the default SSH server has been configured correctly (to prevent

    access to it via backup users).

    """

    with Sudo():

        assert "DenyGroups backup" in File('/etc/ssh/sshd_config').content

def test_backup_ssh_server_configuration_directory(File, Sudo):

    """

    Tests if the backup SSH server configuration directory has been created

    correctly.

    """

    with Sudo():

        backup_ssh_server_directory = File('/etc/ssh-backup')

        assert backup_ssh_server_directory.is_directory

        assert backup_ssh_server_directory.user == 'root'

        assert backup_ssh_server_directory.group == 'root'

        assert backup_ssh_server_directory.mode == 0o700

def test_backup_ssh_server_service_configuration(File):

    """

    Tests if the backup SSH server service configuration file has been set-up

    correctly.

    """

    config_file = File('/etc/default/ssh-backup')

    assert config_file.is_file

    assert config_file.user == 'root'

    assert config_file.group == 'root'

    assert config_file.mode == 0o644

    assert 'SSHD_OPTS="-f /etc/ssh-backup/sshd_config"' in config_file.content

def test_backup_ssh_server_configuration(File, Sudo):

    """

    Tests if the backup SSH server configuration file has been set-up correctly.

    """

    with Sudo():

        config_file = File('/etc/ssh-backup/sshd_config')

        assert config_file.is_file

        assert config_file.user == 'root'

        assert config_file.group == 'root'

        assert config_file.mode == 0o600

        assert "AllowGroups backup" in config_file.content

        assert "ChrootDirectory %h" in config_file.content

        assert "ForceCommand internal-sftp" in config_file.content

        assert "Subsystem sftp internal-sftp" in config_file.content

        assert "PasswordAuthentication no" in config_file.content

        assert "PubkeyAuthentication yes" in config_file.content

        assert "PermitRootLogin no" in config_file.content

        assert "HostKey /etc/ssh-backup/ssh_host_rsa_key" in config_file.content

        assert "HostKey /etc/ssh-backup/ssh_host_dsa_key" in config_file.content

        assert "HostKey /etc/ssh-backup/ssh_host_ecdsa_key" in config_file.content

        assert "HostKey /etc/ssh-backup/ssh_host_ed25519_key" in config_file.content

def test_backup_ssh_server_keys(File, Sudo):

    """

    Tests if the backup SSH server private keys have been deployed correctly.

    """

    with Sudo():

        dsa = File('/etc/ssh-backup/ssh_host_dsa_key')

        assert dsa.is_file

        assert dsa.user == 'root'

        assert dsa.group == 'root'

        assert dsa.mode == 0o600

        assert dsa.content == open('tests/data/ssh/server_dsa', 'r').read()

        rsa = File('/etc/ssh-backup/ssh_host_rsa_key')

        assert rsa.is_file

        assert rsa.user == 'root'

        assert rsa.group == 'root'

        assert rsa.mode == 0o600

        assert rsa.content == open('tests/data/ssh/server_rsa', 'r').read()

        ed25519 = File('/etc/ssh-backup/ssh_host_ed25519_key')

        assert ed25519.is_file

        assert ed25519.user == 'root'

        assert ed25519.group == 'root'

        assert ed25519.mode == 0o600

        assert ed25519.content == open('tests/data/ssh/server_ed25519', 'r').read()

        ecdsa = File('/etc/ssh-backup/ssh_host_ecdsa_key')

        assert ecdsa.is_file

        assert ecdsa.user == 'root'

        assert ecdsa.group == 'root'

        assert ecdsa.mode == 0o600

        assert ecdsa.content == open('tests/data/ssh/server_ecdsa', 'r').read()

def test_backup_ssh_server_systemd_service(File):

    """

    Tests if the backup SSH server systemd service file has been deployed

    correctly.

    """

    service_file = File('/etc/systemd/system/ssh-backup.service')

    assert service_file.is_file

    assert service_file.user == 'root'

    assert service_file.group == 'root'

    assert service_file.mode == 0o644

    assert "EnvironmentFile=-/etc/default/ssh-backup" in service_file.content

def test_backup_ssh_server_service(Service, Socket, Sudo):

    """

    Tests if the backup SSH server service is running and listening on correct

    port.

    """

    with Sudo():

        service = Service('ssh-backup')

        assert service.is_running

        assert service.is_enabled

        assert Socket('tcp://0.0.0.0:2222').is_listening

def test_backup_ssh_server_login_mechanisms():

    """

    Tests available SSH login mechanisms (should be just public key).

    """

    sock = socket.socket()

    sock.connect(('10.31.127.11', 2222))

    transport = paramiko.transport.Transport(sock)

    transport.connect()

    try:

        transport.auth_none('')

    except paramiko.transport.BadAuthenticationType, err:

        assert err.allowed_types == ['publickey']

import testinfra.utils.ansible_runner

testinfra_hosts = testinfra.utils.ansible_runner.AnsibleRunner(

    '.molecule/ansible_inventory').get_hosts('parameters-mandatory')

def test_firewall_configuration(File, Sudo):

    """

    Tests if the firewall configuration file has been deployed correctly.

    """

    with Sudo():

        firewall_config = File('/etc/ferm/conf.d/40-backup.conf')

        assert firewall_config.is_file

        assert firewall_config.user == 'root'

        assert firewall_config.group == 'root'

        assert firewall_config.mode == 0o640

        assert firewall_config.content == ""

import os

import testinfra.utils.ansible_runner

testinfra_hosts = testinfra.utils.ansible_runner.AnsibleRunner(

    '.molecule/ansible_inventory').get_hosts('parameters-optional')

def test_backup_client_users_and_groups(Group, Sudo, User):

    """

    Tests if the system groups and users for backup clients have been set-up

    correctly.

    """

    with Sudo():

        client1_group = Group('bak-client1_backup')

        assert client1_group.exists

        assert client1_group.gid < 1000

        client1_user = User('bak-client1_backup')

        assert client1_user.exists

        assert client1_user.group == 'bak-client1_backup'

        assert sorted(client1_user.groups) == sorted(['bak-client1_backup', 'backup'])

        assert client1_user.home == '/srv/backups/client1.backup'

        assert client1_user.uid < 1000

        assert client1_user.password == '!'

        client2_group = Group('bak-client2-backup')

        assert client2_group.exists

        assert client2_group.gid == 5001

        client2_user = User('bak-client2-backup')

        assert client2_user.exists

        assert client2_user.group == 'bak-client2-backup'

        assert sorted(client2_user.groups) == sorted(['bak-client2-backup', 'backup'])

        assert client2_user.home == '/srv/backups/client2-backup'

        assert client2_user.uid == 5001

        assert client2_user.password == '!'

def test_backup_client_home_directories(File, Sudo, User):

    """

    Tests if the home directory structure has been set-up correctly for the

    backup client system user.

    """

    with Sudo():

        client1_user = User('bak-client1_backup')

        client1_user_home = File(client1_user.home)

        assert client1_user_home.is_directory

        assert client1_user_home.user == 'root'

        assert client1_user_home.group == 'bak-client1_backup'

        assert client1_user_home.mode == 0o750

        client1_user_duplicity = File(os.path.join(client1_user.home, 'duplicity'))

        assert client1_user_duplicity.is_directory

        assert client1_user_duplicity.user == 'bak-client1_backup'

        assert client1_user_duplicity.group == 'bak-client1_backup'

        assert client1_user_duplicity.mode == 0o770

        client1_user_ssh = File(os.path.join(client1_user.home, '.ssh'))

        assert client1_user_ssh.is_directory

        assert client1_user_ssh.user == 'root'

        assert client1_user_ssh.group == 'root'

        assert client1_user_ssh.mode == 0o751

        # This verifies /etc/skel was not used for setting-up home.

        assert not File(os.path.join(client1_user.home, '.bashrc')).exists

        client2_user = User('bak-client2-backup')

        client2_user_home = File(client2_user.home)

        assert client2_user_home.is_directory

        assert client2_user_home.user == 'root'

        assert client2_user_home.group == 'bak-client2-backup'

        assert client2_user_home.mode == 0o750

        client2_user_duplicity = File(os.path.join(client2_user.home, 'duplicity'))

        assert client2_user_duplicity.is_directory

        assert client2_user_duplicity.user == 'bak-client2-backup'

        assert client2_user_duplicity.group == 'bak-client2-backup'

        assert client2_user_duplicity.mode == 0o770

        client2_user_ssh = File(os.path.join(client2_user.home, '.ssh'))

        assert client2_user_ssh.is_directory

        assert client2_user_ssh.user == 'root'

        assert client2_user_ssh.group == 'root'

        assert client2_user_ssh.mode == 0o751

        # This verifies /etc/skel was not used for setting-up home.

        assert not File(os.path.join(client2_user.home, '.bashrc')).exists

def test_backup_client_authorized_keys(File, Sudo, User):

    """

    Tests if the authorized keys for backup client system user have been set-up

    correctly.

    """

    with Sudo():

        client1_user = User('bak-client1_backup')

        client1_user_authorized_keys = File(os.path.join(client1_user.home, '.ssh', 'authorized_keys'))

        assert client1_user_authorized_keys.is_file

        assert client1_user_authorized_keys.user == 'root'

        assert client1_user_authorized_keys.group == 'bak-client1_backup'

        assert client1_user_authorized_keys.mode == 0o640

        assert client1_user_authorized_keys.content == open('tests/data/ssh/client1.pub', 'r').read().strip()

        client2_user = User('bak-client2-backup')

        client2_user_authorized_keys = File(os.path.join(client2_user.home, '.ssh', 'authorized_keys'))

        assert client2_user_authorized_keys.is_file

        assert client2_user_authorized_keys.user == 'root'

        assert client2_user_authorized_keys.group == 'bak-client2-backup'

        assert client2_user_authorized_keys.mode == 0o640

        assert client2_user_authorized_keys.content == open('tests/data/ssh/client2.pub', 'r').read().strip()

def test_firewall_configuration(File, Sudo):

    """

    Tests if the firewall configuration file has been deployed correctly.

    """

    with Sudo():

        firewall_config = File('/etc/ferm/conf.d/40-backup.conf')

        assert firewall_config.is_file

        assert firewall_config.user == 'root'

        assert firewall_config.group == 'root'

        assert firewall_config.mode == 0o640

        assert 'saddr ( 10.31.127.1 10.31.127.3) @subchain "backup_in" {' in firewall_config.content

def test_regular_ssh_server_inaccessible(LocalCommand):

    """

    Tests if the default SSH server is inaccessible for the backup client system

    users.

    """

    # Test connectivity towards regular ssh (should fail).

    login_attempt = LocalCommand("ssh "

                                 "-o PasswordAuthentication=no "

                                 "-o StrictHostKeyChecking=no "

                                 "-o UserKnownHostsFile=/dev/null "

                                 "-i tests/data/ssh/client1 "

                                 "bak-client1_backup@10.31.127.11 "

                                 "/bin/echo sshtest")

    assert login_attempt.rc != 0

    assert "Permission denied (publickey)" in login_attempt.stderr

    login_attempt = LocalCommand("ssh "

                                 "-o PasswordAuthentication=no "

                                 "-o StrictHostKeyChecking=no "

                                 "-o UserKnownHostsFile=/dev/null "

                                 "-i tests/data/ssh/client2 "

                                 "bak-client2-backup@10.31.127.11 "

                                 "/bin/echo sshtest")

    assert login_attempt.rc != 0

    assert "Permission denied (publickey)" in login_attempt.stderr

def test_backup_ssh_service_connectivity(LocalCommand):

    """

    Tests if SFTP (only) is availavble to system users used by backup clients.

    """

    # Test connectivity towards dedicated ssh (should be allowed).

    login_attempt = LocalCommand("ssh -p 2222 "

                                 "-o PasswordAuthentication=no "

                                 "-o StrictHostKeyChecking=no "

                                 "-o UserKnownHostsFile=/dev/null "

                                 "-i tests/data/ssh/client1 "

                                 "bak-client1_backup@10.31.127.11 /bin/echo sshtest")

    assert login_attempt.rc == 1

    assert "This service allows sftp connections only." in login_attempt.stdout

    # Test connectivity towards dedicated ssh (should be allowed).

    login_attempt = LocalCommand("ssh -p 2222 "

                                 "-o PasswordAuthentication=no "

                                 "-o StrictHostKeyChecking=no "

                                 "-o UserKnownHostsFile=/dev/null "

                                 "-i tests/data/ssh/client2 "

                                 "bak-client2-backup@10.31.127.11 /bin/echo sshtest")

    assert login_attempt.rc == 1

    assert "This service allows sftp connections only." in login_attempt.stdout

def test_backup_ssh_service_key_fingerprints(LocalCommand):

    """

    Tests fingerprints of backup SSH server in order to ensure correct keys are

    in use.

    """

    for key_type in ['ssh-dss', 'ssh-rsa', 'ssh-ed25519', 'ecdsa-sha2-nistp256']:

        login_attempt = LocalCommand("ssh -p 2222 "

                                     "-o PasswordAuthentication=no "

                                     "-o StrictHostKeyChecking=yes "

                                     "-o UserKnownHostsFile=tests/data/ssh/known_hosts "

                                     "-i tests/data/ssh/client1 "

                                     "-o HostKeyAlgorithms=%s "

                                     "bak-client1_backup@10.31.127.11 /bin/echo sshtest" % key_type)

        assert login_attempt.rc == 1

        assert "This service allows sftp connections only." in login_attempt.stdout