asyncvnc2.py.orig

import ssl
from asyncio import (
    CancelledError,
    IncompleteReadError,
    Queue,
    QueueFull,
    StreamReader,
    StreamWriter,
    get_running_loop,
    open_connection,
    sleep,
    start_server,
)
from base64 import b64encode
from contextlib import ExitStack, asynccontextmanager, contextmanager
from dataclasses import dataclass, field
from enum import Enum
from fractions import Fraction
from hashlib import sha1
from itertools import product
from os import urandom
from typing import Dict, List, Optional, Set, Tuple
from zlib import decompressobj

import numpy as np
from cryptography.hazmat.primitives.asymmetric import padding, rsa
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.serialization import load_der_public_key
from keysymdef import keysymdef  # type: ignore
from loguru import logger

# Keyboard keys
key_codes: Dict[str, int] = {}
key_codes.update((name, code) for name, code, char in keysymdef)
key_codes.update((chr(char), code) for name, code, char in keysymdef if char)
key_codes['Del'] = key_codes['Delete']
key_codes['Esc'] = key_codes['Escape']
key_codes['Cmd'] = key_codes['Super_L']
key_codes['Alt'] = key_codes['Alt_L']
key_codes['Ctrl'] = key_codes['Control_L']
key_codes['Super'] = key_codes['Super_L']
key_codes['Shift'] = key_codes['Shift_L']
key_codes['Backspace'] = key_codes['BackSpace']

# Common screen aspect ratios
screen_ratios: Set[Fraction] = {
    Fraction(3, 2), Fraction(4, 3), Fraction(16, 10), Fraction(16, 9), Fraction(32, 9), Fraction(64, 27)}

# Colour channel orders
video_modes: Dict[bytes, str] = {
     b'\x20\x18\x00\x01\x00\xff\x00\xff\x00\xff\x10\x08\x00': 'bgra',
     b'\x20\x18\x00\x01\x00\xff\x00\xff\x00\xff\x00\x08\x10': 'rgba',
     b'\x20\x18\x01\x01\x00\xff\x00\xff\x00\xff\x10\x08\x00': 'argb',
     b'\x20\x18\x01\x01\x00\xff\x00\xff\x00\xff\x00\x08\x10': 'abgr',
}

video_definition: Dict[str, bytes] = {v: k for k, v in video_modes.items()}

#: Open H.264 encoding number (0x48323634 = ASCII "H264"). Intentionally NOT an
#: Enc member -- it must never be auto-advertised by EncList.default(), and there
#: is no verified wire format to decode it against. See Video.read() for details.
_H264_ENCODING = 0x48323634

#: Plafond défensif sur toute longueur/taille annoncée par le serveur avant
#: une allocation (Tight, ZlibHex, curseurs). 64 Mo est largement au-dessus
#: de tout besoin légitime -- juste assez pour éviter qu'un serveur hostile
#: ou buggé ne force une allocation démesurée.
_MAX_ALLOC_BYTES = 64 * 1024 * 1024


def _check_alloc_size(n: int, what: str):
    if n > _MAX_ALLOC_BYTES:
        raise ValueError(f"{what}: taille annoncée ({n} octets) dépasse le plafond de sécurité")


class Enc(Enum):
    """
    Supported encodings

    Default priority order is TIGHT, TIGHT_PNG, ZRLE, TRLE, ZLIB, HEXTILE, COPY, RAW.
    ZYWRLE (17) is a supported member but deliberately NOT in this default
    list -- see Enc.ZYWRLE and Enc.default() docstrings.

    The remaining members are pseudo-encodings: they never "win" a rectangle
    like the ones above, they either shape the framebuffer/cursor state
    (CURSOR, X_CURSOR, CURSOR_WITH_ALPHA, DESKTOP_SIZE, EXTENDED_DESKTOP_SIZE,
    LAST_RECT) or just advertise a capability with no rectangle payload of
    their own (FENCE, CONTINUOUS_UPDATES, XVP).
    """

    #: Tight encoding (BasicCompression/FillCompression/JpegCompression).
    TIGHT = 7

    #: Tight encoding, PNG variant (reuses the Tight decoder as-is).
    TIGHT_PNG = -260

    #: ZRLE encoding.
    ZRLE = 16

    #: ZYWRLE encoding (17). On the wire this is the ZRLE format (message-type
    #: 16 sub-stream shape -- 4-byte zlib length then TRLE-style tiles through
    #: the same zlib context): the "ZYWRLE" part is a server-side-only lossy
    #: wavelet preprocessing step applied to pixel data *before* the ordinary
    #: ZRLE/TRLE tile encoder runs (Hitachi's algorithm, see QEMU's
    #: `ui/vnc-enc-zywrle-template.c`). The original hypothesis here was that
    #: this meant no client-side decode changes were needed at all -- **real
    #: testing 2026-09-09 against a real QEMU 8.2.2 server disproves that**:
    #: once QEMU's lossy wavelet path actually engages (`-vnc :N,lossy=on`
    #: plus a jpeg_quality < 9), it sends tile subencoding byte values in the
    #: 17-126 range, which rfbproto.rst/RFC 6143 both reserve as "unused" (no
    #: packed-palette bit-width is defined above size 16) -- this fork's
    #: TRLE/ZRLE decoder correctly raises ValueError on that range rather
    #: than guess at an undocumented tile format. Two candidate guesses at
    #: the real format (palette + 1-byte-per-pixel indices; a couple of
    #: bit-packing width variants) were tried against the live server and
    #: neither produced a coherent image -- see
    #: docs/sessions/session-11-2026-09-09.md for the full trace. Net
    #: result: Enc.ZYWRLE decodes correctly for any *standard* ZRLE-shaped
    #: rectangle (confirmed: subencodings 0/1/2-16/127/128/130-255 all work,
    #: same code as Enc.ZRLE -- see Video.read()), but a real QEMU server
    #: with its lossy path actually engaged WILL eventually send a rectangle
    #: this fork cannot decode. Deliberately excluded from Enc.default() --
    #: see Enc.default() -- opt-in only, and only recommended today against
    #: a server confirmed to encode ZYWRLE within the documented subencoding
    #: range (untested; TigerVNC/Xvnc 1.13.1 doesn't implement ZYWRLE at
    #: all, confirmed the same day -- see docs/features-backlog.md).
    ZYWRLE = 17

    #: TRLE encoding.
    TRLE = 15

    #: Raw encoding with zlib compession.
    ZLIB = 6

    #: Hextile encoding.
    HEXTILE = 5

    #: zlib-compressed Hextile encoding (LibVNCServer/x11vnc).
    ZLIBHEX = 8

    #: CoRRE encoding (Compact RRE -- rectangles limités à 255x255).
    CORRE = 4

    #: RRE encoding (Rise-and-Run-length Encoding).
    RRE = 3

    #: CopyRect encoding.
    COPY = 1

    #: Raw encoding.
    RAW = 0

    #: Rich Cursor pseudo-encoding: client-side cursor rendering.
    CURSOR = -239

    #: X Cursor pseudo-encoding: older/simpler cousin of CURSOR.
    X_CURSOR = -240

    #: Cursor With Alpha pseudo-encoding: true per-pixel alpha, no bitmask.
    CURSOR_WITH_ALPHA = -314

    #: DesktopSize pseudo-encoding: server-initiated framebuffer resize.
    DESKTOP_SIZE = -223

    #: ExtendedDesktopSize pseudo-encoding: resize + multi-screen layout.
    EXTENDED_DESKTOP_SIZE = -308

    #: LastRect pseudo-encoding: marks the end of an update of unknown rectangle count.
    LAST_RECT = -224

    #: Fence pseudo-encoding: enables the ClientFence/ServerFence messages.
    FENCE = -312

    #: ContinuousUpdates pseudo-encoding: enables streaming without polling.
    CONTINUOUS_UPDATES = -313

    #: xvp pseudo-encoding: enables the xvp Client/Server Messages (remote
    #: shutdown/reboot/reset of the system whose framebuffer we're viewing).
    XVP = -309

    @classmethod
    def default(cls):
        """
        Get a list of supported encodings expect RAW.

        Enc.ZYWRLE (17) is deliberately excluded here -- opt-in only via
        EncList([...]) + Enc.ZYWRLE or EncList() + Enc.ZYWRLE. See the
        Enc.ZYWRLE docstring: real testing (2026-09-09) against a real
        QEMU 8.2.2 server found that its ZYWRLE encoder sends tile
        subencoding byte values in the 17-126 range that rfbproto.rst/RFC
        6143 both explicitly reserve as "unused" -- this fork's TRLE/ZRLE
        decoder correctly refuses to decode those (ValueError) rather than
        guess at an undocumented format and risk silently corrupting the
        image. Advertising ZYWRLE by default would make a plain ZRLE-only
        connection to such a server crash the moment the server's lossy
        heuristics kick in (quality<9 with lossy=on) -- entirely outside
        the caller's control once negotiated.
        """
        return filter(lambda x: x.value not in (0, cls.ZYWRLE.value), cls)

class EncList(list):
    """
    A list of Supported encodings and the simple operations:

    "-" -- Exclude from the list

    "+" -- Add to list or rise priority

    EncList()         -- default encoding list

    EncList(Enc.ZLIB) -- encoding list with one encoding

    EncList([Enc.ZLIB,Enc.TRLE]) -- encoding list with two encodings

    EncList() - Enc.ZLIB -- default encoding list except Enc.ZLIB

    EncList() + Enc.ZLIB -- default encoding list with hi-priority of Enc.ZLIB
    """

    def __init__(self, iterable=None):
        if iterable is None:
            super().__init__(Enc.default())
            return
        super().__init__([])
        if isinstance(iterable, Enc):
            iterable = [ iterable ]
        try:
            for x in iterable:
                if isinstance(x, Enc):
                    if x not in self:
                        self.append(x)
                else:
                    raise TypeError
        except TypeError:
            raise TypeError("Arg must be one of the None or the Enc or the iterable of Enc")

    def __add__(self, other):
        if isinstance(other, Enc):
            return EncList([other, *self])
        return EncList([*other, *self])

    def __sub__(self, other):
        if isinstance(other, Enc):
            other = [other]
        a = filter(lambda x: x not in other, self)

        return EncList(a)


async def read_int(reader: StreamReader, length: int) -> int:
    """
    Reads, unpacks, and returns an integer of *length* bytes.
    """

    return int.from_bytes(await reader.readexactly(length), 'big')


async def read_text(reader: StreamReader, encoding: str) -> str:
    """
    Reads, unpacks, and returns length-prefixed text.
    """

    length = await read_int(reader, 4)
    data = await reader.readexactly(length)
    return data.decode(encoding)


def pack_ard(data):
    encoded = data.encode('utf-8')
    if len(encoded) > 63:
        raise ValueError(
            f"Identifiant/mot de passe Apple ARD trop long : {len(encoded)} "
            f"octets UTF-8 (63 maximum, terminateur nul compris dans les 64 "
            f"octets du champ protocolaire)"
        )
    data = encoded + b'\x00'
    data += urandom(64 - len(data))
    return data


# VeNCrypt (security type 19) sub-types, per the only formal spec that was
# ever written for this extension (obtained directly from the original
# VeNCrypt implementation, published on the tigervnc-rfbproto mailing list --
# https://www.mail-archive.com/tigervnc-rfbproto@lists.sourceforge.net/msg00256.html --
# this text never made it into an RFC or into rfbproto.rst itself, but it is
# the same document TigerVNC's own CSecurityVeNCrypt.cxx implements against).
#
# Four sub-types are implemented here, in two families:
# - TLSNone/TLSVnc ("anonymous" TLS): the server presents a self-signed/
#   ephemeral certificate that this client deliberately does not attempt to
#   validate (see _vencrypt_negotiate below) -- that lack of validation *is*
#   the whole meaning of "anonymous" here, matching the "TLS anonyme" wording
#   used for this feature in features.md.
# - X509None/X509Vnc (added 2026-09-05): the server instead presents a real
#   CA-signed certificate that this client *does* validate against a trust
#   store -- by default the system trust store (`ssl.create_default_context()`,
#   the same default a browser or `requests`/`httpx` would use), or an
#   explicit CA supplied by the caller via `ssl_context` (e.g. an internal/
#   private CA). This is what features.md used to describe as "would need a
#   way for the caller to supply a CA certificate, which nothing here
#   currently plumbs through" -- that plumbing is `ssl_context` itself
#   (already threaded through `connect()`/`listen()`/`Client.create()` for
#   TLSNone/TLSVnc) plus the new `server_hostname` parameter below, needed
#   because `loop.start_tls()` refuses to check a certificate's hostname
#   without being told what hostname to check it against.
#
# In both families, "None" means no further authentication after the TLS
# handshake (the SecurityResult below concludes the negotiation directly),
# and "Vnc" means standard VNC Authentication is layered on top, now running
# over the encrypted stream instead of the underlying algorithm changing.
_VENCRYPT_TLS_NONE = 257
_VENCRYPT_TLS_VNC = 258
_VENCRYPT_X509_NONE = 260
_VENCRYPT_X509_VNC = 261


async def _start_tls_client(reader: StreamReader, writer: StreamWriter,
                             ssl_context: ssl.SSLContext,
                             server_hostname: Optional[str] = None) -> Tuple[StreamReader, StreamWriter]:
    """
    Upgrades an already-connected plain-TCP asyncio stream pair to TLS, in
    place. Needed for the VeNCrypt security type (see _vencrypt_negotiate),
    which multiplexes: ProtocolVersion and the VeNCrypt version/sub-type
    negotiation itself happen in clear, then everything from that point on
    (including the rest of the security handshake) must go over TLS on the
    *same* TCP connection -- there is no separate port or new connection.

    asyncio has no public "StreamWriter.start_tls()"; the documented way to
    upgrade an existing connection is `loop.start_tls()`, which swaps the
    transport under an existing protocol instance in place. The StreamReader
    keeps working unmodified afterwards -- `loop.start_tls()` internally
    calls `protocol.connection_made(new_transport)`, and
    `StreamReaderProtocol.connection_made()` reacts to that by calling
    `self._stream_reader.set_transport(new_transport)` on the very same
    StreamReader object we already have. Only the StreamWriter needs to be
    rebuilt: it binds to a specific transport at construction time and has
    no way to swap it out itself, so a fresh StreamWriter is constructed
    around the new (TLS) transport instead, reusing the same reader,
    protocol and event loop.

    *server_hostname* (added for the X509None/X509Vnc VeNCrypt sub-types,
    see _vencrypt_negotiate) is forwarded as-is to `loop.start_tls()`, used
    to check the hostname/IP on the certificate the server presents --
    meaningful whenever `ssl_context.check_hostname` is true, which is the
    default for `ssl.create_default_context()` used below for X509None/
    X509Vnc. Passing `None` here (as `listen()` does, see its own
    docstring) does NOT raise, even with `check_hostname` true: verified by
    real execution in this sandbox that asyncio's `loop.start_tls()`
    (`SSLContext.wrap_bio()` under the hood) silently skips the hostname
    check in that case instead of raising -- unlike the synchronous
    `SSLContext.wrap_socket()` API, which does raise
    `ValueError("check_hostname requires server_hostname")` in the same
    situation (a genuinely different code path; the naive assumption that
    asyncio would raise the same error was wrong and corrected in this
    session, see CLAUDE.md). Certificate CHAIN verification against the
    trust store still applies regardless of `server_hostname` -- only the
    hostname-matching step is affected. Harmless to pass a real hostname
    for TLSNone/TLSVnc, where the default anonymous context explicitly
    disables check_hostname anyway; if a caller supplies their own
    ssl_context for those sub-types with check_hostname enabled instead,
    forwarding it here is the correct behaviour too, not just a no-op.
    """
    loop = get_running_loop()
    transport = writer.transport
    protocol = transport.get_protocol()
    new_transport = await loop.start_tls(transport, protocol, ssl_context, server_side=False,
                                          server_hostname=server_hostname)
    new_writer = StreamWriter(new_transport, protocol, reader, loop)
    return reader, new_writer


