diff --git a/.env b/.env index 40eab3da90..39f0fb4c53 100644 --- a/.env +++ b/.env @@ -30,6 +30,23 @@ VALKEY_HOST=valkey VALKEY_PORT=6379 VALKEY_DB=0 +# API scan settings +# The AWS access key to be used when uploading scan artifacts to an S3 bucket +# If left empty, default AWS credentials resolution behavior will be used +ARTIFACTS_AWS_ACCESS_KEY_ID="" + +# The AWS secret key to be used when uploading scan artifacts to an S3 bucket +ARTIFACTS_AWS_SECRET_ACCESS_KEY="" + +# An optional AWS session token +ARTIFACTS_AWS_SESSION_TOKEN="" + +# The AWS region where your S3 bucket is located (e.g., "us-east-1") +ARTIFACTS_AWS_DEFAULT_REGION="" + +# The name of the S3 bucket where scan artifacts should be stored +ARTIFACTS_AWS_S3_OUTPUT_BUCKET="" + # Django settings DJANGO_ALLOWED_HOSTS=localhost,127.0.0.1,prowler-api DJANGO_BIND_ADDRESS=0.0.0.0 diff --git a/api/src/backend/api/specs/v1.yaml b/api/src/backend/api/specs/v1.yaml index e43b6ba516..fa41cd00b6 100644 --- a/api/src/backend/api/specs/v1.yaml +++ b/api/src/backend/api/specs/v1.yaml @@ -4235,6 +4235,38 @@ paths: schema: $ref: '#/components/schemas/ScanUpdateResponse' description: '' + /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[scans] + schema: + type: array + items: + type: string + enum: [] + 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: + - jwtAuth: [] + responses: + '200': + description: Report obtanined successfully + '404': + description: Report not found /api/v1/schedules/daily: post: operationId: schedules_daily_create diff --git a/api/src/backend/api/v1/serializers.py b/api/src/backend/api/v1/serializers.py index e73825daa1..dc76810e01 100644 --- a/api/src/backend/api/v1/serializers.py +++ b/api/src/backend/api/v1/serializers.py @@ -819,6 +819,14 @@ class ScanTaskSerializer(RLSSerializer): ] +class ScanReportSerializer(RLSSerializer): + class Meta: + model = Scan + fields = [ + "id", + ] + + class ResourceTagSerializer(RLSSerializer): """ Serializer for the ResourceTag model diff --git a/api/src/backend/api/v1/views.py b/api/src/backend/api/v1/views.py index c12da92026..6452cb4306 100644 --- a/api/src/backend/api/v1/views.py +++ b/api/src/backend/api/v1/views.py @@ -1,10 +1,17 @@ +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.db.models.functions import JSONObject +from django.http import HttpResponse from django.urls import reverse from django.utils.decorators import method_decorator from django.views.decorators.cache import cache_control @@ -115,6 +122,7 @@ from api.v1.serializers import ( RoleSerializer, RoleUpdateSerializer, ScanCreateSerializer, + ScanReportSerializer, ScanSerializer, ScanUpdateSerializer, ScheduleDailyCreateSerializer, @@ -127,6 +135,7 @@ from api.v1.serializers import ( UserSerializer, UserUpdateSerializer, ) +from prowler.config.config import tmp_output_directory CACHE_DECORATOR = cache_control( max_age=django_settings.CACHE_MAX_AGE, @@ -1074,6 +1083,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): @@ -1128,6 +1139,101 @@ class ScanViewSet(BaseRLSViewSet): }, ) + @extend_schema( + tags=["Scan"], + summary="Download ZIP report", + description="Returns a ZIP file containing the requested report", + request=ScanReportSerializer, + responses={ + 200: OpenApiResponse(description="Report obtanined successfully"), + 404: OpenApiResponse(description="Report not found"), + }, + ) + @action(detail=True, methods=["get"], url_name="report") + def report(self, request, pk=None): + 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("ARTIFACTS_AWS_ACCESS_KEY_ID"), + aws_secret_access_key=env.str("ARTIFACTS_AWS_SECRET_ACCESS_KEY"), + aws_session_token=env.str("ARTIFACTS_AWS_SESSION_TOKEN"), + region_name=env.str("ARTIFACTS_AWS_DEFAULT_REGION"), + ) + s3_client.list_buckets() + except (ClientError, NoCredentialsError, ParamValidationError) as e: + s3_client = None + + if s3_client: + bucket_name = env.str("ARTIFACTS_AWS_S3_OUTPUT_BUCKET") + s3_prefix = f"{request.tenant_id}/{pk}/" + + try: + response = s3_client.list_objects_v2( + Bucket=bucket_name, Prefix=s3_prefix + ) + if response["KeyCount"] == 0: + return Response( + {"detail": "No files found in S3 storage"}, + status=status.HTTP_404_NOT_FOUND, + ) + + zip_files = [ + obj["Key"] + for obj in response.get("Contents", []) + if obj["Key"].endswith(".zip") + ] + if not zip_files: + return Response( + {"detail": "No ZIP files found in S3 storage"}, + status=status.HTTP_404_NOT_FOUND, + ) + + s3_key = zip_files[0] + s3_object = s3_client.get_object(Bucket=bucket_name, Key=s3_key) + file_content = s3_object["Body"].read() + filename = os.path.basename(s3_key) + + except ClientError: + return Response( + {"detail": "Error accessing cloud storage"}, + status=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + + else: + local_path = os.path.join( + tmp_output_directory, + str(request.tenant_id), + str(pk), + "*.zip", + ) + zip_files = glob.glob(local_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 + @extend_schema_view( list=extend_schema( diff --git a/api/src/backend/tasks/jobs/scan.py b/api/src/backend/tasks/jobs/scan.py index 25993f5bbc..45153094c5 100644 --- a/api/src/backend/tasks/jobs/scan.py +++ b/api/src/backend/tasks/jobs/scan.py @@ -1,8 +1,13 @@ +import os import time +import zipfile from copy import deepcopy from datetime import datetime, timezone +import boto3 +from botocore.exceptions import ClientError from celery.utils.log import get_task_logger +from config.env import env from config.settings.celery import CELERY_DEADLOCK_ATTEMPTS from django.db import IntegrityError, OperationalError from django.db.models import Case, Count, IntegerField, Sum, When @@ -25,15 +30,117 @@ from api.models import ( from api.models import StatusChoices as FindingStatus from api.utils import initialize_prowler_provider from api.v1.serializers import ScanTaskSerializer +from prowler.config.config import ( + csv_file_suffix, + html_file_suffix, + json_asff_file_suffix, + json_ocsf_file_suffix, + output_file_timestamp, + tmp_output_directory, +) +from prowler.lib.outputs.asff.asff import ASFF +from prowler.lib.outputs.csv.csv import CSV from prowler.lib.outputs.finding import Finding as ProwlerFinding +from prowler.lib.outputs.html.html import HTML +from prowler.lib.outputs.ocsf.ocsf import OCSF from prowler.lib.scan.scan import Scan as ProwlerScan logger = get_task_logger(__name__) +# Predefined mapping for output formats and their configurations +OUTPUT_FORMATS_MAPPING = { + "csv": { + "class": CSV, + "suffix": csv_file_suffix, + "kwargs": {}, + }, + "json-asff": {"class": ASFF, "suffix": json_asff_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:///") 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("ARTIFACTS_AWS_S3_OUTPUT_BUCKET", ""): + return + + if env.str("ARTIFACTS_AWS_ACCESS_KEY_ID", ""): + s3 = boto3.client( + "s3", + aws_access_key_id=env.str("ARTIFACTS_AWS_ACCESS_KEY_ID"), + aws_secret_access_key=env.str("ARTIFACTS_AWS_SECRET_ACCESS_KEY"), + aws_session_token=env.str("ARTIFACTS_AWS_SESSION_TOKEN"), + region_name=env.str("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("ARTIFACTS_AWS_S3_OUTPUT_BUCKET"), + Key=s3_key, + ) + return f"s3://{env.str("ARTIFACTS_AWS_S3_OUTPUT_BUCKET")}/{s3_key}" + except ClientError as e: + logger.error(f"S3 upload failed: {str(e)}") + raise e + def _create_finding_delta( last_status: FindingStatus | None | str, new_status: FindingStatus | None -) -> Finding.DeltaChoices: +) -> Finding.DeltaChoices | None: """ Determine the delta status of a finding based on its previous and current status. @@ -53,7 +160,11 @@ def _create_finding_delta( def _store_resources( - finding: ProwlerFinding, tenant_id: str, provider_instance: Provider + finding: ProwlerFinding, + tenant_id: str, + provider_instance: Provider, + resource_cache: dict, + tag_cache: dict, ) -> tuple[Resource, tuple[str, str]]: """ Store resource information from a finding, including tags, in the database. @@ -65,40 +176,91 @@ def _store_resources( Returns: tuple: - - Resource: The resource instance created or retrieved from the database. + - Resource: The resource instance created or updated from the database. - tuple[str, str]: A tuple containing the resource UID and region. """ - with rls_transaction(tenant_id): - resource_instance, created = Resource.objects.get_or_create( - tenant_id=tenant_id, - provider=provider_instance, - uid=finding.resource_uid, - defaults={ - "region": finding.region, - "service": finding.service_name, - "type": finding.resource_type, - }, - ) + resource_uid = finding.resource_uid + + # Check cache or create/update resource + if resource_uid in resource_cache: + resource_instance = resource_cache[resource_uid] + update_fields = [] + for field, value in [ + ("region", finding.region), + ("service", finding.service_name), + ("type", finding.resource_type), + ("name", finding.resource_name), + ]: + if getattr(resource_instance, field) != value: + setattr(resource_instance, field, value) + update_fields.append(field) + if update_fields: + with rls_transaction(tenant_id): + resource_instance.save(update_fields=update_fields) + else: + with rls_transaction(tenant_id): + resource_instance, _ = Resource.objects.update_or_create( + tenant_id=tenant_id, + provider=provider_instance, + uid=resource_uid, + defaults={ + "region": finding.region, + "service": finding.service_name, + "type": finding.resource_type, + "name": finding.resource_name, + }, + ) + resource_cache[resource_uid] = resource_instance + + # Process tags with caching + tags = [] + for key, value in finding.resource_tags.items(): + tag_key = (key, value) + if tag_key not in tag_cache: + with rls_transaction(tenant_id): + tag_instance, _ = ResourceTag.objects.get_or_create( + tenant_id=tenant_id, key=key, value=value + ) + tag_cache[tag_key] = tag_instance + tags.append(tag_cache[tag_key]) - if not created: - resource_instance.region = finding.region - resource_instance.service = finding.service_name - resource_instance.type = finding.resource_type - resource_instance.save() with rls_transaction(tenant_id): - tags = [ - ResourceTag.objects.get_or_create( - tenant_id=tenant_id, key=key, value=value - )[0] - for key, value in finding.resource_tags.items() - ] resource_instance.upsert_or_delete_tags(tags=tags) + return resource_instance, (resource_instance.uid, resource_instance.region) +def _generate_output_directory( + prowler_provider: object, tenant_id: str, scan_id: str +) -> str: + """ + Generate a dynamic output directory path based on the given provider type. + + Args: + prowler_provider (object): An object that has a `type` attribute indicating + the provider type (e.g., "aws", "azure", etc.). + tenant_id (str): A unique identifier for the tenant. Used to build the output path. + scan_id (str): A unique identifier for the scan. Included in the output path. + + Returns: + str: The complete path to the output directory, including the tenant ID, scan ID, + provider identity, and a timestamp. + + """ + provider_type = prowler_provider.type + get_identity = PROVIDER_IDENTITY_MAP.get(provider_type, lambda _: "unknown") + return ( + f"{tmp_output_directory}/{tenant_id}/{scan_id}/prowler-output-" + f"{get_identity(prowler_provider)}-{output_file_timestamp}" + ) + + def perform_prowler_scan( - tenant_id: str, scan_id: str, provider_id: str, checks_to_execute: list[str] = None + tenant_id: str, + scan_id: str, + provider_id: str, + checks_to_execute: list[str] = None, ): """ Perform a scan using Prowler and store the findings and resources in the database. @@ -120,6 +282,9 @@ def perform_prowler_scan( exception = None unique_resources = set() start_time = time.time() + resource_cache = {} + tag_cache = {} + last_status_cache = {} with rls_transaction(tenant_id): provider_instance = Provider.objects.get(pk=provider_id) @@ -129,6 +294,7 @@ def perform_prowler_scan( scan_instance.save() try: + # Provider initialization with rls_transaction(tenant_id): try: prowler_provider = initialize_prowler_provider(provider_instance) @@ -144,105 +310,66 @@ def perform_prowler_scan( ) provider_instance.save() - prowler_scan = ProwlerScan(provider=prowler_provider, checks=checks_to_execute) + # Scan configuration + prowler_scan = ProwlerScan( + provider=prowler_provider, checks=checks_to_execute or [] + ) + output_directory = _generate_output_directory( + prowler_provider, tenant_id, scan_id + ) + os.makedirs("/".join(output_directory.split("/")[:-1]), exist_ok=True) - resource_cache = {} - tag_cache = {} - last_status_cache = {} + all_findings = [] - for progress, findings in prowler_scan.scan(): + # Main scan loop + for progress, findings, stats in prowler_scan.scan(): + # Process findings for finding in findings: + # Resource processing with retries for attempt in range(CELERY_DEADLOCK_ATTEMPTS): try: - with rls_transaction(tenant_id): - # Process resource - resource_uid = finding.resource_uid - if resource_uid not in resource_cache: - # Get or create the resource - resource_instance, _ = Resource.objects.get_or_create( - tenant_id=tenant_id, - provider=provider_instance, - uid=resource_uid, - defaults={ - "region": finding.region, - "service": finding.service_name, - "type": finding.resource_type, - "name": finding.resource_name, - }, - ) - resource_cache[resource_uid] = resource_instance - else: - resource_instance = resource_cache[resource_uid] - - # Update resource fields if necessary - updated_fields = [] - if resource_instance.region != finding.region: - resource_instance.region = finding.region - updated_fields.append("region") - if resource_instance.service != finding.service_name: - resource_instance.service = finding.service_name - updated_fields.append("service") - if resource_instance.type != finding.resource_type: - resource_instance.type = finding.resource_type - updated_fields.append("type") - if updated_fields: - with rls_transaction(tenant_id): - resource_instance.save(update_fields=updated_fields) + resource_instance, resource_uid_region = _store_resources( + finding, + tenant_id, + provider_instance, + resource_cache, + tag_cache, + ) + unique_resources.add(resource_uid_region) + break except (OperationalError, IntegrityError) as db_err: if attempt < CELERY_DEADLOCK_ATTEMPTS - 1: logger.warning( - f"{'Deadlock error' if isinstance(db_err, OperationalError) else 'Integrity error'} " - f"detected when processing resource {resource_uid} on scan {scan_id}. Retrying..." + f"Database error ({type(db_err).__name__}) " + f"processing resource {finding.resource_uid}, retrying..." ) time.sleep(0.1 * (2**attempt)) - continue else: raise db_err - # Update tags - tags = [] - with rls_transaction(tenant_id): - for key, value in finding.resource_tags.items(): - tag_key = (key, value) - if tag_key not in tag_cache: - tag_instance, _ = ResourceTag.objects.get_or_create( - tenant_id=tenant_id, key=key, value=value - ) - tag_cache[tag_key] = tag_instance - else: - tag_instance = tag_cache[tag_key] - tags.append(tag_instance) - resource_instance.upsert_or_delete_tags(tags=tags) - - unique_resources.add((resource_instance.uid, resource_instance.region)) - - # Process finding + # Finding processing with rls_transaction(tenant_id): finding_uid = finding.uid - last_first_seen_at = None if finding_uid not in last_status_cache: - most_recent_finding = ( + most_recent = ( Finding.objects.filter(uid=finding_uid) .order_by("-id") .values("status", "first_seen_at") .first() ) - last_status = None - if most_recent_finding: - last_status = most_recent_finding["status"] - last_first_seen_at = most_recent_finding["first_seen_at"] - last_status_cache[finding_uid] = last_status, last_first_seen_at + last_status, first_seen = ( + (most_recent["status"], most_recent["first_seen_at"]) + if most_recent + else (None, None) + ) + last_status_cache[finding_uid] = (last_status, first_seen) else: - last_status, last_first_seen_at = last_status_cache[finding_uid] + last_status, first_seen = last_status_cache[finding_uid] status = FindingStatus[finding.status] delta = _create_finding_delta(last_status, status) - # For the findings prior to the change, when a first finding is found with delta!="new" it will be assigned a current date as first_seen_at and the successive findings with the same UID will always get the date of the previous finding. - # For new findings, when a finding (delta="new") is found for the first time, the first_seen_at attribute will be assigned the current date, the following findings will get that date. - if not last_first_seen_at: - last_first_seen_at = datetime.now(tz=timezone.utc) + first_seen = first_seen or datetime.now(tz=timezone.utc) - # Create the finding finding_instance = Finding.objects.create( tenant_id=tenant_id, uid=finding_uid, @@ -255,92 +382,96 @@ def perform_prowler_scan( raw_result=finding.raw, check_id=finding.check_id, scan=scan_instance, - first_seen_at=last_first_seen_at, + first_seen_at=first_seen, ) finding_instance.add_resources([resource_instance]) - # Update compliance data if applicable - if finding.status.value == "MUTED": - continue + # Update compliance status + if finding.status.value != "MUTED": + region_data = check_status_by_region.setdefault(finding.region, {}) + if region_data.get(finding.check_id) != "FAIL": + region_data[finding.check_id] = finding.status.value - region_dict = check_status_by_region.setdefault(finding.region, {}) - current_status = region_dict.get(finding.check_id) - if current_status == "FAIL": - continue - region_dict[finding.check_id] = finding.status.value - - # Update scan progress + # Progress updates and output generation with rls_transaction(tenant_id): scan_instance.progress = progress scan_instance.save() + all_findings.extend(findings) + + # Generate output files + for mode, config in OUTPUT_FORMATS_MAPPING.items(): + kwargs = dict(config["kwargs"]) + if mode == "html": + kwargs["provider"] = prowler_provider + kwargs["stats"] = stats + config["class"]( + findings=all_findings, + create_file_descriptor=True, + file_path=output_directory, + file_extension=config["suffix"], + ).batch_write_data_to_file(**kwargs) + scan_instance.state = StateChoices.COMPLETED + # Compress output files + zip_path = _compress_output_files(output_directory) + + # Save to configured storage + _upload_to_s3(tenant_id, zip_path, scan_id) + except Exception as e: - logger.error(f"Error performing scan {scan_id}: {e}") + logger.error(f"Scan {scan_id} failed: {str(e)}") exception = e scan_instance.state = StateChoices.FAILED finally: + # Final scan updates with rls_transaction(tenant_id): scan_instance.duration = time.time() - start_time scan_instance.completed_at = datetime.now(tz=timezone.utc) scan_instance.unique_resource_count = len(unique_resources) scan_instance.save() - if exception is None: - try: - regions = prowler_provider.get_regions() - except AttributeError: - regions = set() - + # Compliance processing + if not exception: compliance_template = PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE[ provider_instance.provider ] - compliance_overview_by_region = { - region: deepcopy(compliance_template) for region in regions + compliance_overview = { + region: deepcopy(compliance_template) + for region in getattr(prowler_provider, "get_regions", lambda: set())() } - for region, check_status in check_status_by_region.items(): - compliance_data = compliance_overview_by_region.setdefault( - region, deepcopy(compliance_template) - ) - for check_name, status in check_status.items(): + for region, checks in check_status_by_region.items(): + for check_id, status in checks.items(): generate_scan_compliance( - compliance_data, + compliance_overview.setdefault( + region, deepcopy(compliance_template) + ), provider_instance.provider, - check_name, + check_id, status, ) - # Prepare compliance overview objects - compliance_overview_objects = [] - for region, compliance_data in compliance_overview_by_region.items(): - for compliance_id, compliance in compliance_data.items(): - compliance_overview_objects.append( - ComplianceOverview( - tenant_id=tenant_id, - scan=scan_instance, - region=region, - compliance_id=compliance_id, - framework=compliance["framework"], - version=compliance["version"], - description=compliance["description"], - requirements=compliance["requirements"], - requirements_passed=compliance["requirements_status"]["passed"], - requirements_failed=compliance["requirements_status"]["failed"], - requirements_manual=compliance["requirements_status"]["manual"], - total_requirements=compliance["total_requirements"], - ) + ComplianceOverview.objects.bulk_create( + [ + ComplianceOverview( + tenant_id=tenant_id, + scan=scan_instance, + region=region, + compliance_id=compliance_id, + **compliance_data, ) - with rls_transaction(tenant_id): - ComplianceOverview.objects.bulk_create(compliance_overview_objects) + for region, data in compliance_overview.items() + for compliance_id, compliance_data in data.items() + ] + ) - if exception is not None: + if exception: raise exception - serializer = ScanTaskSerializer(instance=scan_instance) - return serializer.data + return ScanTaskSerializer(instance=scan_instance).data def aggregate_findings(tenant_id: str, scan_id: str): @@ -372,124 +503,103 @@ def aggregate_findings(tenant_id: str, scan_id: str): - muted_changed: Muted findings with a delta of 'changed'. """ with rls_transaction(tenant_id): - findings = Finding.objects.filter(scan_id=scan_id) - - aggregation = findings.values( - "check_id", - "resources__service", - "severity", - "resources__region", - ).annotate( - fail=Sum( - Case( - When(status="FAIL", then=1), - default=0, - output_field=IntegerField(), - ) - ), - _pass=Sum( - Case( - When(status="PASS", then=1), - default=0, - output_field=IntegerField(), - ) - ), - muted=Sum( - Case( - When(status="MUTED", then=1), - default=0, - output_field=IntegerField(), - ) - ), - total=Count("id"), - new=Sum( - Case( - When(delta="new", then=1), - default=0, - output_field=IntegerField(), - ) - ), - changed=Sum( - Case( - When(delta="changed", then=1), - default=0, - output_field=IntegerField(), - ) - ), - unchanged=Sum( - Case( - When(delta__isnull=True, then=1), - default=0, - output_field=IntegerField(), - ) - ), - fail_new=Sum( - Case( - When(delta="new", status="FAIL", then=1), - default=0, - output_field=IntegerField(), - ) - ), - fail_changed=Sum( - Case( - When(delta="changed", status="FAIL", then=1), - default=0, - output_field=IntegerField(), - ) - ), - pass_new=Sum( - Case( - When(delta="new", status="PASS", then=1), - default=0, - output_field=IntegerField(), - ) - ), - pass_changed=Sum( - Case( - When(delta="changed", status="PASS", then=1), - default=0, - output_field=IntegerField(), - ) - ), - muted_new=Sum( - Case( - When(delta="new", status="MUTED", then=1), - default=0, - output_field=IntegerField(), - ) - ), - muted_changed=Sum( - Case( - When(delta="changed", status="MUTED", then=1), - default=0, - output_field=IntegerField(), - ) - ), + aggregation = ( + Finding.objects.filter(scan_id=scan_id) + .values( + "check_id", + "resources__service", + "severity", + "resources__region", + ) + .annotate( + fail=Sum( + Case( + When(status="FAIL", then=1), + default=0, + output_field=IntegerField(), + ) + ), + _pass=Sum( + Case( + When(status="PASS", then=1), + default=0, + output_field=IntegerField(), + ) + ), + muted=Sum( + Case( + When(status="MUTED", then=1), + default=0, + output_field=IntegerField(), + ) + ), + total=Count("id"), + new=Sum( + Case( + When(delta="new", then=1), + default=0, + output_field=IntegerField(), + ) + ), + changed=Sum( + Case( + When(delta="changed", then=1), + default=0, + output_field=IntegerField(), + ) + ), + unchanged=Sum( + Case( + When(delta__isnull=True, then=1), + default=0, + output_field=IntegerField(), + ) + ), + **{ + f"{status}_new": Sum( + Case( + When(delta="new", status=status.upper(), then=1), + default=0, + output_field=IntegerField(), + ) + ) + for status in ["fail", "pass", "muted"] + }, + **{ + f"{status}_changed": Sum( + Case( + When(delta="changed", status=status.upper(), then=1), + default=0, + output_field=IntegerField(), + ) + ) + for status in ["fail", "pass", "muted"] + }, + ) ) - with rls_transaction(tenant_id): - scan_aggregations = { - ScanSummary( - tenant_id=tenant_id, - scan_id=scan_id, - check_id=agg["check_id"], - service=agg["resources__service"], - severity=agg["severity"], - region=agg["resources__region"], - fail=agg["fail"], - _pass=agg["_pass"], - muted=agg["muted"], - total=agg["total"], - new=agg["new"], - changed=agg["changed"], - unchanged=agg["unchanged"], - fail_new=agg["fail_new"], - fail_changed=agg["fail_changed"], - pass_new=agg["pass_new"], - pass_changed=agg["pass_changed"], - muted_new=agg["muted_new"], - muted_changed=agg["muted_changed"], - ) - for agg in aggregation - } - ScanSummary.objects.bulk_create(scan_aggregations, batch_size=3000) + ScanSummary.objects.bulk_create( + [ + ScanSummary( + tenant_id=tenant_id, + scan_id=scan_id, + check_id=agg["check_id"], + service=agg["resources__service"], + severity=agg["severity"], + region=agg["resources__region"], + **{ + k: v or 0 + for k, v in agg.items() + if k + not in { + "check_id", + "resources__service", + "severity", + "resources__region", + } + }, + ) + for agg in aggregation + ], + batch_size=3000, + ) diff --git a/docker-compose-dev.yml b/docker-compose-dev.yml index d49af9dfbb..dfbea102b5 100644 --- a/docker-compose-dev.yml +++ b/docker-compose-dev.yml @@ -16,6 +16,7 @@ services: volumes: - "./api/src/backend:/home/prowler/backend" - "./api/pyproject.toml:/home/prowler/pyproject.toml" + - "/tmp/prowler_api_output:/tmp/prowler_api_output" depends_on: postgres: condition: service_healthy @@ -82,6 +83,8 @@ services: env_file: - path: .env required: false + volumes: + - "/tmp/prowler_api_output:/tmp/prowler_api_output" depends_on: valkey: condition: service_healthy diff --git a/prowler/config/config.py b/prowler/config/config.py index 6633083a8f..0796a2ee94 100644 --- a/prowler/config/config.py +++ b/prowler/config/config.py @@ -58,6 +58,7 @@ aws_services_json_file = "aws_regions_by_service.json" # gcp_zones_json_file = "gcp_zones.json" default_output_directory = getcwd() + "/output" +tmp_output_directory = "/tmp/prowler_api_output" output_file_timestamp = timestamp.strftime("%Y%m%d%H%M%S") timestamp_iso = timestamp.isoformat(sep=" ", timespec="seconds") csv_file_suffix = ".csv" diff --git a/prowler/lib/outputs/compliance/ens/ens.py b/prowler/lib/outputs/compliance/ens/ens.py index a414a0206d..cd3aa7054b 100644 --- a/prowler/lib/outputs/compliance/ens/ens.py +++ b/prowler/lib/outputs/compliance/ens/ens.py @@ -55,9 +55,9 @@ def get_ens_table( and index not in fail_count ): fail_count.append(index) - marcos[marco_categoria][ - "Estado" - ] = f"{Fore.RED}NO CUMPLE{Style.RESET_ALL}" + marcos[marco_categoria]["Estado"] = ( + f"{Fore.RED}NO CUMPLE{Style.RESET_ALL}" + ) elif finding.status == "PASS" and index not in pass_count: pass_count.append(index) if attribute.Nivel == "opcional": diff --git a/prowler/lib/outputs/output.py b/prowler/lib/outputs/output.py index 290eeabfc9..011c88761e 100644 --- a/prowler/lib/outputs/output.py +++ b/prowler/lib/outputs/output.py @@ -38,16 +38,18 @@ class Output(ABC): file_extension: str = "", ) -> None: self._data = [] + self.file_path = file_path if not file_extension and file_path: self._file_extension = "".join(Path(file_path).suffixes) if file_extension: self._file_extension = file_extension + self.file_path = f"{file_path}{self.file_extension}" if findings: self.transform(findings) if create_file_descriptor and file_path: - self.create_file_descriptor(file_path) + self.create_file_descriptor(self.file_path) @property def data(self): diff --git a/prowler/lib/scan/scan.py b/prowler/lib/scan/scan.py index 213161b64b..ac618292b8 100644 --- a/prowler/lib/scan/scan.py +++ b/prowler/lib/scan/scan.py @@ -1,5 +1,5 @@ import datetime -from typing import Generator +from typing import Dict, Generator, List, Optional, Set, Tuple from prowler.lib.check.check import ( execute, @@ -14,6 +14,7 @@ from prowler.lib.check.models import CheckMetadata, Severity from prowler.lib.logger import logger from prowler.lib.outputs.common import Status from prowler.lib.outputs.finding import Finding +from prowler.lib.outputs.outputs import extract_findings_statistics from prowler.lib.scan.exceptions.exceptions import ( ScanInvalidCategoryError, ScanInvalidCheckError, @@ -28,29 +29,25 @@ from prowler.providers.common.provider import Provider class Scan: _provider: Provider - # Refactor(Core): This should replace the Audit_Metadata - _number_of_checks_to_execute: int = 0 - _number_of_checks_completed: int = 0 - # TODO the str should be a set of Check objects - _checks_to_execute: list[str] - _service_checks_to_execute: dict[str, set[str]] - _service_checks_completed: dict[str, set[str]] + _checks_to_execute: List[str] + _service_checks_map: Dict[str, Set[str]] + _completed_checks: Set[str] _progress: float = 0.0 - _findings: list = [] + _findings: List[Finding] = [] _duration: int = 0 - _status: list[str] = None + _statuses: Optional[List[Status]] = None def __init__( self, provider: Provider, - checks: list[str] = None, - services: list[str] = None, - compliances: list[str] = None, - categories: list[str] = None, - severities: list[str] = None, - excluded_checks: list[str] = None, - excluded_services: list[str] = None, - status: list[str] = None, + checks: Optional[List[str]] = None, + services: Optional[List[str]] = None, + compliances: Optional[List[str]] = None, + categories: Optional[List[str]] = None, + severities: Optional[List[str]] = None, + excluded_checks: Optional[List[str]] = None, + excluded_services: Optional[List[str]] = None, + status: Optional[List[str]] = None, ): """ Scan is the class that executes the checks and yields the progress and the findings. @@ -75,155 +72,190 @@ class Scan: ScanInvalidStatusError: If the status does not exist in the provider. """ self._provider = provider + self._statuses = self._validate_statuses(status) - # Validate the status - if status: - try: - for s in status: - Status(s) - if not self._status: - self._status = [] - if s not in self._status: - self._status.append(s) - except ValueError: - raise ScanInvalidStatusError(f"Invalid status provided: {s}.") - - # Load bulk compliance frameworks bulk_compliance_frameworks = Compliance.get_bulk(provider.type) + bulk_checks_metadata = self._load_checks_metadata(bulk_compliance_frameworks) - # Get bulk checks metadata for the provider - bulk_checks_metadata = CheckMetadata.get_bulk(provider.type) - # Complete checks metadata with the compliance framework specification - bulk_checks_metadata = update_checks_metadata_with_compliance( + self._validate_inputs( + checks, + services, + compliances, + categories, + severities, + bulk_checks_metadata, + bulk_compliance_frameworks, + ) + + self._checks_to_execute = self._init_checks_to_execute( + bulk_checks_metadata, + bulk_compliance_frameworks, + checks, + services, + compliances, + categories, + severities, + excluded_checks, + excluded_services, + ) + + self._service_checks_map = get_service_checks_mapping(self._checks_to_execute) + self._completed_checks = set() + + def _validate_statuses( + self, statuses: Optional[List[str]] + ) -> Optional[List[Status]]: + """Validate and convert status strings to Status enums.""" + if not statuses: + return None + + validated = [] + for status in statuses: + try: + validated.append(Status(status)) + except ValueError: + raise ScanInvalidStatusError(f"Invalid status: {status}") + return validated + + def _load_checks_metadata(self, bulk_compliance_frameworks: dict) -> dict: + """Load and enhance checks metadata with compliance information.""" + bulk_checks_metadata = CheckMetadata.get_bulk(self._provider.type) + return update_checks_metadata_with_compliance( bulk_compliance_frameworks, bulk_checks_metadata ) - # Create a list of valid categories - valid_categories = set() - for check, metadata in bulk_checks_metadata.items(): - for category in metadata.Categories: - if category not in valid_categories: - valid_categories.add(category) + def _validate_inputs( + self, + checks: Optional[List[str]], + services: Optional[List[str]], + compliances: Optional[List[str]], + categories: Optional[List[str]], + severities: Optional[List[str]], + bulk_checks_metadata: dict, + bulk_compliance_frameworks: dict, + ): + """Validate all input parameters against provider capabilities.""" + valid_services = list_services(self._provider.type) + valid_categories = self._extract_valid_categories(bulk_checks_metadata) - # Validate checks + self._validate_checks(checks, bulk_checks_metadata) + self._validate_services(services, valid_services) + self._validate_compliances(compliances, bulk_compliance_frameworks) + self._validate_categories(categories, valid_categories) + self._validate_severities(severities) + + def _extract_valid_categories(self, checks_metadata: dict) -> Set[str]: + """Extract unique categories from checks metadata.""" + return { + category + for metadata in checks_metadata.values() + for category in metadata.Categories + } + + def _validate_checks(self, checks: Optional[List[str]], metadata: dict): + """Validate requested checks exist in provider.""" if checks: - for check in checks: - if check not in bulk_checks_metadata.keys(): - raise ScanInvalidCheckError(f"Invalid check provided: {check}.") + invalid = [check for check in checks if check not in metadata] + if invalid: + raise ScanInvalidCheckError(f"Invalid checks: {', '.join(invalid)}") - # Validate services + def _validate_services(self, services: Optional[List[str]], valid_services: list): + """Validate requested services exist in provider.""" if services: - for service in services: - if service not in list_services(provider.type): - raise ScanInvalidServiceError( - f"Invalid service provided: {service}." - ) + invalid = [srv for srv in services if srv not in valid_services] + if invalid: + raise ScanInvalidServiceError(f"Invalid services: {', '.join(invalid)}") - # Validate compliances + def _validate_compliances(self, compliances: Optional[List[str]], frameworks: dict): + """Validate compliance frameworks exist.""" if compliances: - for compliance in compliances: - if compliance not in bulk_compliance_frameworks.keys(): - raise ScanInvalidComplianceFrameworkError( - f"Invalid compliance provided: {compliance}." - ) + invalid = [comp for comp in compliances if comp not in frameworks] + if invalid: + raise ScanInvalidComplianceFrameworkError( + f"Invalid compliances: {', '.join(invalid)}" + ) - # Validate categories + def _validate_categories( + self, categories: Optional[List[str]], valid_categories: Set[str] + ): + """Validate categories exist in provider checks.""" if categories: - for category in categories: - if category not in valid_categories: - raise ScanInvalidCategoryError( - f"Invalid category provided: {category}." - ) + invalid = [cat for cat in categories if cat not in valid_categories] + if invalid: + raise ScanInvalidCategoryError( + f"Invalid categories: {', '.join(invalid)}" + ) - # Validate severity + def _validate_severities(self, severities: Optional[List[str]]): + """Validate severity values are valid.""" if severities: - for severity in severities: - try: - Severity(severity) - except ValueError: - raise ScanInvalidSeverityError( - f"Invalid severity provided: {severity}." - ) + try: + [Severity(sev) for sev in severities] + except ValueError as e: + raise ScanInvalidSeverityError(f"Invalid severity: {e}") - # Load checks to execute - self._checks_to_execute = sorted( - load_checks_to_execute( - bulk_checks_metadata=bulk_checks_metadata, - bulk_compliance_frameworks=bulk_compliance_frameworks, - check_list=checks, - service_list=services, - compliance_frameworks=compliances, - categories=categories, - severities=severities, - provider=provider.type, - checks_file=None, - ) + def _init_checks_to_execute( + self, + bulk_checks_metadata: dict, + bulk_compliance_frameworks: dict, + checks: Optional[List[str]], + services: Optional[List[str]], + compliances: Optional[List[str]], + categories: Optional[List[str]], + severities: Optional[List[str]], + excluded_checks: Optional[List[str]], + excluded_services: Optional[List[str]], + ) -> List[str]: + """Load and filter checks based on configuration.""" + checks = load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metadata, + bulk_compliance_frameworks=bulk_compliance_frameworks, + check_list=checks, + service_list=services, + compliance_frameworks=compliances, + categories=categories, + severities=severities, + provider=self._provider.type, ) - # Exclude checks if excluded_checks: - for check in excluded_checks: - if check in self._checks_to_execute: - self._checks_to_execute.remove(check) - else: - raise ScanInvalidCheckError( - f"Invalid check provided: {check}. Check does not exist in the provider." - ) + checks = [c for c in checks if c not in excluded_checks] - # Exclude services if excluded_services: - for check in self._checks_to_execute: - if get_service_name_from_check_name(check) in excluded_services: - self._checks_to_execute.remove(check) - else: - raise ScanInvalidServiceError( - f"Invalid service provided: {check}. Service does not exist in the provider." - ) + excluded_services_set = set(excluded_services) + checks = [ + c + for c in checks + if get_service_from_check(c) not in excluded_services_set + ] - self._number_of_checks_to_execute = len(self._checks_to_execute) - - service_checks_to_execute = get_service_checks_to_execute( - self._checks_to_execute - ) - service_checks_completed = dict() - - self._service_checks_to_execute = service_checks_to_execute - self._service_checks_completed = service_checks_completed + return sorted(checks) @property - def checks_to_execute(self) -> list[str]: - return self._checks_to_execute + def total_checks(self) -> int: + return len(self._checks_to_execute) @property - def service_checks_to_execute(self) -> dict[str, set[str]]: - return self._service_checks_to_execute - - @property - def service_checks_completed(self) -> dict[str, set[str]]: - return self._service_checks_completed - - @property - def provider(self) -> Provider: - return self._provider + def completed_checks(self) -> int: + return len(self._completed_checks) @property def progress(self) -> float: return ( - self._number_of_checks_completed / self._number_of_checks_to_execute * 100 + (self.completed_checks / self.total_checks * 100) + if self.total_checks + else 0 ) @property - def duration(self) -> int: - return self._duration + def remaining_services(self) -> Dict[str, Set[str]]: + return { + service: checks + for service, checks in self._service_checks_map.items() + if checks + } - @property - def findings(self) -> list: - return self._findings - - def scan( - self, - custom_checks_metadata: dict = {}, - ) -> Generator[tuple[float, list[Finding]], None, None]: + def scan(self) -> Generator[Tuple[float, List[Finding], dict], None, None]: """ Executes the scan by iterating over the checks to execute and executing each check. Yields the progress and findings for each check. @@ -238,142 +270,97 @@ class Scan: ModuleNotFoundError: If the check does not exist in the provider or is from another provider. Exception: If any other error occurs during the execution of a check. """ + self._provider.audit_metadata = Audit_Metadata( + services_scanned=0, + expected_checks=self._checks_to_execute, + completed_checks=0, + audit_progress=0, + ) + + start_time = datetime.datetime.now() + + for check_name in self._checks_to_execute: + service = get_service_from_check(check_name) + try: + check_module = self._import_check_module(check_name, service) + findings = self._execute_check(check_module, check_name) + filtered_findings = self._filter_findings_by_status(findings) + except Exception as error: + logger.error(f"{check_name} failed: {error}") + continue + + self._update_scan_state(check_name, service, filtered_findings) + stats = extract_findings_statistics(filtered_findings) + + findings = [ + Finding.generate_output(self._provider, finding, output_options=None) + for finding in filtered_findings + ] + + yield self.progress, findings, stats + + self._duration = int((datetime.datetime.now() - start_time).total_seconds()) + + def _import_check_module(self, check_name: str, service: str): + """Dynamically import check module.""" + module_path = ( + f"prowler.providers.{self._provider.type}.services." + f"{service}.{check_name}.{check_name}" + ) try: - checks_to_execute = self.checks_to_execute - # Initialize the Audit Metadata - # TODO: this should be done in the provider class - # Refactor(Core): Audit manager? - self._provider.audit_metadata = Audit_Metadata( - services_scanned=0, - expected_checks=checks_to_execute, - completed_checks=0, - audit_progress=0, - ) - - start_time = datetime.datetime.now() - - for check_name in checks_to_execute: - try: - # Recover service from check name - service = get_service_name_from_check_name(check_name) - try: - # Import check module - check_module_path = f"prowler.providers.{self._provider.type}.services.{service}.{check_name}.{check_name}" - lib = import_check(check_module_path) - # Recover functions from check - check_to_execute = getattr(lib, check_name) - check = check_to_execute() - except ModuleNotFoundError: - logger.error( - f"Check '{check_name}' was not found for the {self._provider.type.upper()} provider" - ) - continue - # Execute the check - check_findings = execute( - check, - self._provider, - custom_checks_metadata, - output_options=None, - ) - - # Filter the findings by the status - if self._status: - for finding in check_findings: - if finding.status not in self._status: - check_findings.remove(finding) - - # Store findings - self._findings.extend(check_findings) - - # Remove the executed check - self._service_checks_to_execute[service].remove(check_name) - if len(self._service_checks_to_execute[service]) == 0: - self._service_checks_to_execute.pop(service, None) - # Add the completed check - if service not in self._service_checks_completed: - self._service_checks_completed[service] = set() - self._service_checks_completed[service].add(check_name) - self._number_of_checks_completed += 1 - - # This should be done just once all the service's checks are completed - # This metadata needs to get to the services not within the provider - # since it is present in the Scan class - self._provider.audit_metadata = update_audit_metadata( - self._provider.audit_metadata, - self.get_completed_services(), - self.get_completed_checks(), - ) - - findings = [ - Finding.generate_output( - self._provider, finding, output_options=None - ) - for finding in check_findings - ] - - yield self.progress, findings - # If check does not exists in the provider or is from another provider - except ModuleNotFoundError: - logger.error( - f"Check '{check_name}' was not found for the {self._provider.type.upper()} provider" - ) - except Exception as error: - logger.error( - f"{check_name} - {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" - ) - # Update the scan duration when all checks are completed - self._duration = int((datetime.datetime.now() - start_time).total_seconds()) - except Exception as error: + return import_check(module_path) + except ModuleNotFoundError: logger.error( - f"{check_name} - {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + f"Check '{check_name}' not found for {self._provider.type.upper()}" ) + raise - def get_completed_services(self) -> set[str]: - """ - get_completed_services returns the services that have been completed. + def _execute_check(self, check_module, check_name: str) -> List[Finding]: + """Execute a single check and return its findings.""" + check_func = getattr(check_module, check_name) + return execute(check_func(), self._provider, {}, None) - Example: - get_completed_services() -> {"ec2", "s3"} - """ - return self._service_checks_completed.keys() + def _filter_findings_by_status(self, findings: List[Finding]) -> List[Finding]: + """Filter findings based on configured status filters.""" + if not self._statuses: + return findings + return [f for f in findings if f.status in self._statuses] - def get_completed_checks(self) -> set[str]: - """ - get_completed_checks returns the checks that have been completed. + def _update_scan_state( + self, check_name: str, service: str, findings: List[Finding] + ): + """Update scan state after check completion.""" + self._service_checks_map[service].discard(check_name) + self._completed_checks.add(check_name) + self._findings.extend(findings) - Example: - get_completed_checks() -> {"ec2_instance_public_ip", "s3_bucket_public"} - """ - completed_checks = set() - for checks in self._service_checks_completed.values(): - completed_checks.update(checks) - return completed_checks + self._provider.audit_metadata = update_audit_metadata( + self._provider.audit_metadata, + self.remaining_services.keys(), + self._completed_checks, + ) -def get_service_name_from_check_name(check_name: str) -> str: +def get_service_from_check(check_name: str) -> str: """ - get_service_name_from_check_name returns the service name for a given check name. + Return the service name for a given check name. Example: - get_service_name_from_check_name("ec2_instance_public") -> "ec2" + get_service_from_check("ec2_instance_public") -> "ec2" """ return check_name.split("_")[0] -def get_service_checks_to_execute(checks_to_execute: set[str]) -> dict[str, set[str]]: +def get_service_checks_mapping(checks: List[str]) -> Dict[str, Set[str]]: """ - get_service_checks_to_execute returns a dictionary with the services and the checks to execute. + Return a dictionary with the services and the checks to execute. Example: - get_service_checks_to_execute({"accessanalyzer_enabled", "ec2_instance_public_ip"}) + get_service_checks_mapping({"accessanalyzer_enabled", "ec2_instance_public_ip"}) -> {"accessanalyzer": {"accessanalyzer_enabled"}, "ec2": {"ec2_instance_public_ip"}} """ - service_checks_to_execute = dict() - for check in checks_to_execute: - # check -> accessanalyzer_enabled - # service -> accessanalyzer - service = get_service_name_from_check_name(check) - if service not in service_checks_to_execute: - service_checks_to_execute[service] = set() - service_checks_to_execute[service].add(check) - return service_checks_to_execute + service_map = {} + for check in checks: + service = get_service_from_check(check) + service_map.setdefault(service, set()).add(check) + return service_map