War Story: Migrating 500+ Codebases from Checkmarx 10.0 to SonarQube 10.0

# migrate_sast.py
# Orchestration script to migrate 517 codebases from Checkmarx 10.0 to SonarQube 10.0
# Requires: aiohttp, asyncio, python-dotenv, pandas
# Environment variables: GITHUB_TOKEN, CHECKMARX_TOKEN, SONARQUBE_TOKEN, SONARQUBE_URL

import asyncio
import aiohttp
import json
import logging
import os
import csv
from dotenv import load_dotenv
from pandas import DataFrame

load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("migration.log"), logging.StreamHandler()],
)

# Constants
GITHUB_API_URL = "https://api.github.com"
CHECKMAKX_API_URL = "https://checkmarx.acme-corp.com/rest"
SONARQUBE_URL = os.getenv("SONARQUBE_URL")
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
CHECKMAKX_TOKEN = os.getenv("CHECKMARX_TOKEN")
SONARQUBE_TOKEN = os.getenv("SONARQUBE_TOKEN")
REPO_CSV = "repos.csv"
REPORT_CSV = "migration_report.csv"
MAX_RETRIES = 3
RETRY_DELAY = 5  # seconds

# Fetch all repos from GitHub organization https://github.com/acme-corp
async def fetch_repos(session):
    repos = []
    page = 1
    while True:
        try:
            async with session.get(
                f"{GITHUB_API_URL}/orgs/acme-corp/repos",
                headers={"Authorization": f"token {GITHUB_TOKEN}"},
                params={"page": page, "per_page": 100, "type": "private"},
            ) as resp:
                if resp.status == 429:
                    logging.warning("GitHub rate limit hit, retrying after 60s")
                    await asyncio.sleep(60)
                    continue
                resp.raise_for_status()
                page_repos = await resp.json()
                if not page_repos:
                    break
                repos.extend(page_repos)
                page += 1
        except aiohttp.ClientError as e:
            logging.error(f"Failed to fetch repos page {page}: {e}")
            if page > MAX_RETRIES:
                break
            await asyncio.sleep(RETRY_DELAY)
            page += 1
    return repos

# Run Checkmarx scan for a single repo
async def run_checkmarx_scan(session, repo_name, repo_url):
    for attempt in range(MAX_RETRIES):
        try:
            # Trigger Checkmarx scan via API
            async with session.post(
                f"{CHECKMAKX_API_URL}/scans",
                headers={"Authorization": f"Bearer {CHECKMAKX_TOKEN}"},
                json={"repoUrl": repo_url, "projectName": repo_name},
            ) as resp:
                if resp.status == 429:
                    await asyncio.sleep(RETRY_DELAY * (attempt + 1))
                    continue
                resp.raise_for_status()
                scan_id = (await resp.json())["scanId"]
                # Wait for scan to complete (simplified for example)
                await asyncio.sleep(30)
                return scan_id
        except aiohttp.ClientError as e:
            logging.error(f"Checkmarx scan failed for {repo_name}: {e}")
            if attempt == MAX_RETRIES - 1:
                return None
            await asyncio.sleep(RETRY_DELAY)
    return None

# Run SonarQube scan for a single repo
async def run_sonar_scan(session, repo_name, repo_url):
    for attempt in range(MAX_RETRIES):
        try:
            # Trigger SonarQube scan via CLI (simulated via API)
            # In production, this would call the SonarQube scanner CLI
            async with session.post(
                f"{SONARQUBE_URL}/api/projects/create",
                headers={"Authorization": f"Bearer {SONARQUBE_TOKEN}"},
                data={"project": repo_name, "name": repo_name},
            ) as resp:
                resp.raise_for_status()
                # Trigger scan via webhook (simplified)
                return True
        except aiohttp.ClientError as e:
            logging.error(f"SonarQube scan failed for {repo_name}: {e}")
            if attempt == MAX_RETRIES - 1:
                return False
            await asyncio.sleep(RETRY_DELAY)
    return False

