How to Integrate Trivy 0.50 with Harbor 2.10 for Private Registry Scanning

import requests
import json
import os
import sys
import logging
from packaging import version  # Requires packaging>=23.0

# Configure logging for actionable output
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Configuration constants - update these for your environment
HARBOR_API_ENDPOINT = os.getenv("HARBOR_API_ENDPOINT", "https://harbor.example.com/api/v2.0")
HARBOR_ADMIN_USER = os.getenv("HARBOR_ADMIN_USER", "admin")
HARBOR_ADMIN_PASSWORD = os.getenv("HARBOR_ADMIN_PASSWORD", "Harbor12345")
MIN_HARBOR_VERSION = "2.10.0"
TRIVY_CLI_PATH = os.getenv("TRIVY_CLI_PATH", "/usr/local/bin/trivy")
MIN_TRIVY_VERSION = "0.50.0"

def check_harbor_version():
    """Validate Harbor is running 2.10.0 or later and API is accessible."""
    try:
        response = requests.get(
            f"{HARBOR_API_ENDPOINT}/systeminfo",
            auth=(HARBOR_ADMIN_USER, HARBOR_ADMIN_PASSWORD),
            timeout=10
        )
        response.raise_for_status()
        system_info = response.json()
        harbor_version = system_info.get("version", "0.0.0")
        logger.info(f"Detected Harbor version: {harbor_version}")

        if version.parse(harbor_version) < version.parse(MIN_HARBOR_VERSION):
            logger.error(f"Harbor version {harbor_version} is below minimum required {MIN_HARBOR_VERSION}")
            sys.exit(1)
        return harbor_version
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to connect to Harbor API: {e}")
        sys.exit(1)

def check_trivy_installation():
    """Validate Trivy 0.50.0 or later is installed and accessible."""
    try:
        import subprocess
        result = subprocess.run(
            [TRIVY_CLI_PATH, "version", "--format", "json"],
            capture_output=True,
            text=True,
            timeout=10
        )
        result.check_returncode()
        trivy_info = json.loads(result.stdout)
        trivy_version = trivy_info.get("Version", "0.0.0")
        logger.info(f"Detected Trivy version: {trivy_version}")

        if version.parse(trivy_version) < version.parse(MIN_TRIVY_VERSION):
            logger.error(f"Trivy version {trivy_version} is below minimum required {MIN_TRIVY_VERSION}")
            sys.exit(1)
        return trivy_version
    except (subprocess.CalledProcessError, FileNotFoundError) as e:
        logger.error(f"Trivy not found or failed to execute: {e}")
        sys.exit(1)

def validate_scan_service_api():
    """Check if Harbor's Scan Service API is enabled and accessible."""
    try:
        response = requests.get(
            f"{HARBOR_API_ENDPOINT}/scanners",
            auth=(HARBOR_ADMIN_USER, HARBOR_ADMIN_PASSWORD),
            timeout=10
        )
        response.raise_for_status()
        logger.info("Harbor Scan Service API is accessible")
        return True
    except requests.exceptions.RequestException as e:
        logger.error(f"Harbor Scan Service API is not accessible: {e}")
        sys.exit(1)

if __name__ == "__main__":
    logger.info("Starting pre-integration validation for Trivy 0.50 + Harbor 2.10")
    harbor_version = check_harbor_version()
    trivy_version = check_trivy_installation()
    validate_scan_service_api()
    logger.info(f"All pre-checks passed. Harbor {harbor_version}, Trivy {trivy_version}")

import requests
import json
import os
import sys
import logging
import uuid
from packaging import version

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Configuration - update these values
HARBOR_API_ENDPOINT = os.getenv("HARBOR_API_ENDPOINT", "https://harbor.example.com/api/v2.0")
HARBOR_ADMIN_USER = os.getenv("HARBOR_ADMIN_USER", "admin")
HARBOR_ADMIN_PASSWORD = os.getenv("HARBOR_ADMIN_PASSWORD", "Harbor12345")
TRIVY_ADAPTER_URL = os.getenv("TRIVY_ADAPTER_URL", "http://harbor-scanner-trivy:8080")
TRIVY_ADAPTER_NAME = "Trivy-0.50"
TRIVY_ADAPTER_VERSION = "0.50.1"
TRIVY_ADAPTER_DESCRIPTION = "Trivy 0.50 vulnerability scanner for Harbor"

