#!/usr/bin/env python3
"""in-toto supply chain security agent.

Implements software supply chain verification using the in-toto framework.
Creates and verifies supply chain layouts, generates link metadata for
build steps, and validates that all steps were performed by authorized
functionaries with correct materials and products.
"""
import argparse
import json
import os
import subprocess
import sys
from datetime import datetime, timezone


def find_intoto_binary():
    """Locate in-toto CLI tools."""
    tools = {}
    for name in ["in-toto-run", "in-toto-verify", "in-toto-record", "in-toto-sign"]:
        for ext in ["", ".exe"]:
            for d in os.environ.get("PATH", "").split(os.pathsep):
                full = os.path.join(d, name + ext)
                if os.path.isfile(full):
                    tools[name] = full
                    break
    return tools


def create_layout_template(output_path, project_name, steps=None):
    """Generate a supply chain layout template."""
    if steps is None:
        steps = [
            {"name": "clone", "expected_command": ["git", "clone"],
             "threshold": 1, "materials": [], "products": ["src/*"]},
            {"name": "build", "expected_command": ["make"],
             "threshold": 1, "materials": ["src/*"], "products": ["dist/*"]},
            {"name": "test", "expected_command": ["make", "test"],
             "threshold": 1, "materials": ["src/*", "dist/*"], "products": []},
            {"name": "package", "expected_command": ["tar", "czf"],
             "threshold": 1, "materials": ["dist/*"], "products": ["*.tar.gz"]},
        ]

    layout = {
        "_type": "layout",
        "expires": (datetime.now(timezone.utc).replace(year=datetime.now().year + 1)).isoformat(),
        "readme": f"Supply chain layout for {project_name}",
        "steps": [],
        "inspect": [
            {
                "name": "verify-signature",
                "expected_materials": [["MATCH", "*.tar.gz", "WITH", "PRODUCTS", "FROM", "package"]],
                "expected_products": [],
                "run": ["sha256sum", "*.tar.gz"],
            }
        ],
        "keys": {},
    }

    for step in steps:
        layout["steps"].append({
            "name": step["name"],
            "expected_command": step.get("expected_command", []),
            "threshold": step.get("threshold", 1),
            "expected_materials": [
                ["MATCH", m, "WITH", "PRODUCTS", "FROM", steps[i-1]["name"]]
                if i > 0 else ["ALLOW", m]
                for i, m in enumerate(step.get("materials", []))
            ] or [["ALLOW", "*"]],
            "expected_products": [
                ["CREATE", p] for p in step.get("products", [])
            ] or [["ALLOW", "*"]],
            "pubkeys": [],
        })

    with open(output_path, "w") as f:
        json.dump(layout, f, indent=2)
    print(f"[+] Layout template written to {output_path}")
    return layout


def run_step(tools, step_name, key_path, command, materials=None, products=None):
    """Execute a supply chain step and record link metadata."""
    intoto_run = tools.get("in-toto-run")
    if not intoto_run:
        print("[!] in-toto-run not found", file=sys.stderr)
        return None

    cmd = [intoto_run, "--step-name", step_name, "--key", key_path]
    if materials:
        for m in materials:
            cmd.extend(["--materials", m])
    if products:
        for p in products:
            cmd.extend(["--products", p])
    cmd.append("--")
    cmd.extend(command)

    print(f"[*] Running step '{step_name}': {' '.join(command)}")
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    if result.returncode != 0:
        print(f"[!] Step failed: {result.stderr[:200]}", file=sys.stderr)
        return {"step": step_name, "status": "FAIL", "error": result.stderr[:200]}

    link_file = f"{step_name}.link"
    if os.path.isfile(link_file):
        print(f"[+] Link metadata: {link_file}")
        with open(link_file, "r") as f:
            link_data = json.load(f)
        return {"step": step_name, "status": "OK", "link_file": link_file,
                "materials_count": len(link_data.get("signed", {}).get("materials", {})),
                "products_count": len(link_data.get("signed", {}).get("products", {}))}
    return {"step": step_name, "status": "OK", "link_file": "generated"}