async def _vencrypt_negotiate(reader: StreamReader, writer: StreamWriter,
                               ssl_context: Optional[ssl.SSLContext],
                               server_hostname: Optional[str] = None) -> Tuple[StreamReader, StreamWriter, int]:
    """
    Runs the VeNCrypt (security type 19) version and sub-type negotiation,
    then upgrades the connection to TLS and returns the new (reader, writer)
    pair together with the sub-type that was selected -- the caller
    (Client.create()) still has to finish authentication itself for the
    "Vnc" sub-types (by falling through into the existing VNC Authentication
    block over the now-encrypted stream) and, either way, to read the final
    SecurityResult the same as for any other RFB 3.7/3.8 security type.

    Byte format verified against the only formal VeNCrypt specification that
    exists -- see the comment above _VENCRYPT_TLS_NONE for the source -- not
    from memory. Only version 0.2 of the extension is implemented (0.1 was
    never documented and, per that same spec author, no known VNC
    implementation was ever found that supports 0.1 but not 0.2).

    *server_hostname* is the hostname/IP the caller believes it is talking
    to (typically the same `host` passed to `connect()`) -- only meaningful
    for the X509None/X509Vnc sub-types added 2026-09-05, see below and the
    comment above _VENCRYPT_TLS_NONE. Ignored (well, harmlessly forwarded)
    for TLSNone/TLSVnc.

    Sub-type preference order when several are offered: X509Vnc, X509None,
    TLSVnc, TLSNone -- real certificate validation (X509*) is preferred over
    anonymous TLS (TLS*), and within each family, requiring VNC
    Authentication on top ("Vnc") is preferred over no further
    authentication at all ("None"), same reasoning as the pre-existing
    outer `for auth_type in (33, 1, 2, 19)` preference a few lines up in
    Client.create(). A server offering only TLSNone/TLSVnc (no X509
    sub-type at all) behaves exactly as before this session -- this
    ordering is a pure addition, not a change to the previous two-way
    choice.

    TLSNone/TLSVnc: only tested in this sandbox against a hand-written fake
    VeNCrypt server over a real loopback TLS socket (see CLAUDE.md) -- never
    against a real server that actually requires VeNCrypt (TigerVNC/RealVNC
    Enterprise are the obvious candidates, none available here).

    X509None/X509Vnc (added 2026-09-05): same hand-written-fake-server
    loopback methodology, this time with a real two-tier CA/server
    certificate chain generated with `openssl` in this sandbox, covering
    both a successful connection (client trusts the CA that signed the
    server certificate) and a rejected one (client trusts an unrelated CA
    instead -- confirms real validation actually happens, unlike TLSNone/
    TLSVnc above). Also never tested against a real server.
    """
    server_major = await read_int(reader, 1)
    server_minor = await read_int(reader, 1)
    if server_major != 0 or server_minor < 2:
        raise ValueError(f'unsupported VeNCrypt version offered by server: {server_major}.{server_minor}')

    writer.write(bytes([0, 2]))
    if await read_int(reader, 1) != 0:
        raise ValueError('server rejected VeNCrypt version 0.2')

    n_subtypes = await read_int(reader, 1)
    if n_subtypes == 0:
        raise ValueError('server advertised no VeNCrypt sub-types')
    raw_subtypes = await reader.readexactly(4 * n_subtypes)
    subtypes = [int.from_bytes(raw_subtypes[i:i + 4], 'big') for i in range(0, len(raw_subtypes), 4)]

    for candidate in (_VENCRYPT_X509_VNC, _VENCRYPT_X509_NONE, _VENCRYPT_TLS_VNC, _VENCRYPT_TLS_NONE):
        if candidate in subtypes:
            chosen = candidate
            break
    else:
        raise ValueError(
            f'unsupported VeNCrypt sub-types offered by server: {subtypes} (only '
            f'TLSNone/TLSVnc/X509None/X509Vnc are implemented)')

    writer.write(chosen.to_bytes(4, 'big'))

    # Point de divergence réel entre sources trouvé et corrigé dans cette
    # session : le tout premier brouillon de spec VeNCrypt (mailing-list
    # tigervnc-rfbproto, 2010, cité plus haut) décrivait ici un U32 de
    # statut (0=OK) suivi, en cas d'échec, d'une chaîne de raison
    # longueur-préfixée -- mais ni la spec communautaire actuelle
    # (rfbproto.rst, section VeNCrypt/"Subtypes with TLS or X509 prefix")
    # ni l'implémentation de référence ne suivent ce format : TigerVNC
    # (`CSecurityTLS::processMsg()`, `common/rfb/CSecurityTLS.cxx`) lit un
    # unique octet U8 et lève une erreur seulement s'il vaut 0 -- aucune
    # chaîne de raison n'est lue. C'est ce format à un octet qui est
    # implémenté ci-dessous ; l'ancien format U32+raison aurait
    # désynchronisé la lecture d'un octet dès la première connexion à un
    # vrai serveur (lu comme les 3 premiers octets de la poignée de main
    # TLS elle-même).
    if await read_int(reader, 1) == 0:
        raise ValueError('server failed to set up VeNCrypt TLS (TLS session initialisation failed)')

    if ssl_context is None:
        if chosen in (_VENCRYPT_X509_NONE, _VENCRYPT_X509_VNC):
            # X509None/X509Vnc (added 2026-09-05): the server presents a
            # real CA-signed certificate, unlike TLSNone/TLSVnc below --
            # this is exactly the "X509" prefix's meaning. The safe default
            # is therefore to actually verify it, against the system trust
            # store, the same default a browser or `requests`/`httpx` would
            # use -- not to silently disable verification as done for the
            # anonymous sub-types just below.
            #
            # A caller whose server uses a private/internal CA (the common
            # case in practice: publicly-trusted CAs essentially never
            # certify VNC endpoints) needs to supply their own ssl_context,
            # e.g.:
            #     ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            #     ctx.load_verify_locations(cafile='ma-ca-interne.pem')
            #     await connect(host, ssl_context=ctx)
            # Passing an explicit ssl_context always takes priority over
            # this default, exactly as it already did for TLSNone/TLSVnc.
            ssl_context = ssl.create_default_context()
        else:
            # TLSNone/TLSVnc are the "anonymous" VeNCrypt sub-types: the
            # server presents a self-signed/ephemeral certificate that
            # there is, by construction, no CA to validate it against --
            # that lack of validation *is* what "anonymous" means here, as
            # opposed to the X509None/X509Vnc sub-types just above, which do
            # carry a real, verifiable certificate. Disabling verification
            # is therefore the correct default for these two sub-types
            # specifically, not a careless bypass -- a caller who wants
            # strict verification instead can always pass their own
            # ssl_context (which is exactly what selects X509None/X509Vnc's
            # verifying behaviour above, if the server also offers one of
            # those two).
            ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            ssl_context.check_hostname = False
            ssl_context.verify_mode = ssl.CERT_NONE
            # 2026-09-07, trouve par test reel contre un vrai serveur
            # TigerVNC (Xvnc, backend GnuTLS) : le contexte ci-dessus,
            # jusque-la, ne suffisait PAS a etablir une session TLSNone/
            # TLSVnc contre ce serveur reel -- echec immediat cote serveur
            # ("TLS Handshake failed: Insufficient credentials for that
            # request."), alors que le meme code fonctionnait contre un
            # faux serveur fabrique a la main lors de la verification du
            # 2026-09-02. Cause reelle : TigerVNC/GnuTLS implemente ces
            # deux sous-types "anonymes" avec de vraies suites Diffie-
            # Hellman anonymes (ADH/AECDH, aucun certificat echange du
            # tout), alors que le faux serveur de test utilisait un
            # certificat auto-signe avec des suites authentifiees
            # normales (CERT_NONE cote client desactive juste la
            # *verification* du certificat, ce qui est different d'une
            # suite reellement anonyme) -- deux implementations
            # differentes d'"anonyme" au sens VeNCrypt du terme.
            # OpenSSL 1.1.0+ retire les suites anonymes de sa liste de
            # suites par defaut (SECLEVEL 1 les exclut deja) ; sans
            # reactivation explicite, aucune suite commune n'existe entre
            # ce client et un serveur qui n'offre QUE de l'anonyme reel --
            # confirme en forcant manuellement `set_ciphers('ADH:AECDH:'
            # '@SECLEVEL=0')` + `maximum_version = TLSv1_2` (les suites
            # anonymes n'existent pas en TLS 1.3, protocole qui les a
            # retirees structurellement) : la connexion contre le meme
            # Xvnc reussit alors. Mais reactiver les suites anonymes ne
            # suffit pas a lui seul : `'DEFAULT:ADH:AECDH:@SECLEVEL=0'`
            # (suites anonymes ajoutees APRES la liste par defaut, pour ne
            # pas casser la non-regression contre un serveur qui
            # presenterait un certificat auto-signe avec des suites
            # authentifiees normales -- exactement le scenario deja
            # verifie le 2026-09-02) echoue tout aussi silencieusement que
            # l'absence totale de suites anonymes. Diagnostic par
            # bissection (`SSLContext.get_ciphers()` sur chaque variante,
            # plus repetition apres avoir decouvert que TigerVNC blackliste
            # temporairement une IP au bout de 5 echecs d'authentification
            # -- `-UseBlacklist=0` cote serveur de test le temps du
            # diagnostic, pour ne pas fausser les mesures) : la ou
            # `'DEFAULT:ADH:AECDH'` echoue, `'ADH:AECDH:DEFAULT'` (memes
            # suites, ordre inverse) reussit. Ce n'est donc pas une
            # question de PRESENCE des suites anonymes dans la liste, mais
            # de leur RANG : `set_ciphers()` restitue une liste ordonnee
            # selon l'ordre d'apparition des tokens dans la chaine, et
            # GnuTLS (backend de Xvnc) ne semble considerer qu'un prefixe
            # de la liste annoncee par le client avant de conclure "No
            # supported cipher suites have been found" -- avec `ADH-AES256
            # -GCM-SHA384` en position 3 (`'ADH:AECDH:HIGH'`) la connexion
            # reussit, avec la meme suite en position 18 (`'ALL:!eNULL'`,
            # ordre par defaut d'OpenSSL) elle echoue, a liste de suites
            # quasi identique par ailleurs (140 vs 141 entrees). Chaine
            # retenue : `'ADH:AECDH:ALL:!eNULL:@SECLEVEL=0'` -- suites
            # anonymes placees EN TETE (prioritaires pour un serveur
            # anonyme reel comme TigerVNC/GnuTLS), suivies de l'ensemble
            # des suites authentifiees normales pour prendre le relais
            # face a un serveur (reel ou fabrique a la main) qui
            # presenterait un certificat auto-signe avec des suites
            # classiques -- verifie sans regression dans les deux sens
            # (voir CLAUDE.md, entree du 2026-09-08). `!eNULL` exclut
            # uniquement les suites sans chiffrement (le "e" de eNULL) ;
            # les suites sans authentification (le "a" de aNULL, ADH/
            # AECDH) restent incluses -- seule l'authentification est
            # optionnelle ici, pas le chiffrement lui-meme. Un caller qui
            # fournit son propre `ssl_context` continue de
            # court-circuiter entierement ce bloc, comme avant.
            ssl_context.set_ciphers('ADH:AECDH:ALL:!eNULL:@SECLEVEL=0')
            ssl_context.maximum_version = ssl.TLSVersion.TLSv1_2


    reader, writer = await _start_tls_client(reader, writer, ssl_context, server_hostname)
    return reader, writer, chosen


@dataclass
class Clipboard:
    """
    Shared clipboard.
    """

    writer: StreamWriter = field(repr=False)

    #: The clipboard text.
    text: str = ''

    def write(self, text: str):
        """
        Sends clipboard text to the server.
        """

        data = text.encode('latin-1')
        self.writer.write(b'\x06\x00\x00\x00' + len(data).to_bytes(4, 'big') + data)


@dataclass
class Keyboard:
    """
    Virtual keyboard.
    """

    writer: StreamWriter = field(repr=False)

    @contextmanager
    def _write(self, key: str):
        data = key_codes[key].to_bytes(4, 'big')
        self.writer.write(b'\x04\x01\x00\x00' + data)
        try:
            yield
        finally:
            self.writer.write(b'\x04\x00\x00\x00' + data)

    @contextmanager
    def hold(self, *keys: str):
        """
        Context manager that pushes the given keys on enter, and releases them (in reverse order) on exit.
        """

        with ExitStack() as stack:
            for key in keys:
                stack.enter_context(self._write(key))
            yield

    def press(self, *keys: str):
        """
        Pushes all the given keys, and then releases them in reverse order.
        """

        with self.hold(*keys):
            pass

    def write(self, text: str):
        """
        Pushes and releases each of the given keys, one after the other.
        """

        for key in text:
            with self.hold(key):
                pass


@dataclass
class Mouse:
    """
    Virtual mouse.
    """

    writer: StreamWriter = field(repr=False)
    buttons: int = 0
    x: int = 0
    y: int = 0

    def _write(self):
        self.writer.write(
            b'\x05' +
            self.buttons.to_bytes(1, 'big') +
            self.x.to_bytes(2, 'big') +
            self.y.to_bytes(2, 'big'))

    @contextmanager
    def hold(self, button: int = 0):
        """
        Context manager that presses a mouse button on enter, and releases it on exit.
        """

        mask = 1 << button
        self.buttons |= mask
        self._write()
        try:
            yield
        finally:
            self.buttons &= ~mask
            self._write()

    def click(self, button: int = 0):
        """
        Presses and releases a mouse button.
        """

        with self.hold(button):
            pass

    def middle_click(self):
        """
        Presses and releases the middle mouse button.
        """

        self.click(1)

    def right_click(self):
        """
        Presses and releases the right mouse button.
        """

        self.click(2)

    def scroll_up(self, repeat=1):
        """
        Scrolls the mouse wheel upwards.
        """

        for _ in range(repeat):
            self.click(3)

    def scroll_down(self, repeat=1):
        """
        Scrolls the mouse wheel downwards.
        """

        for _ in range(repeat):
            self.click(4)

    def move(self, x: int, y: int):
        """
        Moves the mouse cursor to the given co-ordinates.
        """

        self.x = x
        self.y = y
        self._write()


@dataclass
class Screen:
    """
    Computer screen.
    """

    #: Horizontal position in pixels.
    x: int

    #: Vertical position in pixels.
    y: int

    #: Width in pixels.
    width: int

    #: Height in pixels.
    height: int

    #: Arbitrary server/client-assigned identifier (ExtendedDesktopSize only).
    id: int = 0

    #: Reserved flags (ExtendedDesktopSize only, currently always 0).
    flags: int = 0

    @property
    def slices(self) -> Tuple[slice, slice]:
        """
        Object that can be used to crop the video buffer to this screen.
        """

        return slice(self.y, self.y + self.height), slice(self.x, self.x + self.width)

    @property
    def score(self) -> float:
        """
        A measure of our confidence that this represents a real screen. For screens with standard aspect ratios, this
        is proportional to its pixel area. For non-standard aspect ratios, the score is further multiplied by the ratio
        or its reciprocal, whichever is smaller.
        """

        value = float(self.width * self.height)
        ratios = {Fraction(self.width, self.height).limit_denominator(64),
                  Fraction(self.height, self.width).limit_denominator(64)}
        if not ratios & screen_ratios:
            value *= min(ratios) * 0.5
        return value





@dataclass
class Cursor:
    """
    Mouse cursor shape received from the server (Rich Cursor / X Cursor / Cursor With Alpha pseudo-encodings).
    """

    #: Horizontal hotspot position, relative to the cursor image.
    x: int

    #: Vertical hotspot position, relative to the cursor image.
    y: int

    #: Width in pixels.
    width: int

    #: Height in pixels.
    height: int

    #: RGBA image data (alpha is real per-pixel alpha, or derived from the server's transparency bitmask).
    data: np.ndarray


@dataclass
class StreamZReader:
    """
    aio StreamReader wrapper for zlib
    """
    reader: StreamReader = field(repr=False)
    decompress: object = field(repr=False)
    length: int
    _head: int = 0
    _buffer: bytes = b''
    _zbuffer: bytes = b''

    async def read(self, n: int =-1) -> bytes:
        """
        Read up to a maximum of n bytes.
        If n is not provided, or set to -1, read until EOF and return all read bytes.
        When n is provided, data will be returned as soon as it is available.
        """

        if n == -1:
            try:
                self._zbuffer += await self.reader.readexactly(self.length)
            except IncompleteReadError as e:
                self._zbuffer += e.partial
            self.length = 0
            rdata = self._buffer[self._head:] + self.decompress.decompress(self._zbuffer)
            self._buffer = b''
            self._head = 0
            self._zbuffer = self.decompress.unconsumed_tail
            return rdata

        if (n <= 0) or ((len(self._buffer) <= self._head) and (self.length <= 0)):
            return b''

        while (len(self._buffer) == self._head) and (self.length > 0):
            ndata = await self.reader.readexactly(self.length)
            self.length -= len(ndata)
            self._zbuffer += ndata
            self._buffer = self.decompress.decompress(self._zbuffer)
            self._head = 0
            self._zbuffer = self.decompress.unconsumed_tail
        rdata = self._buffer[self._head:self._head + n]
        self._head += len(rdata)
        return rdata

    async def readexactly(self, n: int) -> bytes:
        """
        Read exactly n bytes.
        Raise an asyncio.IncompleteReadError if the end of the stream is reached before n can be read.
        """
        data = b''
        _n = n
        while _n > 0:
            ndata = await self.read(_n)
            if len(ndata) == 0:
                raise IncompleteReadError(data, n)
            data += ndata
            _n = n - len(data)
        return data


