feat: enhance security scanner with reduced false positives and bette…

…r reporting

.github/workflows/security-scan.yml

@@ -4,23 +4,26 @@ on:
   pull_request:
     types: [opened, synchronize, reopened]
 
+# Add top-level permissions blocksadadsad
 permissions:
   contents: read
   pull-requests: write
+  issues: write
+  checks: write
 
 jobs:
   security-scan:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
-
+    - uses: actions/checkout@v4
+    
     - name: Set up Python
-      uses: actions/setup-python@v4
+      uses: actions/setup-python@v5
       with:
         python-version: '3.10'
 
     - name: Cache pip packages
-      uses: actions/cache@v3
+      uses: actions/cache@v4
       with:
         path: ~/.cache/pip
         key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
@@ -47,6 +50,264 @@ jobs:
 
         logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
+        class AdvancedVulnerabilityScanner:
+            def __init__(self, file_path: str):
+                self.file_path = file_path
+                self.vulnerabilities: List[Dict[str, Any]] = []
+                self.code_lines: List[str] = []
+                self.ast_tree: ast.AST = None
+                self.vulnerability_db = self.load_vulnerability_db()
+
+            def load_vulnerability_db(self):
+                return {
+                    'requests': {'2.25.0': ['CVE-2021-12345']},
+                    'django': {'2.2.0': ['CVE-2021-67890']}
+                }
+
+            def parse_file(self):
+                logging.info(f"Parsing file: {self.file_path}")
+                try:
+                    with open(self.file_path, 'r', encoding='utf-8') as file:
+                        self.code_lines = file.readlines()
+                        self.ast_tree = ast.parse(''.join(self.code_lines))
+                    logging.info(f"File parsed. Total lines: {len(self.code_lines)}")
+                except Exception as e:
+                    logging.error(f"Error parsing file {self.file_path}: {str(e)}")
+                    raise
+
+            def run_bandit(self):
+                try:
+                    b_mgr = bandit_manager.BanditManager(bandit.config.BanditConfig(), agg_type='file')
+                    b_mgr.discover_files([self.file_path])
+                    b_mgr.run_tests()
+                    return b_mgr.get_issue_list()
+                except Exception as e:
+                    logging.error(f"Error running Bandit: {str(e)}")
+                    return []
+
+            def add_vulnerability(self, category: str, description: str, line_number: int, severity: str, confidence: str):
+                self.vulnerabilities.append({
+                    'category': category,
+                    'description': description,
+                    'line_number': line_number,
+                    'severity': severity,
+                    'confidence': confidence,
+                    'code_context': self.code_lines[line_number-1].strip() if line_number > 0 else None
+                })
+                logging.info(f"Critical vulnerability added: {category} at line {line_number}")
+
+            def check_high_risk_sql_injection(self):
+                sql_patterns = [
+                    r'(?i)(?:execute|cursor\.execute)\s*\(.*?f["\'].*?\{.*?\}.*?["\'].*?\)',  # f-string in SQL
+                    r'(?i)(?:execute|cursor\.execute)\s*\(.*?\+.*?\)',  # String concatenation in SQL
+                    r'(?i)(?:execute|cursor\.execute)\s*\(.*?%.*?\%.*?\)'  # %-formatting without params
+                ]
+                for i, line in enumerate(self.code_lines):
+                    for pattern in sql_patterns:
+                        if re.search(pattern, line):
+                            self.add_vulnerability(
+                                'SQL Injection',
+                                f"Critical: SQL injection vulnerability detected",
+                                i+1,
+                                'HIGH',
+                                'HIGH'
+                            )
+
+            def check_hardcoded_secrets(self):
+                secret_patterns = [
+                    r'(?i)(password|secret|api_key|token)\s*=\s*["\'][^"\']+["\'](?!\s*{\s*key\s*[=:]\s*|os\.environ)',
+                    r'(?i)auth_token\s*=\s*["\'][0-9a-zA-Z]+["\']',
+                    r'(?i)api_key\s*=\s*["\'][0-9a-zA-Z]+["\']'
+                ]
+                
+                safe_patterns = [
+                    r'key="[\w_-]+"',  # Streamlit/UI keys
+                    r'os\.environ\.get',  # Environment variables
+                    r'load_dotenv',  # Environment loading
+                    r'SECRET_KEY\s*=\s*config\.',  # Config references
+                    r'test|example|dummy|sample'  # Test code
+                ]
+
+                for i, line in enumerate(self.code_lines):
+                    if any(re.search(safe_pat, line, re.IGNORECASE) for safe_pat in safe_patterns):
+                        continue
+                        
+                    for pattern in secret_patterns:
+                        if match := re.search(pattern, line):
+                            self.add_vulnerability(
+                                'Hardcoded Secret',
+                                f"Critical: Hardcoded secret detected",
+                                i+1,
+                                'HIGH',
+                                'HIGH'
+                            )
+
+            def check_command_injection(self):
+                dangerous_patterns = [
+                    r'subprocess\.(?:call|run|Popen)\s*\(.*?\+.*?\)',  # String concatenation
+                    r'os\.system\s*\(.*?\+.*?\)',  # String concatenation
+                    r'subprocess\.(?:call|run|Popen)\s*\(.*?format.*?\)',  # String formatting
+                    r'subprocess\.(?:call|run|Popen)\s*\(.*?f["\'].*?\{.*?\}.*?["\'].*?\)'  # f-strings
+                ]
+                
+                safe_patterns = [
+                    r'subprocess\.run\(\[[^\]]+\],\s*check=True\)',
+                    r'subprocess\.run\(\[[^\]]+\],\s*shell=False\)'
+                ]
+
+                for i, line in enumerate(self.code_lines):
+                    if any(re.search(safe_pat, line) for safe_pat in safe_patterns):
+                        continue
+
+                    for pattern in dangerous_patterns:
+                        if re.search(pattern, line):
+                            self.add_vulnerability(
+                                'Command Injection',
+                                f"Critical: Command injection vulnerability detected",
+                                i+1,
+                                'HIGH',
+                                'HIGH'
+                            )
+
+            def check_dangerous_deserialization(self):
+                dangerous_patterns = [
+                    (r'pickle\.loads?\(.*?\)', "Unsafe pickle deserialization"),
+                    (r'yaml\.load\((?![^)]*Loader=yaml\.SafeLoader)', "Unsafe YAML loading"),
+                    (r'eval\(.*?\)', "Dangerous eval() usage")
+                ]
+                
+                for i, line in enumerate(self.code_lines):
+                    for pattern, message in dangerous_patterns:
+                        if re.search(pattern, line):
+                            self.add_vulnerability(
+                                'Dangerous Deserialization',
+                                f"Critical: {message}",
+                                i+1,
+                                'HIGH',
+                                'HIGH'
+                            )
+
+            def check_path_traversal(self):
+                dangerous_patterns = [
+                    (r'open\s*\([^)]*[\'"][.][.][\/\\]', "Path traversal vulnerability"),
+                    (r'open\s*\([^)]*\+', "Dynamic file path manipulation"),
+                    (r'os\.path\.join\s*\([^)]*\+', "Dynamic path joining")
+                ]
+                
+                for i, line in enumerate(self.code_lines):
+                    for pattern, message in dangerous_patterns:
+                        if re.search(pattern, line):
+                            self.add_vulnerability(
+                                'Path Traversal',
+                                f"Critical: {message}",
+                                i+1,
+                                'HIGH',
+                                'HIGH'
+                            )
+
+            def check_high_risk_ssrf(self):
+                ssrf_patterns = [
+                    r'requests\.(get|post|put|delete)\s*\([^)]*\+',
+                    r'urllib\.request\.urlopen\s*\([^)]*\+',
+                    r'http\.client\.HTTPConnection\s*\([^)]*\+'
+                ]
+                
+                for i, line in enumerate(self.code_lines):
+                    for pattern in ssrf_patterns:
+                        if re.search(pattern, line):
+                            self.add_vulnerability(
+                                'SSRF',
+                                f"Critical: Server-Side Request Forgery vulnerability",
+                                i+1,
+                                'HIGH',
+                                'HIGH'
+                            )
+
+            def analyze(self):
+                try:
+                    self.parse_file()
+                    self.check_high_risk_sql_injection()
+                    self.check_hardcoded_secrets()
+                    self.check_dangerous_deserialization()
+                    self.check_command_injection()
+                    self.check_path_traversal()
+                    self.check_high_risk_ssrf()
+
+                    bandit_issues = self.run_bandit()
+                    for issue in bandit_issues:
+                        if issue.severity.lower() == 'high':
+                            self.add_vulnerability(
+                                f"Critical Security Issue ({issue.test_id})",
+                                issue.text,
+                                issue.lineno,
+                                'HIGH',
+                                issue.confidence
+                            )
+
+                    logging.info("Security analysis completed successfully")
+                except Exception as e:
+                    logging.error(f"An error occurred during analysis: {str(e)}")
+                    raise
+
+            def generate_report(self):
+                report = f"\n🔒 Security Scan Results for {self.file_path} 🔒\n"
+                report += "=" * 50 + "\n"
+                report += f"Lines of Code Analyzed: {len(self.code_lines)}\n\n"
+                
+                # Filter for high-risk vulnerabilities
+                high_risk_vulns = [v for v in self.vulnerabilities 
+                                if v['severity'] == 'HIGH' and v['confidence'] in ['HIGH', 'MEDIUM']]
+                
+                if high_risk_vulns:
+                    report += f"🚨 Found {len(high_risk_vulns)} Critical Security Issues!\n\n"
+                    for vuln in high_risk_vulns:
+                        report += f"CRITICAL: {vuln['category']}\n"
+                        report += f"Description: {vuln['description']}\n"
+                        report += f"Location: Line {vuln['line_number']}\n"
+                        if vuln.get('code_context'):
+                            report += f"Code: {vuln['code_context']}\n"
+                        report += f"Confidence: {vuln['confidence']}\n"
+                        report += "-" * 40 + "\n"
+                else:
+                    report += "✅ No critical security issues detected.\n" 
+                    
+                return report
+
+        def scan_file_or_directory(path):
+            if os.path.isfile(path):
+                scanner = AdvancedVulnerabilityScanner(path)
+                scanner.analyze()
+                return scanner.generate_report()
+            elif os.path.isdir(path):
+                full_report = ""
+                for root, dirs, files in os.walk(path):
+                    for file in files:
+                        if file.endswith('.py'):
+                            file_path = os.path.join(root, file)
+                            scanner = AdvancedVulnerabilityScanner(file_path)
+                            scanner.analyze()
+                            full_report += scanner.generate_report() + "\n\n"
+                return full_report
+            else:
+                return f"Error: {path} is not a valid file or directory."
+
+        def main():
+            path = "."  # Scan the entire repository
+            report = scan_file_or_directory(path)
+            with open('security-scan-results.txt', 'w') as f:
+                f.write(report)
+
+        if __name__ == "__main__":
+            main()
+        EOF
+        import ast
+        import logging
+        from typing import List, Dict, Any
+        import bandit
+        from bandit.core import manager as bandit_manager
+
+        logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
         class AdvancedVulnerabilityScanner:
             def __init__(self, file_path: str):
                 self.file_path = file_path
@@ -274,13 +535,14 @@ jobs:
         fi
 
     - name: Upload scan results
-      uses: actions/upload-artifact@v3
+      uses: actions/upload-artifact@v4
       with:
         name: security-scan-results
         path: security-scan-results.txt
+        retention-days: 90
 
     - name: Comment PR
-      uses: actions/github-script@v6
+      uses: actions/github-script@v7
       if: always()
       with:
         github-token: ${{secrets.GITHUB_TOKEN}}
@@ -301,7 +563,7 @@ jobs:
             comment += '⚠️ **Error: The security scan failed to complete. Please review the workflow logs for more information.**'
           }
           
-          github.rest.issues.createComment({
+          await github.rest.issues.createComment({
             issue_number: context.issue.number,
             owner: context.repo.owner,
             repo: context.repo.repo,