""" SonarQube scanner integration for running code analysis scans. """ import os import subprocess import logging import time import requests import shutil import platform from pathlib import Path from typing import Dict, List, Optional from datetime import datetime logger = logging.getLogger(__name__) def _node_bin_directories_for_path() -> List[str]: """ Directories to prepend to PATH so SonarJS can run `node` (cron/minimal env often omits nvm). Override with NODE_BIN_PATH=/path/to/dir/containing/node or explicit path to node binary. """ dirs: List[str] = [] seen = set() def add(p: str) -> None: if p and p not in seen and Path(p).is_dir(): seen.add(p) dirs.append(p) override = os.environ.get("NODE_BIN_PATH", "").strip() if override: op = Path(override).expanduser() if op.is_file() and os.access(op, os.X_OK): add(str(op.parent)) elif op.is_dir(): add(str(op)) if platform.system() == "Windows": return dirs home = Path.home() default_alias = home / ".nvm" / "alias" / "default" if default_alias.is_file(): ver = default_alias.read_text(encoding="utf-8").strip() if ver and ver != "system": bindir = home / ".nvm" / "versions" / "node" / ver / "bin" if bindir.is_dir(): add(str(bindir)) nvm_node = home / ".nvm" / "versions" / "node" if nvm_node.is_dir(): for child in sorted(nvm_node.iterdir(), key=lambda x: x.name, reverse=True): b = child / "bin" if b.is_dir(): add(str(b)) for extra in (Path("/usr/local/bin"), home / ".local" / "bin"): if extra.is_dir(): add(str(extra)) return dirs def _path_with_node_bins(existing_path: str, node_bins: List[str]) -> str: if not node_bins: return existing_path prefix = os.pathsep.join(node_bins) if existing_path: return prefix + os.pathsep + existing_path return prefix class SonarQubeScanner: """Manages SonarQube scanning operations.""" def __init__(self, sonar_url: str, sonar_token: str, scanner_path: str = "sonar-scanner"): """ Initialize SonarQube scanner. Args: sonar_url: SonarQube server URL sonar_token: SonarQube authentication token scanner_path: Invoke this command to run the scanner (e.g. "sonar-scanner" when in PATH). Set via SONAR_SCANNER_PATH or config scanning.sonar_scanner_path. """ self.sonar_url = sonar_url.rstrip("/") self.sonar_token = sonar_token # Normalize: empty or whitespace -> use CLI command "sonar-scanner" if not (scanner_path and scanner_path.strip()): scanner_path = "sonar-scanner" else: scanner_path = scanner_path.strip() # On Linux/macOS, never use a Windows path (e.g. from shared config) if platform.system() != "Windows": if "\\" in scanner_path or scanner_path.endswith((".bat", ".cmd")) or "C:" in scanner_path: logger.info("Ignoring Windows scanner path on Linux; using sonar-scanner.") scanner_path = "sonar-scanner" # Linux/macOS: resolve bare command to absolute path so subprocess works when PATH is minimal (cron, stripped env). if platform.system() != "Windows": has_sep = os.sep in scanner_path or (os.altsep and os.altsep in scanner_path) if not os.path.isabs(scanner_path) and not has_sep: found = shutil.which(scanner_path) if found: scanner_path = found logger.info("Resolved sonar-scanner via PATH: %s", scanner_path) else: project_root = Path(__file__).resolve().parent.parent candidates = [ project_root / "sonar-scanner" / "bin" / "sonar-scanner", Path("/opt/sonar-scanner/bin/sonar-scanner"), Path("/usr/local/opt/sonar-scanner/bin/sonar-scanner"), ] for cand in candidates: if cand.is_file(): scanner_path = str(cand) logger.info("Using sonar-scanner at: %s", scanner_path) break # On Windows, check if scanner_path needs .bat extension if platform.system() == 'Windows': # If scanner_path doesn't have an extension and doesn't exist, try .bat if not scanner_path.endswith(('.bat', '.exe', '.cmd')): # Check if the command exists as-is if not shutil.which(scanner_path): # Try with .bat extension bat_path = scanner_path + '.bat' if shutil.which(bat_path): scanner_path = bat_path logger.info(f"Using Windows batch file: {scanner_path}") else: # Try to find sonar-scanner in common locations # Check relative to current script location first script_dir = Path(__file__).parent.parent.absolute() common_paths = [ str(script_dir / "sonar-scanner" / "bin" / "sonar-scanner.bat"), r"C:\Program Files\SonarQube\bin\sonar-scanner.bat", r"C:\sonar-scanner\bin\sonar-scanner.bat", os.path.expanduser(r"~\sonar-scanner\bin\sonar-scanner.bat") ] for path in common_paths: if os.path.exists(path): scanner_path = path logger.info(f"Found sonar-scanner at: {scanner_path}") break self.scanner_path = scanner_path self.is_windows = platform.system() == 'Windows' def _token_for_slug(self, repo_slug: Optional[str]) -> str: """ Token for sonar-scanner only (SONAR_TOKEN). Per-repo SONARQUBE_TOKEN_ is for SonarQube *project analysis* tokens (sqp_...). Server REST calls use self.sonar_token (user/global SONARQUBE_TOKEN) — project tokens cannot manage projects or read all metrics APIs. """ if not repo_slug: return self.sonar_token suffix = repo_slug.replace("-", "_").upper() override = os.environ.get(f"SONARQUBE_TOKEN_{suffix}", "").strip() return override if override else self.sonar_token def run_scan(self, repo_path: Path, sonar_properties_path: Path, max_retries: int = 3, retry_delay: int = 60) -> Dict: """ Run SonarQube scan on a repository with retry logic for concurrent scan errors. Args: repo_path: Path to repository directory sonar_properties_path: Path to sonar-project.properties file max_retries: Maximum number of retries for concurrent scan errors (default: 3) retry_delay: Initial delay in seconds between retries (default: 60, doubles each retry) Returns: Dictionary with scan results including: - success: bool - task_id: str (if available) - analysis_id: str (if available) - error: str (if failed) """ logger.info(f"Starting SonarQube scan for {repo_path.name}") result = None for attempt in range(max_retries + 1): if attempt > 0: wait_time = retry_delay * (2 ** (attempt - 1)) # Exponential backoff logger.info(f"Retry attempt {attempt}/{max_retries} after waiting {wait_time} seconds...") time.sleep(wait_time) result = self._execute_scan(repo_path, sonar_properties_path, repo_path.name) # If successful, return immediately if result.get("success"): return result # Check if it's a concurrent scan error and we have retries left error_msg = result.get("error", "") if ("Another SonarQube analysis is already in progress" in error_msg or "already in progress" in error_msg) and attempt < max_retries: logger.warning(f"Another analysis is in progress. Will retry in {retry_delay * (2 ** attempt)} seconds...") continue # For other errors or no retries left, return the result return result # Should not reach here, but return last result if we do return result if result else {"success": False, "error": "Scan failed after all retries"} def _execute_scan(self, repo_path: Path, sonar_properties_path: Path, repo_slug: str) -> Dict: """ Execute a single SonarQube scan attempt. Args: repo_path: Path to repository directory sonar_properties_path: Path to sonar-project.properties file repo_slug: Directory name / repo slug (for per-repo SONARQUBE_TOKEN_) Returns: Dictionary with scan results """ try: # Ensure .scannerwork directory structure exists scannerwork_dir = repo_path / ".scannerwork" scannerwork_dir.mkdir(parents=True, exist_ok=True) # Ensure .sonartmp subdirectory exists (required for temp files) sonartmp_dir = scannerwork_dir / ".sonartmp" sonartmp_dir.mkdir(parents=True, exist_ok=True) logger.debug(f"Ensured scanner work directories exist: {scannerwork_dir}") # Read sonar-project.properties and update with server URL and token self._update_sonar_properties(sonar_properties_path) # Run sonar-scanner command env = os.environ.copy() env["SONAR_TOKEN"] = self._token_for_slug(repo_slug) node_bins = _node_bin_directories_for_path() if node_bins: env["PATH"] = _path_with_node_bins(env.get("PATH", ""), node_bins) logger.debug("Prepended Node bin dirs to PATH for SonarJS: %s", node_bins[:3]) # Set custom cache directory if needed to avoid permission issues # Use a cache directory in the project folder instead of user home project_cache_dir = Path(__file__).parent.parent / ".sonar_cache" project_cache_dir.mkdir(parents=True, exist_ok=True) env["SONAR_USER_HOME"] = str(project_cache_dir.absolute()) logger.debug(f"Using SonarQube cache directory: {project_cache_dir}") # Change to repository directory original_cwd = os.getcwd() os.chdir(repo_path) try: # Execute sonar-scanner # On Windows with .bat files, use shell=True and pass as string # On Linux/Mac, use shell=False and pass as list if self.is_windows and self.scanner_path.endswith(('.bat', '.cmd')): # Windows batch file - use shell=True result = subprocess.run( self.scanner_path, capture_output=True, text=True, env=env, timeout=1800, # 30 minute timeout shell=True ) else: # Linux/Mac or Windows .exe - use shell=False result = subprocess.run( [self.scanner_path], capture_output=True, text=True, env=env, timeout=1800 # 30 minute timeout ) if result.returncode == 0: logger.info(f"SonarQube scan completed successfully for {repo_path.name}") # Try to extract task ID and analysis ID from output task_id = self._extract_task_id(result.stdout) analysis_id = self._extract_analysis_id(result.stdout) return { "success": True, "task_id": task_id, "analysis_id": analysis_id, "output": result.stdout } else: error_msg = result.stderr or result.stdout logger.error(f"SonarQube scan failed for {repo_path.name}: {error_msg}") # Check for concurrent scan error - another analysis is in progress if "Another SonarQube analysis is already in progress" in error_msg or "already in progress" in error_msg: logger.warning("Another SonarQube analysis is already in progress for this project") logger.info("This usually means a previous scan is still running on the SonarQube server") error_msg += "\n\nAnother analysis is in progress. Will retry automatically." # Check for temp file creation errors if "Failed to create temp file" in error_msg or "NoSuchFileException" in error_msg or ".sonartmp" in error_msg: logger.warning("Detected temp file creation error, ensuring scanner directories exist...") scannerwork_dir = repo_path / ".scannerwork" scannerwork_dir.mkdir(parents=True, exist_ok=True) sonartmp_dir = scannerwork_dir / ".sonartmp" sonartmp_dir.mkdir(parents=True, exist_ok=True) logger.info("Scanner work directories created. Please retry the scan.") error_msg += "\n\nScanner directories created. Please retry the scan." # Check for cache/permission errors and try to fix them if "AccessDeniedException" in error_msg or "Failed to extract archive" in error_msg: logger.warning("Detected cache/permission error, attempting to clear SonarQube cache...") cache_cleared = self._clear_sonar_cache() if cache_cleared: logger.info("Cache cleared successfully. Please retry the scan.") error_msg += "\n\nCache cleared. Please retry the scan." else: logger.warning("Could not clear cache automatically. Please manually clear:") logger.warning(f" {os.path.expanduser('~/.sonar/cache')}") error_msg += "\n\nPlease clear the SonarQube cache directory and retry." return { "success": False, "error": error_msg, "output": result.stdout } finally: os.chdir(original_cwd) except subprocess.TimeoutExpired: logger.error(f"SonarQube scan timed out for {repo_path.name}") return { "success": False, "error": "Scan timed out after 30 minutes" } except FileNotFoundError as e: error_msg = f"SonarQube scanner not found: {self.scanner_path}" logger.error(error_msg) logger.error("Please ensure sonar-scanner is installed and either:") logger.error(" 1. Added to your system PATH, or") logger.error(" 2. Set the 'scanning.sonar_scanner_path' config value to the full path") logger.error(f" Example: C:\\Program Files\\SonarQube\\bin\\sonar-scanner.bat") return { "success": False, "error": error_msg } except Exception as e: error_msg = str(e) logger.error(f"Unexpected error during SonarQube scan: {e}") # Check for cache/permission errors and try to fix them if "AccessDeniedException" in error_msg or "Failed to extract archive" in error_msg: logger.warning("Detected cache/permission error, attempting to clear SonarQube cache...") cache_cleared = self._clear_sonar_cache() if cache_cleared: logger.info("Cache cleared successfully. Please retry the scan.") error_msg += "\n\nCache cleared. Please retry the scan." else: logger.warning("Could not clear cache automatically. Please manually clear:") logger.warning(f" {os.path.expanduser('~/.sonar/cache')}") error_msg += "\n\nPlease clear the SonarQube cache directory and retry." return { "success": False, "error": error_msg } def _clear_sonar_cache(self) -> bool: """ Clear SonarQube cache directory to fix permission/corruption issues. Returns: True if cache was cleared successfully, False otherwise """ try: cache_dir = os.path.expanduser("~/.sonar/cache") if os.path.exists(cache_dir): logger.info(f"Clearing SonarQube cache directory: {cache_dir}") shutil.rmtree(cache_dir) logger.info("SonarQube cache cleared successfully") return True else: logger.debug("SonarQube cache directory does not exist") return True # No cache to clear is fine except PermissionError as e: logger.error(f"Permission denied when trying to clear cache: {e}") logger.error("Please manually clear the cache directory or run with elevated permissions") return False except Exception as e: logger.error(f"Failed to clear SonarQube cache: {e}") return False def _update_sonar_properties(self, properties_path: Path) -> None: """ Update sonar-project.properties with server URL and token. Args: properties_path: Path to sonar-project.properties file """ try: # Read existing properties with open(properties_path, "r") as f: content = f.read() # Never persist credentials in the file; scanner uses SONAR_TOKEN from the environment filtered = [] for line in content.splitlines(keepends=True): s = line.strip() if s.startswith("sonar.token") or s.startswith("sonar.login"): continue filtered.append(line) content = "".join(filtered) # Update or add sonar.host.url if "sonar.host.url" in content: # Replace existing import re content = re.sub( r"sonar\.host\.url\s*=.*", f"sonar.host.url={self.sonar_url}", content ) else: # Add new content += f"\nsonar.host.url={self.sonar_url}\n" # Add token (don't store in file, use environment variable instead) # The scanner will pick up SONAR_TOKEN from environment # Write back with open(properties_path, "w") as f: f.write(content) except Exception as e: logger.warning(f"Failed to update sonar-project.properties: {e}") def _extract_task_id(self, output: str) -> Optional[str]: """ Extract task ID from scanner output. Args: output: Scanner output text Returns: Task ID string or None """ import re match = re.search(r"EXECUTION SUCCESS.*?taskId=([^\s]+)", output, re.DOTALL) if match: return match.group(1) # Alternative pattern match = re.search(r"taskId[=:]\s*([^\s\n]+)", output) if match: return match.group(1) return None def _extract_analysis_id(self, output: str) -> Optional[str]: """ Extract analysis ID from scanner output. Args: output: Scanner output text Returns: Analysis ID string or None """ import re match = re.search(r"analysisId[=:]\s*([^\s\n]+)", output) if match: return match.group(1) return None def get_scan_results(self, project_key: str) -> Optional[Dict]: """ Get scan results from SonarQube API (uses global SONARQUBE_TOKEN, not project analysis tokens). Args: project_key: SonarQube project key Returns: Dictionary with scan metrics and quality gate status """ try: headers = {"Authorization": f"Bearer {self.sonar_token}"} status_url = f"{self.sonar_url}/api/qualitygates/project_status" params = {"projectKey": project_key} response = requests.get(status_url, params=params, headers=headers, timeout=30) response.raise_for_status() status_data = response.json() # Get project metrics metrics_url = f"{self.sonar_url}/api/measures/component" metrics_params = { "component": project_key, "metricKeys": "coverage,bugs,vulnerabilities,code_smells,duplicated_lines_density,ncloc" } metrics_response = requests.get( metrics_url, params=metrics_params, headers=headers, timeout=30 ) metrics_data = metrics_response.json() if metrics_response.status_code == 200 else {} return { "quality_gate": status_data.get("projectStatus", {}).get("status", "UNKNOWN"), "metrics": metrics_data.get("component", {}).get("measures", []), "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"Failed to get scan results from SonarQube API: {e}") return None def get_project_key(self, properties_path: Path) -> Optional[str]: """ Extract project key from sonar-project.properties. Args: properties_path: Path to sonar-project.properties file Returns: Project key string or None """ try: with open(properties_path, "r") as f: for line in f: if line.strip().startswith("sonar.projectKey"): return line.split("=", 1)[1].strip() except Exception as e: logger.warning(f"Failed to read project key: {e}") return None def check_project_exists(self, project_key: str) -> bool: """ Check if a SonarQube project exists (uses global SONARQUBE_TOKEN). Args: project_key: SonarQube project key Returns: True if project exists, False otherwise """ try: url = f"{self.sonar_url}/api/projects/search" params = {"projects": project_key} headers = {"Authorization": f"Bearer {self.sonar_token}"} response = requests.get(url, params=params, headers=headers, timeout=30) response.raise_for_status() data = response.json() components = data.get("components", []) # Check if project exists in the results for component in components: if component.get("key") == project_key: logger.debug(f"Project '{project_key}' exists in SonarQube") return True logger.debug(f"Project '{project_key}' not found in SonarQube") return False except requests.HTTPError as e: if e.response.status_code == 404: return False logger.warning(f"Error checking project existence: {e}") return False except Exception as e: logger.warning(f"Failed to check if project exists: {e}") return False def create_project( self, project_key: str, project_name: str, description: str = "", visibility: str = "private", ) -> bool: """ Create a new project in SonarQube (uses global SONARQUBE_TOKEN). Args: project_key: SonarQube project key (must be unique) project_name: Display name for the project description: Project description (optional) visibility: Project visibility - "private" or "public" (default: private) Returns: True if project was created successfully, False otherwise """ try: url = f"{self.sonar_url}/api/projects/create" headers = {"Authorization": f"Bearer {self.sonar_token}"} data = { "project": project_key, "name": project_name, "visibility": visibility } if description: data["description"] = description response = requests.post(url, data=data, headers=headers, timeout=30) if response.status_code == 200: logger.info(f"Successfully created SonarQube project: {project_key} ({project_name})") return True elif response.status_code == 400: # Project might already exist error_text = response.text if "already exists" in error_text.lower(): logger.info(f"Project '{project_key}' already exists in SonarQube") return True logger.warning(f"Failed to create project '{project_key}': {error_text}") return False else: response.raise_for_status() return False except requests.HTTPError as e: logger.error(f"HTTP error creating SonarQube project '{project_key}': {e}") if hasattr(e.response, 'text'): logger.debug(f"Response: {e.response.text}") return False except Exception as e: logger.error(f"Failed to create SonarQube project '{project_key}': {e}") return False def generate_sonar_properties(self, repo_path: Path, project_key: str, project_name: str) -> Path: """ Generate sonar-project.properties file with default values. Args: repo_path: Path to repository directory project_key: SonarQube project key project_name: Project display name Returns: Path to created sonar-project.properties file """ properties_path = repo_path / "sonar-project.properties" try: # Generate properties content properties_content = f"""# SonarQube Project Properties # Auto-generated by SonarQube Scanner Automation # Project identification sonar.projectKey={project_key} sonar.projectName={project_name} # Source code location sonar.sources=. # SonarQube server sonar.host.url={self.sonar_url} # Encoding sonar.sourceEncoding=UTF-8 """ # Write properties file with open(properties_path, "w", encoding="utf-8") as f: f.write(properties_content) logger.info(f"Generated sonar-project.properties for {repo_path.name}: {properties_path}") return properties_path except Exception as e: logger.error(f"Failed to generate sonar-project.properties: {e}") raise