Putting dataset test results behind security flag

src/fides/api/api/v1/endpoints/dataset_endpoints.py

@@ -21,6 +21,7 @@
     HTTP_200_OK,
     HTTP_204_NO_CONTENT,
     HTTP_400_BAD_REQUEST,
+    HTTP_403_FORBIDDEN,
     HTTP_404_NOT_FOUND,
     HTTP_415_UNSUPPORTED_MEDIA_TYPE,
     HTTP_422_UNPROCESSABLE_ENTITY,
@@ -82,6 +83,7 @@
     V1_URL_PREFIX,
     YAML_DATASETS,
 )
+from fides.config import CONFIG
 
 from fides.api.models.sql_models import (  # type: ignore[attr-defined] # isort: skip
     Dataset as CtlDataset,
@@ -856,6 +858,13 @@ def test_connection_datasets(
     fides_key: FidesKey,
     unlabeled_identities: UnlabeledIdentities,
 ) -> Dict[str, Any]:
+
+    if not CONFIG.security.dsr_testing_tools_enabled:
+        raise HTTPException(
+            status_code=HTTP_403_FORBIDDEN,
+            detail="DSR testing tools are not enabled.",
+        )
+
     dataset_config = DatasetConfig.filter(
         db=db,
         conditions=(

src/fides/api/api/v1/endpoints/privacy_request_endpoints.py

@@ -2662,10 +2662,14 @@ def get_test_privacy_request_results(
     # Escape datetime and ObjectId values
     raw_data = privacy_request.get_raw_access_results()
     escaped_json = json.dumps(raw_data, indent=2, default=storage_json_encoder)
-    escaped_data = json.loads(escaped_json)
+    results = json.loads(escaped_json)
 
     return {
         "privacy_request_id": privacy_request.id,
         "status": privacy_request.status,
-        "results": escaped_data,
+        "results": (
+            results
+            if CONFIG.security.dsr_testing_tools_enabled
+            else "DSR testing tools are not enabled, results will not be shown."
+        ),
     }

src/fides/api/models/connectionconfig.py

@@ -220,10 +220,13 @@ def authorized(self) -> bool:
             return False
 
         # hard-coding to avoid cyclic dependency
-        if authentication.strategy not in ["oauth2_authorization_code", "oauth2_client_credentials"]:
+        if authentication.strategy not in [
+            "oauth2_authorization_code",
+            "oauth2_client_credentials",
+        ]:
             return False
 
-        return bool(self.secrets and 'access_token' in self.secrets.keys())
+        return bool(self.secrets and "access_token" in self.secrets.keys())
 
     @property
     def name_or_key(self) -> str:

src/fides/api/schemas/privacy_request.py

@@ -401,4 +401,4 @@ class FilteredPrivacyRequestResults(FidesSchema):
 
     privacy_request_id: str
     status: PrivacyRequestStatus
-    results: Dict[str, Any]
+    results: Union[Dict[str, Any], str]

src/fides/config/security_settings.py

@@ -123,6 +123,10 @@ class SecuritySettings(FidesSettings):
         default=False,
         description="If set to True, the user interface will display a download button for subject requests.",
     )
+    dsr_testing_tools_enabled: bool = Field(
+        default=False,
+        description="If set to True, contributor and owner roles will be able to run test privacy requests.",
+    )
     subject_request_download_link_ttl_seconds: int = Field(
         default=432000,
         description="The number of seconds that a pre-signed download URL when using S3 storage will be valid. The default is equal to 5 days.",

tests/fixtures/application_fixtures.py

@@ -3453,6 +3453,22 @@ def subject_request_download_ui_enabled():
     CONFIG.security.subject_request_download_ui_enabled = original_value
 
 
+@pytest.fixture(scope="function")
+def dsr_testing_tools_enabled():
+    original_value = CONFIG.security.dsr_testing_tools_enabled
+    CONFIG.security.dsr_testing_tools_enabled = True
+    yield
+    CONFIG.security.dsr_testing_tools_enabled = original_value
+
+
+@pytest.fixture(scope="function")
+def dsr_testing_tools_disabled():
+    original_value = CONFIG.security.dsr_testing_tools_enabled
+    CONFIG.security.dsr_testing_tools_enabled = False
+    yield
+    CONFIG.security.dsr_testing_tools_enabled = original_value
+
+
 @pytest.fixture(scope="function")
 def system_with_no_uses(db: Session) -> Generator[System, None, None]:
     system = System.create(

tests/ops/api/v1/endpoints/test_dataset_test_endpoints.py

@@ -208,6 +208,7 @@ def test_dataset_test_not_authenticated(
         response = api_client.post(dataset_url + "/test", headers={})
         assert response.status_code == 401
 
+    @pytest.mark.usefixtures("dsr_testing_tools_enabled")
     def test_dataset_test_wrong_scope(
         self,
         dataset_config,
@@ -220,7 +221,6 @@ def test_dataset_test_wrong_scope(
         response = api_client.post(dataset_url + "/test", headers=auth_header)
         assert response.status_code == 403
 
-    @pytest.mark.usefixtures("default_access_policy")
     @pytest.mark.parametrize(
         "auth_header,expected_status",
         [
@@ -231,6 +231,7 @@ def test_dataset_test_wrong_scope(
             ("approver_auth_header", HTTP_403_FORBIDDEN),
         ],
     )
+    @pytest.mark.usefixtures("default_access_policy", "dsr_testing_tools_enabled")
     def test_dataset_test_with_roles(
         self,
         dataset_config,
@@ -249,6 +250,7 @@ def test_dataset_test_with_roles(
         )
         assert response.status_code == expected_status
 
+    @pytest.mark.usefixtures("dsr_testing_tools_enabled")
     def test_dataset_test_connection_does_not_exist(
         self,
         api_client: TestClient,
@@ -263,6 +265,7 @@ def test_dataset_test_connection_does_not_exist(
         )
         assert response.status_code == 404
 
+    @pytest.mark.usefixtures("dsr_testing_tools_enabled")
     def test_dataset_test_dataset_does_not_exist(
         self,
         connection_config,
@@ -290,7 +293,7 @@ def test_dataset_test_dataset_does_not_exist(
             ),
         ],
     )
-    @pytest.mark.usefixtures("default_access_policy")
+    @pytest.mark.usefixtures("default_access_policy", "dsr_testing_tools_enabled")
     def test_dataset_test_invalid_payloads(
         self,
         connection_config,
@@ -310,7 +313,9 @@ def test_dataset_test_invalid_payloads(
         assert response.status_code == HTTP_400_BAD_REQUEST
         assert response.json()["detail"] == expected_response
 
-    @pytest.mark.usefixtures("default_access_policy", "postgres_integration_db")
+    @pytest.mark.usefixtures(
+        "default_access_policy", "postgres_integration_db", "dsr_testing_tools_enabled"
+    )
     def test_dataset_test(
         self,
         connection_config,
@@ -327,3 +332,23 @@ def test_dataset_test(
         )
         assert response.status_code == HTTP_200_OK
         assert "privacy_request_id" in response.json().keys()
+
+    @pytest.mark.usefixtures(
+        "default_access_policy", "postgres_integration_db", "dsr_testing_tools_disabled"
+    )
+    def test_dataset_test_disabled(
+        self,
+        connection_config,
+        dataset_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = generate_auth_header(scopes=[DATASET_TEST])
+        response = api_client.post(
+            dataset_url + "/test",
+            headers=auth_header,
+            json={"email": "[email protected]"},
+        )
+        assert response.status_code == HTTP_403_FORBIDDEN
+        assert response.json()["detail"] == "DSR testing tools are not enabled."

tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py

@@ -8392,7 +8392,7 @@ def test_filtered_results_wrong_scope(
         response = api_client.get(url, headers=auth_header)
         assert response.status_code == 403
 
-    @pytest.mark.usefixtures("default_access_policy")
+    @pytest.mark.usefixtures("default_access_policy", "dsr_testing_tools_enabled")
     @pytest.mark.parametrize(
         "auth_header,expected_status",
         [
@@ -8427,7 +8427,9 @@ def test_filtered_results_with_roles(
         assert response.status_code == expected_status
 
     @pytest.mark.integration_postgres
-    @pytest.mark.usefixtures("default_access_policy", "postgres_integration_db")
+    @pytest.mark.usefixtures(
+        "default_access_policy", "postgres_integration_db", "dsr_testing_tools_enabled"
+    )
     def test_filtered_results_postgres(
         self,
         connection_config,
@@ -8462,8 +8464,57 @@ def test_filtered_results_postgres(
             "results",
         }
 
+    @pytest.mark.integration_postgres
+    @pytest.mark.usefixtures(
+        "default_access_policy",
+        "postgres_integration_db",
+        "dsr_testing_tools_enabled",
+    )
+    def test_filtered_results_postgres_access_testing_disabled(
+        self,
+        connection_config,
+        postgres_example_test_dataset_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(
+            connection_config, postgres_example_test_dataset_config
+        )
+        auth_header = generate_auth_header(scopes=[DATASET_TEST])
+        response = api_client.post(
+            dataset_url + "/test",
+            headers=auth_header,
+            json={"email": "[email protected]"},
+        )
+        assert response.status_code == HTTP_200_OK
+
+        original_value = CONFIG.security.dsr_testing_tools_enabled
+        CONFIG.security.dsr_testing_tools_enabled = False
+
+        privacy_request_id = response.json()["privacy_request_id"]
+        url = V1_URL_PREFIX + PRIVACY_REQUEST_FILTERED_RESULTS.format(
+            privacy_request_id=privacy_request_id
+        )
+        auth_header = generate_auth_header(scopes=[PRIVACY_REQUEST_READ_ACCESS_RESULTS])
+        response = api_client.get(
+            url,
+            headers=auth_header,
+        )
+        assert response.status_code == HTTP_200_OK
+        assert set(response.json().keys()) == {
+            "privacy_request_id",
+            "status",
+            "results",
+        }
+        assert (
+            response.json()["results"]
+            == "DSR testing tools are not enabled, results will not be shown."
+        )
+
+        CONFIG.security.dsr_testing_tools_enabled = original_value
+
     @pytest.mark.integration_mongo
-    @pytest.mark.usefixtures("default_access_policy")
+    @pytest.mark.usefixtures("default_access_policy", "dsr_testing_tools_enabled")
     def test_filtered_results_mongo(
         self,
         mongo_connection_config,