We Replaced Our Entire QA Team with AI Code Review Tools: Snyk 1.130 vs. SonarQube 10.6

import os
import requests
import json
import time
from typing import List, Dict, Optional
from dataclasses import dataclass

# Data class to represent a Snyk vulnerability finding
@dataclass
class SnykFinding:
    id: str
    severity: str
    package: str
    version: str
    title: str
    is_false_positive: bool = False

class SnykScanner:
    """Client for Snyk 1.130 REST API v1"""
    BASE_URL = "https://api.snyk.io/v1"

    def __init__(self, api_key: str, org_id: str):
        self.api_key = api_key
        self.org_id = org_id
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"token {self.api_key}",
            "Content-Type": "application/json"
        })

    def trigger_scan(self, repo_url: str, branch: str = "main") -> Optional[str]:
        """Trigger a Snyk code scan for a given repository, returns scan ID"""
        endpoint = f"/orgs/{self.org_id}/projects"
        payload = {
            "name": repo_url.split("/")[-1],
            "type": "github",
            "repo_url": repo_url,
            "branch": branch,
            "scan_settings": {
                "ai_review_enabled": True,
                "ai_fix_suggestions": True
            }
        }
        try:
            response = self.session.post(f"{self.BASE_URL}{endpoint}", json=payload, timeout=30)
            response.raise_for_status()
            project_id = response.json().get("id")
            # Wait for scan to complete (polling every 5s)
            scan_id = self._wait_for_scan_completion(project_id)
            return scan_id
        except requests.exceptions.RequestException as e:
            print(f"Failed to trigger Snyk scan: {str(e)}")
            return None
        except KeyError as e:
            print(f"Missing expected key in Snyk response: {str(e)}")
            return None

    def _wait_for_scan_completion(self, project_id: str, timeout: int = 300) -> str:
        """Poll Snyk API for scan completion, returns scan ID"""
        start_time = time.time()
        while time.time() - start_time < timeout:
            endpoint = f"/orgs/{self.org_id}/projects/{project_id}/scans/latest"
            response = self.session.get(f"{self.BASE_URL}{endpoint}", timeout=10)
            if response.json().get("status") == "completed":
                return response.json().get("id")
            time.sleep(5)
        raise TimeoutError(f"Snyk scan for project {project_id} timed out after {timeout}s")

    def get_findings(self, scan_id: str) -> List[SnykFinding]:
        """Retrieve all findings from a completed scan, filter out low severity"""
        endpoint = f"/orgs/{self.org_id}/scans/{scan_id}/issues"
        try:
            response = self.session.get(f"{self.BASE_URL}{endpoint}", timeout=30)
            response.raise_for_status()
            findings = []
            for issue in response.json().get("issues", []):
                # Filter out informational severity findings
                if issue.get("severity") in ["high", "critical", "medium"]:
                    findings.append(SnykFinding(
                        id=issue.get("id"),
                        severity=issue.get("severity"),
                        package=issue.get("package", "N/A"),
                        version=issue.get("version", "N/A"),
                        title=issue.get("title", "No title")
                    ))
            return findings
        except requests.exceptions.RequestException as e:
            print(f"Failed to retrieve Snyk findings: {str(e)}")
            return []

if __name__ == "__main__":
    # Load environment variables (never hardcode API keys!)
    SNYK_API_KEY = os.getenv("SNYK_API_KEY")
    SNYK_ORG_ID = os.getenv("SNYK_ORG_ID")
    if not SNYK_API_KEY or not SNYK_ORG_ID:
        raise ValueError("Missing SNYK_API_KEY or SNYK_ORG_ID environment variables")

    scanner = SnykScanner(SNYK_API_KEY, SNYK_ORG_ID)
    test_repo = "https://github.com/spring-projects/spring-boot"
    print(f"Triggering Snyk 1.130 scan for {test_repo}...")
    scan_id = scanner.trigger_scan(test_repo)
    if scan_id:
        print(f"Scan completed. Scan ID: {scan_id}")
        findings = scanner.get_findings(scan_id)
        print(f"Found {len(findings)} high/medium/critical issues:")
        for f in findings[:5]:  # Print first 5 findings
            print(f" - [{f.severity.upper()}] {f.title} (Package: {f.package}@{f.version})")
    else:
        print("Scan failed to trigger.")

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "os"
    "time"
)

// SonarQubeFinding represents a single issue from SonarQube 10.6 API
type SonarQubeFinding struct {
    Key          string `json:"key"`
    Rule         string `json:"rule"`
    Severity     string `json:"severity"`
    Component    string `json:"component"`
    Line         int    `json:"line"`
    Message      string `json:"message"`
    AIFixAvailable bool `json:"aiFixAvailable"`
}

