Files @ c09b9e29c8a8
Branch filter:

Location: majic-ansible-roles/roles/xmpp_server/molecule/default/tests/test_client.py - annotation

branko
MAR-242: Added initial set of tests for file upload size and user daily quota:

- Tests currently operate on default values for the Prosody modules.
2ada86e90026
c09b9e29c8a8
04a99d1e5a60
2ada86e90026
f8f8d51c3fd5
f8f8d51c3fd5
e970d4afbea4
e970d4afbea4
e970d4afbea4
e970d4afbea4
d62b3adec462
e970d4afbea4
e970d4afbea4
9bc79c136d4f
04a99d1e5a60
72c8481f0498
72c8481f0498
e970d4afbea4
e970d4afbea4
e970d4afbea4
e970d4afbea4
e970d4afbea4
e970d4afbea4
72c8481f0498
9bc79c136d4f
9bc79c136d4f
e970d4afbea4
e970d4afbea4
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
e970d4afbea4
e970d4afbea4
e970d4afbea4
e970d4afbea4
3c51248b600c
3c51248b600c
3c51248b600c
e970d4afbea4
3c51248b600c
e970d4afbea4
3c51248b600c
3c51248b600c
3c51248b600c
e970d4afbea4
3c51248b600c
e970d4afbea4
e970d4afbea4
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
e970d4afbea4
3c51248b600c
e970d4afbea4
e970d4afbea4
3c51248b600c
3c51248b600c
3c51248b600c
3c51248b600c
3c51248b600c
3c51248b600c
e970d4afbea4
e970d4afbea4
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
e970d4afbea4
e970d4afbea4
e970d4afbea4
e970d4afbea4
3c51248b600c
3c51248b600c
3c51248b600c
e970d4afbea4
e970d4afbea4
3c51248b600c
3c51248b600c
3c51248b600c
e970d4afbea4
e970d4afbea4
e970d4afbea4
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
f8f8d51c3fd5
e970d4afbea4
e970d4afbea4
e970d4afbea4
e970d4afbea4
e970d4afbea4
3c51248b600c
3c51248b600c
3c51248b600c
e970d4afbea4
3c51248b600c
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
04a99d1e5a60
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
c09b9e29c8a8
import os
import time
import uuid

import pytest

import testinfra.utils.ansible_runner


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


@pytest.mark.parametrize('server', ['parameters-mandatory', 'parameters-optional'])
@pytest.mark.parametrize('port', [5222, 5223, 5000, 5269, 5281])
@pytest.mark.parametrize('ip_protocol', [4, 6])
def test_connectivity(host, server, port, ip_protocol):
    """
    Tests connectivity to the XMPP server (ports that should be reachable).
    """

    with host.sudo():

        scan = host.run('nmap -%s -p %s -oG - %s', str(ip_protocol), str(port), server)
        assert scan.rc == 0
        assert "Ports: %d/open/tcp//" % port in scan.stdout


@pytest.mark.parametrize("username, password, domain", [
    ["john.doe", "johnpassword", "domain1"],
    ["jane.doe", "janepassword", "domain2"],
])
def test_tls(host, username, password, domain):
    """
    Tests if TLS works as expected.
    """

    send = host.run(f"echo 'Hello' | go-sendxmpp --debug "
                    f"--username {username}@{domain} --password {password} --jserver {domain}:5222 "
                    f"{username}@{domain}")
    assert send.rc == 0
    assert "<body>Hello</body>" in send.stderr

    send = host.run(f"echo 'Hello' | go-sendxmpp --debug --tls "
                    f"--username {username}@{domain} --password {password} --jserver {domain}:5223 "
                    f"{username}@{domain}")
    assert send.rc == 0
    assert "<body>Hello</body>" in send.stderr