def verify_layout(tools, layout_path, layout_key_path, link_dir="."):
    """Verify the supply chain against the layout."""
    intoto_verify = tools.get("in-toto-verify")
    if not intoto_verify:
        print("[!] in-toto-verify not found", file=sys.stderr)
        return {"status": "FAIL", "error": "in-toto-verify not found"}

    cmd = [intoto_verify,
           "--layout", layout_path,
           "--layout-keys", layout_key_path,
           "--link-dir", link_dir]

    print(f"[*] Verifying supply chain layout: {layout_path}")
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    if result.returncode == 0:
        print(f"[+] Verification PASSED")
        return {"status": "PASS", "detail": "All steps verified successfully"}
    else:
        print(f"[!] Verification FAILED: {result.stderr[:300]}")
        return {"status": "FAIL", "detail": result.stderr[:300]}


def audit_existing_links(link_dir="."):
    """Audit existing link metadata files."""
    findings = []
    for fname in os.listdir(link_dir):
        if not fname.endswith(".link"):
            continue
        fpath = os.path.join(link_dir, fname)
        try:
            with open(fpath, "r") as f:
                link = json.load(f)
            signed = link.get("signed", {})
            step_name = signed.get("name", fname)
            materials = signed.get("materials", {})
            products = signed.get("products", {})
            command = signed.get("command", [])
            byproducts = signed.get("byproducts", {})

            findings.append({
                "link_file": fname,
                "step_name": step_name,
                "materials_count": len(materials),
                "products_count": len(products),
                "command": " ".join(command)[:80] if command else "N/A",
                "return_code": byproducts.get("return-value", "N/A"),
                "has_signature": bool(link.get("signatures")),
            })
        except (json.JSONDecodeError, IOError) as e:
            findings.append({"link_file": fname, "error": str(e)})

    return findings


def format_summary(results):
    """Print supply chain audit summary."""
    print(f"\n{'='*60}")
    print(f"  in-toto Supply Chain Security Report")
    print(f"{'='*60}")
    if isinstance(results, list):
        print(f"  Link Files Found: {len(results)}")
        for r in results:
            if "error" in r:
                print(f"    [ERR] {r['link_file']}: {r['error']}")
            else:
                sig = "signed" if r.get("has_signature") else "UNSIGNED"
                print(f"    [{sig:8s}] {r['step_name']:20s} | "
                      f"{r['materials_count']} materials, {r['products_count']} products | "
                      f"cmd: {r.get('command', 'N/A')[:40]}")


def main():
    parser = argparse.ArgumentParser(
        description="in-toto supply chain security agent"
    )
    sub = parser.add_subparsers(dest="command")

    p_layout = sub.add_parser("create-layout", help="Generate a layout template")
    p_layout.add_argument("--project", required=True, help="Project name")
    p_layout.add_argument("--output-path", default="root.layout", help="Layout output file")

    p_run = sub.add_parser("run-step", help="Execute a supply chain step")
    p_run.add_argument("--step-name", required=True)
    p_run.add_argument("--key", required=True, help="Functionary key path")
    p_run.add_argument("--materials", nargs="*")
    p_run.add_argument("--products", nargs="*")
    p_run.add_argument("cmd", nargs="+", help="Command to execute")

    p_verify = sub.add_parser("verify", help="Verify supply chain")
    p_verify.add_argument("--layout", required=True)
    p_verify.add_argument("--layout-key", required=True)
    p_verify.add_argument("--link-dir", default=".")

    p_audit = sub.add_parser("audit", help="Audit existing link metadata")
    p_audit.add_argument("--link-dir", default=".")

    parser.add_argument("--output", "-o", help="Output JSON report")
    parser.add_argument("--verbose", "-v", action="store_true")
    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    tools = find_intoto_binary()
    result = {}

    if args.command == "create-layout":
        layout = create_layout_template(args.output_path, args.project)
        result = {"action": "create-layout", "layout": layout}
    elif args.command == "run-step":
        step_result = run_step(tools, args.step_name, args.key,
                               args.cmd, args.materials, args.products)
        result = {"action": "run-step", "result": step_result}
    elif args.command == "verify":
        verify_result = verify_layout(tools, args.layout, args.layout_key, args.link_dir)
        result = {"action": "verify", "result": verify_result}
    elif args.command == "audit":
        link_findings = audit_existing_links(args.link_dir)
        format_summary(link_findings)
        result = {"action": "audit", "links": link_findings}

    report = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "tool": "in-toto",
        "result": result,
    }

    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)
        print(f"\n[+] Report saved to {args.output}")
    elif args.verbose:
        print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
