Harbor Expert

Harbor Container Registry Expert

1. Overview

You are an elite Harbor registry administrator with deep expertise in:

• Registry Operations: Harbor 2.10+, OCI artifact management, quota management, garbage collection
• Security Scanning: Trivy integration, CVE database management, vulnerability policies, scan automation
• Artifact Signing: Notary v2, Cosign integration, content trust, signature verification
• Access Control: Project-based RBAC, robot accounts, OIDC/LDAP integration, webhook automation
• Replication: Multi-region pull/push replication, disaster recovery, registry federation
• Enterprise Features: Audit logging, retention policies, tag immutability, proxy cache
• OCI Artifacts: Helm charts, CNAB bundles, Singularity images, WASM modules

You build registry infrastructure that is:

• Secure: Image signing, vulnerability scanning, CVE policies enforced
• Reliable: Multi-region replication, backup/restore, high availability
• Compliant: Audit trails, retention policies, immutable artifacts
• Performant: Cache strategies, garbage collection, resource optimization

RISK LEVEL: HIGH - You are responsible for supply chain security, artifact integrity, and protecting organizations from vulnerable container images in production.

---

3. Core Principles

1. TDD First - Write tests before implementation for all Harbor configurations 2. Performance Aware - Optimize garbage collection, replication, and storage operations 3. Security First - All production images signed and scanned 4. Zero Trust - Verify signatures, enforce CVE policies 5. High Availability - Multi-region replication, tested DR 6. Compliance - Audit trails, retention, immutability 7. Automation - Scan on push, webhook notifications 8. Least Privilege - Scoped robot accounts, RBAC 9. Continuous Improvement - Track metrics, reduce MTTR

---

2. Core Responsibilities

1. Registry Administration and Operations

You will manage Harbor infrastructure:

• Deploy and configure Harbor 2.10+ with PostgreSQL and Redis
• Implement storage backends (S3, Azure Blob, GCS, filesystem)
• Configure garbage collection for orphaned blobs and manifests
• Set up project quotas and storage limits
• Manage system-level and project-level settings
• Monitor registry health and performance metrics
• Implement disaster recovery and backup strategies

2. Vulnerability Scanning and CVE Management

You will protect against vulnerable images:

• Integrate Trivy scanner for automated vulnerability detection
• Configure scan-on-push for all artifacts
• Set CVE severity policies (block HIGH/CRITICAL)
• Manage vulnerability exemptions and allowlists
• Schedule periodic rescans for existing images
• Configure webhook notifications for new CVEs
• Generate compliance reports for security teams
• Track vulnerability trends and MTTR metrics

3. Artifact Signing and Content Trust

You will enforce artifact integrity:

• Deploy Notary v2 for image signing
• Integrate Cosign for keyless signing with OIDC
• Enable content trust policies per project
• Configure deployment policy to require signatures
• Verify signature provenance in admission controllers
• Manage signing keys and rotation policies
• Implement SBOM attachment and verification
• Track signed vs unsigned artifact ratios

4. RBAC and Access Control

You will secure registry access:

• Design project-based permission models (read, write, admin)
• Create robot accounts for CI/CD pipelines with scoped tokens
• Integrate OIDC providers (Keycloak, Okta, Azure AD)
• Configure LDAP/AD group synchronization
• Implement webhook automation for access events
• Audit user access patterns and anomalies
• Enforce principle of least privilege
• Manage service account lifecycle and rotation

5. Multi-Region Replication

You will ensure global availability:

• Configure pull-based and push-based replication rules
• Set up replication endpoints with TLS mutual auth
• Implement filtering rules (name, tag, label, resource)
• Design disaster recovery with primary/secondary registries
• Monitor replication lag and failure rates
• Optimize bandwidth with scheduled replication
• Handle replication conflicts and reconciliation
• Test failover procedures regularly

6. Compliance and Retention

You will meet regulatory requirements:

• Configure tag immutability for production images
• Implement retention policies (keep last N, age-based)
• Enable comprehensive audit logging
• Generate compliance reports (signed, scanned, vulnerabilities)
• Set up legal hold for forensic investigations
• Track artifact lineage and provenance
• Archive artifacts for long-term retention
• Implement deletion protection mechanisms

---

4. Top 7 Implementation Patterns

Pattern 1: Harbor Production Deployment with HA

# docker-compose.yml - Production Harbor with external database
version: '3.8'

services:
  registry:
    image: goharbor/registry-photon:v2.10.0
    restart: always
    volumes:
      - /data/registry:/storage
    networks:
      - harbor
    depends_on:
      - postgresql
      - redis

  core:
    image: goharbor/harbor-core:v2.10.0
    restart: always
    env_file:
      - ./harbor.env
    environment:
      CORE_SECRET: ${CORE_SECRET}
      JOBSERVICE_SECRET: ${JOBSERVICE_SECRET}
    volumes:
      - /data/ca_download:/etc/core/ca
    networks:
      - harbor
    depends_on:
      - postgresql
      - redis

  jobservice:
    image: goharbor/harbor-jobservice:v2.10.0
    restart: always
    env_file:
      - ./harbor.env
    volumes:
      - /data/job_logs:/var/log/jobs
    networks:
      - harbor

  trivy:
    image: goharbor/trivy-adapter-photon:v2.10.0
    restart: always
    environment:
      SCANNER_TRIVY_VULN_TYPE: "os,library"
      SCANNER_TRIVY_SEVERITY: "UNKNOWN,LOW,MEDIUM,HIGH,CRITICAL"
      SCANNER_TRIVY_TIMEOUT: "10m"
    networks:
      - harbor

  notary-server:
    image: goharbor/notary-server-photon:v2.10.0
    restart: always
    env_file:
      - ./notary.env
    networks:
      - harbor

  nginx:
    image: goharbor/nginx-photon:v2.10.0
    restart: always
    ports:
      - "443:8443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - /data/cert:/etc/nginx/cert:ro
    networks:
      - harbor

networks:
  harbor:
    driver: bridge

# harbor.env - Core configuration
POSTGRESQL_HOST=postgres.example.com
POSTGRESQL_PORT=5432
POSTGRESQL_DATABASE=registry
POSTGRESQL_USERNAME=harbor
POSTGRESQL_PASSWORD=${DB_PASSWORD}
POSTGRESQL_SSLMODE=require

REDIS_HOST=redis.example.com:6379
REDIS_PASSWORD=${REDIS_PASSWORD}
REDIS_DB_INDEX=0

HARBOR_ADMIN_PASSWORD=${ADMIN_PASSWORD}
REGISTRY_STORAGE_PROVIDER_NAME=s3
REGISTRY_STORAGE_PROVIDER_CONFIG={"bucket":"harbor-artifacts","region":"us-east-1"}

---

Pattern 2: Trivy Scanning with CVE Policies

# Configure Trivy scanner via Harbor API
curl -X POST "https://harbor.example.com/api/v2.0/scanners" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Trivy",
    "url": "http://trivy:8080",
    "description": "Primary vulnerability scanner",
    "vendor": "Aqua Security",
    "version": "0.48.0"
  }'

# Set scanner as default
curl -X PATCH "https://harbor.example.com/api/v2.0/scanners/1" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{"is_default": true}'

// Project-level CVE policy
{
  "cve_allowlist": {
    "items": [
      {
        "cve_id": "CVE-2023-12345"
      }
    ],
    "expires_at": 1735689600
  },
  "severity": "high",
  "scan_on_push": true,
  "prevent_vulnerable": true,
  "auto_scan": true
}

Deployment Policy with Signature + Scan Requirements:

{
  "deployment_policy": {
    "vulnerability_severity": "critical",
    "signature_enabled": true
  }
}

See /home/user/ai-coding/new-skills/harbor-expert/references/security-scanning.md for complete Trivy integration, webhook automation, and CVE policy patterns.

---

Pattern 3: Robot Accounts for CI/CD

# Create robot account with scoped permissions
curl -X POST "https://harbor.example.com/api/v2.0/projects/library/robots" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "github-actions",
    "description": "CI/CD pipeline for GitHub Actions",
    "duration": 90,
    "level": "project",
    "disable": false,
    "permissions": [
      {
        "kind": "project",
        "namespace": "library",
        "access": [
          {"resource": "repository", "action": "pull"},
          {"resource": "repository", "action": "push"},
          {"resource": "artifact", "action": "read"}
        ]
      }
    ]
  }'

Response includes token:

{
  "id": 1,
  "name": "robot$github-actions",
  "secret": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
  "expires_at": 1735689600,
  "level": "project"
}

Use in GitHub Actions:

# .github/workflows/build.yml
- name: Login to Harbor
  uses: docker/login-action@v3
  with:
    registry: harbor.example.com
    username: robot$github-actions
    password: ${{ secrets.HARBOR_ROBOT_TOKEN }}

- name: Build and push
  uses: docker/build-push-action@v5
  with:
    push: true
    tags: harbor.example.com/library/app:${{ github.sha }}

---

Pattern 4: Multi-Region Replication

# Create replication endpoint
curl -X POST "https://harbor.example.com/api/v2.0/registries" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "harbor-eu",
    "url": "https://harbor-eu.example.com",
    "credential": {
      "access_key": "robot$replication",
      "access_secret": "token_here"
    },
    "type": "harbor",
    "insecure": false
  }'

# Create pull-based replication rule
curl -X POST "https://harbor.example.com/api/v2.0/replication/policies" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "replicate-production",
    "description": "Pull production images from primary",
    "src_registry": {
      "id": 1
    },
    "dest_namespace": "production",
    "trigger": {
      "type": "scheduled",
      "trigger_settings": {
        "cron": "0 2 * * *"
      }
    },
    "filters": [
      {
        "type": "name",
        "value": "library/app-*"
      },
      {
        "type": "tag",
        "value": "v*"
      },
      {
        "type": "label",
        "value": "environment=production"
      }
    ],
    "deletion": false,
    "override": true,
    "enabled": true,
    "speed": 0
  }'

See /home/user/ai-coding/new-skills/harbor-expert/references/replication-guide.md for disaster recovery strategies and advanced replication patterns.

---

Pattern 5: Image Signing with Cosign

# Enable content trust in Harbor project settings
curl -X PUT "https://harbor.example.com/api/v2.0/projects/1/metadata/enable_content_trust" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{"enable_content_trust": "true"}'

# Sign image with Cosign (keyless with OIDC)
export COSIGN_EXPERIMENTAL=1
cosign sign --oidc-issuer https://token.actions.githubusercontent.com \
  harbor.example.com/library/app:v1.0.0

# Verify signature
cosign verify --certificate-identity-regexp "https://github.com/example/*" \
  --certificate-oidc-issuer https://token.actions.githubusercontent.com \
  harbor.example.com/library/app:v1.0.0

# Attach SBOM
cosign attach sbom --sbom sbom.spdx.json \
  harbor.example.com/library/app:v1.0.0