// SonarQubeClient is a client for SonarQube 10.6 REST API
type SonarQubeClient struct {
    baseURL string
    token   string
    client  *http.Client
}

// NewSonarQubeClient initializes a new client with timeout
func NewSonarQubeClient(baseURL, token string) *SonarQubeClient {
    return &SonarQubeClient{
        baseURL: baseURL,
        token:   token,
        client: &http.Client{
            Timeout: 30 * time.Second,
        },
    }
}

// TriggerScan triggers a new scan for a given project key
func (c *SonarQubeClient) TriggerScan(projectKey string) error {
    endpoint := fmt.Sprintf("%s/api/projects/analyze", c.baseURL)
    payload := map[string]string{
        "project": projectKey,
        "scanMode": "ai_assisted",
    }
    jsonPayload, err := json.Marshal(payload)
    if err != nil {
        return fmt.Errorf("failed to marshal payload: %w", err)
    }

    req, err := http.NewRequest("POST", endpoint, bytes.NewBuffer(jsonPayload))
    if err != nil {
        return fmt.Errorf("failed to create request: %w", err)
    }
    req.SetBasicAuth(c.token, "")
    req.Header.Set("Content-Type", "application/json")

    resp, err := c.client.Do(req)
    if err != nil {
        return fmt.Errorf("failed to trigger scan: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        body, _ := io.ReadAll(resp.Body)
        return fmt.Errorf("scan trigger failed with status %d: %s", resp.StatusCode, string(body))
    }
    fmt.Printf("Triggered SonarQube 10.6 scan for project %s\n", projectKey)
    return nil
}

// GetFindings retrieves all open findings for a project with AI fix available
func (c *SonarQubeClient) GetFindings(projectKey string) ([]SonarQubeFinding, error) {
    endpoint := fmt.Sprintf("%s/api/issues/search", c.baseURL)
    req, err := http.NewRequest("GET", endpoint, nil)
    if err != nil {
        return nil, fmt.Errorf("failed to create request: %w", err)
    }
    req.SetBasicAuth(c.token, "")
    q := req.URL.Query()
    q.Add("projects", projectKey)
    q.Add("severities", "CRITICAL,MAJOR")
    q.Add("aiFixAvailable", "true")
    req.URL.RawQuery = q.Encode()

    resp, err := c.client.Do(req)
    if err != nil {
        return nil, fmt.Errorf("failed to get findings: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        body, _ := io.ReadAll(resp.Body)
        return nil, fmt.Errorf("findings request failed with status %d: %s", resp.StatusCode, string(body))
    }

    var result struct {
        Issues []SonarQubeFinding `json:"issues"`
    }
    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return nil, fmt.Errorf("failed to decode response: %w", err)
    }

    return result.Issues, nil
}

func main() {
    // Load config from environment variables
    sonarURL := os.Getenv("SONARQUBE_URL")
    sonarToken := os.Getenv("SONARQUBE_TOKEN")
    projectKey := os.Getenv("SONARQUBE_PROJECT_KEY")
    if sonarURL == "" || sonarToken == "" || projectKey == "" {
        fmt.Fprintf(os.Stderr, "Missing required env vars: SONARQUBE_URL, SONARQUBE_TOKEN, SONARQUBE_PROJECT_KEY\n")
        os.Exit(1)
    }

    client := NewSonarQubeClient(sonarURL, sonarToken)
    if err := client.TriggerScan(projectKey); err != nil {
        fmt.Fprintf(os.Stderr, "Error triggering scan: %v\n", err)
        os.Exit(1)
    }

    // Wait for scan to complete (simplified polling)
    time.Sleep(60 * time.Second)

    findings, err := client.GetFindings(projectKey)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error getting findings: %v\n", err)
        os.Exit(1)
    }

    fmt.Printf("Found %d findings with AI fixes available:\n", len(findings))
    for i, f := range findings {
        if i >= 5 { // Print first 5
            break
        }
        fmt.Printf(" - [%s] %s (Line %d: %s)\n", f.Severity, f.Message, f.Line, f.Rule)
    }
}

import os
import json
import sys
from typing import List, Dict
from snyk_client import SnykScanner  # Assumes Code Example 1 class
from sonar_client import SonarQubeClient  # Assumes Code Example 2 class

SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")
if not SLACK_WEBHOOK_URL:
    raise ValueError("Missing SLACK_WEBHOOK_URL environment variable")

def load_config() -> Dict:
    """Load all required config from environment variables"""
    required = [
        "SNYK_API_KEY", "SNYK_ORG_ID", "SONARQUBE_URL",
        "SONARQUBE_TOKEN", "SONARQUBE_PROJECT_KEY", "REPO_URL"
    ]
    config = {}
    for var in required:
        val = os.getenv(var)
        if not val:
            raise ValueError(f"Missing required environment variable: {var}")
        config[var.lower()] = val
    return config

