Tutorial: How to Implement Security Scanning in GitLab CI 16.8 with Trivy 0.50 and Grype 0.70

# Sample pipeline output (truncated)
$ trivy image --severity CRITICAL,HIGH --format sarif --output trivy-report.sarif my-app:latest
2024-03-15T10:23:45Z INFO Scanning my-app:latest...
2024-03-15T10:23:49Z INFO Detected 2 critical, 5 high vulnerabilities
$ grype my-app:latest -o sarif --file grype-report.sarif --fail-on critical
2024-03-15T10:23:47Z INFO Scanning image...
2024-03-15T10:23:52Z INFO Found 2 critical, 4 high vulnerabilities
$ gitlab-ci-security-report --merge-reports trivy-report.sarif grype-report.sarif --output combined.sarif
2024-03-15T10:23:53Z INFO Merged 4 critical, 9 high vulnerabilities into combined report
Pipeline failed: 2 critical vulnerabilities detected (CVSS ≥ 9.0)

# setup_project.py
# Initializes a sample vulnerable Node.js app for security scanning tutorials
# Requires Python 3.10+, requests library (pip install requests)

import os
import json
import shutil
import requests
from pathlib import Path
from typing import Dict, Any

# Configuration constants
SAMPLE_APP_DIR = Path("./sample-app")
NODE_VERSION = "20.9.0"
VULNERABLE_DEP_VERSION = "3.29.0"  # express 3.29.0 has known CVEs
DOCKERFILE_CONTENT = f"""FROM node:{NODE_VERSION}-alpine3.19
WORKDIR /app
COPY package*.json ./
RUN npm install --production
COPY . .
EXPOSE 3000
CMD ["node", "app.js"]
"""

APP_JS_CONTENT = """const express = require("express");
const app = express();
const port = 3000;

app.get("/", (req, res) => {{
  res.send("Hello World from vulnerable app!");
}});

app.listen(port, () => {{
  console.log(`App listening on port ${{port}}`);
}});
"""

PACKAGE_JSON_CONTENT = json.dumps({{
  "name": "sample-vulnerable-app",
  "version": "1.0.0",
  "dependencies": {{
    "express": VULNERABLE_DEP_VERSION  # CVE-2017-16026, CVE-2018-3715
  }}
}}, indent=2)


def create_directory_structure() -> None:
  """Creates the sample app directory and subdirectories."""
  try:
    if SAMPLE_APP_DIR.exists():
      shutil.rmtree(SAMPLE_APP_DIR)
    SAMPLE_APP_DIR.mkdir(parents=True)
    (SAMPLE_APP_DIR / "src").mkdir()
    print(f"Created directory structure at {SAMPLE_APP_DIR.absolute()}")
  except PermissionError as e:
    print(f"ERROR: Permission denied creating directories: {e}")
    raise
  except Exception as e:
    print(f"ERROR: Failed to create directory structure: {e}")
    raise


def write_app_files() -> None:
  """Writes Dockerfile, app.js, and package.json to the sample app directory."""
  try:
    # Write Dockerfile
    (SAMPLE_APP_DIR / "Dockerfile").write_text(DOCKERFILE_CONTENT)
    print("Wrote Dockerfile with Node.js 20.9.0 Alpine base")

    # Write app.js
    (SAMPLE_APP_DIR / "src" / "app.js").write_text(APP_JS_CONTENT)
    print("Wrote vulnerable express app (v3.29.0)")

    # Write package.json
    (SAMPLE_APP_DIR / "package.json").write_text(PACKAGE_JSON_CONTENT)
    print("Wrote package.json with vulnerable express dependency")
  except IOError as e:
    print(f"ERROR: Failed to write app files: {e}")
    raise
  except Exception as e:
    print(f"ERROR: Unexpected error writing files: {e}")
    raise


def verify_setup() -> bool:
  """Verifies the setup is correct by checking for required files."""
  required_files = ["Dockerfile", "package.json", "src/app.js"]
  for file in required_files:
    if not (SAMPLE_APP_DIR / file).exists():
      print(f"ERROR: Missing required file {file}")
      return False
  print("Setup verification passed")
  return True


if __name__ == "__main__":
  print("Starting sample app setup for security scanning tutorial...")
  try:
    create_directory_structure()
    write_app_files()
    if verify_setup():
      print("Sample app setup complete. Next step: Initialize GitLab CI pipeline.")
    else:
      print("Setup failed verification. Check error logs above.")
      exit(1)
  except Exception as e:
    print(f"FATAL: Setup failed: {e}")
    exit(1)

