vt1200/xie/evidence/xie_raw.py

#!/usr/bin/env python3
"""Minimal X11 client: send one raw XIE request and report the response.

Used to probe the VT1200's XIE 2.3 request sizes/opcodes without XIElib.
Read-only probes only.
"""
import socket
import struct
import sys

HOST, PORT = "192.168.99.2", 6000
LAST_SETUP = {}


def read_exact(sock, n):
    buf = b""
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:
            raise EOFError("server closed connection")
        buf += chunk
    return buf


def setup(sock):
    # little-endian, protocol 11.0, no auth
    sock.sendall(struct.pack("<cxHHHHxx", b"l", 11, 0, 0, 0))
    head = read_exact(sock, 8)
    status, reason_len, major, minor, length = struct.unpack("<BBHHH", head)
    body = read_exact(sock, length * 4) if length else b""
    if status != 1:
        raise RuntimeError(f"setup failed: status={status} reason={body[:reason_len]!r}")
    if len(body) >= 32:
        LAST_SETUP.update(
            release=struct.unpack("<I", body[0:4])[0],
            id_base=struct.unpack("<I", body[4:8])[0],
            id_mask=struct.unpack("<I", body[8:12])[0],
            vendor_len=struct.unpack("<H", body[16:18])[0],
            max_req=struct.unpack("<H", body[18:20])[0],
        )
        vendor_len = LAST_SETUP["vendor_len"]
        n_formats = body[21] if len(body) > 21 else 0
        off = 32 + ((vendor_len + 3) // 4) * 4 + n_formats * 8
        if len(body) >= off + 40:
            LAST_SETUP.update(
                root=struct.unpack("<I", body[off:off + 4])[0],
                screen_width=struct.unpack("<H", body[off + 20:off + 22])[0],
                screen_height=struct.unpack("<H", body[off + 22:off + 24])[0],
                root_depth=body[off + 38],
            )
    return major, minor


def query_extension(sock, name):
    b = name.encode()
    padded = b + b"\0" * ((4 - len(b) % 4) % 4)
    length = 2 + len(padded) // 4
    req = struct.pack("<BBHHH", 98, 0, length, len(b), 0) + padded
    sock.sendall(req)
    rep = read_exact(sock, 32)
    present, opcode, first_event, first_error = rep[8], rep[9], rep[10], rep[11]
    return present, opcode, first_event, first_error


def read_packet(sock):
    head = read_exact(sock, 32)
    kind = head[0]
    if kind == 1:  # reply
        length = struct.unpack("<I", head[4:8])[0]
        return (1, head, read_exact(sock, length * 4))
    if kind == 0:  # error
        return (0, head, b"")
    # event
    return (kind, head, b"")


def probe(minor, words, payload=b"", timeout=3.0):
    s = socket.create_connection((HOST, PORT), timeout=timeout)
    s.settimeout(timeout)
    try:
        setup(s)
        present, opcode, ev, er = query_extension(s, "Xie")
        if not present:
            return f"minor={minor:<2} words={words}: Xie not present"
        data = payload.ljust((words * 4) - 4, b"\0") if words >= 1 else b""
        req = struct.pack("<BBH", opcode, minor, words) + data
        s.sendall(req)
        try:
            kind, head, extra = read_packet(s)
        except socket.timeout:
            return f"minor={minor:<2} words={words}: TIMEOUT (no reply/error)"
        if kind == 1:
            seq = struct.unpack("<H", head[2:4])[0]
            return (f"minor={minor:<2} words={words}: REPLY seq={seq} "
                    f"len={struct.unpack('<I', head[4:8])[0]} "
                    f"bytes={head[8:24].hex()} extra={len(extra)}")
        if kind == 0:
            code = head[1]
            seq = struct.unpack("<H", head[2:4])[0]
            bad = struct.unpack("<I", head[4:8])[0]
            minor_rep = struct.unpack("<H", head[8:10])[0]
            major_rep = head[10]
            return (f"minor={minor:<2} words={words}: ERROR code={code} seq={seq} "
                    f"bad={bad:#x} minor_op={minor_rep} major_op={major_rep}")
        return f"minor={minor:<2} words={words}: event type={kind}"
    finally:
        s.close()


if __name__ == "__main__":
    # baseline: QueryImageExtension, 5.0 header (2 words)
    print(probe(1, 2, struct.pack("<HH", 5, 0)))
    # QueryTechniques: vary the declared request length
    for w in range(1, 9):
        print(probe(2, w))