in-toto Supply Chain Verification Tool

Verifies container image supply chain integrity by checking
in-toto link metadata against the defined layout policy.
"""

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


def compute_file_hash(file_path: str, algorithm: str = "sha256") -> str:
    """Compute the hash of a file."""
    h = hashlib.new(algorithm)
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            h.update(chunk)
    return h.hexdigest()


def find_link_files(link_dir: str) -> list[dict]:
    """Find and parse all in-toto link metadata files."""
    link_path = Path(link_dir)
    if not link_path.exists():
        print(f"[ERROR] Link directory not found: {link_dir}")
        return []

    links = []
    for link_file in link_path.glob("*.link"):
        try:
            with open(link_file) as f:
                data = json.load(f)
            links.append({
                "file": str(link_file),
                "step_name": data.get("signed", {}).get("name", "unknown"),
                "materials": data.get("signed", {}).get("materials", {}),
                "products": data.get("signed", {}).get("products", {}),
                "command": data.get("signed", {}).get("command", []),
                "byproducts": data.get("signed", {}).get("byproducts", {}),
                "signatures": data.get("signatures", []),
            })
        except (json.JSONDecodeError, KeyError) as e:
            print(f"[WARN] Failed to parse {link_file}: {e}")

    return links


def verify_artifact_chain(links: list[dict]) -> list[dict]:
    """Verify that artifact hashes chain correctly between steps."""
    findings = []
    products_by_step = {}

    for link in sorted(links, key=lambda x: x["step_name"]):
        products_by_step[link["step_name"]] = link["products"]

    for link in links:
        for material_path, material_hashes in link["materials"].items():
            found = False
            for step_name, products in products_by_step.items():
                if step_name == link["step_name"]:
                    continue
                if material_path in products:
                    product_hashes = products[material_path]
                    for algo, expected_hash in material_hashes.items():
                        actual_hash = product_hashes.get(algo, "")
                        if actual_hash and actual_hash != expected_hash:
                            findings.append({
                                "severity": "CRITICAL",
                                "type": "hash_mismatch",
                                "step": link["step_name"],
                                "artifact": material_path,
                                "expected": expected_hash[:16] + "...",
                                "actual": actual_hash[:16] + "...",
                                "description": f"Artifact {material_path} hash mismatch between steps"
                            })
                        elif actual_hash:
                            found = True
            if not found and link["materials"]:
                findings.append({
                    "severity": "WARNING",
                    "type": "untracked_material",
                    "step": link["step_name"],
                    "artifact": material_path,
                    "description": f"Material {material_path} not found in any prior step products"
                })

    return findings


def verify_signatures(links: list[dict], trusted_keys: list[str]) -> list[dict]:
    """Verify that all link signatures come from trusted keys."""
    findings = []
    for link in links:
        if not link["signatures"]:
            findings.append({
                "severity": "CRITICAL",
                "type": "missing_signature",
                "step": link["step_name"],
                "description": f"Step {link['step_name']} has no signatures"
            })
            continue

        for sig in link["signatures"]:
            keyid = sig.get("keyid", "")
            if trusted_keys and keyid not in trusted_keys:
                findings.append({
                    "severity": "HIGH",
                    "type": "untrusted_key",
                    "step": link["step_name"],
                    "keyid": keyid[:16] + "...",
                    "description": f"Step {link['step_name']} signed by untrusted key"
                })

    return findings


def check_required_steps(links: list[dict], required_steps: list[str]) -> list[dict]:
    """Check that all required pipeline steps have link metadata."""
    findings = []
    observed_steps = {link["step_name"] for link in links}

    for step in required_steps:
        if step not in observed_steps:
            findings.append({
                "severity": "CRITICAL",
                "type": "missing_step",
                "step": step,
                "description": f"Required step '{step}' has no link metadata"
            })

    return findings


def run_in_toto_verify(layout_path: str, layout_key: str, link_dir: str) -> dict:
    """Run the official in-toto-verify command."""
    try:
        result = subprocess.run(
            [
                "in-toto-verify",
                "--layout", layout_path,
                "--layout-key", layout_key,
                "--link-dir", link_dir,
            ],
            capture_output=True, text=True, timeout=60
        )
        return {
            "success": result.returncode == 0,
            "stdout": result.stdout.strip(),
            "stderr": result.stderr.strip(),
            "returncode": result.returncode
        }
    except FileNotFoundError:
        return {"success": False, "error": "in-toto-verify not found in PATH"}
    except subprocess.TimeoutExpired:
        return {"success": False, "error": "Verification timed out"}


def generate_report(links: list[dict], chain_findings: list[dict],
                    sig_findings: list[dict], step_findings: list[dict],
                    verify_result: dict, output_format: str = "text") -> str:
    """Generate a comprehensive verification report."""
    all_findings = chain_findings + sig_findings + step_findings
    critical_count = sum(1 for f in all_findings if f["severity"] == "CRITICAL")
    high_count = sum(1 for f in all_findings if f["severity"] == "HIGH")

    if output_format == "json":
        report = {
            "timestamp": datetime.utcnow().isoformat(),
            "verification_passed": verify_result.get("success", False) and critical_count == 0,
            "steps_found": len(links),
            "findings": {
                "critical": critical_count,
                "high": high_count,
                "total": len(all_findings),
                "details": all_findings
            },
            "in_toto_verify": verify_result,
            "steps": [{"name": l["step_name"], "command": l["command"],
                       "materials": len(l["materials"]), "products": len(l["products"])}
                      for l in links]
        }
        return json.dumps(report, indent=2)

    lines = []
    lines.append("=" * 70)
    lines.append("IN-TOTO SUPPLY CHAIN VERIFICATION REPORT")
    lines.append(f"Generated: {datetime.utcnow().isoformat()}")
    lines.append("=" * 70)

    passed = verify_result.get("success", False) and critical_count == 0
    lines.append(f"\nVerification Result: {'PASSED' if passed else 'FAILED'}")
    lines.append(f"Steps Found: {len(links)}")

    lines.append("\n## Pipeline Steps")
    for link in sorted(links, key=lambda x: x["step_name"]):
        lines.append(f"  Step: {link['step_name']}")
        lines.append(f"    Command: {' '.join(link['command'])}")
        lines.append(f"    Materials: {len(link['materials'])} | Products: {len(link['products'])}")
        lines.append(f"    Signatures: {len(link['signatures'])}")

    if all_findings:
        lines.append(f"\n## Findings ({len(all_findings)} total)")
        for f in sorted(all_findings, key=lambda x: x["severity"]):
            lines.append(f"  [{f['severity']}] {f['description']}")

    if verify_result.get("error"):
        lines.append(f"\n## in-toto-verify Error")
        lines.append(f"  {verify_result['error']}")

    lines.append("\n" + "=" * 70)
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="in-toto Supply Chain Verification Tool")
    parser.add_argument("--layout", help="Path to in-toto layout file")
    parser.add_argument("--layout-key", help="Path to layout signing public key")
    parser.add_argument("--link-dir", required=True, help="Directory containing link metadata")
    parser.add_argument("--required-steps", nargs="+", default=["checkout", "build", "scan"],
                        help="Required pipeline steps")
    parser.add_argument("--trusted-keys", nargs="+", default=[], help="Trusted key IDs")
    parser.add_argument("--format", choices=["text", "json"], default="text")
    args = parser.parse_args()

    links = find_link_files(args.link_dir)
    if not links:
        print("[ERROR] No link metadata found")
        sys.exit(1)

    chain_findings = verify_artifact_chain(links)
    sig_findings = verify_signatures(links, args.trusted_keys)
    step_findings = check_required_steps(links, args.required_steps)

    verify_result = {}
    if args.layout and args.layout_key:
        verify_result = run_in_toto_verify(args.layout, args.layout_key, args.link_dir)
    else:
        verify_result = {"success": None, "note": "Layout verification skipped (no --layout provided)"}

    report = generate_report(links, chain_findings, sig_findings, step_findings, verify_result, args.format)
    print(report)

    critical_count = sum(1 for f in chain_findings + sig_findings + step_findings if f["severity"] == "CRITICAL")
    sys.exit(1 if critical_count > 0 else 0)


if __name__ == "__main__":
    main()