#!/usr/bin/env python3
"""Agent for verifying container image provenance using Sigstore Cosign."""

import json
import argparse
import subprocess
from datetime import datetime


def run_cosign(args_list):
    """Run a cosign CLI command and return output."""
    cmd = ["cosign"] + args_list
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    return {
        "stdout": result.stdout.strip(),
        "stderr": result.stderr.strip(),
        "returncode": result.returncode,
    }


def verify_image(image_ref, key=None, certificate_identity=None, certificate_oidc_issuer=None):
    """Verify a container image signature with Cosign."""
    args = ["verify"]
    if key:
        args.extend(["--key", key])
    elif certificate_identity and certificate_oidc_issuer:
        args.extend(["--certificate-identity", certificate_identity,
                      "--certificate-oidc-issuer", certificate_oidc_issuer])
    args.append(image_ref)
    result = run_cosign(args)
    verified = result["returncode"] == 0
    attestations = []
    if verified and result["stdout"]:
        try:
            attestations = json.loads(result["stdout"])
        except json.JSONDecodeError:
            pass
    return {
        "image": image_ref,
        "verified": verified,
        "attestations": attestations if isinstance(attestations, list) else [attestations],
        "error": result["stderr"] if not verified else None,
    }


def verify_attestation(image_ref, predicate_type, key=None):
    """Verify an in-toto attestation attached to an image."""
    args = ["verify-attestation", "--type", predicate_type]
    if key:
        args.extend(["--key", key])
    args.append(image_ref)
    result = run_cosign(args)
    return {
        "image": image_ref,
        "predicate_type": predicate_type,
        "verified": result["returncode"] == 0,
        "output": result["stdout"][:500] if result["stdout"] else None,
        "error": result["stderr"] if result["returncode"] != 0 else None,
    }


def sign_image(image_ref, key=None, keyless=False):
    """Sign a container image with Cosign."""
    args = ["sign"]
    if key:
        args.extend(["--key", key])
    elif keyless:
        args.append("--yes")
    args.append(image_ref)
    result = run_cosign(args)
    return {
        "image": image_ref,
        "signed": result["returncode"] == 0,
        "error": result["stderr"] if result["returncode"] != 0 else None,
    }


def triangulate_image(image_ref):
    """Get the signature and attestation image references."""
    result = run_cosign(["triangulate", image_ref])
    return {
        "image": image_ref,
        "signature_ref": result["stdout"] if result["returncode"] == 0 else None,
    }


def audit_registry_images(images_list, key=None, identity=None, issuer=None):
    """Audit multiple container images for valid signatures."""
    results = []
    for image in images_list:
        result = verify_image(image, key=key, certificate_identity=identity,
                              certificate_oidc_issuer=issuer)
        result["severity"] = "INFO" if result["verified"] else "HIGH"
        results.append(result)
    signed = sum(1 for r in results if r["verified"])
    return {
        "total_images": len(results),
        "signed": signed,
        "unsigned": len(results) - signed,
        "signing_rate": round(signed / len(results) * 100, 1) if results else 0,
        "details": results,
    }


def generate_kyverno_policy(image_patterns, key=None, identity=None, issuer=None):
    """Generate Kyverno ClusterPolicy for image verification."""
    policy = {
        "apiVersion": "kyverno.io/v1",
        "kind": "ClusterPolicy",
        "metadata": {"name": "verify-image-signatures"},
        "spec": {
            "validationFailureAction": "Enforce",
            "webhookTimeoutSeconds": 30,
            "rules": [{
                "name": "verify-cosign-signature",
                "match": {"any": [{"resources": {"kinds": ["Pod"]}}]},
                "verifyImages": [{
                    "imageReferences": image_patterns,
                    "attestors": [{
                        "entries": [{
                            "keyless": {
                                "subject": identity or "*",
                                "issuer": issuer or "https://token.actions.githubusercontent.com",
                            }
                        }] if not key else [{"keys": {"publicKeys": key}}]
                    }],
                }],
            }],
        },
    }
    return policy


