From 8cb8e7533e1359df242ac37fde7091a85e4a18c1 Mon Sep 17 00:00:00 2001 From: Ishaan Jain Date: Thu, 9 Jul 2026 10:02:49 +0530 Subject: [PATCH] feat: add --severity threshold gating for SonarQube SAST Allow sq-sast scans to fail when SonarQube issue severities or hotspot vulnerability probabilities match a configured --severity list. --- README.md | 9 ++++++ aspm_cli/scan/sq_sast.py | 43 ++++++++++++++++++++++++++-- aspm_cli/scanners/sq_sast_scanner.py | 15 ++++++++-- aspm_cli/utils/config.py | 27 +++++++++++++++-- 4 files changed, 88 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index d9853ee..65fc479 100644 --- a/README.md +++ b/README.md @@ -572,6 +572,7 @@ Flags used after `sq-sast`: - `--skip-sonar-scan` - `--container-mode` +- `--severity` — Comma-separated severities that fail the scan. Matches SonarQube issue severity (`INFO,MINOR,MAJOR,CRITICAL,BLOCKER`) and hotspot `vulnerabilityProbability` (`LOW,MEDIUM,HIGH`). Defaults to all. - `--repo-url` - `--branch` - `--commit-sha` @@ -589,6 +590,14 @@ Example: accuknox-aspm-scanner scan --skip-upload --keep-results sq-sast --command "-Dsonar.projectKey= -Dsonar.host.url= -Dsonar.token= -Dsonar.organization=" ``` +Fail the pipeline when Critical/Blocker issues (or High hotspots) are found: + +```bash +accuknox-aspm-scanner scan --skip-upload --keep-results sq-sast \ + --command "-Dsonar.projectKey= -Dsonar.host.url= -Dsonar.token= -Dsonar.organization=" \ + --severity "CRITICAL,BLOCKER,HIGH" +``` + Important note: - Even with `--skip-sonar-scan`, `--command` is still required by the current parser diff --git a/aspm_cli/scan/sq_sast.py b/aspm_cli/scan/sq_sast.py index 01a12f0..abdbaaa 100644 --- a/aspm_cli/scan/sq_sast.py +++ b/aspm_cli/scan/sq_sast.py @@ -11,9 +11,10 @@ class SQSASTScanner: sast_image = os.getenv("SCAN_IMAGE", "public.ecr.aws/k9v9d5v2/sonarsource/sonar-scanner-cli:11.4") + DEFAULT_SEVERITY = "INFO,MINOR,MAJOR,CRITICAL,BLOCKER,LOW,MEDIUM,HIGH" def __init__(self, skip_sonar_scan, command, container_mode=False, repo_url=None, branch=None, - commit_sha=None, pipeline_url=None): + commit_sha=None, pipeline_url=None, severity=None): """ :param command: Raw command string (e.g., "-Dsonar.projectKey=... -Dsonar.token=...") :param container_mode: If True, run sonar-scanner natively instead of Docker @@ -21,6 +22,7 @@ def __init__(self, skip_sonar_scan, command, container_mode=False, repo_url=None :param branch: Branch name :param commit_sha: Git commit SHA :param pipeline_url: CI/CD pipeline URL + :param severity: Comma-separated severities that fail the scan """ self.skip_sonar_scan = skip_sonar_scan self.command = command @@ -29,6 +31,11 @@ def __init__(self, skip_sonar_scan, command, container_mode=False, repo_url=None self.branch = branch self.commit_sha = commit_sha self.pipeline_url = pipeline_url + self.severity = [ + s.strip().upper() + for s in (severity or self.DEFAULT_SEVERITY).split(",") + if s.strip() + ] # Extract needed values from command (for fetcher) self.sonar_project_key = self._extract_arg("-Dsonar.projectKey") @@ -45,7 +52,15 @@ def run(self): Logger.get_logger().info("SQ SAST scan skipped.") result_file = self._run_ak_scan() self._process_result_file(result_file) - return returncode, result_file + if self._severity_threshold_met(result_file): + Logger.get_logger().error( + f"Vulnerabilities matching severities: {', '.join(self.severity)} found." + ) + return 1, result_file + # Prefer severity gating for findings; still surface scanner hard failures. + if returncode != 0: + return returncode, result_file + return 0, result_file except subprocess.CalledProcessError as e: Logger.get_logger().error(f"SonarQube-based AccuKnox SAST scan failed: {e}") raise @@ -117,6 +132,30 @@ def _process_result_file(self, file_path): Logger.get_logger().error(f"Error during SQ SAST scan: {e}") raise + def _severity_threshold_met(self, file_path): + """ + Return True if any SonarQube issue severity or hotspot vulnerability + probability matches the configured --severity list. + """ + try: + with open(file_path, "r") as f: + data = json.load(f) + + for issue in data.get("issues") or []: + severity = (issue.get("severity") or "").upper() + if severity in self.severity: + return True + + for hotspot in data.get("hotspots") or []: + probability = (hotspot.get("vulnerabilityProbability") or "").upper() + if probability in self.severity: + return True + + return False + except Exception as e: + Logger.get_logger().error(f"Error evaluating SQ SAST severity threshold: {e}") + raise + def _extract_arg(self, key): """Extracts a value from the raw command string (e.g. -Dsonar.projectKey=foo).""" for arg in shlex.split(self.command): diff --git a/aspm_cli/scanners/sq_sast_scanner.py b/aspm_cli/scanners/sq_sast_scanner.py index b7ca770..ab7ba2a 100644 --- a/aspm_cli/scanners/sq_sast_scanner.py +++ b/aspm_cli/scanners/sq_sast_scanner.py @@ -21,6 +21,15 @@ def add_arguments(self, parser: argparse.ArgumentParser): action="store_true", help="Run in container mode" ) + parser.add_argument( + "--severity", + default="INFO,MINOR,MAJOR,CRITICAL,BLOCKER,LOW,MEDIUM,HIGH", + help=( + "Comma-separated severities that fail the scan. " + "Matches SonarQube issue severity (INFO,MINOR,MAJOR,CRITICAL,BLOCKER) " + "and hotspot vulnerabilityProbability (LOW,MEDIUM,HIGH). Defaults to all." + ), + ) parser.add_argument("--repo-url", default=GitInfo.get_repo_url(), help="Git repository URL") parser.add_argument("--branch", default=GitInfo.get_branch_name(), help="Git repository branch") parser.add_argument("--commit-sha", default=GitInfo.get_commit_sha(), help="Commit SHA for scanning") @@ -29,12 +38,14 @@ def add_arguments(self, parser: argparse.ArgumentParser): def validate_config(self, args: argparse.Namespace, validator: ConfigValidator): validator.validate_sq_sast_scan( args.skip_sonar_scan, args.command, args.container_mode, - args.repo_url, args.branch, args.commit_sha, args.pipeline_url + args.repo_url, args.branch, args.commit_sha, args.pipeline_url, + args.severity, ) def run_scan(self, args: argparse.Namespace) -> tuple[int, str]: scanner = OriginalSQSASTScanner( args.skip_sonar_scan, args.command, args.container_mode, - args.repo_url, args.branch, args.commit_sha, args.pipeline_url + args.repo_url, args.branch, args.commit_sha, args.pipeline_url, + severity=args.severity, ) return scanner.run() \ No newline at end of file diff --git a/aspm_cli/utils/config.py b/aspm_cli/utils/config.py index 42b1b4b..497162e 100644 --- a/aspm_cli/utils/config.py +++ b/aspm_cli/utils/config.py @@ -136,7 +136,7 @@ def validate_severity(cls, v: str, info: FieldValidationInfo): Logger.get_logger().debug(f"IAC scan configuration error: {concise_msg}") raise ValueError(concise_msg) - def validate_sq_sast_scan(self, skip_sonar_scan: bool, command: str, container_mode: bool, repo_url: Optional[str], branch: Optional[str], commit_sha: Optional[str], pipeline_url: Optional[str]): + def validate_sq_sast_scan(self, skip_sonar_scan: bool, command: str, container_mode: bool, repo_url: Optional[str], branch: Optional[str], commit_sha: Optional[str], pipeline_url: Optional[str], severity: str): class SQSAScanConfig(BaseModel): skip_sonar_scan: bool command: str = Field(..., min_length=1, description="Command arguments for SQ SAST scanner") @@ -145,6 +145,28 @@ class SQSAScanConfig(BaseModel): branch: Optional[str] commit_sha: Optional[str] pipeline_url: Optional[str] + severity: str = Field(..., description="Comma-separated list of severities") + + @field_validator("severity", mode="before") + @classmethod + def validate_severity(cls, v: str, info: FieldValidationInfo): + # Issue severity + hotspot vulnerabilityProbability + allowed_severities = { + "INFO", "MINOR", "MAJOR", "CRITICAL", "BLOCKER", + "LOW", "MEDIUM", "HIGH", + } + if not isinstance(v, str): + raise ValueError("Severity must be a comma-separated string.") + provided_severities = {s.strip().upper() for s in v.split(",") if s.strip()} + if not provided_severities: + raise ValueError("At least one severity must be provided.") + invalid = provided_severities - allowed_severities + if invalid: + raise ValueError( + f"Invalid severity values: {', '.join(sorted(invalid))}. " + f"Allowed values: {', '.join(sorted(allowed_severities))}." + ) + return ",".join(sorted(provided_severities)) @model_validator(mode='after') def check_sonar_command_if_not_skipped(self) -> 'SQSAScanConfig': @@ -160,7 +182,8 @@ def check_sonar_command_if_not_skipped(self) -> 'SQSAScanConfig': repo_url=repo_url, branch=branch, commit_sha=commit_sha, - pipeline_url=pipeline_url + pipeline_url=pipeline_url, + severity=severity, ) self._log_validation_success("SQ SAST") except ValidationError as e: