feat(export): add api export system

This commit is contained in:
Adrián Jesús Peña Rodríguez committed 2025-02-10 16:09:18 +01:00
1 parent 80267c389b
commit a790a5060e
18 files changed
+657 -91

No files matched your search

+25 -16
View File
@@ -7,7 +7,7 @@ from rest_framework_json_api.serializers import ValidationError
from api.db_utils import POSTGRES_TENANT_VAR, SET_CONFIG_QUERY
def set_tenant(func):
def set_tenant(func=None, *, keep_tenant=False):
"""
Decorator to set the tenant context for a Celery task based on the provided tenant_id.
@@ -40,20 +40,29 @@ def set_tenant(func):
# The tenant context will be set before the task logic executes.
"""
@wraps(func)
@transaction.atomic
def wrapper(*args, **kwargs):
try:
tenant_id = kwargs.pop("tenant_id")
except KeyError:
raise KeyError("This task requires the tenant_id")
try:
uuid.UUID(tenant_id)
except ValueError:
raise ValidationError("Tenant ID must be a valid UUID")
with connection.cursor() as cursor:
cursor.execute(SET_CONFIG_QUERY, [POSTGRES_TENANT_VAR, tenant_id])
def decorator(func):
@wraps(func)
@transaction.atomic
def wrapper(*args, **kwargs):
try:
if not keep_tenant:
tenant_id = kwargs.pop("tenant_id")
else:
tenant_id = kwargs["tenant_id"]
except KeyError:
raise KeyError("This task requires the tenant_id")
try:
uuid.UUID(tenant_id)
except ValueError:
raise ValidationError("Tenant ID must be a valid UUID")
with connection.cursor() as cursor:
cursor.execute(SET_CONFIG_QUERY, [POSTGRES_TENANT_VAR, tenant_id])
return func(*args, **kwargs)
return func(*args, **kwargs)
return wrapper
return wrapper
if func is None:
return decorator
else:
return decorator(func)
@@ -0,0 +1,22 @@
# Generated by Django 5.1.5 on 2025-02-07 10:59
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("api", "0008_daily_scheduled_tasks_update"),
]
operations = [
migrations.AddField(
model_name="scan",
name="output_path",
field=models.CharField(blank=True, max_length=200, null=True),
),
migrations.AddField(
model_name="scan",
name="upload_to_s3",
field=models.BooleanField(blank=True, null=True),
),
]
+2
View File
@@ -414,6 +414,8 @@ class Scan(RowLevelSecurityProtectedModel):
scheduler_task = models.ForeignKey(
PeriodicTask, on_delete=models.CASCADE, null=True, blank=True
)
output_path = models.CharField(blank=True, null=True, max_length=200)
upload_to_s3 = models.BooleanField(blank=True, null=True)
# TODO: mutelist foreign key
class Meta(RowLevelSecurityProtectedModel.Meta):
+1 -3
View File
@@ -88,9 +88,7 @@ class RowLevelSecurityConstraint(models.BaseConstraint):
f"{grant_queries}{self.grant_sql_query.format(statement=statement)}"
)
full_create_sql_query = (
f"{self.rls_sql_query}" f"{policy_queries}" f"{grant_queries}"
)
full_create_sql_query = f"{self.rls_sql_query}{policy_queries}{grant_queries}"
table_name = model._meta.db_table
if self.partition_name:
+4 -3
View File
@@ -274,9 +274,10 @@ class TestValidateInvitation:
expired_time = datetime.now(timezone.utc) - timedelta(days=1)
invitation.expires_at = expired_time
with patch("api.utils.Invitation.objects.using") as mock_using, patch(
"api.utils.datetime"
) as mock_datetime:
with (
patch("api.utils.Invitation.objects.using") as mock_using,
patch("api.utils.datetime") as mock_datetime,
):
mock_db = mock_using.return_value
mock_db.get.return_value = invitation
mock_datetime.now.return_value = datetime.now(timezone.utc)
+21 -21
View File
@@ -847,9 +847,9 @@ class TestProviderViewSet:
included_data = response.json()["included"]
for expected_type in expected_resources:
assert any(
d.get("type") == expected_type for d in included_data
), f"Expected type '{expected_type}' not found in included data"
assert any(d.get("type") == expected_type for d in included_data), (
f"Expected type '{expected_type}' not found in included data"
)
def test_providers_retrieve(self, authenticated_client, providers_fixture):
provider1, *_ = providers_fixture
@@ -2226,9 +2226,9 @@ class TestResourceViewSet:
included_data = response.json()["included"]
for expected_type in expected_resources:
assert any(
d.get("type") == expected_type for d in included_data
), f"Expected type '{expected_type}' not found in included data"
assert any(d.get("type") == expected_type for d in included_data), (
f"Expected type '{expected_type}' not found in included data"
)
@pytest.mark.parametrize(
"filter_name, filter_value, expected_count",
@@ -2412,9 +2412,9 @@ class TestFindingViewSet:
included_data = response.json()["included"]
for expected_type in expected_resources:
assert any(
d.get("type") == expected_type for d in included_data
), f"Expected type '{expected_type}' not found in included data"
assert any(d.get("type") == expected_type for d in included_data), (
f"Expected type '{expected_type}' not found in included data"
)
@pytest.mark.parametrize(
"filter_name, filter_value, expected_count",
@@ -2692,9 +2692,9 @@ class TestJWTFields:
reverse("token-obtain"), data, format="json"
)
assert (
response.status_code == status.HTTP_200_OK
), f"Unexpected status code: {response.status_code}"
assert response.status_code == status.HTTP_200_OK, (
f"Unexpected status code: {response.status_code}"
)
access_token = response.data["attributes"]["access"]
payload = jwt.decode(access_token, options={"verify_signature": False})
@@ -2708,23 +2708,23 @@ class TestJWTFields:
# Verify expected fields
for field in expected_fields:
assert field in payload, f"The field '{field}' is not in the JWT"
assert (
payload[field] == expected_fields[field]
), f"The value of '{field}' does not match"
assert payload[field] == expected_fields[field], (
f"The value of '{field}' does not match"
)
# Verify time fields are integers
for time_field in ["exp", "iat", "nbf"]:
assert time_field in payload, f"The field '{time_field}' is not in the JWT"
assert isinstance(
payload[time_field], int
), f"The field '{time_field}' is not an integer"
assert isinstance(payload[time_field], int), (
f"The field '{time_field}' is not an integer"
)
# Verify identification fields are non-empty strings
for id_field in ["jti", "sub", "tenant_id"]:
assert id_field in payload, f"The field '{id_field}' is not in the JWT"
assert (
isinstance(payload[id_field], str) and payload[id_field]
), f"The field '{id_field}' is not a valid string"
assert isinstance(payload[id_field], str) and payload[id_field], (
f"The field '{id_field}' is not a valid string"
)
@pytest.mark.django_db
+8
View File
@@ -819,6 +819,14 @@ class ScanTaskSerializer(RLSSerializer):
]
class ScanReportSerializer(RLSSerializer):
class Meta:
model = Scan
fields = [
"id",
]
class ResourceTagSerializer(RLSSerializer):
"""
Serializer for the ResourceTag model
+86 -5
View File
@@ -1,9 +1,16 @@
import glob
import os
import boto3
from botocore.exceptions import ClientError, NoCredentialsError, ParamValidationError
from celery.result import AsyncResult
from config.env import env
from django.conf import settings as django_settings
from django.contrib.postgres.aggregates import ArrayAgg
from django.contrib.postgres.search import SearchQuery
from django.db import transaction
from django.db.models import Count, F, OuterRef, Prefetch, Q, Subquery, Sum
from django.http import HttpResponse
from django.db.models.functions import Coalesce
from django.urls import reverse
from django.utils.decorators import method_decorator
@@ -35,7 +42,6 @@ from tasks.tasks import (
check_provider_connection_task,
delete_provider_task,
delete_tenant_task,
perform_scan_summary_task,
perform_scan_task,
)
@@ -114,6 +120,7 @@ from api.v1.serializers import (
RoleSerializer,
RoleUpdateSerializer,
ScanCreateSerializer,
ScanReportSerializer,
ScanSerializer,
ScanUpdateSerializer,
ScheduleDailyCreateSerializer,
@@ -1073,6 +1080,8 @@ class ScanViewSet(BaseRLSViewSet):
return ScanCreateSerializer
elif self.action == "partial_update":
return ScanUpdateSerializer
elif self.action == "report":
return ScanReportSerializer
return super().get_serializer_class()
def partial_update(self, request, *args, **kwargs):
@@ -1090,6 +1099,82 @@ class ScanViewSet(BaseRLSViewSet):
)
return Response(data=read_serializer.data, status=status.HTTP_200_OK)
@extend_schema(
tags=["Scan"],
summary="Download ZIP report",
description="Returns a ZIP file containing the requested report",
request=ScanReportSerializer,
responses={
200: OpenApiResponse(description="Report obtained successfully"),
423: OpenApiResponse(
description="There is a problem with the AWS credentials"
),
},
)
@action(detail=True, methods=["get"], url_name="report")
def report(self, request, pk=None):
scan_instance = Scan.objects.get(pk=pk)
output_path = scan_instance.output_path
if 1 == 2 and scan_instance.upload_to_s3:
s3_client = None
try:
s3_client = boto3.client("s3")
s3_client.list_buckets()
except (ClientError, NoCredentialsError, ParamValidationError):
try:
s3_client = boto3.client(
"s3",
aws_access_key_id=env.str("DJANGO_ARTIFACTS_AWS_ACCESS_KEY_ID"),
aws_secret_access_key=env.str(
"DJANGO_ARTIFACTS_AWS_SECRET_ACCESS_KEY"
),
aws_session_token=env.str("DJANGO_ARTIFACTS_AWS_SESSION_TOKEN"),
region_name=env.str("DJANGO_ARTIFACTS_AWS_DEFAULT_REGION"),
)
s3_client.list_buckets()
except (ClientError, NoCredentialsError, ParamValidationError):
return Response(
{"detail": "There is a problem with the AWS credentials."},
status=status.HTTP_423_LOCKED,
)
bucket_name = env.str("DJANGO_ARTIFACTS_AWS_S3_OUTPUT_BUCKET")
try:
key = output_path[len(f"s3://{bucket_name}/") :]
s3_object = s3_client.get_object(Bucket=bucket_name, Key=key)
file_content = s3_object["Body"].read()
filename = os.path.basename(output_path.split("/")[-1])
except ClientError:
return Response(
{"detail": "Error accessing cloud storage"},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
else:
zip_files = glob.glob(output_path)
if not zip_files:
return Response(
{"detail": "No local files found"}, status=status.HTTP_404_NOT_FOUND
)
try:
file_path = zip_files[0]
with open(file_path, "rb") as f:
file_content = f.read()
filename = os.path.basename(file_path)
except IOError:
return Response(
{"detail": "Error reading local file"},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
response = HttpResponse(
file_content, content_type="application/x-zip-compressed"
)
response["Content-Disposition"] = f'attachment; filename="{filename}"'
return response
def create(self, request, *args, **kwargs):
input_serializer = self.get_serializer(data=request.data)
input_serializer.is_valid(raise_exception=True)
@@ -1104,10 +1189,6 @@ class ScanViewSet(BaseRLSViewSet):
# Disabled for now
# checks_to_execute=scan.scanner_args.get("checks_to_execute"),
},
link=perform_scan_summary_task.si(
tenant_id=self.request.tenant_id,
scan_id=str(scan.id),
),
)
scan.task_id = task.id
+138
View File
@@ -0,0 +1,138 @@
import os
import zipfile
import boto3
from botocore.exceptions import ClientError
from celery.utils.log import get_task_logger
from config.env import env
from prowler.config.config import (
csv_file_suffix,
html_file_suffix,
json_ocsf_file_suffix,
output_file_timestamp,
)
from prowler.lib.outputs.csv.csv import CSV
from prowler.lib.outputs.html.html import HTML
from prowler.lib.outputs.ocsf.ocsf import OCSF
logger = get_task_logger(__name__)
tmp_output_directory = "/tmp/prowler_api_output"
# Predefined mapping for output formats and their configurations
OUTPUT_FORMATS_MAPPING = {
"csv": {
"class": CSV,
"suffix": csv_file_suffix,
"kwargs": {},
},
"json-ocsf": {"class": OCSF, "suffix": json_ocsf_file_suffix, "kwargs": {}},
"html": {"class": HTML, "suffix": html_file_suffix, "kwargs": {"stats": {}}},
}
# Mapping provider types to their identity components for output paths
PROVIDER_IDENTITY_MAP = {
"aws": lambda p: p.identity.account,
"azure": lambda p: p.identity.tenant_domain,
"gcp": lambda p: p.identity.profile,
"kubernetes": lambda p: p.identity.context.replace(":", "_").replace("/", "_"),
}
def _compress_output_files(output_directory: str) -> str:
"""
Compress output files from all configured output formats into a ZIP archive.
Args:
output_directory (str): The directory where the output files are located.
The function looks up all known suffixes in OUTPUT_FORMATS_MAPPING
and compresses those files into a single ZIP.
Returns:
str: The full path to the newly created ZIP archive.
"""
zip_path = f"{output_directory}.zip"
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
for suffix in [config["suffix"] for config in OUTPUT_FORMATS_MAPPING.values()]:
zipf.write(
f"{output_directory}{suffix}",
f"artifacts/{output_directory.split('/')[-1]}{suffix}",
)
return zip_path
def _upload_to_s3(tenant_id: str, zip_path: str, scan_id: str) -> str:
"""
Upload the specified ZIP file to an S3 bucket.
If the S3 bucket environment variables are not configured,
the function returns None without performing an upload.
Args:
tenant_id (str): The tenant identifier, used as part of the S3 key prefix.
zip_path (str): The local file system path to the ZIP file to be uploaded.
scan_id (str): The scan identifier, used as part of the S3 key prefix.
Returns:
str: The S3 URI of the uploaded file (e.g., "s3://<bucket>/<key>") if successful.
None: If the required environment variables for the S3 bucket are not set.
Raises:
botocore.exceptions.ClientError: If the upload attempt to S3 fails for any reason.
"""
if not env.str("DJANGO_ARTIFACTS_AWS_S3_OUTPUT_BUCKET", ""):
return
if env.str("DJANGO_ARTIFACTS_AWS_ACCESS_KEY_ID", ""):
s3 = boto3.client(
"s3",
aws_access_key_id=env.str("DJANGO_ARTIFACTS_AWS_ACCESS_KEY_ID"),
aws_secret_access_key=env.str("DJANGO_ARTIFACTS_AWS_SECRET_ACCESS_KEY"),
aws_session_token=env.str("DJANGO_ARTIFACTS_AWS_SESSION_TOKEN"),
region_name=env.str("DJANGO_ARTIFACTS_AWS_DEFAULT_REGION"),
)
else:
s3 = boto3.client("s3")
s3_key = f"{tenant_id}/{scan_id}/{os.path.basename(zip_path)}"
try:
s3.upload_file(
Filename=zip_path,
Bucket=env.str("DJANGO_ARTIFACTS_AWS_S3_OUTPUT_BUCKET"),
Key=s3_key,
)
return f"s3://{env.str('DJANGO_ARTIFACTS_AWS_S3_OUTPUT_BUCKET')}/{s3_key}"
except ClientError as e:
logger.error(f"S3 upload failed: {str(e)}")
raise e
def _generate_output_directory(
prowler_provider: object, tenant_id: str, scan_id: str
) -> str:
"""
Generate a file system path for the output directory of a prowler scan.
This function constructs the output directory path by combining a base
temporary output directory, the tenant ID, the scan ID, and details about
the prowler provider along with a timestamp. The resulting path is used to
store the output files of a prowler scan.
Note:
This function depends on two external variables:
- `tmp_output_directory`: The base directory where temporary outputs are stored.
- `output_file_timestamp`: A timestamp (as a string) used to uniquely identify the output.
Args:
prowler_provider (object): An identifier or descriptor for the prowler provider.
Typically, this is a string indicating the provider (e.g., "aws").
tenant_id (str): The unique identifier for the tenant.
scan_id (str): The unique identifier for the scan.
Returns:
str: The constructed file system path for the prowler scan output directory.
Example:
>>> _generate_output_directory("aws", "tenant-1234", "scan-5678")
'/tmp/tenant-1234/scan-5678/prowler-output-aws-2023-02-15T12:34:56'
"""
return (
f"{tmp_output_directory}/{tenant_id}/{scan_id}/prowler-output-"
f"{prowler_provider}-{output_file_timestamp}"
)
+139 -9
View File
@@ -1,14 +1,24 @@
from celery import shared_task
import os
from celery import chain, shared_task
from config.celery import RLSTask
from api.utils import initialize_prowler_provider
from django_celery_beat.models import PeriodicTask
from tasks.jobs.connection import check_provider_connection
from tasks.jobs.deletion import delete_provider, delete_tenant
from tasks.jobs.export import (
OUTPUT_FORMATS_MAPPING,
_compress_output_files,
_generate_output_directory,
_upload_to_s3,
)
from tasks.jobs.scan import aggregate_findings, perform_prowler_scan
from tasks.utils import get_next_execution_datetime
from api.db_utils import rls_transaction
from api.decorators import set_tenant
from api.models import Scan, StateChoices
from api.models import Finding, Provider, Scan, ScanSummary, StateChoices
from prowler.lib.outputs.finding import Finding as FindingOutput
@shared_task(base=RLSTask, name="provider-connection-check")
@@ -68,13 +78,20 @@ def perform_scan_task(
Returns:
dict: The result of the scan execution, typically including the status and results of the performed checks.
"""
return perform_prowler_scan(
result = perform_prowler_scan(
tenant_id=tenant_id,
scan_id=scan_id,
provider_id=provider_id,
checks_to_execute=checks_to_execute,
)
chain(
perform_scan_summary_task.s(tenant_id, scan_id),
generate_outputs.si(scan_id, provider_id, tenant_id=tenant_id),
).apply_async()
return result
@shared_task(base=RLSTask, bind=True, name="scan-perform-scheduled", queue="scans")
def perform_scheduled_scan_task(self, tenant_id: str, provider_id: str):
@@ -135,12 +152,11 @@ def perform_scheduled_scan_task(self, tenant_id: str, provider_id: str):
scheduler_task_id=periodic_task_instance.id,
)
perform_scan_summary_task.apply_async(
kwargs={
"tenant_id": tenant_id,
"scan_id": str(scan_instance.id),
}
)
chain(
perform_scan_summary_task.s(tenant_id, scan_instance.id),
generate_outputs.si(str(scan_instance.id), provider_id, tenant_id=tenant_id),
).apply_async()
return result
@@ -152,3 +168,117 @@ def perform_scan_summary_task(tenant_id: str, scan_id: str):
@shared_task(name="tenant-deletion")
def delete_tenant_task(tenant_id: str):
return delete_tenant(pk=tenant_id)
def batched(iterable, batch_size):
"""
Yield successive batches from an iterable.
Args:
iterable: An iterable source of items.
batch_size (int): The number of items per batch.
Yields:
tuple: A pair (batch, is_last_batch) where:
- batch (list): A list of items (with length equal to batch_size,
except possibly for the last batch).
- is_last_batch (bool): True if this is the final batch, False otherwise.
"""
batch = []
for item in iterable:
batch.append(item)
if len(batch) == batch_size:
yield batch, False
batch = []
yield batch, True
@shared_task(base=RLSTask, name="scan-output", queue="scans")
@set_tenant(keep_tenant=True)
def generate_outputs(scan_id: str, provider_id: str, tenant_id: str):
"""
Process findings in batches and generate output files in multiple formats.
This function retrieves findings associated with a scan, processes them
in batches of 50, and writes each batch to the corresponding output files.
It reuses output writer instances across batches, updates them with each
batch of transformed findings, and uses a flag to indicate when the final
batch is being processed. Finally, the output files are compressed and
uploaded to S3.
Args:
tenant_id (str): The tenant identifier.
scan_id (str): The scan identifier.
provider_id (str): The provider_id id to be used in generating outputs.
"""
# Generate and ensure the output directory exists
output_directory = _generate_output_directory(provider_id, tenant_id, scan_id)
os.makedirs("/".join(output_directory.split("/")[:-1]), exist_ok=True)
# Define auxiliary variables
output_writers = {}
scan_summary = FindingOutput._transform_findings_stats(
ScanSummary.objects.filter(scan_id=scan_id)
)
# Retrieve findings queryset
findings_qs = Finding.objects.filter(scan_id=scan_id)
# Process findings in batches
for batch, is_last_batch in batched(findings_qs.iterator(), 50):
finding_outputs = [
FindingOutput.transform_api_finding(finding) for finding in batch
]
# Generate output files
for mode, config in OUTPUT_FORMATS_MAPPING.items():
kwargs = dict(config.get("kwargs", {}))
if mode == "html":
kwargs["provider"] = initialize_prowler_provider(
Provider.objects.get(id=provider_id)
)
kwargs["stats"] = scan_summary
writer_class = config["class"]
if writer_class in output_writers:
# Reuse the existing writer and update with the new batch
writer = output_writers[writer_class]
writer.findings = finding_outputs
writer.close_file = is_last_batch
else:
# Create a new writer for this format
writer = writer_class(
findings=finding_outputs,
file_path=output_directory,
file_extension=config["suffix"],
)
writer.close_file = is_last_batch
output_writers[writer_class] = writer
# Write the current batch using the writer
writer.batch_write_data_to_file(**kwargs)
# Compress output files
output_directory = _compress_output_files(output_directory)
# Save to configured storage
uploaded = _upload_to_s3(tenant_id, output_directory, scan_id)
if uploaded:
output_directory = uploaded
uploaded = True
else:
uploaded = False
# Update the scan instance with the output path and upload status
Scan.objects.filter(id=scan_id).update(
output_path=output_directory, upload_to_s3=uploaded
)
return {
"output_path": output_directory,
"upload_to_s3": uploaded,
"scan_id": scan_id,
"provider_id": provider_id,
}