def generate_cosign_ci_workflow(image_ref, registry):
    """Generate GitHub Actions workflow for Cosign signing."""
    return f"""name: Sign Container Image
on:
  push:
    branches: [main]
jobs:
  sign:
    runs-on: ubuntu-latest
    permissions:
      contents: read
      id-token: write
      packages: write
    steps:
      - uses: actions/checkout@v4
      - uses: sigstore/cosign-installer@v3
      - name: Login to registry
        uses: docker/login-action@v3
        with:
          registry: {registry}
      - name: Build and push
        run: |
          docker build -t {image_ref} .
          docker push {image_ref}
      - name: Sign image (keyless)
        run: cosign sign --yes {image_ref}
      - name: Verify signature
        run: cosign verify --certificate-identity-regexp=".*" --certificate-oidc-issuer="https://token.actions.githubusercontent.com" {image_ref}
"""


def main():
    parser = argparse.ArgumentParser(description="Cosign Image Provenance Agent")
    parser.add_argument("--verify", help="Image to verify")
    parser.add_argument("--sign", help="Image to sign")
    parser.add_argument("--audit", nargs="+", help="Images to audit for signatures")
    parser.add_argument("--key", help="Cosign public key for verification")
    parser.add_argument("--identity", help="Certificate identity for keyless verification")
    parser.add_argument("--issuer", help="OIDC issuer for keyless verification")
    parser.add_argument("--gen-policy", nargs="+", help="Image patterns for Kyverno policy")
    parser.add_argument("--gen-workflow", help="Image ref for CI workflow generation")
    parser.add_argument("--output", default="cosign_provenance_report.json")
    args = parser.parse_args()

    report = {"generated_at": datetime.utcnow().isoformat(), "results": {}}

    if args.verify:
        result = verify_image(args.verify, key=args.key,
                              certificate_identity=args.identity,
                              certificate_oidc_issuer=args.issuer)
        report["results"]["verification"] = result
        status = "VERIFIED" if result["verified"] else "FAILED"
        print(f"[+] {args.verify}: {status}")

    if args.audit:
        result = audit_registry_images(args.audit, key=args.key,
                                       identity=args.identity, issuer=args.issuer)
        report["results"]["audit"] = result
        print(f"[+] Audit: {result['signed']}/{result['total_images']} signed")

    if args.gen_policy:
        policy = generate_kyverno_policy(args.gen_policy, key=args.key,
                                          identity=args.identity, issuer=args.issuer)
        report["results"]["kyverno_policy"] = policy
        print("[+] Kyverno policy generated")

    if args.gen_workflow:
        workflow = generate_cosign_ci_workflow(args.gen_workflow, "ghcr.io")
        report["results"]["workflow"] = workflow
        print("[+] CI workflow generated")

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Cosign Image Provenance Manager - Sign, verify, and audit container
image signatures using Sigstore Cosign.
"""

import json
import subprocess
import sys
import argparse
from pathlib import Path


def run_cosign(args: list) -> dict:
    """Execute cosign command and return output."""
    cmd = ["cosign"] + args
    result = subprocess.run(cmd, capture_output=True, text=True)
    return {
        "stdout": result.stdout,
        "stderr": result.stderr,
        "returncode": result.returncode,
    }


def sign_image(image: str, key: str = None, annotations: dict = None, keyless: bool = False) -> bool:
    """Sign a container image."""
    args = ["sign"]
    if key:
        args.extend(["--key", key])
    if keyless:
        args.append("--yes")
    if annotations:
        for k, v in annotations.items():
            args.extend(["-a", f"{k}={v}"])
    args.append(image)

    result = run_cosign(args)
    if result["returncode"] == 0:
        print(f"Successfully signed: {image}")
        return True
    else:
        print(f"Failed to sign: {result['stderr']}", file=sys.stderr)
        return False


def verify_image(image: str, key: str = None, identity: str = None,
                 issuer: str = None) -> dict:
    """Verify a container image signature."""
    args = ["verify"]
    if key:
        args.extend(["--key", key])
    if identity:
        args.extend(["--certificate-identity", identity])
    if issuer:
        args.extend(["--certificate-oidc-issuer", issuer])
    args.append(image)

    result = run_cosign(args)
    verified = result["returncode"] == 0

    signatures = []
    if verified and result["stdout"].strip():
        try:
            signatures = json.loads(result["stdout"])
        except json.JSONDecodeError:
            pass

    return {
        "image": image,
        "verified": verified,
        "signatures": signatures,
        "error": result["stderr"] if not verified else None,
    }


def verify_attestation(image: str, att_type: str, key: str = None,
                       identity: str = None, issuer: str = None) -> dict:
    """Verify an attestation on a container image."""
    args = ["verify-attestation", "--type", att_type]
    if key:
        args.extend(["--key", key])
    if identity:
        args.extend(["--certificate-identity", identity])
    if issuer:
        args.extend(["--certificate-oidc-issuer", issuer])
    args.append(image)

    result = run_cosign(args)
    return {
        "image": image,
        "type": att_type,
        "verified": result["returncode"] == 0,
        "output": result["stdout"],
        "error": result["stderr"] if result["returncode"] != 0 else None,
    }


def audit_images(images: list, key: str = None, identity: str = None,
                 issuer: str = None) -> list:
    """Audit multiple images for valid signatures."""
    results = []
    for image in images:
        result = verify_image(image, key=key, identity=identity, issuer=issuer)
        results.append(result)
    return results


def generate_report(audit_results: list) -> str:
    """Generate markdown audit report."""
    signed = sum(1 for r in audit_results if r["verified"])
    total = len(audit_results)

    report = f"""# Image Signature Audit Report

