import os 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 "Hello" 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 "Hello" 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 "" 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