# validate_ci_config.py
# Validates GitLab CI 16.8 pipeline configuration for Trivy 0.50 integration
# Requires Python 3.10+, PyYAML library (pip install pyyaml)

import yaml
import sys
import os
from typing import List, Dict, Any, Optional
from packaging import version  # Requires packaging library: pip install packaging

# Expected configuration constants for GitLab CI 16.8 + Trivy 0.50
MIN_GITLAB_CI_VERSION = "16.8"
TRIVY_TARGET_VERSION = "0.50"
REQUIRED_STAGES = ["build", "security-scan", "report"]
REQUIRED_TRIVY_JOB_KEYS = ["image", "stage", "script", "variables", "artifacts"]


def load_ci_config(config_path: str = ".gitlab-ci.yml") -> Optional[Dict[str, Any]]:
  """Loads and parses the GitLab CI configuration YAML file."""
  try:
    if not os.path.exists(config_path):
      print(f"ERROR: CI config file {config_path} not found.")
      return None
    with open(config_path, "r") as f:
      config = yaml.safe_load(f)
    print(f"Loaded CI config from {config_path}")
    return config
  except yaml.YAMLError as e:
    print(f"ERROR: Failed to parse YAML: {e}")
    return None
  except IOError as e:
    print(f"ERROR: Failed to read file: {e}")
    return None
  except Exception as e:
    print(f"ERROR: Unexpected error loading config: {e}")
    return None


def validate_gitlab_ci_version(config: Dict[str, Any]) -> bool:
  """Validates that the pipeline targets GitLab CI 16.8 or higher."""
  # Note: GitLab CI doesn't have a version key in config, so we validate features
  # GitLab CI 16.8 introduced native SARIF report support in Security Dashboard
  for job_name, job_config in config.items():
    if isinstance(job_config, dict) and "artifacts" in job_config:
      artifacts = job_config["artifacts"]
      if "reports" in artifacts and "sast" in artifacts["reports"]:
        if artifacts["reports"]["sast"].endswith(".sarif"):
          print(f"Validated SARIF support (GitLab CI 16.8+ feature) in job {job_name}")
          return True
  print("ERROR: No SARIF report configuration found (requires GitLab CI 16.8+)")
  return False


def validate_trivy_version(job_config: Dict[str, Any]) -> bool:
  """Validates that the Trivy job uses version 0.50 exactly."""
  try:
    job_image = job_config.get("image", "")
    # Trivy official image is aquasec/trivy:0.50
    if "trivy" not in job_image:
      print(f"ERROR: Job image {job_image} is not a Trivy image")
      return False
    # Extract version from image tag
    image_tag = job_image.split(":")[-1]
    if version.parse(image_tag) != version.parse(TRIVY_TARGET_VERSION):
      print(f"ERROR: Trivy version {image_tag} does not match target {TRIVY_TARGET_VERSION}")
      return False
    print(f"Validated Trivy version {TRIVY_TARGET_VERSION} in job image")
    return True
  except Exception as e:
    print(f"ERROR: Failed to validate Trivy version: {e}")
    return False


def validate_security_stage(config: Dict[str, Any]) -> bool:
  """Validates that the security-scan stage exists and has required jobs."""
  stages = config.get("stages", [])
  if "security-scan" not in stages:
    print("ERROR: Missing required stage 'security-scan'")
    return False
  # Check for at least one Trivy job in security-scan stage
  trivy_jobs = [k for k, v in config.items() if isinstance(v, dict) and v.get("stage") == "security-scan" and "trivy" in v.get("image", "")]
  if not trivy_jobs:
    print("ERROR: No Trivy jobs found in security-scan stage")
    return False
  print(f"Validated security-scan stage with Trivy jobs: {trivy_jobs}")
  return True


if __name__ == "__main__":
  print("Starting GitLab CI configuration validation for Trivy 0.50...")
  try:
    ci_config = load_ci_config()
    if not ci_config:
      print("FATAL: Failed to load CI config")
      sys.exit(1)
    # Run validations
    validations = [
      validate_gitlab_ci_version(ci_config),
      validate_security_stage(ci_config)
    ]
    # Validate Trivy job specifically
    for job_name, job_config in ci_config.items():
      if isinstance(job_config, dict) and "trivy" in job_config.get("image", ""):
        validations.append(validate_trivy_version(job_config))
    if all(validations):
      print("CI configuration validation passed!")
      sys.exit(0)
    else:
      print("CI configuration validation failed. Check errors above.")
      sys.exit(1)
  except Exception as e:
    print(f"FATAL: Validation failed: {e}")
    sys.exit(1)