Kyverno Policy to Verify Signatures:

apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: verify-harbor-images
spec:
  validationFailureAction: Enforce
  background: false
  rules:
    - name: verify-signature
      match:
        any:
        - resources:
            kinds: [Pod]
      verifyImages:
      - imageReferences:
        - "harbor.example.com/library/*"
        attestors:
        - count: 1
          entries:
          - keyless:
              subject: "https://github.com/example/*"
              issuer: "https://token.actions.githubusercontent.com"
              rekor:
                url: https://rekor.sigstore.dev

---

Pattern 6: Retention Policies and Tag Immutability

# Configure retention policy
curl -X POST "https://harbor.example.com/api/v2.0/projects/library/retentions" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "rules": [
      {
        "disabled": false,
        "action": "retain",
        "template": "latestPushedK",
        "params": {
          "latestPushedK": 10
        },
        "tag_selectors": [
          {
            "kind": "doublestar",
            "decoration": "matches",
            "pattern": "v*"
          }
        ],
        "scope_selectors": {
          "repository": [
            {
              "kind": "doublestar",
              "decoration": "repoMatches",
              "pattern": "**"
            }
          ]
        }
      },
      {
        "disabled": false,
        "action": "retain",
        "template": "nDaysSinceLastPush",
        "params": {
          "nDaysSinceLastPush": 90
        },
        "tag_selectors": [
          {
            "kind": "doublestar",
            "decoration": "matches",
            "pattern": "main-*"
          }
        ]
      }
    ],
    "algorithm": "or",
    "trigger": {
      "kind": "Schedule",
      "settings": {
        "cron": "0 0 * * 0"
      }
    }
  }'

# Enable tag immutability for production
curl -X POST "https://harbor.example.com/api/v2.0/projects/library/immutabletagrules" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "tag_selectors": [
      {
        "kind": "doublestar",
        "decoration": "matches",
        "pattern": "v*.*.*"
      }
    ],
    "scope_selectors": {
      "repository": [
        {
          "kind": "doublestar",
          "decoration": "repoMatches",
          "pattern": "production/**"
        }
      ]
    }
  }'

---

Pattern 7: Webhook Automation and Event Handling

# Configure webhook for vulnerability scan results
curl -X POST "https://harbor.example.com/api/v2.0/projects/library/webhook/policies" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "notify-security-team",
    "description": "Alert on critical vulnerabilities",
    "enabled": true,
    "event_types": [
      "SCANNING_COMPLETED",
      "SCANNING_FAILED"
    ],
    "targets": [
      {
        "type": "http",
        "address": "https://slack.com/api/webhooks/xxx",
        "skip_cert_verify": false,
        "payload_format": "CloudEvents"
      }
    ]
  }'

Webhook Payload Structure:

{
  "specversion": "1.0",
  "type": "harbor.scanning.completed",
  "source": "harbor.example.com",
  "id": "unique-id",
  "time": "2024-01-15T10:30:00Z",
  "data": {
    "repository": "library/app",
    "tag": "v1.0.0",
    "scan_overview": {
      "severity": "High",
      "total_count": 5,
      "fixable_count": 3,
      "summary": {
        "Critical": 0,
        "High": 5,
        "Medium": 12
      }
    }
  }
}

---

6. Implementation Workflow (TDD)

Step 1: Write Failing Test First

Before implementing any Harbor configuration, write tests to verify expected behavior:

# tests/test_harbor_config.py
import pytest
import requests
from unittest.mock import patch, MagicMock

class TestHarborProjectConfiguration:
    """Test Harbor project settings before implementation."""

    def test_project_vulnerability_policy_blocks_critical(self):
        """Test that CVE policy blocks critical vulnerabilities."""
        # Arrange
        project_config = {
            "prevent_vulnerable": True,
            "severity": "critical",
            "scan_on_push": True
        }

        # Act
        result = validate_vulnerability_policy(project_config)

        # Assert
        assert result["blocks_critical"] == True
        assert result["scan_enabled"] == True

    def test_robot_account_follows_least_privilege(self):
        """Test robot account has minimal required permissions."""
        # Arrange
        robot_permissions = {
            "namespace": "library",
            "access": [
                {"resource": "repository", "action": "pull"},
                {"resource": "repository", "action": "push"}
            ]
        }

        # Act
        result = validate_robot_permissions(robot_permissions)

        # Assert
        assert result["is_scoped"] == True
        assert result["has_admin"] == False
        assert len(result["permissions"]) <= 3

    def test_replication_policy_has_filters(self):
        """Test replication policy includes proper filters."""
        # Arrange
        replication_config = {
            "filters": [
                {"type": "name", "value": "library/app-*"},
                {"type": "tag", "value": "v*"}
            ],
            "trigger": {"type": "scheduled"}
        }

        # Act
        result = validate_replication_policy(replication_config)

        # Assert
        assert result["has_name_filter"] == True
        assert result["has_tag_filter"] == True
        assert result["is_scheduled"] == True


class TestHarborAPIIntegration:
    """Integration tests for Harbor API operations."""

    @pytest.fixture
    def harbor_client(self):
        """Create Harbor API client for testing."""
        return HarborClient(
            url="https://harbor.example.com",
            username="admin",
            password="test"
        )

    def test_create_project_with_security_policies(self, harbor_client):
        """Test project creation includes security policies."""
        # Arrange
        project_spec = {
            "project_name": "test-project",
            "public": False,
            "metadata": {
                "enable_content_trust": "true",
                "prevent_vul": "true",
                "severity": "high",
                "auto_scan": "true"
            }
        }

        # Act
        result = harbor_client.create_project(project_spec)

        # Assert
        assert result.status_code == 201
        project = harbor_client.get_project("test-project")
        assert project["metadata"]["enable_content_trust"] == "true"
        assert project["metadata"]["prevent_vul"] == "true"

    def test_garbage_collection_schedule_configured(self, harbor_client):
        """Test GC schedule is properly configured."""
        # Arrange
        gc_schedule = {
            "schedule": {
                "type": "Weekly",
                "cron": "0 2 * * 6"
            },
            "parameters": {
                "delete_untagged": True,
                "dry_run": False
            }
        }

        # Act
        result = harbor_client.set_gc_schedule(gc_schedule)

        # Assert
        assert result.status_code == 200
        current_schedule = harbor_client.get_gc_schedule()
        assert current_schedule["schedule"]["cron"] == "0 2 * * 6"

Step 2: Implement Minimum to Pass

# harbor_client.py
import requests
from typing import Dict, Any

class HarborClient:
    """Harbor API client with security-first defaults."""

    def __init__(self, url: str, username: str, password: str):
        self.url = url.rstrip('/')
        self.auth = (username, password)
        self.session = requests.Session()
        self.session.auth = self.auth
        self.session.headers.update({"Content-Type": "application/json"})

    def create_project(self, spec: Dict[str, Any]) -> requests.Response:
        """Create project with security policies."""
        # Ensure security defaults
        if "metadata" not in spec:
            spec["metadata"] = {}

        spec["metadata"].setdefault("enable_content_trust", "true")
        spec["metadata"].setdefault("prevent_vul", "true")
        spec["metadata"].setdefault("severity", "high")
        spec["metadata"].setdefault("auto_scan", "true")

        return self.session.post(
            f"{self.url}/api/v2.0/projects",
            json=spec
        )

    def set_gc_schedule(self, schedule: Dict[str, Any]) -> requests.Response:
        """Configure garbage collection schedule."""
        return self.session.post(
            f"{self.url}/api/v2.0/system/gc/schedule",
            json=schedule
        )

Step 3: Refactor If Needed

After tests pass, refactor for better error handling and performance:

# Refactored with retry logic and connection pooling
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class HarborClient:
    def __init__(self, url: str, username: str, password: str):
        self.url = url.rstrip('/')
        self.auth = (username, password)
        self.session = self._create_session()

    def _create_session(self) -> requests.Session:
        """Create session with retry and connection pooling."""
        session = requests.Session()
        session.auth = self.auth
        session.headers.update({"Content-Type": "application/json"})

        # Configure retries for resilience
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        adapter = HTTPAdapter(
            max_retries=retry_strategy,
            pool_connections=10,
            pool_maxsize=10
        )
        session.mount("https://", adapter)

        return session

Step 4: Run Full Verification

# Run all tests
pytest tests/test_harbor_config.py -v

# Run with coverage
pytest tests/test_harbor_config.py --cov=harbor_client --cov-report=term-missing

# Validate actual Harbor configuration
curl -s "https://harbor.example.com/api/v2.0/systeminfo" \
  -u "admin:password" | jq '.harbor_version'

# Test scanner connectivity
curl -s "https://harbor.example.com/api/v2.0/scanners" \
  -u "admin:password" | jq '.[].is_default'

# Verify replication endpoints
curl -s "https://harbor.example.com/api/v2.0/registries" \
  -u "admin:password" | jq '.[].status'

---

7. Performance Patterns

Pattern 1: Garbage Collection Optimization

Bad - Infrequent GC causes storage bloat:

# ❌ Monthly GC - storage fills up
{
  "schedule": {
    "type": "Custom",
    "cron": "0 0 1 * *"
  },
  "parameters": {
    "delete_untagged": false
  }
}

Good - Regular GC with untagged deletion:

# ✅ Weekly GC with untagged cleanup
curl -X POST "https://harbor.example.com/api/v2.0/system/gc/schedule" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "schedule": {
      "type": "Weekly",
      "cron": "0 2 * * 6"
    },
    "parameters": {
      "delete_untagged": true,
      "dry_run": false,
      "workers": 4
    }
  }'

# Monitor GC performance
curl -s "https://harbor.example.com/api/v2.0/system/gc" \
  -u "admin:password" | jq '.[-1] | {status, deleted, duration: (.end_time - .start_time)}'

Pattern 2: Replication Optimization

Bad - Unfiltered full replication:

# ❌ Replicate everything - wastes bandwidth
{
  "name": "replicate-all",
  "filters": [],
  "trigger": {"type": "event_based"},
  "speed": 0
}

Good - Filtered scheduled replication with bandwidth control:

# ✅ Filtered replication with scheduling and rate limiting
curl -X POST "https://harbor.example.com/api/v2.0/replication/policies" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "replicate-production",
    "filters": [
      {"type": "name", "value": "production/**"},
      {"type": "tag", "value": "v*"},
      {"type": "label", "value": "approved=true"}
    ],
    "trigger": {
      "type": "scheduled",
      "trigger_settings": {
        "cron": "0 */4 * * *"
      }
    },
    "speed": 10485760,
    "override": true,
    "enabled": true
  }'

# Monitor replication performance
curl -s "https://harbor.example.com/api/v2.0/replication/executions?policy_id=1" \
  -u "admin:password" | jq '[.[] | select(.status=="Succeed")] | length'

Pattern 3: Caching and Proxy Configuration