def _tile_1d_gen(tw: int, x: int, w: int):
    for cx in range(x, x + w - tw + 1, tw):
        yield (cx, tw)
    mod = w % tw
    if mod:
        yield (x + w - mod, mod)

def _tile_gen(tw: int, th: int, x: int, y: int, w: int, h: int):
    for (cy, ch) in _tile_1d_gen(th, y, h):
        for (cx, cw) in _tile_1d_gen(tw, x, w):
            yield (cx, cy, cw, ch)

class _BytesReader:
    """
    Minimal async-readexactly wrapper over an in-memory buffer, so decompressed
    ZlibHex tile data (already fully available) can be fed through the same
    tile-body decoder used for plain Hextile (which reads from a live StreamReader).
    """

    def __init__(self, data: bytes):
        self._data = data
        self._pos = 0

    async def readexactly(self, n: int) -> bytes:
        chunk = self._data[self._pos:self._pos + n]
        if len(chunk) < n:
            raise IncompleteReadError(chunk, n)
        self._pos += n
        return chunk

async def _hextile_tile_body(reader, mask: int, bg, fg: bytes, cw: int, ch: int):
    """
    Reads a Hextile tile's background/foreground/subrects (everything except the
    Raw case, which the caller handles separately since it short-circuits
    everything else) and returns (tile_array, new_bg, new_fg).

    Shared between process_hextile (reading live from the network) and
    process_zlibhex's Zlib-compressed tiles (reading from an already-decompressed
    in-memory buffer via _BytesReader) -- both follow identical Hextile rules
    once the subencoding byte and a byte source are in hand.
    """
    if mask & 2:
        bg = await reader.readexactly(4)
    if bg is None:
        raise ValueError("Hextile: first tile of the rectangle did not specify a background colour")
    if mask & 4:
        fg = await reader.readexactly(4)

    tile = np.ndarray((ch, cw, 4), 'B', bg * (cw * ch)).copy()
    if mask & 8:
        n = await read_int(reader, 1)
        coloured = bool(mask & 16)
        for _ in range(n):
            colour = (await reader.readexactly(4)) if coloured else fg
            pos = await read_int(reader, 1)
            size = await read_int(reader, 1)
            sx = (pos >> 4) & 0x0F
            sy = pos & 0x0F
            sw = ((size >> 4) & 0x0F) + 1
            sh = (size & 0x0F) + 1
            tile[sy:sy + sh, sx:sx + sw, :] = np.ndarray((sh, sw, 4), 'B', colour * (sw * sh))
    return tile, bg, fg

async def _rle_len(reader: StreamReader):
    _pixels = 0
    while True:
        n = await read_int(reader, 1)
        _pixels += n
        if n != 255:
            break
    return _pixels + 1

async def _update_palette(reader: StreamReader, n: int, palette: Optional[list] = None):
    pal_bytes = await reader.readexactly(3*n)
    pal_list = [pal_bytes[i:i+3] for i in range(0, 3*n, 3)]
    palette.clear()
    palette.extend(pal_list)

async def _rle_packedbits(reader: StreamReader, cw: int, ch: int, subencoding: int, palette: list) -> bytes:
    frame = b''
    if subencoding <= 16:
        # subencoding == 2 .. 16
        await _update_palette(reader, subencoding, palette)
    else:
        # subencoding == 127
        subencoding = len(palette)

    if subencoding == 2:
        bits = 1
    elif subencoding <= 4:
        bits = 2
    else:
        bits = 4
    rowlen = (bits * cw + 7) // 8
    mask = (1 << bits) - 1

    for _ in range(ch):
        row = await reader.readexactly(rowlen)
        offset = 0
        rowit = iter(row)
        for _ in range(cw):
            if offset == 0:
                offset = 8
                packcol = next(rowit)
            offset -= bits
            frame += palette[(packcol >> offset) & mask]
    return frame

async def _rle_rle(reader: StreamReader, cw: int, ch: int, subencoding: int, palette: list) -> bytes:
    frame = b''
    subencoding -= 128
    if subencoding > 1:
        await _update_palette(reader, subencoding, palette)
    elif subencoding == 1:
        subencoding = len(palette)

    tile_pixels = ch * cw
    pixels = 0

    while pixels < tile_pixels:
        if subencoding:
            pal_index = await read_int(reader, 1)
            if pal_index < 128:
                frame += palette[pal_index]
                pixels += 1
                continue
            else:
                pal_index -= 128
                color = palette[pal_index]
        else:
            color = await reader.readexactly(3)
        p = await _rle_len(reader)
        frame += color * p
        pixels += p
    if pixels > tile_pixels:
        raise ValueError("Too many pixels")
    return frame


async def _tight_read_length(reader: StreamReader) -> int:
    """
    Reads a Tight "compact" length (1-3 bytes) and returns it.
    """
    b0 = await read_int(reader, 1)
    length = b0 & 0x7F
    if b0 & 0x80:
        b1 = await read_int(reader, 1)
        length |= (b1 & 0x7F) << 7
        if b1 & 0x80:
            b2 = await read_int(reader, 1)
            length |= b2 << 14
    return length


