ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2025-12-19 05:17:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(api): restore compliance overview endpoint (#9330)

Browse Source
This commit is contained in:
Adrián Jesús Peña Rodríguez
2025-11-27 13:31:15 +01:00
committed by GitHub
parent 79ec53bfc5
commit 7e0c5540bb
5 changed files with 218 additions and 639 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
api/CHANGELOG.md
View File

@@ -2,6 +2,13 @@
All notable changes to the **Prowler API** are documented in this file.
## [1.16.0] (Unreleased)
### Changed
- Restore the compliance overview endpoint's mandatory filters [(#9330)](https://github.com/prowler-cloud/prowler/pull/9330)
---
## [1.15.1] (Prowler v5.14.1)
### Fixed

8
api/src/backend/api/filters.py
View File

@@ -761,14 +761,6 @@ class RoleFilter(FilterSet):
class ComplianceOverviewFilter(FilterSet):
inserted_at = DateFilter(field_name="inserted_at", lookup_expr="date")
scan_id = UUIDFilter(field_name="scan_id")
provider_id = UUIDFilter(field_name="scan__provider__id", lookup_expr="exact")
provider_id__in = UUIDInFilter(field_name="scan__provider__id", lookup_expr="in")
provider_type = ChoiceFilter(
field_name="scan__provider__provider", choices=Provider.ProviderChoices.choices
)
provider_type__in = ChoiceInFilter(
field_name="scan__provider__provider", choices=Provider.ProviderChoices.choices
)
region = CharFilter(field_name="region")
class Meta:

334
api/src/backend/api/specs/v1.yaml
View File

@@ -283,11 +283,8 @@ paths:
/api/v1/compliance-overviews:
get:
operationId: compliance_overviews_list
description: Retrieve an overview of all compliance frameworks. If scan_id is
provided, returns compliance data for that specific scan. If scan_id is omitted,
returns compliance data aggregated from the latest completed scan of each
provider.
summary: List compliance overviews
description: Retrieve an overview of all the compliance in a given scan.
summary: List compliance overviews for a scan
parameters:
- in: query
name: fields[compliance-overviews]
@@ -346,32 +343,6 @@ paths:
schema:
type: string
format: date-time
- in: query
name: filter[provider_id]
schema:
type: string
format: uuid
description: Filter by specific provider ID.
- in: query
name: filter[provider_id__in]
schema:
type: array
items:
type: string
format: uuid
description: Filter by multiple provider IDs (comma-separated).
- in: query
name: filter[provider_type]
schema:
type: string
description: Filter by provider type (e.g., aws, azure, gcp).
- in: query
name: filter[provider_type__in]
schema:
type: array
items:
type: string
description: Filter by multiple provider types (comma-separated).
- in: query
name: filter[region]
schema:
@@ -394,8 +365,8 @@ paths:
schema:
type: string
format: uuid
description: Optional scan ID. If provided, returns compliance for that scan.
If omitted, returns compliance for the latest completed scan per provider.
description: Related scan ID.
required: true
- name: filter[search]
required: false
in: query
@@ -635,77 +606,6 @@ paths:
schema:
type: string
format: date-time
- in: query
name: filter[provider_id]
schema:
type: string
format: uuid
- in: query
name: filter[provider_id__in]
schema:
type: array
items:
type: string
format: uuid
description: Multiple values may be separated by commas.
explode: false
style: form
- in: query
name: filter[provider_type]
schema:
type: string
x-spec-enum-id: eca8c51e6bd28935
enum:
- aws
- azure
- gcp
- github
- iac
- kubernetes
- m365
- mongodbatlas
- oraclecloud
description: |-
* `aws` - AWS
* `azure` - Azure
* `gcp` - GCP
* `kubernetes` - Kubernetes
* `m365` - M365
* `github` - GitHub
* `mongodbatlas` - MongoDB Atlas
* `iac` - IaC
* `oraclecloud` - Oracle Cloud Infrastructure
- in: query
name: filter[provider_type__in]
schema:
type: array
items:
type: string
x-spec-enum-id: eca8c51e6bd28935
enum:
- aws
- azure
- gcp
- github
- iac
- kubernetes
- m365
- mongodbatlas
- oraclecloud
description: |-
Multiple values may be separated by commas.
* `aws` - AWS
* `azure` - Azure
* `gcp` - GCP
* `kubernetes` - Kubernetes
* `m365` - M365
* `github` - GitHub
* `mongodbatlas` - MongoDB Atlas
* `iac` - IaC
* `oraclecloud` - Oracle Cloud Infrastructure
explode: false
style: form
- in: query
name: filter[region]
schema:
@@ -5068,6 +4968,8 @@ paths:
type: string
enum:
- id
- provider_type
- region
- total
- fail
- muted
@@ -5200,6 +5102,10 @@ paths:
enum:
- id
- -id
- provider_type
- -provider_type
- region
- -region
- total
- -total
- fail
@@ -8984,116 +8890,12 @@ paths:
description: CSV file containing the compliance report
'404':
description: Compliance report not found
/api/v1/scans/{id}/report:
get:
operationId: scans_report_retrieve
description: Returns a ZIP file containing the requested report
summary: Download ZIP report
parameters:
- in: query
name: fields[scan-reports]
schema:
type: array
items:
type: string
enum:
- id
description: endpoint return only specific fields in the response on a per-type
basis by including a fields[TYPE] query parameter.
explode: false
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this scan.
required: true
tags:
- Scan
security:
- JWT or API Key: []
responses:
'200':
description: Report obtained successfully
'202':
description: The task is in progress
'403':
description: There is a problem with credentials
'404':
description: The scan has no reports, or the report generation task has
not started yet
/api/v1/scans/{id}/threatscore:
get:
operationId: scans_threatscore_retrieve
description: Download a specific threatscore report (e.g., 'prowler_threatscore_aws')
as a PDF file.
summary: Retrieve threatscore report
parameters:
- in: query
name: fields[scans]
schema:
type: array
items:
type: string
enum:
- name
- trigger
- state
- unique_resource_count
- progress
- duration
- provider
- task
- inserted_at
- started_at
- completed_at
- scheduled_at
- next_scan_at
- processor
- url
description: endpoint return only specific fields in the response on a per-type
basis by including a fields[TYPE] query parameter.
explode: false
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this scan.
required: true
- in: query
name: include
schema:
type: array
items:
type: string
enum:
- provider
description: include query parameter to allow the client to customize which
related resources should be returned.
explode: false
tags:
- Scan
security:
- JWT or API Key: []
responses:
'200':
description: PDF file containing the threatscore report
'202':
description: The task is in progress
'401':
description: API key missing or user not Authenticated
'403':
description: There is a problem with credentials
'404':
description: The scan has no threatscore reports, or the threatscore report
generation task has not started yet
/api/v1/scans/{id}/ens:
get:
operationId: scans_ens_retrieve
description: Download a specific ENS compliance report (e.g., 'prowler_ens_aws')
description: Download ENS RD2022 compliance report (e.g., 'ens_rd2022_aws')
as a PDF file.
summary: Retrieve ENS compliance report
summary: Retrieve ENS RD2022 compliance report
parameters:
- in: query
name: fields[scans]
@@ -9220,6 +9022,110 @@ paths:
'404':
description: The scan has no NIS2 reports, or the NIS2 report generation
task has not started yet
/api/v1/scans/{id}/report:
get:
operationId: scans_report_retrieve
description: Returns a ZIP file containing the requested report
summary: Download ZIP report
parameters:
- in: query
name: fields[scan-reports]
schema:
type: array
items:
type: string
enum:
- id
description: endpoint return only specific fields in the response on a per-type
basis by including a fields[TYPE] query parameter.
explode: false
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this scan.
required: true
tags:
- Scan
security:
- JWT or API Key: []
responses:
'200':
description: Report obtained successfully
'202':
description: The task is in progress
'403':
description: There is a problem with credentials
'404':
description: The scan has no reports, or the report generation task has
not started yet
/api/v1/scans/{id}/threatscore:
get:
operationId: scans_threatscore_retrieve
description: Download a specific threatscore report (e.g., 'prowler_threatscore_aws')
as a PDF file.
summary: Retrieve threatscore report
parameters:
- in: query
name: fields[scans]
schema:
type: array
items:
type: string
enum:
- name
- trigger
- state
- unique_resource_count
- progress
- duration
- provider
- task
- inserted_at
- started_at
- completed_at
- scheduled_at
- next_scan_at
- processor
- url
description: endpoint return only specific fields in the response on a per-type
basis by including a fields[TYPE] query parameter.
explode: false
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this scan.
required: true
- in: query
name: include
schema:
type: array
items:
type: string
enum:
- provider
description: include query parameter to allow the client to customize which
related resources should be returned.
explode: false
tags:
- Scan
security:
- JWT or API Key: []
responses:
'200':
description: PDF file containing the threatscore report
'202':
description: The task is in progress
'401':
description: API key missing or user not Authenticated
'403':
description: There is a problem with credentials
'404':
description: The scan has no threatscore reports, or the threatscore report
generation task has not started yet
/api/v1/schedules/daily:
post:
operationId: schedules_daily_create
@@ -13558,6 +13464,11 @@ components:
properties:
id:
type: string
readOnly: true
provider_type:
type: string
region:
type: string
total:
type: integer
fail:
@@ -13567,7 +13478,8 @@ components:
pass:
type: integer
required:
- id
- provider_type
- region
- total
- fail
- muted

159
api/src/backend/api/tests/test_views.py
View File

@@ -35,8 +35,6 @@ from rest_framework.response import Response
from api.compliance import get_compliance_frameworks
from api.db_router import MainRouter
from api.models import (
ComplianceOverviewSummary,
ComplianceRequirementOverview,
Finding,
Integration,
Invitation,
@@ -57,7 +55,6 @@ from api.models import (
Scan,
ScanSummary,
StateChoices,
StatusChoices,
Task,
TenantAPIKey,
ThreatScoreSnapshot,
@@ -5315,9 +5312,11 @@ class TestUserRoleRelationshipViewSet:
def test_create_relationship_already_exists(
self, authenticated_client, roles_fixture, create_test_user
):
# Only add Role One (which has manage_account=True) to ensure
# the second request has permission to add roles
data = {
"data": [
{"type": "roles", "id": str(role.id)} for role in roles_fixture[:2]
{"type": "roles", "id": str(roles_fixture[0].id)},
]
}
authenticated_client.post(
@@ -5820,44 +5819,16 @@ class TestProviderGroupMembershipViewSet:
@pytest.mark.django_db
class TestComplianceOverviewViewSet:
@pytest.fixture(autouse=True)
def mock_backfill_task(self):
with patch("api.v1.views.backfill_compliance_summaries_task.delay") as mock:
yield mock
def test_compliance_overview_list_none(
self,
authenticated_client,
tenants_fixture,
providers_fixture,
mock_backfill_task,
):
tenant = tenants_fixture[0]
provider = providers_fixture[0]
scan = Scan.objects.create(
name="empty-compliance-scan",
provider=provider,
trigger=Scan.TriggerChoices.MANUAL,
state=StateChoices.COMPLETED,
tenant=tenant,
)
def test_compliance_overview_list_none(self, authenticated_client):
response = authenticated_client.get(
reverse("complianceoverview-list"),
{"filter[scan_id]": str(scan.id)},
{"filter[scan_id]": "8d20ac7d-4cbc-435e-85f4-359be37af821"},
)
assert response.status_code == status.HTTP_200_OK
assert len(response.json()["data"]) == 0
mock_backfill_task.assert_called_once()
_, kwargs = mock_backfill_task.call_args
assert kwargs["scan_id"] == str(scan.id)
assert str(kwargs["tenant_id"]) == str(tenant.id)
def test_compliance_overview_list(
self,
authenticated_client,
compliance_requirements_overviews_fixture,
mock_backfill_task,
self, authenticated_client, compliance_requirements_overviews_fixture
):
# List compliance overviews with existing data
requirement_overview1 = compliance_requirements_overviews_fixture[0]
@@ -5887,112 +5858,6 @@ class TestComplianceOverviewViewSet:
assert "requirements_failed" in attributes
assert "requirements_manual" in attributes
assert "total_requirements" in attributes
mock_backfill_task.assert_called_once()
_, kwargs = mock_backfill_task.call_args
assert kwargs["scan_id"] == scan_id
def test_compliance_overview_list_uses_preaggregated_summaries(
self,
authenticated_client,
tenants_fixture,
providers_fixture,
mock_backfill_task,
):
tenant = tenants_fixture[0]
provider = providers_fixture[0]
scan = Scan.objects.create(
name="preaggregated-scan",
provider=provider,
trigger=Scan.TriggerChoices.MANUAL,
state=StateChoices.COMPLETED,
tenant=tenant,
)
ComplianceRequirementOverview.objects.create(
tenant=tenant,
scan=scan,
compliance_id="cis_1.4_aws",
framework="CIS-1.4-AWS",
version="1.4",
description="CIS AWS Foundations Benchmark v1.4.0",
region="eu-west-1",
requirement_id="framework-metadata",
requirement_status=StatusChoices.PASS,
passed_checks=1,
failed_checks=0,
total_checks=1,
)
ComplianceOverviewSummary.objects.create(
tenant=tenant,
scan=scan,
compliance_id="cis_1.4_aws",
requirements_passed=5,
requirements_failed=1,
requirements_manual=2,
total_requirements=8,
)
response = authenticated_client.get(
reverse("complianceoverview-list"),
{"filter[scan_id]": str(scan.id)},
)
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert len(data) == 1
overview = data[0]
assert overview["id"] == "cis_1.4_aws"
assert overview["attributes"]["requirements_passed"] == 5
assert overview["attributes"]["requirements_failed"] == 1
assert overview["attributes"]["requirements_manual"] == 2
assert overview["attributes"]["total_requirements"] == 8
assert "framework" in overview["attributes"]
assert "version" in overview["attributes"]
mock_backfill_task.assert_not_called()
def test_compliance_overview_region_filter_skips_backfill(
self,
authenticated_client,
compliance_requirements_overviews_fixture,
mock_backfill_task,
):
requirement_overview = compliance_requirements_overviews_fixture[0]
scan_id = str(requirement_overview.scan.id)
response = authenticated_client.get(
reverse("complianceoverview-list"),
{
"filter[scan_id]": scan_id,
"filter[region]": requirement_overview.region,
},
)
assert response.status_code == status.HTTP_200_OK
assert len(response.json()["data"]) >= 1
mock_backfill_task.assert_not_called()
def test_compliance_overview_list_without_scan_id(
self, authenticated_client, compliance_requirements_overviews_fixture
):
# Ensure the endpoint works without passing a scan filter
response = authenticated_client.get(reverse("complianceoverview-list"))
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert len(data) == 3
# Validate payload structure
first_item = data[0]
assert "id" in first_item
assert "attributes" in first_item
attributes = first_item["attributes"]
assert "framework" in attributes
assert "version" in attributes
assert "requirements_passed" in attributes
assert "requirements_failed" in attributes
assert "requirements_manual" in attributes
assert "total_requirements" in attributes
def test_compliance_overview_metadata(
self, authenticated_client, compliance_requirements_overviews_fixture
@@ -6146,11 +6011,6 @@ class TestComplianceOverviewViewSet:
requirement_overview1 = compliance_requirements_overviews_fixture[0]
scan_id = str(requirement_overview1.scan.id)
# Remove existing compliance data so the view falls back to task checks
scan = requirement_overview1.scan
ComplianceOverviewSummary.objects.filter(scan=scan).delete()
ComplianceRequirementOverview.objects.filter(scan=scan).delete()
# Mock a running task
with patch.object(
ComplianceOverviewViewSet, "get_task_response_if_running"
@@ -6178,11 +6038,6 @@ class TestComplianceOverviewViewSet:
requirement_overview1 = compliance_requirements_overviews_fixture[0]
scan_id = str(requirement_overview1.scan.id)
# Remove existing compliance data so the view falls back to task checks
scan = requirement_overview1.scan
ComplianceOverviewSummary.objects.filter(scan=scan).delete()
ComplianceRequirementOverview.objects.filter(scan=scan).delete()
# Mock a failed task
with patch.object(
ComplianceOverviewViewSet, "get_task_response_if_running"
@@ -6206,8 +6061,6 @@ class TestComplianceOverviewViewSet:
("framework", "framework", 1),
("version", "version", 1),
("region", "region", 1),
("region__in", "region", 1),
("region.in", "region", 1),
],
)
def test_compliance_overview_filters(

349
api/src/backend/api/v1/views.py
View File

@@ -75,7 +75,6 @@ from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from tasks.beat import schedule_provider_scan
from tasks.jobs.export import get_s3_client
from tasks.tasks import (
backfill_compliance_summaries_task,
backfill_scan_resource_summaries_task,
check_integration_connection_task,
check_lighthouse_connection_task,
@@ -126,7 +125,6 @@ from api.filters import (
UserFilter,
)
from api.models import (
ComplianceOverviewSummary,
ComplianceRequirementOverview,
Finding,
Integration,
@@ -3359,50 +3357,15 @@ class RoleProviderGroupRelationshipView(RelationshipView, BaseRLSViewSet):
@extend_schema_view(
list=extend_schema(
tags=["Compliance Overview"],
summary="List compliance overviews",
description=(
"Retrieve an overview of all compliance frameworks. "
"If scan_id is provided, returns compliance data for that specific scan. "
"If scan_id is omitted, returns compliance data aggregated from the latest completed scan of each provider."
),
summary="List compliance overviews for a scan",
description="Retrieve an overview of all the compliance in a given scan.",
parameters=[
OpenApiParameter(
name="filter[scan_id]",
required=False,
required=True,
type=OpenApiTypes.UUID,
location=OpenApiParameter.QUERY,
description=(
"Optional scan ID. If provided, returns compliance for that scan. "
"If omitted, returns compliance for the latest completed scan per provider."
),
),
OpenApiParameter(
name="filter[provider_id]",
required=False,
type=OpenApiTypes.UUID,
location=OpenApiParameter.QUERY,
description="Filter by specific provider ID.",
),
OpenApiParameter(
name="filter[provider_id__in]",
required=False,
type={"type": "array", "items": {"type": "string", "format": "uuid"}},
location=OpenApiParameter.QUERY,
description="Filter by multiple provider IDs (comma-separated).",
),
OpenApiParameter(
name="filter[provider_type]",
required=False,
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by provider type (e.g., aws, azure, gcp).",
),
OpenApiParameter(
name="filter[provider_type__in]",
required=False,
type={"type": "array", "items": {"type": "string"}},
location=OpenApiParameter.QUERY,
description="Filter by multiple provider types (comma-separated).",
description="Related scan ID.",
),
],
responses={
@@ -3559,114 +3522,6 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
def retrieve(self, request, *args, **kwargs):
raise MethodNotAllowed(method="GET")
def _compliance_summaries_queryset(self, scan_id):
"""Return pre-aggregated summaries constrained by RBAC visibility."""
role = get_role(self.request.user)
unlimited_visibility = getattr(
role, Permissions.UNLIMITED_VISIBILITY.value, False
)
summaries = ComplianceOverviewSummary.objects.filter(
tenant_id=self.request.tenant_id,
scan_id=scan_id,
)
if not unlimited_visibility:
providers = Provider.all_objects.filter(
provider_groups__in=role.provider_groups.all()
).distinct()
summaries = summaries.filter(scan__provider__in=providers)
return summaries
def _get_compliance_template(self, *, provider=None, scan_id=None):
"""Return the compliance template for the given provider or scan."""
if provider is None and scan_id is not None:
scan = Scan.all_objects.select_related("provider").get(pk=scan_id)
provider = scan.provider
if not provider:
return {}
return PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE.get(provider.provider, {})
def _aggregate_compliance_overview(self, queryset, template_metadata=None):
"""
Aggregate requirement rows into compliance overview dictionaries.
Args:
queryset: ComplianceRequirementOverview queryset already filtered.
template_metadata: Optional dict mapping compliance_id -> metadata.
"""
template_metadata = template_metadata or {}
requirement_status_subquery = queryset.values(
"compliance_id", "requirement_id"
).annotate(
fail_count=Count("id", filter=Q(requirement_status="FAIL")),
pass_count=Count("id", filter=Q(requirement_status="PASS")),
total_count=Count("id"),
)
compliance_data = {}
fallback_metadata = {
item["compliance_id"]: {
"framework": item["framework"],
"version": item["version"],
}
for item in queryset.values(
"compliance_id", "framework", "version"
).distinct()
}
for item in requirement_status_subquery:
compliance_id = item["compliance_id"]
if item["fail_count"] > 0:
req_status = "FAIL"
elif item["pass_count"] == item["total_count"]:
req_status = "PASS"
else:
req_status = "MANUAL"
compliance_status = compliance_data.setdefault(
compliance_id,
{
"total_requirements": 0,
"requirements_passed": 0,
"requirements_failed": 0,
"requirements_manual": 0,
},
)
compliance_status["total_requirements"] += 1
if req_status == "PASS":
compliance_status["requirements_passed"] += 1
elif req_status == "FAIL":
compliance_status["requirements_failed"] += 1
else:
compliance_status["requirements_manual"] += 1
response_data = []
for compliance_id, data in compliance_data.items():
template = template_metadata.get(compliance_id, {})
fallback = fallback_metadata.get(compliance_id, {})
response_data.append(
{
"id": compliance_id,
"compliance_id": compliance_id,
"framework": template.get("framework")
or fallback.get("framework", ""),
"version": template.get("version") or fallback.get("version", ""),
"requirements_passed": data["requirements_passed"],
"requirements_failed": data["requirements_failed"],
"requirements_manual": data["requirements_manual"],
"total_requirements": data["total_requirements"],
}
)
serializer = self.get_serializer(response_data, many=True)
return serializer.data
def _task_response_if_running(self, scan_id):
"""Check for an in-progress task only when no compliance data exists."""
try:
@@ -3681,135 +3536,95 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
def _list_with_region_filter(self, scan_id, region_filter):
"""
Fall back to detailed ComplianceRequirementOverview query when region filter is applied.
This uses the original aggregation logic across filtered regions.
"""
regions = region_filter.split(",") if "," in region_filter else [region_filter]
queryset = self.filter_queryset(self.get_queryset()).filter(
scan_id=scan_id,
region__in=regions,
)
data = self._aggregate_compliance_overview(queryset)
if data:
return Response(data)
task_response = self._task_response_if_running(scan_id)
if task_response:
return task_response
return Response(data)
def _list_without_region_aggregation(self, scan_id):
"""
Fall back aggregation when compliance summaries don't exist yet.
Aggregates ComplianceRequirementOverview data across ALL regions.
"""
queryset = self.filter_queryset(self.get_queryset()).filter(scan_id=scan_id)
compliance_template = self._get_compliance_template(scan_id=scan_id)
data = self._aggregate_compliance_overview(
queryset, template_metadata=compliance_template
)
if data:
return Response(data)
task_response = self._task_response_if_running(scan_id)
if task_response:
return task_response
return Response(data)
def list(self, request, *args, **kwargs):
scan_id = request.query_params.get("filter[scan_id]")
tenant_id = self.request.tenant_id
if scan_id:
# Specific scan requested - use optimized summaries with region support
region_filter = request.query_params.get(
"filter[region]"
) or request.query_params.get("filter[region__in]")
if region_filter:
# Fall back to detailed query with region filtering
return self._list_with_region_filter(scan_id, region_filter)
summaries = list(self._compliance_summaries_queryset(scan_id))
if not summaries:
# Trigger async backfill for next time
backfill_compliance_summaries_task.delay(
tenant_id=self.request.tenant_id, scan_id=scan_id
)
# Use fallback aggregation for this request
return self._list_without_region_aggregation(scan_id)
# Get compliance template for provider to enrich with framework/version
compliance_template = self._get_compliance_template(scan_id=scan_id)
# Convert to response format with framework/version enrichment
response_data = []
for summary in summaries:
compliance_metadata = compliance_template.get(summary.compliance_id, {})
response_data.append(
if not scan_id:
raise ValidationError(
[
{
"id": summary.compliance_id,
"compliance_id": summary.compliance_id,
"framework": compliance_metadata.get("framework", ""),
"version": compliance_metadata.get("version", ""),
"requirements_passed": summary.requirements_passed,
"requirements_failed": summary.requirements_failed,
"requirements_manual": summary.requirements_manual,
"total_requirements": summary.total_requirements,
"detail": "This query parameter is required.",
"status": 400,
"source": {"pointer": "filter[scan_id]"},
"code": "required",
}
)
]
)
try:
if task := self.get_task_response_if_running(
task_name="scan-compliance-overviews",
task_kwargs={"tenant_id": self.request.tenant_id, "scan_id": scan_id},
raise_on_not_found=False,
):
return task
except TaskFailedException:
return Response(
{"detail": "Task failed to generate compliance overview data."},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
queryset = self.filter_queryset(self.filter_queryset(self.get_queryset()))
serializer = self.get_serializer(response_data, many=True)
return Response(serializer.data)
else:
# No scan_id provided - use latest scans per provider
# First, check if provider filters are present
provider_id = request.query_params.get("filter[provider_id]")
provider_id__in = request.query_params.get("filter[provider_id__in]")
provider_type = request.query_params.get("filter[provider_type]")
provider_type__in = request.query_params.get("filter[provider_type__in]")
requirement_status_subquery = queryset.values(
"compliance_id", "requirement_id"
).annotate(
fail_count=Count("id", filter=Q(requirement_status="FAIL")),
pass_count=Count("id", filter=Q(requirement_status="PASS")),
total_count=Count("id"),
)
scan_filters = {"tenant_id": tenant_id, "state": StateChoices.COMPLETED}
compliance_data = {}
framework_info = {}
# Apply provider ID filters
if provider_id:
scan_filters["provider_id"] = provider_id
elif provider_id__in:
# Convert comma-separated string to list
provider_ids = [pid.strip() for pid in provider_id__in.split(",")]
scan_filters["provider_id__in"] = provider_ids
for item in queryset.values("compliance_id", "framework", "version").distinct():
framework_info[item["compliance_id"]] = {
"framework": item["framework"],
"version": item["version"],
}
# Apply provider type filters
if provider_type:
scan_filters["provider__provider"] = provider_type
elif provider_type__in:
# Convert comma-separated string to list
provider_types = [pt.strip() for pt in provider_type__in.split(",")]
scan_filters["provider__provider__in"] = provider_types
for item in requirement_status_subquery:
compliance_id = item["compliance_id"]
latest_scan_ids = (
Scan.all_objects.filter(**scan_filters)
.order_by("provider_id", "-inserted_at")
.distinct("provider_id")
.values_list("id", flat=True)
if item["fail_count"] > 0:
req_status = "FAIL"
elif item["pass_count"] == item["total_count"]:
req_status = "PASS"
else:
req_status = "MANUAL"
if compliance_id not in compliance_data:
compliance_data[compliance_id] = {
"total_requirements": 0,
"requirements_passed": 0,
"requirements_failed": 0,
"requirements_manual": 0,
}
compliance_data[compliance_id]["total_requirements"] += 1
if req_status == "PASS":
compliance_data[compliance_id]["requirements_passed"] += 1
elif req_status == "FAIL":
compliance_data[compliance_id]["requirements_failed"] += 1
else:
compliance_data[compliance_id]["requirements_manual"] += 1
response_data = []
for compliance_id, data in compliance_data.items():
framework = framework_info.get(compliance_id, {})
response_data.append(
{
"id": compliance_id,
"compliance_id": compliance_id,
"framework": framework.get("framework", ""),
"version": framework.get("version", ""),
"requirements_passed": data["requirements_passed"],
"requirements_failed": data["requirements_failed"],
"requirements_manual": data["requirements_manual"],
"total_requirements": data["total_requirements"],
}
)
base_queryset = self.get_queryset()
queryset = self.filter_queryset(
base_queryset.filter(scan_id__in=latest_scan_ids)
)
# Aggregate compliance data across latest scans
compliance_template = self._get_compliance_template()
data = self._aggregate_compliance_overview(
queryset, template_metadata=compliance_template
)
return Response(data)
serializer = self.get_serializer(response_data, many=True)
return Response(serializer.data)
@action(detail=False, methods=["get"], url_name="metadata")
def metadata(self, request):