# run_trivy_scan.py
# Executes Trivy 0.50 container and filesystem scans, parses results, and fails on critical CVEs
# Requires Python 3.10+, Trivy 0.50 installed (https://github.com/aquasecurity/trivy), requests library

import subprocess
import json
import sys
import os
from typing import List, Dict, Any, Optional
from packaging import version

# Configuration
TRIVY_VERSION = "0.50"
MIN_CRITICAL_CVSS = 9.0
SCAN_TARGETS = ["image:my-app:latest", "filesystem:./sample-app"]
SARIF_OUTPUT_PATH = "trivy-report.sarif"
FAIL_ON_SEVERITY = "CRITICAL"


def check_trivy_version() -> bool:
  """Verifies that Trivy 0.50 is installed."""
  try:
    result = subprocess.run(
      ["trivy", "--version"],
      capture_output=True,
      text=True,
      check=True
    )
    installed_version = result.stdout.split("\n")[0].split(" ")[1]
    if version.parse(installed_version) != version.parse(TRIVY_VERSION):
      print(f"ERROR: Trivy version {installed_version} does not match required {TRIVY_VERSION}")
      return False
    print(f"Verified Trivy version {TRIVY_VERSION}")
    return True
  except subprocess.CalledProcessError as e:
    print(f"ERROR: Failed to run Trivy: {e.stderr}")
    return False
  except FileNotFoundError:
    print("ERROR: Trivy not found in PATH. Install Trivy 0.50 first.")
    return False
  except Exception as e:
    print(f"ERROR: Unexpected error checking Trivy version: {e}")
    return False


def run_image_scan(image_tag: str) -> Optional[Dict[str, Any]]:
  """Runs Trivy image scan and returns results as JSON."""
  try:
    print(f"Starting Trivy image scan for {image_tag}...")
    result = subprocess.run(
      [
        "trivy", "image",
        "--severity", "CRITICAL,HIGH,MEDIUM",
        "--format", "json",
        "--output", "-",
        image_tag
      ],
      capture_output=True,
      text=True,
      check=True
    )
    scan_results = json.loads(result.stdout)
    print(f"Completed image scan for {image_tag}: {len(scan_results.get('Results', []))} results")
    return scan_results
  except subprocess.CalledProcessError as e:
    print(f"ERROR: Trivy image scan failed: {e.stderr}")
    return None
  except json.JSONDecodeError as e:
    print(f"ERROR: Failed to parse Trivy JSON output: {e}")
    return None
  except Exception as e:
    print(f"ERROR: Unexpected error during image scan: {e}")
    return None


def run_filesystem_scan(target_path: str) -> Optional[Dict[str, Any]]:
  """Runs Trivy filesystem scan and returns results as JSON."""
  try:
    print(f"Starting Trivy filesystem scan for {target_path}...")
    result = subprocess.run(
      [
        "trivy", "fs",
        "--severity", "CRITICAL,HIGH,MEDIUM",
        "--format", "json",
        "--output", "-",
        target_path
      ],
      capture_output=True,
      text=True,
      check=True
    )
    scan_results = json.loads(result.stdout)
    print(f"Completed filesystem scan for {target_path}: {len(scan_results.get('Results', []))} results")
    return scan_results
  except subprocess.CalledProcessError as e:
    print(f"ERROR: Trivy filesystem scan failed: {e.stderr}")
    return None
  except json.JSONDecodeError as e:
    print(f"ERROR: Failed to parse Trivy JSON output: {e}")
    return None
  except Exception as e:
    print(f"ERROR: Unexpected error during filesystem scan: {e}")
    return None


def generate_sarif_report(scan_results: List[Dict[str, Any]]) -> bool:
  """Generates a SARIF report from Trivy scan results."""
  try:
    print(f"Generating SARIF report at {SARIF_OUTPUT_PATH}...")
    # Run Trivy again with SARIF output
    with open(SARIF_OUTPUT_PATH, "w") as f:
      subprocess.run(
        ["trivy", "image", "--severity", "CRITICAL,HIGH", "--format", "sarif", "my-app:latest"],
        stdout=f,
        check=True
      )
    print(f"SARIF report generated at {SARIF_OUTPUT_PATH}")
    return True
  except Exception as e:
    print(f"ERROR: Failed to generate SARIF report: {e}")
    return False


