feat: add --severity threshold gating for SonarQube SAST

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=<PROJECT_KEY> -Dsonar.host.url=<HOST_URL> -Dsonar.token=<TOKEN> -Dsonar.organization=<ORG_ID>"
 ```
 
+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=<PROJECT_KEY> -Dsonar.host.url=<HOST_URL> -Dsonar.token=<TOKEN> -Dsonar.organization=<ORG_ID>" \
+  --severity "CRITICAL,BLOCKER,HIGH"
+```
+
 Important note:
 
 - Even with `--skip-sonar-scan`, `--command` is still required by the current parser

aspm_cli/scan/sq_sast.py

@@ -11,16 +11,18 @@
 
 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
         :param repo_url: Git repository URL
         :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):

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()

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: