# .semgrep.yml
rules:
  # SQL Injection Detection
  - id: sql-injection-string-concat
    patterns:
      - pattern-either:
          - pattern: $QUERY = "..." + $USER_INPUT + "..."
          - pattern: $QUERY = f"...{$USER_INPUT}..."
          - pattern: |
              $QUERY = "..." % $USER_INPUT
    message: "Potential SQL injection via string concatenation"
    languages: [python]
    severity: ERROR
    metadata:
      cwe: "CWE-89"
      owasp: "A03:2021 - Injection"
      category: security
      confidence: HIGH
 
  # Hardcoded Secrets
  - id: hardcoded-api-key
    patterns:
      - pattern-regex: |
          (?i)(api[_-]?key|apikey|secret[_-]?key|auth[_-]?token)\s*[=:]\s*["'][a-zA-Z0-9]{16,}["']
    message: "Hardcoded API key or secret detected"
    languages: [python, javascript, typescript, java]
    severity: ERROR
    metadata:
      cwe: "CWE-798"
      category: security
 
  # Insecure Deserialization
  - id: insecure-pickle-load
    patterns:
      - pattern: pickle.load($SOURCE)
      - pattern: pickle.loads($SOURCE)
    message: "Insecure deserialization with pickle - use safer alternatives"
    languages: [python]
    severity: ERROR
    metadata:
      cwe: "CWE-502"
 
  # Command Injection
  - id: command-injection-subprocess
    patterns:
      - pattern: subprocess.call($CMD, shell=True, ...)
      - pattern: subprocess.Popen($CMD, shell=True, ...)
      - pattern: os.system($CMD)
    message: "Potential command injection vulnerability"
    languages: [python]
    severity: ERROR
    metadata:
      cwe: "CWE-78"
 
  # Path Traversal
  - id: path-traversal-open
    patterns:
      - pattern-inside: |
          def $FUNC(..., $PATH, ...):
            ...
      - pattern-not-inside: |
          $PATH = os.path.basename(...)
          ...
      - pattern: open($PATH, ...)
    message: "Potential path traversal - validate file paths"
    languages: [python]
    severity: WARNING
    metadata:
      cwe: "CWE-22"

# custom_semgrep_rules.py
"""
Generate custom Semgrep rules programmatically.
"""
 
import yaml
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict
import json
 
 
@dataclass
class RuleMetadata:
    """Metadata for a Semgrep rule."""
    cwe: str
    owasp: Optional[str] = None
    category: str = "security"
    confidence: str = "HIGH"
    references: List[str] = None
 
 
@dataclass
class SemgrepRule:
    """Represents a Semgrep rule."""
    id: str
    message: str
    severity: str
    languages: List[str]
    patterns: List[Dict[str, Any]]
    metadata: RuleMetadata
    fix: Optional[str] = None
    fix_regex: Optional[Dict[str, str]] = None
 
 
class SemgrepRuleGenerator:
    """Generate custom Semgrep rules for security scanning."""
 
    def __init__(self):
        self.rules: List[SemgrepRule] = []
 
    def add_injection_rule(
        self,
        rule_id: str,
        sink_patterns: List[str],
        languages: List[str],
        injection_type: str = "SQL",
        cwe: str = "CWE-89"
    ):
        """Add an injection detection rule."""
        patterns = []
 
        for sink in sink_patterns:
            patterns.append({
                "pattern-either": [
                    {"pattern": f"{sink}($USER_INPUT)"},
                    {"pattern": f'{sink}("..." + $USER_INPUT + "...")'},
                    {"pattern": f'{sink}(f"...{{$USER_INPUT}}...")'},
                ]
            })
 
        rule = SemgrepRule(
            id=rule_id,
            message=f"Potential {injection_type} injection vulnerability",
            severity="ERROR",
            languages=languages,
            patterns=patterns,
            metadata=RuleMetadata(
                cwe=cwe,
                owasp="A03:2021 - Injection",
                confidence="HIGH"
            )
        )
 
        self.rules.append(rule)
 
    def add_sensitive_data_rule(
        self,
        rule_id: str,
        data_patterns: List[str],
        context: str = "logging"
    ):
        """Add a rule for detecting sensitive data exposure."""
        pattern_either = []
 
        for data_pattern in data_patterns:
            if context == "logging":
                pattern_either.extend([
                    {"pattern": f"logging.info(..., {data_pattern}, ...)"},
                    {"pattern": f"logging.debug(..., {data_pattern}, ...)"},
                    {"pattern": f"print(..., {data_pattern}, ...)"},
                    {"pattern": f"logger.info(..., {data_pattern}, ...)"},
                ])
            elif context == "response":
                pattern_either.extend([
                    {"pattern": f"return {{{data_pattern}: ...}}"},
                    {"pattern": f"json.dumps({{..., {data_pattern}: ..., ...}})"},
                ])
 
        rule = SemgrepRule(
            id=rule_id,
            message=f"Sensitive data may be exposed in {context}",
            severity="WARNING",
            languages=["python", "javascript", "typescript"],
            patterns=[{"pattern-either": pattern_either}],
            metadata=RuleMetadata(
                cwe="CWE-532" if context == "logging" else "CWE-200",
                category="security",
                confidence="MEDIUM"
            )
        )
 
        self.rules.append(rule)
 
    def add_authentication_rule(
        self,
        rule_id: str,
        weak_patterns: List[Dict[str, str]]
    ):
        """Add authentication security rules."""
        patterns = []
 
        for pattern_config in weak_patterns:
            patterns.append({
                "pattern": pattern_config["pattern"],
                "pattern-not": pattern_config.get("pattern-not")
            })
 
        rule = SemgrepRule(
            id=rule_id,
            message="Weak authentication pattern detected",
            severity="ERROR",
            languages=["python", "javascript", "java"],
            patterns=patterns,
            metadata=RuleMetadata(
                cwe="CWE-287",
                owasp="A07:2021 - Identification and Authentication Failures"
            )
        )
 
        self.rules.append(rule)
 
    def add_crypto_rule(
        self,
        rule_id: str,
        weak_algorithms: List[str],
        languages: List[str]
    ):
        """Add weak cryptography detection rules."""
        pattern_either = []
 
        for algo in weak_algorithms:
            pattern_either.extend([
                {"pattern": f'hashlib.{algo}(...)'},
                {"pattern": f'Cipher.getInstance("{algo}")'},
                {"pattern": f'crypto.create{algo.title()}(...)'},
            ])
 
        rule = SemgrepRule(
            id=rule_id,
            message=f"Weak cryptographic algorithm detected: {', '.join(weak_algorithms)}",
            severity="WARNING",
            languages=languages,
            patterns=[{"pattern-either": pattern_either}],
            metadata=RuleMetadata(
                cwe="CWE-327",
                owasp="A02:2021 - Cryptographic Failures"
            )
        )
 
        self.rules.append(rule)
 
    def export_rules(self, output_path: str):
        """Export rules to YAML file."""
        rules_dict = {
            "rules": []
        }
 
        for rule in self.rules:
            rule_dict = {
                "id": rule.id,
                "message": rule.message,
                "severity": rule.severity,
                "languages": rule.languages,
                "patterns": rule.patterns,
                "metadata": {
                    "cwe": rule.metadata.cwe,
                    "category": rule.metadata.category,
                    "confidence": rule.metadata.confidence
                }
            }
 
            if rule.metadata.owasp:
                rule_dict["metadata"]["owasp"] = rule.metadata.owasp
 
            if rule.fix:
                rule_dict["fix"] = rule.fix
 
            rules_dict["rules"].append(rule_dict)
 
        with open(output_path, 'w') as f:
            yaml.dump(rules_dict, f, default_flow_style=False, sort_keys=False)
 
 
# Generate custom rules
generator = SemgrepRuleGenerator()
 
# SQL Injection rules
generator.add_injection_rule(
    "custom-sql-injection",
    ["cursor.execute", "db.execute", "connection.execute"],
    ["python"],
    "SQL",
    "CWE-89"
)
 
# Sensitive data logging
generator.add_sensitive_data_rule(
    "sensitive-data-logging",
    ["password", "secret", "token", "api_key", "ssn", "credit_card"],
    "logging"
)
 
# Weak crypto
generator.add_crypto_rule(
    "weak-crypto-hash",
    ["md5", "sha1"],
    ["python", "java", "javascript"]
)
 
# Export
generator.export_rules("custom-security-rules.yml")

# sast_scanner.py
"""
SAST scanning orchestration with multiple tools.
"""
 
import subprocess
import json
import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import xml.etree.ElementTree as ET
 
 
@dataclass
class Finding:
    """Represents a security finding."""
    tool: str
    rule_id: str
    severity: str
    message: str
    file_path: str
    line_number: int
    code_snippet: Optional[str]
    cwe: Optional[str]
    fix_suggestion: Optional[str]
 
 
class SASTScanner:
    """
    Orchestrates multiple SAST tools for comprehensive scanning.
    """
 
    def __init__(self, project_path: str):
        self.project_path = project_path
        self.findings: List[Finding] = []
 
    def run_semgrep(
        self,
        config: str = "auto",
        additional_rules: Optional[str] = None
    ) -> List[Finding]:
        """Run Semgrep scan."""
        cmd = [
            "semgrep", "scan",
            "--config", config,
            "--json",
            "--no-git-ignore",
            self.project_path
        ]
 
        if additional_rules:
            cmd.extend(["--config", additional_rules])
 
        result = subprocess.run(cmd, capture_output=True, text=True)
 
        findings = []
        try:
            data = json.loads(result.stdout)
 
            for match in data.get("results", []):
                findings.append(Finding(
                    tool="semgrep",
                    rule_id=match.get("check_id", "unknown"),
                    severity=match.get("extra", {}).get("severity", "INFO"),
                    message=match.get("extra", {}).get("message", ""),
                    file_path=match.get("path", ""),
                    line_number=match.get("start", {}).get("line", 0),
                    code_snippet=match.get("extra", {}).get("lines", ""),
                    cwe=match.get("extra", {}).get("metadata", {}).get("cwe"),
                    fix_suggestion=match.get("extra", {}).get("fix")
                ))
        except json.JSONDecodeError:
            pass
 
        self.findings.extend(findings)
        return findings
 
    def run_bandit(self, confidence: str = "MEDIUM") -> List[Finding]:
        """Run Bandit for Python security scanning."""
        cmd = [
            "bandit",
            "-r", self.project_path,
            "-f", "json",
            "-ll",  # Only medium and higher severity
            f"-c", confidence
        ]
 
        result = subprocess.run(cmd, capture_output=True, text=True)
 
        findings = []
        try:
            data = json.loads(result.stdout)
 
            for issue in data.get("results", []):
                severity_map = {"LOW": "INFO", "MEDIUM": "WARNING", "HIGH": "ERROR"}
 
                findings.append(Finding(
                    tool="bandit",
                    rule_id=issue.get("test_id", "unknown"),
                    severity=severity_map.get(issue.get("severity", "LOW"), "INFO"),
                    message=issue.get("issue_text", ""),
                    file_path=issue.get("filename", ""),
                    line_number=issue.get("line_number", 0),
                    code_snippet=issue.get("code", ""),
                    cwe=issue.get("cwe", {}).get("id") if issue.get("cwe") else None,
                    fix_suggestion=None
                ))
        except json.JSONDecodeError:
            pass
 
        self.findings.extend(findings)
        return findings
 
    def run_eslint_security(self) -> List[Finding]:
        """Run ESLint with security plugins for JavaScript/TypeScript."""
        cmd = [
            "npx", "eslint",
            "--plugin", "security",
            "--format", "json",
            self.project_path
        ]
 
        result = subprocess.run(cmd, capture_output=True, text=True)
 
        findings = []
        try:
            data = json.loads(result.stdout)
 
            for file_result in data:
                for message in file_result.get("messages", []):
                    if "security" in message.get("ruleId", ""):
                        severity_map = {1: "WARNING", 2: "ERROR"}
 
                        findings.append(Finding(
                            tool="eslint-security",
                            rule_id=message.get("ruleId", "unknown"),
                            severity=severity_map.get(message.get("severity", 1), "INFO"),
                            message=message.get("message", ""),
                            file_path=file_result.get("filePath", ""),
                            line_number=message.get("line", 0),
                            code_snippet=message.get("source", ""),
                            cwe=None,
                            fix_suggestion=message.get("fix")
                        ))
        except json.JSONDecodeError:
            pass
 
        self.findings.extend(findings)
        return findings
 
    def run_all_scanners(self) -> Dict[str, Any]:
        """Run all configured SAST scanners."""
        results = {
            "scan_time": datetime.utcnow().isoformat(),
            "project_path": self.project_path,
            "scanners": {},
            "summary": {}
        }
 
        # Run each scanner
        scanners = [
            ("semgrep", self.run_semgrep),
            ("bandit", self.run_bandit),
            ("eslint-security", self.run_eslint_security),
        ]
 
        for name, scanner_fn in scanners:
            try:
                findings = scanner_fn()
                results["scanners"][name] = {
                    "status": "success",
                    "findings_count": len(findings)
                }
            except Exception as e:
                results["scanners"][name] = {
                    "status": "error",
                    "error": str(e)
                }
 
        # Generate summary
        results["summary"] = self.generate_summary()
 
        return results
 
    def generate_summary(self) -> Dict[str, Any]:
        """Generate summary of all findings."""
        summary = {
            "total_findings": len(self.findings),
            "by_severity": {},
            "by_tool": {},
            "by_cwe": {},
            "critical_files": []
        }
 
        # Count by severity
        for finding in self.findings:
            severity = finding.severity
            summary["by_severity"][severity] = summary["by_severity"].get(severity, 0) + 1
 
        # Count by tool
        for finding in self.findings:
            tool = finding.tool
            summary["by_tool"][tool] = summary["by_tool"].get(tool, 0) + 1
 
        # Count by CWE
        for finding in self.findings:
            if finding.cwe:
                summary["by_cwe"][finding.cwe] = summary["by_cwe"].get(finding.cwe, 0) + 1
 
        # Find critical files
        file_counts = {}
        for finding in self.findings:
            if finding.severity in ["ERROR", "HIGH"]:
                file_counts[finding.file_path] = file_counts.get(finding.file_path, 0) + 1
 
        summary["critical_files"] = sorted(
            file_counts.items(),
            key=lambda x: x[1],
            reverse=True
        )[:10]
 
        return summary
 
    def export_sarif(self, output_path: str):
        """Export findings in SARIF format for GitHub integration."""
        sarif = {
            "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/master/Schemata/sarif-schema-2.1.0.json",
            "version": "2.1.0",
            "runs": [{
                "tool": {
                    "driver": {
                        "name": "SAST Scanner",
                        "version": "1.0.0",
                        "rules": []
                    }
                },
                "results": []
            }]
        }
 
        # Add unique rules
        rules_seen = set()
        for finding in self.findings:
            if finding.rule_id not in rules_seen:
                rules_seen.add(finding.rule_id)
                sarif["runs"][0]["tool"]["driver"]["rules"].append({
                    "id": finding.rule_id,
                    "shortDescription": {"text": finding.message[:100]},
                    "help": {"text": finding.message}
                })
 
        # Add results
        for i, finding in enumerate(self.findings):
            sarif["runs"][0]["results"].append({
                "ruleId": finding.rule_id,
                "level": "error" if finding.severity in ["ERROR", "HIGH"] else "warning",
                "message": {"text": finding.message},
                "locations": [{
                    "physicalLocation": {
                        "artifactLocation": {"uri": finding.file_path},
                        "region": {"startLine": finding.line_number}
                    }
                }]
            })
 
        with open(output_path, 'w') as f:
            json.dump(sarif, f, indent=2)

# zap-automation.yaml
env:
  contexts:
    - name: "Application Context"
      urls:
        - "https://app.example.com"
      includePaths:
        - "https://app.example.com/.*"
      excludePaths:
        - "https://app.example.com/logout.*"
        - "https://app.example.com/static/.*"
      authentication:
        method: "form"
        parameters:
          loginUrl: "https://app.example.com/login"
          loginRequestData: "username={%username%}&password={%password%}"
        verification:
          method: "response"
          loggedInRegex: "\\QLogout\\E"
          loggedOutRegex: "\\QLogin\\E"
      users:
        - name: "test-user"
          credentials:
            username: "${ZAP_AUTH_USER}"
            password: "${ZAP_AUTH_PASS}"
 
  parameters:
    failOnError: true
    failOnWarning: false
    progressToStdout: true
 
jobs:
  # Spider the application
  - type: spider
    parameters:
      context: "Application Context"
      user: "test-user"
      maxDuration: 10
      maxDepth: 5
      maxChildren: 10
 
  # AJAX Spider for JavaScript-heavy apps
  - type: spiderAjax
    parameters:
      context: "Application Context"
      user: "test-user"
      maxDuration: 10
      maxCrawlDepth: 5
      numberOfBrowsers: 2
 
  # Passive scan
  - type: passiveScan-wait
    parameters:
      maxDuration: 5
 
  # Active scan
  - type: activeScan
    parameters:
      context: "Application Context"
      user: "test-user"
      policy: "API-Scan"
      maxRuleDurationInMins: 5
      maxScanDurationInMins: 30
 
  # Generate reports
  - type: report
    parameters:
      template: "traditional-html"
      reportDir: "/zap/reports"
      reportFile: "zap-report.html"
    risks:
      - high
      - medium
      - low
 
  - type: report
    parameters:
      template: "sarif-json"
      reportDir: "/zap/reports"
      reportFile: "zap-report.sarif"

# zap_custom_scanner.py
"""
Custom ZAP scanning with Python API.
"""
 
from zapv2 import ZAPv2
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import json
 
 
@dataclass
class DASTFinding:
    """DAST vulnerability finding."""
    alert_id: str
    name: str
    risk: str
    confidence: str
    url: str
    parameter: Optional[str]
    evidence: Optional[str]
    description: str
    solution: str
    cwe_id: Optional[str]
    wasc_id: Optional[str]
 
 
class ZAPScanner:
    """
    Custom ZAP scanner with enhanced functionality.
    """
 
    def __init__(
        self,
        target_url: str,
        api_key: str = "",
        proxy_host: str = "localhost",
        proxy_port: int = 8080
    ):
        self.target_url = target_url
        self.zap = ZAPv2(
            apikey=api_key,
            proxies={
                "http": f"http://{proxy_host}:{proxy_port}",
                "https": f"http://{proxy_host}:{proxy_port}"
            }
        )
        self.context_id = None
        self.user_id = None
 
    def setup_context(
        self,
        context_name: str,
        include_paths: List[str],
        exclude_paths: List[str]
    ):
        """Set up scanning context."""
        # Create context
        self.context_id = self.zap.context.new_context(context_name)
 
        # Add include patterns
        for path in include_paths:
            self.zap.context.include_in_context(context_name, path)
 
        # Add exclude patterns
        for path in exclude_paths:
            self.zap.context.exclude_from_context(context_name, path)
 
        return self.context_id
 
    def setup_authentication(
        self,
        auth_type: str = "form",
        login_url: str = None,
        login_data: str = None,
        logged_in_indicator: str = None
    ):
        """Configure authentication for authenticated scanning."""
        if auth_type == "form":
            self.zap.authentication.set_authentication_method(
                self.context_id,
                "formBasedAuthentication",
                f"loginUrl={login_url}&loginRequestData={login_data}"
            )
 
        if logged_in_indicator:
            self.zap.authentication.set_logged_in_indicator(
                self.context_id,
                logged_in_indicator
            )
 
    def add_user(
        self,
        username: str,
        password: str,
        user_name: str = "test-user"
    ):
        """Add a user for authenticated scanning."""
        self.user_id = self.zap.users.new_user(self.context_id, user_name)
 
        self.zap.users.set_authentication_credentials(
            self.context_id,
            self.user_id,
            f"username={username}&password={password}"
        )
 
        self.zap.users.set_user_enabled(self.context_id, self.user_id, True)
 
        return self.user_id
 
    def run_spider(
        self,
        max_depth: int = 5,
        max_duration: int = 0
    ) -> int:
        """Run the spider to discover URLs."""
        if self.user_id:
            scan_id = self.zap.spider.scan_as_user(
                self.context_id,
                self.user_id,
                self.target_url,
                maxchildren=10,
                recurse=True,
                subtreeonly=False
            )
        else:
            scan_id = self.zap.spider.scan(
                self.target_url,
                maxchildren=10,
                recurse=True,
                subtreeonly=False
            )
 
        # Wait for spider to complete
        while int(self.zap.spider.status(scan_id)) < 100:
            time.sleep(2)
 
        return len(self.zap.spider.results(scan_id))
 
    def run_ajax_spider(self, max_duration: int = 10):
        """Run AJAX spider for JavaScript applications."""
        if self.user_id:
            self.zap.ajaxSpider.scan_as_user(
                self.context_id,
                self.user_id,
                self.target_url,
                subtreeonly=False
            )
        else:
            self.zap.ajaxSpider.scan(self.target_url)
 
        # Wait for AJAX spider
        start_time = time.time()
        while self.zap.ajaxSpider.status == "running":
            if time.time() - start_time > max_duration * 60:
                self.zap.ajaxSpider.stop()
                break
            time.sleep(5)
 
    def run_passive_scan(self, max_wait: int = 300):
        """Wait for passive scanning to complete."""
        start_time = time.time()
 
        while int(self.zap.pscan.records_to_scan) > 0:
            if time.time() - start_time > max_wait:
                break
            time.sleep(2)
 
    def run_active_scan(
        self,
        scan_policy: Optional[str] = None
    ) -> str:
        """Run active security scan."""
        if self.user_id:
            scan_id = self.zap.ascan.scan_as_user(
                self.target_url,
                self.context_id,
                self.user_id,
                recurse=True,
                scanpolicyname=scan_policy
            )
        else:
            scan_id = self.zap.ascan.scan(
                self.target_url,
                recurse=True,
                scanpolicyname=scan_policy
            )
 
        # Monitor progress
        while int(self.zap.ascan.status(scan_id)) < 100:
            progress = self.zap.ascan.status(scan_id)
            print(f"Active scan progress: {progress}%")
            time.sleep(10)
 
        return scan_id
 
    def get_alerts(
        self,
        min_risk: str = "Low"
    ) -> List[DASTFinding]:
        """Get all alerts from the scan."""
        risk_levels = ["Informational", "Low", "Medium", "High"]
        min_risk_index = risk_levels.index(min_risk)
 
        alerts = self.zap.core.alerts(baseurl=self.target_url)
        findings = []
 
        for alert in alerts:
            alert_risk = alert.get("risk", "Informational")
            if risk_levels.index(alert_risk) >= min_risk_index:
                findings.append(DASTFinding(
                    alert_id=alert.get("id", ""),
                    name=alert.get("name", ""),
                    risk=alert_risk,
                    confidence=alert.get("confidence", ""),
                    url=alert.get("url", ""),
                    parameter=alert.get("param"),
                    evidence=alert.get("evidence"),
                    description=alert.get("description", ""),
                    solution=alert.get("solution", ""),
                    cwe_id=alert.get("cweid"),
                    wasc_id=alert.get("wascid")
                ))
 
        return findings
 
    def run_full_scan(self) -> Dict[str, Any]:
        """Run complete DAST scan."""
        results = {
            "target": self.target_url,
            "start_time": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "phases": {}
        }
 
        # Spider
        print("Running spider...")
        urls_found = self.run_spider()
        results["phases"]["spider"] = {"urls_discovered": urls_found}
 
        # AJAX Spider
        print("Running AJAX spider...")
        self.run_ajax_spider()
        results["phases"]["ajax_spider"] = {"status": "completed"}
 
        # Passive scan
        print("Running passive scan...")
        self.run_passive_scan()
        results["phases"]["passive_scan"] = {"status": "completed"}
 
        # Active scan
        print("Running active scan...")
        self.run_active_scan()
        results["phases"]["active_scan"] = {"status": "completed"}
 
        # Get findings
        findings = self.get_alerts()
        results["findings"] = [
            {
                "name": f.name,
                "risk": f.risk,
                "confidence": f.confidence,
                "url": f.url,
                "parameter": f.parameter,
                "cwe": f.cwe_id
            }
            for f in findings
        ]
 
        # Summary
        results["summary"] = {
            "total_findings": len(findings),
            "high_risk": len([f for f in findings if f.risk == "High"]),
            "medium_risk": len([f for f in findings if f.risk == "Medium"]),
            "low_risk": len([f for f in findings if f.risk == "Low"])
        }
 
        results["end_time"] = time.strftime("%Y-%m-%dT%H:%M:%SZ")
 
        return results

# .github/workflows/security-testing.yml
name: Security Testing
 
on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
 
jobs:
  sast:
    name: SAST Scanning
    runs-on: ubuntu-latest
    permissions:
      security-events: write
      contents: read
 
    steps:
      - name: Checkout code
        uses: actions/checkout@v4
 
      - name: Run Semgrep
        uses: returntocorp/semgrep-action@v1
        with:
          config: >-
            p/security-audit
            p/owasp-top-ten
            p/cwe-top-25
          generateSarif: true
 
      - name: Upload Semgrep SARIF
        uses: github/codeql-action/upload-sarif@v2
        with:
          sarif_file: semgrep.sarif
 
      - name: Run Bandit (Python)
        run: |
          pip install bandit
          bandit -r . -f sarif -o bandit.sarif || true
 
      - name: Upload Bandit SARIF
        uses: github/codeql-action/upload-sarif@v2
        with:
          sarif_file: bandit.sarif
        if: always()
 
      - name: Run npm audit
        run: |
          npm audit --json > npm-audit.json || true
 
      - name: Process npm audit results
        run: |
          python scripts/process-npm-audit.py npm-audit.json
 
  dast:
    name: DAST Scanning
    runs-on: ubuntu-latest
    needs: [sast]
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
 
    services:
      app:
        image: ${{ github.repository }}:${{ github.sha }}
        ports:
          - 3000:3000
 
    steps:
      - name: Checkout
        uses: actions/checkout@v4
 
      - name: OWASP ZAP Full Scan
        uses: zaproxy/action-full-scan@v0.8.0
        with:
          target: 'http://localhost:3000'
          rules_file_name: '.zap/rules.tsv'
          cmd_options: '-a'
 
      - name: Upload ZAP Report
        uses: actions/upload-artifact@v4
        with:
          name: zap-report
          path: report_html.html
 
      - name: Upload ZAP SARIF
        uses: github/codeql-action/upload-sarif@v2
        with:
          sarif_file: zap.sarif
 
  security-gate:
    name: Security Gate
    runs-on: ubuntu-latest
    needs: [sast, dast]
    if: always()
 
    steps:
      - name: Check for critical findings
        run: |
          echo "Checking security scan results..."
          # Add logic to fail if critical vulnerabilities found