#!/usr/bin/env python3
"""Agent for securing container registry with Harbor.

Audits Harbor registry security configuration including RBAC,
vulnerability scanning policies, content trust, immutable tags,
and OIDC authentication via Harbor REST API v2.0.
"""

import json
import sys
from pathlib import Path
from datetime import datetime

try:
    import requests
except ImportError:
    requests = None


class HarborSecurityAgent:
    """Audits and hardens Harbor container registry security."""

    def __init__(self, harbor_url, username="admin", password="",
                 output_dir="./harbor_audit"):
        self.base_url = harbor_url.rstrip("/") + "/api/v2.0"
        self.auth = (username, password)
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.findings = []

    def _get(self, path, params=None):
        if not requests:
            return {"error": "requests library required"}
        resp = requests.get(f"{self.base_url}{path}", auth=self.auth,
                            params=params, verify=True, timeout=15)
        try:
            return resp.json()
        except (json.JSONDecodeError, ValueError):
            return {"status": resp.status_code}

    def audit_projects(self):
        """Audit all projects for security configuration."""
        projects = self._get("/projects", {"page_size": 100})
        if isinstance(projects, dict) and "error" in projects:
            return projects

        results = []
        for p in projects:
            meta = p.get("metadata", {})
            name = p.get("name", "")
            issues = []

            if meta.get("public") == "true":
                issues.append("Project is public - images accessible without auth")
            if meta.get("auto_scan") != "true":
                issues.append("Auto-scan on push not enabled")
            if meta.get("prevent_vul") != "true":
                issues.append("Vulnerability prevention not enabled")
            if meta.get("enable_content_trust") != "true":
                issues.append("Content trust (Notary) not enabled")
            if meta.get("enable_content_trust_cosign") != "true":
                issues.append("Cosign content trust not enabled")

            for issue in issues:
                self.findings.append({
                    "severity": "high" if "public" in issue or "prevent_vul" in issue else "medium",
                    "project": name,
                    "issue": issue,
                })

            results.append({
                "name": name,
                "public": meta.get("public"),
                "auto_scan": meta.get("auto_scan"),
                "prevent_vul": meta.get("prevent_vul"),
                "content_trust": meta.get("enable_content_trust"),
                "cosign": meta.get("enable_content_trust_cosign"),
                "issues": issues,
            })
        return results

    def audit_system_config(self):
        """Check system-level security configuration."""
        config = self._get("/configurations")
        if isinstance(config, dict) and "error" in config:
            return config

        checks = []
        auth_mode = config.get("auth_mode", {}).get("value", "db_auth")
        if auth_mode == "db_auth":
            self.findings.append({
                "severity": "medium",
                "issue": "Using local database auth instead of OIDC/LDAP",
            })
            checks.append({"check": "auth_mode", "value": auth_mode, "status": "WARN"})
        else:
            checks.append({"check": "auth_mode", "value": auth_mode, "status": "OK"})

        self_reg = config.get("self_registration", {}).get("value", True)
        if self_reg:
            self.findings.append({
                "severity": "high",
                "issue": "Self-registration enabled - anyone can create accounts",
            })
        checks.append({"check": "self_registration", "value": self_reg,
                        "status": "FAIL" if self_reg else "OK"})

        return checks

    def list_project_members(self, project_name):
        """List members and their roles for a project."""
        members = self._get(f"/projects/{project_name}/members")
        if isinstance(members, dict) and "error" in members:
            return members
        role_map = {1: "ProjectAdmin", 2: "Maintainer", 3: "Developer",
                    4: "Guest", 5: "LimitedGuest"}
        return [
            {"username": m.get("entity_name", ""), "role": role_map.get(m.get("role_id"), "Unknown")}
            for m in (members if isinstance(members, list) else [])
        ]

    def check_immutable_tags(self, project_name):
        """Check immutable tag rules for a project."""
        rules = self._get(f"/projects/{project_name}/immutabletagrules")
        if not rules or (isinstance(rules, dict) and "error" in rules):
            self.findings.append({
                "severity": "medium",
                "project": project_name,
                "issue": "No immutable tag rules configured",
            })
            return []
        return rules

    def audit_logs(self, page_size=20):
        """Retrieve recent audit log entries."""
        logs = self._get("/audit-logs", {"page_size": page_size})
        if isinstance(logs, dict) and "error" in logs:
            return logs
        return [
            {"operation": l.get("operation"), "resource": l.get("resource"),
             "username": l.get("username"), "time": l.get("op_time")}
            for l in (logs if isinstance(logs, list) else [])
        ]

    def generate_report(self):
        projects = self.audit_projects()
        sys_config = self.audit_system_config()

        report = {
            "report_date": datetime.utcnow().isoformat(),
            "harbor_url": self.base_url.replace("/api/v2.0", ""),
            "projects_audited": len(projects) if isinstance(projects, list) else 0,
            "system_config": sys_config,
            "projects": projects,
            "findings": self.findings,
            "finding_count": len(self.findings),
        }
        out = self.output_dir / "harbor_audit_report.json"
        with open(out, "w") as f:
            json.dump(report, f, indent=2)
        print(json.dumps(report, indent=2))
        return report


def main():
    if len(sys.argv) < 2:
        print("Usage: agent.py <harbor_url> [--user admin] [--pass Harbor12345]")
        sys.exit(1)

    url = sys.argv[1]
    user = "admin"
    password = "Harbor12345"
    if "--user" in sys.argv:
        user = sys.argv[sys.argv.index("--user") + 1]
    if "--pass" in sys.argv:
        password = sys.argv[sys.argv.index("--pass") + 1]

    agent = HarborSecurityAgent(url, user, password)
    agent.generate_report()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Harbor Container Registry Security Auditor