def check_critical_vulns(scan_results: List[Dict[str, Any]]) -> bool:
  """Checks for critical vulnerabilities with CVSS ≥ MIN_CRITICAL_CVSS."""
  critical_count = 0
  for result in scan_results:
    for vuln in result.get("Vulnerabilities", []):
      cvss_score = vuln.get("CVSS", {}).get("nvd", {}).get("V3Score", 0)
      if vuln.get("Severity") == "CRITICAL" and cvss_score >= MIN_CRITICAL_CVSS:
        critical_count += 1
        print(f"CRITICAL VULN: {vuln.get('VulnerabilityID')} in {vuln.get('PkgName')} (CVSS: {cvss_score})")
  if critical_count > 0:
    print(f"ERROR: Detected {critical_count} critical vulnerabilities (CVSS ≥ {MIN_CRITICAL_CVSS})")
    return False
  print("No critical vulnerabilities detected")
  return True


if __name__ == "__main__":
  print("Starting Trivy 0.50 security scan workflow...")
  try:
    # Check Trivy version
    if not check_trivy_version():
      sys.exit(1)
    # Run scans
    image_results = run_image_scan("my-app:latest")
    fs_results = run_filesystem_scan("./sample-app")
    all_results = []
    if image_results:
      all_results.append(image_results)
    if fs_results:
      all_results.append(fs_results)
    if not all_results:
      print("ERROR: No scan results generated")
      sys.exit(1)
    # Generate SARIF report
    if not generate_sarif_report(all_results):
      sys.exit(1)
    # Check for critical vulns
    if not check_critical_vulns(all_results):
      sys.exit(1)
    print("Trivy scan completed successfully")
  except Exception as e:
    print(f"FATAL: Trivy scan failed: {e}")
    sys.exit(1)

# run_grype_scan.py
# Executes Grype 0.70 container and filesystem scans, parses results, and fails on critical CVEs
# Requires Python 3.10+, Grype 0.70 installed (https://github.com/anchore/grype), requests library

import subprocess
import json
import sys
import os
from typing import List, Dict, Any, Optional
from packaging import version

# Configuration
GRYPE_VERSION = "0.70"
MIN_CRITICAL_CVSS = 9.0
SCAN_TARGETS = ["my-app:latest", "./sample-app"]
SARIF_OUTPUT_PATH = "grype-report.sarif"
FAIL_ON_SEVERITY = "critical"


def check_grype_version() -> bool:
  """Verifies that Grype 0.70 is installed."""
  try:
    result = subprocess.run(
      ["grype", "--version"],
      capture_output=True,
      text=True,
      check=True
    )
    installed_version = result.stdout.split("\n")[0].split(" ")[1].strip()
    if version.parse(installed_version) != version.parse(GRYPE_VERSION):
      print(f"ERROR: Grype version {installed_version} does not match required {GRYPE_VERSION}")
      return False
    print(f"Verified Grype version {GRYPE_VERSION}")
    return True
  except subprocess.CalledProcessError as e:
    print(f"ERROR: Failed to run Grype: {e.stderr}")
    return False
  except FileNotFoundError:
    print("ERROR: Grype not found in PATH. Install Grype 0.70 first.")
    return False
  except Exception as e:
    print(f"ERROR: Unexpected error checking Grype version: {e}")
    return False


def run_image_scan(image_tag: str) -> Optional[Dict[str, Any]]:
  """Runs Grype image scan and returns results as JSON."""
  try:
    print(f"Starting Grype image scan for {image_tag}...")
    result = subprocess.run(
      [
        "grype", image_tag,
        "-o", "json",
        "--fail-on", FAIL_ON_SEVERITY
      ],
      capture_output=True,
      text=True,
      check=False  # Don't fail immediately, parse results even if critical found
    )
    scan_results = json.loads(result.stdout)
    print(f"Completed image scan for {image_tag}: {len(scan_results.get('matches', []))} vulnerabilities")
    return scan_results
  except subprocess.CalledProcessError as e:
    print(f"ERROR: Grype image scan failed: {e.stderr}")
    return None
  except json.JSONDecodeError as e:
    print(f"ERROR: Failed to parse Grype JSON output: {e}")
    return None
  except Exception as e:
    print(f"ERROR: Unexpected error during image scan: {e}")
    return None