Bad - No caching, direct pulls every time:

# ❌ Every pull hits upstream registry
docker pull docker.io/library/nginx:latest
# Slow and uses bandwidth

Good - Harbor as proxy cache:

# ✅ Configure proxy cache endpoint
curl -X POST "https://harbor.example.com/api/v2.0/registries" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dockerhub-cache",
    "type": "docker-hub",
    "url": "https://hub.docker.com",
    "credential": {
      "access_key": "username",
      "access_secret": "token"
    }
  }'

# Create proxy cache project
curl -X POST "https://harbor.example.com/api/v2.0/projects" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "project_name": "dockerhub-proxy",
    "registry_id": 1,
    "public": true
  }'

# Pull through cache - subsequent pulls are instant
docker pull harbor.example.com/dockerhub-proxy/library/nginx:latest

Pattern 4: Storage Backend Optimization

Bad - Local filesystem storage:

# ❌ Filesystem storage - no HA, backup complexity
storage_service:
  filesystem:
    rootdirectory: /data/registry

Good - Object storage with lifecycle policies:

# ✅ S3 storage with intelligent tiering
REGISTRY_STORAGE_PROVIDER_NAME=s3
REGISTRY_STORAGE_PROVIDER_CONFIG='{
  "bucket": "harbor-artifacts",
  "region": "us-east-1",
  "rootdirectory": "/harbor",
  "storageclass": "INTELLIGENT_TIERING",
  "multipartcopythresholdsize": 33554432,
  "multipartcopychunksize": 33554432,
  "multipartcopymaxconcurrency": 100,
  "encrypt": true,
  "v4auth": true
}'

# Configure lifecycle policy for old artifacts
aws s3api put-bucket-lifecycle-configuration \
  --bucket harbor-artifacts \
  --lifecycle-configuration '{
    "Rules": [{
      "ID": "archive-old-artifacts",
      "Status": "Enabled",
      "Filter": {"Prefix": "harbor/"},
      "Transitions": [{
        "Days": 90,
        "StorageClass": "GLACIER"
      }],
      "NoncurrentVersionTransitions": [{
        "NoncurrentDays": 30,
        "StorageClass": "GLACIER"
      }]
    }]
  }'

Pattern 5: Database Connection Pooling

Bad - Default database connections:

# ❌ Default connections - bottleneck under load
POSTGRESQL_MAX_OPEN_CONNS=0
POSTGRESQL_MAX_IDLE_CONNS=2

Good - Optimized connection pool:

# ✅ Tuned connection pool for production
POSTGRESQL_HOST=postgres.example.com
POSTGRESQL_PORT=5432
POSTGRESQL_MAX_OPEN_CONNS=100
POSTGRESQL_MAX_IDLE_CONNS=50
POSTGRESQL_CONN_MAX_LIFETIME=5m
POSTGRESQL_SSLMODE=require

# Redis connection optimization
REDIS_HOST=redis.example.com:6379
REDIS_PASSWORD=${REDIS_PASSWORD}
REDIS_DB_INDEX=0
REDIS_IDLE_TIMEOUT_SECONDS=30

# Monitor connection usage
psql -h postgres.example.com -U harbor -c \
  "SELECT count(*) as active_connections FROM pg_stat_activity WHERE datname='registry';"

Pattern 6: Scan Performance Tuning

Bad - Sequential scanning with long timeout:

# ❌ Slow scanning blocks pushes
SCANNER_TRIVY_TIMEOUT=30m
# No parallelization

Good - Parallel scanning with optimized settings:

# ✅ Optimized Trivy scanner configuration
trivy:
  environment:
    SCANNER_TRIVY_TIMEOUT: "10m"
    SCANNER_TRIVY_VULN_TYPE: "os,library"
    SCANNER_TRIVY_SEVERITY: "UNKNOWN,LOW,MEDIUM,HIGH,CRITICAL"
    SCANNER_TRIVY_SKIP_UPDATE: "false"
    SCANNER_TRIVY_GITHUB_TOKEN: "${GITHUB_TOKEN}"
    SCANNER_TRIVY_CACHE_DIR: "/home/scanner/.cache/trivy"
    SCANNER_STORE_REDIS_URL: "redis://redis:6379/5"
    SCANNER_JOB_QUEUE_REDIS_URL: "redis://redis:6379/6"
  volumes:
    - trivy-cache:/home/scanner/.cache/trivy
  deploy:
    replicas: 3
    resources:
      limits:
        memory: 4G
        cpus: '2'

# Pre-download vulnerability database
docker exec trivy trivy image --download-db-only

---

5. Security Standards

5.1 Image Signing Requirements

Content Trust Policy:

• All production images MUST be signed before deployment
• Use Cosign with keyless signing (OIDC) for transparency
• Attach SBOMs to all signed images
• Verify signatures in admission controllers (Kyverno)
• Track signature coverage metrics (target: 100% for prod)

Signing Workflow: 1. Build image in CI/CD pipeline 2. Scan with Trivy (must pass CVE policy) 3. Generate SBOM with Syft or Trivy 4. Sign image with Cosign (ephemeral keys via OIDC) 5. Attach SBOM as artifact 6. Push to Harbor registry 7. Verify signature before Kubernetes deployment

---

5.2 Vulnerability Management

CVE Policy Enforcement:

• CRITICAL: Block all deployments, require immediate fix
• HIGH: Block production, allow dev with time-bound exemption
• MEDIUM: Alert only, track in security dashboard
• LOW/UNKNOWN: Log for awareness

Scan Configuration:

• Scan on push: Enabled for all projects
• Automatic rescan: Daily at 2 AM UTC
• Vulnerability database update: Every 6 hours
• Scan timeout: 10 minutes per image
• Retention: Keep scan results for 90 days

Exemption Process: 1. Security team reviews CVE impact 2. Create allowlist entry with expiration date 3. Document mitigation or compensating controls 4. Track exemptions in compliance reports 5. Alert 7 days before exemption expires

---

5.3 RBAC and Access Control

Project Roles:

• Project Admin: Full control, manage members, configure policies
• Developer: Push/pull images, view scan results, cannot change policies
• Guest: Pull images only, read-only access to metadata
• Limited Guest: Pull specific repositories only

Robot Account Best Practices:

• Use robot accounts for all automation (never user credentials)
• Scope to single project with minimal permissions
• Set expiration (90 days max, rotate at 60 days)
• Use descriptive names: robot$service-environment-action
• Audit robot account usage weekly
• Revoke immediately when service is decommissioned

OIDC Integration:

# Harbor OIDC configuration
auth_mode: oidc_auth
oidc_name: Keycloak
oidc_endpoint: https://keycloak.example.com/auth/realms/harbor
oidc_client_id: harbor
oidc_client_secret: ${OIDC_SECRET}
oidc_scope: openid,profile,email,groups
oidc_verify_cert: true
oidc_auto_onboard: true
oidc_user_claim: preferred_username
oidc_group_claim: groups

---

5.4 Supply Chain Security

Artifact Integrity:

• Enable content trust for all production projects
• Require signatures from trusted issuers only
• Verify SBOM presence and completeness
• Track artifact provenance from source to deployment
• Implement cosign verification in admission controllers

Base Image Security:

• Use official minimal base images (distroless, alpine, chainguard)
• Scan base images before use
• Pin base images with digest (not tags)
• Monitor base image CVE notifications
• Update base images within 7 days of security patches

Compliance Tracking:

• Generate weekly compliance reports
• Track metrics: signature coverage, scan pass rate, CVE MTTR
• Audit artifact access patterns
• Alert on unsigned production deployments
• Monthly security review with stakeholders

---

8. Common Mistakes

Mistake 1: Allowing Unsigned Images in Production

Problem:

# ❌ No signature verification
apiVersion: v1
kind: Pod
spec:
  containers:
  - image: harbor.example.com/library/app:latest

Solution:

# ✅ Kyverno enforces signatures
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: require-signed-images
spec:
  validationFailureAction: Enforce
  rules:
    - name: verify-signature
      verifyImages:
      - imageReferences: ["harbor.example.com/library/*"]
        required: true

---

Mistake 2: Overly Permissive Robot Accounts

Problem:

# ❌ Project admin for CI/CD
{
  "permissions": [{
    "namespace": "library",
    "access": [{"resource": "*", "action": "*"}]
  }]
}

Solution:

# ✅ Minimal scoped permissions
{
  "name": "ci-pipeline",
  "duration": 90,
  "permissions": [{
    "namespace": "library",
    "access": [
      {"resource": "repository", "action": "pull"},
      {"resource": "repository", "action": "push"},
      {"resource": "artifact-label", "action": "create"}
    ]
  }]
}

---

Mistake 3: No CVE Blocking Policy

Problem:

// ❌ Scan only, no enforcement
{
  "scan_on_push": true,
  "prevent_vulnerable": false
}

Solution:

// ✅ Block critical/high CVEs
{
  "scan_on_push": true,
  "prevent_vulnerable": true,
  "severity": "high",
  "auto_scan": true
}

---

Mistake 4: Missing Replication Monitoring

Problem:

# ❌ Set and forget replication
# No monitoring, failures go unnoticed

Solution:

# ✅ Monitor replication health
curl "https://harbor.example.com/api/v2.0/replication/executions?policy_id=1" \
  -u "admin:password" | jq -r '.[] | select(.status=="Failed")'

# Alert on replication lag > 1 hour
LAST_SUCCESS=$(curl -s "..." | jq -r '.[-1].end_time')
LAG=$(( $(date +%s) - $(date -d "$LAST_SUCCESS" +%s) ))
if [ $LAG -gt 3600 ]; then
  alert "Replication lag detected"
fi

---

Mistake 5: No Garbage Collection

Problem:

# ❌ Storage grows indefinitely
# Deleted artifacts never cleaned up

Solution:

# ✅ Scheduled garbage collection
# Harbor UI: Administration > Garbage Collection > Schedule
# Cron: 0 2 * * 6 (every Saturday 2 AM)

# Or via API
curl -X POST "https://harbor.example.com/api/v2.0/system/gc/schedule" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d '{
    "schedule": {
      "type": "Weekly",
      "cron": "0 2 * * 6"
    },
    "parameters": {
      "delete_untagged": true,
      "dry_run": false
    }
  }'

---

Mistake 6: Using :latest Tag in Production

Problem:

# ❌ Non-deterministic deployments
image: harbor.example.com/library/app:latest

Solution:

# ✅ Immutable digest-based references
image: harbor.example.com/library/app@sha256:abc123...

# Or immutable semantic version
image: harbor.example.com/library/app:v1.2.3
# + tag immutability rule for v*.*.* pattern

---

9. Testing

Unit Testing Harbor Configurations

# tests/test_harbor_policies.py
import pytest
from harbor_client import HarborClient, validate_project_config

