Lucian Petrut
2099591245
VDDK always decompresses the chunks it retrieves. However, some callers may be interested in the compressed chunks, which may be forwarded to another service involved in the backup process. We'll add a read flag (skip_decompression). If set, the read operation will return a "ReadResult" container, containing a list of fragments and their compressed / decompressed lengths. If the returned buffer size matches the AIO buffer size (which now becomes configurable), at most one fragment will be returned. While at it, we're adding perf tests that check various AIO buffer sizes, cross checking against VDDK.
677 lines
25 KiB
Python
677 lines
25 KiB
Python
# Copyright 2026 Cloudbase Solutions Srl
|
|
# All Rights Reserved.
|
|
|
|
"""VDDK-compatible NFC disk open, sector read, and sector write.
|
|
|
|
After ``nfc_auth.connect_authd`` returns ``200 Connect``, NBD
|
|
(``useSSL=0``) stops using ``SSL_write`` and speaks NFC as raw TCP on
|
|
that file descriptor. NBDSSL (``useSSL=1``) starts a second TLS
|
|
handshake on the same TCP connection (``200 Connect ha-nfcssl``) and
|
|
speaks the same NFC frames as TLS application data. This module dups
|
|
the authd fd and speaks:
|
|
|
|
1. Classic 264-byte NFC messages (handshake, version, connection data,
|
|
AIO session open).
|
|
2. NFC AIO frames (16-byte header plus payload) to open a VMDK and read
|
|
or write sectors.
|
|
|
|
pyVmomi is not involved here; the ticket and TLS authd handshake already
|
|
happened in ``nfc_auth``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import socket
|
|
import ssl
|
|
import struct
|
|
from dataclasses import dataclass
|
|
|
|
from openvixdisklib import fastlz
|
|
from openvixdisklib.nfc_auth import NfcAuthSession, _ssl_client_context
|
|
|
|
NFC_MSG_SIZE = 264
|
|
NFC_AIO_MAGIC = 0xA100DA7A
|
|
NFC_AIO_HDR_SIZE = 16
|
|
NFC_SECTOR_SIZE = 512
|
|
NFC_PROTOCOL_VERSION = 11
|
|
# Max data bytes in one AIO IO request/reply fragment. Sent as
|
|
# OPEN_SESSION ``bufSize`` (VDDK ``vixDiskLib.nfcAio.Session.BufSizeIn64KB``
|
|
# times 64 KiB). ESXi read extras use this size; 2 MiB (32) works on
|
|
# ESXi 8, 16 MiB and 32 MiB do not.
|
|
NFC_AIO_BUFFER_SIZE = 65536
|
|
NFC_AIO_BUFFER_COUNT = 1
|
|
|
|
# Classic NFC message types observed on the wire (uint32 at offset 0).
|
|
NFC_MSG_SESSION_COMPLETE = 4
|
|
NFC_MSG_SESSION_PARAMS = 33
|
|
NFC_MSG_SESSION_PARAMS_REPLY = 36
|
|
NFC_MSG_HANDSHAKE = 43
|
|
NFC_MSG_VERSION = 51
|
|
NFC_MSG_AIO_SESSION_OPEN = 52
|
|
NFC_MSG_CONNECTION_DATA = 54
|
|
NFC_MSG_SESSION_FEATURES = 55
|
|
|
|
# SessionParams / feature bits from VDDK logs (interruption | switch).
|
|
NFC_SESSION_FEATURE_INTERRUPTION_SWITCH = 3
|
|
|
|
# AIO message types (NfcAioSendMessage "type = N").
|
|
NFC_AIO_MSG_ERROR = 1
|
|
NFC_AIO_MSG_OPEN_SESSION = 2
|
|
NFC_AIO_MSG_CLOSE_SESSION = 3
|
|
NFC_AIO_MSG_OPEN_FILE = 4
|
|
NFC_AIO_MSG_CLOSE_FILE = 5
|
|
NFC_AIO_MSG_IO = 7
|
|
NFC_AIO_MSG_SET_SOCK_OPTS = 9
|
|
NFC_AIO_MSG_DDB_GET = 11
|
|
NFC_AIO_MSG_SET_RES_POOL = 22
|
|
|
|
# Open-file body: file type NFC_DISK. 0x1e is what VDDK sends for
|
|
# VIXDISKLIB_FLAG_OPEN_READ_ONLY; writable opens clear bit 0x04 (0x1a).
|
|
NFC_DISK = 2
|
|
NFC_OPEN_FLAGS_READ_ONLY = 0x1E
|
|
NFC_OPEN_FLAGS_READ_WRITE = 0x1A
|
|
|
|
NFC_AIO_IO_WRITE = 0
|
|
NFC_AIO_IO_READ = 1
|
|
|
|
# High 32 bits of the IO opcode uint64. Captured from VDDK FASTLZ:
|
|
# writes that shrink go on the wire as type 2; incompressible writes
|
|
# fall back to type 0 with raw extra data.
|
|
NFC_COMPRESSION_NONE = 0
|
|
NFC_COMPRESSION_FASTLZ = 2
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ReadFragment:
|
|
"""One NFC AIO extra in a packed skip-decompression ``buf``.
|
|
|
|
Views into ``buf`` (``buf[offset:offset + length]``) are valid
|
|
until the next ``read`` into the same buffer.
|
|
|
|
``dest`` is NFC payload offset 28: the byte offset of this fragment
|
|
**inside this uncompressed read**, starting at 0. It is not a disk
|
|
LBA and not a byte offset from the start of the VMDK. The disk byte
|
|
address is ``start_sector * sector_size + dest``.
|
|
|
|
``offset`` is where this extra sits in packed ``buf`` (densely from
|
|
0 in receive order). ``length`` is the extra on the wire.
|
|
``uncompressed_length`` is NFC payload offset 32.
|
|
"""
|
|
|
|
dest: int
|
|
uncompressed_length: int
|
|
compression_type: int
|
|
offset: int
|
|
length: int
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ReadResult:
|
|
"""Outcome of ``NfcDisk.readinto`` / ``VixDiskLibHandle.read``.
|
|
|
|
``compressed_length`` is bytes of extra on the wire. With
|
|
``skip_decompression=True``, extras are packed in ``buf`` from
|
|
offset 0 and ``fragments`` describes them. The decompressing path
|
|
sets ``fragments`` to empty so callers can still use the lengths
|
|
for metrics.
|
|
"""
|
|
|
|
uncompressed_length: int
|
|
compressed_length: int
|
|
fragments: tuple[ReadFragment, ...]
|
|
|
|
|
|
class NfcProtocolError(ConnectionError):
|
|
"""Raised when an NFC message is malformed or reports failure."""
|
|
|
|
|
|
def takeover_authd_socket(ssock: ssl.SSLSocket) -> socket.socket:
|
|
"""Return a raw socket on the authd TCP connection.
|
|
|
|
VDDK writes NFC with ``write(SSL_get_fd(ssl), ...)`` after PROXY, so
|
|
those bytes are not TLS records. Duping the fd lets Python do the
|
|
same without ``SSLSocket.send`` re-encrypting, and without
|
|
``SSL_shutdown``.
|
|
|
|
Args:
|
|
ssock: The TLS socket from ``nfc_auth.connect_authd``.
|
|
"""
|
|
timeout = ssock.gettimeout()
|
|
raw = socket.socket(
|
|
family=ssock.family,
|
|
type=ssock.type,
|
|
proto=ssock.proto,
|
|
fileno=os.dup(ssock.fileno()),
|
|
)
|
|
raw.settimeout(timeout)
|
|
_enable_tcp_nodelay(raw)
|
|
return raw
|
|
|
|
|
|
def wrap_nfcssl_socket(ssock: ssl.SSLSocket, server_hostname: str) -> ssl.SSLSocket:
|
|
"""Start the second TLS session used by NBDSSL after PROXY.
|
|
|
|
After ``200 Connect ha-nfcssl``, authd TLS is finished and
|
|
``ha-nfcssl`` expects a new ClientHello on the same TCP connection.
|
|
The fd is dup'd so the original authd ``SSLSocket`` can be closed
|
|
later without ``SSL_shutdown`` of this NFCSSL session.
|
|
|
|
Args:
|
|
ssock: The TLS socket from ``nfc_auth.connect_authd``.
|
|
server_hostname: Host name passed to ``SSLContext.wrap_socket``.
|
|
"""
|
|
raw = takeover_authd_socket(ssock)
|
|
ssl_context = _ssl_client_context(verify=False)
|
|
try:
|
|
return ssl_context.wrap_socket(raw, server_hostname=server_hostname)
|
|
except Exception:
|
|
raw.close()
|
|
raise
|
|
|
|
|
|
def _enable_tcp_nodelay(sock: socket.socket) -> None:
|
|
"""Disable Nagle so a small AIO header is not held back from its extra."""
|
|
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
|
|
|
|
|
|
def _recvn(sock: socket.socket, size: int) -> bytes:
|
|
buf = bytearray(size)
|
|
_recvn_into(sock, memoryview(buf))
|
|
return bytes(buf)
|
|
|
|
|
|
def _recvn_into(sock: socket.socket, buf: memoryview) -> None:
|
|
"""Read exactly ``len(buf)`` bytes into ``buf``."""
|
|
view = buf.cast("B") if buf.format != "B" else buf
|
|
filled = 0
|
|
n = len(view)
|
|
while filled < n:
|
|
got = sock.recv_into(view[filled:n])
|
|
if not got:
|
|
raise NfcProtocolError(
|
|
f"NFC connection closed, needed {n} bytes, got {filled}"
|
|
)
|
|
filled += got
|
|
|
|
|
|
def _writable_bytes(buf: bytearray | memoryview, length: int) -> memoryview:
|
|
"""Return a writable ``B`` view of the first ``length`` bytes of ``buf``."""
|
|
view = buf if isinstance(buf, memoryview) else memoryview(buf)
|
|
if view.readonly:
|
|
raise TypeError("read buffer is read-only")
|
|
raw = view.cast("B") if view.format != "B" else view
|
|
if len(raw) < length:
|
|
raise RuntimeError(f"read buffer is {len(raw)} bytes, need {length}")
|
|
return raw[:length]
|
|
|
|
|
|
def _aio_extra_len(ctype: int, body: bytes, chunk_len: int) -> int:
|
|
"""Return this fragment's extra size on the wire."""
|
|
if ctype == NFC_COMPRESSION_FASTLZ:
|
|
extra_len = struct.unpack_from("<I", body, 36)[0]
|
|
if extra_len < 1:
|
|
raise NfcProtocolError(
|
|
f"FastLZ extra length {extra_len} is invalid, chunk {chunk_len}"
|
|
)
|
|
return extra_len
|
|
if ctype == NFC_COMPRESSION_NONE:
|
|
return chunk_len
|
|
raise NfcProtocolError(f"unsupported NFC IO compression type {ctype}")
|
|
|
|
|
|
def _send_nfc_msg(sock: socket.socket, msg_type: int, body: bytes = b"") -> None:
|
|
if len(body) > NFC_MSG_SIZE - 4:
|
|
raise ValueError("NFC classic message body too large")
|
|
frame = struct.pack("<I", msg_type) + body
|
|
sock.sendall(frame.ljust(NFC_MSG_SIZE, b"\x00"))
|
|
|
|
|
|
def _recv_nfc_msg(sock: socket.socket) -> tuple[int, bytes]:
|
|
frame = _recvn(sock, NFC_MSG_SIZE)
|
|
msg_type = struct.unpack_from("<I", frame)[0]
|
|
return msg_type, frame[4:]
|
|
|
|
|
|
def _pack_aio_hdr(msg_type: int, payload_size: int, op_id: int) -> bytes:
|
|
return struct.pack("<IIII", NFC_AIO_MAGIC, msg_type, payload_size, op_id)
|
|
|
|
|
|
def _unpack_aio_hdr(hdr: bytes) -> tuple[int, int, int]:
|
|
magic, msg_type, size, op_id = struct.unpack_from("<IIII", hdr)
|
|
if magic != NFC_AIO_MAGIC:
|
|
raise NfcProtocolError(
|
|
f"AIO header magic mismatch: 0x{magic:x}, expected 0x{NFC_AIO_MAGIC:x}"
|
|
)
|
|
if msg_type == NFC_AIO_MSG_ERROR:
|
|
raise NfcProtocolError(f"AIO error opId={op_id} size={size}")
|
|
return msg_type, size, op_id
|
|
|
|
|
|
class NfcDisk:
|
|
"""An NFC AIO session with one VMDK opened for I/O."""
|
|
|
|
def __init__(
|
|
self,
|
|
sock: socket.socket,
|
|
path: str,
|
|
handle: int,
|
|
sector_size: int,
|
|
compression: int = NFC_COMPRESSION_NONE,
|
|
aio_buffer_size: int = NFC_AIO_BUFFER_SIZE,
|
|
aio_buffer_count: int = NFC_AIO_BUFFER_COUNT,
|
|
) -> None:
|
|
"""Wrap an AIO session that already has ``path`` open.
|
|
|
|
Args:
|
|
sock: NFC socket after handshake (raw TCP for nbd, TLS for nbdssl).
|
|
path: Datastore path that was opened.
|
|
handle: Server file handle from OPEN_FILE.
|
|
sector_size: Sector size from the OPEN_FILE reply.
|
|
compression: NFC IO compression type (``NFC_COMPRESSION_NONE``
|
|
or ``NFC_COMPRESSION_FASTLZ``).
|
|
aio_buffer_size: OPEN_SESSION extra size in bytes (default
|
|
``NFC_AIO_BUFFER_SIZE``, 64 KiB). ESXi read extras are
|
|
at most this large.
|
|
aio_buffer_count: OPEN_SESSION buffer pool count (default
|
|
``NFC_AIO_BUFFER_COUNT``).
|
|
"""
|
|
self._sock = sock
|
|
self._op_id = 0
|
|
self.path = path
|
|
self.handle = handle
|
|
self.sector_size = sector_size
|
|
self.compression = compression
|
|
self.aio_buffer_size = aio_buffer_size
|
|
self.aio_buffer_count = aio_buffer_count
|
|
self._closed = False
|
|
|
|
def _next_op_id(self) -> int:
|
|
op_id = self._op_id
|
|
self._op_id += 1
|
|
return op_id
|
|
|
|
def _aio_send(self, msg_type: int, payload: bytes, extra: bytes = b"") -> int:
|
|
"""Send one AIO request (header, payload, and extra in one write)."""
|
|
op_id = self._next_op_id()
|
|
self._sock.sendall(
|
|
_pack_aio_hdr(msg_type, len(payload), op_id) + payload + extra
|
|
)
|
|
return op_id
|
|
|
|
def _aio_recv_reply(self) -> tuple[int, int, bytes]:
|
|
"""Read the next AIO reply. Returns ``(type, op_id, payload)``."""
|
|
rhdr = _recvn(self._sock, NFC_AIO_HDR_SIZE)
|
|
magic, rtype, rsize, rop = struct.unpack_from("<IIII", rhdr)
|
|
if magic != NFC_AIO_MAGIC:
|
|
raise NfcProtocolError(
|
|
f"AIO header magic mismatch: 0x{magic:x}, expected 0x{NFC_AIO_MAGIC:x}"
|
|
)
|
|
body = _recvn(self._sock, rsize) if rsize else b""
|
|
if rtype == NFC_AIO_MSG_ERROR:
|
|
raise NfcProtocolError(f"AIO error opId={rop} size={rsize} {body.hex()}")
|
|
return rtype, rop, body
|
|
|
|
def _aio_roundtrip(
|
|
self, msg_type: int, payload: bytes, extra: bytes = b"", extra_recv: int = 0
|
|
) -> bytes:
|
|
"""Send one AIO request and return the reply payload (+ extra)."""
|
|
op_id = self._aio_send(msg_type, payload, extra)
|
|
rtype, rop, body = self._aio_recv_reply()
|
|
if rtype != msg_type or rop != op_id:
|
|
raise NfcProtocolError(
|
|
f"AIO reply type={rtype} opId={rop}, "
|
|
f"expected type={msg_type} opId={op_id}"
|
|
)
|
|
if extra_recv:
|
|
body += _recvn(self._sock, extra_recv)
|
|
return body
|
|
|
|
def read(self, start_sector: int, num_sectors: int = 1) -> bytes:
|
|
"""Read ``num_sectors`` starting at ``start_sector``.
|
|
|
|
Matches ``VixDiskLib_Read``: one ``NFC_AIO_MSG_IO`` request in
|
|
byte units. If the length exceeds the session AIO buffer the
|
|
server replies with several same-``opId`` fragments, which are
|
|
placed by the fragment byte offset in the reply (they may arrive
|
|
out of order). FASTLZ open requests compression in the opcode;
|
|
each reply fragment may be compressed (type 2) or raw (type 0).
|
|
|
|
Args:
|
|
start_sector: Sector offset from the start of the disk.
|
|
num_sectors: Number of sectors to read.
|
|
"""
|
|
if num_sectors < 1:
|
|
raise ValueError("num_sectors must be at least 1")
|
|
buf = bytearray(num_sectors * self.sector_size)
|
|
self.readinto(start_sector, num_sectors, buf)
|
|
return bytes(buf)
|
|
|
|
def readinto(
|
|
self,
|
|
start_sector: int,
|
|
num_sectors: int,
|
|
buf: bytearray | memoryview,
|
|
skip_decompression: bool = False,
|
|
) -> ReadResult:
|
|
"""Read ``num_sectors`` into ``buf`` starting at ``start_sector``.
|
|
|
|
Uncompressed fragments are received directly into ``buf``. FastLZ
|
|
still decompresses into a temporary buffer, then copies the
|
|
result, unless ``skip_decompression`` is set. ``buf`` must be
|
|
writable and at least ``num_sectors * sector_size`` bytes (a
|
|
``get_buffer`` ctypes array is wrapped with ``memoryview`` by
|
|
the VDDK-shaped handle).
|
|
|
|
Args:
|
|
start_sector: Sector offset from the start of the disk.
|
|
num_sectors: Number of sectors to read.
|
|
buf: Destination buffer.
|
|
skip_decompression: When True, pack NFC extras densely from
|
|
offset 0 without FastLZ decompress. Fragment metadata
|
|
is in the returned ``ReadResult``. Completion still
|
|
uses uncompressed chunk lengths.
|
|
|
|
Returns:
|
|
Lengths of the uncompressed request and of extras on the
|
|
wire. ``fragments`` is populated only when skipping
|
|
decompression.
|
|
"""
|
|
if num_sectors < 1:
|
|
raise ValueError("num_sectors must be at least 1")
|
|
length = num_sectors * self.sector_size
|
|
data = _writable_bytes(buf, length)
|
|
offset = start_sector * self.sector_size
|
|
opcode = NFC_AIO_IO_READ | (self.compression << 32)
|
|
payload = struct.pack(
|
|
"<QQQQIII", self.handle, opcode, offset, length, length, length, 0
|
|
)
|
|
op_id = self._next_op_id()
|
|
self._sock.sendall(_pack_aio_hdr(NFC_AIO_MSG_IO, len(payload), op_id) + payload)
|
|
filled = 0
|
|
wire_bytes = 0
|
|
packed_offset = 0
|
|
fragments: list[ReadFragment] = []
|
|
seen: set[int] = set()
|
|
while filled < length:
|
|
rhdr = _recvn(self._sock, NFC_AIO_HDR_SIZE)
|
|
rtype, rsize, rop = _unpack_aio_hdr(rhdr)
|
|
if rtype != NFC_AIO_MSG_IO or rop != op_id:
|
|
raise NfcProtocolError(
|
|
f"AIO IO reply type={rtype} opId={rop}, "
|
|
f"expected type={NFC_AIO_MSG_IO} opId={op_id}"
|
|
)
|
|
body = _recvn(self._sock, rsize)
|
|
if rsize < 36:
|
|
raise NfcProtocolError(f"AIO IO reply payload too short: {rsize}")
|
|
# Fragments may arrive out of order. Offset 28 is the byte
|
|
# offset of this chunk within the request (0, 65536, …),
|
|
# not a 0-based index. Offset 32 is the uncompressed
|
|
# fragment length. When the opcode high bits are FastLZ
|
|
# (2), extra data is compressed and offset 36 is its size.
|
|
opcode = struct.unpack_from("<Q", body, 8)[0]
|
|
dest, chunk_len = struct.unpack_from("<II", body, 28)
|
|
if dest in seen or chunk_len == 0 or dest + chunk_len > length:
|
|
raise NfcProtocolError(
|
|
f"AIO IO chunk offset={dest} length={chunk_len} invalid, "
|
|
f"request {length}"
|
|
)
|
|
seen.add(dest)
|
|
ctype = opcode >> 32
|
|
extra_len = _aio_extra_len(ctype, body, chunk_len)
|
|
if skip_decompression:
|
|
end = packed_offset + extra_len
|
|
if end > length:
|
|
raise NfcProtocolError(
|
|
f"packed extras {end} bytes exceed request {length}"
|
|
)
|
|
_recvn_into(self._sock, data[packed_offset:end])
|
|
fragments.append(
|
|
ReadFragment(
|
|
dest=dest,
|
|
uncompressed_length=chunk_len,
|
|
compression_type=ctype,
|
|
offset=packed_offset,
|
|
length=extra_len,
|
|
)
|
|
)
|
|
packed_offset = end
|
|
elif ctype == NFC_COMPRESSION_FASTLZ:
|
|
extra = _recvn(self._sock, extra_len)
|
|
try:
|
|
chunk = fastlz.decompress(extra, chunk_len)
|
|
except ValueError as exc:
|
|
raise NfcProtocolError(
|
|
f"FastLZ read fragment failed: {exc}"
|
|
) from exc
|
|
if len(chunk) != chunk_len:
|
|
raise NfcProtocolError(
|
|
f"FastLZ read got {len(chunk)} bytes, expected {chunk_len}"
|
|
)
|
|
data[dest : dest + chunk_len] = chunk
|
|
elif ctype == NFC_COMPRESSION_NONE:
|
|
_recvn_into(self._sock, data[dest : dest + chunk_len])
|
|
else:
|
|
raise NfcProtocolError(f"unsupported NFC IO compression type {ctype}")
|
|
wire_bytes += extra_len
|
|
filled += chunk_len
|
|
return ReadResult(
|
|
uncompressed_length=length,
|
|
compressed_length=wire_bytes,
|
|
fragments=tuple(fragments) if skip_decompression else (),
|
|
)
|
|
|
|
def write(self, start_sector: int, num_sectors: int, data: bytes) -> None:
|
|
"""Write ``num_sectors`` starting at ``start_sector``.
|
|
|
|
Matches ``VixDiskLib_Write``: one ``NFC_AIO_MSG_IO`` ``opId``
|
|
for the whole call. Chunks larger than the session AIO buffer
|
|
are extra fragments with that same ``opId``; the server replies
|
|
once. FASTLZ open compresses each fragment when that shrinks it.
|
|
|
|
Args:
|
|
start_sector: Sector offset from the start of the disk.
|
|
num_sectors: Number of sectors to write.
|
|
data: Bytes to write; length must be ``num_sectors * sector_size``.
|
|
"""
|
|
if num_sectors < 1:
|
|
raise ValueError("num_sectors must be at least 1")
|
|
length = num_sectors * self.sector_size
|
|
if len(data) != length:
|
|
raise ValueError(f"write data is {len(data)} bytes, need {length}")
|
|
disk_offset = start_sector * self.sector_size
|
|
op_id = self._next_op_id()
|
|
frag_offset = 0
|
|
while frag_offset < length:
|
|
chunk = data[frag_offset : frag_offset + self.aio_buffer_size]
|
|
extra = chunk
|
|
extra_len = len(chunk)
|
|
ctype = NFC_COMPRESSION_NONE
|
|
if self.compression == NFC_COMPRESSION_FASTLZ and extra_len >= 16:
|
|
compressed = fastlz.compress(chunk)
|
|
if compressed and len(compressed) < extra_len:
|
|
extra = compressed
|
|
extra_len = len(compressed)
|
|
ctype = NFC_COMPRESSION_FASTLZ
|
|
opcode = NFC_AIO_IO_WRITE | (ctype << 32)
|
|
payload = struct.pack(
|
|
"<QQQIIIII",
|
|
self.handle,
|
|
opcode,
|
|
disk_offset,
|
|
length,
|
|
frag_offset,
|
|
len(chunk),
|
|
extra_len,
|
|
0,
|
|
)
|
|
self._sock.sendall(
|
|
_pack_aio_hdr(NFC_AIO_MSG_IO, len(payload), op_id) + payload + extra
|
|
)
|
|
frag_offset += len(chunk)
|
|
rtype, rop, _body = self._aio_recv_reply()
|
|
if rtype != NFC_AIO_MSG_IO or rop != op_id:
|
|
raise NfcProtocolError(
|
|
f"AIO IO write reply type={rtype} opId={rop}, "
|
|
f"expected type={NFC_AIO_MSG_IO} opId={op_id}"
|
|
)
|
|
|
|
def close(self) -> None:
|
|
"""Close the VMDK, the AIO session, and the classic NFC session."""
|
|
if self._closed:
|
|
return
|
|
self._closed = True
|
|
try:
|
|
self._aio_roundtrip(NFC_AIO_MSG_CLOSE_FILE, struct.pack("<Q", self.handle))
|
|
self._aio_roundtrip(NFC_AIO_MSG_CLOSE_SESSION, struct.pack("<I", 0))
|
|
_send_nfc_msg(self._sock, NFC_MSG_SESSION_COMPLETE)
|
|
finally:
|
|
try:
|
|
self._sock.close()
|
|
except OSError:
|
|
pass
|
|
|
|
def __enter__(self) -> NfcDisk:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
self.close()
|
|
|
|
|
|
def _handshake(sock: socket.socket, client_name: str, op_id: str, version: int) -> None:
|
|
"""Run the classic NFC session handshake used by VDDK NBD."""
|
|
_send_nfc_msg(sock, NFC_MSG_HANDSHAKE, b"PlainText")
|
|
_send_nfc_msg(sock, NFC_MSG_SESSION_PARAMS)
|
|
reply_type, _ = _recv_nfc_msg(sock)
|
|
if reply_type != NFC_MSG_SESSION_PARAMS_REPLY:
|
|
raise NfcProtocolError(
|
|
f"expected session-params reply {NFC_MSG_SESSION_PARAMS_REPLY}, "
|
|
f"got {reply_type}"
|
|
)
|
|
|
|
_send_nfc_msg(sock, NFC_MSG_VERSION, struct.pack("<I", version))
|
|
reply_type, body = _recv_nfc_msg(sock)
|
|
if reply_type != NFC_MSG_VERSION:
|
|
raise NfcProtocolError(
|
|
f"expected version reply {NFC_MSG_VERSION}, got {reply_type}"
|
|
)
|
|
remote_version = struct.unpack_from("<I", body)[0]
|
|
if remote_version < 3:
|
|
raise NfcProtocolError(
|
|
f"NFC server version {remote_version} is too old for AIO"
|
|
)
|
|
|
|
name_b = client_name.encode("ascii")
|
|
op_b = op_id.encode("ascii")
|
|
_send_nfc_msg(
|
|
sock, NFC_MSG_CONNECTION_DATA, struct.pack("<II", len(name_b), len(op_b))
|
|
)
|
|
sock.sendall(name_b)
|
|
sock.sendall(op_b)
|
|
_send_nfc_msg(
|
|
sock,
|
|
NFC_MSG_SESSION_FEATURES,
|
|
struct.pack("<I", NFC_SESSION_FEATURE_INTERRUPTION_SWITCH),
|
|
)
|
|
_send_nfc_msg(sock, NFC_MSG_AIO_SESSION_OPEN)
|
|
reply_type, _ = _recv_nfc_msg(sock)
|
|
if reply_type != NFC_MSG_AIO_SESSION_OPEN:
|
|
raise NfcProtocolError(
|
|
f"expected AIO session-open reply "
|
|
f"{NFC_MSG_AIO_SESSION_OPEN}, got {reply_type}"
|
|
)
|
|
|
|
|
|
def _aio_prepare(disk: NfcDisk) -> None:
|
|
open_session = struct.pack(
|
|
"<IIII", 0, disk.aio_buffer_size, disk.aio_buffer_count, 0
|
|
)
|
|
disk._aio_roundtrip(NFC_AIO_MSG_OPEN_SESSION, open_session)
|
|
disk._aio_roundtrip(NFC_AIO_MSG_SET_SOCK_OPTS, bytes(12))
|
|
disk._aio_roundtrip(NFC_AIO_MSG_SET_RES_POOL, struct.pack("<I", 1))
|
|
|
|
|
|
def _parse_open_reply(body: bytes) -> tuple[int, int]:
|
|
if len(body) < 40:
|
|
raise NfcProtocolError(f"OPEN_FILE reply too short: {len(body)}")
|
|
handle, file_type, _flags = struct.unpack_from("<QII", body, 8)
|
|
sector_size = struct.unpack_from("<I", body, 36)[0]
|
|
if file_type != NFC_DISK:
|
|
raise NfcProtocolError(f"opened file type {file_type}, expected NFC_DISK")
|
|
if sector_size == 0:
|
|
sector_size = NFC_SECTOR_SIZE
|
|
return handle, sector_size
|
|
|
|
|
|
def open_disk(
|
|
session: NfcAuthSession,
|
|
disk_path: str,
|
|
client_name: str = "vddk",
|
|
op_id: str = "nbdmode",
|
|
version: int = NFC_PROTOCOL_VERSION,
|
|
read_only: bool = True,
|
|
compression: int = NFC_COMPRESSION_NONE,
|
|
aio_buffer_size: int = NFC_AIO_BUFFER_SIZE,
|
|
aio_buffer_count: int = NFC_AIO_BUFFER_COUNT,
|
|
) -> NfcDisk:
|
|
"""Open ``disk_path`` over the authenticated authd socket.
|
|
|
|
Matches VDDK ``VixDiskLib_Open`` of a datastore path after the NFC
|
|
ticket and authd PROXY handshake: session init, AIO open, then
|
|
``NFC_AIO_MSG_OPEN_FILE`` with type ``NFC_DISK``. NBD dups the
|
|
authd fd and sends plaintext NFC. NBDSSL wraps that dup in a
|
|
second TLS session (``session.nfc_ssl``).
|
|
|
|
Args:
|
|
session: Result of ``nfc_auth.authenticate``.
|
|
disk_path: Datastore path, for example
|
|
``[datastore0] vm/vm.vmdk``.
|
|
client_name: NFC client name; VDDK sends ``vddk``.
|
|
op_id: NFC operation id; VDDK NBD sends ``nbdmode``.
|
|
version: Client NFC protocol version (lab ESXi answered 11).
|
|
read_only: When True, open with VDDK's read-only NFC flags.
|
|
compression: ``NFC_COMPRESSION_NONE`` or ``NFC_COMPRESSION_FASTLZ``.
|
|
OPEN_FILE flags are unchanged; compression is per IO message.
|
|
aio_buffer_size: Extra size advertised in OPEN_SESSION (bytes).
|
|
aio_buffer_count: Buffer pool count advertised in OPEN_SESSION.
|
|
"""
|
|
if aio_buffer_size < 1:
|
|
raise ValueError("aio_buffer_size must be at least 1")
|
|
if aio_buffer_count < 1:
|
|
raise ValueError("aio_buffer_count must be at least 1")
|
|
if compression not in (NFC_COMPRESSION_NONE, NFC_COMPRESSION_FASTLZ):
|
|
raise NotImplementedError(
|
|
f"NFC compression type {compression} is not supported"
|
|
)
|
|
sock: socket.socket
|
|
if session.nfc_ssl:
|
|
sock = wrap_nfcssl_socket(session.authd_sock, session.ticket.host)
|
|
else:
|
|
sock = takeover_authd_socket(session.authd_sock)
|
|
try:
|
|
_handshake(sock, client_name, op_id, version)
|
|
disk = NfcDisk(
|
|
sock,
|
|
disk_path,
|
|
handle=0,
|
|
sector_size=NFC_SECTOR_SIZE,
|
|
compression=compression,
|
|
aio_buffer_size=aio_buffer_size,
|
|
aio_buffer_count=aio_buffer_count,
|
|
)
|
|
_aio_prepare(disk)
|
|
path_b = disk_path.encode("utf-8")
|
|
open_flags = (
|
|
NFC_OPEN_FLAGS_READ_ONLY if read_only else NFC_OPEN_FLAGS_READ_WRITE
|
|
)
|
|
open_body = struct.pack("<IIIIII", len(path_b), 0, 0, 0, NFC_DISK, open_flags)
|
|
open_body = open_body.ljust(60, b"\x00")
|
|
reply = disk._aio_roundtrip(NFC_AIO_MSG_OPEN_FILE, open_body, extra=path_b)
|
|
handle, sector_size = _parse_open_reply(reply)
|
|
disk.handle = handle
|
|
disk.sector_size = sector_size
|
|
return disk
|
|
except Exception:
|
|
sock.close()
|
|
raise
|