Audits Harbor registry configuration for security best practices
including scanning policies, content trust, RBAC, and TLS.
"""

import json
import sys
import urllib.request
import urllib.error
import ssl
import base64
from dataclasses import dataclass, field


@dataclass
class HarborFinding:
    category: str
    title: str
    severity: str
    details: str
    remediation: str


@dataclass
class HarborAuditReport:
    findings: list = field(default_factory=list)
    harbor_url: str = ""
    projects_audited: int = 0


def harbor_api_call(base_url: str, endpoint: str, username: str, password: str) -> dict:
    """Call Harbor API and return JSON response."""
    url = f"{base_url}/api/v2.0{endpoint}"
    credentials = base64.b64encode(f"{username}:{password}".encode()).decode()

    req = urllib.request.Request(url)
    req.add_header("Authorization", f"Basic {credentials}")
    req.add_header("Content-Type", "application/json")

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    try:
        with urllib.request.urlopen(req, context=ctx, timeout=10) as response:
            return json.loads(response.read().decode())
    except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError) as e:
        print(f"[!] API call failed for {endpoint}: {e}")
        return {}


def audit_projects(base_url: str, username: str, password: str, report: HarborAuditReport):
    """Audit all projects for security configurations."""
    projects = harbor_api_call(base_url, "/projects?page=1&page_size=100", username, password)
    if not projects:
        return

    for project in projects:
        name = project.get("name", "unknown")
        metadata = project.get("metadata", {})
        report.projects_audited += 1

        # Check auto-scan
        if metadata.get("auto_scan") != "true":
            report.findings.append(HarborFinding(
                category="Scanning",
                title=f"Auto-scan disabled for project: {name}",
                severity="HIGH",
                details="Images pushed to this project are not automatically scanned",
                remediation=f"Enable auto_scan for project '{name}'"
            ))

        # Check vulnerability prevention
        if metadata.get("prevent_vul") != "true":
            report.findings.append(HarborFinding(
                category="Scanning",
                title=f"Vulnerability prevention disabled for project: {name}",
                severity="HIGH",
                details="Vulnerable images can be pulled from this project",
                remediation=f"Enable prevent_vul for project '{name}'"
            ))

        # Check content trust
        if metadata.get("enable_content_trust") != "true" and metadata.get("enable_content_trust_cosign") != "true":
            report.findings.append(HarborFinding(
                category="Content Trust",
                title=f"Content trust not enforced for project: {name}",
                severity="MEDIUM",
                details="Unsigned images can be used from this project",
                remediation=f"Enable content trust (Cosign) for project '{name}'"
            ))

        # Check public visibility
        if metadata.get("public") == "true":
            report.findings.append(HarborFinding(
                category="Access Control",
                title=f"Project is publicly accessible: {name}",
                severity="MEDIUM",
                details="Anyone can pull images from this project",
                remediation=f"Set project '{name}' to private unless intentionally public"
            ))


def audit_system_config(base_url: str, username: str, password: str, report: HarborAuditReport):
    """Audit system-level Harbor configuration."""
    config = harbor_api_call(base_url, "/configurations", username, password)
    if not config:
        return

    # Check auth mode
    auth_mode = config.get("auth_mode", {}).get("value", "db_auth")
    if auth_mode == "db_auth":
        report.findings.append(HarborFinding(
            category="Authentication",
            title="Using local database authentication",
            severity="MEDIUM",
            details="Harbor uses local DB auth instead of enterprise IdP",
            remediation="Configure OIDC or LDAP authentication"
        ))

    # Check self-registration
    self_reg = config.get("self_registration", {}).get("value", False)
    if self_reg:
        report.findings.append(HarborFinding(
            category="Authentication",
            title="Self-registration is enabled",
            severity="HIGH",
            details="Anyone can create an account on this Harbor instance",
            remediation="Disable self-registration in Harbor configuration"
        ))


def print_report(report: HarborAuditReport):
    print("\n" + "=" * 70)
    print("HARBOR REGISTRY SECURITY AUDIT")
    print("=" * 70)
    print(f"Harbor URL:         {report.harbor_url}")
    print(f"Projects Audited:   {report.projects_audited}")
    print(f"Findings:           {len(report.findings)}")
    print("=" * 70)

    for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW"]:
        findings = [f for f in report.findings if f.severity == sev]
        if findings:
            print(f"\n{sev} ({len(findings)}):")
            for f in findings:
                print(f"  [{f.category}] {f.title}")
                print(f"    Fix: {f.remediation}")


def main():
    import argparse
    parser = argparse.ArgumentParser(description="Harbor Registry Security Auditor")
    parser.add_argument("--url", required=True, help="Harbor URL (e.g., https://harbor.example.com)")
    parser.add_argument("--username", default="admin", help="Harbor admin username")
    parser.add_argument("--password", required=True, help="Harbor admin password")
    args = parser.parse_args()

    report = HarborAuditReport(harbor_url=args.url)

    print(f"[*] Auditing Harbor at {args.url}")
    audit_projects(args.url, args.username, args.password, report)
    audit_system_config(args.url, args.username, args.password, report)
    print_report(report)

    with open("harbor_audit_report.json", "w") as f:
        json.dump({
            "harbor_url": report.harbor_url,
            "findings": [{"category": f.category, "title": f.title,
                         "severity": f.severity, "remediation": f.remediation}
                        for f in report.findings]
        }, f, indent=2)
    print("\n[*] Report saved to harbor_audit_report.json")


if __name__ == "__main__":
    main()