def generate_adapter_uuid():
    """Generate a UUID for the scanner adapter registration."""
    return str(uuid.uuid4())

def check_existing_scanners():
    """Check if Trivy adapter is already registered to avoid duplicates."""
    try:
        response = requests.get(
            f"{HARBOR_API_ENDPOINT}/scanners",
            auth=(HARBOR_ADMIN_USER, HARBOR_ADMIN_PASSWORD),
            timeout=10
        )
        response.raise_for_status()
        scanners = response.json()
        for scanner in scanners:
            if scanner.get("name") == TRIVY_ADAPTER_NAME:
                logger.info(f"Trivy adapter already registered with ID: {scanner.get('id')}")
                return scanner.get("id")
        return None
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to fetch existing scanners: {e}")
        sys.exit(1)

def register_trivy_adapter():
    """Register Trivy scanner adapter with Harbor's Scan Service API."""
    existing_id = check_existing_scanners()
    if existing_id:
        logger.info(f"Skipping registration, adapter already exists with ID: {existing_id}")
        return existing_id

    adapter_payload = {
        "uuid": generate_adapter_uuid(),
        "name": TRIVY_ADAPTER_NAME,
        "description": TRIVY_ADAPTER_DESCRIPTION,
        "url": TRIVY_ADAPTER_URL,
        "version": TRIVY_ADAPTER_VERSION,
        "metadata": {
            "capabilities": [
                {
                    "type": "vulnerability",
                    "features": ["os", "library", "secret", "config"]
                }
            ]
        }
    }

    try:
        response = requests.post(
            f"{HARBOR_API_ENDPOINT}/scanners",
            auth=(HARBOR_ADMIN_USER, HARBOR_ADMIN_PASSWORD),
            json=adapter_payload,
            timeout=10
        )
        response.raise_for_status()
        adapter_id = response.json().get("id")
        logger.info(f"Successfully registered Trivy adapter with ID: {adapter_id}")
        return adapter_id
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to register Trivy adapter: {e}")
        if hasattr(e, "response") and e.response is not None:
            logger.error(f"API response: {e.response.text}")
        sys.exit(1)

def set_default_scanner(adapter_id):
    """Set Trivy as the default scanner for all Harbor projects."""
    try:
        response = requests.patch(
            f"{HARBOR_API_ENDPOINT}/scanners/{adapter_id}/default",
            auth=(HARBOR_ADMIN_USER, HARBOR_ADMIN_PASSWORD),
            timeout=10
        )
        response.raise_for_status()
        logger.info(f"Set Trivy adapter {adapter_id} as default scanner")
        return True
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to set default scanner: {e}")
        sys.exit(1)

def validate_adapter_health(adapter_id):
    """Check if the registered Trivy adapter is healthy and responding."""
    try:
        response = requests.get(
            f"{HARBOR_API_ENDPOINT}/scanners/{adapter_id}/health",
            auth=(HARBOR_ADMIN_USER, HARBOR_ADMIN_PASSWORD),
            timeout=10
        )
        response.raise_for_status()
        health_status = response.json().get("status", "unknown")
        if health_status != "healthy":
            logger.error(f"Trivy adapter health status: {health_status}")
            sys.exit(1)
        logger.info(f"Trivy adapter health status: {health_status}")
        return True
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to check adapter health: {e}")
        sys.exit(1)

if __name__ == "__main__":
    logger.info("Starting Trivy adapter registration with Harbor")
    adapter_id = register_trivy_adapter()
    set_default_scanner(adapter_id)
    validate_adapter_health(adapter_id)
    logger.info("Trivy adapter registration completed successfully")

import requests
import json
import os
import sys
import logging
from http.server import HTTPServer, BaseHTTPRequestHandler
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Configuration
WEBHOOK_PORT = int(os.getenv("WEBHOOK_PORT", 8080))
SLACK_TOKEN = os.getenv("SLACK_TOKEN", "xoxb-your-slack-token")
SLACK_CHANNEL = os.getenv("SLACK_CHANNEL", "#security-alerts")
HARBOR_API_ENDPOINT = os.getenv("HARBOR_API_ENDPOINT", "https://harbor.example.com/api/v2.0")
HARBOR_ADMIN_USER = os.getenv("HARBOR_ADMIN_USER", "admin")
HARBOR_ADMIN_PASSWORD = os.getenv("HARBOR_ADMIN_PASSWORD", "Harbor12345")
CRITICAL_CVSS_THRESHOLD = 9.0
HIGH_CVSS_THRESHOLD = 7.0