class TestProjectPolicies:
    """Unit tests for Harbor project configuration."""

    def test_vulnerability_policy_requires_scanning(self):
        """Verify CVE policy requires scan_on_push."""
        config = {
            "prevent_vulnerable": True,
            "severity": "high",
            "scan_on_push": False  # Invalid combination
        }

        result = validate_project_config(config)
        assert result["valid"] == False
        assert "scan_on_push required" in result["errors"]

    def test_content_trust_requires_notary(self):
        """Verify content trust needs Notary configured."""
        config = {
            "enable_content_trust": True,
            "notary_url": None
        }

        result = validate_project_config(config)
        assert result["valid"] == False

    def test_retention_policy_validation(self):
        """Verify retention rules are valid."""
        policy = {
            "rules": [{
                "template": "latestPushedK",
                "params": {"latestPushedK": -1}  # Invalid
            }]
        }

        result = validate_retention_policy(policy)
        assert result["valid"] == False


class TestRobotAccounts:
    """Test robot account permission validation."""

    def test_robot_account_expiration_required(self):
        """Robot accounts must have expiration."""
        robot = {
            "name": "ci-pipeline",
            "duration": 0,  # Never expires - bad
            "permissions": [{"resource": "repository", "action": "push"}]
        }

        result = validate_robot_account(robot)
        assert result["valid"] == False
        assert "expiration required" in result["errors"]

    def test_robot_account_max_duration(self):
        """Robot account max duration is 90 days."""
        robot = {
            "name": "ci-pipeline",
            "duration": 365,  # Too long
            "permissions": [{"resource": "repository", "action": "push"}]
        }

        result = validate_robot_account(robot)
        assert result["valid"] == False
        assert "max duration 90 days" in result["errors"]

Integration Testing with Harbor API

# tests/integration/test_harbor_api.py
import pytest
import os
from harbor_client import HarborClient

@pytest.fixture(scope="module")
def harbor():
    """Create Harbor client for integration tests."""
    return HarborClient(
        url=os.getenv("HARBOR_URL", "https://harbor.example.com"),
        username=os.getenv("HARBOR_USER", "admin"),
        password=os.getenv("HARBOR_PASSWORD")
    )

class TestHarborAPIIntegration:
    """Integration tests against live Harbor instance."""

    def test_health_check(self, harbor):
        """Verify Harbor API is accessible."""
        result = harbor.health()
        assert result.status_code == 200
        assert result.json()["status"] == "healthy"

    def test_scanner_configured(self, harbor):
        """Verify Trivy scanner is default."""
        scanners = harbor.get_scanners()
        default_scanner = next(
            (s for s in scanners if s["is_default"]), None
        )
        assert default_scanner is not None
        assert "trivy" in default_scanner["name"].lower()

    def test_project_security_defaults(self, harbor):
        """Verify projects have security settings."""
        # Create test project
        project = harbor.create_project({
            "project_name": "test-security-defaults",
            "public": False
        })

        # Verify security defaults applied
        metadata = harbor.get_project("test-security-defaults")["metadata"]
        assert metadata.get("enable_content_trust") == "true"
        assert metadata.get("prevent_vul") == "true"
        assert metadata.get("auto_scan") == "true"

        # Cleanup
        harbor.delete_project("test-security-defaults")

    def test_gc_schedule_exists(self, harbor):
        """Verify garbage collection is scheduled."""
        schedule = harbor.get_gc_schedule()
        assert schedule["schedule"]["type"] in ["Weekly", "Daily", "Custom"]
        assert schedule["parameters"]["delete_untagged"] == True


class TestReplicationPolicies:
    """Test replication policy configurations."""

    def test_replication_endpoint_tls(self, harbor):
        """Verify replication endpoints use TLS."""
        endpoints = harbor.get_registries()
        for endpoint in endpoints:
            assert endpoint["url"].startswith("https://")
            assert endpoint["insecure"] == False

    def test_replication_has_filters(self, harbor):
        """Verify replication policies have filters."""
        policies = harbor.get_replication_policies()
        for policy in policies:
            if policy["enabled"]:
                assert len(policy.get("filters", [])) > 0, \
                    f"Policy {policy['name']} has no filters"

End-to-End Testing

#!/bin/bash
# tests/e2e/test_harbor_workflow.sh

set -e

HARBOR_URL="${HARBOR_URL:-https://harbor.example.com}"
PROJECT="e2e-test-$(date +%s)"

echo "=== Harbor E2E Test Suite ==="

# Test 1: Create project with security defaults
echo "Test 1: Creating project with security defaults..."
curl -s -X POST "${HARBOR_URL}/api/v2.0/projects" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d "{\"project_name\": \"${PROJECT}\", \"public\": false}" \
  -o /dev/null -w "%{http_code}" | grep -q "201"
echo "✓ Project created"

# Test 2: Verify security policies applied
echo "Test 2: Verifying security policies..."
METADATA=$(curl -s "${HARBOR_URL}/api/v2.0/projects/${PROJECT}" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" | jq '.metadata')

echo "$METADATA" | jq -e '.auto_scan == "true"' > /dev/null
echo "✓ Auto scan enabled"

echo "$METADATA" | jq -e '.prevent_vul == "true"' > /dev/null
echo "✓ Vulnerability prevention enabled"

# Test 3: Push and scan image
echo "Test 3: Pushing and scanning image..."
docker pull alpine:latest
docker tag alpine:latest "${HARBOR_URL}/${PROJECT}/alpine:test"
docker push "${HARBOR_URL}/${PROJECT}/alpine:test"

# Wait for scan
sleep 30

SCAN_STATUS=$(curl -s "${HARBOR_URL}/api/v2.0/projects/${PROJECT}/repositories/alpine/artifacts/test" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" | jq -r '.scan_overview.scan_status')

[ "$SCAN_STATUS" == "Success" ]
echo "✓ Image scanned successfully"

# Test 4: Create robot account
echo "Test 4: Creating robot account..."
ROBOT=$(curl -s -X POST "${HARBOR_URL}/api/v2.0/projects/${PROJECT}/robots" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "e2e-test",
    "duration": 1,
    "permissions": [{"namespace": "'${PROJECT}'", "access": [{"resource": "repository", "action": "pull"}]}]
  }')

echo "$ROBOT" | jq -e '.secret' > /dev/null
echo "✓ Robot account created"

# Cleanup
echo "Cleaning up..."
curl -s -X DELETE "${HARBOR_URL}/api/v2.0/projects/${PROJECT}" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}"
echo "✓ Cleanup complete"

echo "=== All E2E tests passed ==="

Running Tests

# Run unit tests
pytest tests/test_harbor_policies.py -v

# Run integration tests (requires HARBOR_URL, HARBOR_USER, HARBOR_PASSWORD)
pytest tests/integration/ -v --tb=short

# Run E2E tests
./tests/e2e/test_harbor_workflow.sh

# Run all tests with coverage
pytest tests/ --cov=harbor_client --cov-report=html

# Specific test markers
pytest -m "not integration"  # Skip integration tests
pytest -m "security"         # Run only security tests

---

13. Critical Reminders

Pre-Implementation Checklist

Phase 1: Before Writing Code

• [ ] Read existing Harbor configuration and version
• [ ] Identify affected projects and replication policies
• [ ] Review current security policies (CVE blocking, content trust)
• [ ] Check existing robot accounts and their permissions
• [ ] Document current garbage collection schedule
• [ ] Write failing tests for new functionality
• [ ] Review Harbor API documentation for changes

Phase 2: During Implementation

• [ ] Follow TDD workflow (test first, implement, refactor)
• [ ] Apply security defaults to all new projects
• [ ] Use least privilege for robot accounts
• [ ] Configure filters for replication policies
• [ ] Enable scan-on-push for all artifacts
• [ ] Set appropriate retention policies
• [ ] Test all API calls return expected results

Phase 3: Before Committing

• [ ] Run full test suite (unit, integration, E2E)
• [ ] Verify all security policies are enforced
• [ ] Check garbage collection is scheduled
• [ ] Validate replication endpoints are healthy
• [ ] Confirm scanner is operational
• [ ] Review audit logs for anomalies
• [ ] Update documentation if needed

---

Pre-Production Deployment Checklist

Registry Configuration:

• [ ] PostgreSQL and Redis externalized (not embedded)
• [ ] Storage backend configured (S3/GCS/Azure, not filesystem)
• [ ] TLS certificates valid and auto-renewing
• [ ] Backup strategy configured and tested
• [ ] Resource limits set (CPU, memory, storage quota)

Security Hardening:

• [ ] Trivy scanner integrated and set as default
• [ ] Scan-on-push enabled for all projects
• [ ] CVE blocking policy configured (HIGH/CRITICAL)
• [ ] Content trust enabled for production projects
• [ ] Tag immutability enabled for release tags
• [ ] Robot accounts follow least privilege
• [ ] OIDC/LDAP authentication configured
• [ ] Audit logging enabled

Replication and DR:

• [ ] Multi-region replication configured
• [ ] Replication monitoring and alerting active
• [ ] Disaster recovery runbook documented
• [ ] Failover tested within last 90 days
• [ ] RTO/RPO requirements met

Compliance:

• [ ] Retention policies configured
• [ ] Webhook notifications for security events
• [ ] Compliance reports generated weekly
• [ ] Signature coverage >95% for production
• [ ] CVE MTTR <7 days for critical

Operational Readiness:

• [ ] Garbage collection scheduled weekly
• [ ] Database vacuum scheduled monthly
• [ ] Monitoring dashboards configured
• [ ] Runbooks for common incidents
• [ ] On-call team trained on Harbor administration

---

Critical Security Controls

NEVER:

• Deploy unsigned images to production
• Allow scan-failing images with CRITICAL CVEs
• Use user credentials in CI/CD (use robot accounts)
• Share robot account tokens across services
• Disable content trust for production projects
• Skip replication testing before DR events
• Allow public access to private registries

ALWAYS:

• Scan all images before deployment
• Sign production images with provenance
• Rotate robot account tokens every 90 days
• Monitor replication lag and failures
• Test backup/restore procedures quarterly
• Update Trivy vulnerability database daily
• Audit unusual access patterns weekly
• Document CVE exemptions with expiration

---

14. Summary

You are a Harbor expert who manages secure container registries with comprehensive vulnerability scanning, artifact signing, and multi-region replication. You implement defense-in-depth security with Trivy CVE scanning, Cosign image signing, RBAC controls, and deployment policies that block vulnerable or unsigned images.

You design highly available registry infrastructure with PostgreSQL/Redis backends, S3 storage, and pull-based replication to secondary regions for disaster recovery. You implement compliance automation with retention policies, tag immutability, audit logging, and webhook notifications for security events.

You protect the software supply chain by requiring signed artifacts, enforcing CVE policies, generating compliance reports, and integrating signature verification in Kubernetes admission controllers. You optimize registry operations with garbage collection, quota management, and performance monitoring.

Your mission: Provide secure, reliable container registry infrastructure that protects organizations from supply chain attacks while enabling developer velocity.

Reference Materials:

• Security Scanning: /home/user/ai-coding/new-skills/harbor-expert/references/security-scanning.md
• Replication Guide: /home/user/ai-coding/new-skills/harbor-expert/references/replication-guide.md

Harbor Replication and Disaster Recovery Guide

Multi-Region Replication Strategies

This reference provides comprehensive patterns for multi-region replication, disaster recovery, and registry federation in Harbor.

---

1. Replication Architecture Patterns

Pattern 1: Hub-and-Spoke (Single Primary)

┌─────────────────┐
│  Primary Harbor │ (us-east-1)
│   (Read/Write)  │
└────────┬────────┘
         │ Push replication
         ├──────────────┬──────────────┬──────────────┐
         │              │              │              │
    ┌────▼────┐    ┌────▼────┐    ┌────▼────┐   ┌────▼────┐
    │ Harbor  │    │ Harbor  │    │ Harbor  │   │ Harbor  │
    │ EU-West │    │ AP-SE   │    │ US-West │   │ SA-East │
    │ (RO)    │    │ (RO)    │    │ (RO)    │   │ (RO)    │
    └─────────┘    └─────────┘    └─────────┘   └─────────┘

Use case: Global content delivery, single source of truth Pros: Simple, consistent, no conflicts Cons: Single point of failure, higher latency for writes

Pattern 2: Active-Active (Multi-Primary)

┌─────────────────┐          ┌─────────────────┐
│  Harbor US-East │◄────────►│  Harbor EU-West │
│   (Read/Write)  │  Bidirectional │   (Read/Write)  │
└────────┬────────┘  Replication   └────────┬────────┘
         │                                   │
         │ Push replication                 │ Push replication
         │                                   │
    ┌────▼────┐                         ┌────▼────┐
    │ Harbor  │                         │ Harbor  │
    │ AP-SE   │                         │ SA-East │
    │ (RO)    │                         │ (RO)    │
    └─────────┘                         └─────────┘

Use case: Global development teams, regional autonomy Pros: No single point of failure, low latency Cons: Conflict resolution needed, complex

Pattern 3: Disaster Recovery (Primary-Secondary)

┌─────────────────┐
│  Primary Harbor │ (us-east-1)
│   (Read/Write)  │
└────────┬────────┘
         │ Continuous pull replication
         │ + Manual failover
    ┌────▼────┐
    │ Secondary│
    │ Harbor  │ (us-west-2)
    │(Standby)│
    └─────────┘

Use case: Business continuity, failover Pros: Simple DR, tested backup Cons: Manual failover, RPO/RTO considerations

---

2. Replication Configuration

Create Replication Endpoints

#!/bin/bash
# configure-endpoints.sh - Set up replication endpoints

HARBOR_URL="https://harbor-primary.example.com"
HARBOR_USER="admin"
HARBOR_PASSWORD="${HARBOR_ADMIN_PASSWORD}"

# Create endpoint for EU region
curl -X POST "${HARBOR_URL}/api/v2.0/registries" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "harbor-eu-west",
    "description": "EU West Harbor registry",
    "url": "https://harbor-eu.example.com",
    "credential": {
      "access_key": "robot$replication-eu",
      "access_secret": "'${EU_ROBOT_TOKEN}'"
    },
    "type": "harbor",
    "insecure": false
  }'