# Main migration logic
async def main():
    async with aiohttp.ClientSession() as session:
        # Fetch all repos
        logging.info("Fetching repos from GitHub...")
        repos = await fetch_repos(session)
        logging.info(f"Fetched {len(repos)} repos")

        # Write repo list to CSV
        with open(REPO_CSV, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=["name", "url", "language"])
            writer.writeheader()
            for repo in repos:
                writer.writerow({
                    "name": repo["name"],
                    "url": repo["html_url"],
                    "language": repo["language"],
                })

        # Process repos in batches of 10 to avoid rate limits
        report_rows = []
        for i in range(0, len(repos), 10):
            batch = repos[i:i+10]
            tasks = []
            for repo in batch:
                repo_name = repo["name"]
                repo_url = repo["html_url"]
                # Run both scans in parallel
                tasks.append(asyncio.gather(
                    run_checkmarx_scan(session, repo_name, repo_url),
                    run_sonar_scan(session, repo_name, repo_url),
                ))
            results = await asyncio.gather(*tasks)
            for repo, (cx_scan, sonar_scan) in zip(batch, results):
                report_rows.append({
                    "repo": repo["name"],
                    "checkmarx_scan": "success" if cx_scan else "failed",
                    "sonarqube_scan": "success" if sonar_scan else "failed",
                    "language": repo["language"],
                })
            logging.info(f"Processed batch {i//10 + 1}, total repos: {len(report_rows)}")

        # Write migration report
        with open(REPORT_CSV, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=["repo", "checkmarx_scan", "sonarqube_scan", "language"])
            writer.writeheader()
            writer.writerows(report_rows)
        logging.info(f"Migration report written to {REPORT_CSV}")

if __name__ == "__main__":
    asyncio.run(main())

# configure_sonar_gates.py
# Converts Checkmarx 10.0 policies to SonarQube 10.0 quality gates via API
# Requires: aiohttp, asyncio, pyyaml, python-dotenv
# Environment variables: SONARQUBE_URL, SONARQUBE_TOKEN
# Input: checkmarx_policies.yaml (exported from Checkmarx 10.0)

import asyncio
import aiohttp
import yaml
import logging
import os
from dotenv import load_dotenv

load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("gate_config.log"), logging.StreamHandler()],
)

# Constants
SONARQUBE_URL = os.getenv("SONARQUBE_URL")
SONARQUBE_TOKEN = os.getenv("SONARQUBE_TOKEN")
POLICIES_FILE = "checkmarx_policies.yaml"
MAX_RETRIES = 3
RETRY_DELAY = 3

# Load Checkmarx policies from YAML
def load_policies():
    try:
        with open(POLICIES_FILE, "r") as f:
            return yaml.safe_load(f)
    except FileNotFoundError:
        logging.error(f"Policies file {POLICIES_FILE} not found")
        raise
    except yaml.YAMLError as e:
        logging.error(f"Failed to parse YAML: {e}")
        raise

# Create a SonarQube quality gate
async def create_quality_gate(session, gate_name, language):
    for attempt in range(MAX_RETRIES):
        try:
            async with session.post(
                f"{SONARQUBE_URL}/api/qualitygates/create",
                headers={"Authorization": f"Bearer {SONARQUBE_TOKEN}"},
                data={"name": f"{gate_name}_{language}"},
            ) as resp:
                if resp.status == 409:
                    logging.warning(f"Quality gate {gate_name}_{language} already exists")
                    return gate_name
                resp.raise_for_status()
                return gate_name
        except aiohttp.ClientError as e:
            logging.error(f"Failed to create quality gate {gate_name}: {e}")
            if attempt == MAX_RETRIES - 1:
                return None
            await asyncio.sleep(RETRY_DELAY)
    return None

