""" Web Sentinel Source Code Security Scanner Main orchestrator for static application security testing (SAST). Integrates with the license system and supports multiple programming languages. """ from __future__ import annotations import logging import mimetypes import os import re from dataclasses import dataclass, field from pathlib import Path from typing import Dict, Iterable, List, Optional, Set, Union from ...localization import t from ...localization.helpers import create_translated_finding from ...model import Finding, ScanRequest LOGGER = logging.getLogger("web_sentinel.checks.source_code") # Supported file extensions by language LANGUAGE_EXTENSIONS = { # Backend Languages (Advanced Analysis) 'php': {'.php', '.phtml', '.php3', '.php4', '.php5', '.phar'}, 'javascript': {'.js', '.jsx', '.ts', '.tsx', '.vue', '.mjs', '.svelte'}, 'python': {'.py', '.pyw', '.wsgi', '.pyx'}, 'java': {'.java', '.jsp', '.jspx'}, 'csharp': {'.cs', '.aspx', '.ascx', '.ashx', '.asmx'}, 'go': {'.go'}, 'ruby': {'.rb', '.erb', '.rake', '.gemspec'}, 'rust': {'.rs'}, 'cpp': {'.c', '.cpp', '.cxx', '.cc', '.h', '.hpp', '.hxx'}, 'kotlin': {'.kt', '.kts'}, 'scala': {'.scala', '.sc'}, 'swift': {'.swift'}, 'objc': {'.m', '.mm', '.h'}, 'dart': {'.dart'}, # DevOps & Infrastructure 'dockerfile': {'dockerfile', '.dockerfile'}, 'shell': {'.sh', '.bash', '.zsh', '.fish', '.ksh'}, 'powershell': {'.ps1', '.psm1', '.psd1'}, 'terraform': {'.tf', '.tfvars', '.hcl'}, # Data & Config Languages 'sql': {'.sql', '.mysql', '.pgsql', '.sqlite'}, 'html': {'.html', '.htm', '.xhtml'}, 'css': {'.css', '.scss', '.sass', '.less', '.styl'}, 'xml': {'.xml', '.config', '.xsd', '.xsl', '.xaml'}, 'json': {'.json', '.jsonc'}, 'yaml': {'.yml', '.yaml'}, } # Common exclusion patterns DEFAULT_EXCLUSIONS = { # Dependencies 'vendor/', 'node_modules/', '.venv/', 'venv/', '__pycache__/', '.git/', '.svn/', '.hg/', '.bzr/', # Build outputs 'build/', 'dist/', 'target/', 'bin/', 'obj/', # Minified files '*.min.js', '*.min.css', '*.bundle.js', # IDE files '.vscode/', '.idea/', '*.iml', '.project', '.classpath', # OS files '.DS_Store', 'Thumbs.db', 'desktop.ini', # Logs and temporary files '*.log', '*.tmp', '*.temp', '*.bak', } @dataclass class SourceScanConfig: """Configuration for source code scanning.""" source_path: Optional[Path] = None source_files: List[Path] = field(default_factory=list) languages: Set[str] = field(default_factory=set) exclude_patterns: Set[str] = field(default_factory=lambda: DEFAULT_EXCLUSIONS.copy()) min_severity: str = "medium" max_files: int = -1 # -1 for unlimited max_size_mb: int = -1 # -1 for unlimited advanced_rules: bool = False recursive: bool = True detailed_report: bool = False def __post_init__(self): if not self.languages: self.languages = {"auto"} # Auto-detect class SourceCodeScanner: """Main scanner for static application security testing.""" def __init__(self, config: SourceScanConfig): self.config = config self._total_size = 0 self._file_count = 0 self._root_path = self._determine_root() def _determine_root(self) -> Optional[Path]: """Determine the root path used for relative exclusions.""" if self.config.source_path: if self.config.source_path.is_dir(): return self.config.source_path return self.config.source_path.parent if self.config.source_files: parents = [path.parent for path in self.config.source_files if path.exists()] if parents: try: common = Path(os.path.commonpath([str(parent) for parent in parents])) return common except ValueError: return parents[0] return None def _categorize_file(self, file_path: Path) -> str: """ Catégorise un fichier : 'production', 'test', 'dependency' Args: file_path: Chemin du fichier à catégoriser Returns: Catégorie du fichier ('production', 'test', 'dependency') """ file_str = str(file_path).lower() file_name = file_path.name.lower() # 1. Dépendances (priorité haute) dependency_patterns = [ 'node_modules/', 'vendor/', 'packages/', 'libs/', 'lib/', 'third_party/', 'external/', 'dependencies/', '.venv/', 'venv/', '__pycache__/', 'site-packages/', 'dist/', 'build/', 'target/', 'bin/', 'obj/', '.min.', '.bundle.', 'chunk-', 'webpack', 'rollup', # Headers système C/C++ '.h"', 'include/', 'usr/include/', 'windows kits/', ] for pattern in dependency_patterns: if pattern in file_str: return 'dependency' # Fichiers headers système (OpenSSL, zlib, etc.) if file_path.suffix.lower() in {'.h', '.hpp'} and 'openssl' in file_str or 'zlib' in file_name: return 'dependency' # 2. Fichiers de test (priorité moyenne) test_patterns = [ # Dossiers de test '/test/', '/tests/', '/__test__/', '/__tests__/', '/spec/', '/specs/', '\\test\\', '\\tests\\', # Noms de fichiers 'test_', '_test.', '.test.', '.spec.', 'spec_', '_spec.', 'conftest.', 'pytest.', # Frameworks de test 'jest.config', 'karma.conf', 'mocha.opts', # Mocks et fixtures 'mock_', '_mock.', 'stub_', '_stub.', 'fixture_', '_fixture.', 'factory_', ] for pattern in test_patterns: if pattern in file_str or pattern in file_name: return 'test' # 3. Production (par défaut) return 'production' def scan(self) -> List[Finding]: """Execute the source code security scan.""" findings: List[Finding] = [] if not self.config.source_path and not self.config.source_files: findings.append(create_translated_finding( check="source-code", i18n_key="sast.missing_target", severity="info" )) return findings # Validate source path if self.config.source_path and not self.config.source_path.exists(): findings.append(create_translated_finding( check="source-code", i18n_key="sast.path_not_found", severity="critical", i18n_params={"path": str(self.config.source_path)}, evidence=f"Path: {self.config.source_path}" )) return findings # Get files to scan files_to_scan = self._discover_files() if not files_to_scan: target_info = ( str(self.config.source_path) if self.config.source_path else ", ".join(str(path) for path in self.config.source_files[:5]) ) findings.append(create_translated_finding( check="source-code", i18n_key="sast.no_files_found", severity="info", i18n_params={ "languages": ', '.join(self.config.languages), "path": target_info }, evidence=f"Languages: {self.config.languages}, Target: {target_info}" )) return findings # Check file limits if self.config.max_files > 0 and len(files_to_scan) > self.config.max_files: findings.append(create_translated_finding( check="source-code", i18n_key="sast.file_limit_exceeded", severity="medium", i18n_params={ "found": len(files_to_scan), "limit": self.config.max_files }, evidence=f"Files found: {len(files_to_scan)}, Limit: {self.config.max_files}" )) files_to_scan = files_to_scan[:self.config.max_files] self._file_count = len(files_to_scan) # Scan each file for file_path in files_to_scan: try: findings.extend(self._scan_file(file_path)) except Exception as exc: LOGGER.warning(f"Failed to scan {file_path}: {exc}") findings.append(create_translated_finding( check="source-code", i18n_key="sast.scan_error", severity="low", i18n_params={"file": str(file_path), "error": str(exc)}, evidence=f"Fichier: {file_path}, Erreur: {exc}" )) return findings def _discover_files(self) -> List[Path]: """Discover source files to scan.""" files: List[Path] = [] def add_candidate(file_path: Path) -> None: if not self._should_scan_file(file_path): return if self.config.max_size_mb > 0: file_size_mb = file_path.stat().st_size / (1024 * 1024) if self._total_size + file_size_mb > self.config.max_size_mb: LOGGER.info(t("messages.size_limit_skipping").format(file=file_path)) return self._total_size += file_size_mb files.append(file_path) if self.config.source_files: for file_path in self.config.source_files: path_obj = Path(file_path) if path_obj.exists() and path_obj.is_file(): add_candidate(path_obj) self._file_count = len(files) return files base_path = self.config.source_path if not base_path: return files if base_path.is_file(): add_candidate(base_path) self._file_count = len(files) return files iterator = base_path.rglob("*") if self.config.recursive else base_path.glob("*") for file_path in iterator: try: if file_path.is_file(): add_candidate(file_path) except (OSError, PermissionError) as e: # Ignorer les erreurs d'accès (symlinks cassés, permissions, etc.) LOGGER.debug(f"Skipping {file_path}: {e}") continue self._file_count = len(files) return files def _should_scan_file(self, file_path: Path) -> bool: """Determine if a file should be scanned.""" # Check exclusion patterns relative_path = self._relative_path(file_path) for pattern in self.config.exclude_patterns: if self._matches_pattern(relative_path, pattern): return False # Exclure automatiquement les fichiers compilés/minifiés pour éviter les faux positifs compiled_patterns = [ 'chunk-', 'bundle-', '.min.', 'vendor/', 'node_modules/', '.map', 'dist/', 'build/', '.compiled.', '.minified.', 'webpack', 'rollup', 'parcel' ] # Exclure UNIQUEMENT les fichiers demo et exemples (garder les tests pour les catégoriser) demo_patterns = [ 'demo_', 'demo/', 'demos/', 'example_', 'examples/', 'sample_', 'samples/', 'config_generator.py', ] file_str = str(file_path).lower() file_name = file_path.name.lower() # Vérifier les patterns de fichiers compilés if any(pattern in file_str for pattern in compiled_patterns): return False # Vérifier les patterns de demo/exemples uniquement if any(pattern in file_str or pattern in file_name for pattern in demo_patterns): return False # Check file extension if 'auto' in self.config.languages: return self._is_source_file(file_path) # Check specific languages file_ext = file_path.suffix.lower() for lang in self.config.languages: if file_ext in LANGUAGE_EXTENSIONS.get(lang, set()): return True return False def _relative_path(self, file_path: Path) -> str: """Return a normalized relative path for exclusion matching.""" candidate = str(file_path) if self._root_path: try: candidate = str(file_path.relative_to(self._root_path)) except ValueError: pass return candidate.replace("\\", "/") def _matches_pattern(self, path: str, pattern: str) -> bool: """Check if path matches exclusion pattern.""" if pattern.endswith('/'): return pattern in path + '/' if '*' in pattern: import fnmatch return fnmatch.fnmatch(path, pattern) return pattern in path def _is_source_file(self, file_path: Path) -> bool: """Check if file is a source code file.""" ext = file_path.suffix.lower() for extensions in LANGUAGE_EXTENSIONS.values(): if ext in extensions: return True return False def _detect_language(self, file_path: Path) -> Optional[str]: """Detect programming language from file extension.""" ext = file_path.suffix.lower() for lang, extensions in LANGUAGE_EXTENSIONS.items(): if ext in extensions: return lang return None def _scan_file(self, file_path: Path) -> List[Finding]: """Scan a single file for security vulnerabilities.""" findings: List[Finding] = [] language = self._detect_language(file_path) if not language: return findings try: content = file_path.read_text(encoding='utf-8', errors='ignore') except Exception as exc: LOGGER.warning(f"Could not read {file_path}: {exc}") return findings # Déterminer la catégorie du fichier file_category = self._categorize_file(file_path) # Language-specific scanning if language == 'php': findings.extend(self._scan_php(file_path, content)) elif language == 'javascript': findings.extend(self._scan_javascript(file_path, content)) elif language == 'python': findings.extend(self._scan_python(file_path, content)) elif language == 'java': findings.extend(self._scan_java(file_path, content)) elif language == 'csharp': findings.extend(self._scan_csharp(file_path, content)) elif language == 'go': findings.extend(self._scan_go(file_path, content)) elif language == 'ruby': findings.extend(self._scan_ruby(file_path, content)) elif language == 'rust': findings.extend(self._scan_rust(file_path, content)) elif language == 'cpp': findings.extend(self._scan_cpp(file_path, content)) elif language == 'kotlin': findings.extend(self._scan_kotlin(file_path, content)) elif language == 'scala': findings.extend(self._scan_scala(file_path, content)) elif language == 'swift': findings.extend(self._scan_swift(file_path, content)) elif language == 'objc': findings.extend(self._scan_objc(file_path, content)) elif language == 'dart': findings.extend(self._scan_dart(file_path, content)) elif language == 'dockerfile': findings.extend(self._scan_dockerfile(file_path, content)) elif language == 'shell': findings.extend(self._scan_shell(file_path, content)) elif language == 'powershell': findings.extend(self._scan_powershell(file_path, content)) elif language == 'terraform': findings.extend(self._scan_terraform(file_path, content)) elif language == 'sql': findings.extend(self._scan_sql(file_path, content)) # Generic patterns applicable to all languages findings.extend(self._scan_generic_patterns(file_path, content)) # Ajouter la catégorie et ajuster la sévérité findings = self._apply_file_category(findings, file_path, file_category) return findings def _apply_file_category(self, findings: List[Finding], file_path: Path, category: str) -> List[Finding]: """ Applique la catégorie aux findings et ajuste la sévérité. Args: findings: Liste des vulnérabilités détectées file_path: Chemin du fichier scanné category: Catégorie ('production', 'test', 'dependency') Returns: Liste des findings avec catégorie et sévérité ajustées """ categorized_findings = [] for finding in findings: # Mapper les sévérités severity_map = { 'critical': 3, 'high': 2, 'medium': 1, 'low': 0, 'info': -1 } reverse_map = {3: 'critical', 2: 'high', 1: 'medium', 0: 'low', -1: 'info'} # Récupérer la sévérité actuelle current_severity_level = severity_map.get(finding.severity, 1) new_severity_level = current_severity_level # Ajuster selon la catégorie if category == 'test': # Réduire de 2 niveaux pour les tests (sauf si déjà info) new_severity_level = max(current_severity_level - 2, -1) category_label = "[TEST]" category_color = "🧪" elif category == 'dependency': # Réduire de 1 niveau pour les dépendances new_severity_level = max(current_severity_level - 1, -1) category_label = "[DEP]" category_color = "📦" else: # production category_label = "[PROD]" category_color = "🔴" new_severity = reverse_map.get(new_severity_level, 'low') # Mettre à jour le titre avec la catégorie new_title = f"{category_color} {category_label} {finding.title}" # Créer un nouveau finding avec les valeurs ajustées categorized_finding = Finding( check=finding.check, title=new_title, severity=new_severity, description=finding.description, remediation=finding.remediation, impact=finding.impact, evidence=finding.evidence ) categorized_findings.append(categorized_finding) return categorized_findings def _is_real_eval_usage(self, line: str, match: str) -> bool: """Vérifier si c'est un vrai usage d'eval() et non un faux positif.""" line_lower = line.lower() # Exclusions pour éviter les faux positifs false_positive_patterns = [ 'test', 'spec', 'mock', 'stub', 'example', # Fichiers de test 'retrieval', 'evaluation', 'medieval', # Mots contenant 'eval' 'development', 'developer', # Mots de dev 'validation', 'valideval', # Validation 'levels', 'leveling', # Gaming terms ] # Si la ligne contient des patterns de faux positifs, ignorer if any(pattern in line_lower for pattern in false_positive_patterns): return False # Si c'est dans un commentaire if '//' in line or '/*' in line or '*/' in line: return False # Si c'est juste du texte ou des noms de fonction/variable if 'function ' in line_lower and 'eval' in line_lower and 'eval(' not in line_lower: return False return True def _is_real_crypto_usage(self, line: str, algorithm: str) -> bool: """Vérifier si c'est un vrai usage de crypto faible et non un faux positif.""" line_lower = line.lower() algorithm_lower = algorithm.lower() # Exclusions pour éviter les faux positifs sur les méthodes natives JavaScript false_positive_patterns = [ '.includes', '.filter', '.split', '.map', '.reduce', # Méthodes natives JS '.test', '.match', '.search', '.replace', # Méthodes String JS 'description', 'includes', 'excluded', # Mots génériques 'modal', 'model', 'details', # UI/données 'filtered', 'includes', # États de données ] # Si la ligne contient des patterns de faux positifs, ignorer if any(pattern in line_lower for pattern in false_positive_patterns): return False # Si c'est dans un commentaire if '//' in line or '/*' in line or '*/' in line or '#' in line: return False # Vérifier si c'est vraiment un appel de fonction crypto crypto_function_patterns = [ rf'\b{algorithm_lower}\s*\(', # md5(), sha1(), etc. rf'create.*{algorithm_lower}\s*\(', # createMd5(), etc. rf'{algorithm_lower}\.digest', # md5.digest(), etc. rf'CryptoJS\.{algorithm_lower.upper()}', # CryptoJS.MD5 ] return any(re.search(pattern, line_lower) for pattern in crypto_function_patterns) def _is_comment_or_example(self, line: str) -> bool: """Vérifier si une ligne est un commentaire, documentation ou exemple.""" line_stripped = line.strip().lower() # Commentaires évidents comment_patterns = ['#', '//', '/*', '*/', '"""', "'''", '--', ''] if any(line_stripped.startswith(pattern) or pattern in line_stripped for pattern in comment_patterns): return True # Patterns d'exemples et tests example_patterns = [ 'example', 'exemple', 'test', 'demo', 'sample', 'placeholder', 'fake', 'dummy', 'mock', 'stub', 'development', 'dev', 'clé exposée', 'exposed', 'hardcoded', 'supersecret', 'write_text', '.write(', 'f"', 'f\'', # Génération de code 'evidence=', 'code:', 'line:', 'file:', # Évidence/logs ] if any(pattern in line_stripped for pattern in example_patterns): return True return False def _scan_php(self, file_path: Path, content: str) -> List[Finding]: """Scan PHP file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') # PHP-specific vulnerability patterns php_patterns = { 'sql_injection': [ r'mysql_query\s*\(\s*[\'"]?.*\$.*[\'"]?\s*\)', r'mysqli_query\s*\(\s*\$\w+\s*,\s*[\'"]?.*\$.*[\'"]?\s*\)', r'\$\w+->query\s*\(\s*[\'"]?.*\$.*[\'"]?\s*\)', r'[\'"]SELECT.*\$.*[\'"]', r'[\'"]INSERT.*\$.*[\'"]', r'[\'"]UPDATE.*\$.*[\'"]', r'[\'"]DELETE.*\$.*[\'"]', ], 'xss': [ r'echo\s+\$_GET\[', r'echo\s+\$_POST\[', r'print\s+\$_REQUEST\[', r'echo.*\$_.*\[.*\]', r'<\?=\s*\$_', ], 'file_inclusion': [ r'include\s*\(\s*\$_', r'require\s*\(\s*\$_', r'include_once\s*\(\s*\$_', r'require_once\s*\(\s*\$_', ], 'command_injection': [ r'exec\s*\(\s*[\'"]?.*\$.*[\'"]?\s*\)', r'shell_exec\s*\(\s*[\'"]?.*\$.*[\'"]?\s*\)', r'system\s*\(\s*[\'"]?.*\$.*[\'"]?\s*\)', r'passthru\s*\(\s*[\'"]?.*\$.*[\'"]?\s*\)', ], } for line_num, line in enumerate(lines, 1): # Ignorer les commentaires et exemples if self._is_comment_or_example(line): continue # SQL Injection for pattern in php_patterns['sql_injection']: if re.search(pattern, line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code-php", i18n_key="sast.php.sql_injection", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # XSS for pattern in php_patterns['xss']: if re.search(pattern, line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code-php", i18n_key="sast.php.xss", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # File Inclusion for pattern in php_patterns['file_inclusion']: if re.search(pattern, line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code-php", i18n_key="sast.php.file_inclusion", severity="critical", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # Command Injection for pattern in php_patterns['command_injection']: if re.search(pattern, line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code-php", i18n_key="sast.php.command_injection", severity="critical", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_javascript(self, file_path: Path, content: str) -> List[Finding]: """Scan JavaScript file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') # Patterns plus précis pour éviter les faux positifs js_patterns = { # Détection plus précise d'eval - éviter les faux positifs comme "testFailRetrieval" 'eval': [ r'\beval\s*\(', # eval() réel uniquement r'\bnew\s+Function\s*\(', # new Function() constructor r'setTimeout\s*\(\s*[\'"][^\'\"]*[\'\"]\s*,', # setTimeout avec string (dangereuse) ], # DOM XSS seulement avec concaténation dangereuse 'dom_xss': [ r'innerHTML\s*=\s*[^\'\"]*\+', # innerHTML avec concaténation r'document\.write\s*\([^)]*\+[^)]*\)', # document.write avec concaténation ], 'prototype_pollution': [r'__proto__', r'constructor\.prototype'], } for line_num, line in enumerate(lines, 1): line_stripped = line.strip() # Ignorer les lignes de commentaires if line_stripped.startswith('//') or line_stripped.startswith('/*'): continue # Dangerous eval usage - avec vérification contextuelle for pattern in js_patterns['eval']: match = re.search(pattern, line, re.IGNORECASE) if match: # Vérification supplémentaire pour éviter les faux positifs if self._is_real_eval_usage(line, match.group()): findings.append(create_translated_finding( check="source-code-js", i18n_key="sast.js.eval_usage", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # DOM XSS for pattern in js_patterns['dom_xss']: if re.search(pattern, line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code-js", i18n_key="sast.js.dom_xss", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_python(self, file_path: Path, content: str) -> List[Finding]: """Scan Python file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') python_patterns = { 'eval': [r'eval\s*\(', r'exec\s*\('], 'sql_injection': [ r'\.execute\s*\([\'"].*%.*[\'"]', # .execute("... %s ...") r'\.execute\s*\([\'"].*\+.*[\'"]', # .execute("... + ...") r'["\'].*SELECT.*["\'].*\+', # query = "SELECT ..." + var r'["\'].*INSERT.*["\'].*\+', # query = "INSERT ..." + var r'["\'].*UPDATE.*["\'].*\+', # query = "UPDATE ..." + var r'["\'].*DELETE.*["\'].*\+', # query = "DELETE ..." + var ], 'command_injection': [r'os\.system\s*\(', r'subprocess\.call\s*\(.*shell=True'], 'pickle': [r'pickle\.loads?\s*\(', r'cPickle\.loads?\s*\('], } for line_num, line in enumerate(lines, 1): # Ignorer les commentaires et exemples if self._is_comment_or_example(line): continue # Eval/exec usage avec validation contextuelle for pattern in python_patterns['eval']: match = re.search(pattern, line, re.IGNORECASE) if match: if not self._is_real_eval_usage(line, match.group()): continue findings.append(create_translated_finding( check="source-code-python", i18n_key="sast.python.eval_usage", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # SQL Injection for pattern in python_patterns['sql_injection']: if re.search(pattern, line): findings.append(create_translated_finding( check="source-code-python", i18n_key="sast.python.sql_injection", severity="critical", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # Command Injection for pattern in python_patterns['command_injection']: if re.search(pattern, line): findings.append(create_translated_finding( check="source-code-python", i18n_key="sast.python.command_injection", severity="critical", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # Pickle deserialization for pattern in python_patterns['pickle']: if re.search(pattern, line): findings.append(create_translated_finding( check="source-code-python", i18n_key="sast.python.pickle_usage", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # Hardcoded credentials (simple password detection) if re.search(r'(password|secret|api_key)\s*=\s*["\'][^"\']{8,}["\']', line, re.IGNORECASE): if 'todo' not in line.lower() and 'example' not in line.lower(): findings.append(create_translated_finding( check="source-code-python", i18n_key="sast.python.hardcoded_credential", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_java(self, file_path: Path, content: str) -> List[Finding]: """Scan Java file for security vulnerabilities.""" findings: List[Finding] = [] # TODO: Implement Java-specific patterns return findings def _scan_csharp(self, file_path: Path, content: str) -> List[Finding]: """Scan C# file for security vulnerabilities.""" findings: List[Finding] = [] # TODO: Implement C#-specific patterns return findings def _scan_go(self, file_path: Path, content: str) -> List[Finding]: """Scan Go file for security vulnerabilities.""" findings: List[Finding] = [] # TODO: Implement Go-specific patterns return findings def _scan_sql(self, file_path: Path, content: str) -> List[Finding]: """Scan SQL file for security issues.""" findings: List[Finding] = [] # TODO: Implement SQL-specific patterns return findings def _scan_ruby(self, file_path: Path, content: str) -> List[Finding]: """Scan Ruby file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue line_lower = line.lower().strip() # Ruby-specific vulnerabilities # 1. Command Injection if re.search(r'system\s*\(\s*["\'].*#\{', line) or re.search(r'`.*#\{', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.ruby.command_injection", severity="critical", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. SQL Injection in Rails if re.search(r'\.where\s*\(\s*["\'][^"\']*#\{', line) or re.search(r'\.find_by_sql\s*\(\s*["\'][^"\']*#\{', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.ruby.sql_injection", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 3. Unsafe eval if re.search(r'\beval\s*\(', line) and not self._is_comment_or_example(line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.ruby.eval_usage", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_rust(self, file_path: Path, content: str) -> List[Finding]: """Scan Rust file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # Rust-specific vulnerabilities # 1. Unsafe blocks if re.search(r'\bunsafe\s*\{', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.rust.unsafe_block", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. Transmute usage if re.search(r'std::mem::transmute|mem::transmute', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.rust.transmute_usage", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 1. Raw pointer dereference if re.search(r'\*\s*\w+\s*as\s*\*', line) or re.search(r'from_raw\s*\(', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.rust.raw_pointer", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_cpp(self, file_path: Path, content: str) -> List[Finding]: """Scan C/C++ file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # C/C++ specific vulnerabilities # 1. Buffer overflow functions dangerous_functions = ['strcpy', 'strcat', 'sprintf', 'gets', 'scanf'] for func in dangerous_functions: if re.search(rf'\b{func}\s*\(', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.cpp.dangerous_function", severity="high", i18n_params={ "func": func, "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. Format string vulnerabilities if re.search(r'printf\s*\(\s*[a-zA-Z_]\w*\s*\)', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.cpp.format_string", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 3. malloc without free (memory leak indication) if re.search(r'\bmalloc\s*\(', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.cpp.memory_allocation", severity="low", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_kotlin(self, file_path: Path, content: str) -> List[Finding]: """Scan Kotlin file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # Kotlin-specific vulnerabilities # 1. SQL Injection in Android if re.search(r'rawQuery\s*\(\s*["\'][^"\']*\$', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.kotlin.sql_injection", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. Intent vulnerabilities if re.search(r'startActivity\s*\(\s*Intent\s*\(\s*["\']', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.kotlin.intent_vulnerability", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_scala(self, file_path: Path, content: str) -> List[Finding]: """Scan Scala file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # Scala-specific vulnerabilities # 1. Deserialization if re.search(r'ObjectInputStream|readObject', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.scala.deserialization", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_swift(self, file_path: Path, content: str) -> List[Finding]: """Scan Swift file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # Swift-specific vulnerabilities # 1. Keychain issues if re.search(r'kSecAttrAccessibleAlways', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.swift.keychain_issue", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. URL loading without validation if re.search(r'NSURLSession.*dataTask.*http://', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.swift.http_connection", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_objc(self, file_path: Path, content: str) -> List[Finding]: """Scan Objective-C file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # Objective-C specific vulnerabilities # 1. Format string vulnerabilities if re.search(r'NSLog\s*\(\s*[a-zA-Z_]\w*\s*\)', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.objc.format_string", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. Weak cryptography if re.search(r'CC_MD5|CC_SHA1', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.objc.weak_crypto", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_dart(self, file_path: Path, content: str) -> List[Finding]: """Scan Dart/Flutter file for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # Dart/Flutter specific vulnerabilities # 1. Insecure HTTP if re.search(r'http://(?!localhost|127\.0\.0\.1)', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.dart.http_connection", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. Unsafe WebView settings if re.search(r'debuggingEnabled.*true', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.dart.webview_debugging", severity="low", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_dockerfile(self, file_path: Path, content: str) -> List[Finding]: """Scan Dockerfile for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue line_upper = line.upper().strip() # Dockerfile security issues # 1. Running as root if line_upper.startswith('USER ROOT') or line_upper == 'USER 0': findings.append(create_translated_finding( check="source-code", i18n_key="sast.dockerfile.root_user", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. Secrets in environment variables if re.search(r'ENV.*(?:PASSWORD|SECRET|KEY|TOKEN).*=', line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code", i18n_key="sast.dockerfile.secrets_in_env", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 3. Privileged mode if re.search(r'--privileged', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.dockerfile.privileged_mode", severity="critical", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_shell(self, file_path: Path, content: str) -> List[Finding]: """Scan shell scripts for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # Shell script vulnerabilities # 1. Command injection via unquoted variables if re.search(r'\$\{?\w+\}?(?!\s*["\'])', line) and re.search(r'(?:rm|mv|cp|chmod|chown)\s', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.shell.command_injection", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. Dangerous eval usage if re.search(r'\beval\s', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.shell.eval_usage", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 3. Hardcoded passwords if re.search(r'(?:password|passwd|pwd)\s*=\s*["\'][^"\']+["\']', line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code", i18n_key="sast.shell.hardcoded_password", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_powershell(self, file_path: Path, content: str) -> List[Finding]: """Scan PowerShell scripts for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # PowerShell vulnerabilities # 1. Invoke-Expression (dangerous eval equivalent) if re.search(r'Invoke-Expression|iex\s', line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code", i18n_key="sast.powershell.invoke_expression", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. Execution policy bypass if re.search(r'-ExecutionPolicy\s+Bypass', line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code", i18n_key="sast.powershell.execution_policy", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 3. Hardcoded credentials if re.search(r'ConvertTo-SecureString.*-AsPlainText', line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code", i18n_key="sast.powershell.hardcoded_credentials", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_terraform(self, file_path: Path, content: str) -> List[Finding]: """Scan Terraform files for security vulnerabilities.""" findings: List[Finding] = [] lines = content.split('\n') for line_num, line in enumerate(lines, 1): if self._is_comment_or_example(line): continue # Terraform security issues # 1. Hardcoded secrets if re.search(r'(?:password|secret|key|token)\s*=\s*"[^"]{8,}"', line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code", i18n_key="sast.terraform.hardcoded_secret", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 2. Public access if re.search(r'cidr_blocks\s*=\s*\["0\.0\.0\.0/0"\]', line): findings.append(create_translated_finding( check="source-code", i18n_key="sast.terraform.overly_permissive", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # 3. Unencrypted storage if re.search(r'encrypted\s*=\s*false', line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code", i18n_key="sast.terraform.unencrypted_storage", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def _scan_generic_patterns(self, file_path: Path, content: str) -> List[Finding]: """Scan for generic security patterns applicable to all languages.""" findings: List[Finding] = [] lines = content.split('\n') # Generic patterns sensitive_patterns = { 'hardcoded_secrets': [ r'password\s*=\s*[\'"][^\'\"]{8,}[\'"]', r'api[_-]?key\s*=\s*[\'"][^\'\"]{16,}[\'"]', r'secret\s*=\s*[\'"][^\'\"]{16,}[\'"]', r'token\s*=\s*[\'"][^\'\"]{20,}[\'"]', ], 'weak_crypto': [ (r'\bmd5\s*\(', 'MD5'), (r'\bsha1\s*\(', 'SHA1'), (r'\bDES\s*\(', 'DES'), (r'\bRC4\s*\(', 'RC4'), (r'CryptoJS\.MD5\b', 'MD5'), (r'CryptoJS\.SHA1\b', 'SHA1'), (r'createHash\s*\(\s*[\'"]md5[\'"]', 'MD5'), (r'createHash\s*\(\s*[\'"]sha1[\'"]', 'SHA1'), ], } for line_num, line in enumerate(lines, 1): # Ignorer les commentaires et exemples if self._is_comment_or_example(line): continue # Hardcoded secrets avec validation contextuelle for pattern in sensitive_patterns['hardcoded_secrets']: if re.search(pattern, line, re.IGNORECASE): findings.append(create_translated_finding( check="source-code-generic", i18n_key="sast.generic.hardcoded_secrets", severity="high", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip() }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) # Weak cryptography avec validation contextuelle for pattern, algorithm in sensitive_patterns['weak_crypto']: if re.search(pattern, line, re.IGNORECASE): # Vérification supplémentaire pour éviter les faux positifs if self._is_real_crypto_usage(line, algorithm): findings.append(create_translated_finding( check="source-code-generic", i18n_key="sast.generic.weak_crypto", severity="medium", i18n_params={ "file": str(file_path), "line": line_num, "code": line.strip(), "algorithm": algorithm }, evidence=f"Fichier: {file_path.name}, Ligne: {line_num}\nCode: {line.strip()}" )) return findings def evaluate_source_code(request: ScanRequest) -> Iterable[Finding]: """ Entry point for source code security scanning. This function integrates with the existing Web Sentinel architecture and is compatible with the license system and CLI/GUI interfaces. """ findings: List[Finding] = [] def _normalize_iterable(value, *, split_tokens: bool = True) -> List[str]: if not value: return [] if isinstance(value, (list, tuple, set, frozenset)): normalized: List[str] = [] for item in value: normalized.extend(_normalize_iterable(item, split_tokens=split_tokens)) return normalized if isinstance(value, str): if split_tokens: parts = [part.strip() for part in re.split(r"[;,]\s*|\s+", value) if part.strip()] return parts return [value.strip()] return [str(value).strip()] raw_source_path = getattr(request, "source_path", None) source_path = None if isinstance(raw_source_path, (str, Path)) and raw_source_path: source_path = Path(raw_source_path).expanduser() raw_source_files = getattr(request, "source_files", ()) normalized_files = _normalize_iterable(raw_source_files, split_tokens=False) source_files = [Path(item).expanduser() for item in normalized_files if item] if source_path is None and not source_files: findings.append( create_translated_finding( check="source-code", i18n_key="sast.not_configured", severity="info", ) ) return findings raw_languages = getattr(request, "source_languages", ()) or getattr(request, "languages", ()) languages_set = {lang.lower() for lang in _normalize_iterable(raw_languages)} or {"auto"} raw_exclude = getattr(request, "source_exclude", ()) or getattr(request, "exclude_patterns", ()) exclude_patterns = set(DEFAULT_EXCLUSIONS) exclude_patterns.update(_normalize_iterable(raw_exclude)) config = SourceScanConfig( source_path=source_path, source_files=source_files, languages=languages_set, exclude_patterns=exclude_patterns, min_severity=getattr(request, "source_min_severity", "medium"), max_files=getattr(request, "source_max_files", -1), max_size_mb=getattr(request, "source_max_size_mb", -1), advanced_rules=getattr(request, "source_advanced_rules", False), recursive=getattr(request, "source_recursive", True), detailed_report=getattr(request, "source_detailed_report", False), ) # Execute scan scanner = SourceCodeScanner(config) findings.extend(scanner.scan()) return findings