def merge_findings(snyk_findings: List[Dict], sonar_findings: List[Dict]) -> List[Dict]:
    """Merge findings from both tools, deduplicate by rule + line + package"""
    merged = {}
    for f in snyk_findings:
        key = f"{f['rule']}-{f['line']}-{f['package']}"
        merged[key] = {**f, "source": "snyk"}
    for f in sonar_findings:
        key = f"{f['rule']}-{f['line']}-{f['component']}"
        if key in merged:
            merged[key]["source"] = "both"
        else:
            merged[key] = {**f, "source": "sonarqube"}
    return list(merged.values())

def calculate_metrics(merged_findings: List[Dict], known_false_positives: List[str]) -> Dict:
    """Calculate true positive, false positive rates"""
    tp = 0
    fp = 0
    for f in merged_findings:
        if f["id"] in known_false_positives:
            fp += 1
        else:
            tp += 1
    total = len(merged_findings)
    return {
        "true_positives": tp,
        "false_positives": fp,
        "tp_rate": (tp / total) * 100 if total > 0 else 0,
        "fp_rate": (fp / total) * 100 if total > 0 else 0
    }

def post_to_slack(metrics: Dict, merged_count: int):
    """Post scan results to Slack via webhook"""
    import requests
    payload = {
        "text": "🔍 AI Code Review Scan Complete",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Scan Results for {os.getenv('REPO_URL')}*"
                }
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Total Findings:* {merged_count}"},
                    {"type": "mrkdwn", "text": f"*True Positive Rate:* {metrics['tp_rate']:.1f}%"},
                    {"type": "mrkdwn", "text": f"*False Positive Rate:* {metrics['fp_rate']:.1f}%"},
                    {"type": "mrkdwn", "text": f"*Tools Used:* Snyk 1.130 + SonarQube 10.6"}
                ]
            }
        ]
    }
    try:
        response = requests.post(SLACK_WEBHOOK_URL, json=payload, timeout=10)
        response.raise_for_status()
        print("Posted results to Slack successfully")
    except requests.exceptions.RequestException as e:
        print(f"Failed to post to Slack: {str(e)}")

def main():
    try:
        config = load_config()
    except ValueError as e:
        print(f"Config error: {str(e)}")
        sys.exit(1)

    # Initialize clients
    snyk = SnykScanner(config["snyk_api_key"], config["snyk_org_id"])
    sonar = SonarQubeClient(config["sonarqube_url"], config["sonarqube_token"])

    # Run Snyk scan
    print("Running Snyk 1.130 scan...")
    snyk_scan_id = snyk.trigger_scan(config["repo_url"])
    if not snyk_scan_id:
        print("Snyk scan failed, exiting")
        sys.exit(1)
    snyk_findings = snyk.get_findings(snyk_scan_id)
    print(f"Snyk found {len(snyk_findings)} findings")

    # Run SonarQube scan
    print("Running SonarQube 10.6 scan...")
    if err := sonar.TriggerScan(config["sonarqube_project_key"]); err != nil:
        print(f"SonarQube scan failed: {err}")
        sys.exit(1)
    sonar_findings, err := sonar.GetFindings(config["sonarqube_project_key"])
    if err != nil:
        print(f"Failed to get SonarQube findings: {err}")
        sys.exit(1)
    print(f"SonarQube found {len(sonar_findings)} findings")

    # Merge and process
    merged = merge_findings(snyk_findings, sonar_findings)
    # Load known false positives from file (curated list from past scans)
    known_fp = []
    if os.path.exists("known_false_positives.json"):
        with open("known_false_positives.json") as f:
            known_fp = json.load(f)
    metrics = calculate_metrics(merged, known_fp)

    # Post results
    post_to_slack(metrics, len(merged))

    # Save full report
    with open("scan_report.json", "w") as f:
        json.dump({"metrics": metrics, "findings": merged}, f, indent=2)
    print("Full report saved to scan_report.json")

if __name__ == "__main__":
    main()

# Snyk API snippet to update AI fix confidence threshold
import requests
response = requests.patch(
    "https://api.snyk.io/v1/orgs/{org_id}/settings/ai",
    headers={"Authorization": "token {api_key}"},
    json={"fix_confidence_threshold": 85}
)

# Deduplicate Snyk findings by package + vulnerability
seen = set()
unique_findings = []
for f in snyk_findings:
    key = f"{f.package}-{f.version}-{f.id}"
    if key not in seen:
        seen.add(key)
        unique_findings.append(f)

# GitHub Actions step to block PRs with Snyk high/critical issues
- name: Run Snyk Scan
  uses: snyk/actions@v1.130
  with:
    args: --severity-threshold=high
  env:
    SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}