# Add condition to quality gate (map Checkmarx rule to SonarQube metric)
async def add_condition(session, gate_name, rule_id, severity, language):
    # Map Checkmarx severity to SonarQube metric threshold
    severity_map = {
        "High": "critical",
        "Medium": "major",
        "Low": "minor",
    }
    sonar_severity = severity_map.get(severity, "major")
    for attempt in range(MAX_RETRIES):
        try:
            async with session.post(
                f"{SONARQUBE_URL}/api/qualitygates/add-condition",
                headers={"Authorization": f"Bearer {SONARQUBE_TOKEN}"},
                data={
                    "gateName": gate_name,
                    "metric": f"security_hotspots_{sonar_severity}",
                    "op": "LT",
                    "warning": "1",
                    "error": "1",
                },
            ) as resp:
                if resp.status == 400:
                    logging.warning(f"Invalid condition for rule {rule_id}")
                    return False
                resp.raise_for_status()
                return True
        except aiohttp.ClientError as e:
            logging.error(f"Failed to add condition for rule {rule_id}: {e}")
            if attempt == MAX_RETRIES - 1:
                return False
            await asyncio.sleep(RETRY_DELAY)
    return False

# Assign quality gate to a project
async def assign_gate_to_project(session, project_key, gate_name):
    for attempt in range(MAX_RETRIES):
        try:
            async with session.post(
                f"{SONARQUBE_URL}/api/qualitygates/select",
                headers={"Authorization": f"Bearer {SONARQUBE_TOKEN}"},
                data={"gateName": gate_name, "projectKey": project_key},
            ) as resp:
                resp.raise_for_status()
                return True
        except aiohttp.ClientError as e:
            logging.error(f"Failed to assign gate {gate_name} to {project_key}: {e}")
            if attempt == MAX_RETRIES - 1:
                return False
            await asyncio.sleep(RETRY_DELAY)
    return False

# Main logic
async def main():
    # Load policies
    try:
        policies = load_policies()
    except Exception:
        return
    logging.info(f"Loaded {len(policies)} policies from {POLICIES_FILE}")

    async with aiohttp.ClientSession() as session:
        for policy in policies:
            policy_name = policy["name"]
            language = policy["language"]
            rules = policy["rules"]
            severity = policy["severity"]

            # Create quality gate for language
            gate_name = await create_quality_gate(session, policy_name, language)
            if not gate_name:
                continue

            # Add conditions for each rule
            for rule in rules:
                await add_condition(session, gate_name, rule["id"], severity, language)

            # Assign to all projects of that language (simplified)
            logging.info(f"Configured quality gate {gate_name} for {language}")

if __name__ == "__main__":
    asyncio.run(main())

# pr_decoration_webhook.py
# Flask webhook handler to replace Checkmarx PR comments with SonarQube results
# Requires: flask, requests, python-dotenv, hmac, hashlib
# Environment variables: GITHUB_TOKEN, SONARQUBE_TOKEN, WEBHOOK_SECRET
# GitHub repos: https://github.com/acme-corp/*

import hmac
import hashlib
import logging
import os
from flask import Flask, request, jsonify
import requests
from dotenv import load_dotenv

load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)

# Constants
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
SONARQUBE_TOKEN = os.getenv("SONARQUBE_TOKEN")
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
SONARQUBE_URL = os.getenv("SONARQUBE_URL")
GITHUB_API_URL = "https://api.github.com"

app = Flask(__name__)

# Verify GitHub webhook signature
def verify_signature(data, signature):
    if not WEBHOOK_SECRET:
        logging.warning("No webhook secret configured, skipping verification")
        return True
    mac = hmac.new(WEBHOOK_SECRET.encode(), msg=data, digestmod=hashlib.sha256)
    expected = f"sha256={mac.hexdigest()}"
    return hmac.compare_digest(expected, signature)

# Post comment to GitHub PR
def post_github_comment(repo, pr_number, comment):
    try:
        resp = requests.post(
            f"{GITHUB_API_URL}/repos/acme-corp/{repo}/issues/{pr_number}/comments",
            headers={
                "Authorization": f"token {GITHUB_TOKEN}",
                "Accept": "application/vnd.github.v3+json",
            },
            json={"body": comment},
        )
        resp.raise_for_status()
        return True
    except requests.exceptions.RequestException as e:
        logging.error(f"Failed to post comment to {repo} PR {pr_number}: {e}")
        return False