# Initialize Slack client
slack_client = WebClient(token=SLACK_TOKEN)

class HarborWebhookHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        """Handle POST requests from Harbor webhooks."""
        content_length = int(self.headers.get("Content-Length", 0))
        post_data = self.rfile.read(content_length)
        try:
            webhook_payload = json.loads(post_data)
            logger.info(f"Received webhook event: {webhook_payload.get('type')}")
            self.handle_scan_completed(webhook_payload)
            self.send_response(200)
            self.end_headers()
        except json.JSONDecodeError as e:
            logger.error(f"Failed to parse webhook payload: {e}")
            self.send_response(400)
            self.end_headers()
        except Exception as e:
            logger.error(f"Error processing webhook: {e}")
            self.send_response(500)
            self.end_headers()

    def handle_scan_completed(self, payload):
        """Process scan completed events from Harbor."""
        if payload.get("type") != "SCAN_COMPLETED":
            logger.info(f"Ignoring non-scan event: {payload.get('type')}")
            return

        scan_data = payload.get("event_data", {})
        image_name = scan_data.get("repository", {}).get("name")
        image_tag = scan_data.get("tag", {}).get("name")
        scan_id = scan_data.get("scan_id")
        logger.info(f"Processing scan {scan_id} for {image_name}:{image_tag}")

        # Fetch full scan report from Harbor API
        try:
            response = requests.get(
                f"{HARBOR_API_ENDPOINT}/projects/{scan_data.get('project', {}).get('name')}/repositories/{image_name}/artifacts/{image_tag}/scan/{scan_id}/report",
                auth=(HARBOR_ADMIN_USER, HARBOR_ADMIN_PASSWORD),
                timeout=10
            )
            response.raise_for_status()
            scan_report = response.json()
            self.process_trivy_report(scan_report, image_name, image_tag)
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to fetch scan report {scan_id}: {e}")

    def process_trivy_report(self, report, image_name, image_tag):
        """Parse Trivy scan report and send Slack alerts for high/critical CVEs."""
        vulnerabilities = report.get("vulnerabilities", [])
        critical_vulns = []
        high_vulns = []

        for vuln in vulnerabilities:
            cvss_score = vuln.get("cvss_score", 0.0)
            if cvss_score >= CRITICAL_CVSS_THRESHOLD:
                critical_vulns.append(vuln)
            elif cvss_score >= HIGH_CVSS_THRESHOLD:
                high_vulns.append(vuln)

        if not critical_vulns and not high_vulns:
            logger.info(f"No high/critical vulnerabilities found for {image_name}:{image_tag}")
            return

        # Build Slack message
        message_blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"Vulnerabilities Found: {image_name}:{image_tag}"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Critical CVEs:* {len(critical_vulns)}\n*High CVEs:* {len(high_vulns)}"
                }
            }
        ]

        # Add top 5 critical vulns
        for vuln in critical_vulns[:5]:
            message_blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"• *{vuln.get('id')}* (CVSS {vuln.get('cvss_score')}): {vuln.get('description')[:100]}... Fix version: {vuln.get('fix_version', 'None')}"
                }
            })

        # Send to Slack
        try:
            slack_client.chat_postMessage(
                channel=SLACK_CHANNEL,
                blocks=message_blocks,
                text=f"Vulnerabilities found in {image_name}:{image_tag}"
            )
            logger.info(f"Sent Slack alert for {image_name}:{image_tag}")
        except SlackApiError as e:
            logger.error(f"Failed to send Slack message: {e}")

if __name__ == "__main__":
    logger.info(f"Starting Harbor webhook handler on port {WEBHOOK_PORT}")
    server = HTTPServer(("0.0.0.0", WEBHOOK_PORT), HarborWebhookHandler)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        logger.info("Shutting down webhook handler")
        server.shutdown()

import boto3
import subprocess
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

S3_BUCKET = "trivy-db-cache"
S3_PREFIX = "vulndb"
TRIVY_DB_PATH = "/tmp/trivy-db"