# Create endpoint for AP region
curl -X POST "${HARBOR_URL}/api/v2.0/registries" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "harbor-ap-southeast",
    "description": "Asia Pacific Harbor registry",
    "url": "https://harbor-ap.example.com",
    "credential": {
      "access_key": "robot$replication-ap",
      "access_secret": "'${AP_ROBOT_TOKEN}'"
    },
    "type": "harbor",
    "insecure": false
  }'

# Verify endpoints
curl -X GET "${HARBOR_URL}/api/v2.0/registries" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" | jq

Push-Based Replication Rules

#!/bin/bash
# create-push-replication.sh - Configure push replication

HARBOR_URL="https://harbor-primary.example.com"

# Production images to EU (immediate push)
curl -X POST "${HARBOR_URL}/api/v2.0/replication/policies" \
  -u "admin:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "production-to-eu",
    "description": "Replicate production images to EU on push",
    "dest_registry": {
      "id": 1
    },
    "src_registry": null,
    "dest_namespace": "production",
    "dest_namespace_replace_count": 0,
    "trigger": {
      "type": "event_based",
      "trigger_settings": null
    },
    "filters": [
      {
        "type": "name",
        "value": "production/**"
      },
      {
        "type": "tag",
        "value": "v[0-9]*"
      },
      {
        "type": "label",
        "value": "replicate=true"
      }
    ],
    "replicate_deletion": false,
    "deletion": false,
    "override": true,
    "enabled": true,
    "speed": -1
  }'

# All images to AP (scheduled nightly)
curl -X POST "${HARBOR_URL}/api/v2.0/replication/policies" \
  -u "admin:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "all-to-ap-nightly",
    "description": "Nightly sync to Asia Pacific",
    "dest_registry": {
      "id": 2
    },
    "dest_namespace": "",
    "trigger": {
      "type": "scheduled",
      "trigger_settings": {
        "cron": "0 2 * * *"
      }
    },
    "filters": [
      {
        "type": "name",
        "value": "**"
      }
    ],
    "deletion": true,
    "override": true,
    "enabled": true,
    "speed": 0
  }'

Pull-Based Replication Rules

#!/bin/bash
# create-pull-replication.sh - Configure pull replication

SECONDARY_HARBOR="https://harbor-dr.example.com"

# Pull all production images from primary
curl -X POST "${SECONDARY_HARBOR}/api/v2.0/replication/policies" \
  -u "admin:${DR_HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dr-pull-production",
    "description": "DR: Pull production images from primary",
    "src_registry": {
      "id": 1
    },
    "dest_registry": null,
    "dest_namespace": "production",
    "trigger": {
      "type": "scheduled",
      "trigger_settings": {
        "cron": "*/15 * * * *"
      }
    },
    "filters": [
      {
        "type": "name",
        "value": "production/**"
      },
      {
        "type": "resource",
        "value": "image"
      }
    ],
    "replicate_deletion": true,
    "deletion": false,
    "override": true,
    "enabled": true,
    "speed": -1
  }'

---

3. Advanced Filtering Strategies

Label-Based Replication

# Only replicate images with specific labels

# 1. Add label to image
curl -X POST "https://harbor.example.com/api/v2.0/projects/library/repositories/app/artifacts/v1.0.0/labels" \
  -u "admin:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "id": 1
  }'

# 2. Create label filter in replication rule
{
  "filters": [
    {
      "type": "label",
      "value": "production"
    }
  ]
}

Resource Type Filtering

# Replicate only specific artifact types
{
  "filters": [
    {
      "type": "resource",
      "value": "image"
    }
  ]
}

# Available resource types:
# - image: Docker/OCI images
# - chart: Helm charts
# - cnab: CNAB bundles
# - all: All artifact types

Complex Pattern Matching

# Replicate semantic versioned images only
{
  "filters": [
    {
      "type": "tag",
      "value": "v[0-9]+\\.[0-9]+\\.[0-9]+"
    }
  ]
}

# Exclude development/testing images
{
  "filters": [
    {
      "type": "tag",
      "value": "{v*,release-*}"
    },
    {
      "type": "tag",
      "value": "!{dev-*,test-*,pr-*}"
    }
  ]
}

---

4. Disaster Recovery Procedures

DR Setup and Configuration

#!/bin/bash
# setup-dr.sh - Configure disaster recovery harbor

PRIMARY="https://harbor-primary.example.com"
DR="https://harbor-dr.example.com"

# 1. Create robot account on primary for DR pulls
PRIMARY_ROBOT=$(curl -X POST "${PRIMARY}/api/v2.0/robots" \
  -u "admin:${PRIMARY_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dr-replication",
    "description": "DR replication robot",
    "duration": -1,
    "level": "system",
    "permissions": [
      {
        "kind": "project",
        "namespace": "*",
        "access": [
          {"resource": "repository", "action": "pull"},
          {"resource": "repository", "action": "list"}
        ]
      }
    ]
  }' | jq -r '.secret')

# 2. Register primary as source on DR
curl -X POST "${DR}/api/v2.0/registries" \
  -u "admin:${DR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "primary-harbor",
    "url": "'${PRIMARY}'",
    "credential": {
      "access_key": "robot$dr-replication",
      "access_secret": "'${PRIMARY_ROBOT}'"
    },
    "type": "harbor"
  }'

# 3. Create comprehensive pull replication
curl -X POST "${DR}/api/v2.0/replication/policies" \
  -u "admin:${DR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dr-full-sync",
    "description": "Full disaster recovery sync",
    "src_registry": {"id": 1},
    "dest_namespace": "",
    "trigger": {
      "type": "scheduled",
      "trigger_settings": {"cron": "*/10 * * * *"}
    },
    "filters": [{"type": "name", "value": "**"}],
    "replicate_deletion": true,
    "override": true,
    "enabled": true
  }'

Failover Procedure

#!/bin/bash
# failover-to-dr.sh - Failover from primary to DR

DR_HARBOR="https://harbor-dr.example.com"
DNS_ZONE="example.com"

echo "=== Harbor Failover Procedure ==="
echo "This will failover from primary to DR harbor"
read -p "Are you sure? (yes/no): " CONFIRM

if [ "$CONFIRM" != "yes" ]; then
  echo "Failover cancelled"
  exit 0
fi

# 1. Verify DR is healthy
echo "[1/5] Checking DR harbor health..."
DR_HEALTH=$(curl -s "${DR_HARBOR}/api/v2.0/health" | jq -r '.status')
if [ "$DR_HEALTH" != "healthy" ]; then
  echo "ERROR: DR harbor is not healthy (status: ${DR_HEALTH})"
  exit 1
fi

# 2. Trigger final sync from primary
echo "[2/5] Triggering final sync..."
EXEC_ID=$(curl -s -X POST "${DR_HARBOR}/api/v2.0/replication/executions" \
  -u "admin:${DR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{"policy_id": 1}' | jq -r '.id')

# Wait for sync to complete
while true; do
  STATUS=$(curl -s "${DR_HARBOR}/api/v2.0/replication/executions/${EXEC_ID}" \
    -u "admin:${DR_PASSWORD}" | jq -r '.status')
  if [ "$STATUS" == "Succeed" ]; then
    break
  elif [ "$STATUS" == "Failed" ]; then
    echo "ERROR: Final sync failed"
    exit 1
  fi
  echo "Waiting for sync... (status: ${STATUS})"
  sleep 5
done

# 3. Disable replication on DR (prevent pulls from failed primary)
echo "[3/5] Disabling replication..."
curl -X PUT "${DR_HARBOR}/api/v2.0/replication/policies/1" \
  -u "admin:${DR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{"enabled": false}'

# 4. Update DNS to point to DR
echo "[4/5] Updating DNS..."
# This is provider-specific; example for Route53:
aws route53 change-resource-record-sets \
  --hosted-zone-id Z1234567890ABC \
  --change-batch '{
    "Changes": [{
      "Action": "UPSERT",
      "ResourceRecordSet": {
        "Name": "harbor.'${DNS_ZONE}'",
        "Type": "CNAME",
        "TTL": 60,
        "ResourceRecords": [{"Value": "harbor-dr.'${DNS_ZONE}'"}]
      }
    }]
  }'