# Fetch SonarQube scan results for a project
def fetch_sonar_results(project_key):
    try:
        resp = requests.get(
            f"{SONARQUBE_URL}/api/qualitygates/project_status",
            headers={"Authorization": f"Bearer {SONARQUBE_TOKEN}"},
            params={"projectKey": project_key},
        )
        resp.raise_for_status()
        return resp.json()
    except requests.exceptions.RequestException as e:
        logging.error(f"Failed to fetch SonarQube results for {project_key}: {e}")
        return None

# Format SonarQube results into GitHub comment
def format_comment(sonar_results, repo, pr_number):
    status = sonar_results["projectStatus"]["status"]
    conditions = sonar_results["projectStatus"]["conditions"]
    comment = f"## SonarQube Scan Results for PR #{pr_number}\n"
    comment += f"**Status**: {'✅ Passed' if status == 'OK' else '❌ Failed'}\n\n"
    comment += "### Conditions:\n"
    for cond in conditions:
        metric = cond["metricKey"]
        value = cond["actualValue"]
        status = cond["status"]
        comment += f"- {metric}: {value} ({'✅' if status == 'OK' else '❌'})\n"
    comment += f"\nView full report: {SONARQUBE_URL}/dashboard?id={repo}"
    return comment

# Webhook endpoint for GitHub PR events
@app.route("/github-webhook", methods=["POST"])
def github_webhook():
    # Verify signature
    signature = request.headers.get("X-Hub-Signature-256")
    if not verify_signature(request.data, signature):
        logging.error("Invalid webhook signature")
        return jsonify({"error": "Invalid signature"}), 403

    # Parse payload
    payload = request.json
    if not payload:
        return jsonify({"error": "No payload"}), 400

    # Handle pull request events only
    if payload.get("action") not in ["opened", "synchronize"]:
        return jsonify({"status": "ignored"}), 200

    repo = payload["repository"]["name"]
    pr_number = payload["number"]
    repo_url = payload["repository"]["html_url"]
    # Extract project key from repo URL (https://github.com/acme-corp/repo -> repo)
    project_key = repo

    # Fetch SonarQube results
    sonar_results = fetch_sonar_results(project_key)
    if not sonar_results:
        return jsonify({"error": "Failed to fetch SonarQube results"}), 500

    # Format and post comment
    comment = format_comment(sonar_results, repo, pr_number)
    if post_github_comment(repo, pr_number, comment):
        return jsonify({"status": "success"}), 200
    else:
        return jsonify({"error": "Failed to post comment"}), 500

# Webhook endpoint for SonarQube scan completion
@app.route("/sonar-webhook", methods=["POST"])
def sonar_webhook():
    payload = request.json
    if not payload:
        return jsonify({"error": "No payload"}), 400

    project_key = payload["project"]["key"]
    # Trigger GitHub PR comment update (simplified)
    logging.info(f"SonarQube scan completed for {project_key}")
    return jsonify({"status": "success"}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=False)

import pandas as pd
import json

def diff_findings(checkmarx_json, sonarqube_json):
    cx_df = pd.DataFrame(checkmarx_json["findings"])
    sonar_df = pd.DataFrame(sonarqube_json["issues"])
    # Merge on vulnerability type and file path
    merged = pd.merge(cx_df, sonar_df, left_on=["vulnType", "filePath"], right_on=["rule", "component"], how="outer")
    # Find rules only in Checkmarx
    only_cx = merged[merged["rule"].isna()]
    # Find rules only in SonarQube
    only_sonar = merged[merged["vulnType"].isna()]
    return only_cx, only_sonar

resource "sonarqube_project" "java_microservice" {
  count        = length(var.java_repos)
  project      = var.java_repos[count.index]
  name         = var.java_repos[count.index]
  visibility   = "private"
}

resource "sonarqube_quality_gate_project_association" "java_gate" {
  count        = length(sonarqube_project.java_microservice)
  gate_name    = "java_security_gate"
  project_key  = sonarqube_project.java_microservice[count.index].project
}

- name: Run SAST Scan
  run: |
    if curl -s "https://launchdarkly.acme-corp.com/api/flags/sast-sonar-rollout" | grep -q "${{ github.repository }}"; then
      sonar-scanner -Dsonar.projectKey=${{ github.repository }}
    else
      checkmarx-cli scan -p ${{ github.repository }}
    fi