diff --git a/api/changelog.d/scan-report-writer-transaction.fixed.md b/api/changelog.d/scan-report-writer-transaction.fixed.md new file mode 100644 index 0000000000..52a0932f77 --- /dev/null +++ b/api/changelog.d/scan-report-writer-transaction.fixed.md @@ -0,0 +1 @@ +Scan report generation no longer holds a database transaction open while it renders and uploads files, which could block schema changes for the length of the report diff --git a/api/src/backend/tasks/tasks.py b/api/src/backend/tasks/tasks.py index 7a1ff54131..02a66e178c 100644 --- a/api/src/backend/tasks/tasks.py +++ b/api/src/backend/tasks/tasks.py @@ -738,7 +738,6 @@ class ScanReportRLSTask(RLSTask): name="scan-report", queue="scan-reports", ) -@set_tenant(keep_tenant=True) @handle_provider_deletion def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str): """ @@ -751,6 +750,11 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str): batch is being processed. Finally, the output files are compressed and uploaded to S3. + Rendering and uploading take tens of minutes, so every database access is + scoped to its own short `rls_transaction`. A task-wide transaction would hold + ACCESS SHARE on the tables it read for the whole run, and any DDL waiting on + one of those tables would queue every later reader behind itself. + Args: tenant_id (str): The tenant identifier. scan_id (str): The scan identifier. @@ -769,15 +773,21 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str): error, ) - # Check if the scan has findings - if not ScanSummary.objects.filter(scan_id=scan_id).exists(): - logger.info(f"No findings found for scan {scan_id}") - return {"upload": False} + with rls_transaction(tenant_id): + # Check if the scan has findings + if not ScanSummary.objects.filter(scan_id=scan_id).exists(): + logger.info(f"No findings found for scan {scan_id}") + return {"upload": False} - provider_obj = Provider.objects.get(id=provider_id) + # `secret` is selected eagerly because initialize_prowler_provider reads it + # but runs outside this transaction, where RLS would resolve it to nothing. + provider_obj = Provider.objects.select_related("secret").get(id=provider_id) + provider_uid = provider_obj.uid + provider_type = provider_obj.provider + + # Kept outside the transaction: provider initialization authenticates against + # the cloud provider, and that network call must not hold the providers lock. prowler_provider = initialize_prowler_provider(provider_obj) - provider_uid = provider_obj.uid - provider_type = provider_obj.provider # Per-framework exporters in `COMPLIANCE_CLASS_MAP` consume the legacy bulk. frameworks_bulk = Compliance.get_bulk(provider_type) @@ -818,19 +828,22 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str): universal_base_dir = os.path.dirname(out_dir) universal_output_filename = os.path.basename(out_dir) - scan_summary = FindingOutput._transform_findings_stats( - ScanSummary.objects.filter(scan_id=scan_id) - ) - - # Check if we need to generate ASFF output for AWS providers with SecurityHub integration - generate_asff = False - if provider_type == "aws": - security_hub_integrations = Integration.objects.filter( - integrationproviderrelationship__provider_id=provider_id, - integration_type=Integration.IntegrationChoices.AWS_SECURITY_HUB, - enabled=True, + with rls_transaction(tenant_id): + # `_transform_findings_stats` walks the queryset and follows its `scan` + # relation, so it has to run with the tenant context live. + scan_summary = FindingOutput._transform_findings_stats( + ScanSummary.objects.filter(scan_id=scan_id) ) - generate_asff = security_hub_integrations.exists() + + # Check if we need to generate ASFF output for AWS providers with SecurityHub integration + generate_asff = False + if provider_type == "aws": + security_hub_integrations = Integration.objects.filter( + integrationproviderrelationship__provider_id=provider_id, + integration_type=Integration.IntegrationChoices.AWS_SECURITY_HUB, + enabled=True, + ) + generate_asff = security_hub_integrations.exists() qs = ( Finding.all_objects.filter(tenant_id=tenant_id, scan_id=scan_id) @@ -943,10 +956,15 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str): # S3 integrations (need output_directory) with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS): - s3_integrations = Integration.objects.filter( - integrationproviderrelationship__provider_id=provider_id, - integration_type=Integration.IntegrationChoices.AMAZON_S3, - enabled=True, + # Materialized inside the transaction: a lazy queryset would run its query + # on the next access, once the tenant context is gone and RLS filters + # everything out. + s3_integrations = list( + Integration.objects.filter( + integrationproviderrelationship__provider_id=provider_id, + integration_type=Integration.IntegrationChoices.AMAZON_S3, + enabled=True, + ) ) if s3_integrations: @@ -977,7 +995,8 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str): else: final_location, did_upload = compressed, False - Scan.all_objects.filter(id=scan_id).update(output_location=final_location) + with rls_transaction(tenant_id): + Scan.all_objects.filter(id=scan_id).update(output_location=final_location) logger.info(f"Scan outputs at {final_location}") return { diff --git a/api/src/backend/tasks/tests/test_generate_outputs_rls.py b/api/src/backend/tasks/tests/test_generate_outputs_rls.py new file mode 100644 index 0000000000..bfbd74701a --- /dev/null +++ b/api/src/backend/tasks/tests/test_generate_outputs_rls.py @@ -0,0 +1,158 @@ +import pytest +from api.db_utils import rls_transaction +from api.models import ( + Integration, + IntegrationProviderRelationship, + Provider, + ProviderSecret, +) +from django.db import DataError, connections + + +@pytest.fixture +def enforce_rls(): + """ + Run the assertions under the unprivileged `test` role, on a fresh session. + + The suite normally connects as `prowler_admin`, a superuser, and superusers + bypass row-level security even on FORCE ROW LEVEL SECURITY tables. Under that + role a query with no tenant context still returns rows, so nothing here would + fail. `test` is the role the RLS policies are actually written for. + + The reconnect matters: a session that has already run an `rls_transaction` + reports the tenant variable as "" rather than NULL, which sends the policy down + a different branch. Starting clean keeps each test independent of suite order. + """ + connections["default"].close() + with connections["default"].cursor() as cursor: + cursor.execute("SET ROLE test;") + yield + with connections["default"].cursor() as cursor: + cursor.execute("RESET ROLE;") + connections["default"].close() + + +@pytest.fixture +def provider_with_secret(tenants_fixture): + tenant = tenants_fixture[0] + provider = Provider.objects.create( + tenant_id=tenant.id, + provider=Provider.ProviderChoices.AWS.value, + uid="123456789012", + alias="rls-scoping", + ) + ProviderSecret.objects.create( + tenant_id=tenant.id, + provider=provider, + secret_type=ProviderSecret.TypeChoices.STATIC, + secret={"key": "value"}, + name=provider.alias, + ) + integration = Integration.objects.create( + tenant_id=tenant.id, + enabled=True, + connected=True, + integration_type=Integration.IntegrationChoices.AMAZON_S3, + configuration={"key": "value"}, + credentials={"psswd": "1234"}, + ) + IntegrationProviderRelationship.objects.create( + tenant_id=tenant.id, + integration=integration, + provider=provider, + ) + return str(tenant.id), str(provider.id) + + +@pytest.mark.django_db(transaction=True) +class TestGenerateOutputsTenantScoping: + """ + Pins the assumptions that let `generate_outputs_task` drop `@set_tenant`: every + query it runs must sit inside an `rls_transaction`, and anything it reads after + one closes must already be in memory. + + `transaction=True` is required, not incidental. Plain `django_db` wraps the test + in its own atomic block, which demotes each `rls_transaction` to a savepoint and + leaves the transaction-scoped tenant variable set for the rest of the test. Every + assertion about a query running *outside* tenant context would then pass for the + wrong reason. + """ + + def test_rls_denies_reads_without_tenant_context( + self, provider_with_secret, enforce_rls + ): + # enforce_rls reconnects as the unprivileged `test` role for the whole + # test; pytest injects it by parameter name, so it is referenced + # explicitly to keep static analysers from flagging it as unused. + del enforce_rls + # Guard for the tests below: proves the `test` role really is subject to + # RLS, otherwise passing assertions here would mean nothing. + _, provider_id = provider_with_secret + + assert not Provider.objects.filter(id=provider_id).exists() + + def test_rls_allows_reads_inside_an_rls_transaction( + self, provider_with_secret, enforce_rls + ): + del enforce_rls + tenant_id, provider_id = provider_with_secret + + with rls_transaction(tenant_id): + assert Provider.objects.filter(id=provider_id).exists() + + def test_select_related_secret_survives_the_transaction( + self, provider_with_secret, enforce_rls + ): + del enforce_rls + tenant_id, provider_id = provider_with_secret + + with rls_transaction(tenant_id): + provider = Provider.objects.select_related("secret").get(id=provider_id) + + # initialize_prowler_provider reads `secret` out here, with no tenant + # context. It only works because select_related already cached it. + assert provider.secret.secret == {"key": "value"} + + def test_secret_without_select_related_is_lost_after_the_transaction( + self, provider_with_secret, enforce_rls + ): + del enforce_rls + tenant_id, provider_id = provider_with_secret + + with rls_transaction(tenant_id): + provider = Provider.objects.get(id=provider_id) + + # The regression select_related guards against: a lazy relation resolved + # after the transaction never reaches the row. + with pytest.raises(DataError): + provider.secret + + def test_lazy_queryset_does_not_reach_the_rows_after_the_transaction( + self, provider_with_secret, enforce_rls + ): + del enforce_rls + tenant_id, provider_id = provider_with_secret + + with rls_transaction(tenant_id): + lazy = Integration.objects.filter( + integrationproviderrelationship__provider_id=provider_id, + integration_type=Integration.IntegrationChoices.AMAZON_S3, + enabled=True, + ) + materialized = list( + Integration.objects.filter( + integrationproviderrelationship__provider_id=provider_id, + integration_type=Integration.IntegrationChoices.AMAZON_S3, + enabled=True, + ) + ) + + # The lazy queryset only runs its query here, and by then the tenant + # variable is gone. Once a transaction has set and discarded it, + # current_setting returns "" rather than NULL, so the policy's + # ""::uuid cast errors instead of taking the NULL branch that yields no + # rows (see test_rls_denies_reads_without_tenant_context). Wrong either + # way, which is why the S3 lookup is materialized with list(). + assert len(materialized) == 1 + with pytest.raises(DataError): + list(lazy) diff --git a/api/src/backend/tasks/tests/test_tasks.py b/api/src/backend/tasks/tests/test_tasks.py index 6622ab10a5..b9f7746e2c 100644 --- a/api/src/backend/tasks/tests/test_tasks.py +++ b/api/src/backend/tasks/tests/test_tasks.py @@ -290,14 +290,14 @@ class TestGenerateOutputs: @patch("tasks.tasks.get_compliance_frameworks") @patch("tasks.tasks.Compliance.get_bulk") @patch("tasks.tasks.initialize_prowler_provider") - @patch("tasks.tasks.Provider.objects.get") + @patch("tasks.tasks.Provider.objects.select_related") @patch("tasks.tasks.ScanSummary.objects.filter") @patch("tasks.tasks.Finding.all_objects.filter") def test_generate_outputs_happy_path( self, mock_finding_filter, mock_scan_summary_filter, - mock_provider_get, + mock_provider_select_related, mock_initialize_provider, mock_compliance_get_bulk, mock_get_available_frameworks, @@ -309,7 +309,7 @@ class TestGenerateOutputs: mock_provider = MagicMock() mock_provider.uid = "provider-uid" mock_provider.provider = "aws" - mock_provider_get.return_value = mock_provider + mock_provider_select_related.return_value.get.return_value = mock_provider prowler_provider = MagicMock() mock_initialize_provider.return_value = prowler_provider @@ -375,7 +375,7 @@ class TestGenerateOutputs: def test_generate_outputs_fails_upload(self): with ( patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter, - patch("tasks.tasks.Provider.objects.get"), + patch("tasks.tasks.Provider.objects.select_related"), patch("tasks.tasks.initialize_prowler_provider"), patch("tasks.tasks.Compliance.get_bulk"), patch("tasks.tasks.get_compliance_frameworks"), @@ -446,7 +446,10 @@ class TestGenerateOutputs: with ( patch("tasks.tasks.get_prowler_provider_compliance", return_value={}), patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter, - patch("tasks.tasks.Provider.objects.get", return_value=mock_provider), + patch( + "tasks.tasks.Provider.objects.select_related", + **{"return_value.get.return_value": mock_provider}, + ), patch("tasks.tasks.initialize_prowler_provider"), patch("tasks.tasks.Compliance.get_bulk", return_value={"cis": MagicMock()}), patch("tasks.tasks.get_compliance_frameworks", return_value=["cis"]), @@ -523,7 +526,7 @@ class TestGenerateOutputs: with ( patch("tasks.tasks.ScanSummary.objects.filter") as mock_summary, - patch("tasks.tasks.Provider.objects.get"), + patch("tasks.tasks.Provider.objects.select_related"), patch("tasks.tasks.initialize_prowler_provider"), patch("tasks.tasks.Compliance.get_bulk"), patch("tasks.tasks.get_compliance_frameworks", return_value=[]), @@ -603,8 +606,12 @@ class TestGenerateOutputs: patch("tasks.tasks.get_prowler_provider_compliance", return_value={}), patch("tasks.tasks.ScanSummary.objects.filter") as mock_summary, patch( - "tasks.tasks.Provider.objects.get", - return_value=MagicMock(uid="UID", provider="aws"), + "tasks.tasks.Provider.objects.select_related", + **{ + "return_value.get.return_value": MagicMock( + uid="UID", provider="aws" + ) + }, ), patch("tasks.tasks.initialize_prowler_provider"), patch( @@ -677,7 +684,10 @@ class TestGenerateOutputs: with ( patch("tasks.tasks.get_prowler_provider_compliance", return_value={}), patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter, - patch("tasks.tasks.Provider.objects.get", return_value=mock_provider), + patch( + "tasks.tasks.Provider.objects.select_related", + **{"return_value.get.return_value": mock_provider}, + ), patch("tasks.tasks.initialize_prowler_provider"), patch("tasks.tasks.Compliance.get_bulk", return_value={"cis": MagicMock()}), patch("tasks.tasks.get_compliance_frameworks", return_value=["cis"]), @@ -735,7 +745,7 @@ class TestGenerateOutputs: """Test that generate_outputs_task only processes enabled S3 integrations.""" with ( patch("tasks.tasks.ScanSummary.objects.filter") as mock_summary, - patch("tasks.tasks.Provider.objects.get"), + patch("tasks.tasks.Provider.objects.select_related"), patch("tasks.tasks.initialize_prowler_provider"), patch("tasks.tasks.Compliance.get_bulk"), patch("tasks.tasks.get_compliance_frameworks", return_value=[]), @@ -1123,7 +1133,7 @@ class TestCheckIntegrationsTask: @patch("tasks.tasks.s3_integration_task") @patch("tasks.tasks.Integration.objects.filter") @patch("tasks.tasks.ScanSummary.objects.filter") - @patch("tasks.tasks.Provider.objects.get") + @patch("tasks.tasks.Provider.objects.select_related") @patch("tasks.tasks.initialize_prowler_provider") @patch("tasks.tasks.Compliance.get_bulk") @patch("tasks.tasks.get_compliance_frameworks") @@ -1148,7 +1158,7 @@ class TestCheckIntegrationsTask: mock_get_frameworks, mock_compliance_bulk, mock_initialize_provider, - mock_provider_get, + mock_provider_select_related, mock_scan_summary, mock_integration_filter, mock_s3_task, @@ -1164,7 +1174,7 @@ class TestCheckIntegrationsTask: mock_provider = MagicMock() mock_provider.uid = "aws-account-123" mock_provider.provider = "aws" - mock_provider_get.return_value = mock_provider + mock_provider_select_related.return_value.get.return_value = mock_provider # Mock SecurityHub integration exists mock_security_hub_integrations = MagicMock() @@ -1253,7 +1263,7 @@ class TestCheckIntegrationsTask: @patch("tasks.tasks.s3_integration_task") @patch("tasks.tasks.Integration.objects.filter") @patch("tasks.tasks.ScanSummary.objects.filter") - @patch("tasks.tasks.Provider.objects.get") + @patch("tasks.tasks.Provider.objects.select_related") @patch("tasks.tasks.initialize_prowler_provider") @patch("tasks.tasks.Compliance.get_bulk") @patch("tasks.tasks.get_compliance_frameworks") @@ -1278,7 +1288,7 @@ class TestCheckIntegrationsTask: mock_get_frameworks, mock_compliance_bulk, mock_initialize_provider, - mock_provider_get, + mock_provider_select_related, mock_scan_summary, mock_integration_filter, mock_s3_task, @@ -1294,7 +1304,7 @@ class TestCheckIntegrationsTask: mock_provider = MagicMock() mock_provider.uid = "aws-account-123" mock_provider.provider = "aws" - mock_provider_get.return_value = mock_provider + mock_provider_select_related.return_value.get.return_value = mock_provider # Mock NO SecurityHub integration mock_security_hub_integrations = MagicMock() @@ -1378,7 +1388,7 @@ class TestCheckIntegrationsTask: @patch("tasks.tasks.get_prowler_provider_compliance", return_value={}) @patch("tasks.tasks.ScanSummary.objects.filter") - @patch("tasks.tasks.Provider.objects.get") + @patch("tasks.tasks.Provider.objects.select_related") @patch("tasks.tasks.initialize_prowler_provider") @patch("tasks.tasks.Compliance.get_bulk") @patch("tasks.tasks.get_compliance_frameworks") @@ -1403,7 +1413,7 @@ class TestCheckIntegrationsTask: mock_get_frameworks, mock_compliance_bulk, mock_initialize_provider, - mock_provider_get, + mock_provider_select_related, mock_scan_summary, mock_get_prowler_compliance, ): @@ -1417,7 +1427,7 @@ class TestCheckIntegrationsTask: mock_provider = MagicMock() mock_provider.uid = "azure-subscription-123" mock_provider.provider = "azure" # Non-AWS provider - mock_provider_get.return_value = mock_provider + mock_provider_select_related.return_value.get.return_value = mock_provider # Mock other necessary components mock_initialize_provider.return_value = MagicMock()