# 5. Verify failover
echo "[5/5] Verifying failover..."
sleep 10
NEW_TARGET=$(dig +short harbor.${DNS_ZONE} | tail -n1)
echo "DNS now points to: ${NEW_TARGET}"

echo ""
echo "=== Failover Complete ==="
echo "Harbor is now serving from DR location"
echo "RTO achieved: $(date)"
echo ""
echo "Next steps:"
echo "- Notify teams of failover"
echo "- Update monitoring dashboards"
echo "- Investigate primary failure"
echo "- Plan failback when primary is restored"

Failback Procedure

#!/bin/bash
# failback-to-primary.sh - Restore primary as active

PRIMARY_HARBOR="https://harbor-primary.example.com"
DR_HARBOR="https://harbor-dr.example.com"

echo "=== Harbor Failback Procedure ==="
read -p "Is primary fully restored and healthy? (yes/no): " CONFIRM

if [ "$CONFIRM" != "yes" ]; then
  echo "Failback cancelled. Restore primary first."
  exit 0
fi

# 1. Verify primary is healthy
echo "[1/6] Checking primary harbor health..."
PRIMARY_HEALTH=$(curl -s "${PRIMARY_HARBOR}/api/v2.0/health" | jq -r '.status')
if [ "$PRIMARY_HEALTH" != "healthy" ]; then
  echo "ERROR: Primary harbor is not healthy"
  exit 1
fi

# 2. Sync DR changes back to primary (catch-up)
echo "[2/6] Syncing DR changes to primary..."
# Temporarily create reverse replication
# (This assumes primary was configured with DR as endpoint)
curl -X POST "${PRIMARY_HARBOR}/api/v2.0/replication/executions" \
  -u "admin:${PRIMARY_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{"policy_id": 99}' # Reverse sync policy

# 3. Wait for sync
echo "[3/6] Waiting for sync completion..."
sleep 60  # Adjust based on data size

# 4. Update DNS back to primary
echo "[4/6] Updating DNS to primary..."
aws route53 change-resource-record-sets \
  --hosted-zone-id Z1234567890ABC \
  --change-batch '{
    "Changes": [{
      "Action": "UPSERT",
      "ResourceRecordSet": {
        "Name": "harbor.example.com",
        "Type": "CNAME",
        "TTL": 60,
        "ResourceRecords": [{"Value": "harbor-primary.example.com"}]
      }
    }]
  }'

# 5. Re-enable forward replication (primary -> DR)
echo "[5/6] Re-enabling DR replication..."
curl -X PUT "${DR_HARBOR}/api/v2.0/replication/policies/1" \
  -u "admin:${DR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{"enabled": true}'

# 6. Verify
echo "[6/6] Verifying failback..."
sleep 10
echo "DNS now points to: $(dig +short harbor.example.com | tail -n1)"

echo ""
echo "=== Failback Complete ==="
echo "Primary harbor is now active"
echo "DR replication is re-enabled"

---

5. Replication Monitoring

Monitoring Script

#!/usr/bin/env python3
# monitor-replication.py - Monitor replication health

import requests
from requests.auth import HTTPBasicAuth
import time
import os
from datetime import datetime, timedelta

HARBOR_URL = "https://harbor.example.com"
USERNAME = "admin"
PASSWORD = os.environ["HARBOR_PASSWORD"]

auth = HTTPBasicAuth(USERNAME, PASSWORD)

def check_replication_health():
    """Check health of all replication policies"""
    policies_url = f"{HARBOR_URL}/api/v2.0/replication/policies"
    policies = requests.get(policies_url, auth=auth).json()

    alerts = []

    for policy in policies:
        policy_id = policy["id"]
        policy_name = policy["name"]
        enabled = policy["enabled"]

        if not enabled:
            continue

        # Get recent executions
        exec_url = f"{HARBOR_URL}/api/v2.0/replication/executions"
        params = {"policy_id": policy_id, "page_size": 10}
        executions = requests.get(exec_url, auth=auth, params=params).json()

        if not executions:
            alerts.append({
                "severity": "warning",
                "policy": policy_name,
                "message": "No executions found"
            })
            continue

        latest = executions[0]
        status = latest["status"]
        end_time = latest.get("end_time")

        # Check for failures
        if status == "Failed":
            alerts.append({
                "severity": "critical",
                "policy": policy_name,
                "message": f"Replication failed: {latest.get('status_text')}"
            })

        # Check for staleness (>24h since last success)
        if end_time:
            end_dt = datetime.fromisoformat(end_time.replace('Z', '+00:00'))
            age = datetime.now(end_dt.tzinfo) - end_dt
            if age > timedelta(hours=24) and status == "Succeed":
                alerts.append({
                    "severity": "warning",
                    "policy": policy_name,
                    "message": f"Last successful replication {age.total_seconds() / 3600:.1f}h ago"
                })

        # Check replication lag
        if status == "InProgress":
            start_time = latest.get("start_time")
            if start_time:
                start_dt = datetime.fromisoformat(start_time.replace('Z', '+00:00'))
                duration = datetime.now(start_dt.tzinfo) - start_dt
                if duration > timedelta(hours=1):
                    alerts.append({
                        "severity": "warning",
                        "policy": policy_name,
                        "message": f"Replication in progress for {duration.total_seconds() / 60:.0f} minutes"
                    })

    return alerts

def send_alerts(alerts):
    """Send alerts to monitoring system"""
    if not alerts:
        print(f"[{datetime.now()}] All replication policies healthy")
        return

    for alert in alerts:
        severity = alert["severity"]
        policy = alert["policy"]
        message = alert["message"]
        print(f"[{datetime.now()}] [{severity.upper()}] {policy}: {message}")

        # Send to alerting system (PagerDuty, Slack, etc.)
        # Example: send_to_pagerduty(alert)

if __name__ == "__main__":
    while True:
        try:
            alerts = check_replication_health()
            send_alerts(alerts)
        except Exception as e:
            print(f"[{datetime.now()}] [ERROR] Monitoring failed: {e}")

        time.sleep(300)  # Check every 5 minutes

Prometheus Metrics Exporter

#!/usr/bin/env python3
# harbor-replication-exporter.py - Export replication metrics to Prometheus

from prometheus_client import start_http_server, Gauge, Counter
import requests
from requests.auth import HTTPBasicAuth
import time
import os

HARBOR_URL = "https://harbor.example.com"
USERNAME = "admin"
PASSWORD = os.environ["HARBOR_PASSWORD"]

auth = HTTPBasicAuth(USERNAME, PASSWORD)

# Metrics
replication_status = Gauge('harbor_replication_status',
                           'Replication status (1=success, 0=failed, -1=in_progress)',
                           ['policy', 'destination'])
replication_lag_seconds = Gauge('harbor_replication_lag_seconds',
                                'Time since last successful replication',
                                ['policy', 'destination'])
replication_total = Counter('harbor_replication_total',
                            'Total replication executions',
                            ['policy', 'destination', 'status'])

def collect_metrics():
    """Collect replication metrics"""
    policies_url = f"{HARBOR_URL}/api/v2.0/replication/policies"
    policies = requests.get(policies_url, auth=auth).json()

    for policy in policies:
        policy_id = policy["id"]
        policy_name = policy["name"]

        # Get destination registry name
        dest_registry = policy.get("dest_registry", {})
        dest_name = dest_registry.get("name", "local") if dest_registry else "local"

        # Get latest execution
        exec_url = f"{HARBOR_URL}/api/v2.0/replication/executions"
        params = {"policy_id": policy_id, "page_size": 1}
        executions = requests.get(exec_url, auth=auth, params=params).json()

        if executions:
            latest = executions[0]
            status = latest["status"]

            # Status metric
            if status == "Succeed":
                replication_status.labels(policy=policy_name, destination=dest_name).set(1)
            elif status == "Failed":
                replication_status.labels(policy=policy_name, destination=dest_name).set(0)
            else:
                replication_status.labels(policy=policy_name, destination=dest_name).set(-1)

            # Lag metric
            end_time = latest.get("end_time")
            if end_time and status == "Succeed":
                from datetime import datetime
                end_dt = datetime.fromisoformat(end_time.replace('Z', '+00:00'))
                lag = (datetime.now(end_dt.tzinfo) - end_dt).total_seconds()
                replication_lag_seconds.labels(policy=policy_name, destination=dest_name).set(lag)

if __name__ == "__main__":
    start_http_server(9100)
    print("Harbor replication exporter running on :9100")

    while True:
        try:
            collect_metrics()
        except Exception as e:
            print(f"Error collecting metrics: {e}")
        time.sleep(60)

---

6. Bandwidth Optimization

Scheduled Replication for Large Datasets

# Replicate during off-peak hours
{
  "trigger": {
    "type": "scheduled",
    "trigger_settings": {
      "cron": "0 2 * * *"  # 2 AM daily
    }
  },
  "speed": 10485760  # Limit to 10 MB/s
}

Incremental Replication

# Only replicate changed artifacts
{
  "filters": [
    {
      "type": "name",
      "value": "**"
    }
  ],
  "override": false  # Don't re-replicate existing artifacts
}

Compression and Deduplication

Harbor automatically uses registry blob deduplication. Images sharing layers only transfer unique blobs.

---

7. Conflict Resolution

Handling Tag Conflicts

# Strategy 1: Override (last write wins)
{
  "override": true
}

# Strategy 2: Preserve destination
{
  "override": false
}

Bidirectional Replication Conflicts

For active-active setups, implement tag naming conventions:

# Region-specific tag prefixes
# us-east: myapp:v1.0.0-use1
# eu-west: myapp:v1.0.0-euw1

# Replicate with filters
{
  "filters": [
    {
      "type": "tag",
      "value": "!*-use1"  # Don't replicate US tags back to US
    }
  ]
}

---

Summary

This guide provides production-ready patterns for:

• Architecture Patterns: Hub-and-spoke, active-active, disaster recovery
• Replication Configuration: Push/pull rules, advanced filtering
• Disaster Recovery: Complete failover/failback procedures
• Monitoring: Health checks, Prometheus metrics, alerting
• Optimization: Bandwidth management, incremental sync
• Conflict Resolution: Tag management, bidirectional strategies

Use these patterns to implement highly available, globally distributed Harbor registries with tested disaster recovery procedures.

---

RTO/RPO Targets

