Merge pull request #36 from Eshrath027/MK-1945

[MK-1947]:Changes to integrate codeassure in ASPM scanner cli

Author: Ishaanj18

.github/workflows/release.yml

@@ -117,7 +117,7 @@ jobs:
       contents: write
 
     env:
-      PYTHON_VERSION: "3.10"
+      PYTHON_VERSION: "3.12"
 
     steps:
       - name: Checkout code
@@ -152,6 +152,12 @@ jobs:
           ls -lahR ../../
           cd ../../
 
+      - name: Upload Scanner Tarballs to Release
+        uses: softprops/action-gh-release@v1
+        with:
+          files: |
+            utils/*.tar.gz
+
       - name: Upload ASPM DEB and Scanners for Linux
         uses: softprops/action-gh-release@v1
         with:

aspm_cli/scan/sast.py

@@ -12,12 +12,12 @@
 
 class SASTScanner:
     opengrep_image = os.getenv("SCAN_IMAGE", "public.ecr.aws/k9v9d5v2/accuknox/opengrepjob:0.1.0")
-    claude_image = os.getenv("CLAUDE_IMAGE", "public.ecr.aws/k9v9d5v2/accuknox/ai-sast-claude-cli:latest")
+    codeassure_image = os.getenv("CODEASSURE_IMAGE", "public.ecr.aws/k9v9d5v2/accuknox/ai-sast-codeassure-cli:0.1.0")
     result_file = "results.json"
 
     def __init__(self, command=None, container_mode=True, severity = None,
                  repo_url=None, commit_ref=None, commit_sha=None,
-                 pipeline_id=None, job_url=None, anthropic_api_key=None, ai_analysis=False):
+                 pipeline_id=None, job_url=None, ai_analysis=True, codeassure_config=None,aiscan_severity=None):
         """
         :param command: Raw OpenGrep CLI args (string)
         :param container_mode: Run in Docker if True, else use local binary
@@ -26,19 +26,19 @@ def __init__(self, command=None, container_mode=True, severity = None,
         :param commit_sha: Commit SHA
         :param pipeline_id: CI pipeline ID
         :param job_url: CI job URL
-        :param anthropic_api_key: Anthropic API key for AI analysis
         :param ai_analysis: Enable AI analysis of results
         """
         self.command = command
         self.container_mode = container_mode
         self.severity = [s.strip().upper() for s in (severity).split(',')]
+        self.aiscan_severity = [s.strip().upper() for s in aiscan_severity.split(',')] if ai_analysis and isinstance(aiscan_severity, str) else []
         self.repo_url = repo_url
         self.commit_ref = commit_ref
         self.commit_sha = commit_sha
         self.pipeline_id = pipeline_id
         self.job_url = job_url
-        self.anthropic_api_key = anthropic_api_key or os.getenv("ANTHROPIC_API_KEY")
         self.ai_analysis = ai_analysis
+        self.codeassure_config = codeassure_config
 
     def run(self):
         try:
@@ -79,6 +79,7 @@ def run(self):
                     try:
                         Logger.get_logger().debug("Starting AI analysis of SAST results...")
                         self._run_ai_analysis()
+                        self._apply_verification_fields()
                     except Exception as e:
                         Logger.get_logger().error(f"AI analysis failed: {e}")
 
@@ -101,169 +102,103 @@ def _run_ai_analysis(self):
         This ensures the scan continues successfully even if AI analysis fails.
         """
         try:
-            if not self.anthropic_api_key:
-                Logger.get_logger().warning("Anthropic API key not provided. Skipping AI analysis.")
-                return
-            
             if self.container_mode:
-                docker_pull(self.claude_image)
+                docker_pull(self.codeassure_image)
+            else:
+                ToolManager.get_path("codeassure")  # raises FileNotFoundError if not installed
 
             # Check if there are any results to analyze
             with open(self.result_file, 'r') as f:
                 current_data = json.load(f)
 
             results = current_data.get("results", [])
-            Logger.get_logger().info(f"Running Claude AI analysis: {len(results)} findings to analyze.")
+            Logger.get_logger().info(f"Running AI analysis: {len(results)} findings to analyze.")
 
             if not results or len(results) == 0:
                 Logger.get_logger().debug("No results to analyze. Skipping AI analysis.")
                 return
 
 
-            cmd = self._build_claude_command()
-            ai_result = subprocess.run(cmd, capture_output=True, text=True, check=False)
-
-            if ai_result.stderr:
-                Logger.get_logger().info(f"Claude analysis stderr: {ai_result.stderr}")
+            cmd = self._build_ai_analysis_command()
+ 
+            ai_result = subprocess.run(cmd, check=False)
 
             if ai_result.returncode != 0:
-                Logger.get_logger().info(f"AI analysis failed with exit code: {ai_result.returncode}.")
-                if ai_result.stderr:
-                    Logger.get_logger().warning(f"Error details: {ai_result.stderr[:500]}")
-                if ai_result.stdout:
-                    Logger.get_logger().info(f"Claude stdout: {ai_result.stdout[:500]}")
-                Logger.get_logger().warning("Continuing with original results.")
-                return
-
-            if not ai_result.stdout:
-                Logger.get_logger().warning("AI analysis returned empty output. Continuing with original results.")
+                Logger.get_logger().warning(f"AI analysis failed with exit code: {ai_result.returncode}. Continuing with original results.")
                 return
 
-            # Extract JSON from Claude's output (remove markdown code blocks if present)
-            output = ai_result.stdout.strip()
-            # Try to extract JSON from markdown code blocks
-            if "```json" in output:
-                json_start = output.find("```json") + 7
-                json_end = output.find("```", json_start)
-                output = output[json_start:json_end].strip()
-               
-            elif "```" in output:
-                json_start = output.find("```") + 3
-                json_end = output.find("```", json_start)
-                output = output[json_start:json_end].strip()
-                
-
-            # Try to find JSON object/array if there's extra text
-            if not output.startswith('{') and not output.startswith('['):
-                json_start = min(
-                    output.find('{') if '{' in output else len(output),
-                    output.find('[') if '[' in output else len(output)
-                )
-                if json_start < len(output):
-                    output = output[json_start:]
-                   
-
-            # Parse the AI output
-            updated_results = json.loads(output)
-            
+        except Exception as e:
+            Logger.get_logger().warning(f"Unexpected error during AI analysis: {e}. Continuing with original results.")
+            Logger.get_logger().debug(f"Exception details: {str(e)}")
 
-            # Validate the structure
-            if not isinstance(updated_results, dict) or "results" not in updated_results:
-                Logger.get_logger().warning("AI analysis output missing 'results' field. Continuing with original results.")
-                return
+        return
+    
+    def _apply_verification_fields(self):
+        """
+        After codeassure writes verification data, promote is_false_positive
+        and validation_reason to the top level of each finding.
+        """
+        try:
+            with open(self.result_file, 'r') as f:
+                data = json.load(f)
 
-            # Preserve metadata fields that were added by process_result_file
-            metadata_fields = ['repo', 'sha', 'ref', 'run_id', 'repo_url', 'repo_run_url']
-            for field in metadata_fields:
-                if field in current_data:
-                    updated_results[field] = current_data[field]
+            for finding in data.get("results", []):
+                verification = finding.get("verification", {})
+                is_vuln = verification.get("is_security_vulnerability")
+                finding["is_false_positive"] = not bool(is_vuln) if is_vuln is not None else None
+                finding["validation_reason"] = verification.get("reason")
 
-            # Write the updated results
             with open(self.result_file, 'w') as f:
-                json.dump(updated_results, f, indent=2)
-
-           
-            Logger.get_logger().debug("AI analysis completed successfully and results updated.")
+                json.dump(data, f, indent=2)
 
-        except json.JSONDecodeError as e:
-            Logger.get_logger().warning(f"Failed to parse Claude output as JSON: {e}. Continuing with original results.")
-            Logger.get_logger().debug(f"Raw output: {ai_result.stdout[:500] if 'ai_result' in locals() else 'N/A'}...")
-        except FileNotFoundError as e:
-            Logger.get_logger().warning(f"Result file not found during AI analysis: {e}. Continuing without AI analysis.")
+            Logger.get_logger().debug("Verification fields applied to results.")
         except Exception as e:
-            Logger.get_logger().warning(f"Unexpected error during AI analysis: {e}. Continuing with original results.")
-            Logger.get_logger().debug(f"Exception details: {str(e)}")
+            Logger.get_logger().warning(f"Could not apply verification fields: {e}")
 
-        return
-    
     def validate_updated_results(self, results):
         if not isinstance(results, dict) or "results" not in results:
             raise ValueError("AI analysis output is not in the expected format.")
 
-        # for item in results["results"]:
-        #     if "is_false_positive" not in item or "validation_reason" not in item:
-        #         raise ValueError("Missing required fields in AI analysis results.")
 
-    def _build_claude_command(self) -> list[str]:
+    def _build_ai_analysis_command(self) -> list[str]:
         """
-        Build the Claude CLI command (local or Docker).
+        Build the AI Analysis CLI command (local or Docker).
 
         :return: List of command arguments
         """
 
-        system_prompt = """You are a security analysis expert with direct file access.
-
-        Your task: Analyze SAST findings and determine if they are real vulnerabilities or false positives.
-
-        CRITICAL RULES:
-        1. Use bash tool to read files directly
-        2. DO NOT generate Python scripts
-        3. DO NOT output file contents as-is
-        4. Output ONLY the MODIFIED JSON object with your analysis
-
-        For each finding:
-        1. Read the source code file at the specified path
-        2. Examine the code at the specified line numbers
-        3. Analyze: input validation, framework protections, code context, exploitability
-        4. Add two fields: "is_false_positive" (boolean) and "validation_reason" (string)
-
-        Output format: Pure JSON object with all original fields + your two new analysis fields in each finding of results."""
-
-        user_prompt = """Task: Analyze security findings in /workspace/results.json
-
-        Steps to execute:
-        1. Read /workspace/results.json using bash
-        2. For EACH item in the "results" array:
-        - Read the source file specified in "path" field
-        - Examine code at the "line" number
-        - Determine if it's a false positive based on actual code review
-        - Add "is_false_positive": true/false
-        - Add "validation_reason": "brief explanation based on code analysis"
-
-        3. Output the COMPLETE JSON structure with:
-        - ALL original fields preserved exactly as they were
-        - The two new analysis fields added to each result
-        - Proper JSON formatting
-
-        IMPORTANT: 
-        - Output ONLY the final JSON object, nothing else
-        - No markdown code blocks (no ```)
-        - No explanations or commentary
-        - Start output with { and end with }
-        - The output should be parseable JSON"""
-
-
         if not self.container_mode:
-            cmd = ["claude", "--system-prompt", system_prompt, user_prompt]
+            cmd = [
+                ToolManager.get_path("codeassure"),
+                "--codebase", os.getcwd(),
+                "--findings", self.result_file,
+                "--output", self.result_file,
+            ]
+            if self.codeassure_config:
+                cmd.extend(["--config", self.codeassure_config])
+            if self.aiscan_severity:
+                cmd.extend(["--severity", ",".join(self.aiscan_severity)])
+
         else:
             cmd = [
                 "docker", "run", "--rm",
-                "-e", f"ANTHROPIC_API_KEY={self.anthropic_api_key}",
-                "-e", "CLAUDE_CODE_MAX_OUTPUT_TOKENS=200000",
                 "-v", f"{os.getcwd()}:/workspace",
-                self.claude_image,
-                "claude", "--system-prompt", system_prompt, user_prompt
             ]
+            # if self.codeassure_config:
+            #     config_path = os.path.abspath(self.codeassure_config)
+            #     cmd.extend(["-v", f"{config_path}:/workspace/codeassure.json"])
+
+            cmd.extend([
+                self.codeassure_image,
+                "--codebase", "/workspace",
+                "--findings", "/workspace/results.json",
+                "--output", "/workspace/results.json",
+            ])
+            if self.codeassure_config:
+                cmd.extend(["--config", "/workspace/codeassure.json"])
+            if self.aiscan_severity:
+                cmd.extend(["--severity", ",".join(self.aiscan_severity)])
+
 
         return cmd    
 

aspm_cli/scanners/sast_scanner.py

@@ -24,13 +24,18 @@ def add_arguments(self, parser: argparse.ArgumentParser):
             default="INFO,WARNING,LOW,MEDIUM,HIGH,CRITICAL",
             help="Comma-separated list of severities to check. If any match, the scan will fail. Defaults to all severities."
         )
+        parser.add_argument(
+            "--aiscan-severity",
+            default=None,
+            help="Comma-separated list of severities to check for AI analysis. If any match, AI analysis will run on those findings."
+        )
         parser.add_argument("--repo-url", default=GitInfo.get_repo_url(), help="Git repository URL")
         parser.add_argument("--commit-ref", default=GitInfo.get_commit_ref(), help="Commit reference for scanning")
         parser.add_argument("--commit-sha", default=GitInfo.get_commit_sha(), help="Commit SHA for scanning")
         parser.add_argument("--pipeline-id", help="Pipeline ID for scanning")
         parser.add_argument("--job-url", help="Job URL for scanning")
         parser.add_argument("--ai-analysis", action="store_true", help="Enable AI analysis of results")
-        parser.add_argument("--anthropic-api-key", help="Anthropic API key for AI analysis")
+        parser.add_argument("--codeassure-config", help="Path to codeassure.json config file for AI analysis")
 
     def validate_config(self, args: argparse.Namespace, validator: ConfigValidator):
         validator.validate_sast_scan(
@@ -43,12 +48,13 @@ def run_scan(self, args: argparse.Namespace) -> tuple[int, str]:
             command=args.command,
             container_mode=args.container_mode,
             severity=args.severity,
+            aiscan_severity=args.aiscan_severity if args.ai_analysis else [],
             repo_url=args.repo_url,
             commit_ref=args.commit_ref,
             commit_sha=args.commit_sha,
             pipeline_id=args.pipeline_id,
             job_url=args.job_url,
-            anthropic_api_key=getattr(args, 'anthropic_api_key', None),
-            ai_analysis=args.ai_analysis
+            ai_analysis=args.ai_analysis,
+            codeassure_config=getattr(args, 'codeassure_config', None)
         )
         return scanner.run()

aspm_cli/tool/download.py

@@ -19,7 +19,8 @@ class ToolDownloader:
             "secret": "https://github.com/accuknox/aspm-scanner-cli/releases/download/v0.10.1/secret.tar.gz",
             "sq-sast": "https://github.com/accuknox/aspm-scanner-cli/releases/download/v0.10.1/sq-sast.tar.gz",
             "sast": "https://github.com/accuknox/aspm-scanner-cli/releases/download/v0.10.1/sast.tar.gz",
-            "dast": "https://github.com/accuknox/aspm-scanner-cli/releases/download/v0.10.1/dast.tar.gz"
+            "dast": "https://github.com/accuknox/aspm-scanner-cli/releases/download/v0.10.1/dast.tar.gz",
+            "codeassure": "https://github.com/accuknox/aspm-scanner-cli/releases/download/v0.14.2/codeassure.tar.gz"
         },
     }
 

aspm_cli/tool/manager.py

@@ -38,6 +38,7 @@ class ToolManager:
         "dast-java": Path("dast") / "java" / "bin",
         "sast": Path("sast") / "sast",
         "sast-rules": Path("sast") / "rules",
+        "codeassure": Path("codeassure") / "codeassure",
     }
 
 

aspm_cli/utils/validation.py

@@ -1,7 +1,7 @@
 from pydantic import BaseModel, Field, field_validator, model_validator
 from typing import Literal, Optional
 
-ALLOWED_TOOL_TYPES = ["iac", "sast", "secret", "container", "dast", "sq-sast"]
+ALLOWED_TOOL_TYPES = ["iac", "sast", "secret", "container", "dast", "sq-sast", "codeassure"]
 
 class ToolDownloadConfig(BaseModel):
     tooltype: Optional[str] = Field(default=None)

utils/prepare-aspm-scanners.sh

@@ -132,4 +132,28 @@ tar -czvf "$DAST_TAR" "$DAST_FOLDER"
 rm -rf "$TEMP_DIR" "$DAST_FOLDER"
 echo "✅ OpenJDK ${JDK_VERSION} + ZAP ${ZAP_VERSION} packaged into '$DAST_TAR'"
 
+### CodeAssure -> codeassure.tar.gz
+echo "=== [7/7] Building CodeAssure ==="
+CODEASSURE_REPO="https://github.com/accuknox/codeassure-cli.git"
+CODEASSURE_FOLDER="codeassure"
+CODEASSURE_TAR="codeassure.tar.gz"
+TEMP_CODEASSURE="temp_codeassure_build"
+
+rm -rf "$CODEASSURE_FOLDER" "$TEMP_CODEASSURE"
+git clone --branch main "$CODEASSURE_REPO" "$TEMP_CODEASSURE"
+cd "$TEMP_CODEASSURE"
+
+pip install uv
+uv pip install --system -e ".[build]"
+pyinstaller codeassure.spec --clean
+
+mkdir -p "../$CODEASSURE_FOLDER"
+cp dist/codeassure "../$CODEASSURE_FOLDER/codeassure"
+cd ..
+
+cp -r "$CODEASSURE_FOLDER" "$PACKAGE_BIN_DIR"
+tar -czvf "$CODEASSURE_TAR" "$CODEASSURE_FOLDER"
+rm -rf "$TEMP_CODEASSURE" "$CODEASSURE_FOLDER"
+echo "✅ CodeAssure binary packaged as $CODEASSURE_TAR"
+
 echo "🎉 All tools downloaded and prepared successfully."