@pytest.mark.parametrize("username, password, domain", [
    ["john.doe", "johnpassword", "domain1"],
    ["jane.doe", "janepassword", "domain2"],
])
def test_authentication_requires_tls(host, username, password, domain):
    """
    Tests if STARTTLS is required.
    """

    send = host.run(f"echo 'Hello' | go-sendxmpp --debug "
                    f"--username {username}@{domain} --password {password} --jserver {domain}:5222 "
                    f"{username}@{domain}")

    assert send.rc == 0
    assert "<starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'><required/></starttls>" in send.stderr


@pytest.mark.parametrize("username, password, domain", [
    ["john.doe", "johnpassword", "domain1"],
    ["jane.doe", "janepassword", "domain2"],
    ["mick.doe", "mickpassword", "domain3"],
])
def test_authentication(host, username, password, domain):
    """
    Tests if authentication works correctly.
    """

    send = host.run(f"echo 'Hello' | go-sendxmpp --debug "
                    f"--username {username}@{domain} --password {password} --jserver {domain}:5222 "
                    f"{username}@{domain}")
    assert send.rc == 0

    send = host.run(f"echo 'Hello' | go-sendxmpp --debug --tls "
                    f"--username {username}@{domain} --password {password} --jserver {domain}:5223 "
                    f"{username}@{domain}")
    assert send.rc == 0


@pytest.mark.parametrize("target_username, target_domain", [
    ["john.doe", "domain1"],
    ["jane.doe", "domain2"],
])
def test_unauthorized_users_rejected(host, target_username, target_domain):
    """
    Tests if unauthorized users (present in LDAP, but not member of correct
    group) are rejected from accessing the XMPP server.
    """

    send = host.run(f"echo 'Hello' | go-sendxmpp --debug "
                    f"--username noxmpp@{target_domain} --password noxmpppassword --jserver {target_domain}:5222 "
                    f"{target_username}@{target_domain}")
    assert send.rc != 0
    assert "Unable to authorize you with the authentication credentials you've sent" in send.stderr


@pytest.mark.parametrize("username, password, domain, server", [
    ["john.doe", "johnpassword", "domain1", "parameters-mandatory"],
    ["jane.doe", "janepassword", "domain2", "parameters-optional"],
    ["mick.doe", "mickpassword", "domain3", "parameters-optional"],
])
def test_http_file_upload(host, username, password, domain, server):
    """
    Tests if http file upload works correctly.
    """

    # Prepare file for transfer.
    expected_content = str(uuid.uuid4())
    create_sample_file = host.run("echo -n %s > /tmp/http_file_upload_sample.txt", expected_content)
    assert create_sample_file.rc == 0

    # Path where uploaded file will end-up.
    upload_directory_path = f"/var/lib/prosody/upload%2e{domain}/http_file_share"

    # Find the host that serves the domain. Used for validating uploaded content.
    ansible_facts = host.ansible("setup")["ansible_facts"]
    ansible_distribution_release = ansible_facts['ansible_distribution_release']
    ansible_runner = testinfra.utils.ansible_runner.AnsibleRunner(os.environ['MOLECULE_INVENTORY_FILE'])
    server_host = ansible_runner.get_host(ansible_runner.get_hosts(f'{server}-{ansible_distribution_release}')[0])

    # Clean up leftovers from previous run.
    with server_host.sudo():
        server_host.run("rm -rf %s/*", upload_directory_path)

    # Upload the file.
    send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
                    f"--http-upload /tmp/http_file_upload_sample.txt "
                    f"{username}@{domain}")
    assert send.rc == 0
    assert "No http upload component found." not in send.stderr

    # Verify content on server.
    with server_host.sudo():
        upload_directory = server_host.file(upload_directory_path)
        assert upload_directory.is_directory
        assert upload_directory.user == "prosody"
        assert upload_directory.group == "prosody"
        assert upload_directory.mode == 0o750
        assert len(upload_directory.listdir()) == 1

        uploaded_file_name = upload_directory.listdir()[0]
        uploaded_file = server_host.file(os.path.join(upload_directory_path, uploaded_file_name))
        assert uploaded_file.is_file
        assert uploaded_file.user == "prosody"
        assert uploaded_file.group == "prosody"
        assert uploaded_file.mode == 0o640
        assert uploaded_file.content_string == expected_content