Adjust replication frequency based on your specific RTO/RPO requirements.

Harbor Security Scanning Reference

Trivy Integration and CVE Management

This reference provides comprehensive patterns for vulnerability scanning, CVE policy enforcement, and webhook automation in Harbor.

---

1. Trivy Scanner Configuration

Installing and Configuring Trivy Adapter

# docker-compose.yml - Trivy service
services:
  trivy:
    image: goharbor/trivy-adapter-photon:v2.10.0
    container_name: trivy-adapter
    restart: always
    cap_drop:
      - ALL
    cap_add:
      - CHOWN
      - SETGID
      - SETUID
    dns_search: .
    environment:
      SCANNER_LOG_LEVEL: info
      SCANNER_TRIVY_CACHE_DIR: /home/scanner/.cache/trivy
      SCANNER_TRIVY_REPORTS_DIR: /home/scanner/.cache/reports
      SCANNER_TRIVY_DEBUG_MODE: "false"
      SCANNER_TRIVY_VULN_TYPE: "os,library"
      SCANNER_TRIVY_SECURITY_CHECKS: "vuln,config,secret"
      SCANNER_TRIVY_SEVERITY: "UNKNOWN,LOW,MEDIUM,HIGH,CRITICAL"
      SCANNER_TRIVY_IGNORE_UNFIXED: "false"
      SCANNER_TRIVY_SKIP_UPDATE: "false"
      SCANNER_TRIVY_OFFLINE_SCAN: "false"
      SCANNER_TRIVY_INSECURE: "false"
      SCANNER_TRIVY_TIMEOUT: "10m"
      SCANNER_API_SERVER_ADDR: ":8080"
      SCANNER_STORE_REDIS_URL: redis://redis:6379
      SCANNER_STORE_REDIS_NAMESPACE: harbor.scanner.trivy:store
      SCANNER_JOB_QUEUE_REDIS_URL: redis://redis:6379
      SCANNER_JOB_QUEUE_REDIS_NAMESPACE: harbor.scanner.trivy:job-queue
    volumes:
      - trivy_cache:/home/scanner/.cache
    networks:
      - harbor
    depends_on:
      - redis

volumes:
  trivy_cache:

Register Trivy Scanner via API

#!/bin/bash
# register-trivy.sh - Register and configure Trivy scanner

HARBOR_URL="https://harbor.example.com"
HARBOR_USER="admin"
HARBOR_PASSWORD="${HARBOR_ADMIN_PASSWORD}"

# Register scanner
SCANNER_ID=$(curl -X POST "${HARBOR_URL}/api/v2.0/scanners" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Trivy",
    "description": "Aqua Security Trivy vulnerability scanner",
    "url": "http://trivy:8080",
    "vendor": "Aqua Security",
    "version": "0.48.0",
    "auth": "",
    "access_credential": "",
    "skip_cert_verify": false,
    "use_internal_addr": true
  }' | jq -r '.id')

echo "Trivy scanner registered with ID: ${SCANNER_ID}"

# Set as default scanner
curl -X PATCH "${HARBOR_URL}/api/v2.0/scanners/${SCANNER_ID}" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{"is_default": true}'

echo "Trivy set as default scanner"

# Verify configuration
curl -X GET "${HARBOR_URL}/api/v2.0/scanners" \
  -u "${HARBOR_USER}:${HARBOR_PASSWORD}" | jq

---

2. CVE Policy Configuration

Project-Level Vulnerability Policies

#!/bin/bash
# configure-cve-policy.sh - Set up CVE policies for projects

HARBOR_URL="https://harbor.example.com"
PROJECT_NAME="production"

# Get project ID
PROJECT_ID=$(curl -s "${HARBOR_URL}/api/v2.0/projects?name=${PROJECT_NAME}" \
  -u "admin:${HARBOR_PASSWORD}" | jq -r '.[0].project_id')

# Configure strict CVE policy for production
curl -X PUT "${HARBOR_URL}/api/v2.0/projects/${PROJECT_ID}" \
  -u "admin:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "metadata": {
      "auto_scan": "true",
      "severity": "critical",
      "reuse_sys_cve_allowlist": "false",
      "prevent_vul": "true",
      "enable_content_trust": "true",
      "public": "false"
    }
  }'

echo "CVE policy configured for ${PROJECT_NAME}"

CVE Allowlist Management

#!/bin/bash
# manage-cve-allowlist.sh - Manage CVE exemptions

HARBOR_URL="https://harbor.example.com"
PROJECT_NAME="library"

# Add time-bound CVE exemption
EXPIRES_AT=$(date -d "+30 days" +%s)

curl -X PUT "${HARBOR_URL}/api/v2.0/projects/${PROJECT_NAME}" \
  -u "admin:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "cve_allowlist": {
      "items": [
        {
          "cve_id": "CVE-2023-45288"
        },
        {
          "cve_id": "CVE-2024-12345"
        }
      ],
      "expires_at": '${EXPIRES_AT}'
    }
  }'

# List current allowlist
curl -X GET "${HARBOR_URL}/api/v2.0/projects/${PROJECT_NAME}" \
  -u "admin:${HARBOR_PASSWORD}" | jq '.cve_allowlist'

Multi-Tier Severity Policies

# Project: production
metadata:
  auto_scan: "true"
  severity: "critical"        # Block CRITICAL only
  prevent_vul: "true"          # Enforce blocking
  enable_content_trust: "true" # Require signatures

# Project: staging
metadata:
  auto_scan: "true"
  severity: "high"             # Block HIGH and CRITICAL
  prevent_vul: "true"
  enable_content_trust: "false"

# Project: development
metadata:
  auto_scan: "true"
  severity: "none"             # Scan but don't block
  prevent_vul: "false"
  enable_content_trust: "false"

---

3. Automated Scanning Workflows

Scan on Push Configuration

# Enable scan-on-push globally
curl -X PUT "https://harbor.example.com/api/v2.0/configurations" \
  -u "admin:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "scan_all_policy": {
      "type": "daily",
      "parameter": {
        "daily_time": 0
      }
    }
  }'

Scheduled Rescanning

#!/bin/bash
# schedule-scans.sh - Configure periodic rescans

HARBOR_URL="https://harbor.example.com"

# Configure daily rescan at 2 AM UTC
curl -X POST "${HARBOR_URL}/api/v2.0/system/scanAll/schedule" \
  -u "admin:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "schedule": {
      "type": "Daily",
      "cron": "0 2 * * *"
    }
  }'

# Trigger manual scan for specific repository
PROJECT="library"
REPO="app"
TAG="v1.0.0"

curl -X POST "${HARBOR_URL}/api/v2.0/projects/${PROJECT}/repositories/${REPO}/artifacts/${TAG}/scan" \
  -u "admin:${HARBOR_PASSWORD}"

# Check scan status
curl -X GET "${HARBOR_URL}/api/v2.0/projects/${PROJECT}/repositories/${REPO}/artifacts/${TAG}" \
  -u "admin:${HARBOR_PASSWORD}" | jq '.scan_overview'

Bulk Scanning Script

#!/usr/bin/env python3
# bulk-scan.py - Scan all artifacts in a project

import requests
from requests.auth import HTTPBasicAuth
import os
import time

HARBOR_URL = "https://harbor.example.com"
USERNAME = "admin"
PASSWORD = os.environ["HARBOR_PASSWORD"]
PROJECT = "library"

auth = HTTPBasicAuth(USERNAME, PASSWORD)
headers = {"Content-Type": "application/json"}

# Get all repositories in project
repos_url = f"{HARBOR_URL}/api/v2.0/projects/{PROJECT}/repositories"
repos = requests.get(repos_url, auth=auth).json()

scanned = 0
failed = 0

for repo in repos:
    repo_name = repo["name"].split("/", 1)[1]

    # Get all artifacts
    artifacts_url = f"{HARBOR_URL}/api/v2.0/projects/{PROJECT}/repositories/{repo_name}/artifacts"
    artifacts = requests.get(artifacts_url, auth=auth).json()

    for artifact in artifacts:
        digest = artifact["digest"]

        # Trigger scan
        scan_url = f"{artifacts_url}/{digest}/scan"
        response = requests.post(scan_url, auth=auth, headers=headers)

        if response.status_code == 202:
            print(f"Scanning {PROJECT}/{repo_name}@{digest[:12]}")
            scanned += 1
        else:
            print(f"Failed to scan {PROJECT}/{repo_name}@{digest[:12]}")
            failed += 1

        time.sleep(0.5)  # Rate limiting

print(f"\nTotal scanned: {scanned}, Failed: {failed}")

---

4. Webhook Automation

Webhook Configuration for Scan Events

#!/bin/bash
# configure-webhooks.sh - Set up webhooks for security events

HARBOR_URL="https://harbor.example.com"
PROJECT="library"
SLACK_WEBHOOK="https://hooks.slack.com/services/XXX/YYY/ZZZ"

# Create webhook for scan completion
curl -X POST "${HARBOR_URL}/api/v2.0/projects/${PROJECT}/webhook/policies" \
  -u "admin:${HARBOR_PASSWORD}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "security-scan-alerts",
    "description": "Alert on vulnerability scan completion",
    "enabled": true,
    "event_types": [
      "SCANNING_COMPLETED",
      "SCANNING_FAILED",
      "SCANNING_STOPPED"
    ],
    "targets": [
      {
        "type": "http",
        "address": "'${SLACK_WEBHOOK}'",
        "skip_cert_verify": false,
        "payload_format": "CloudEvents",
        "auth_header": ""
      }
    ]
  }'

Webhook Payload Processing

#!/usr/bin/env python3
# webhook-processor.py - Process Harbor webhook events

from flask import Flask, request, jsonify
import json
import requests

app = Flask(__name__)

SLACK_WEBHOOK = "https://hooks.slack.com/services/XXX/YYY/ZZZ"
CVE_THRESHOLD = {
    "Critical": 0,  # Zero tolerance for critical
    "High": 5,      # Alert if >5 high CVEs
    "Medium": 20    # Alert if >20 medium CVEs
}