def run_filesystem_scan(target_path: str) -> Optional[Dict[str, Any]]:
  """Runs Grype filesystem scan and returns results as JSON."""
  try:
    print(f"Starting Grype filesystem scan for {target_path}...")
    result = subprocess.run(
      [
        "grype", f"dir:{target_path}",
        "-o", "json",
        "--fail-on", FAIL_ON_SEVERITY
      ],
      capture_output=True,
      text=True,
      check=False
    )
    scan_results = json.loads(result.stdout)
    print(f"Completed filesystem scan for {target_path}: {len(scan_results.get('matches', []))} vulnerabilities")
    return scan_results
  except subprocess.CalledProcessError as e:
    print(f"ERROR: Grype filesystem scan failed: {e.stderr}")
    return None
  except json.JSONDecodeError as e:
    print(f"ERROR: Failed to parse Grype JSON output: {e}")
    return None
  except Exception as e:
    print(f"ERROR: Unexpected error during filesystem scan: {e}")
    return None


def generate_sarif_report(scan_results: Dict[str, Any]) -> bool:
  """Generates a SARIF report from Grype scan results."""
  try:
    print(f"Generating SARIF report at {SARIF_OUTPUT_PATH}...")
    # Run Grype again with SARIF output
    with open(SARIF_OUTPUT_PATH, "w") as f:
      subprocess.run(
        ["grype", "my-app:latest", "-o", "sarif"],
        stdout=f,
        check=True
      )
    print(f"SARIF report generated at {SARIF_OUTPUT_PATH}")
    return True
  except Exception as e:
    print(f"ERROR: Failed to generate SARIF report: {e}")
    return False


def check_critical_vulns(scan_results: Dict[str, Any]) -> bool:
  """Checks for critical vulnerabilities with CVSS ≥ MIN_CRITICAL_CVSS."""
  critical_count = 0
  for match in scan_results.get("matches", []):
    vuln = match.get("vulnerability", {})
    cvss_score = vuln.get("cvss", {}).get("nvd", {}).get("v3Score", 0)
    severity = vuln.get("severity", "")
    if severity == "critical" and cvss_score >= MIN_CRITICAL_CVSS:
      critical_count += 1
      print(f"CRITICAL VULN: {vuln.get('id')} in {match.get('artifact', {}).get('name')} (CVSS: {cvss_score})")
  if critical_count > 0:
    print(f"ERROR: Detected {critical_count} critical vulnerabilities (CVSS ≥ {MIN_CRITICAL_CVSS})")
    return False
  print("No critical vulnerabilities detected")
  return True


if __name__ == "__main__":
  print("Starting Grype 0.70 security scan workflow...")
  try:
    # Check Grype version
    if not check_grype_version():
      sys.exit(1)
    # Run scans
    image_results = run_image_scan("my-app:latest")
    fs_results = run_filesystem_scan("./sample-app")
    # Process results
    if not image_results and not fs_results:
      print("ERROR: No scan results generated")
      sys.exit(1)
    # Generate SARIF report
    if image_results and not generate_sarif_report(image_results):
      sys.exit(1)
    # Check for critical vulns
    all_critical = True
    if image_results and not check_critical_vulns(image_results):
      all_critical = False
    if fs_results and not check_critical_vulns(fs_results):
      all_critical = False
    if not all_critical:
      sys.exit(1)
    print("Grype scan completed successfully")
  except Exception as e:
    print(f"FATAL: Grype scan failed: {e}")
    sys.exit(1)

cache:
  key: "${CI_JOB_NAME}-${TRIVY_VERSION}-${GRYPE_VERSION}-${CI_OS_FAMILY}"
  paths:
    - ~/.cache/trivy
    - ~/.cache/grype

security-scan:
  stage: security-scan
  parallel:
    matrix:
      - SCANNER: trivy
      - SCANNER: grype
  script:
    - run-${SCANNER}-scan.py

trivy-scan:
  script:
    - if [ "$CI_ENVIRONMENT_NAME" = "production" ]; then SEVERITY="CRITICAL"; else SEVERITY="HIGH,CRITICAL"; fi
    - trivy image --severity $SEVERITY --exit-code 1 my-app:latest

gitlab-ci-scanning-tutorial/
├── sample-app/
│   ├── Dockerfile
│   ├── package.json
│   └── src/
│       └── app.js
├── scripts/
│   ├── setup_project.py
│   ├── validate_ci_config.py
│   ├── run_trivy_scan.py
│   └── run_grype_scan.py
├── .gitlab-ci.yml
├── .trivyignore
├── .grypeignore
└── README.md