diff --git a/securedrop/tests/functional/conftest.py b/securedrop/tests/functional/conftest.py index 24e735f446..7957c88d6e 100644 --- a/securedrop/tests/functional/conftest.py +++ b/securedrop/tests/functional/conftest.py @@ -1,6 +1,6 @@ +import multiprocessing from contextlib import contextmanager from dataclasses import dataclass -from multiprocessing.context import Process from pathlib import Path from typing import Generator import requests @@ -100,9 +100,21 @@ def spawn_sd_servers( # Spawn the source and journalist web apps in separate processes source_port = _get_unused_port() journalist_port = _get_unused_port() - source_app_process = Process(target=_start_source_server, args=(source_port, config_to_use)) + + # Start the server subprocesses using the "spawn" method instead of "fork". + # This is needed for the config_to_use argument to work; if "fork" is used, the subprocess + # will inherit all the globals from the parent process, which will include the Python + # variables declared as "global" in the SD code base + # (example: see _DesignationGenerator.get_default()). + # This means the config_to_use will be ignored if these globals have already been + # initialized (for example by tests running before the code here). + mp_spawn_ctx = multiprocessing.get_context("spawn") + + source_app_process = mp_spawn_ctx.Process( # type: ignore + target=_start_source_server, args=(source_port, config_to_use) + ) source_app_process.start() - journalist_app_process = Process( + journalist_app_process = mp_spawn_ctx.Process( # type: ignore target=_start_journalist_server, args=(journalist_port, config_to_use) ) journalist_app_process.start() diff --git a/securedrop/tests/functional/factories.py b/securedrop/tests/functional/factories.py index 4c4e54e10d..d17ec71e97 100644 --- a/securedrop/tests/functional/factories.py +++ b/securedrop/tests/functional/factories.py @@ -36,8 +36,8 @@ def create( ) -> SecureDropConfig: """Create a securedrop config suitable for the unit tests. - It will automatically create an initialized DB at SECUREDROP_DATA_ROOT/db.sqlite which will - be set as the DATABASE_FILE. + It will erase any existing file within SECUREDROP_DATA_ROOT and then create an initialized + DB at SECUREDROP_DATA_ROOT/db.sqlite which will be set as the DATABASE_FILE. """ # Clear the data root directory if SECUREDROP_DATA_ROOT.exists(): diff --git a/securedrop/tests/functional/source_navigation_steps.py b/securedrop/tests/functional/source_navigation_steps.py index 2b593a6add..d56929ce4f 100644 --- a/securedrop/tests/functional/source_navigation_steps.py +++ b/securedrop/tests/functional/source_navigation_steps.py @@ -127,14 +127,6 @@ def submit_page_loaded(): self.wait_for(submit_page_loaded) - def _source_continues_to_submit_page_with_colliding_journalist_designation(self): - self.safe_click_by_id("continue-button") - - self.wait_for(lambda: self.driver.find_element_by_css_selector(".error")) - flash_error = self.driver.find_element_by_css_selector(".error") - assert "There was a temporary problem creating your account. Please try again." \ - == flash_error.text - def _source_submits_a_file(self): with tempfile.NamedTemporaryFile() as file: file.write(self.secret_message.encode("utf-8")) diff --git a/securedrop/tests/functional/test_source.py b/securedrop/tests/functional/test_source.py index 7da0739f90..511bfecc93 100644 --- a/securedrop/tests/functional/test_source.py +++ b/securedrop/tests/functional/test_source.py @@ -1,205 +1,241 @@ +import requests import werkzeug -import source_user +from tests.functional.app_navigators import SourceAppNagivator from ..test_journalist import VALID_PASSWORD -from . import source_navigation_steps, journalist_navigation_steps -from . import functional_test -from sdconfig import config +from tests.functional import tor_utils -class TestSourceInterfaceDesignationCollision( - functional_test.FunctionalTest, - source_navigation_steps.SourceNavigationStepsMixin): +class TestSourceAppCodenameHints: - def start_source_server(self, source_port): - # Generator that always returns the same journalist designation - source_user._default_designation_generator = source_user._DesignationGenerator( - nouns=["accent"], - adjectives=["tonic"], + FIRST_SUBMISSION_TEXT = "Please check back later for replies" + SUBMISSION_ON_FIRST_LOGIN_TEXT = "Forgot your codename?" + + def test_no_codename_hint_on_second_login(self, sd_servers_v2, tor_browser_web_driver): + navigator = SourceAppNagivator( + source_app_base_url=sd_servers_v2.source_app_base_url, + web_driver=tor_browser_web_driver, + ) + + # Given a source user who creates an account + # When they first login + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + navigator.source_continues_to_submit_page() + + # Then they are able to retrieve their codename from the UI + source_codename = navigator.source_retrieves_codename_from_hint() + assert source_codename + + # And they are able to close the codename hint UI + content = navigator.driver.find_element_by_css_selector("details#codename-hint") + assert content.get_attribute("open") is not None + navigator.nav_helper.safe_click_by_id("codename-hint") + assert content.get_attribute("open") is None + + # And on their second login + navigator.source_logs_out() + navigator.source_visits_source_homepage() + navigator.source_chooses_to_login() + navigator.source_proceeds_to_login(codename=source_codename) + + # The codename hint UI is no longer present + codename = navigator.driver.find_elements_by_css_selector(".code-reminder") + assert len(codename) == 0 + + def test_submission_notifications_on_first_login(self, sd_servers_v2, tor_browser_web_driver): + navigator = SourceAppNagivator( + source_app_base_url=sd_servers_v2.source_app_base_url, + web_driver=tor_browser_web_driver, ) - config.SESSION_EXPIRATION_MINUTES = self.session_expiration / 60.0 - - self.source_app.run(port=source_port, debug=True, use_reloader=False, threaded=True) - - def test_journalist_designation_collisions(self): - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - self._source_continues_to_submit_page() - self._source_logs_out() - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - self._source_continues_to_submit_page_with_colliding_journalist_designation() - - -class TestSourceInterface( - functional_test.FunctionalTest, - source_navigation_steps.SourceNavigationStepsMixin): - - def test_lookup_codename_hint(self): - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - self._source_continues_to_submit_page() - self._source_shows_codename() - self._source_hides_codename() - self._source_logs_out() - self._source_visits_source_homepage() - self._source_chooses_to_login() - self._source_proceeds_to_login() - self._source_sees_no_codename() - - def test_lookup_submit_notification_first_login(self): - """Test that on a first login, the submission notification includes the 'Please check back - later' and 'Forgot your codename?' messages. Also verify that those messages are not - present on a subsequent submission.""" - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - self._source_continues_to_submit_page() - self._source_submits_a_message( - verify_notification=True, first_submission=True, first_login=True + # Given a source user who creates an account + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + navigator.source_continues_to_submit_page() + + # When they submit a message during their first login + # Then it succeeds + confirmation_text_first_submission = navigator.source_submits_a_message() + + # And they see the expected confirmation messages for a first submission on first login + assert self.SUBMISSION_ON_FIRST_LOGIN_TEXT in confirmation_text_first_submission + assert self.FIRST_SUBMISSION_TEXT in confirmation_text_first_submission + + # And when they submit a second message + confirmation_text_second_submission = navigator.source_submits_a_message() + + # Then they don't see the messages since it's not their first submission + assert self.SUBMISSION_ON_FIRST_LOGIN_TEXT not in confirmation_text_second_submission + assert self.FIRST_SUBMISSION_TEXT not in confirmation_text_second_submission + + def test_submission_notifications_on_second_login(self, sd_servers_v2, tor_browser_web_driver): + navigator = SourceAppNagivator( + source_app_base_url=sd_servers_v2.source_app_base_url, + web_driver=tor_browser_web_driver, + ) + + # Given a source user who creates an account + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + navigator.source_continues_to_submit_page() + source_codename = navigator.source_retrieves_codename_from_hint() + assert source_codename + + # When they submit a message during their second login + navigator.source_logs_out() + navigator.source_visits_source_homepage() + navigator.source_chooses_to_login() + navigator.source_proceeds_to_login(codename=source_codename) + + # Then it succeeds + confirmation_text_first_submission = navigator.source_submits_a_message() + + # And they see the expected confirmation messages for a first submission on second login + assert self.SUBMISSION_ON_FIRST_LOGIN_TEXT not in confirmation_text_first_submission + assert self.FIRST_SUBMISSION_TEXT in confirmation_text_first_submission + + # And when they submit a second message + confirmation_text_second_submission = navigator.source_submits_a_message() + + # Then they don't see the messages since it's not their first submission + assert self.SUBMISSION_ON_FIRST_LOGIN_TEXT not in confirmation_text_second_submission + assert self.FIRST_SUBMISSION_TEXT not in confirmation_text_second_submission + + +class TestSourceAppDownloadJournalistKey: + def test(self, sd_servers_v2): + # Given a source app, when fetching the instance's journalist public key + url = f"{sd_servers_v2.source_app_base_url}/public-key" + response = requests.get(url=url, proxies=tor_utils.proxies_for_url(url)) + + # Then it succeeds and the right data is returned + assert "BEGIN PGP PUBLIC KEY BLOCK" in response.content.decode("utf-8") + + +class TestSourceAppCodenamesInMultipleTabs: + """Test generation of multiple codenames in different browser tabs, ref. issue 4458.""" + + @staticmethod + def _assert_is_on_lookup_page(navigator: SourceAppNagivator) -> None: + navigator.nav_helper.wait_for(lambda: navigator.driver.find_element_by_id("upload")) + + @staticmethod + def _extract_generated_codename(navigator: SourceAppNagivator) -> str: + codename = navigator.driver.find_element_by_css_selector("#codename").text + assert codename + return codename + + def test_generate_codenames_in_multiple_tabs(self, sd_servers_v2, tor_browser_web_driver): + navigator = SourceAppNagivator( + source_app_base_url=sd_servers_v2.source_app_base_url, + web_driver=tor_browser_web_driver, ) - self._source_submits_a_message(verify_notification=True) - - def test_lookup_submit_notification_2nd_login(self): - """Test that on a second login, the first submission notification includes the 'Please - check back later' message but not the 'Forgot your codename?' message since the codename - hint section is not present on a second login (Ref. issue #5101). Also verify that none of - those messages are present on a subsequent submission.""" - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - self._source_continues_to_submit_page() - self._source_logs_out() - self._source_visits_source_homepage() - self._source_chooses_to_login() - self._source_proceeds_to_login() - self._source_submits_a_message(verify_notification=True, first_submission=True) - self._source_submits_a_message(verify_notification=True) - - -class TestDownloadKey( - functional_test.FunctionalTest, - journalist_navigation_steps.JournalistNavigationStepsMixin): - - def test_journalist_key_from_source_interface(self): - data = self.return_downloaded_content(self.source_location + - "/public-key", None) - - data = data.decode('utf-8') - assert "BEGIN PGP PUBLIC KEY BLOCK" in data - - -class TestDuplicateSourceInterface( - functional_test.FunctionalTest, - source_navigation_steps.SourceNavigationStepsMixin): - - def get_codename_generate(self): - return self.driver.find_element_by_css_selector("#codename").text - - def get_codename_lookup(self): - return self.driver.find_element_by_css_selector("#codename-hint mark").text - - def test_duplicate_generate_pages(self): - # Test generation of multiple codenames in different browser tabs, ref. issue 4458. - - # Generate a codename in Tab A - assert len(self.driver.window_handles) == 1 - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - codename_a = self.get_codename_generate() - - # Generate a different codename in Tab B - self.driver.execute_script("window.open('about:blank', '_blank')") - tab_b = self.driver.window_handles[1] - assert len(self.driver.window_handles) == 2 - self.driver.switch_to.window(tab_b) - assert self.driver.current_window_handle == tab_b - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - codename_b = self.get_codename_generate() - - tab_a = self.driver.window_handles[0] + + # Given a user who generated a codename in Tab A + tab_a = navigator.driver.window_handles[0] + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + codename_a = self._extract_generated_codename(navigator) + + # And they then opened a new tab, Tab B + navigator.driver.execute_script("window.open('about:blank', '_blank')") + tab_b = navigator.driver.window_handles[1] + navigator.driver.switch_to.window(tab_b) assert tab_a != tab_b + + # And they also generated another codename in Tab B + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + codename_b = self._extract_generated_codename(navigator) assert codename_a != codename_b - # Proceed to submit documents in Tab A - assert len(self.driver.window_handles) == 2 - self.driver.switch_to.window(tab_a) - assert self.driver.current_window_handle == tab_a - self._source_continues_to_submit_page() - assert self._is_on_lookup_page() - self._source_shows_codename(verify_source_name=False) - codename_lookup_a = self.get_codename_lookup() - assert codename_lookup_a == codename_a - self._source_submits_a_message() - - # Proceed to submit documents in Tab B - self.driver.switch_to.window(tab_b) - assert self.driver.current_window_handle == tab_b - self._source_continues_to_submit_page() - assert self._is_on_lookup_page() - self._source_sees_already_logged_in_in_other_tab_message() - self._source_shows_codename(verify_source_name=False) - codename_lookup_b = self.get_codename_lookup() - # We expect the codename to be the one from Tab A - assert codename_lookup_b == codename_a - self._source_submits_a_message() - - def test_refreshed_duplicate_generate_pages(self): - # Test generation of multiple codenames in different browser tabs, including behavior - # of refreshing the codename in each tab. Ref. issue 4458. - - # Generate a codename in Tab A - assert len(self.driver.window_handles) == 1 - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - codename_a1 = self.get_codename_generate() - # Regenerate codename in Tab A - self._source_regenerates_codename() - codename_a2 = self.get_codename_generate() + # And they ended up creating their account and submitting documents in Tab A + navigator.driver.switch_to.window(tab_a) + navigator.source_continues_to_submit_page() + self._assert_is_on_lookup_page(navigator) + assert navigator.source_retrieves_codename_from_hint() == codename_a + navigator.source_submits_a_message() + + # When the user tries to create an account and submit documents in Tab B + navigator.driver.switch_to.window(tab_b) + navigator.source_continues_to_submit_page() + + # Then the submission fails and the user sees the corresponding flash message in Tab B + self._assert_is_on_lookup_page(navigator) + notification = navigator.source_sees_flash_message() + if not navigator.accept_languages: + assert "You are already logged in." in notification.text + + # And the user's actual codename is the one initially generated in Tab A + assert navigator.source_retrieves_codename_from_hint() == codename_a + + def test_generate_and_refresh_codenames_in_multiple_tabs( + self, sd_servers_v2, tor_browser_web_driver + ): + navigator = SourceAppNagivator( + source_app_base_url=sd_servers_v2.source_app_base_url, + web_driver=tor_browser_web_driver, + ) + + # Given a user who generated a codename in Tab A + tab_a = navigator.driver.window_handles[0] + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + codename_a1 = self._extract_generated_codename(navigator) + + # And they then re-generated their codename in Tab + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + codename_a2 = self._extract_generated_codename(navigator) assert codename_a1 != codename_a2 - # Generate a different codename in Tab B - self.driver.execute_script("window.open('about:blank', '_blank')") - tab_a = self.driver.window_handles[0] - tab_b = self.driver.window_handles[1] - self.driver.switch_to.window(tab_b) - assert self.driver.current_window_handle == tab_b - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - codename_b = self.get_codename_generate() - assert codename_b != codename_a1 != codename_a2 - - # Proceed to submit documents in Tab A - self.driver.switch_to.window(tab_a) - assert self.driver.current_window_handle == tab_a - self._source_continues_to_submit_page() - assert self._is_on_lookup_page() - self._source_shows_codename(verify_source_name=False) - codename_lookup_a = self.get_codename_lookup() - assert codename_lookup_a == codename_a2 - self._source_submits_a_message() - - # Regenerate codename in Tab B - self.driver.switch_to.window(tab_b) - assert self.driver.current_window_handle == tab_b - self._source_regenerates_codename() - # We expect the source to be directed to /lookup with a flash message - assert self._is_on_lookup_page() - self._source_sees_redirect_already_logged_in_message() - # Check codename - self._source_shows_codename(verify_source_name=False) - codename_lookup_b = self.get_codename_lookup() - assert codename_lookup_b == codename_a2 - self._source_submits_a_message() - - def test_codenames_exceed_max_cookie_size(self): - # Test generation of enough codenames (from multiple tabs and/or serial - # refreshes) that the resulting cookie exceeds the recommended - # `werkzeug.Response.max_cookie_size` = 4093 bytes. (#6043) - - too_many = 2*(werkzeug.Response.max_cookie_size // len(VALID_PASSWORD)) - for i in range(too_many): - self._source_visits_source_homepage() - self._source_chooses_to_submit_documents() - - self._source_continues_to_submit_page() + # And they then opened a new tab, Tab B + navigator.driver.execute_script("window.open('about:blank', '_blank')") + tab_b = navigator.driver.window_handles[1] + navigator.driver.switch_to.window(tab_b) + assert tab_a != tab_b + + # And they also generated another codename in Tab B + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + codename_b = self._extract_generated_codename(navigator) + assert codename_a2 != codename_b + + # And they ended up creating their account and submitting documents in Tab A + navigator.driver.switch_to.window(tab_a) + navigator.source_continues_to_submit_page() + self._assert_is_on_lookup_page(navigator) + assert navigator.source_retrieves_codename_from_hint() == codename_a2 + navigator.source_submits_a_message() + + # When they try to re-generate a codename in Tab B + navigator.driver.switch_to.window(tab_b) + navigator.source_visits_source_homepage() + navigator.nav_helper.safe_click_by_id("submit-documents-button") + + # Then they get redirected to /lookup with the corresponding flash message + self._assert_is_on_lookup_page(navigator) + notification = navigator.source_sees_flash_message() + if not navigator.accept_languages: + assert "You were redirected because you are already logged in." in notification.text + + # And the user's actual codename is the expected one + assert navigator.source_retrieves_codename_from_hint() == codename_a2 + + # TODO(AD): This test takes ~50s ; we could refactor it to speed it up + def test_codenames_exceed_max_cookie_size(self, sd_servers_v2, tor_browser_web_driver): + """Test generation of enough codenames that the resulting cookie exceeds the recommended + `werkzeug.Response.max_cookie_size` = 4093 bytes. (#6043) + """ + navigator = SourceAppNagivator( + source_app_base_url=sd_servers_v2.source_app_base_url, + web_driver=tor_browser_web_driver, + ) + + too_many = 2 * (werkzeug.Response.max_cookie_size // len(VALID_PASSWORD)) + for _ in range(too_many): + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + + navigator.source_continues_to_submit_page() diff --git a/securedrop/tests/functional/test_source_designation_collision.py b/securedrop/tests/functional/test_source_designation_collision.py new file mode 100644 index 0000000000..0e926ff9a1 --- /dev/null +++ b/securedrop/tests/functional/test_source_designation_collision.py @@ -0,0 +1,61 @@ +from pathlib import Path + +import pytest + +from .app_navigators import SourceAppNagivator +from .conftest import spawn_sd_servers +from .factories import SecureDropConfigFactory + + +@pytest.fixture(scope="session") +def _sd_servers_with_designation_collisions(setup_journalist_key_and_gpg_folder): + """Spawn source and journalist apps that can only generate a single journalist designation.""" + # Generate a config that can only generate a single journalist designation + folder_for_fixture_path = Path("/tmp/sd-tests/functional-designation-collisions") + folder_for_fixture_path.mkdir(parents=True, exist_ok=True) + nouns_path = folder_for_fixture_path / "nouns.txt" + nouns_path.touch(exist_ok=True) + nouns_path.write_text("accent") + adjectives_path = folder_for_fixture_path / "adjectives.txt" + adjectives_path.touch(exist_ok=True) + adjectives_path.write_text("tonic") + config_for_collisions = SecureDropConfigFactory.create( + SECUREDROP_DATA_ROOT=folder_for_fixture_path / "sd_data_root", + NOUNS=nouns_path, + ADJECTIVES=adjectives_path, + ) + + # Ensure the GPG settings match the one in the config to use, to ensure consistency + journalist_key_fingerprint, gpg_dir = setup_journalist_key_and_gpg_folder + assert Path(config_for_collisions.GPG_KEY_DIR) == gpg_dir + assert config_for_collisions.JOURNALIST_KEY == journalist_key_fingerprint + + # Spawn the apps in separate processes + with spawn_sd_servers(config_to_use=config_for_collisions) as sd_servers_result: + yield sd_servers_result + + +class TestSourceAppDesignationCollision: + def test(self, _sd_servers_with_designation_collisions, tor_browser_web_driver): + navigator = SourceAppNagivator( + source_app_base_url=_sd_servers_with_designation_collisions.source_app_base_url, + web_driver=tor_browser_web_driver, + ) + + # Given a source user who created an account + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + navigator.source_continues_to_submit_page() + navigator.source_logs_out() + + # When another source user creates an account but gets the same journalist designation + navigator.source_visits_source_homepage() + navigator.source_clicks_submit_documents_on_homepage() + + # Then the right error message is displayed + navigator.nav_helper.safe_click_by_id("continue-button") + navigator.nav_helper.wait_for( + lambda: navigator.driver.find_element_by_css_selector(".error") + ) + flash_error = navigator.driver.find_element_by_css_selector(".error") + assert "There was a temporary problem creating your account" in flash_error.text