undefined

```bash

---

undefined

```json
// 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": {
    "vulnerability_severity": "critical",
    "signature_enabled": true
  }
}

undefined

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

undefined

---

undefined

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

---

undefined

**Kyverno Policy to Verify Signatures**:
```yaml
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

undefined

---

undefined

**Webhook Payload Structure**:
```json
{
  "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
      }
    }
  }
}

undefined

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

@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"

undefined

undefined

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
    )

undefined

undefined

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

undefined

undefined

---

undefined

**Good** - Regular GC with untagged deletion:
```bash

undefined

undefined

**Good** - Filtered scheduled replication with bandwidth control:
```bash

undefined

undefined

**Good** - Harbor as proxy cache:
```bash

undefined

undefined

**Good** - Object storage with lifecycle policies:
```bash

undefined

undefined

**Good** - Optimized connection pool:
```bash

undefined

undefined

**Good** - Parallel scanning with optimized settings:
```bash

---

undefined

---

undefined

**Solution**:
```yaml

---

undefined

**Solution**:
```bash

---

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

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

undefined

**Solution**:
```bash

---

undefined

**Solution**:
```bash

---

undefined

**Solution**:
```yaml

---

undefined

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

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"]

undefined

undefined

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

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"

undefined

#!/bin/bash

undefined

undefined

---