ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 04:51:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(csv): Common output for all the providers (#3513)

Browse Source 
Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
Co-authored-by: Sergio Garcia <sergargar1@gmail.com>
This commit is contained in:
Pepe Fagoaga
2024-03-13 17:31:35 +01:00
committed by GitHub
parent b539a888b1
commit 87893bd54b
28 changed files with 516 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/lib/outputs/compliance/aws_well_architected_framework.py
+2 -4
View File
@@ -2,10 +2,8 @@ from csv import DictWriter
from prowler.config.config import timestamp
from prowler.lib.logger import logger
from prowler.lib.outputs.models import (
Check_Output_CSV_AWS_Well_Architected,
generate_csv_fields,
)
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.outputs.models import Check_Output_CSV_AWS_Well_Architected
from prowler.lib.utils.utils import outputs_unix_timestamp
prowler/lib/outputs/compliance/cis.py
+1 -1
View File
@@ -1,6 +1,6 @@
from prowler.lib.outputs.compliance.cis_aws import generate_compliance_row_cis_aws
from prowler.lib.outputs.compliance.cis_gcp import generate_compliance_row_cis_gcp
from prowler.lib.outputs.csv import write_csv
from prowler.lib.outputs.csv.csv import write_csv
def write_compliance_row_cis(
prowler/lib/outputs/compliance/cis_aws.py
+2 -1
View File
@@ -1,5 +1,6 @@
from prowler.config.config import timestamp
from prowler.lib.outputs.models import Check_Output_CSV_AWS_CIS, generate_csv_fields
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.outputs.models import Check_Output_CSV_AWS_CIS
from prowler.lib.utils.utils import outputs_unix_timestamp
prowler/lib/outputs/compliance/cis_gcp.py
+2 -1
View File
@@ -1,5 +1,6 @@
from prowler.config.config import timestamp
from prowler.lib.outputs.models import Check_Output_CSV_GCP_CIS, generate_csv_fields
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.outputs.models import Check_Output_CSV_GCP_CIS
from prowler.lib.utils.utils import outputs_unix_timestamp
prowler/lib/outputs/compliance/ens_rd2022_aws.py
+2 -1
View File
@@ -1,7 +1,8 @@
from csv import DictWriter
from prowler.config.config import timestamp
from prowler.lib.outputs.models import Check_Output_CSV_ENS_RD2022, generate_csv_fields
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.outputs.models import Check_Output_CSV_ENS_RD2022
from prowler.lib.utils.utils import outputs_unix_timestamp
prowler/lib/outputs/compliance/generic.py
+2 -4
View File
@@ -1,10 +1,8 @@
from csv import DictWriter
from prowler.config.config import timestamp
from prowler.lib.outputs.models import (
Check_Output_CSV_Generic_Compliance,
generate_csv_fields,
)
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.outputs.models import Check_Output_CSV_Generic_Compliance
from prowler.lib.utils.utils import outputs_unix_timestamp
prowler/lib/outputs/compliance/iso27001_2013_aws.py
+2 -4
View File
@@ -1,10 +1,8 @@
from csv import DictWriter
from prowler.config.config import timestamp
from prowler.lib.outputs.models import (
Check_Output_CSV_AWS_ISO27001_2013,
generate_csv_fields,
)
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.outputs.models import Check_Output_CSV_AWS_ISO27001_2013
from prowler.lib.utils.utils import outputs_unix_timestamp
prowler/lib/outputs/compliance/mitre_attack_aws.py
+2 -5
View File
@@ -1,11 +1,8 @@
from csv import DictWriter
from prowler.config.config import timestamp
from prowler.lib.outputs.models import (
Check_Output_MITRE_ATTACK,
generate_csv_fields,
unroll_list,
)
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.outputs.models import Check_Output_MITRE_ATTACK, unroll_list
from prowler.lib.utils.utils import outputs_unix_timestamp
prowler/lib/outputs/csv.py
-10
View File
@@ -1,10 +0,0 @@
from csv import DictWriter
def write_csv(file_descriptor, headers, row):
csv_writer = DictWriter(
file_descriptor,
fieldnames=headers,
delimiter=";",
)
csv_writer.writerow(row.__dict__)
prowler/lib/outputs/csv/__init__.py
View File
prowler/lib/outputs/csv/csv.py
+161
View File
@@ -0,0 +1,161 @@
from csv import DictWriter
from operator import attrgetter
from typing import Any
from prowler.config.config import timestamp
from prowler.lib.logger import logger
from prowler.lib.outputs.csv.models import CSVRow
from prowler.lib.outputs.models import unroll_list, unroll_tags
from prowler.lib.utils.utils import outputs_unix_timestamp
def get_provider_data_mapping(provider) -> dict:
data = {}
for generic_field, provider_field in provider.get_output_mapping.items():
try:
provider_value = attrgetter(provider_field)(provider)
data[generic_field] = provider_value
except AttributeError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
data[generic_field] = None
return data
def fill_common_data_csv(finding: dict, unix_timestamp: bool) -> dict:
data = {
"timestamp": outputs_unix_timestamp(unix_timestamp, timestamp),
"check_id": finding.check_metadata.CheckID,
"check_title": finding.check_metadata.CheckTitle,
"check_type": ",".join(finding.check_metadata.CheckType),
"status": finding.status,
"status_extended": finding.status_extended,
"service_name": finding.check_metadata.ServiceName,
"subservice_name": finding.check_metadata.SubServiceName,
"severity": finding.check_metadata.Severity,
"resource_type": finding.check_metadata.ResourceType,
"resource_details": finding.resource_details,
"resource_tags": unroll_tags(finding.resource_tags),
"description": finding.check_metadata.Description,
"risk": finding.check_metadata.Risk,
"related_url": finding.check_metadata.RelatedUrl,
"remediation_recommendation_text": (
finding.check_metadata.Remediation.Recommendation.Text
),
"remediation_recommendation_url": (
finding.check_metadata.Remediation.Recommendation.Url
),
"remediation_code_nativeiac": (
finding.check_metadata.Remediation.Code.NativeIaC
),
"remediation_code_terraform": (
finding.check_metadata.Remediation.Code.Terraform
),
"remediation_code_cli": (finding.check_metadata.Remediation.Code.CLI),
"remediation_code_other": (finding.check_metadata.Remediation.Code.Other),
"categories": unroll_list(finding.check_metadata.Categories),
"depends_on": unroll_list(finding.check_metadata.DependsOn),
"related_to": unroll_list(finding.check_metadata.RelatedTo),
"notes": finding.check_metadata.Notes,
}
return data
def generate_provider_output_csv(provider, finding, csv_data):
"""
generate_provider_output_csv creates the provider's CSV output
"""
# TODO: we have to standardize this between the above mapping and the provider.get_output_mapping()
try:
if provider.type == "aws":
csv_data["auth_method"] = f"profile: {csv_data['auth_method']}"
csv_data["resource_name"] = finding.resource_id
csv_data["resource_uid"] = finding.resource_arn
csv_data["region"] = finding.region
elif provider.type == "azure":
# TODO: we should show the authentication method used I think
csv_data["auth_method"] = (
f"{provider.identity.identity_type}: {provider.identity.identity_id}"
)
csv_data["account_uid"] = provider.identity.subscriptions[
finding.subscription
]
csv_data["account_name"] = finding.subscription
# Get the first tenant domain ID, just in case
csv_data["account_organization_uid"] = csv_data["account_organization_uid"][
0
]
csv_data["resource_name"] = finding.resource_name
csv_data["resource_uid"] = finding.resource_id
# TODO: pending to get location from Azure resources (finding.location)
csv_data["region"] = ""
elif provider.type == "gcp":
csv_data["auth_method"] = f"Account: {csv_data['auth_method']}"
csv_data["account_uid"] = provider.projects[finding.project_id].number
csv_data["account_name"] = provider.projects[finding.project_id].name
csv_data["account_tags"] = provider.projects[finding.project_id].labels
csv_data["resource_name"] = finding.resource_name
csv_data["resource_uid"] = finding.resource_id
csv_data["region"] = finding.location
if (
provider.projects
and finding.project_id in provider.projects
and getattr(provider.projects[finding.project_id], "organization")
):
csv_data["account_organization_uid"] = provider.projects[
finding.project_id
].organization.id
# TODO: for now is None since we don't retrieve that data
csv_data["account_organization"] = provider.projects[
finding.project_id
].organization.display_name
elif provider.type == "kubernetes":
if provider.identity.context == "In-Cluster":
csv_data["auth_method"] = "in-cluster"
else:
csv_data["auth_method"] = "kubeconfig"
csv_data["resource_name"] = finding.resource_name
csv_data["resource_uid"] = finding.resource_id
csv_data["account_name"] = f"context: {provider.identity.context}"
csv_data["region"] = f"namespace: {finding.namespace}"
# Finding Unique ID
# TODO: move this to a function
# TODO: in Azure, GCP and K8s there are fidings without resource_name
csv_data["finding_uid"] = (
f"prowler-{provider.type}-{finding.check_metadata.CheckID}-{csv_data['account_uid']}-{csv_data['region']}-{csv_data['resource_name']}"
)
finding_output = CSVRow(**csv_data)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
return finding_output
def write_csv(file_descriptor, headers, row):
csv_writer = DictWriter(
file_descriptor,
fieldnames=headers,
delimiter=";",
)
csv_writer.writerow(row.__dict__)
def generate_csv_fields(format: Any) -> list[str]:
"""Generates the CSV headers for the given class"""
csv_fields = []
# __fields__ is always available in the Pydantic's BaseModel class
for field in format.__dict__.get("__fields__").keys():
csv_fields.append(field)
return csv_fields
prowler/lib/outputs/csv/models.py
+74
View File
@@ -0,0 +1,74 @@
from datetime import datetime
from enum import Enum
from typing import Optional
from pydantic import BaseModel
class Status(str, Enum):
PASS = "PASS"
FAIL = "FAIL"
MANUAL = "MANUAL"
class Severity(str, Enum):
critical = "critical"
high = "high"
medium = "medium"
low = "low"
informational = "informational"
class CSVRow(BaseModel):
"""
CSVRow generates a finding's output in CSV format.
This is the base CSV output model for every provider.
"""
auth_method: str
timestamp: datetime
account_uid: str
# Optional since depends on permissions
account_name: Optional[str]
# Optional since depends on permissions
account_email: Optional[str]
# Optional since depends on permissions
account_organization_uid: Optional[str]
# Optional since depends on permissions
account_organization: Optional[str]
# Optional since depends on permissions
account_tags: Optional[str]
finding_uid: str
provider: str
check_id: str
check_title: str
check_type: str
status: Status
status_extended: str
muted: bool = False
service_name: str
subservice_name: str
severity: Severity
resource_type: str
resource_uid: str
resource_name: str
resource_details: str
resource_tags: str
# Only present for AWS and Azure
partition: Optional[str]
region: str
description: str
risk: str
related_url: str
remediation_recommendation_text: str
remediation_recommendation_url: str
remediation_code_nativeiac: str
remediation_code_terraform: str
remediation_code_cli: str
remediation_code_other: str
compliance: str
categories: str
depends_on: str
related_to: str
notes: str
prowler/lib/outputs/file_descriptors.py
+11 -26
View File
@@ -9,6 +9,8 @@ from prowler.config.config import (
json_ocsf_file_suffix,
)
from prowler.lib.logger import logger
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.outputs.csv.models import CSVRow
from prowler.lib.outputs.models import (
Check_Output_CSV_AWS_CIS,
Check_Output_CSV_AWS_ISO27001_2013,
@@ -17,20 +19,14 @@ from prowler.lib.outputs.models import (
Check_Output_CSV_GCP_CIS,
Check_Output_CSV_Generic_Compliance,
Check_Output_MITRE_ATTACK,
generate_csv_fields,
)
from prowler.lib.utils.utils import file_exists, open_file
from prowler.providers.common.outputs import get_provider_output_model
def initialize_file_descriptor(
filename: str,
output_mode: str,
# TODO: review this provider, maybe it's not needed
provider: Any,
format: Any = None,
filename: str, output_mode: str, format: Any = CSVRow
) -> TextIOWrapper:
"""Open/Create the output file. If needed include headers or the required format"""
"""Open/Create the output file. If needed include headers or the required format, by default will use the CSVRow"""
try:
if file_exists(filename):
file_descriptor = open_file(
@@ -52,13 +48,12 @@ def initialize_file_descriptor(
file_descriptor, fieldnames=csv_header, delimiter=";"
)
csv_writer.writeheader()
return file_descriptor
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return file_descriptor
def fill_file_descriptors(output_modes, output_directory, output_filename, provider):
try:
@@ -67,36 +62,31 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
for output_mode in output_modes:
if output_mode == "csv":
filename = f"{output_directory}/{output_filename}{csv_file_suffix}"
output_model = get_provider_output_model(provider.type)
output_model = CSVRow
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
provider,
output_model,
)
file_descriptors.update({output_mode: file_descriptor})
elif output_mode == "json":
filename = f"{output_directory}/{output_filename}{json_file_suffix}"
file_descriptor = initialize_file_descriptor(
filename, output_mode, provider
)
file_descriptor = initialize_file_descriptor(filename, output_mode)
file_descriptors.update({output_mode: file_descriptor})
elif output_mode == "json-ocsf":
filename = (
f"{output_directory}/{output_filename}{json_ocsf_file_suffix}"
)
file_descriptor = initialize_file_descriptor(
filename, output_mode, provider
)
file_descriptor = initialize_file_descriptor(filename, output_mode)
file_descriptors.update({output_mode: file_descriptor})
elif provider.type == "gcp":
filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
if "cis_" in output_mode:
file_descriptor = initialize_file_descriptor(
filename, output_mode, provider, Check_Output_CSV_GCP_CIS
filename, output_mode, Check_Output_CSV_GCP_CIS
)
file_descriptors.update({output_mode: file_descriptor})
else:
@@ -112,7 +102,7 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
if output_mode == "json-asff":
filename = f"{output_directory}/{output_filename}{json_asff_file_suffix}"
file_descriptor = initialize_file_descriptor(
filename, output_mode, provider
filename, output_mode
)
file_descriptors.update({output_mode: file_descriptor})
else: # Compliance frameworks
@@ -121,7 +111,6 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
provider,
Check_Output_CSV_ENS_RD2022,
)
file_descriptors.update({output_mode: file_descriptor})
@@ -130,7 +119,6 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
provider,
Check_Output_CSV_AWS_CIS,
)
file_descriptors.update({output_mode: file_descriptor})
@@ -139,7 +127,6 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
provider,
Check_Output_CSV_AWS_Well_Architected,
)
file_descriptors.update({output_mode: file_descriptor})
@@ -148,7 +135,6 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
provider,
Check_Output_CSV_AWS_ISO27001_2013,
)
file_descriptors.update({output_mode: file_descriptor})
@@ -157,16 +143,15 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
provider,
Check_Output_MITRE_ATTACK,
)
file_descriptors.update({output_mode: file_descriptor})
else:
# Generic Compliance framework
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
provider,
Check_Output_CSV_Generic_Compliance,
)
file_descriptors.update({output_mode: file_descriptor})
prowler/lib/outputs/json.py
+2 -2
View File
@@ -187,8 +187,8 @@ def fill_json_ocsf(provider, finding, output_options) -> Check_Output_JSON_OCSF:
uid=finding.subscription,
)
org = Organization(
name=provider.identity.domain,
uid=provider.identity.domain,
name=provider.identity.tenant_domain,
uid=provider.identity.tenant_domain,
)
resource_name = finding.resource_name
resource_uid = finding.resource_id
prowler/lib/outputs/models.py
+2 -241
View File
@@ -1,8 +1,7 @@
import importlib
import sys
from csv import DictWriter
from datetime import datetime
from typing import Any, List, Literal, Optional
from typing import List, Literal, Optional
from pydantic import BaseModel
@@ -46,148 +45,6 @@ def get_check_compliance(finding, provider_type, output_options) -> dict:
sys.exit(1)
def generate_provider_output_csv(provider, finding, mode: str, fd, output_options):
"""
generate_provider_output_csv creates the provider's CSV output
"""
try:
# Dynamically load the Provider_Output_Options class
finding_output_model = (
f"{provider.type.capitalize()}_Check_Output_{mode.upper()}"
)
output_model = getattr(importlib.import_module(__name__), finding_output_model)
# Fill common data among providers
data = fill_common_data_csv(finding, output_options.unix_timestamp)
if provider.type == "azure":
data["resource_id"] = finding.resource_id
data["resource_name"] = finding.resource_name
data["subscription"] = finding.subscription
data["tenant_domain"] = provider.identity.domain
data["finding_unique_id"] = (
f"prowler-{provider.type}-{finding.check_metadata.CheckID}-{finding.subscription}-{finding.resource_id}"
)
data["compliance"] = unroll_dict(
get_check_compliance(finding, provider.type, output_options)
)
finding_output = output_model(**data)
if provider.type == "gcp":
data["resource_id"] = finding.resource_id
data["resource_name"] = finding.resource_name
data["project_id"] = finding.project_id
data["location"] = finding.location.lower()
data["finding_unique_id"] = (
f"prowler-{provider.type}-{finding.check_metadata.CheckID}-{finding.project_id}-{finding.resource_id}"
)
data["compliance"] = unroll_dict(
get_check_compliance(finding, provider.type, output_options)
)
finding_output = output_model(**data)
if provider.type == "kubernetes":
data["resource_id"] = finding.resource_id
data["resource_name"] = finding.resource_name
data["namespace"] = finding.namespace
data["context"] = provider.identity.context
data["finding_unique_id"] = (
f"prowler-{provider.type}-{finding.check_metadata.CheckID}-{finding.namespace}-{finding.resource_id}"
)
data["compliance"] = unroll_dict(
get_check_compliance(finding, provider.type, output_options)
)
finding_output = output_model(**data)
if provider.type == "aws":
data["profile"] = provider.identity.profile
data["account_id"] = provider.identity.account
data["region"] = finding.region
data["resource_id"] = finding.resource_id
data["resource_arn"] = finding.resource_arn
data["finding_unique_id"] = (
f"prowler-{provider.type}-{finding.check_metadata.CheckID}-{provider.identity.account}-{finding.region}-{finding.resource_id}"
)
data["compliance"] = unroll_dict(
get_check_compliance(finding, provider.type, output_options)
)
finding_output = output_model(**data)
if provider.organizations_metadata:
finding_output.account_name = (
provider.organizations_metadata.account_details_name
)
finding_output.account_email = (
provider.organizations_metadata.account_details_email
)
finding_output.account_arn = (
provider.organizations_metadata.account_details_arn
)
finding_output.account_org = (
provider.organizations_metadata.account_details_org
)
finding_output.account_tags = (
provider.organizations_metadata.account_details_tags
)
csv_writer = DictWriter(
fd,
fieldnames=generate_csv_fields(output_model),
delimiter=";",
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
return csv_writer, finding_output
def fill_common_data_csv(finding: dict, unix_timestamp: bool) -> dict:
data = {
"assessment_start_time": outputs_unix_timestamp(unix_timestamp, timestamp),
"finding_unique_id": "",
"provider": finding.check_metadata.Provider,
"check_id": finding.check_metadata.CheckID,
"check_title": finding.check_metadata.CheckTitle,
"check_type": ",".join(finding.check_metadata.CheckType),
"status": finding.status,
"status_extended": finding.status_extended,
"service_name": finding.check_metadata.ServiceName,
"subservice_name": finding.check_metadata.SubServiceName,
"severity": finding.check_metadata.Severity,
"resource_type": finding.check_metadata.ResourceType,
"resource_details": finding.resource_details,
"resource_tags": unroll_tags(finding.resource_tags),
"description": finding.check_metadata.Description,
"risk": finding.check_metadata.Risk,
"related_url": finding.check_metadata.RelatedUrl,
"remediation_recommendation_text": (
finding.check_metadata.Remediation.Recommendation.Text
),
"remediation_recommendation_url": (
finding.check_metadata.Remediation.Recommendation.Url
),
"remediation_recommendation_code_nativeiac": (
finding.check_metadata.Remediation.Code.NativeIaC
),
"remediation_recommendation_code_terraform": (
finding.check_metadata.Remediation.Code.Terraform
),
"remediation_recommendation_code_cli": (
finding.check_metadata.Remediation.Code.CLI
),
"remediation_recommendation_code_other": (
finding.check_metadata.Remediation.Code.Other
),
"categories": unroll_list(finding.check_metadata.Categories),
"depends_on": unroll_list(finding.check_metadata.DependsOn),
"related_to": unroll_list(finding.check_metadata.RelatedTo),
"notes": finding.check_metadata.Notes,
}
return data
def unroll_list(listed_items: list):
unrolled_items = ""
separator = "|"
@@ -276,102 +133,6 @@ def parse_json_tags(tags: list):
return dict_tags
def generate_csv_fields(format: Any) -> list[str]:
"""Generates the CSV headers for the given class"""
csv_fields = []
# __fields__ is always available in the Pydantic's BaseModel class
for field in format.__dict__.get("__fields__").keys():
csv_fields.append(field)
return csv_fields
class Check_Output_CSV(BaseModel):
"""
Check_Output_CSV generates a finding's output in CSV format.
This is the base CSV output model for every provider.
"""
assessment_start_time: str
finding_unique_id: str
provider: str
check_id: str
check_title: str
check_type: str
status: str
status_extended: str
service_name: str
subservice_name: str
severity: str
resource_type: str
resource_details: str
resource_tags: str
description: str
risk: str
related_url: str
remediation_recommendation_text: str
remediation_recommendation_url: str
remediation_recommendation_code_nativeiac: str
remediation_recommendation_code_terraform: str
remediation_recommendation_code_cli: str
remediation_recommendation_code_other: str
compliance: str
categories: str
depends_on: str
related_to: str
notes: str
class Aws_Check_Output_CSV(Check_Output_CSV):
"""
Aws_Check_Output_CSV generates a finding's output in CSV format for the AWS provider.
"""
profile: Optional[str]
account_id: int
account_name: Optional[str]
account_email: Optional[str]
account_arn: Optional[str]
account_org: Optional[str]
account_tags: Optional[str]
region: str
resource_id: str
resource_arn: str
class Azure_Check_Output_CSV(Check_Output_CSV):
"""
Azure_Check_Output_CSV generates a finding's output in CSV format for the Azure provider.
"""
tenant_domain: str = ""
subscription: str = ""
resource_id: str = ""
resource_name: str = ""
class Gcp_Check_Output_CSV(Check_Output_CSV):
"""
Gcp_Check_Output_CSV generates a finding's output in CSV format for the GCP provider.
"""
project_id: str = ""
location: str = ""
resource_id: str = ""
resource_name: str = ""
class Kubernetes_Check_Output_CSV(Check_Output_CSV):
"""
Kubernetes_Check_Output_CSV generates a finding's output in CSV format for the Kubernetes provider.
"""
context: str = ""
namespace: str = ""
resource_id: str = ""
resource_name: str = ""
def generate_provider_output_json(provider, finding, mode: str, output_options):
"""
generate_provider_output_json configures automatically the outputs based on the selected provider and returns the Check_Output_JSON object.
@@ -393,7 +154,7 @@ def generate_provider_output_json(provider, finding, mode: str, output_options):
finding_output.ResourceDetails = finding.resource_details
if provider.type == "azure":
finding_output.Tenant_Domain = provider.identity.domain
finding_output.Tenant_Domain = provider.identity.tenant_domain
finding_output.Subscription = finding.subscription
finding_output.ResourceId = finding.resource_id
finding_output.ResourceName = finding.resource_name
prowler/lib/outputs/outputs.py
+35 -9
View File
@@ -1,4 +1,5 @@
import json
from csv import DictWriter
from colorama import Fore, Style
@@ -8,12 +9,20 @@ from prowler.lib.outputs.compliance.compliance import (
add_manual_controls,
fill_compliance,
)
from prowler.lib.outputs.csv.csv import (
fill_common_data_csv,
generate_csv_fields,
generate_provider_output_csv,
get_provider_data_mapping,
)
from prowler.lib.outputs.csv.models import CSVRow
from prowler.lib.outputs.file_descriptors import fill_file_descriptors
from prowler.lib.outputs.json import fill_json_asff, fill_json_ocsf
from prowler.lib.outputs.models import (
Check_Output_JSON_ASFF,
generate_provider_output_csv,
generate_provider_output_json,
get_check_compliance,
unroll_dict,
)
@@ -104,16 +113,33 @@ def report(check_findings, provider):
)
file_descriptors["json-asff"].write(",")
# Common outputs
# CSV
if "csv" in file_descriptors:
csv_writer, finding_output = generate_provider_output_csv(
provider,
finding,
"csv",
file_descriptors["csv"],
output_options,
provider_data = get_provider_data_mapping(provider)
common_data = fill_common_data_csv(
finding, output_options.unix_timestamp
)
csv_writer.writerow(finding_output.__dict__)
compliance_data = unroll_dict(
get_check_compliance(
finding, provider.type, output_options
)
)
csv_data = {}
csv_data.update(provider_data)
csv_data.update(common_data)
csv_data["compliance"] = compliance_data
csv_writer = DictWriter(
file_descriptors["csv"],
fieldnames=generate_csv_fields(CSVRow),
delimiter=";",
)
finding_output = generate_provider_output_csv(
provider, finding, csv_data
)
csv_writer.writerow(finding_output.dict())
if "json" in file_descriptors:
finding_output = generate_provider_output_json(
prowler/lib/outputs/summary_table.py
+2 -2
View File
@@ -25,11 +25,11 @@ def display_summary_table(
audited_entities = provider.identity.account
elif provider.type == "azure":
if (
provider.identity.domain
provider.identity.tenant_domain
!= "Unknown tenant domain (missing AAD permissions)"
):
entity_type = "Tenant Domain"
audited_entities = provider.identity.domain
audited_entities = provider.identity.tenant_domain
else:
entity_type = "Tenant ID/s"
audited_entities = " ".join(provider.identity.tenant_ids)
prowler/providers/aws/aws_provider.py
+14
View File
@@ -282,6 +282,20 @@ class AwsProvider(Provider):
mutelist = {}
self._mutelist = mutelist
@property
def get_output_mapping(self):
return {
"auth_method": "identity.profile",
"provider": "type",
"account_uid": "identity.account",
"account_name": "organizations_metadata.account_details_name",
"account_email": "organizations_metadata.account_details_email",
"account_organization_uid": "organizations_metadata.account_details_arn",
"account_organization": "organizations_metadata.account_details_org",
"account_tags": "organizations_metadata.account_details_tags",
"partition": "identity.partition",
}
# TODO: This can be moved to another class since it doesn't need self
def get_organizations_info(
self, organizations_session: Session, aws_account_id: str
prowler/providers/aws/lib/organizations/organizations.py
+1 -1
View File
@@ -31,7 +31,7 @@ def parse_organizations_metadata(metadata: dict, tags: dict) -> AWSOrganizations
# Convert Tags dictionary to String
account_details_tags = ""
for tag in tags.get("Tags", {}):
account_details_tags += tag["Key"] + ":" + tag["Value"] + ","
account_details_tags += f"{tag['Key']}:{tag['Value']},"
account_details = metadata.get("Account", {})
organizations_info = AWSOrganizationsInfo(
prowler/providers/azure/azure_provider.py
+25 -3
View File
@@ -105,6 +105,25 @@ class AzureProvider(Provider):
arguments, bulk_checks_metadata, self._identity
)
@property
def get_output_mapping(self):
return {
# identity_type: identity_id
# "auth_method": "identity.profile",
"provider": "type",
# "account_uid": "identity.account",
# TODO: store subscription_name + id pairs
# "account_name": "organizations_metadata.account_details_name",
# "account_email": "organizations_metadata.account_details_email",
# TODO: check the tenant_ids
# TODO: we have to get the account organization, the tenant is not that
"account_organization_uid": "identity.tenant_ids",
"account_organization": "identity.tenant_domain",
# TODO: pending to get the subscription tags
# "account_tags": "organizations_metadata.account_details_tags",
"partition": "region_config.name",
}
# TODO: pending to implement
# @property
# def mutelist(self):
@@ -160,7 +179,7 @@ class AzureProvider(Provider):
report = f"""
This report is being generated using the identity below:
Azure Tenant IDs: {Fore.YELLOW}[{" ".join(self._identity.tenant_ids)}]{Style.RESET_ALL} Azure Tenant Domain: {Fore.YELLOW}[{self._identity.domain}]{Style.RESET_ALL} Azure Region: {Fore.YELLOW}[{self.region_config.name}]{Style.RESET_ALL}
Azure Tenant ID: {Fore.YELLOW}[{self._identity.tenant_ids[0]}]{Style.RESET_ALL} Azure Tenant Domain: {Fore.YELLOW}[{self._identity.tenant_domain}]{Style.RESET_ALL} Azure Region: {Fore.YELLOW}[{self.region_config.name}]{Style.RESET_ALL}
Azure Subscriptions: {Fore.YELLOW}{printed_subscriptions}{Style.RESET_ALL}
Azure Identity Type: {Fore.YELLOW}[{self._identity.identity_type}]{Style.RESET_ALL} Azure Identity ID: {Fore.YELLOW}[{self._identity.identity_id}]{Style.RESET_ALL}
"""
@@ -248,7 +267,7 @@ Azure Identity Type: {Fore.YELLOW}[{self._identity.identity_type}]{Style.RESET_A
domain_result = await client.domains.get()
if getattr(domain_result, "value"):
if getattr(domain_result.value[0], "id"):
identity.domain = domain_result.value[0].id
identity.tenant_domain = domain_result.value[0].id
except Exception as error:
logger.error(
@@ -300,6 +319,8 @@ Azure Identity Type: {Fore.YELLOW}[{self._identity.identity_type}]{Style.RESET_A
if not subscription_ids:
logger.info("Scanning all the Azure subscriptions...")
for subscription in subscriptions_client.subscriptions.list():
# TODO: get tags or labels
# TODO: fill with AzureSubscription
identity.subscriptions.update(
{subscription.display_name: subscription.subscription_id}
)
@@ -333,7 +354,7 @@ Azure Identity Type: {Fore.YELLOW}[{self._identity.identity_type}]{Style.RESET_A
return identity
def get_locations(self, credentials, region_config):
def get_locations(self, credentials, region_config) -> dict[str, list[str]]:
locations = None
if credentials and region_config:
subscriptions_client = SubscriptionClient(
@@ -342,6 +363,7 @@ Azure Identity Type: {Fore.YELLOW}[{self._identity.identity_type}]{Style.RESET_A
credential_scopes=region_config.credential_scopes,
)
list_subscriptions = subscriptions_client.subscriptions.list()
# TODO: use the identity subscritions
list_subscriptions_ids = [
subscription.subscription_id for subscription in list_subscriptions
]
prowler/providers/azure/models.py
+13 -3
View File
@@ -8,7 +8,7 @@ class AzureIdentityInfo(BaseModel):
identity_id: str = ""
identity_type: str = ""
tenant_ids: list[str] = []
domain: str = "Unknown tenant domain (missing AAD permissions)"
tenant_domain: str = "Unknown tenant domain (missing AAD permissions)"
subscriptions: dict = {}
locations: dict = {}
@@ -20,6 +20,13 @@ class AzureRegionConfig(BaseModel):
credential_scopes: list = []
class AzureSubscription(BaseModel):
id: str
subscription_id: str
display_name: str
state: str
class AzureOutputOptions(ProviderOutputOptions):
def __init__(self, arguments, bulk_checks_metadata, identity):
# First call Provider_Output_Options init
@@ -37,9 +44,12 @@ class AzureOutputOptions(ProviderOutputOptions):
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
if identity.domain != "Unknown tenant domain (missing AAD permissions)":
if (
identity.tenant_domain
!= "Unknown tenant domain (missing AAD permissions)"
):
self.output_filename = (
f"prowler-output-{identity.domain}-{output_file_timestamp}"
f"prowler-output-{identity.tenant_domain}-{output_file_timestamp}"
)
else:
self.output_filename = f"prowler-output-{'-'.join(identity.tenant_ids)}-{output_file_timestamp}"
prowler/providers/common/outputs.py
-78
View File
@@ -1,78 +0,0 @@
import importlib
# TODO: remove after fixing tests
# import sys
# from dataclasses import dataclass
# from os import makedirs
# from os.path import isdir
# from prowler.lib.logger import logger
# TODO: remove after fixing tests
# def set_provider_output_options(
# provider: str, arguments, identity, mutelist_file, bulk_checks_metadata
# ):
# """
# set_provider_output_options configures automatically the outputs based on the selected provider and returns the Provider_Output_Options object.
# """
# try:
# # Dynamically load the Provider_Output_Options class
# provider_output_class = f"{provider.capitalize()}_Output_Options"
# provider_output_options = getattr(
# importlib.import_module(__name__), provider_output_class
# )(arguments, identity, mutelist_file, bulk_checks_metadata)
# except Exception as error:
# logger.critical(
# f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
# )
# sys.exit(1)
# else:
# return provider_output_options
# TODO: review this function, probably is not needed anymore
def get_provider_output_model(provider_type):
"""
get_provider_output_model returns the model <provider>_Check_Output_CSV for each provider
"""
# TODO: classes should be AwsCheckOutputCSV
output_provider_model_name = f"{provider_type.capitalize()}_Check_Output_CSV"
output_provider_models_path = "prowler.lib.outputs.models"
output_provider_model = getattr(
importlib.import_module(output_provider_models_path), output_provider_model_name
)
return output_provider_model
# TODO: remove after fixing tests
# @dataclass
# class Provider_Output_Options:
# status: bool
# output_modes: list
# output_directory: str
# mutelist_file: str
# bulk_checks_metadata: dict
# verbose: str
# output_filename: str
# only_logs: bool
# unix_timestamp: bool
# def __init__(self, arguments, mutelist_file, bulk_checks_metadata):
# self.status = arguments.status
# self.output_modes = arguments.output_modes
# self.output_directory = arguments.output_directory
# self.verbose = arguments.verbose
# self.bulk_checks_metadata = bulk_checks_metadata
# self.mutelist_file = mutelist_file
# self.only_logs = arguments.only_logs
# self.unix_timestamp = arguments.unix_timestamp
# # Check output directory, if it is not created -> create it
# if arguments.output_directory:
# if not isdir(arguments.output_directory):
# if arguments.output_modes:
# makedirs(arguments.output_directory, exist_ok=True)
# if not isdir(arguments.output_directory + "/compliance"):
# if arguments.output_modes:
# makedirs(arguments.output_directory + "/compliance", exist_ok=True)
prowler/providers/common/provider.py
+8
View File
@@ -81,6 +81,14 @@ class Provider(ABC):
This method needs to be created in each provider.
"""
@abstractmethod
def get_output_mapping(self):
"""
get_output_mapping return the CSV output mapping between the provider and the generic model.
This method needs to be created in each provider.
"""
# TODO: probably this won't be here since we want to do the arguments validation during the parse()
def validate_arguments(self):
pass
prowler/providers/gcp/gcp_provider.py
+113 -12
View File
@@ -5,12 +5,18 @@ from colorama import Fore, Style
from google import auth
from google.oauth2.credentials import Credentials
from googleapiclient import discovery
from googleapiclient.errors import HttpError
from prowler.config.config import load_and_validate_config_file
from prowler.lib.logger import logger
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.provider import Provider
from prowler.providers.gcp.models import GCPIdentityInfo, GCPOutputOptions
from prowler.providers.gcp.models import (
GCPIdentityInfo,
GCPOrganization,
GCPOutputOptions,
GCPProject,
)
class GcpProvider(Provider):
@@ -34,15 +40,17 @@ class GcpProvider(Provider):
self._session, default_project_id = self.setup_session(credentials_file)
self._project_ids = []
accessible_projects = self.get_project_ids()
self._projects = {}
accessible_projects = self.get_projects()
if not accessible_projects:
logger.critical("No Project IDs can be accessed via Google Credentials.")
sys.exit(1)
if input_project_ids:
for input_project in input_project_ids:
if input_project in accessible_projects:
self._project_ids.append(input_project)
if input_project in accessible_projects.keys():
self._projects[input_project] = accessible_projects[input_project]
self._project_ids.append(accessible_projects[input_project].id)
else:
logger.critical(
f"Project {input_project} cannot be accessed via Google Credentials."
@@ -50,7 +58,12 @@ class GcpProvider(Provider):
sys.exit(1)
else:
# If not projects were input, all accessible projects are scanned by default
self._project_ids = accessible_projects
for project_id, project in accessible_projects.items():
self._projects[project_id] = project
self._project_ids.append(project_id)
# Update organizations info
self.update_projects_with_organizations()
self._identity = GCPIdentityInfo(
profile=getattr(self.session, "_service_account_email", "default"),
@@ -75,6 +88,10 @@ class GcpProvider(Provider):
def session(self):
return self._session
@property
def projects(self):
return self._projects
@property
def project_ids(self):
return self._project_ids
@@ -94,6 +111,27 @@ class GcpProvider(Provider):
arguments, bulk_checks_metadata, self._identity
)
@property
def get_output_mapping(self):
return {
# Account: identity.profile
"auth_method": "identity.profile",
"provider": "type",
# TODO: comes from finding, finding.project_id
# "account_uid": "",
# TODO: get project name from GCP
# "account_name": "organizations_metadata.account_details_name",
# There is no concept as project email in GCP
# "account_email": "organizations_metadata.account_details_email",
# TODO: get project organization ID from GCP
# "account_organization_uid": "organizations_metadata.account_details_arn",
# TODO: get project organization from GCP
# "account_organization": "",
# TODO: get project tags organization from GCP
# "account_tags": "organizations_metadata.account_details_tags",
# "partition": "identity.partition",
}
# TODO: pending to implement
# @property
# def mutelist(self):
@@ -131,8 +169,8 @@ class GcpProvider(Provider):
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = client_secrets_path
def print_credentials(self):
# Beautify audited profile, set "default" if there is no profile set
# TODO: Beautify audited profile, set "default" if there is no profile set
# TODO: improve print_credentials with more data like name, number, organization
report = f"""
This report is being generated using credentials below:
@@ -140,9 +178,9 @@ GCP Account: {Fore.YELLOW}[{self.identity.profile}]{Style.RESET_ALL} GCP Projec
"""
print(report)
def get_project_ids(self):
def get_projects(self) -> dict[str, GCPProject]:
try:
project_ids = []
projects = {}
service = discovery.build(
"cloudresourcemanager", "v1", credentials=self.session
@@ -154,18 +192,81 @@ GCP Account: {Fore.YELLOW}[{self.identity.profile}]{Style.RESET_ALL} GCP Projec
response = request.execute()
for project in response.get("projects", []):
project_ids.append(project["projectId"])
labels = ""
for key, value in project.get("labels", {}).items():
labels += f"{key}:{value},"
project_id = project["projectId"]
gcp_project = GCPProject(
number=project["projectNumber"],
id=project_id,
name=project["name"],
lifecycle_state=project["lifecycleState"],
labels=labels.rstrip(","),
)
if (
"parent" in project
and "type" in project["parent"]
and project["parent"]["type"] == "organization"
):
organization_id = project["parent"]["id"]
gcp_project.organization = GCPOrganization(
id=organization_id, name=f"organizations/{organization_id}"
)
projects[project_id] = gcp_project
request = service.projects().list_next(
previous_request=request, previous_response=response
)
return project_ids
except HttpError as http_error:
if http_error.status_code == 403 and "organizations" in http_error.uri:
logger.error(
f"{http_error.__class__.__name__}[{http_error.__traceback__.tb_lineno}]: {http_error.error_details} to get Organizations data."
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
# TODO: we cannot print this for whatever exception
print(
f"\n{Fore.YELLOW}Cloud Resource Manager API {Style.RESET_ALL}has not been used before or it is disabled.\nEnable it by visiting https://console.developers.google.com/apis/api/cloudresourcemanager.googleapis.com/ then retry."
)
return []
finally:
return projects
def update_projects_with_organizations(self):
try:
service = discovery.build(
"cloudresourcemanager", "v1", credentials=self._session
)
# TODO: this call requires more permissions to get that data
# resourcemanager.organizations.get --> add to the docs
for project in self._projects.values():
if project.organization:
request = service.organizations().get(
name=f"organizations/{project.organization.id}"
)
while request is not None:
response = request.execute()
project.organization.display_name = response.get("displayName")
request = service.projects().list_next(
previous_request=request, previous_response=response
)
except HttpError as http_error:
if http_error.status_code == 403 and "organizations" in http_error.uri:
logger.error(
f"{http_error.__class__.__name__}[{http_error.__traceback__.tb_lineno}]: {http_error.error_details} to get Organizations data."
)
else:
logger.error(
f"{http_error.__class__.__name__}[{http_error.__traceback__.tb_lineno}]: {http_error}"
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
prowler/providers/gcp/models.py
+20 -3
View File
@@ -1,15 +1,32 @@
from dataclasses import dataclass
from typing import Optional
from pydantic import BaseModel
from prowler.config.config import output_file_timestamp
from prowler.providers.common.models import ProviderOutputOptions
@dataclass
class GCPIdentityInfo:
class GCPIdentityInfo(BaseModel):
profile: str
default_project_id: str
class GCPOrganization(BaseModel):
id: str
name: str
# TODO: the name needs to be retrieved from another API
display_name: Optional[str]
class GCPProject(BaseModel):
number: str
id: str
name: str
organization: Optional[GCPOrganization]
labels: str
lifecycle_state: str
class GCPOutputOptions(ProviderOutputOptions):
def __init__(self, arguments, bulk_checks_metadata, identity):
# First call ProviderOutputOptions init
prowler/providers/gcp/services/iam/iam_no_service_roles_at_project_level/iam_no_service_roles_at_project_level.py
+1 -1
View File
@@ -28,7 +28,7 @@ class iam_no_service_roles_at_project_level(Check):
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.resource_name = project
report.status = "PASS"
report.location = cloudresourcemanager_client.region
report.status_extended = f"No IAM Users assigned to service roles at project level {project}."
prowler/providers/kubernetes/kubernetes_provider.py
+17 -1
View File
@@ -49,7 +49,7 @@ class KubernetesProvider(Provider):
self._identity = KubernetesIdentityInfo(
context=self._session.context["name"].replace(":", "_").replace("/", "_"),
user=self._session.context["context"]["user"],
cluster=self._session.context["context"]["user"],
cluster=self._session.context["context"]["cluster"],
)
# TODO: move this to the providers, pending for AWS, GCP, AZURE and K8s
@@ -89,6 +89,22 @@ class KubernetesProvider(Provider):
arguments, bulk_checks_metadata, self._identity
)
@property
def get_output_mapping(self):
return {
# "in-cluster/kubeconfig"
# "auth_method": "identity.profile",
"provider": "type",
# cluster: <context>
"account_uid": "identity.cluster",
# "account_name": "organizations_metadata.account_details_name",
# "account_email": "organizations_metadata.account_details_email",
# "account_organization_uid": "organizations_metadata.account_details_arn",
# "account_organization": "organizations_metadata.account_details_org",
# "account_tags": "organizations_metadata.account_details_tags",
# "partition": "identity.partition",
}
# TODO: pending to implement
# @property
# def mutelist(self):
tests/providers/common/common_outputs_test.py
+2 -17
View File
@@ -16,7 +16,6 @@ from prowler.providers.common.outputs import (
Azure_Output_Options,
Gcp_Output_Options,
Kubernetes_Output_Options,
get_provider_output_model,
set_provider_output_options,
)
from prowler.providers.gcp.lib.audit_info.models import GCP_Audit_Info
@@ -251,7 +250,7 @@ class Test_Common_Output_Options:
# Mock Azure Audit Info
audit_info = self.set_mocked_azure_audit_info()
audit_info.identity.domain = "test-domain"
audit_info.identity.tenant_domain = "test-domain"
mutelist_file = ""
bulk_checks_metadata = {}
@@ -270,7 +269,7 @@ class Test_Common_Output_Options:
assert output_options.verbose
assert (
output_options.output_filename
== f"prowler-output-{audit_info.identity.domain}-{DATETIME}"
== f"prowler-output-{audit_info.identity.tenant_domain}-{DATETIME}"
)
# Delete testing directory
@@ -316,17 +315,3 @@ class Test_Common_Output_Options:
# Delete testing directory
rmdir(arguments.output_directory)
def test_get_provider_output_model(self):
audit_info_class_names = [
"AWS_Audit_Info",
"GCP_Audit_Info",
"Azure_Audit_Info",
"Kubernetes_Audit_Info",
]
for class_name in audit_info_class_names:
provider_prefix = class_name.split("_", 1)[0].lower().capitalize()
assert (
get_provider_output_model(class_name).__name__
== f"{provider_prefix}_Check_Output_CSV"
)