def update_trivy_db():
    """Update local Trivy DB and push to S3 cache."""
    try:
        subprocess.run(["trivy", "vuln", "--download-db-only", "--cache-dir", TRIVY_DB_PATH], check=True)
        logger.info("Trivy DB updated successfully")
    except subprocess.CalledProcessError as e:
        logger.error(f"Failed to update Trivy DB: {e}")
        raise

def upload_db_to_s3():
    """Upload updated Trivy DB to S3 for distributed access."""
    s3 = boto3.client("s3")
    timestamp = datetime.utcnow().strftime("%Y%m%d%H%M%S")
    try:
        import shutil
        shutil.make_archive(f"/tmp/trivy-db-{timestamp}", "zip", TRIVY_DB_PATH)
        s3.upload_file(f"/tmp/trivy-db-{timestamp}.zip", S3_BUCKET, f"{S3_PREFIX}/{timestamp}.zip")
        logger.info(f"Uploaded Trivy DB to s3://{S3_BUCKET}/{S3_PREFIX}/{timestamp}.zip")
    except Exception as e:
        logger.error(f"Failed to upload DB to S3: {e}")
        raise

if __name__ == "__main__":
    update_trivy_db()
    upload_db_to_s3()

package harbor.admission

import future.keywords.if
import future.keywords.in

# Deny if image has critical vulnerability and is not a hotfix
deny[msg] if {
    input.request.operation == "CREATE"
    scan_report = input.request.object.scan_report
    vuln = scan_report.vulnerabilities[_]
    vuln.cvss_score >= 9.0
    not input.request.object.tag in ["hotfix", "hotfix-*"]
    msg := sprintf("Critical CVE %s (CVSS %f) found in %s:%s", [vuln.id, vuln.cvss_score, input.request.object.repository, input.request.object.tag])
}

# Allow all other requests
allow if {
    not deny[_]
}

import requests
import pandas as pd
import os
from datetime import datetime, timedelta

HARBOR_API = os.getenv("HARBOR_API_ENDPOINT")
HARBOR_USER = os.getenv("HARBOR_ADMIN_USER")
HARBOR_PASS = os.getenv("HARBOR_ADMIN_PASSWORD")

def fetch_scan_reports(days=7):
    """Fetch all scan reports from Harbor for the last N days."""
    reports = []
    projects = requests.get(f"{HARBOR_API}/projects", auth=(HARBOR_USER, HARBOR_PASS)).json()
    for project in projects:
        project_name = project["name"]
        repos = requests.get(f"{HARBOR_API}/projects/{project_name}/repositories", auth=(HARBOR_USER, HARBOR_PASS)).json()
        for repo in repos:
            artifacts = requests.get(f"{HARBOR_API}/projects/{project_name}/repositories/{repo['name']}/artifacts", auth=(HARBOR_USER, HARBOR_PASS)).json()
            for artifact in artifacts:
                if datetime.fromtimestamp(artifact["push_time"]) > datetime.now() - timedelta(days=days):
                    scan_report = requests.get(f"{HARBOR_API}/projects/{project_name}/repositories/{repo['name']}/artifacts/{artifact['digest']}/scan/report", auth=(HARBOR_USER, HARBOR_PASS)).json()
                    reports.append({
                        "project": project_name,
                        "repo": repo["name"],
                        "tag": artifact["tags"][0]["name"] if artifact.get("tags") else "latest",
                        "vuln_count": len(scan_report.get("vulnerabilities", [])),
                        "critical_count": len([v for v in scan_report.get("vulnerabilities", []) if v["cvss_score"] >= 9.0]),
                        "scan_date": datetime.fromtimestamp(artifact["push_time"])
                    })
    return reports

def generate_compliance_report():
    """Generate weekly compliance report from scan data."""
    reports = fetch_scan_reports(days=7)
    df = pd.DataFrame(reports)
    summary = df.groupby("project").agg({
        "vuln_count": "sum",
        "critical_count": "sum",
        "repo": "nunique"
    }).reset_index()
    summary.to_csv(f"compliance_report_{datetime.now().strftime('%Y%m%d')}.csv", index=False)
    logger.info(f"Generated compliance report with {len(summary)} projects")

if __name__ == "__main__":
    generate_compliance_report()