**Total Images:** {total}
**Signed:** {signed}
**Unsigned:** {total - signed}

## Results

| Image | Signed | Signatures | Error |
|-------|--------|------------|-------|
"""
    for r in audit_results:
        status = "YES" if r["verified"] else "NO"
        sig_count = len(r.get("signatures", []))
        error = r.get("error", "")[:50] if r.get("error") else "-"
        report += f"| `{r['image']}` | {status} | {sig_count} | {error} |\n"

    return report


def main():
    parser = argparse.ArgumentParser(description="Cosign Image Provenance Manager")
    subparsers = parser.add_subparsers(dest="command")

    sign_cmd = subparsers.add_parser("sign", help="Sign an image")
    sign_cmd.add_argument("image", help="Image reference")
    sign_cmd.add_argument("--key", help="Signing key path")
    sign_cmd.add_argument("--keyless", action="store_true", help="Use keyless signing")
    sign_cmd.add_argument("--annotation", "-a", action="append", help="key=value annotations")

    verify_cmd = subparsers.add_parser("verify", help="Verify image signature")
    verify_cmd.add_argument("image", help="Image reference")
    verify_cmd.add_argument("--key", help="Public key path")
    verify_cmd.add_argument("--identity", help="Certificate identity")
    verify_cmd.add_argument("--issuer", help="OIDC issuer")

    audit_cmd = subparsers.add_parser("audit", help="Audit multiple images")
    audit_cmd.add_argument("--images-file", required=True, help="File with image refs (one per line)")
    audit_cmd.add_argument("--key", help="Public key path")
    audit_cmd.add_argument("--identity", help="Certificate identity")
    audit_cmd.add_argument("--issuer", help="OIDC issuer")
    audit_cmd.add_argument("--report", help="Output report path")

    args = parser.parse_args()

    if args.command == "sign":
        annotations = {}
        if args.annotation:
            for a in args.annotation:
                k, v = a.split("=", 1)
                annotations[k] = v
        sign_image(args.image, key=args.key, annotations=annotations,
                  keyless=args.keyless)

    elif args.command == "verify":
        result = verify_image(args.image, key=args.key,
                            identity=args.identity, issuer=args.issuer)
        print(json.dumps(result, indent=2))
        sys.exit(0 if result["verified"] else 1)

    elif args.command == "audit":
        images = Path(args.images_file).read_text().strip().split("\n")
        results = audit_images(images, key=args.key,
                             identity=args.identity, issuer=args.issuer)
        report = generate_report(results)
        if args.report:
            Path(args.report).write_text(report)
            print(f"Report written to {args.report}")
        else:
            print(report)

    else:
        parser.print_help()


if __name__ == "__main__":
    main()