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

fix(github_actions): use zizmor v1 format

Browse Source
This commit is contained in:
Andoni A.
2025-11-06 15:23:55 +01:00
parent 1f1e905d9e
commit aadcebfa0e
15 changed files with 862 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/__main__.py
+7 -4
View File
@@ -104,6 +104,7 @@ from prowler.providers.iac.models import IACOutputOptions
from prowler.providers.kubernetes.models import KubernetesOutputOptions
from prowler.providers.m365.models import M365OutputOptions
from prowler.providers.nhn.models import NHNOutputOptions
from prowler.providers.pipeline.models import PipelineOutputOptions
def prowler():
@@ -178,8 +179,8 @@ def prowler():
# Load compliance frameworks
logger.debug("Loading compliance frameworks from .json files")
# Skip compliance frameworks for IAC and GitHub Action providers
if provider not in ["iac", "github_action"]:
# Skip compliance frameworks for IAC, GitHub Action, and Pipeline providers
if provider not in ["iac", "github_action", "pipeline"]:
bulk_compliance_frameworks = Compliance.get_bulk(provider)
# Complete checks metadata with the compliance framework specification
bulk_checks_metadata = update_checks_metadata_with_compliance(
@@ -309,6 +310,8 @@ def prowler():
output_options = IACOutputOptions(args, bulk_checks_metadata)
elif provider == "github_action":
output_options = GithubActionOutputOptions(args, bulk_checks_metadata)
elif provider == "pipeline":
output_options = PipelineOutputOptions(args, bulk_checks_metadata)
# Run the quick inventory for the provider if available
if hasattr(args, "quick_inventory") and args.quick_inventory:
@@ -318,8 +321,8 @@ def prowler():
# Execute checks
findings = []
if provider in ["iac", "github_action"]:
# For IAC and GitHub Action providers, run the scan directly
if provider in ["iac", "github_action", "pipeline"]:
# For IAC, GitHub Action, and Pipeline providers, run the scan directly
findings = global_provider.run()
elif len(checks_to_execute):
findings = execute_checks(
prowler/lib/check/checks_loader.py
+2 -2
View File
@@ -20,8 +20,8 @@ def load_checks_to_execute(
) -> set:
"""Generate the list of checks to execute based on the cloud provider and the input arguments given"""
try:
# Bypass check loading for IAC and GitHub Action providers since they use external tools directly
if provider in ["iac", "github_action"]:
# Bypass check loading for IAC, GitHub Action, and Pipeline providers since they use external tools directly
if provider in ["iac", "github_action", "pipeline"]:
return set()
# Local subsets
prowler/lib/check/models.py
+18 -9
View File
@@ -685,25 +685,34 @@ class CheckReportGithubAction(Check_Report):
resource_name: str
resource_line_range: str
@dataclass
class CheckReportPipeline(Check_Report):
"""Contains the Pipeline Check's finding information using poutine."""
resource_name: str
resource_line_range: str
def __init__(
self, metadata: dict = {}, finding: dict = {}, workflow_path: str = ""
self, metadata: dict = {}, finding: dict = {}, pipeline_path: str = ""
) -> None:
"""
Initialize the GitHub Action Check's finding information from a zizmor finding dict.
Initialize the Pipeline Check's finding information from a poutine finding dict.
Args:
metadata (Dict): Optional check metadata (can be None).
finding (dict): A single finding result from zizmor's JSON output.
workflow_path (str): Path to the workflow file.
metadata (Dict): Check metadata as JSON string.
finding (dict): A single finding result from poutine's JSON output.
pipeline_path (str): Path to the pipeline file.
"""
super().__init__(metadata, finding)
self.resource = finding
self.resource_name = workflow_path
# Extract line range from location if available
self.resource_name = pipeline_path
# Extract line range from finding location
location = finding.get("location", {})
if location.get("line"):
start_line = location.get("line", "")
if location.get("start_line"):
start_line = location.get("start_line", "")
end_line = location.get("end_line", start_line)
self.resource_line_range = f"{start_line}:{end_line}" if start_line else ""
else:
prowler/lib/check/utils.py
+4 -4
View File
@@ -14,8 +14,8 @@ def recover_checks_from_provider(
Returns a list of tuples with the following format (check_name, check_path)
"""
try:
# Bypass check loading for IAC and GitHub Action providers since they use external tools directly
if provider in ["iac", "github_action"]:
# Bypass check loading for IAC, GitHub Action, and Pipeline providers since they use external tools directly
if provider in ["iac", "github_action", "pipeline"]:
return []
checks = []
@@ -63,8 +63,8 @@ def recover_checks_from_service(service_list: list, provider: str) -> set:
Returns a set of checks from the given services
"""
try:
# Bypass check loading for IAC and GitHub Action providers since they use external tools directly
if provider in ["iac", "github_action"]:
# Bypass check loading for IAC, GitHub Action, and Pipeline providers since they use external tools directly
if provider in ["iac", "github_action", "pipeline"]:
return set()
checks = set()
prowler/lib/cli/parser.py
+3 -2
View File
@@ -26,10 +26,10 @@ class ProwlerArgumentParser:
self.parser = argparse.ArgumentParser(
prog="prowler",
formatter_class=RawTextHelpFormatter,
usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,nhn,dashboard,iac,github_action} ...",
usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,nhn,dashboard,iac,github_action,pipeline} ...",
epilog="""
Available Cloud Providers:
{aws,azure,gcp,kubernetes,m365,github,iac,nhn,github_action}
{aws,azure,gcp,kubernetes,m365,github,iac,nhn,github_action,pipeline}
aws AWS Provider
azure Azure Provider
gcp GCP Provider
@@ -39,6 +39,7 @@ Available Cloud Providers:
iac IaC Provider (Preview)
nhn NHN Provider (Unofficial)
github_action GitHub Actions Security Provider
pipeline CI/CD Pipeline Security Provider
Available components:
dashboard Local dashboard
prowler/lib/outputs/finding.py
+9
View File
@@ -311,6 +311,15 @@ class Finding(BaseModel):
output_data["region"] = check_output.resource_line_range
output_data["resource_line_range"] = check_output.resource_line_range
output_data["framework"] = "zizmor"
elif provider.type == "pipeline":
output_data["auth_method"] = provider.auth_method
output_data["account_uid"] = provider.identity.scan_type
output_data["account_name"] = provider.identity.platform
output_data["resource_name"] = check_output.resource_name
output_data["resource_uid"] = check_output.resource_name
output_data["region"] = check_output.resource_line_range
output_data["resource_line_range"] = check_output.resource_line_range
output_data["framework"] = "poutine"
# check_output Unique ID
# TODO: move this to a function
prowler/lib/outputs/summary_table.py
+10
View File
@@ -68,6 +68,16 @@ def display_summary_table(
else:
entity_type = "Directory"
audited_entities = provider.workflow_path
elif provider.type == "pipeline":
if provider.organization:
entity_type = "Organization"
audited_entities = provider.organization
elif provider.repository_url:
entity_type = "Repository"
audited_entities = provider.repository_url
else:
entity_type = "Directory"
audited_entities = provider.scan_path
# Check if there are findings and that they are not all MANUAL
if findings and not all(finding.status == "MANUAL" for finding in findings):
prowler/providers/common/provider.py
+17 -2
View File
@@ -149,9 +149,11 @@ class Provider(ABC):
provider_class_path = (
f"{providers_path}.{arguments.provider}.{arguments.provider}_provider"
)
# Special handling for github_action provider
# Special handling for certain providers
if arguments.provider == "github_action":
provider_class_name = "GithubActionProvider"
elif arguments.provider == "pipeline":
provider_class_name = "PipelineProvider"
else:
provider_class_name = f"{arguments.provider.capitalize()}Provider"
provider_class = getattr(
@@ -241,6 +243,17 @@ class Provider(ABC):
mutelist_path=arguments.mutelist_file,
fixer_config=fixer_config,
)
elif "pipeline" in provider_class_name.lower():
provider_class(
scan_path=getattr(arguments, "scan_path", "."),
repository_url=getattr(arguments, "repository_url", None),
organization=getattr(arguments, "organization", None),
platform=getattr(arguments, "platform", "github"),
token=getattr(arguments, "token", None),
exclude_paths=getattr(arguments, "exclude_paths", []),
config_path=getattr(arguments, "config_file", None),
fixer_config=fixer_config,
)
elif "githubaction" in provider_class_name.lower():
provider_class(
workflow_path=getattr(arguments, "workflow_path", "."),
@@ -249,7 +262,9 @@ class Provider(ABC):
config_path=getattr(arguments, "config_file", None),
fixer_config=fixer_config,
github_username=getattr(arguments, "github_username", None),
personal_access_token=getattr(arguments, "personal_access_token", None),
personal_access_token=getattr(
arguments, "personal_access_token", None
),
oauth_app_token=getattr(arguments, "oauth_app_token", None),
)
elif "github" in provider_class_name.lower():
prowler/providers/github_action/github_action_provider.py
+111 -52
View File
@@ -14,7 +14,7 @@ from prowler.config.config import (
default_config_file_path,
load_and_validate_config_file,
)
from prowler.lib.check.models import CheckReportGithubAction, CheckMetadata
from prowler.lib.check.models import CheckMetadata, CheckReportGithubAction
from prowler.lib.logger import logger
from prowler.lib.utils.utils import print_boxes
from prowler.providers.common.models import Audit_Metadata
@@ -132,17 +132,48 @@ class GithubActionProvider(Provider):
"""GitHub Action provider doesn't need a session since it uses zizmor directly"""
return None
def _extract_workflow_file_from_location(self, location: dict) -> str:
"""
Extract the workflow file path from a location object.
Supports zizmor v1.x+ JSON format.
Args:
location: The location object from zizmor output
Returns:
The workflow file path, or None if not found
"""
try:
symbolic = location.get("symbolic", {})
# v1.x+ format: symbolic.key.Local.given_path
if "key" in symbolic:
key = symbolic["key"]
if isinstance(key, dict) and "Local" in key:
local = key["Local"]
if isinstance(local, dict) and "given_path" in local:
return local["given_path"]
logger.warning(f"Could not extract workflow file from location: {location}")
return None
except Exception as error:
logger.error(
f"Error extracting workflow file from location: {error.__class__.__name__} - {error}"
)
return None
def _process_zizmor_finding(
self, finding: dict, workflow_file: str, location: dict
) -> CheckReportGithubAction:
"""
Process a zizmor finding with the new JSON structure.
Process a zizmor finding (v1.x+ format).
Args:
finding: The finding object from zizmor output
workflow_file: The path to the workflow file
location: The specific location object for this finding
Returns:
CheckReportGithubAction: The processed check report
"""
@@ -151,7 +182,7 @@ class GithubActionProvider(Provider):
concrete_location = location.get("concrete", {}).get("location", {})
start = concrete_location.get("start_point", {})
end = concrete_location.get("end_point", {})
# Format line range
if start and end:
if start.get("row") == end.get("row"):
@@ -160,75 +191,86 @@ class GithubActionProvider(Provider):
line_range = f"lines {start.get('row', 'unknown')}-{end.get('row', 'unknown')}"
else:
line_range = "location unknown"
# Get determinations (severity/confidence)
determinations = finding.get("determinations", {})
severity = determinations.get("severity", "Unknown").lower()
confidence = determinations.get("confidence", "Unknown")
# Map zizmor severity to Prowler severity
severity_map = {
"critical": "critical",
"high": "high",
"high": "high",
"medium": "medium",
"low": "low",
"informational": "informational",
"unknown": "medium"
"unknown": "medium",
}
prowler_severity = severity_map.get(severity, "medium")
# Create CheckReport
finding_id = f"githubaction_{finding.get('ident', 'unknown').replace('-', '_')}"
finding_id = (
f"githubaction_{finding.get('ident', 'unknown').replace('-', '_')}"
)
# Prepare metadata dict
metadata = {
"Provider": "github_action",
"CheckID": finding_id,
"CheckTitle": finding.get("desc", "Unknown GitHub Actions Security Issue"),
"CheckTitle": finding.get(
"desc", "Unknown GitHub Actions Security Issue"
),
"CheckType": ["Security"],
"ServiceName": "githubaction",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": prowler_severity,
"ResourceType": "GitHubActionsWorkflow",
"Description": finding.get("desc", "Security issue detected in GitHub Actions workflow"),
"Risk": location.get("symbolic", {}).get("annotation", "Security risk in workflow"),
"Description": finding.get(
"desc", "Security issue detected in GitHub Actions workflow"
),
"Risk": location.get("symbolic", {}).get(
"annotation", "Security risk in workflow"
),
"RelatedUrl": finding.get("url", "https://docs.zizmor.sh/"),
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Review and fix the security issue in your GitHub Actions workflow",
"Terraform": ""
"Terraform": "",
},
"Recommendation": {
"Text": f"Review the security issue at {line_range} in {workflow_file}. {finding.get('desc', '')}",
"Url": finding.get("url", "https://docs.zizmor.sh/")
}
"Url": finding.get("url", "https://docs.zizmor.sh/"),
},
},
"Categories": ["security"],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
"Notes": "",
}
# Create the report with metadata as JSON string and finding
# Create the report - need to pass all required fields to the dataclass
# Since CheckReportGithubAction is a dataclass without custom __init__,
# we need to create it with all required fields from Check_Report
report = CheckReportGithubAction(
metadata=json.dumps(metadata),
finding=finding,
workflow_path=workflow_file
status="FAIL",
status_extended=(
f"GitHub Actions security issue found in {workflow_file} at {line_range}: "
f"{finding.get('desc', 'Unknown issue')}. "
f"Confidence: {confidence}. "
f"Details: {location.get('symbolic', {}).get('annotation', 'No details available')}"
),
check_metadata=CheckMetadata.parse_raw(json.dumps(metadata)),
resource=finding,
resource_details="",
resource_tags=[],
muted=False,
resource_name=workflow_file,
resource_line_range=line_range,
)
report.resource_name = workflow_file
report.resource_line_range = line_range
report.status = "FAIL"
report.status_extended = (
f"GitHub Actions security issue found in {workflow_file} at {line_range}: "
f"{finding.get('desc', 'Unknown issue')}. "
f"Confidence: {confidence}. "
f"Details: {location.get('symbolic', {}).get('annotation', 'No details available')}"
)
return report
except Exception as error:
logger.critical(
@@ -309,8 +351,21 @@ class GithubActionProvider(Provider):
self, directory: str, exclude_workflows: list[str]
) -> List[CheckReportGithubAction]:
try:
# Check zizmor version
try:
version_result = subprocess.run(
["zizmor", "--version"],
capture_output=True,
text=True,
timeout=5,
)
zizmor_version = version_result.stdout.strip()
logger.info(f"Using {zizmor_version}")
except Exception as version_error:
logger.warning(f"Could not determine zizmor version: {version_error}")
logger.info(f"Running GitHub Actions security scan on {directory} ...")
# Build zizmor command
zizmor_command = [
"zizmor",
@@ -318,11 +373,11 @@ class GithubActionProvider(Provider):
"--format",
"json",
]
# Add exclude patterns if provided
for exclude_pattern in exclude_workflows:
zizmor_command.extend(["--exclude", exclude_pattern])
with alive_bar(
ctrl_c=False,
bar="blocks",
@@ -331,8 +386,12 @@ class GithubActionProvider(Provider):
enrich_print=False,
) as bar:
try:
bar.title = f"-> Running GitHub Actions security scan on {directory} ..."
bar.title = (
f"-> Running GitHub Actions security scan on {directory} ..."
)
# Run zizmor with JSON output
# Note: zizmor exits with non-zero code when findings exist (e.g., 14)
# This is expected behavior, not an error
process = subprocess.run(
zizmor_command,
capture_output=True,
@@ -342,7 +401,7 @@ class GithubActionProvider(Provider):
except Exception as error:
bar.title = "-> Scan failed!"
raise error
# Log zizmor's stderr output
if process.stderr:
for line in process.stderr.strip().split("\n"):
@@ -356,12 +415,12 @@ class GithubActionProvider(Provider):
else:
logger.warning("No output returned from zizmor scan")
return []
# zizmor returns an array of findings directly
if not output or (isinstance(output, list) and len(output) == 0):
logger.info("No security issues found in GitHub Actions workflows")
return []
except json.JSONDecodeError as error:
# zizmor might not output JSON for certain cases
logger.warning(f"Failed to parse zizmor output as JSON: {error}")
@@ -380,12 +439,14 @@ class GithubActionProvider(Provider):
# Extract workflow file from the finding's location
if "locations" in finding and finding["locations"]:
for location in finding["locations"]:
if "symbolic" in location and "key" in location["symbolic"]:
key = location["symbolic"]["key"]
if "Local" in key:
workflow_file = key["Local"]["given_path"]
report = self._process_zizmor_finding(finding, workflow_file, location)
reports.append(report)
workflow_file = self._extract_workflow_file_from_location(
location
)
if workflow_file:
report = self._process_zizmor_finding(
finding, workflow_file, location
)
reports.append(report)
return reports
@@ -410,9 +471,7 @@ class GithubActionProvider(Provider):
f"Repository: {Fore.YELLOW}{self.repository_url}{Style.RESET_ALL}",
]
else:
report_title = (
f"{Style.BRIGHT}Scanning local GitHub Actions workflows:{Style.RESET_ALL}"
)
report_title = f"{Style.BRIGHT}Scanning local GitHub Actions workflows:{Style.RESET_ALL}"
report_lines = [
f"Directory: {Fore.YELLOW}{self.workflow_path}{Style.RESET_ALL}",
]
@@ -426,4 +485,4 @@ class GithubActionProvider(Provider):
f"Authentication method: {Fore.YELLOW}{self.auth_method}{Style.RESET_ALL}"
)
print_boxes(report_lines, report_title)
print_boxes(report_lines, report_title)
prowler/providers/pipeline/__init__.py
+5
View File
@@ -0,0 +1,5 @@
# Pipeline Provider
"""
Pipeline security provider for scanning CI/CD pipelines using Poutine.
Supports GitHub Actions, GitLab CI, Azure DevOps, and Tekton pipelines.
"""
prowler/providers/pipeline/lib/__init__.py
View File
prowler/providers/pipeline/lib/arguments/__init__.py
View File
prowler/providers/pipeline/lib/arguments/arguments.py
+74
View File
@@ -0,0 +1,74 @@
"""Pipeline Provider Arguments."""
def init_parser(self):
"""Init the Pipeline Provider CLI parser."""
pipeline_parser = self.subparsers.add_parser(
"pipeline",
parents=[self.common_providers_parser],
help="CI/CD Pipeline Security Provider (using Poutine)",
)
# Add common arguments
pipeline_parser.add_argument(
"--scan-path",
nargs="?",
default=".",
help="Path to local directory containing pipeline files (default: current directory)",
)
pipeline_parser.add_argument(
"--repository-url",
help="URL of remote repository to scan (e.g., https://github.com/org/repo)",
)
pipeline_parser.add_argument(
"--organization",
help="Organization name to scan all repositories",
)
pipeline_parser.add_argument(
"--platform",
choices=["github", "gitlab", "azure", "tekton"],
default="github",
help="CI/CD platform type (default: github)",
)
pipeline_parser.add_argument(
"--token",
help="Authentication token for the CI/CD platform",
)
pipeline_parser.add_argument(
"--exclude-paths",
nargs="+",
default=[],
help="Paths to exclude from scanning",
)
def validate_arguments(arguments):
"""Validate Pipeline Provider arguments."""
# Check for conflicting scan targets
targets = [
bool(arguments.scan_path and arguments.scan_path != "."),
bool(arguments.repository_url),
bool(arguments.organization),
]
if sum(targets) > 1:
return (
False,
"Only one of --scan-path, --repository-url, or --organization can be specified",
)
# Warn if token not provided for remote scanning
if (arguments.repository_url or arguments.organization) and not arguments.token:
print(
"\nWarning: Remote scanning without --token may have limited functionality. "
"Some security checks require authentication to detect.\n"
)
# Validate platform-specific requirements
if arguments.platform == "gitlab" and arguments.organization:
return (
False,
"Organization scanning is not supported for GitLab. Use --repository-url instead.",
)
return True, ""
prowler/providers/pipeline/models.py
+27
View File
@@ -0,0 +1,27 @@
"""Pipeline Provider Models."""
from pydantic import BaseModel
from prowler.config.config import output_file_timestamp
from prowler.providers.common.models import ProviderOutputOptions
class PipelineIdentityInfo(BaseModel):
"""Model for Pipeline Provider Identity Information."""
platform: str = "unknown"
organization: str = ""
repository: str = ""
scan_type: str = "local" # local, repository, or organization
class PipelineOutputOptions(ProviderOutputOptions):
"""Output options for the Pipeline provider."""
def __init__(self, arguments, bulk_checks_metadata=None):
"""Initialize the Pipeline output options."""
super().__init__(arguments, bulk_checks_metadata)
# Set default output filename if not specified
if not getattr(arguments, "output_filename", None):
self.output_filename = f"prowler-output-pipeline-{output_file_timestamp}"
prowler/providers/pipeline/pipeline_provider.py
+575
View File
@@ -0,0 +1,575 @@
"""Pipeline Provider for scanning CI/CD pipelines with Poutine."""
import json
import subprocess
import sys
from typing import List
from alive_progress import alive_bar
from prowler.config.config import (
default_config_file_path,
load_and_validate_config_file,
)
from prowler.lib.check.models import CheckReportPipeline
from prowler.lib.logger import logger
from prowler.lib.utils.utils import print_boxes
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.provider import Provider
from prowler.providers.pipeline.models import PipelineIdentityInfo
class PipelineProvider(Provider):
"""
PipelineProvider scans CI/CD pipelines for security vulnerabilities using Poutine.
Supports multiple pipeline platforms:
- GitHub Actions
- GitLab CI
- Azure DevOps
- Tekton (Pipelines as Code)
Detects security issues such as:
- Injection vulnerabilities
- Insecure secret usage
- Excessive permissions
- Supply chain risks
- Misconfigurations
"""
_type: str = "pipeline"
audit_metadata: Audit_Metadata
def __init__(
self,
scan_path: str = ".",
repository_url: str = None,
organization: str = None,
platform: str = "github", # github, gitlab, azure, tekton
token: str = None,
exclude_paths: List[str] = None,
config_path: str = None,
fixer_config: dict = None,
):
"""
Initialize the Pipeline Provider.
Args:
scan_path: Local directory path to scan
repository_url: Remote repository URL to scan
organization: Organization name to scan all repos
platform: CI/CD platform type (github, gitlab, azure, tekton)
token: Authentication token for the platform
exclude_paths: List of paths to exclude from scanning
config_path: Path to Prowler config file
fixer_config: Configuration for the fixer
"""
logger.info("Initializing Pipeline Provider for CI/CD security scanning")
self.scan_path = scan_path
self.repository_url = repository_url
self.organization = organization
self.platform = platform
self.token = token
self.exclude_paths = exclude_paths or []
# Provider type
self._type = "pipeline"
# Determine scan type
if organization:
scan_type = "organization"
elif repository_url:
scan_type = "repository"
else:
scan_type = "local"
# Set authentication method
if token:
self._auth_method = f"{platform.capitalize()} Token"
else:
self._auth_method = "No auth (local scan only)"
# Identity configuration
self._identity = PipelineIdentityInfo(
platform=platform,
organization=organization or "",
repository=repository_url or "",
scan_type=scan_type,
)
self.audited_account = organization or "local-pipelines"
self.region = "global"
# Session is not needed for Pipeline provider
self._session = None
# Load configurations
if config_path:
self._audit_config = load_and_validate_config_file(self._type, config_path)
else:
self._audit_config = load_and_validate_config_file(
self._type, default_config_file_path
)
self._fixer_config = fixer_config
# Initialize audit metadata
self.audit_metadata = Audit_Metadata(
provider=self._type,
account_id=scan_type,
account_name=platform,
region="global",
services_scanned=0, # Pipeline doesn't use services
expected_checks=[], # Pipeline doesn't use checks
completed_checks=0,
audit_progress=0,
)
# Setup session (no-op for pipeline provider)
self.setup_session()
# Set the global provider
Provider.set_global_provider(self)
@property
def auth_method(self):
return self._auth_method
@property
def type(self):
return self._type
@property
def identity(self):
return self._identity
@property
def session(self):
return self._session
@property
def audit_config(self):
return self._audit_config
@property
def fixer_config(self):
return self._fixer_config
def setup_session(self):
"""Pipeline provider doesn't need a session since it uses poutine directly"""
return None
def _process_poutine_finding(
self, finding: dict, file_path: str
) -> CheckReportPipeline:
"""
Process a poutine finding and create a CheckReportPipeline object.
Args:
finding: The finding object from poutine JSON output
file_path: The path to the pipeline file
Returns:
CheckReportPipeline: The processed check report
"""
try:
# Extract finding details from poutine structure
# Finding has: rule_id, purl, meta (with details, job, line, path, step)
rule_id = (
finding.get("rule_id", "unknown").replace("-", "_").replace(".", "_")
)
finding_id = f"pipeline_{rule_id}"
# Get metadata
meta = finding.get("meta", {})
# For poutine, we need to get severity from the rules dict in the parent output
# For now, use a default mapping based on rule_id
severity_map = {
"injection": "high",
"debug_enabled": "low",
"default_permissions_on_risky_events": "medium",
"github_action_from_unverified_creator_used": "low",
"if_always_true": "high",
"job_all_secrets": "medium",
"known_vulnerability_in_build_component": "high",
"known_vulnerability_in_build_platform": "high",
"pr_runs_on_self_hosted": "medium",
"unpinnable_action": "low",
"untrusted_checkout_exec": "critical",
"unverified_script_exec": "high",
}
prowler_severity = severity_map.get(rule_id, "medium")
# Extract location information from meta
start_line = meta.get("line", 0)
end_line = start_line # Poutine doesn't provide end_line
if start_line:
if start_line == end_line:
line_range = f"line {start_line}"
else:
line_range = f"lines {start_line}-{end_line}"
else:
line_range = "location unknown"
# Get rule name from rule_id (use title case)
rule_title = rule_id.replace("_", " ").title()
# Build description from meta details
details = meta.get("details", "")
job = meta.get("job", "")
step = meta.get("step", "")
description = f"Security issue detected in pipeline: {rule_title}"
if job:
description += f" in job '{job}'"
if step:
description += f" at step {step}"
if details:
description += f". {details}"
# Prepare metadata
metadata = {
"Provider": "pipeline",
"CheckID": finding_id,
"CheckTitle": rule_title,
"CheckType": ["Security", "CI/CD"],
"ServiceName": "pipeline",
"SubServiceName": self.platform,
"ResourceIdTemplate": "",
"Severity": prowler_severity,
"ResourceType": "Pipeline",
"Description": description,
"Risk": f"Potential {prowler_severity} severity security risk in CI/CD pipeline",
"RelatedUrl": "https://boostsecurityio.github.io/poutine/rules/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": f"Review and fix the {rule_title.lower()} issue in your pipeline",
"Terraform": "",
},
"Recommendation": {
"Text": f"Fix the security issue at {line_range} in {file_path}",
"Url": f"https://boostsecurityio.github.io/poutine/rules/{rule_id}",
},
},
"Categories": ["security", "cicd"],
"DependsOn": [],
"RelatedTo": [],
"Notes": "",
}
# Create the report
report = CheckReportPipeline(
metadata=json.dumps(metadata), finding=finding, pipeline_path=file_path
)
report.resource_name = file_path
report.resource_line_range = line_range
report.status = "FAIL"
report.status_extended = (
f"Pipeline security issue found in {file_path} at {line_range}: "
f"{rule_title}. "
f"Severity: {prowler_severity}. "
f"Details: {details if details else 'No additional details available'}"
)
return report
except Exception as error:
logger.critical(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
)
sys.exit(1)
def run(self) -> List[CheckReportPipeline]:
"""
Execute poutine scan on the configured target.
Returns:
List of CheckReportPipeline objects with findings
"""
try:
if self.repository_url:
return self._scan_repository(self.repository_url)
elif self.organization:
return self._scan_organization(self.organization)
else:
return self._scan_local(self.scan_path)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
)
sys.exit(1)
def _scan_local(self, path: str) -> List[CheckReportPipeline]:
"""
Scan a local directory for pipeline security issues.
Args:
path: Directory path to scan
Returns:
List of CheckReportPipeline objects
"""
try:
# Build poutine command
poutine_command = ["poutine", "analyze_local", path, "--format", "json"]
# Add exclude paths if provided
for exclude_path in self.exclude_paths:
poutine_command.extend(["--exclude", exclude_path])
with alive_bar(
ctrl_c=False,
bar="blocks",
spinner="classic",
stats=False,
enrich_print=False,
) as bar:
try:
bar.title = f"-> Running pipeline security scan on {path} ..."
# Run poutine with JSON output
process = subprocess.run(
poutine_command,
capture_output=True,
text=True,
)
bar.title = "-> Scan completed!"
except Exception as error:
bar.title = "-> Scan failed!"
raise error
# Log poutine's stderr output
if process.stderr:
for line in process.stderr.strip().split("\n"):
if line.strip():
logger.debug(f"poutine: {line}")
try:
# Parse poutine JSON output
if process.stdout:
output = json.loads(process.stdout)
else:
logger.warning("No output returned from poutine scan")
return []
# Check if poutine found any issues
if not output:
logger.info("No security issues found in pipelines")
return []
except json.JSONDecodeError as error:
logger.warning(f"Failed to parse poutine output as JSON: {error}")
logger.debug(f"Raw output: {process.stdout}")
return []
reports = []
# Process poutine findings
if isinstance(output, dict) and "findings" in output:
# Poutine returns findings in a "findings" array
for finding in output["findings"]:
# Extract path from meta field
file_path = finding.get("meta", {}).get("path", "unknown")
report = self._process_poutine_finding(finding, file_path)
reports.append(report)
elif isinstance(output, list):
# Alternative format: array of findings
for finding in output:
file_path = finding.get("path", "unknown")
report = self._process_poutine_finding(finding, file_path)
reports.append(report)
return reports
except Exception as error:
if "No such file or directory: 'poutine'" in str(error):
logger.critical(
"poutine binary not found. Please install poutine from https://github.com/boostsecurityio/poutine "
"or use 'brew install poutine' on macOS"
)
sys.exit(1)
logger.critical(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
)
sys.exit(1)
def _scan_repository(self, repository_url: str) -> List[CheckReportPipeline]:
"""
Scan a remote repository for pipeline security issues.
Args:
repository_url: URL of the repository to scan
Returns:
List of CheckReportPipeline objects
"""
try:
# Extract org and repo from URL
# Format: https://github.com/org/repo
parts = repository_url.rstrip("/").split("/")
if len(parts) >= 2:
repo_path = f"{parts[-2]}/{parts[-1]}"
else:
logger.error(f"Invalid repository URL: {repository_url}")
return []
# Build poutine command
poutine_command = ["poutine", "analyze_repo", repo_path, "--format", "json"]
# Add token if provided
if self.token:
poutine_command.extend(["--token", self.token])
with alive_bar(
ctrl_c=False,
bar="blocks",
spinner="classic",
stats=False,
enrich_print=False,
) as bar:
try:
bar.title = f"-> Scanning repository {repo_path} ..."
process = subprocess.run(
poutine_command,
capture_output=True,
text=True,
)
bar.title = "-> Scan completed!"
except Exception as error:
bar.title = "-> Scan failed!"
raise error
# Process output same as local scan
return self._process_scan_output(process)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
)
sys.exit(1)
def _scan_organization(self, organization: str) -> List[CheckReportPipeline]:
"""
Scan all repositories in an organization.
Args:
organization: Organization name
Returns:
List of CheckReportPipeline objects
"""
try:
# Build poutine command
poutine_command = [
"poutine",
"analyze_org",
organization,
"--format",
"json",
]
# Add token if provided
if self.token:
poutine_command.extend(["--token", self.token])
with alive_bar(
ctrl_c=False,
bar="blocks",
spinner="classic",
stats=False,
enrich_print=False,
) as bar:
try:
bar.title = f"-> Scanning organization {organization} ..."
process = subprocess.run(
poutine_command,
capture_output=True,
text=True,
)
bar.title = "-> Scan completed!"
except Exception as error:
bar.title = "-> Scan failed!"
raise error
# Process output same as local scan
return self._process_scan_output(process)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
)
sys.exit(1)
def _process_scan_output(self, process) -> List[CheckReportPipeline]:
"""Process the output from poutine scan."""
# Log stderr
if process.stderr:
for line in process.stderr.strip().split("\n"):
if line.strip():
logger.debug(f"poutine: {line}")
try:
# Parse JSON output
if process.stdout:
output = json.loads(process.stdout)
else:
logger.warning("No output returned from poutine scan")
return []
if not output:
logger.info("No security issues found in pipelines")
return []
except json.JSONDecodeError as error:
logger.warning(f"Failed to parse poutine output as JSON: {error}")
return []
reports = []
# Process findings based on output format
if isinstance(output, dict) and "findings" in output:
for file_path, findings in output["findings"].items():
for finding in findings:
report = self._process_poutine_finding(finding, file_path)
reports.append(report)
elif isinstance(output, list):
for finding in output:
file_path = finding.get("path", "unknown")
report = self._process_poutine_finding(finding, file_path)
reports.append(report)
return reports
def print_credentials(self):
"""Print the credentials used for the scan."""
if self.organization:
scan_info = [[f"Organization: {self.organization}"]]
elif self.repository_url:
scan_info = [[f"Repository: {self.repository_url}"]]
else:
scan_info = [[f"Directory: {self.scan_path}"]]
scan_info.append([f"Platform: {self.platform}"])
scan_info.append([f"Authentication method: {self._auth_method}"])
if self.exclude_paths:
scan_info.append([f"Excluded paths: {', '.join(self.exclude_paths)}"])
title = (
"Scanning CI/CD pipelines for security issues:"
if not self.organization and not self.repository_url
else f"Scanning {self.platform.capitalize()} pipelines:"
)
print_boxes(scan_info, title)
def test_connection(self):
"""Test the connection to the platform if using remote scanning."""
if self.token and (self.repository_url or self.organization):
# Could implement a test API call here
return True
return True