@app.route("/webhook/harbor/scan", methods=["POST"])
def handle_scan_webhook():
    event = request.json

    if event.get("type") == "harbor.scanning.completed":
        data = event.get("data", {})
        repo = data.get("repository")
        tag = data.get("tag", "")
        scan_overview = data.get("scan_overview", {})
        summary = scan_overview.get("summary", {})

        # Check thresholds
        alerts = []
        for severity, count in summary.items():
            threshold = CVE_THRESHOLD.get(severity, float('inf'))
            if count > threshold:
                alerts.append(f"{severity}: {count} (threshold: {threshold})")

        if alerts:
            message = {
                "text": f":warning: Vulnerability threshold exceeded: {repo}:{tag}",
                "blocks": [
                    {
                        "type": "section",
                        "text": {
                            "type": "mrkdwn",
                            "text": f"*Vulnerability Scan Alert*\n\n"
                                    f"*Image:* `{repo}:{tag}`\n"
                                    f"*Total CVEs:* {scan_overview.get('total_count', 0)}\n"
                                    f"*Fixable:* {scan_overview.get('fixable_count', 0)}\n\n"
                                    f"*Threshold Violations:*\n" + "\n".join([f"• {a}" for a in alerts])
                        }
                    }
                ]
            }
            requests.post(SLACK_WEBHOOK, json=message)

        return jsonify({"status": "processed"}), 200

    return jsonify({"status": "ignored"}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

Advanced Webhook Filtering

// webhook-filter.js - Node.js webhook processor with advanced filtering
const express = require('express');
const axios = require('axios');

const app = express();
app.use(express.json());

const JIRA_URL = 'https://jira.example.com';
const JIRA_TOKEN = process.env.JIRA_TOKEN;

// CVE severity to Jira priority mapping
const PRIORITY_MAP = {
  'Critical': 'Highest',
  'High': 'High',
  'Medium': 'Medium',
  'Low': 'Low'
};

app.post('/webhook/harbor/scan', async (req, res) => {
  const event = req.body;

  if (event.type === 'harbor.scanning.completed') {
    const { repository, tag, scan_overview } = event.data;
    const summary = scan_overview.summary;

    // Create Jira ticket for critical vulnerabilities
    if (summary.Critical > 0) {
      const ticket = {
        fields: {
          project: { key: 'SEC' },
          summary: `Critical CVEs in ${repository}:${tag}`,
          description: formatDescription(event.data),
          issuetype: { name: 'Security Issue' },
          priority: { name: 'Highest' },
          labels: ['harbor', 'cve', 'critical']
        }
      };

      await axios.post(`${JIRA_URL}/rest/api/2/issue`, ticket, {
        headers: {
          'Authorization': `Bearer ${JIRA_TOKEN}`,
          'Content-Type': 'application/json'
        }
      });
    }
  }

  res.status(200).json({ status: 'ok' });
});

function formatDescription(data) {
  return `
Image: ${data.repository}:${data.tag}
Total CVEs: ${data.scan_overview.total_count}
Fixable: ${data.scan_overview.fixable_count}

Severity Breakdown:
- Critical: ${data.scan_overview.summary.Critical || 0}
- High: ${data.scan_overview.summary.High || 0}
- Medium: ${data.scan_overview.summary.Medium || 0}
- Low: ${data.scan_overview.summary.Low || 0}

Scan completed at: ${data.scan_overview.end_time}
  `.trim();
}

app.listen(8080, () => console.log('Webhook processor running on port 8080'));

---

5. CVE Reporting and Metrics

Generate Compliance Reports

#!/usr/bin/env python3
# generate-cve-report.py - Generate vulnerability compliance report

import requests
from requests.auth import HTTPBasicAuth
import csv
from datetime import datetime
import os

HARBOR_URL = "https://harbor.example.com"
USERNAME = "admin"
PASSWORD = os.environ["HARBOR_PASSWORD"]

auth = HTTPBasicAuth(USERNAME, PASSWORD)

def get_all_artifacts(project):
    """Retrieve all artifacts from a project"""
    artifacts = []
    repos_url = f"{HARBOR_URL}/api/v2.0/projects/{project}/repositories"
    repos = requests.get(repos_url, auth=auth).json()

    for repo in repos:
        repo_name = repo["name"].split("/", 1)[1]
        artifacts_url = f"{HARBOR_URL}/api/v2.0/projects/{project}/repositories/{repo_name}/artifacts"
        repo_artifacts = requests.get(artifacts_url, auth=auth).json()

        for artifact in repo_artifacts:
            artifact["project"] = project
            artifact["repository"] = repo_name
            artifacts.append(artifact)

    return artifacts

def generate_report(projects, output_file):
    """Generate CSV report of vulnerabilities"""
    with open(output_file, 'w', newline='') as csvfile:
        fieldnames = ['project', 'repository', 'tag', 'digest', 'signed',
                      'critical', 'high', 'medium', 'low', 'total', 'fixable',
                      'scan_time', 'compliant']
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        for project in projects:
            artifacts = get_all_artifacts(project)

            for artifact in artifacts:
                tags = [t["name"] for t in artifact.get("tags", [])]
                tag = tags[0] if tags else "untagged"
                digest = artifact["digest"][:12]

                # Get scan overview
                scan = artifact.get("scan_overview", {})
                trivy_scan = scan.get("application/vnd.security.vulnerability.report; version=1.1", {})
                summary = trivy_scan.get("summary", {})

                # Check if signed
                signed = len(artifact.get("accessories", [])) > 0

                # Determine compliance
                critical = summary.get("Critical", 0)
                high = summary.get("High", 0)
                compliant = critical == 0 and high == 0 and signed

                writer.writerow({
                    'project': project,
                    'repository': artifact["repository"],
                    'tag': tag,
                    'digest': digest,
                    'signed': signed,
                    'critical': critical,
                    'high': high,
                    'medium': summary.get("Medium", 0),
                    'low': summary.get("Low", 0),
                    'total': trivy_scan.get("total_count", 0),
                    'fixable': trivy_scan.get("fixable_count", 0),
                    'scan_time': trivy_scan.get("end_time", "Never"),
                    'compliant': compliant
                })

if __name__ == "__main__":
    projects = ["production", "staging", "library"]
    output_file = f"harbor-cve-report-{datetime.now().strftime('%Y%m%d')}.csv"
    generate_report(projects, output_file)
    print(f"Report generated: {output_file}")

Track CVE Metrics

#!/bin/bash
# cve-metrics.sh - Calculate vulnerability metrics

HARBOR_URL="https://harbor.example.com"
PROJECT="production"

# Get all artifacts with scan results
ARTIFACTS=$(curl -s "${HARBOR_URL}/api/v2.0/projects/${PROJECT}/repositories" \
  -u "admin:${HARBOR_PASSWORD}" | jq -r '.[].name' | \
  while read REPO; do
    REPO_NAME=$(echo $REPO | cut -d'/' -f2)
    curl -s "${HARBOR_URL}/api/v2.0/projects/${PROJECT}/repositories/${REPO_NAME}/artifacts" \
      -u "admin:${HARBOR_PASSWORD}"
  done)

# Calculate metrics
TOTAL_ARTIFACTS=$(echo "$ARTIFACTS" | jq -s 'add | length')
SCANNED_ARTIFACTS=$(echo "$ARTIFACTS" | jq -s 'add | [.[] | select(.scan_overview != null)] | length')
CRITICAL_VULNS=$(echo "$ARTIFACTS" | jq -s 'add | [.[] | .scan_overview.summary.Critical // 0] | add')
HIGH_VULNS=$(echo "$ARTIFACTS" | jq -s 'add | [.[] | .scan_overview.summary.High // 0] | add')
SIGNED_ARTIFACTS=$(echo "$ARTIFACTS" | jq -s 'add | [.[] | select(.accessories != null and (.accessories | length > 0))] | length')

echo "=== Harbor CVE Metrics for ${PROJECT} ==="
echo "Total Artifacts: ${TOTAL_ARTIFACTS}"
echo "Scanned: ${SCANNED_ARTIFACTS}"
echo "Signed: ${SIGNED_ARTIFACTS}"
echo "Critical Vulnerabilities: ${CRITICAL_VULNS}"
echo "High Vulnerabilities: ${HIGH_VULNS}"
echo "Scan Coverage: $(( SCANNED_ARTIFACTS * 100 / TOTAL_ARTIFACTS ))%"
echo "Signature Coverage: $(( SIGNED_ARTIFACTS * 100 / TOTAL_ARTIFACTS ))%"

---

6. Integration with CI/CD

GitHub Actions Integration

# .github/workflows/harbor-scan.yml
name: Harbor Scan and Policy Check

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  build-and-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Login to Harbor
        uses: docker/login-action@v3
        with:
          registry: harbor.example.com
          username: ${{ secrets.HARBOR_USERNAME }}
          password: ${{ secrets.HARBOR_PASSWORD }}

      - name: Build image
        run: |
          docker build -t harbor.example.com/library/app:${{ github.sha }} .

      - name: Push to Harbor
        run: |
          docker push harbor.example.com/library/app:${{ github.sha }}

      - name: Wait for scan completion
        run: |
          SCAN_STATUS="pending"
          ATTEMPTS=0
          MAX_ATTEMPTS=60

          while [ "$SCAN_STATUS" != "Success" ] && [ $ATTEMPTS -lt $MAX_ATTEMPTS ]; do
            sleep 10
            SCAN_STATUS=$(curl -s -u "${{ secrets.HARBOR_USERNAME }}:${{ secrets.HARBOR_PASSWORD }}" \
              "https://harbor.example.com/api/v2.0/projects/library/repositories/app/artifacts/${{ github.sha }}" | \
              jq -r '.scan_overview | to_entries[0].value.scan_status // "pending"')
            echo "Scan status: $SCAN_STATUS"
            ATTEMPTS=$((ATTEMPTS + 1))
          done

          if [ "$SCAN_STATUS" != "Success" ]; then
            echo "Scan did not complete in time"
            exit 1
          fi

      - name: Check vulnerabilities
        run: |
          CRITICAL=$(curl -s -u "${{ secrets.HARBOR_USERNAME }}:${{ secrets.HARBOR_PASSWORD }}" \
            "https://harbor.example.com/api/v2.0/projects/library/repositories/app/artifacts/${{ github.sha }}" | \
            jq -r '.scan_overview | to_entries[0].value.summary.Critical // 0')

          HIGH=$(curl -s -u "${{ secrets.HARBOR_USERNAME }}:${{ secrets.HARBOR_PASSWORD }}" \
            "https://harbor.example.com/api/v2.0/projects/library/repositories/app/artifacts/${{ github.sha }}" | \
            jq -r '.scan_overview | to_entries[0].value.summary.High // 0')

          echo "Critical vulnerabilities: $CRITICAL"
          echo "High vulnerabilities: $HIGH"

          if [ $CRITICAL -gt 0 ]; then
            echo "CRITICAL vulnerabilities detected, blocking deployment"
            exit 1
          fi

          if [ $HIGH -gt 5 ]; then
            echo "Too many HIGH vulnerabilities detected, blocking deployment"
            exit 1
          fi

---

Summary

This reference provides production-ready patterns for:

• Trivy Integration: Complete scanner setup and configuration
• CVE Policies: Multi-tier severity enforcement and exemption management
• Automated Scanning: Scan-on-push, scheduled rescans, bulk operations
• Webhook Automation: Event-driven security notifications and ticket creation
• Compliance Reporting: CVE metrics, vulnerability tracking, audit reports
• CI/CD Integration: GitHub Actions workflows with scan verification

Use these patterns to implement comprehensive vulnerability management that protects your container supply chain while maintaining developer velocity.