def _tight_palette_filter(data: bytes, cw: int, ch: int, palette: list) -> bytes:
    """
    Expands Tight PaletteFilter indices back into TPIXEL (RGB) data.
    """
    n = len(palette)
    bits = 1 if n == 2 else 8
    rowlen = (bits * cw + 7) // 8
    out = bytearray(cw * ch * 3)
    pos = 0
    for cy in range(ch):
        row = data[pos:pos + rowlen]
        pos += rowlen
        base = cy * cw * 3
        if bits == 1:
            for cx in range(cw):
                byte = row[cx // 8]
                idx = (byte >> (7 - (cx % 8))) & 1
                out[base:base + 3] = palette[idx]
                base += 3
        else:
            for cx in range(cw):
                out[base:base + 3] = palette[row[cx]]
                base += 3
    return bytes(out)


def _tight_gradient_filter(data: bytes, cw: int, ch: int) -> bytes:
    """
    Reverses the Tight GradientFilter prediction to recover raw TPIXEL data.
    """
    out = bytearray(cw * ch * 3)
    prev_row = [(0, 0, 0)] * cw
    for cy in range(ch):
        left = (0, 0, 0)
        cur_row = [(0, 0, 0)] * cw
        base = cy * cw * 3
        for cx in range(cw):
            up = prev_row[cx]
            upleft = prev_row[cx - 1] if cx else (0, 0, 0)
            pixel = [0, 0, 0]
            for c in range(3):
                pred = left[c] + up[c] - upleft[c]
                if pred < 0:
                    pred = 0
                elif pred > 255:
                    pred = 255
                pixel[c] = (data[base + c] + pred) & 0xFF
            pixel = tuple(pixel)
            out[base:base + 3] = bytes(pixel)
            cur_row[cx] = pixel
            left = pixel
            base += 3
        prev_row = cur_row
    return bytes(out)


def _tight_decode_image(data: bytes, width: int, height: int) -> np.ndarray:
    """
    Decodes a JPEG (Tight) or PNG (TightPng) sub-rectangle. Pillow auto-detects
    the format from the file's magic bytes, so both encodings share this function.

    Requires Pillow (an optional dependency, only needed if the server actually
    sends Jpeg/PngCompression sub-rectangles): pip install Pillow.
    """
    try:
        from PIL import Image
    except ImportError as exc:
        raise RuntimeError("Decoding Tight/TightPng image sub-rectangles requires Pillow (pip install Pillow)") \
            from exc
    import io
    image = Image.open(io.BytesIO(data)).convert('RGB')
    array = np.asarray(image, dtype='B')
    if array.shape != (height, width, 3):
        array = array.reshape((height, width, 3))
    return array


#: ClientFence/ServerFence flag bits.
FENCE_FLAG_BLOCK_BEFORE = 0x00000001
FENCE_FLAG_BLOCK_AFTER = 0x00000002
FENCE_FLAG_SYNC_NEXT = 0x00000004
FENCE_FLAG_REQUEST = 0x80000000

#: Bits the client understands and is willing to echo back in a ServerFence reply.
_FENCE_KNOWN_FLAGS = FENCE_FLAG_BLOCK_BEFORE | FENCE_FLAG_BLOCK_AFTER | FENCE_FLAG_SYNC_NEXT

# xvp-message-code values (both directions -- xvp is a "bidirectional"
# message type, code 250, cf. RFC RFB / rfbproto.rst "xvp Client Message"
# and "xvp Server Message"). Only version 1 of the xvp extension is
# currently defined by the spec.
XVP_FAIL = 0      # server -> client: most recent xvp request could not be honoured
XVP_INIT = 1      # server -> client: confirms xvp support, gives highest version supported
XVP_SHUTDOWN = 2  # client -> server: request a clean shutdown
XVP_REBOOT = 3    # client -> server: request a clean reboot
XVP_RESET = 4     # client -> server: request an abrupt reset
_XVP_VERSION = 1  # extension version this client speaks (only version defined so far)


@dataclass
class Video:
    """
    Video buffer.
    """

    reader: StreamReader = field(repr=False)
    writer: StreamWriter = field(repr=False)
    decompress: object = field(repr=False)
#    lastrefresh = None;

    #: Desktop name.
    name: str

    #: Width in pixels.
    width: int

    #: Height in pixels.
    height: int

    #: Colour channel order.
    mode: str

    #: Allowed encodings list
    encodings: Optional[list] = None

    #: Four independent zlib streams used by Tight's BasicCompression.
    tight_streams: list = field(default_factory=lambda: [decompressobj() for _ in range(4)], repr=False)

    #: Two independent zlib streams used by ZlibHex (one for ZlibRaw tiles, one for Zlib tiles).
    zlibhex_streams: list = field(default_factory=lambda: [decompressobj(), decompressobj()], repr=False)

    #: 8bpp indexed-colour palette {index: (r, g, b)} (0-255 each), from SetColourMapEntries.
    #: Only meaningful when Video.create() was called with allow_indexed_colour=True and the
    #: server actually proposed an 8bpp PseudoColor format (self.mode == 'indexed8').
    palette: dict = field(default_factory=dict, repr=False)

    #: Current mouse cursor shape (Rich Cursor / X Cursor / Cursor With Alpha), if any.
    cursor: Optional[Cursor] = None

    #: Physical screen layout (ExtendedDesktopSize pseudo-encoding).
    screens: list = field(default_factory=list, repr=False)

    #: Serial number
    serial = 0

    #: 3D numpy array of colour data.
    data: Optional[np.ndarray] = None

    @classmethod
    async def create(cls, reader: StreamReader, writer: StreamWriter, encodings: list,
                      jpeg_quality: Optional[int] = None, compression_level: Optional[int] = None,
                      allow_indexed_colour: bool = False, shared: bool = True) -> 'Video':
        encodings = EncList(encodings)

        writer.write(bytes([1 if shared else 0]))
        width = await read_int(reader, 2)
        height = await read_int(reader, 2)
        mode_data = bytearray(await reader.readexactly(13))
        bits_per_pixel = mode_data[0]
        true_colour = mode_data[3] & 1
        mode_data[2] &= 1  # set big endian flag to 0 or 1
        mode_data[3] &= 1  # set true colour flag to 0 or 1
        mode = video_modes.get(bytes(mode_data))
        await reader.readexactly(3)  # padding
        name = await read_text(reader, 'utf-8')

        if mode is None:
            if allow_indexed_colour and not true_colour and bits_per_pixel == 8:
                # Accept the server's native 8bpp indexed-colour format instead of overriding
                # it with our own truecolour SetPixelFormat. Only Enc.RAW is meaningful here --
                # see process_raw_indexed and the note on the allow_indexed_colour parameter.
                mode = 'indexed8'
            else:
                mode = 'rgba'
                writer.write(b'\x00\x00\x00\x00' + video_definition.get(mode) + b'\x00\x00\x00')

        # jpeg_quality/compression_level are one-shot hints, not real encodings: they never come
        # back as a rectangle, so they're appended as raw values instead of going through EncList
        # (which only accepts Enc members).
        values = [e.value for e in encodings]
        if jpeg_quality is not None:
            if not 0 <= jpeg_quality <= 9:
                raise ValueError("jpeg_quality must be between 0 and 9")
            values.append(-32 + jpeg_quality)
        if compression_level is not None:
            if not 0 <= compression_level <= 9:
                raise ValueError("compression_level must be between 0 and 9")
            values.append(-256 + compression_level)

        writer.write(b'\x02\x00' + len(values).to_bytes(2, 'big'))
        writer.write(b''.join(v.to_bytes(4, 'big', signed=True) for v in values))

        decompress = decompressobj()

        return cls(reader, writer, decompress, name, width, height, mode, encodings)

    def get_rect(self, x: int = 0, y: int = 0, width: Optional[int] = None, height: Optional[int] = None) -> tuple:
        """
        Crops the rectangle according to the video buffer.
        """
        if x < 0:
            x = 0
        elif x > self.width:
            x = self.width
        if y < 0:
            y = 0
        elif y > self.height:
            y = self.height
        if (width is None) or (width + x > self.width):
            width = self.width - x
        elif width < 0:
            width = 0
        if (height is None)  or (height + y > self.height):
            height = self.height - y
        elif height < 0:
            height = 0

        return (x, y, width, height)

    def refresh(self, x: int = 0, y: int = 0, width: Optional[int] = None, height: Optional[int] = None):
        """
        Sends a video buffer update request to the server.
        """

        incremental = self.data is not None

        (x, y, width, height) = self.get_rect(x, y, width, height)

        self.writer.write(
            b'\x03' +
            incremental.to_bytes(1, 'big') +
            x.to_bytes(2, 'big') +
            y.to_bytes(2, 'big') +
            width.to_bytes(2, 'big') +
            height.to_bytes(2, 'big'))

    def _update_rect(self, x1: int, x2: int, y1: int, y2: int, data: np.ndarray):
        """
        Fills the space of the rectangle with the selected data
        Accepts various input shapes:
            HxWx4 -- raw copy
            1x1x4 -- fills with one color
            HxWx3 -- adds alpha
            1x1x3 -- fills with one color and adds alpha
        """

#        print(f"_update_rect {x1} +{x2-x1} {y1} +{y2-y1} {data.shape}")
        if self.data is None:
            self.data = np.zeros((self.height, self.width, 4), 'B')
        # indexed8 has no channel-order letters like the truecolour modes ('rgba', 'bgra', ...) --
        # process_raw_indexed always builds its lookup table in RGBA order, so alpha is always
        # channel 3 there.
        a_index = 3 if self.mode == 'indexed8' else self.mode.index('a')
        if data.shape[2] == 4:
            self.data[y1:y2, x1:x2, :] = data
        if data.shape[2] == 3:
            if a_index:
                self.data[y1:y2, x1:x2, 0:3] = data
            else:
                self.data[y1:y2, x1:x2, 1:4] = data
        self.data[y1:y2, x1:x2, a_index] = 255
        self.serial = (self.serial + 1) & 0xfffffff

    async def process_raw(self, _reader, x, y, width, height):
        ff = height * width * 4
        (cx, cy) = (x, y)
        data = b''
        while ff:
            newdata = await _reader.read(ff)
            ff -= len(newdata)
            data += newdata
            rows = len(data) // (width * 4)
            if rows > 0:
                await sleep(0)
                self._update_rect(cx, cx + width, cy, cy + rows, np.ndarray((rows, width, 4), 'B', data))
                cy += rows
                data = data[rows * width * 4:]
                await sleep(0)

    async def process_raw_indexed(self, _reader, x, y, width, height):
        """
        Decodes a Raw rectangle in 8bpp indexed-colour mode: 1 byte per pixel,
        each byte an index into self.palette (populated by SetColourMapEntries).
        Indices with no known palette entry fall back to grayscale (index value
        repeated across R/G/B).

        IMPORTANT -- ordering hazard confirmed by real-world testing: the RFB
        spec does not guarantee SetColourMapEntries arrives before the first
        FramebufferUpdate that uses those colours. If pixel data is decoded
        here before the relevant palette entries are known, those pixels get
        the grayscale fallback rather than their real colour, and this method
        has no way to go back and fix already-decoded pixels once the palette
        arrives later. If precise colours matter immediately, request a fresh
        (non-incremental) FramebufferUpdate after any SetColourMapEntries you
        receive, so the affected area gets redecoded with the palette now known.
        """
        indices = np.frombuffer(await _reader.readexactly(width * height), 'B').reshape(height, width)
        lut = np.empty((256, 4), 'B')
        lut[:, 0] = lut[:, 1] = lut[:, 2] = np.arange(256, dtype='B')
        lut[:, 3] = 255
        for idx, (r, g, b) in self.palette.items():
            if 0 <= idx < 256:
                lut[idx] = (r, g, b, 255)
        self._update_rect(x, x + width, y, y + height, lut[indices])
        await sleep(0)

    async def process_xrle(self, _reader, x, y, width, height, tile_sz):
        palette = []

        for (cx, cy, cw, ch) in _tile_gen(tile_sz, tile_sz, x, y, width, height):
            subencoding  = await read_int(_reader, 1)
            await sleep(0)
            if subencoding == 0:
                block = await _reader.readexactly(ch * cw * 3)
                self._update_rect(cx, cx + cw, cy, cy + ch, np.ndarray((ch, cw, 3), 'B', block))
            elif subencoding == 1:
                block = await _reader.readexactly(3)
                self._update_rect(cx, cx + cw, cy, cy + ch, np.ndarray((1, 1, 3), 'B', block))
            elif (subencoding <= 16) or (subencoding == 127):
                block = await _rle_packedbits(_reader, cw, ch, subencoding, palette)
                self._update_rect(cx, cx + cw, cy, cy + ch, np.ndarray((ch, cw, 3), 'B', block))
            elif subencoding < 128:
                # Valeurs 17-126 : "unused" selon rfbproto.rst / RFC 6143 (aucun format de
                # bit-packing n'est défini au-delà de la taille de palette 16) -- MAIS un vrai
                # QEMU 8.2.2 (`-vnc :N,lossy=on`) les émet réellement une fois son extension
                # ZYWRLE engagée (quality<9). Deux hypothèses de format testées en conditions
                # réelles le 2026-09-09 (palette de `subencoding` couleurs + 1 octet d'index par
                # pixel ; et quelques variantes de largeur de bit-packing) : aucune ne redonne un
                # flux cohérent avec le contenu de test connu -- voir
                # docs/sessions/session-11-2026-09-09.md pour le détail des essais et pourquoi ce
                # n'est délibérément PAS deviné plus loin ici (le risque est de décoder du faux
                # silencieusement). On refuse donc explicitement plutôt que de corrompre l'image.
                raise ValueError(
                    f"Palette subencoding {subencoding} (plage 17-126, réservée par la RFC) reçue "
                    "-- probablement du ZYWRLE réel (QEMU) dont le format de tuile exact pour "
                    "cette plage n'a pas été déterminé avec certitude dans ce fork ; voir "
                    "Enc.ZYWRLE et docs/features-backlog.md"
                )
            else:
                block = await _rle_rle(_reader, cw, ch, subencoding, palette)
                self._update_rect(cx, cx + cw, cy, cy + ch, np.ndarray((ch, cw, 3), 'B', block))
            await sleep(0)

    async def process_copy(self, _reader, x, y, width, height):
        srcx = await read_int(_reader, 2)
        srcy = await read_int(_reader, 2)
#        print(f"Copy Rect {width}x{height} ({srcx},{srcy}) ==> ({x},{y})")
        self._update_rect(x, x + width, y, y + height, self.data[srcy:srcy + height, srcx:srcx + width, :])

    async def process_tight(self, _reader, x, y, width, height):
        """
        Decodes a Tight or TightPng-encoded rectangle (BasicCompression, FillCompression,
        JpegCompression or PngCompression).
        """
        ctl = await read_int(_reader, 1)
        for i in range(4):
            if ctl & (1 << i):
                self.tight_streams[i] = decompressobj()
        await sleep(0)

        if ctl & 0x80:
            kind = ctl & 0xF0
            if kind == 0x80:
                pixel = await _reader.readexactly(3)
                block = pixel * (width * height)
                self._update_rect(x, x + width, y, y + height, np.ndarray((height, width, 3), 'B', block))
                return
            if kind == 0x90:
                length = await _tight_read_length(_reader)
                _check_alloc_size(length, "Tight (JPEG/PNG)")
                image_data = await _reader.readexactly(length)
                block = _tight_decode_image(image_data, width, height)
                self._update_rect(x, x + width, y, y + height, block)
                return
            raise ValueError(f"Invalid Tight compression-control byte: {ctl:#x}")

        stream_id = (ctl >> 4) & 0x03
        filter_id = (await read_int(_reader, 1)) if (ctl & 0x40) else 0
        palette: list = []

        if filter_id == 1:
            n = (await read_int(_reader, 1)) + 1
            pal_bytes = await _reader.readexactly(3 * n)
            palette = [pal_bytes[i:i + 3] for i in range(0, 3 * n, 3)]
            bits = 1 if n == 2 else 8
            data_size = ((bits * width + 7) // 8) * height
        elif filter_id in (0, 2):
            data_size = width * height * 3
        else:
            raise ValueError(f"Unknown Tight filter-id: {filter_id}")

        if data_size < 12:
            raw = await _reader.readexactly(data_size)
        else:
            length = await _tight_read_length(_reader)
            _check_alloc_size(length, "Tight (BasicCompression)")
            comp = await _reader.readexactly(length)
            raw = self.tight_streams[stream_id].decompress(comp)

        if filter_id == 1:
            block = _tight_palette_filter(raw, width, height, palette)
        elif filter_id == 2:
            block = _tight_gradient_filter(raw, width, height)
        else:
            block = raw

        self._update_rect(x, x + width, y, y + height, np.ndarray((height, width, 3), 'B', block))
        await sleep(0)

    async def process_hextile(self, _reader, x, y, width, height):
        """
        Decodes a Hextile-encoded rectangle.
        """
        bg = None
        fg = bytes(4)
        for (cx, cy, cw, ch) in _tile_gen(16, 16, x, y, width, height):
            mask = await read_int(_reader, 1)
            await sleep(0)

            if mask & 1:
                block = await _reader.readexactly(cw * ch * 4)
                self._update_rect(cx, cx + cw, cy, cy + ch, np.ndarray((ch, cw, 4), 'B', block))
                continue

            tile, bg, fg = await _hextile_tile_body(_reader, mask, bg, fg, cw, ch)
            self._update_rect(cx, cx + cw, cy, cy + ch, tile)
            await sleep(0)

    async def process_rre(self, _reader, x, y, width, height):
        """
        Decodes an RRE-encoded rectangle: a background pixel plus N
        axis-aligned solid-colour subrectangles drawn on top.
        """
        n = await read_int(_reader, 4)
        bg = await _reader.readexactly(4)
        tile = np.ndarray((height, width, 4), 'B', bg * (width * height)).copy()
        for _ in range(n):
            colour = await _reader.readexactly(4)
            sx = await read_int(_reader, 2)
            sy = await read_int(_reader, 2)
            sw = await read_int(_reader, 2)
            sh = await read_int(_reader, 2)
            tile[sy:sy + sh, sx:sx + sw, :] = np.ndarray((sh, sw, 4), 'B', colour * (sw * sh))
            await sleep(0)
        self._update_rect(x, x + width, y, y + height, tile)

    async def process_corre(self, _reader, x, y, width, height):
        """
        Decodes a CoRRE-encoded rectangle: same idea as RRE, but with
        1-byte subrectangle coordinates/dimensions (each subrect confined
        to 0-255) instead of RRE's 2-byte fields -- more compact for small
        tiles. The RFB spec implies the rectangle itself must be <=255x255
        for CoRRE to be legal; enforced explicitly below rather than
        trusting the encoder.
        """
        if width > 255 or height > 255:
            raise ValueError(
                f"CoRRE: rectangle {width}x{height} exceeds the 255x255 limit "
                f"implied by its 1-byte subrectangle fields"
            )
        n = await read_int(_reader, 4)
        bg = await _reader.readexactly(4)
        tile = np.ndarray((height, width, 4), 'B', bg * (width * height)).copy()
        for _ in range(n):
            colour = await _reader.readexactly(4)
            sx = await read_int(_reader, 1)
            sy = await read_int(_reader, 1)
            sw = await read_int(_reader, 1)
            sh = await read_int(_reader, 1)
            tile[sy:sy + sh, sx:sx + sw, :] = np.ndarray((sh, sw, 4), 'B', colour * (sw * sh))
            await sleep(0)
        self._update_rect(x, x + width, y, y + height, tile)

    async def process_zlibhex(self, _reader, x, y, width, height):
        """
        Decodes a ZlibHex-encoded rectangle (Hextile + zlib, LibVNCServer/x11vnc).

        Extends the Hextile subencoding byte with two more bits: ZlibRaw (0x20,
        cancels all other bits like Raw does in plain Hextile -- the decompressed
        data is w*h raw pixels) and Zlib (0x40, the decompressed data is a plain
        Hextile tile body, parsed per the *same* subencoding byte's lower 5 bits).
        Two independent zlib streams are used: one for ZlibRaw tiles, one for
        Zlib tiles -- both persist for the life of the connection (no reset bits
        in this encoding, unlike Tight).
        """
        bg = None
        fg = bytes(4)
        for (cx, cy, cw, ch) in _tile_gen(16, 16, x, y, width, height):
            mask = await read_int(_reader, 1)
            await sleep(0)

            if mask & 0x20:  # ZlibRaw -- cancels all other bits
                length = await read_int(_reader, 2)
                _check_alloc_size(length, "ZlibHex (ZlibRaw)")
                comp = await _reader.readexactly(length)
                raw = self.zlibhex_streams[0].decompress(comp)
                self._update_rect(cx, cx + cw, cy, cy + ch, np.ndarray((ch, cw, 4), 'B', raw))
                continue

            if mask & 0x40:  # Zlib -- decompressed data is a plain Hextile tile body
                length = await read_int(_reader, 2)
                _check_alloc_size(length, "ZlibHex (Zlib)")
                comp = await _reader.readexactly(length)
                raw = self.zlibhex_streams[1].decompress(comp)
                tile, bg, fg = await _hextile_tile_body(_BytesReader(raw), mask, bg, fg, cw, ch)
                self._update_rect(cx, cx + cw, cy, cy + ch, tile)
                await sleep(0)
                continue

            if mask & 1:  # plain Raw (uncompressed)
                block = await _reader.readexactly(cw * ch * 4)
                self._update_rect(cx, cx + cw, cy, cy + ch, np.ndarray((ch, cw, 4), 'B', block))
                continue

            tile, bg, fg = await _hextile_tile_body(_reader, mask, bg, fg, cw, ch)
            self._update_rect(cx, cx + cw, cy, cy + ch, tile)
            await sleep(0)

    async def process_cursor(self, x, y, width, height):
        """
        Decodes a Rich Cursor pseudo-encoding rectangle into self.cursor.
        """
        if width == 0 or height == 0:
            self.cursor = Cursor(x, y, width, height, np.zeros((0, 0, 4), 'B'))
            return
        _check_alloc_size(width * height * 4, "Curseur (Rich Cursor)")
        pixel_bytes = await self.reader.readexactly(width * height * 4)
        rowlen = (width + 7) // 8
        mask_bytes = await self.reader.readexactly(rowlen * height)
        await sleep(0)

        pixels = np.ndarray((height, width, 4), 'B', pixel_bytes)
        rgb = np.dstack((
            pixels[:, :, self.mode.index('r')],
            pixels[:, :, self.mode.index('g')],
            pixels[:, :, self.mode.index('b')]))
        mask_bits = np.unpackbits(np.frombuffer(mask_bytes, 'B').reshape(height, rowlen), axis=1)[:, :width]
        alpha = (mask_bits * 255).astype('B')
        self.cursor = Cursor(x, y, width, height, np.dstack((rgb, alpha)))

    async def process_x_cursor(self, x, y, width, height):
        """
        Decodes an X Cursor pseudo-encoding rectangle into self.cursor (older,
        simpler cousin of Rich Cursor: two RGB triplets instead of full
        pixel-format colours, plus a 1-bit-per-pixel bitmap selecting between them).
        """
        if width == 0 or height == 0:
            self.cursor = Cursor(x, y, width, height, np.zeros((0, 0, 4), 'B'))
            return

        _check_alloc_size(width * height * 4, "Curseur (X Cursor)")
        fore = np.frombuffer(await self.reader.readexactly(3), 'B')
        back = np.frombuffer(await self.reader.readexactly(3), 'B')
        rowlen = (width + 7) // 8
        bitmap_bytes = await self.reader.readexactly(rowlen * height)
        mask_bytes = await self.reader.readexactly(rowlen * height)
        await sleep(0)

        bitmap = np.unpackbits(np.frombuffer(bitmap_bytes, 'B').reshape(height, rowlen), axis=1)[:, :width]
        mask = np.unpackbits(np.frombuffer(mask_bytes, 'B').reshape(height, rowlen), axis=1)[:, :width]
        rgb = np.where(bitmap[:, :, None].astype(bool), fore, back).astype('B')
        alpha = (mask * 255).astype('B')
        self.cursor = Cursor(x, y, width, height, np.dstack((rgb, alpha)))

    async def process_cursor_with_alpha(self, x, y, width, height):
        """
        Decodes a Cursor With Alpha pseudo-encoding rectangle into self.cursor.

        Unlike Rich Cursor / X Cursor, transparency here is a real 8-bit alpha
        channel per pixel (no separate bitmask), always sent as raw 32-bit
        RGBA regardless of the negotiated pixel format. Only the Raw
        sub-encoding (by far the common case in practice) is supported here.
        """
        sub_encoding = await read_int(self.reader, 4)
        if width == 0 or height == 0:
            self.cursor = Cursor(x, y, width, height, np.zeros((0, 0, 4), 'B'))
            return
        if sub_encoding != 0:
            raise ValueError(f"Cursor With Alpha: unsupported sub-encoding {sub_encoding} (only Raw is handled)")
        _check_alloc_size(width * height * 4, "Curseur (Cursor With Alpha)")
        pixel_bytes = await self.reader.readexactly(width * height * 4)
        await sleep(0)
        self.cursor = Cursor(x, y, width, height, np.ndarray((height, width, 4), 'B', pixel_bytes).copy())

    async def process_desktop_size(self, width, height):
        """
        Handles the (simple) DesktopSize pseudo-encoding: resizes the video buffer.
        """
        self.width = width
        self.height = height
        self.data = None
        await sleep(0)

    async def process_extended_desktop_size(self, x, y, width, height):
        """
        Handles the ExtendedDesktopSize pseudo-encoding: resizes the video buffer
        and refreshes the physical screen layout.

        The rectangle's x-position carries the reason code (0 = server-side
        change or reply to a non-incremental request, 1 = reply to this
        client's SetDesktopSize, 2 = another client's request was approved)
        and y-position carries the status code (0 = success) when x is 1.
        """
        reason = x
        status = y
        count = await read_int(self.reader, 1)
        await self.reader.readexactly(3)  # padding
        screens = []
        for _ in range(count):
            screen_id = await read_int(self.reader, 4)
            sx = await read_int(self.reader, 2)
            sy = await read_int(self.reader, 2)
            sw = await read_int(self.reader, 2)
            sh = await read_int(self.reader, 2)
            flags = await read_int(self.reader, 4)
            screens.append(Screen(sx, sy, sw, sh, id=screen_id, flags=flags))
        if (reason != 1 or status == 0) and (self.width, self.height) != (width, height):
            self.width = width
            self.height = height
            self.data = None
        self.screens = screens
        await sleep(0)

    async def read(self):
        x = await read_int(self.reader, 2)
        y = await read_int(self.reader, 2)
        width = await read_int(self.reader, 2)
        height = await read_int(self.reader, 2)
        raw_encoding = await read_int(self.reader, 4)
        if raw_encoding >= 0x80000000:  # read_int is unsigned; pseudo-encodings are negative S32 values
            raw_encoding -= 0x100000000
        if raw_encoding == _H264_ENCODING:
            raise NotImplementedError(
                "Received an Open H.264 rectangle (encoding 0x48323634). This encoding is "
                "intentionally not decoded here: it comes from an experimental, unmerged "
                "TigerVNC extension (PR #1194) with no stable, canonical wire-format spec -- "
                "unlike Tight/Hextile/ZlibHex/etc., which are all documented in rfbproto.rst "
                "or LibVNCServer's rfbproto.h. Guessing at the byte layout risked silently "
                "desyncing or corrupting the stream, so this raises instead. It is also never "
                "advertised automatically (it is not an Enc member, so EncList.default() never "
                "includes it) -- seeing this error means a server sent it unprompted. If you "
                "need this encoding, get the exact wire format from the specific server build "
                "in use (it is not guaranteed to be identical across TigerVNC/TurboVNC forks) "
                "before implementing a decoder."
            )
        encoding = Enc(raw_encoding)
        length = height * width * 4
#        print(f"GET VIDEO REC: {x} {y} {width}x{height} / {encoding}")

        if encoding is Enc.RAW:
            if self.mode == 'indexed8':
                await self.process_raw_indexed(self.reader, x, y, width, height)
            else:
                await self.process_raw(self.reader,
                           x, y, width, height)

        elif encoding is Enc.ZLIB:
            length = await read_int(self.reader, 4)
            await self.process_raw(StreamZReader(self.reader, self.decompress, length),
                       x, y, width, height)

        elif encoding is Enc.TRLE:
            await self.process_xrle(self.reader,
                       x, y, width, height, 16)

        elif (encoding is Enc.ZRLE) or (encoding is Enc.ZYWRLE):
            # Format fil identique pour les deux (voir la docstring de Enc.ZYWRLE) tant que le
            # serveur ne produit que des subencodings standard (0/1/2-16/127/128/130-255) --
            # confirmé 2026-09-09 contre un vrai QEMU 8.2.2 pour une tuile Raw (subenc=0), MAIS
            # ce même serveur envoie aussi, une fois son vrai chemin ZYWRLE engagé, des
            # subencodings 17-126 (réservés par la RFC) que cette fonction refuse explicitement
            # plus bas (voir le ValueError dédié) plutôt que de deviner un format non documenté.
            length = await read_int(self.reader, 4)
            await self.process_xrle(StreamZReader(self.reader, self.decompress, length),
                       x, y, width, height, 64)

        elif encoding is Enc.HEXTILE:
            await self.process_hextile(self.reader,
                       x, y, width, height)

        elif encoding is Enc.ZLIBHEX:
            await self.process_zlibhex(self.reader,
                       x, y, width, height)

        elif encoding is Enc.CORRE:
            await self.process_corre(self.reader,
                       x, y, width, height)

        elif encoding is Enc.RRE:
            await self.process_rre(self.reader,
                       x, y, width, height)

        elif encoding is Enc.COPY:
            await self.process_copy(self.reader,
                                 x, y, width, height)

        elif (encoding is Enc.TIGHT) or (encoding is Enc.TIGHT_PNG):
            await self.process_tight(self.reader,
                       x, y, width, height)

        elif encoding is Enc.CURSOR:
            await self.process_cursor(x, y, width, height)

        elif encoding is Enc.X_CURSOR:
            await self.process_x_cursor(x, y, width, height)

        elif encoding is Enc.CURSOR_WITH_ALPHA:
            await self.process_cursor_with_alpha(x, y, width, height)

        elif encoding is Enc.DESKTOP_SIZE:
            await self.process_desktop_size(width, height)

        elif encoding is Enc.EXTENDED_DESKTOP_SIZE:
            await self.process_extended_desktop_size(x, y, width, height)

        elif encoding is Enc.LAST_RECT:
            return True  # signals Client.read() to stop looping over rectangles

        else:
            raise ValueError(encoding)

        return False

    def as_rgba(self, x: int = 0, y: int = 0, width: Optional[int] = None, height: Optional[int] = None) -> np.ndarray:
        """
        Returns the video buffer or the selected part of it as a 3D RGBA array.
        """

        (x, y, width, height) = self.get_rect(x, y, width, height)

        if self.data is None:
            return np.zeros((height, width, 4), 'B')
        if self.mode == 'rgba':
            return self.data[y:y + height, x:x + width, :]
        if self.mode == 'abgr':
            return self.data[y:y + height, x:x + width, ::-1]
        return np.dstack((
            self.data[y:y + height, x:x + width, self.mode.index('r')],
            self.data[y:y + height, x:x + width, self.mode.index('g')],
            self.data[y:y + height, x:x + width, self.mode.index('b')],
            self.data[y:y + height, x:x + width, self.mode.index('a')]))

    def is_complete(self, x: int = 0, y: int = 0, width: Optional[int] = None, height: Optional[int] = None):
        """
        Returns true if the video buffer or the selected part of it is entirely opaque.
        """

        if self.data is None:
            return False

        (x, y, width, height) = self.get_rect(x, y, width, height)

        return self.data[y:y + height, x:x + width, self.mode.index('a')].all()

    def detect_screens(self) -> List[Screen]:
        """
        Detect physical screens by inspecting the alpha channel.
        """

        if self.data is None:
            return []

        mask = self.data[:, :, self.mode.index('a')]
        mask = np.pad(mask // 255, ((1, 1), (1, 1))).astype(np.int8)
        mask_a = mask[1:, 1:]
        mask_b = mask[1:, :-1]
        mask_c = mask[:-1, 1:]
        mask_d = mask[:-1, :-1]

        screens = []
        while True:
            # Detect corners by ANDing perpendicular pairs of differences.
            corners = product(
                np.argwhere(mask_b - mask_a & mask_c - mask_a == -1),  # top left
                np.argwhere(mask_a - mask_b & mask_d - mask_b == -1),  # top right
                np.argwhere(mask_d - mask_c & mask_a - mask_c == -1),  # bottom left
                np.argwhere(mask_c - mask_d & mask_b - mask_d == -1))  # bottom right

            # Find cases where 3 corners align, forming an  'L' shape.
            rects = set()
            for a, b, c, d in corners:
                ab = a[0] == b[0] and a[1] < b[1]  # top
                cd = c[0] == d[0] and c[1] < d[1]  # bottom
                ac = a[1] == c[1] and a[0] < c[0]  # left
                bd = b[1] == d[1] and b[0] < d[0]  # right
                if ab and ac:
                    rects.add((a[1], a[0], b[1], c[0]))
                if ab and bd:
                    rects.add((a[1], a[0], d[1], d[0]))
                if cd and ac:
                    rects.add((a[1], a[0], d[1], d[0]))
                if cd and bd:
                    rects.add((c[1], b[0], d[1], d[0]))

            # Create screen objects and sort them by their scores.
            candidates = [Screen(int(x0), int(y0), int(x1 - x0), int(y1 - y0)) for x0, y0, x1, y1 in rects]
            candidates.sort(key=lambda screen: screen.score, reverse=True)

            # Find a single fully-opaque screen
            for screen in candidates:
                if mask_a[screen.slices].all():
                    mask_a[screen.slices] = 0
                    screens.append(screen)
                    break

            # Finish up if no screens remain
            else:
                return screens


class UpdateType(Enum):
    """
    Update from server to client.
    """

    #: Video update.
    VIDEO = 0

    #: SetColourMapEntries: server-defined palette for indexed (non-truecolour) pixel formats.
    #: Only sent by a server that ignored our SetPixelFormat truecolour override, or before the
    #: client's first FramebufferUpdateRequest -- see Client.process_set_colour_map_entries.
    SET_COLOUR_MAP_ENTRIES = 1

    #: Bell update.
    BELL = 2

    #: Clipboard update.
    CLIPBOARD = 3

    #: EndOfContinuousUpdates: server confirms continuous updates have stopped.
    END_OF_CONTINUOUS_UPDATES = 150

    #: ServerFence: part of the Fence pseudo-encoding's synchronisation handshake.
    SERVER_FENCE = 248

    #: xvp Server Message: XVP_INIT (support confirmed) or XVP_FAIL (last
    #: shutdown/reboot/reset request refused). See Client.process_xvp().
    XVP = 250


@dataclass
class Client:
    """
    VNC client.
    """

    reader: StreamReader = field(repr=False)
    writer: StreamWriter = field(repr=False)

    #: The shared clipboard.
    clipboard: Clipboard

    #: The virtual keyboard.
    keyboard: Keyboard

    #: The virtual mouse.
    mouse: Mouse

    #: The video buffer.
    video: Video

    #: The server's public key (Mac only)
    host_key: Optional[rsa.RSAPublicKey]

    #: Version RFB effectivement négociée avec le serveur, sous la forme
    #: (major, minor) -- (3, 3), (3, 7) ou (3, 8). Voir Client.create()
    #: pour le repli de version (RFC 6143 §7.1.1 + Appendix A) : ce client
    #: ne demande jamais une version supérieure à celle offerte par le
    #: serveur, et adapte la poignée de main de sécurité en conséquence.
    protocol_version: tuple = (3, 8)

    #: Whether the server is currently streaming continuous updates.
    continuous_updates_active: bool = False

    #: Whether the server has confirmed xvp support via an XVP_INIT message
    #: (sent by the server after the client requested the XVP pseudo-encoding).
    #: False until that first XVP_INIT arrives -- send_xvp() does not check
    #: this itself, see its docstring.
    xvp_supported: bool = False

    #: Highest xvp-extension-version the server said it supports, from the
    #: most recent XVP_INIT message. 0 until one has been received. Only
    #: version 1 is defined by the spec at this time.
    xvp_version: int = 0

    #: Palette from the most recent SetColourMapEntries message, if any: {index: (r, g, b)}
    #: with each component 0-65535 (16-bit, per spec) regardless of the actual colour depth in
    #: use. Populated defensively -- see process_set_colour_map_entries for why this should
    #: never actually happen given Video.create() always forces a truecolour pixel format.
    colour_map: dict = field(default_factory=dict, repr=False)

    @classmethod
    async def create(
            cls,
            reader: StreamReader,
            writer: StreamWriter,
            username: Optional[str] = None,
            password: Optional[str] = None,
            host_key: Optional[rsa.RSAPublicKey] = None,
            encodings: Optional[list] = None,
            jpeg_quality: Optional[int] = None,
            compression_level: Optional[int] = None,
            allow_indexed_colour: bool = False,
            shared: bool = True,
            ssl_context: Optional[ssl.SSLContext] = None,
            server_hostname: Optional[str] = None) -> 'Client':
        """
        *server_hostname* (added 2026-09-05) is only used for the VeNCrypt
        X509None/X509Vnc sub-types, and only when *ssl_context* is not
        supplied (see _vencrypt_negotiate) -- it is the hostname/IP the
        caller believes it is connecting to, needed so the real certificate
        presented by the server can be checked against it. `connect()`
        passes its own `host` parameter here automatically; `listen()`
        leaves it at `None` since reverse connections have no such target
        hostname to check (see `listen()`'s own docstring).
        """

        intro = await reader.readline()
        if intro[:4] != b'RFB ':
            raise ValueError('not a VNC server')
        try:
            server_major = int(intro[4:7])
            server_minor = int(intro[8:11])
        except ValueError:
            raise ValueError('not a VNC server')

        # Repli de version de protocole (RFC 6143 §7.1.1 + Appendix A) : ce
        # client ne doit jamais réclamer une version supérieure à celle
        # offerte par le serveur. Seules 3.3, 3.7 et 3.8 sont "publiées" --
        # tout minor 3.x non reconnu doit être traité comme 3.3 ("Any
        # version reported other than 3.7 or 3.8 should be treated as 3.3",
        # Appendix A). Un major > 3 est traité comme 3.8 plutôt que comme un
        # repli 3.3 : en pratique, seuls des serveurs RealVNC Enterprise
        # récents annoncent des majors 4/5 ("RFB 004.001", "RFB 005.000" --
        # confirmé par un rapport de bug tiers qui a dû corriger exactement
        # ce point), et il s'agit là de leur propre numérotation produit,
        # pas d'une évolution du protocole RFB lui-même -- eux-mêmes
        # attendent une réponse "003.008" et se comportent alors en 3.8
        # standard.
        if server_major < 3:
            raise ValueError(f'unsupported RFB protocol version: {intro!r}')
        elif server_major == 3 and server_minor == 7:
            server_version = (3, 7)
        elif server_major == 3 and server_minor < 7:
            server_version = (3, 3)
        else:
            server_version = (3, 8)

        writer.write('RFB {:03d}.{:03d}\n'.format(*server_version).encode('ascii'))

        if server_version == (3, 3):
            # 3.3 : pas de négociation à deux sens -- le serveur décide
            # seul et envoie directement un U32 (0=échec, 1=None,
            # 2=VNC Authentication) ; le client n'envoie aucun octet de
            # sélection ici (Appendix A.1).
            auth_type = await read_int(reader, 4)
            if auth_type == 0:
                raise ValueError(await read_text(reader, 'utf-8'))
            if auth_type not in (1, 2):
                raise ValueError(f'unsupported security type: {auth_type}')
        else:
            auth_types = set(await reader.readexactly(await read_int(reader, 1)))
            if not auth_types:
                raise ValueError(await read_text(reader, 'utf-8'))
            # VeNCrypt (19) est délibérément en dernier de cette liste de
            # préférence : c'est une fonctionnalité neuve, jamais testée
            # contre un vrai serveur (voir CLAUDE.md), alors que 33/1/2 sont
            # éprouvés. Un serveur qui propose aussi 33/1/2 en plus de 19
            # continue donc à se comporter exactement comme avant ce patch --
            # seul un serveur qui n'offre QUE VeNCrypt (auparavant un échec
            # sec ici, "unsupported auth types") en bénéficie.
            for auth_type in (33, 1, 2, 19):
                if auth_type in auth_types:
                    writer.write(auth_type.to_bytes(1, 'big'))
                    break
            else:
                raise ValueError(f'unsupported auth types: {auth_types}')

        # VeNCrypt (TLS anonyme TLSNone/TLSVnc, ou authentifié par certificat
        # X.509 X509None/X509Vnc depuis 2026-09-05) -- voir
        # _vencrypt_negotiate() pour le détail du format. Après ce bloc,
        # `reader`/`writer` pointent vers le flux TLS et non plus vers la
        # socket TCP en clair d'origine -- tout le reste de cette méthode
        # (SecurityResult, ClientInit/ServerInit, etc.) continue de
        # fonctionner sans changement puisqu'il ne manipule que
        # `reader`/`writer`, jamais le transport sous-jacent directement.
        if auth_type == 19:
            reader, writer, vencrypt_subtype = await _vencrypt_negotiate(
                reader, writer, ssl_context, server_hostname)
            if vencrypt_subtype in (_VENCRYPT_TLS_VNC, _VENCRYPT_X509_VNC):
                # Les sous-types "Vnc" (anonyme ou X.509) enchaînent sur une
                # authentification VNC Auth standard, désormais sur le flux
                # chiffré -- on réutilise donc tel quel le bloc "VNC
                # authentication" ci-dessous en se faisant simplement passer
                # pour lui. Seule la validation (ou non) du certificat TLS
                # en amont diffère entre les deux ; ce qui suit est
                # identique.
                auth_type = 2
            # Les sous-types "None" (anonyme ou X.509) n'ont plus rien à
            # faire ici : ni le bloc Apple ni le bloc VNC authentication
            # ci-dessous ne se déclenchent (auth_type reste 19), et le
            # SecurityResult lu plus
            # bas conclut la négociation -- comportement standard de tout
            # type de sécurité RFB 3.7/3.8, VeNCrypt inclus (le fait que la
            # spec VeNCrypt d'origine ne mentionne pas elle-même ce
            # SecurityResult final s'explique par le fait qu'elle documente
            # uniquement ce qui est spécifique à VeNCrypt, pas la poignée de
            # main RFB générique dans laquelle il s'insère).

        # Apple authentication
        if auth_type == 33:
            if username is None or password is None:
                raise ValueError('server requires username and password')
            if host_key is None:
                writer.write(b'\x00\x00\x00\x0a\x01\x00RSA1\x00\x00\x00\x00')
                await reader.readexactly(4)  # packet length
                await reader.readexactly(2)  # packet version
                host_key_length = await read_int(reader, 4)
                host_key = await reader.readexactly(host_key_length)
                host_key = load_der_public_key(host_key)
                await reader.readexactly(1)  # unknown
            aes_key = urandom(16)
            cipher = Cipher(algorithms.AES(aes_key), modes.ECB())
            encryptor = cipher.encryptor()
            credentials = pack_ard(username) + pack_ard(password)
            writer.write(
                b'\x00\x00\x01\x8a\x01\x00RSA1' +
                b'\x00\x01' + encryptor.update(credentials) +
                b'\x00\x01' + host_key.encrypt(aes_key, padding=padding.PKCS1v15()))
            await reader.readexactly(4)  # unknown

        # VNC authentication
        if auth_type == 2:
            if password is None:
                raise ValueError('server requires password')
            try:
                raw_password = password.encode('ascii')
            except UnicodeEncodeError:
                # La VNC Auth classique n'a jamais eu d'encodage standard
                # pour les mots de passe non-ASCII. Latin-1 couvre les
                # caractères accentués courants (é, è, à, ç, ù...) en un
                # seul octet chacun -- meilleur compromis possible ici,
                # mais ça ne garantit pas l'interopérabilité avec un
                # serveur qui aurait encodé le mot de passe différemment
                # (limite inhérente au protocole, pas résoluble côté client seul).
                raw_password = password.encode('latin-1', errors='replace')
            des_key = raw_password[:8].ljust(8, b'\x00')
            des_key = bytes(int(bin(n)[:1:-1].ljust(8, '0'), 2) for n in des_key)
            encryptor = Cipher(algorithms.TripleDES(des_key), modes.ECB()).encryptor()
            challenge = await reader.readexactly(16)
            writer.write(encryptor.update(challenge) + encryptor.finalize())

        # En 3.3 et 3.7 (mais pas 3.8), le type "None" saute le message
        # SecurityResult et enchaîne directement sur ClientInit/ServerInit
        # (RFC 6143 Appendix A.1 et A.2 -- seule 3.8 envoie un
        # SecurityResult après None).
        if auth_type == 1 and server_version in ((3, 3), (3, 7)):
            auth_result = 0
        else:
            auth_result = await read_int(reader, 4)
        if auth_result == 0:
            return cls(
                reader=reader,
                writer=writer,
                host_key=host_key,
                protocol_version=server_version,
                clipboard=Clipboard(writer),
                keyboard=Keyboard(writer),
                mouse=Mouse(writer),
                video=await Video.create(reader, writer, encodings, jpeg_quality, compression_level,
                                          allow_indexed_colour, shared))
        elif auth_result == 1:
            raise PermissionError('Auth failed')
        elif auth_result == 2:
            raise PermissionError('Auth failed (too many attempts)')
        else:
            reason = await reader.readexactly(auth_result)
            raise PermissionError(reason.decode('utf-8'))

    async def read(self) -> UpdateType:
        """
        Reads an update from the server and returns its type.
        """

        update_type = UpdateType(await read_int(self.reader, 1))

        if update_type is UpdateType.CLIPBOARD:
            await self.reader.readexactly(3)  # padding
            self.clipboard.text = await read_text(self.reader, 'latin-1')

        if update_type is UpdateType.SET_COLOUR_MAP_ENTRIES:
            await self.process_set_colour_map_entries()

        if update_type is UpdateType.VIDEO:
            await self.reader.readexactly(1)  # padding
            cnt = await read_int(self.reader, 2)
#            print(f"UpdateType.VIDEO count={cnt}")
            if cnt == 0xFFFF:
                # Rectangle count unknown: the server signals the end of this
                # update with a LastRect pseudo-rectangle instead.
                while not await self.video.read():
                    pass
            else:
                for _ in range(cnt):
                    await self.video.read()

        if update_type is UpdateType.SERVER_FENCE:
            await self.process_server_fence()

        if update_type is UpdateType.END_OF_CONTINUOUS_UPDATES:
            self.continuous_updates_active = False

        if update_type is UpdateType.XVP:
            await self.process_xvp()

        return update_type

    async def process_set_colour_map_entries(self):
        """
        Reads a SetColourMapEntries message body and stores it in self.colour_map
        (raw 16-bit components, as the spec defines them) and, downsampled to
        8-bit, in self.video.palette (what process_raw_indexed actually renders
        with, when Video.create() was called with allow_indexed_colour=True and
        the server proposed a genuine 8bpp PseudoColor format).

        Outside of that opt-in indexed-colour mode, Video.create() always sends
        a SetPixelFormat forcing 32bpp truecolour, so a compliant server should
        never send this message at all -- it's stored here regardless, purely
        defensively, in case a non-compliant/legacy server (e.g. some embedded
        OOB-management firmware) ignores the client's pixel format anyway.
        """
        await self.reader.readexactly(1)  # padding
        first_colour = await read_int(self.reader, 2)
        n_colours = await read_int(self.reader, 2)
        for i in range(n_colours):
            r = await read_int(self.reader, 2)
            g = await read_int(self.reader, 2)
            b = await read_int(self.reader, 2)
            self.colour_map[first_colour + i] = (r, g, b)
            self.video.palette[first_colour + i] = (r >> 8, g >> 8, b >> 8)

    async def process_server_fence(self):
        """
        Reads a ServerFence message body and, if it's a request, replies with
        a ClientFence carrying the same flags and payload but with the
        Request bit -- and any bits we don't understand -- cleared.
        """
        await self.reader.readexactly(3)  # padding
        flags = await read_int(self.reader, 4)
        length = await read_int(self.reader, 1)
        payload = await self.reader.readexactly(length) if length else b''

        if flags & FENCE_FLAG_REQUEST:
            await self.send_client_fence(flags & _FENCE_KNOWN_FLAGS, payload)

    async def process_xvp(self):
        """
        Reads a server xvp message body (type 250, cf. Enc.XVP).

        XVP_INIT confirms the server supports the extension and gives the
        highest xvp-extension-version it supports -- stored in
        self.xvp_supported / self.xvp_version. XVP_FAIL means the most
        recent xvp Client Message (send_xvp_shutdown/_reboot/_reset) was
        refused by the server; it does *not* mean the extension itself is
        unsupported, so it deliberately leaves self.xvp_supported alone.

        Format vérifié contre la spec RFB officielle (rfbproto.rst,
        section "xvp Server Message" + le layout partagé avec "xvp Client
        Message", tous deux type 250 -- message bidirectionnel d'après le
        même document).

        **2026-09-09** : cette méthode ne se déclenche en pratique que si
        le serveur connaît réellement xvp et envoie donc un message de
        type 250 en bonne et due forme. Testé contre un vrai Xvnc
        (TigerVNC 1.13.1) : ce serveur ne supporte PAS xvp et **ferme la
        connexion** ("unknown message type") dès qu'il reçoit un
        message-type 250 -- process_xvp() n'est alors jamais atteint, le
        prochain client.read() échoue avec IncompleteReadError. Voir
        send_xvp() ci-dessous pour le détail complet de ce constat, qui
        concerne surtout l'appelant de send_xvp() plutôt que cette
        méthode-ci (qui ne fait que décoder un message xvp *reçu*, et
        reste correcte pour un serveur qui, lui, supporte réellement
        l'extension).
        """
        await self.reader.readexactly(1)  # padding
        version = await read_int(self.reader, 1)
        code = await read_int(self.reader, 1)
        if code == XVP_INIT:
            self.xvp_supported = True
            self.xvp_version = version

    async def send_client_fence(self, flags: int, payload: bytes = b''):
        """
        Sends a ClientFence message (type 248). Requires the server to have
        advertised support via the Fence pseudo-encoding.
        """
        if len(payload) > 64:
            raise ValueError("ClientFence payload is limited to 64 bytes")
        self.writer.write(bytes([248, 0, 0, 0]) + flags.to_bytes(4, 'big') + bytes([len(payload)]) + payload)

    async def send_enable_continuous_updates(self, enable: bool, x: int, y: int, width: int, height: int):
        """
        Sends an EnableContinuousUpdates message (type 150). Only send this
        after the server has confirmed support via the ContinuousUpdates
        pseudo-encoding -- there's no other handshake for it.
        """
        self.writer.write(
            bytes([150, 1 if enable else 0]) +
            x.to_bytes(2, 'big') +
            y.to_bytes(2, 'big') +
            width.to_bytes(2, 'big') +
            height.to_bytes(2, 'big'))
        self.continuous_updates_active = enable

    async def send_set_desktop_size(self, width: int, height: int, screens: Optional[list] = None):
        """
        Demande au serveur de redimensionner le bureau distant (message
        client SetDesktopSize, type 251). Le serveur peut refuser -- la
        réponse arrive en tant que rectangle ExtendedDesktopSize avec
        reason=1 (cf. process_extended_desktop_size), à surveiller côté
        appelant pour savoir si la demande a abouti.

        Si `screens` n'est pas fourni, envoie une disposition à un seul
        écran couvrant tout le nouveau bureau (comportement par défaut
        raisonnable pour un client qui ne gère pas explicitement le
        multi-écran lors du resize). Pour préserver une disposition
        multi-écran existante, passer self.video.screens (éventuellement
        ajustée) en argument.
        """
        if screens is None:
            screens = [Screen(0, 0, width, height, id=0, flags=0)]

        body = (
            width.to_bytes(2, 'big') +
            height.to_bytes(2, 'big') +
            len(screens).to_bytes(1, 'big') +
            b'\x00'  # padding
        )
        for screen in screens:
            body += (
                screen.id.to_bytes(4, 'big') +
                screen.x.to_bytes(2, 'big') +
                screen.y.to_bytes(2, 'big') +
                screen.width.to_bytes(2, 'big') +
                screen.height.to_bytes(2, 'big') +
                screen.flags.to_bytes(4, 'big')
            )

        logger.info(
            "send_set_desktop_size | width={width} height={height} screen_count={screen_count}",
            width=width,
            height=height,
            screen_count=len(screens),
        )
        self.writer.write(b'\xfb\x00' + body)  # 251, padding, puis le corps

    async def send_qemu_extended_key_event(self, down: bool, keysym: int, keycode: int):
        """
        Envoie un événement clavier étendu QEMU : keysym ET keycode matériel
        (espace de numérotation Linux evdev) dans le même message, ce qui
        évite l'ambiguïté de traduction keysym<->touche physique que peut
        introduire un layout clavier compliqué (AZERTY inclus).

        Format vérifié le 2026-09-02 contre le code source QEMU officiel
        (github.com/qemu/qemu, fichier ui/vnc.c, commit
        a925240509d1b4b656cc480f1cc79ba4d7c8bc08, branche master) -- cette
        extension n'étant pas documentée dans la RFC RFB, c'est la seule
        source faisant foi. Confirmé côté serveur dans
        protocol_client_msg() : message-type 255 (VNC_MSG_CLIENT_QEMU),
        submessage-type 0 (VNC_MSG_CLIENT_QEMU_EXT_KEY_EVENT), puis
        down-flag en U16 (lu via read_u16(data, 2)), keysym en U32
        (read_u32(data, 4)) et keycode en U32 (read_u32(data, 8)), le tout
        en network byte order -- 12 octets au total ("if (len == 2) return
        12;"). Vérification octet par octet faite dans ce sandbox contre 3
        messages reconstruits à la main d'après cette lecture de source
        (dont des valeurs hautes de bornes U32) : implémentation déjà
        conforme, aucune correction de code nécessaire.

        **2026-09-09 : testé de bout en bout contre un vrai QEMU 8.2.2**
        (`qemu-system-x86_64`, paquet Ubuntu, aucun Proxmox disponible dans
        ce sandbox -- mais c'est le même code serveur `ui/vnc.c` que
        Proxmox utilise pour sa propre console VNC). Vérification par
        preuve visuelle : un secteur de boot x86 minimal écrit pour cette
        session (moins de 50 octets de code, BIOS INT 16h) affiche le caractère
        exact renvoyé par le clavier virtuel du guest -- capture d'écran
        avant/après via `screenshot()` de ce même fichier, donc bien la
        chaîne complète client -> serveur -> périphérique clavier émulé
        -> BIOS guest testée, pas seulement l'envoi. 3 constats réels :

        1. Keysym et keycode cohérents (chiffres, lettres minuscules) :
           le caractère exact apparaît, à chaque fois -- cas nominal
           confirmé de bout en bout.
        2. **`keycode=0` est silencieusement ignoré par ce QEMU** (aucun
           caractère ne remonte), et ne déclenche PAS un repli sur le
           keysym seul -- reconfirmé deux fois avec des keysyms distincts
           (`'3'` puis `'z'`), écran inchangé (0 pixel modifié) dans les
           deux cas. Un appelant doit donc toujours fournir un vrai
           keycode evdev non nul.
        3. **Si le QEMU/Proxmox cible a été démarré avec un layout clavier
           explicite (option `-k <layout>`), `keycode` est intégralement
           ignoré.** Confirmé à la fois en lisant `ext_key_event()` dans
           `ui/vnc.c` (`if (keyboard_layout) { key_event(vs, down, sym); }
           else { do_key_event(vs, down, keycode, sym); }` -- un layout
           explicite bascule sur le chemin keysym-seul, identique au
           vieux message RFB `SetKeyEvent` non étendu) et en relançant le
           même test avec `-k en-us` : un `keycode=0` qui ne produisait
           rien sans `-k` produit alors le caractère attendu à partir du
           seul keysym. **Conséquence pratique : tout l'intérêt de cette
           extension (ne pas dépendre d'un layout côté serveur) disparaît
           silencieusement dès que la cible a un `-k` explicite** -- aucun
           moyen pour ce client de le détecter à l'avance.
        4. Keysym et keycode délibérément incohérents pour une touche
           lettre (ex. keycode de la touche physique A + keysym de 'B') :
           le caractère obtenu n'est ni celui du keycode seul ('a') ni
           celui du keysym seul ('B'), mais 'A' -- hypothèse la plus
           probable (cohérente avec 2 essais indépendants, mais PAS
           confirmée en relisant l'intégralité de `do_key_event()`) : un
           keysym ASCII majuscule A-Z déclenche une touche Shift
           synthétisée côté serveur, appliquée à la touche physique
           désignée par keycode -- comportement observé dans
           `key_event()` pour le chemin non étendu, qui repasse `sym`
           (pas seulement un `sym` déjà mis en minuscule) à
           `do_key_event()`. **Ne pas s'appuyer sur ce cas : toujours
           garder keysym et keycode mutuellement cohérents**, comme le
           ferait un vrai clavier physique -- c'est l'usage pour lequel
           cette extension est conçue, pas un moyen de forcer un
           caractère différent de la touche physique pressée.
        """
        logger.info(
            "send_qemu_extended_key_event | down={down} keysym={keysym:#x} keycode={keycode}",
            down=down,
            keysym=keysym,
            keycode=keycode,
        )
        self.writer.write(
            b'\xff\x00' +                            # message-type 255, submessage-type 0
            (1 if down else 0).to_bytes(2, 'big') +  # down-flag (U16)
            keysym.to_bytes(4, 'big') +
            keycode.to_bytes(4, 'big')
        )

    async def send_xvp(self, code: int, confirm: bool = False):
        """
        Envoie un message client xvp (type 250) demandant au serveur
        d'effectuer un shutdown propre (XVP_SHUTDOWN), un reboot propre
        (XVP_REBOOT) ou un reset brutal (XVP_RESET) du système dont ce
        client affiche le framebuffer. Préférer send_xvp_shutdown() /
        send_xvp_reboot() / send_xvp_reset() ci-dessous plutôt que
        d'appeler cette méthode directement.

        *confirm* (ajouté 2026-09-06) doit être explicitement mis à
        `True` par l'appelant, faute de quoi une `PermissionError` est
        levée avant tout envoi -- garde-fou minimal contre un envoi
        accidentel de cette commande destructive (ex. mauvais bouton
        cliqué dans une UI, appel automatisé mal branché). Ce n'est
        volontairement PAS une confirmation interactive (cette
        bibliothèque bas niveau n'a aucune notion d'UI) : c'est à
        l'appelant final de décider comment obtenir cette confirmation
        (boîte de dialogue, double-clic, argument CLI `--yes`, etc.) et
        de ne passer `confirm=True` qu'une fois cette confirmation
        réellement obtenue. Voir features.md/CLAUDE.md, section
        « Confirmation avant commande xvp destructive ».

        Cette méthode n'exige pas self.xvp_supported -- mais contrairement
        à ce qu'affirmait une version précédente de cette docstring
        (avant test réel), un serveur sans support xvp ne se contente
        **pas forcément** d'ignorer silencieusement le message : **testé
        le 2026-09-09 contre un vrai Xvnc (TigerVNC 1.13.1, aucun support
        xvp)**, ce serveur journalise "unknown message type 250" et
        **ferme immédiatement la connexion** ("unknown message type"),
        exactement comme n'importe quel type de message RFB non reconnu
        qu'il traiterait de la même façon générique -- ce n'est donc pas
        un traitement spécifique à xvp, mais bien le comportement par
        défaut de ce serveur face à un message-type inconnu, quel qu'il
        soit. Un appelant DOIT donc attendre qu'un message XVP_INIT ait
        été reçu (self.xvp_supported == True, via process_xvp()/
        self.read()) avant d'appeler cette méthode, sous peine de
        perdre la connexion RFB entière -- ce n'est plus une simple
        recommandation de prudence mais une précaution nécessaire au vu
        de ce test réel, au moins pour les serveurs qui réagissent comme
        TigerVNC à un message-type inconnu (RFC 6143 ne précise pas de
        comportement obligatoire dans ce cas -- silence, fermeture ou
        autre réaction restent tous conformes à la spec du point de vue
        du serveur).

        Format vérifié contre la spec RFB officielle (rfbproto.rst,
        section "xvp Client Message") -- structure interne testée dans ce
        sandbox contre un flux serveur XVP_INIT fabriqué à la main.
        **2026-09-09 : testé pour de vrai contre TigerVNC/Xvnc 1.13.1**
        (installé via apt dans ce sandbox, `SecurityTypes None`) -- ce
        serveur ne supporte pas xvp (jamais de XVP_INIT reçu, y compris
        après plusieurs FramebufferUpdate) et ferme la connexion dès
        réception d'un send_xvp_shutdown()/_reboot()/_reset() (voir
        détail complet dans docs/sessions/session-10-2026-09-09.md et
        features-backlog.md). Le format binaire lui-même (4 octets,
        message-type 250 + version + code) reste conforme à la spec et
        inchangé par ce constat -- seul le comportement *serveur* face à
        ce message est désormais mieux compris. Aucun serveur supportant
        réellement xvp n'a pu être testé dans ce sandbox (le logiciel
        `xvp` lui-même, distinct d'un serveur VNC/Xvnc classique, n'a pas
        été installé -- voir features-backlog.md pour la discussion de sa
        pertinence).
        """
        if code not in (XVP_SHUTDOWN, XVP_REBOOT, XVP_RESET):
            raise ValueError(f'invalid xvp-message-code: {code}')
        code_name = {XVP_SHUTDOWN: 'XVP_SHUTDOWN', XVP_REBOOT: 'XVP_REBOOT', XVP_RESET: 'XVP_RESET'}[code]
        if not confirm:
            logger.warning(
                "send_xvp | code={code} ({code_name}) | REFUSÉ -- confirm=True requis",
                code=code,
                code_name=code_name,
            )
            raise PermissionError(
                f'send_xvp({code_name}) is a destructive remote command and requires confirm=True'
            )
        logger.warning(
            "send_xvp | code={code} ({code_name}) | commande destructive envoyée au serveur distant",
            code=code,
            code_name=code_name,
        )
        self.writer.write(bytes([250, 0, _XVP_VERSION, code]))

    async def send_xvp_shutdown(self, confirm: bool = False):
        """Demande un arrêt propre du système distant (xvp). Voir send_xvp() pour *confirm*."""
        await self.send_xvp(XVP_SHUTDOWN, confirm=confirm)

    async def send_xvp_reboot(self, confirm: bool = False):
        """Demande un redémarrage propre du système distant (xvp). Voir send_xvp() pour *confirm*."""
        await self.send_xvp(XVP_REBOOT, confirm=confirm)

    async def send_xvp_reset(self, confirm: bool = False):
        """Demande un reset brutal du système distant (xvp). Voir send_xvp() pour *confirm*."""
        await self.send_xvp(XVP_RESET, confirm=confirm)

    async def drain(self):
        """
        Waits for data to be written to the server.
        """

        await self.writer.drain()

    async def screenshot(self, x: int = 0, y: int = 0, width: Optional[int] = None, height: Optional[int] = None):
        """
        Takes a screenshot and returns a 3D RGBA array.
        """

        self.video.data = None
        self.video.refresh(x, y, width, height)
        while True:
            update_type = await self.read()
            if update_type is UpdateType.VIDEO and self.video.is_complete(x, y, width, height):
                return self.video.as_rgba(x, y, width, height)


_WEBSOCKET_GUID = b'258EAFA5-E914-47DA-95CA-C5AB0DC85B11'  # RFC 6455 §1.3, constante fixe


def _ws_frame(payload: bytes, opcode: int = 0x2, mask: bool = True) -> bytes:
    """
    Construit une trame WebSocket (RFC 6455 §5.2) contenant *payload*.
    *opcode* 0x2 = binaire (celui utilisé pour tout le trafic RFB une fois
    la poignée de main HTTP terminée), 0x8 = close, 0x9 = ping, 0xA = pong.
    RFC 6455 §5.1 impose que toute trame envoyée par un client soit
    masquée (*mask* reste donc à `True` dans tous les usages internes de
    cette fonction) ; la clé de masquage est tirée au hasard à chaque
    trame, comme l'exige la RFC (une clé fixe ou prévisible affaiblirait
    la protection - certes limitée - que le masquage apporte contre le
    cache poisoning de proxies intermédiaires mal écrits, sa seule raison
    d'être réelle d'après la RFC elle-même).
    """
    header = bytearray()
    header.append(0x80 | (opcode & 0x0F))  # FIN=1, RSV1-3=0, opcode
    length = len(payload)
    mask_bit = 0x80 if mask else 0x00
    if length <= 125:
        header.append(mask_bit | length)
    elif length <= 0xFFFF:
        header.append(mask_bit | 126)
        header += length.to_bytes(2, 'big')
    else:
        header.append(mask_bit | 127)
        header += length.to_bytes(8, 'big')
    if mask:
        key = urandom(4)
        header += key
        masked = bytes(b ^ key[i % 4] for i, b in enumerate(payload))
        return bytes(header) + masked
    return bytes(header) + payload


class _WebSocketWriter:
    """
    Enveloppe un `StreamWriter` TCP brut pour que chaque `write()` soit
    automatiquement encadré dans une trame WebSocket binaire avant d'être
    réellement écrit sur le socket -- transparent pour tout le reste
    d'`asyncvnc2.py`, qui continue à voir un objet exposant `write()`/
    `drain()`/`close()`/`wait_closed()` comme n'importe quel `StreamWriter`
    (voir `_ws_wrap_connection()` ci-dessous pour le pendant côté lecture).

    Volontairement minimal : ne cherche pas à imiter la totalité de
    l'API `StreamWriter` (pas de `.transport`, pas de `get_extra_info()`),
    seulement ce dont `Client.create()` et le reste de ce fichier se
    servent réellement. Si un futur patch a besoin de combiner ceci avec
    VeNCrypt (RFB-sur-WebSockets-sur-TLS-VeNCrypt, un empilement exotique
    mais pas interdit par les specs respectives), `_start_tls_client()`
    devra être adapté pour ne plus supposer l'existence de `.transport`.

    *mask* détermine si les trames sortantes sont masquées : `True`
    (défaut) côté client (`_ws_wrap_connection()`, utilisé par
    `connect()`), `False` côté serveur (`_ws_wrap_connection_server()`,
    utilisé par `listen()`) -- RFC 6455 §5.3 : seules les trames
    client-vers-serveur doivent être masquées, jamais l'inverse.
    """

    def __init__(self, raw_writer: StreamWriter, pump_task=None, mask: bool = True):
        self._raw = raw_writer
        self._pump_task = pump_task
        self._mask = mask

    def write(self, data: bytes) -> None:
        self._raw.write(_ws_frame(bytes(data), mask=self._mask))

    async def drain(self) -> None:
        await self._raw.drain()

    def close(self) -> None:
        try:
            self._raw.write(_ws_frame(b'', opcode=0x8, mask=self._mask))
        except Exception:
            # La connexion peut déjà être à moitié fermée (par ex. si le
            # serveur a fermé le premier) -- on n'empêche pas la fermeture
            # du côté client pour autant, le close WebSocket n'est qu'une
            # politesse protocolaire, pas une étape obligatoire pour que
            # la socket TCP sous-jacente se ferme correctement.
            pass
        self._raw.close()
        if self._pump_task is not None:
            # Découvert par un test réel contre un vrai `websockify` en
            # TLS (`wss://`) : sans cette annulation explicite, la tâche
            # de fond `_ws_pump()` peut se retrouver à lire sur une
            # socket que `self._raw.close()` vient de faire passer en
            # cours de fermeture TLS, et lever une `ssl.SSLError`
            # ("application data after close notify") jamais récupérée
            # par personne -- désormais rattrapée dans `_ws_pump()` lui-
            # même (voir plus bas), mais autant arrêter proprement cette
            # tâche dès que le côté écriture se ferme plutôt que
            # d'attendre qu'elle échoue de son côté.
            self._pump_task.cancel()

    async def wait_closed(self) -> None:
        try:
            await self._raw.wait_closed()
        except ssl.SSLError as exc:
            if 'APPLICATION_DATA_AFTER_CLOSE_NOTIFY' not in str(exc):
                raise
            # Découvert par un test réel contre un vrai `websockify` en
            # TLS (`wss://`, `--run-once`) : quand le pair ferme la
            # connexion TCP juste après avoir envoyé sa dernière trame
            # WebSocket, il arrive qu'OpenSSL considère qu'une trace
            # d'"application data" traîne encore après le close_notify au
            # moment où *nous* tentons notre propre `unwrap()` de
            # fermeture -- pas une corruption de données (la session RFB
            # entière a déjà été lue et vérifiée correcte avant d'arriver
            # ici, cette erreur ne survient que dans le ménage de
            # fermeture), plutôt une course bénigne bien connue entre
            # asyncio et OpenSSL sur la fin de vie d'une connexion TLS
            # (le même symptôme existe indépendamment de WebSockets --
            # voir bpo-39951 -- mais ne se manifestait pas avec le simple
            # `opener` TLS déjà existant testé ailleurs dans ce fichier ;
            # la combinaison avec la tâche de fond `_ws_pump()`, qui
            # continue de lire sur ce même flux jusqu'à son annulation
            # dans `close()` ci-dessus, semble être ce qui la révèle).
            # Ignorée ici uniquement pour ce message d'erreur précis --
            # toute autre `ssl.SSLError` continue de remonter normalement.


async def _ws_pump(raw_reader: StreamReader, raw_writer: StreamWriter, ws_reader: StreamReader) -> None:
    """
    Tourne en tâche de fond pendant toute la durée de vie d'une connexion
    RFB-sur-WebSockets : lit en continu des trames WebSocket sur
    *raw_reader*, et pousse le contenu utile dans *ws_reader* via
    `feed_data()` pour que le reste d'`asyncvnc2.py` -- qui n'a jamais
    entendu parler de WebSockets -- puisse continuer à faire du
    `await reader.readexactly(n)` normalement (RFC 6455 §5.4 : les
    trames de données peuvent être fragmentées sur plusieurs trames
    physiques par le réseau, indépendamment de tout découpage en messages
    applicatifs RFB -- ce pompage réassemble tout ça en un flux d'octets
    continu, exactement le service que rendrait un vrai `StreamReader` TCP
    si WebSockets n'existait pas).

    Répond aux trames *ping* par un *pong* immédiat (RFC 6455 §5.5.2 :
    "A Pong frame sent in response to a Ping frame must have identical
    'Application data' as found in the message body of the Ping frame
    being replied to"), et à une trame *close* en fermant proprement le
    flux côté lecture (`feed_eof()`) puis en arrêtant la tâche.
    """
    try:
        while True:
            first2 = await raw_reader.readexactly(2)
            opcode = first2[0] & 0x0F
            masked = bool(first2[1] & 0x80)
            length = first2[1] & 0x7F
            if length == 126:
                length = int.from_bytes(await raw_reader.readexactly(2), 'big')
            elif length == 127:
                length = int.from_bytes(await raw_reader.readexactly(8), 'big')
            mask_key = await raw_reader.readexactly(4) if masked else None
            payload = await raw_reader.readexactly(length) if length else b''
            if mask_key is not None:
                # Un serveur WebSocket conforme à la RFC ne masque jamais
                # ses trames (seul le client y est tenu) -- mais on
                # démasque quand même si le bit est mis, par tolérance
                # envers un serveur non conforme, plutôt que de planter
                # sur un flux par ailleurs parfaitement exploitable (même
                # philosophie que le traitement de "RFB 004.001" pour
                # RealVNC Enterprise ailleurs dans ce fichier).
                payload = bytes(b ^ mask_key[i % 4] for i, b in enumerate(payload))

            if opcode in (0x0, 0x2, 0x1):  # continuation / binaire / texte
                ws_reader.feed_data(payload)
            elif opcode == 0x9:  # ping -> pong immédiat, même charge utile
                raw_writer.write(_ws_frame(payload, opcode=0xA))
                await raw_writer.drain()
            elif opcode == 0xA:  # pong : rien à faire
                pass
            elif opcode == 0x8:  # close
                ws_reader.feed_eof()
                return
            # Les opcodes réservés/inconnus sont ignorés plutôt que de
            # faire échouer toute la connexion -- la RFB elle-même adopte
            # ce principe de tolérance pour ses propres extensions
            # (encodages/pseudo-encodages inconnus simplement ignorés).
    except CancelledError:
        # Une annulation explicite de cette tâche (voir
        # `_WebSocketWriter.close()`) ne doit surtout pas être avalée ici
        # -- cela romprait la sémantique normale d'annulation d'asyncio
        # (l'appelant de `task.cancel()` ne serait jamais notifié).
        raise
    except (IncompleteReadError, ConnectionError, OSError):
        # Fin de connexion pendant la lecture d'une trame -- que ce soit
        # une simple coupure TCP (`IncompleteReadError`/`ConnectionError`)
        # ou, découvert par un test réel contre un vrai `websockify` en
        # TLS (`wss://`), une erreur `ssl.SSLError` du type
        # "application data after close notify" quand le pair envoie son
        # message de fermeture TLS avant que cette tâche n'ait fini de
        # lire (`ssl.SSLError` est une sous-classe d'`OSError`, donc
        # couverte ici sans import supplémentaire). Dans tous les cas, le
        # comportement attendu (déjà celui adopté ailleurs dans ce
        # fichier pour le flux RFB lui-même) est de signaler l'EOF au
        # lecteur applicatif plutôt que de laisser une exception sortir
        # sans jamais être récupérée d'une tâche de fond que personne
        # n'attend directement (ce qui produirait le message asyncio
        # "Task exception was never retrieved" sur la sortie standard,
        # sans conséquence fonctionnelle mais trompeur pour l'appelant).
        ws_reader.feed_eof()


async def _ws_handshake(reader: StreamReader, writer: StreamWriter,
                         host: str, port: int, path: str) -> None:
    """
    Poignée de main HTTP d'upgrade vers WebSocket (RFC 6455 §4.1/§4.2),
    côté client. Envoie `Sec-WebSocket-Protocol: binary`, le sous-protocole
    que `noVNC`/`websockify` (les deux implémentations de référence pour
    RFB-sur-WebSockets) attendent -- mais n'exige pas que le serveur le
    confirme dans sa réponse : certains proxys RFB-sur-WS plus permissifs
    ne le renvoient pas alors qu'ils fonctionnent très bien par ailleurs,
    et la seule chose qui compte réellement pour la suite est que le
    serveur accepte de faire transiter des octets RFB bruts dans des
    trames binaires, ce qu'un statut 101 garantit déjà en pratique pour
    ces deux implémentations.
    """
    key = b64encode(urandom(16)).decode('ascii')
    request = (
        f'GET {path} HTTP/1.1\r\n'
        f'Host: {host}:{port}\r\n'
        f'Upgrade: websocket\r\n'
        f'Connection: Upgrade\r\n'
        f'Sec-WebSocket-Key: {key}\r\n'
        f'Sec-WebSocket-Version: 13\r\n'
        f'Sec-WebSocket-Protocol: binary\r\n'
        f'\r\n'
    ).encode('ascii')
    writer.write(request)
    await writer.drain()

    status_line = await reader.readline()
    if b'101' not in status_line:
        raise ValueError(f'WebSocket handshake refusée par le serveur : {status_line!r}')

    headers = {}
    while True:
        line = await reader.readline()
        if line in (b'\r\n', b'\n', b''):
            break
        name, _, value = line.decode('iso-8859-1').partition(':')
        headers[name.strip().lower()] = value.strip()

    expected_accept = b64encode(sha1(key.encode('ascii') + _WEBSOCKET_GUID).digest()).decode('ascii')
    if headers.get('sec-websocket-accept') != expected_accept:
        raise ValueError(
            "WebSocket handshake invalide : Sec-WebSocket-Accept ne correspond pas "
            "à Sec-WebSocket-Key (RFC 6455 §4.2.2) -- le serveur n'est probablement "
            "pas un serveur WebSocket conforme")


async def _ws_wrap_connection(reader: StreamReader, writer: StreamWriter,
                               host: str, port: int, path: str) -> Tuple[StreamReader, StreamWriter]:
    """
    Poignée de main WebSocket puis mise en place du tunnel bidirectionnel
    RFB-sur-WebSockets : à partir d'ici, tout le reste de ce fichier
    (`Client.create()` inclus, sans aucune modification) peut continuer à
    parler RFB brut sur le couple `(reader, writer)` retourné exactement
    comme s'il s'agissait d'une connexion TCP ordinaire -- c'est tout
    l'intérêt de ce point d'insertion : contrairement à VeNCrypt (qui
    s'immisce *au milieu* de la poignée de main RFB, côté sécurité),
    WebSockets se négocie entièrement *avant* que le premier octet RFB
    (`ProtocolVersion`) ne soit échangé, donc aucune des fonctions
    `_vencrypt_negotiate()` etc. n'a besoin d'avoir connaissance de ceci.
    """
    await _ws_handshake(reader, writer, host, port, path)

    ws_reader = StreamReader()
    loop = get_running_loop()
    pump_task = loop.create_task(_ws_pump(reader, writer, ws_reader))
    # Le rattacher au StreamReader lui-même (plutôt que de le laisser
    # simplement s'exécuter en arrière-plan sans référence) pour qu'il ne
    # puisse pas être ramassé par le garbage collector en cours de route
    # (piège classique documenté par asyncio lui-même pour
    # `loop.create_task()`) et pour qu'un futur appelant qui voudrait
    # l'annuler explicitement à la fermeture puisse le faire.
    ws_reader._ws_pump_task = pump_task  # type: ignore[attr-defined]

    ws_writer = _WebSocketWriter(writer, pump_task=pump_task)
    return ws_reader, ws_writer


async def _ws_handshake_server(reader: StreamReader, writer: StreamWriter) -> None:
    """
    Pendant côté serveur de `_ws_handshake()` -- utilisé par
    `listen(websocket=True)` pour accepter une connexion WebSocket
    entrante au lieu d'en initier une (RFC 6455 §4.1/§4.2, côté serveur
    cette fois) : lit la requête HTTP d'upgrade envoyée par le pair
    (typiquement un proxy tiers configuré pour relayer une reconnexion
    inversée -- `websocat` en pratique, PAS `websockify`, voir le
    commentaire dans `listen()` -- ou directement un vrai client
    WebSocket), vérifie qu'il s'agit bien d'une demande d'upgrade WebSocket
    conforme, puis répond `101 Switching Protocols` avec le
    `Sec-WebSocket-Accept` calculé à partir de la `Sec-WebSocket-Key`
    reçue.

    Volontairement tolérant sur deux points où la RFC laisse une marge
    d'interprétation aux implémentations réelles plutôt que d'imposer un
    format strict : la méthode HTTP n'est vérifiée que pour être `GET`
    (la version HTTP exacte du client n'est pas vérifiée, `HTTP/1.0`
    comme `HTTP/1.1` sont acceptés) et l'en-tête `Sec-WebSocket-Protocol`
    n'est renvoyé dans la réponse que si le client l'a lui-même proposé,
    jamais imposé unilatéralement -- même philosophie de tolérance que
    `_ws_handshake()` (client) vis-à-vis de ce même en-tête.
    """
    request_line = await reader.readline()
    if not request_line.upper().startswith(b'GET '):
        raise ValueError(f"requête HTTP invalide pour une poignée de main WebSocket : {request_line!r}")

    headers: Dict[str, str] = {}
    while True:
        line = await reader.readline()
        if line in (b'\r\n', b'\n', b''):
            break
        name, _, value = line.decode('iso-8859-1').partition(':')
        headers[name.strip().lower()] = value.strip()

    if headers.get('upgrade', '').lower() != 'websocket':
        raise ValueError(f"en-tête 'Upgrade: websocket' manquant ou incorrect : {headers.get('upgrade')!r}")
    key = headers.get('sec-websocket-key')
    if not key:
        raise ValueError("en-tête 'Sec-WebSocket-Key' manquant dans la requête d'upgrade")

    accept = b64encode(sha1(key.encode('ascii') + _WEBSOCKET_GUID).digest()).decode('ascii')
    response_lines = [
        'HTTP/1.1 101 Switching Protocols',
        'Upgrade: websocket',
        'Connection: Upgrade',
        f'Sec-WebSocket-Accept: {accept}',
    ]
    # Le sous-protocole n'est confirmé que si le client (le proxy WS en
    # face, généralement) l'a lui-même demandé -- voir docstring.
    requested_protocols = [p.strip() for p in headers.get('sec-websocket-protocol', '').split(',') if p.strip()]
    if 'binary' in requested_protocols:
        response_lines.append('Sec-WebSocket-Protocol: binary')
    response = ('\r\n'.join(response_lines) + '\r\n\r\n').encode('ascii')
    writer.write(response)
    await writer.drain()


async def _ws_wrap_connection_server(reader: StreamReader, writer: StreamWriter) -> Tuple[StreamReader, StreamWriter]:
    """
    Pendant côté serveur de `_ws_wrap_connection()`, pour `listen()`
    (reconnexion inversée) au lieu de `connect()` : accepte la poignée
    de main WebSocket entrante puis met en place le même tunnel
    bidirectionnel que côté client, à la seule différence que les trames
    sortantes ne sont pas masquées (RFC 6455 §5.3, réservé au sens
    client-vers-serveur -- voir `_WebSocketWriter`). `_ws_pump()` lui-même
    n'a besoin d'aucune adaptation : il démasque déjà les trames entrantes
    quand leur bit de masque est positionné, quel que soit le sens
    d'établissement de la connexion.
    """
    await _ws_handshake_server(reader, writer)

    ws_reader = StreamReader()
    loop = get_running_loop()
    pump_task = loop.create_task(_ws_pump(reader, writer, ws_reader))
    ws_reader._ws_pump_task = pump_task  # type: ignore[attr-defined]

    ws_writer = _WebSocketWriter(writer, pump_task=pump_task, mask=False)
    return ws_reader, ws_writer


@asynccontextmanager
async def connect(
        host: str,
        port: int = 5900,
        username: Optional[str] = None,
        password: Optional[str] = None,
        host_key: Optional[rsa.RSAPublicKey] = None,
        encodings: Optional[list] = None,
        jpeg_quality: Optional[int] = None,
        compression_level: Optional[int] = None,
        allow_indexed_colour: bool = False,
        shared: bool = True,
        opener=None,
        ssl_context: Optional[ssl.SSLContext] = None,
        websocket: bool = False,
        ws_path: str = '/'):
    """
    Make a VNC client connection. This is an async context manager that returns a connected :class:`Client` instance.

    *ssl_context*/VeNCrypt : si le serveur exige le type de sécurité 19
    (VeNCrypt) et propose le sous-type X509None ou X509Vnc (authentifié par
    certificat, ajouté le 2026-09-05 -- voir `Client.create()`/
    `_vencrypt_negotiate()`), *host* ci-dessus est automatiquement transmis
    comme nom d'hôte à vérifier contre le certificat présenté par le
    serveur -- aucun paramètre supplémentaire à passer pour ça. Voir
    `_vencrypt_negotiate()` pour la CA utilisée par défaut (magasin système)
    et comment fournir une CA privée via *ssl_context*.
    """

    opener = opener or open_connection
    reader, writer = await opener(host, port)
    if websocket:
        # Négociée entièrement avant tout octet RFB -- voir
        # _ws_wrap_connection() pour le détail ; `reader`/`writer` sont
        # remplacés ici par leurs équivalents "démasqués" du tunnel
        # WebSocket, transparents pour Client.create() ci-dessous.
        reader, writer = await _ws_wrap_connection(reader, writer, host, port, ws_path)

    client = await Client.create(reader, writer, username, password, host_key, encodings,
                                  jpeg_quality, compression_level, allow_indexed_colour, shared,
                                  ssl_context, server_hostname=host)
    try:
        yield client
    finally:
        writer.close()
        await writer.wait_closed()


@asynccontextmanager
async def listen(
        host: str = '0.0.0.0',
        port: int = 5500,
        username: Optional[str] = None,
        password: Optional[str] = None,
        host_key: Optional[rsa.RSAPublicKey] = None,
        encodings: Optional[list] = None,
        jpeg_quality: Optional[int] = None,
        compression_level: Optional[int] = None,
        allow_indexed_colour: bool = False,
        shared: bool = True,
        ssl_context: Optional[ssl.SSLContext] = None,
        websocket: bool = False):
    """
    Mode "reconnexion inversée" (reverse connection). Au lieu que ce
    client compose vers le serveur VNC (comme le fait `connect()`), on
    ouvre une socket d'écoute et on attend que ce soit le serveur qui
    vienne se connecter à nous -- cas d'usage classique quand la machine
    à administrer est derrière un NAT/pare-feu qu'elle seule peut
    traverser (ex. Xvnc + `vncconfig -connect`, WinVNC "Add new client",
    UltraVNC `-connect`). Port conventionnel pour l'écoute inversée :
    5500 (par opposition à 5900 pour le mode normal) -- confirmé par la
    documentation RealVNC/TigerVNC/TightVNC/UltraVNC, qui l'utilisent
    tous comme valeur par défaut de leur option `-listen`.

    Une fois la connexion TCP acceptée, le déroulé du protocole RFB lui-
    même est STRICTEMENT IDENTIQUE au mode normal (ProtocolVersion,
    Security, ClientInit/ServerInit, etc.) -- seul le sens
    d'établissement de la connexion TCP change, le rôle RFB de chaque
    extrémité (le serveur RFB envoie toujours le premier message, le
    client RFB répond toujours en premier lieu par ClientInit) restant
    identique. Cette fonction ne fait donc que brancher un listener
    devant `Client.create()`, déjà vérifié par ailleurs : aucun nouveau
    format de message à risque ici.

    Ferme la socket d'écoute dès qu'une connexion est acceptée (mode "un
    seul essai", le plus courant) : si le serveur distant se déconnecte,
    il faut rappeler `listen()` pour une nouvelle tentative. Une
    éventuelle deuxième connexion entrante avant la fermeture de la
    socket d'écoute est immédiatement refermée sans négociation.

    Vérifié par exécution réelle dans ce sandbox : un socket TCP
    localhost bouclé sur lui-même, avec un script jouant le rôle du
    serveur VNC distant (ProtocolVersion, Security de type None,
    SecurityResult, ServerInit fabriqués à la main), confirme que
    `listen()` accepte la connexion entrante et que le `Client` retourné
    est bien utilisable (dimensions, nom de bureau lus correctement).

    *websocket* (2026-09-04, topologie "reverse" complète confirmée le
    2026-09-10) : si `True`, la connexion TCP entrante est d'abord
    traitée comme une poignée de main WebSocket côté serveur (RFC 6455)
    avant que le protocole RFB ne démarre -- utile quand la reconnexion
    inversée doit transiter par un proxy WebSocket plutôt qu'arriver en
    TCP brut. Voir `CLAUDE.md` pour le détail des deux vérifications :
    **2026-09-04**, contre un vrai client WebSocket tiers (bibliothèque
    Python `websockets`) parlant directement à `listen()`, puis
    **2026-09-10**, contre un vrai proxy tiers (`websocat`) relayant
    effectivement une connexion TCP entrante vers une connexion
    WebSocket sortante -- la topologie "reverse" complète, qui restait
    jusque-là non couverte. **`websockify` ne convient PAS pour ce rôle
    de proxy** malgré son nom (voir le commentaire dans le corps de
    cette fonction, plus bas, pour le détail vérifié de pourquoi).

    *ssl_context*/VeNCrypt X509None/X509Vnc (2026-09-05) : contrairement à
    `connect()`, ce mode n'a pas de nom d'hôte "cible" à vérifier -- c'est
    tout le sens de la reconnexion inversée que l'appelant écoute sans
    savoir à l'avance qui va se connecter. `Client.create()` est donc
    appelé ici avec `server_hostname=None`. Vérifié par exécution réelle
    dans ce sandbox (voir CLAUDE.md) : avec `server_hostname=None`, asyncio
    (`loop.start_tls()`/`SSLContext.wrap_bio()`) NE lève PAS d'erreur même
    si *ssl_context* a `check_hostname=True` (le défaut de
    `ssl.create_default_context()`, utilisé si *ssl_context* n'est pas
    fourni) -- il ignore silencieusement la vérification du nom d'hôte,
    faute d'hôte à vérifier, sans désactiver pour autant la vérification de
    la CHAINE de certification contre le magasin de CA de confiance
    (celle-ci continue de s'appliquer normalement). Concrètement : un
    serveur distant qui présente un certificat signé par une CA que
    l'appelant reconnaît est accepté ; un serveur qui présente un
    certificat signé par une CA différente est rejeté avec une vraie
    `ssl.SSLCertVerificationError`, exactement comme pour `connect()` --
    seule la correspondance nom-d'hôte/certificat n'est pas vérifiée ici
    (elle n'aurait de toute façon aucun sens en reconnexion inversée). Pour
    une CA privée, fournir son propre *ssl_context* avec
    `load_verify_locations()`, exactement comme pour `connect()` -- aucun
    besoin de désactiver `check_hostname` soi-même, contrairement à ce
    qu'une lecture rapide de la documentation `ssl` standard (pensée pour
    l'API synchrone `SSLContext.wrap_socket()`, qui elle lève bien
    `ValueError("check_hostname requires server_hostname")` dans cette
    situation -- un chemin de code différent de celui utilisé ici) pourrait
    laisser croire.
    """
    incoming: Queue = Queue(maxsize=1)

    async def _accept(reader: StreamReader, writer: StreamWriter):
        try:
            incoming.put_nowait((reader, writer))
        except QueueFull:
            # Une connexion inattendue en plus de la première -- ce mode
            # ne gère qu'une seule reconnexion inversée à la fois.
            writer.close()

    server = await start_server(_accept, host, port)
    try:
        reader, writer = await incoming.get()
    finally:
        # On ferme la socket d'écoute (on n'accepte plus de nouvelles
        # connexions) mais SANS attendre server.wait_closed() : sur les
        # versions récentes d'asyncio, wait_closed() attend aussi la
        # fermeture des connexions déjà acceptées -- exactement celle
        # qu'on vient d'extraire de la queue et qu'on s'apprête à garder
        # ouverte pour toute la session RFB. L'attendre ici bloquerait
        # indéfiniment (vérifié dans ce sandbox : deadlock reproductible
        # sous Python 3.12 tant que la connexion RFB reste active).
        server.close()

    if websocket:
        # Rôle serveur WebSocket -- voir _ws_wrap_connection_server()
        # pour le détail. Utile quand la reconnexion inversée doit
        # transiter par un proxy qui relaie une connexion TCP entrante
        # vers une connexion WebSocket sortante, plutôt que la connexion
        # TCP brute que ce mode attend par défaut.
        #
        # ATTENTION, vérifié réellement le 2026-09-10 : `websockify` (le
        # proxy de référence utilisé ailleurs dans ce fichier pour le
        # rôle CLIENT, voir `connect()`) NE PEUT PAS jouer ce rôle-ci.
        # `websockify --help` ne propose que des adresses d'ÉCOUTE en
        # WS/HTTP (`source_port`, `--unix-listen`) et des CIBLES en TCP,
        # socket Unix (`--unix-target`) ou sous-processus
        # (`-- WRAP_COMMAND_LINE`) -- jamais l'inverse (pas de mode
        # "écoute TCP, cible WS"). Un exemple d'invocation
        # (`websockify LISTEN_HOST:LISTEN_PORT --wait`) figurait ici par
        # erreur : `--wait` n'est même pas une option reconnue, et sans
        # elle, une adresse seule échoue avec "Too few arguments" --
        # confirmé par exécution réelle des deux cas dans ce sandbox.
        # Outil qui convient réellement pour ce rôle, vérifié de bout en
        # bout le 2026-09-10 (voir `CLAUDE.md`) : `websocat`
        # (<https://github.com/vi/websocat>, indépendant de ce fork et
        # de `websockify`), par ex.
        # `websocat --binary tcp-listen:BRIDGE_HOST:BRIDGE_PORT ws://LISTEN_HOST:LISTEN_PORT/`.
        reader, writer = await _ws_wrap_connection_server(reader, writer)

    client = await Client.create(reader, writer, username, password, host_key, encodings,
                                  jpeg_quality, compression_level, allow_indexed_colour, shared,
                                  ssl_context)
    try:
        yield client
    finally:
        writer.close()
        await writer.wait_closed()