@pytest.mark.parametrize("username, password, domain, server", [
    ["john.doe", "johnpassword", "domain1", "parameters-mandatory"],
    ["jane.doe", "janepassword", "domain2", "parameters-optional"],
    ["mick.doe", "mickpassword", "domain3", "parameters-optional"],
])
def test_http_file_share_size_limit(host, username, password, domain, server):
    """
    Tests the maximum file size for files uploaded via XEP-0363.
    """

    file_size_limit = 10 * 1024 * 1024

    # Find the host that serves the domain. Used for validating uploaded content.
    ansible_facts = host.ansible("setup")["ansible_facts"]
    ansible_distribution_release = ansible_facts['ansible_distribution_release']
    ansible_runner = testinfra.utils.ansible_runner.AnsibleRunner(os.environ['MOLECULE_INVENTORY_FILE'])
    server_host = ansible_runner.get_host(ansible_runner.get_hosts(f'{server}-{ansible_distribution_release}')[0])

    # Clean up leftovers from previous run.
    with server_host.sudo():
        server_host.run("rm -rf %s/*", f"/var/lib/prosody/upload%2e{domain}/")
        server_host.run("systemctl restart prosody")
        time.sleep(1)

    # Test exact size limit.
    create_sample_file = host.run("dd if=/dev/zero of=/tmp/http_file_upload_sample.txt bs=%sB count=1", str(file_size_limit))
    assert create_sample_file.rc == 0

    send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
                    f"--http-upload /tmp/http_file_upload_sample.txt "
                    f"{username}@{domain}")
    assert "file-too-large" not in send.stderr

    # Test exceeded size limit.
    create_sample_file = host.run("dd if=/dev/zero of=/tmp/http_file_upload_sample.txt bs=%sB count=1", str(file_size_limit + 1))
    assert create_sample_file.rc == 0

    send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
                    f"--http-upload /tmp/http_file_upload_sample.txt "
                    f"{username}@{domain}")
    assert "file-too-large" in send.stderr


@pytest.mark.parametrize("username, password, domain, server", [
    ["john.doe", "johnpassword", "domain1", "parameters-mandatory"],
    ["jane.doe", "janepassword", "domain2", "parameters-optional"],
    ["mick.doe", "mickpassword", "domain3", "parameters-optional"],
])
def test_http_file_share_daily_quota(host, username, password, domain, server):
    """
    Tests the user's daily quota for files uploaded via XEP-0363.
    """

    file_size_limit = 10 * 1024 * 1024

    # Find the host that serves the domain. Used for validating uploaded content.
    ansible_facts = host.ansible("setup")["ansible_facts"]
    ansible_distribution_release = ansible_facts['ansible_distribution_release']
    ansible_runner = testinfra.utils.ansible_runner.AnsibleRunner(os.environ['MOLECULE_INVENTORY_FILE'])
    server_host = ansible_runner.get_host(ansible_runner.get_hosts(f'{server}-{ansible_distribution_release}')[0])

    # Clean up leftovers from previous run.
    with server_host.sudo():
        server_host.run("rm -rf %s/*", f"/var/lib/prosody/upload%2e{domain}/")
        server_host.run("systemctl restart prosody")
        time.sleep(1)

    # Fill-up the daily quota.
    create_sample_file = host.run("dd if=/dev/zero of=/tmp/http_file_upload_sample.txt bs=%sB count=1", str(file_size_limit))
    assert create_sample_file.rc == 0
    for _ in range(10):
        send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
                        f"--http-upload /tmp/http_file_upload_sample.txt "
                        f"{username}@{domain}")
        assert send.rc == 0

    # Test exceeded daily quota.
    create_sample_file = host.run("dd if=/dev/zero of=/tmp/http_file_upload_sample.txt bs=1B count=1")
    assert create_sample_file.rc == 0

    send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
                    f"--http-upload /tmp/http_file_upload_sample.txt "
                    f"{username}@{domain}")
    assert "Daily quota reached" in send.stderr