{"uuid": "0f95e496-c31e-43fc-a4dd-9cba51d1277c", "vulnerability_lookup_origin": "1a89b78e-f703-45f3-bb86-59eb712668bd", "author": "9f56dd64-161d-43a6-b9c3-555944290a09", "vulnerability": "CVE-2026-66066", "type": "seen", "source": "https://gist.github.com/cskartikey/e1db6fa823980f7d870432de59e29462", "content": "#!/usr/bin/env python3\n\"\"\"Fast Cloudflare R2 scan for CVE-2026-66066 / GHSA-xr9x-r78c-5hrm uploads.\n\nExample: scan_r2_cve2026_66066.py --workers 100 --jsonl r2-hits.jsonl\nCredentials via environment:\n\n    export R2_ACCOUNT_ID=...\n    export R2_ACCESS_KEY_ID=...\n    export R2_SECRET_ACCESS_KEY=...\n    export R2_BUCKET=...\n\n    python3 -m pip install boto3\n    python3 scan_r2_cve2026_66066.py\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport json\nimport os\nimport sys\nimport threading\nfrom concurrent.futures import ThreadPoolExecutor, as_completed\nfrom dataclasses import asdict, dataclass\n\ntry:\n    import boto3\n    from botocore.client import Config\n    from botocore.exceptions import ClientError\nexcept ModuleNotFoundError:\n    print(\"error: install boto3: python3 -m pip install boto3\", file=sys.stderr)\n    raise SystemExit(2)\n\n\nMATLAB_PREFIX = b\"MATLAB 5.0\"\nHDF5_SIGNATURE = b\"\\x89HDF\\r\\n\\x1a\\n\"\nHDF5_OFFSET = 512\nPOC_TRAILER_MAGIC = b\"RAILS_GHSA_OAST_PAYLOAD_V1\"\nBMP_MAGIC = b\"BM\"\nTHREAD_LOCAL = threading.local()\n\n\n@dataclass\nclass Hit:\n    key: str\n    size: int\n    etag: str\n    reasons: list[str]\n    head_hex: str\n    content_type: str = \"\"\n    tail_hex: str = \"\"\n\n\ndef parse_args() -> argparse.Namespace:\n    parser = argparse.ArgumentParser(\n        description=(\n            \"Parallel R2 byte-sample scan for MATLAB/HDF5 Active Storage \"\n            \"exploit uploads (CVE-2026-66066).\"\n        )\n    )\n    parser.add_argument(\n        \"--bucket\",\n        default=os.environ.get(\"R2_BUCKET\"),\n        help=\"R2 bucket name (or set R2_BUCKET)\",\n    )\n    parser.add_argument(\n        \"--prefix\",\n        default=os.environ.get(\"R2_PREFIX\", \"\"),\n        help=\"optional key prefix (Active Storage root prefix, if any)\",\n    )\n    parser.add_argument(\n        \"--endpoint\",\n        default=os.environ.get(\"R2_ENDPOINT\"),\n        help=\"R2 S3 API endpoint (default derived from R2_ACCOUNT_ID)\",\n    )\n    parser.add_argument(\n        \"--head-bytes\",\n        type=int,\n        default=768,\n        help=\"bytes to fetch from object start (default: %(default)s)\",\n    )\n    parser.add_argument(\n        \"--tail-bytes\",\n        type=int,\n        default=1024,\n        help=\"bytes to fetch from object end only on head hits (default: %(default)s)\",\n    )\n    parser.add_argument(\n        \"--min-size\",\n        type=int,\n        default=1024,\n        help=(\n            \"skip objects smaller than this (default: %(default)s; \"\n            \"lab artifact is ~2-4 KiB)\"\n        ),\n    )\n    parser.add_argument(\n        \"--max-size\",\n        type=int,\n        default=256 * 1024,\n        help=(\n            \"skip objects larger than this (default: 256KiB). Observed prod \"\n            \"MATLAB probes were ~74KiB declared as image/png.\"\n        ),\n    )\n    parser.add_argument(\n        \"--workers\",\n        type=int,\n        default=32,\n        help=\"parallel Range-GET workers (default: %(default)s)\",\n    )\n    parser.add_argument(\n        \"--max-objects\",\n        type=int,\n        default=0,\n        help=\"stop after sampling N objects (0 = no limit)\",\n    )\n    parser.add_argument(\n        \"--jsonl\",\n        type=str,\n        default=\"\",\n        help=\"optional path to write hit records as JSONL\",\n    )\n    return parser.parse_args()\n\n\ndef require_env(name: str) -> str:\n    value = os.environ.get(name, \"\").strip()\n    if not value:\n        raise SystemExit(f\"error: set {name} in the environment\")\n    return value\n\n\ndef make_client(endpoint: str | None):\n    account_id = os.environ.get(\"R2_ACCOUNT_ID\", \"\").strip()\n    access_key = require_env(\"R2_ACCESS_KEY_ID\")\n    secret_key = require_env(\"R2_SECRET_ACCESS_KEY\")\n    if not endpoint:\n        if not account_id:\n            raise SystemExit(\n                \"error: set R2_ENDPOINT or R2_ACCOUNT_ID in the environment\"\n            )\n        endpoint = f\"https://{account_id}.r2.cloudflarestorage.com\"\n    return boto3.client(\n        \"s3\",\n        endpoint_url=endpoint,\n        aws_access_key_id=access_key,\n        aws_secret_access_key=secret_key,\n        region_name=os.environ.get(\"AWS_REGION\", \"auto\"),\n        config=Config(\n            signature_version=\"s3v4\",\n            max_pool_connections=64,\n            retries={\"max_attempts\": 3, \"mode\": \"standard\"},\n        ),\n    )\n\n\ndef client_for_thread(endpoint: str | None):\n    client = getattr(THREAD_LOCAL, \"client\", None)\n    if client is None:\n        client = make_client(endpoint)\n        THREAD_LOCAL.client = client\n    return client\n\n\ndef get_range(client, bucket: str, key: str, start: int, end: int) -> bytes:\n    response = client.get_object(\n        Bucket=bucket,\n        Key=key,\n        Range=f\"bytes={start}-{end}\",\n    )\n    return response[\"Body\"].read()\n\n\ndef analyze_head(key: str, size: int, etag: str, head: bytes) -> Hit | None:\n    reasons: list[str] = []\n\n    if head.startswith(MATLAB_PREFIX):\n        reasons.append(\"matlab_5_header\")\n    if len(head) >= HDF5_OFFSET + len(HDF5_SIGNATURE):\n        if head[HDF5_OFFSET : HDF5_OFFSET + len(HDF5_SIGNATURE)] == HDF5_SIGNATURE:\n            reasons.append(\"hdf5_sig_at_512\")\n    if POC_TRAILER_MAGIC in head:\n        reasons.append(\"poc_trailer_magic\")\n    for marker in (b\"MATLAB_class\", b\"/proc/\", b\"SECRET_KEY_BASE\"):\n        if marker in head:\n            reasons.append(f\"marker:{marker.decode('ascii', 'replace')}\")\n\n    # Declared BMP is checked only if we later learn content-type; for raw\n    # bytes, a non-BM object that still looks like MAT/HDF5 is enough.\n    if \"matlab_5_header\" in reasons and \"hdf5_sig_at_512\" in reasons:\n        reasons.append(\"mat_hdf5_hybrid_layout\")\n\n    if not reasons:\n        return None\n\n    return Hit(\n        key=key,\n        size=size,\n        etag=etag.strip('\"'),\n        reasons=sorted(set(reasons)),\n        head_hex=head[:64].hex(),\n    )\n\n\ndef enrich_tail(\n    client,\n    bucket: str,\n    hit: Hit,\n    size: int,\n    head: bytes,\n    tail_bytes: int,\n) -> Hit:\n    if tail_bytes <= 0:\n        return hit\n    if size <= len(head):\n        tail = head\n    else:\n        start = max(0, size - tail_bytes)\n        try:\n            tail = get_range(client, bucket, hit.key, start, size - 1)\n        except ClientError as error:\n            print(f\"warn: tail get failed key={hit.key}: {error}\", file=sys.stderr)\n            return hit\n\n    extra: list[str] = []\n    if POC_TRAILER_MAGIC in tail:\n        extra.append(\"poc_trailer_magic\")\n    for marker in (b\"MATLAB_class\", b\"/proc/\", b\"SECRET_KEY_BASE\", b\"rails_ghsa\"):\n        if marker in tail:\n            extra.append(f\"marker:{marker.decode('ascii', 'replace')}\")\n    if extra:\n        hit.reasons = sorted(set(hit.reasons) | set(extra))\n    hit.tail_hex = tail[-64:].hex()\n    return hit\n\n\ndef iter_candidates(client, bucket: str, prefix: str, min_size: int, max_size: int):\n    token = None\n    listed = 0\n    kept = 0\n    while True:\n        kwargs = {\"Bucket\": bucket, \"Prefix\": prefix, \"MaxKeys\": 1000}\n        if token:\n            kwargs[\"ContinuationToken\"] = token\n        page = client.list_objects_v2(**kwargs)\n        for item in page.get(\"Contents\", []):\n            listed += 1\n            size = int(item[\"Size\"])\n            if size < min_size or size > max_size:\n                continue\n            kept += 1\n            yield item\n        if not page.get(\"IsTruncated\"):\n            break\n        token = page.get(\"NextContinuationToken\")\n        print(\n            f\"list progress listed={listed} candidates={kept}\",\n            file=sys.stderr,\n            flush=True,\n        )\n    print(\n        f\"list done listed={listed} candidates={kept}\",\n        file=sys.stderr,\n        flush=True,\n    )\n\n\ndef sample_one(\n    endpoint: str | None,\n    bucket: str,\n    item: dict,\n    head_bytes: int,\n    tail_bytes: int,\n) -> Hit | None:\n    client = client_for_thread(endpoint)\n    key = item[\"Key\"]\n    size = int(item[\"Size\"])\n    etag = str(item.get(\"ETag\", \"\"))\n    head_end = min(size, head_bytes) - 1\n    try:\n        head = get_range(client, bucket, key, 0, head_end)\n    except ClientError as error:\n        print(f\"warn: range get failed key={key}: {error}\", file=sys.stderr)\n        return None\n\n    hit = analyze_head(key, size, etag, head)\n    if hit is None:\n        return None\n    return enrich_tail(client, bucket, hit, size, head, tail_bytes)\n\n\ndef main() -> int:\n    args = parse_args()\n    if not args.bucket:\n        raise SystemExit(\"error: pass --bucket or set R2_BUCKET\")\n    if args.head_bytes < 520:\n        raise SystemExit(\"error: --head-bytes must be >= 520 to catch HDF5@512\")\n    if args.workers < 1:\n        raise SystemExit(\"error: --workers must be >= 1\")\n    if args.min_size < 0 or args.max_size < args.min_size:\n        raise SystemExit(\"error: invalid size bounds\")\n\n    list_client = make_client(args.endpoint)\n    candidates = []\n    for item in iter_candidates(\n        list_client,\n        args.bucket,\n        args.prefix,\n        args.min_size,\n        args.max_size,\n    ):\n        candidates.append(item)\n        if args.max_objects and len(candidates) >= args.max_objects:\n            break\n\n    hits: list[Hit] = []\n    sampled = 0\n    lock = threading.Lock()\n    jsonl = open(args.jsonl, \"w\", encoding=\"utf-8\") if args.jsonl else None\n\n    try:\n        with ThreadPoolExecutor(max_workers=args.workers) as pool:\n            futures = [\n                pool.submit(\n                    sample_one,\n                    args.endpoint,\n                    args.bucket,\n                    item,\n                    args.head_bytes,\n                    args.tail_bytes,\n                )\n                for item in candidates\n            ]\n            for future in as_completed(futures):\n                sampled += 1\n                if sampled % 200 == 0 or sampled == len(candidates):\n                    print(\n                        f\"sample progress sampled={sampled}/{len(candidates)} \"\n                        f\"hits={len(hits)}\",\n                        file=sys.stderr,\n                        flush=True,\n                    )\n                hit = future.result()\n                if hit is None:\n                    continue\n                with lock:\n                    hits.append(hit)\n                    line = json.dumps(\n                        asdict(hit), separators=(\",\", \":\"), sort_keys=True\n                    )\n                    print(line, flush=True)\n                    if jsonl is not None:\n                        jsonl.write(line + \"\\n\")\n                        jsonl.flush()\n    finally:\n        if jsonl is not None:\n            jsonl.close()\n\n    print(\n        f\"summary candidates={len(candidates)} sampled={sampled} hits={len(hits)} \"\n        f\"bucket={args.bucket} prefix={args.prefix!r} \"\n        f\"size_window={args.min_size}-{args.max_size}\",\n        file=sys.stderr,\n    )\n    if not hits:\n        print(\n            \"no MATLAB/HDF5 hybrid uploads found in size window. \"\n            \"If Active Storage keeps blob metadata in Postgres, also query \"\n            \"active_storage_blobs for content_type like image/bmp and \"\n            \"byte_size between ~1KiB and ~64KiB.\",\n            file=sys.stderr,\n        )\n    return 1 if hits else 0\n\n\nif __name__ == \"__main__\":\n    raise SystemExit(main())\n", "creation_timestamp": "2026-07-30T14:46:38.372236Z"}