fix(api): scope writer transactions in the scan report task

generate_outputs_task ran inside one transaction for its entire duration
because @set_tenant wraps the task to set the RLS tenant variable, which is
transaction-scoped. The task runs for a long time, so the writer connection
held a lock on the providers table for work that no longer touched it.

Drop the decorator and scope each writer access to its own short
rls_transaction, the pattern already used elsewhere in this file. Provider
is now fetched with select_related("secret") so provider initialization,
which authenticates over the network, runs outside the transaction. Also
fixes a lazy queryset that was built inside a transaction but evaluated
outside it, which only worked by accident under the old decorator.
This commit is contained in:
César Arroba committed 2026-07-15 18:51:18 +02:00
1 parent 0d899b3076
commit d558d35107
4 files changed
+232 -44

No files matched your search

@@ -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
+44 -25
View File
@@ -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 {
@@ -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)
+29 -19
View File
@@ -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()