From 336ee1e4ca0d1d52643f396becc6aabbd2c699ea Mon Sep 17 00:00:00 2001 From: doccaz Date: Sat, 19 Sep 2026 10:37:20 -0300 Subject: [PATCH 1/3] Add VixDiskLib_GetInfo (capacity, geometry, DDB_GET fields) Capacity and physical geometry come free from the OPEN_FILE reply: found via an SSL-hook capture of VixDiskLib_GetInfo that dumped every byte of the reply rather than just the fields an earlier Open-only capture had labeled. Offset 28 (uint64, bytes) matches VixDiskLibInfo.capacity; offsets 40/44/48 match physGeo exactly. No extra NFC round trip needed for these two fields. biosGeo, adapterType, and uuid come from DDB_GET (a generic VMDK descriptor key/value NFC message, implemented here too): request is a 16-byte fixed payload plus the key name as a raw ASCII extra; reply is 16 bytes plus a value extra that is ASCII text on the wire (not binary) -- geometry.cylinders comes back as the literal bytes b"2088", matching how a VMDK descriptor's DDB section stores key/value pairs as plain text. VixDiskLibHandle.get_info() now issues the same 5 DDB_GET round trips real VDDK's VixDiskLib_GetInfo pays on every call, matching its behavior and cost exactly (previously it only returned the two free OPEN_FILE-derived fields). Adds unit tests for the OPEN_FILE reply parsing, the DDB_GET wire format, and query_full_info's field combination/fallback logic -- no lab needed. Validated against a live standalone ESXi 8.0.3 host: output matches native VDDK's own GetInfo on the same disk exactly (adapterType=3 <-> "lsilogic", same uuid string, same zeroed biosGeo). Full protocol details in docs/nfc_open.md. --- README.md | 7 +- docs/nfc_open.md | 88 ++++++++++-- docs/nfc_read.md | 1 - docs/reverse_engineering_procedure.md | 62 +++++++- openvixdisklib/nfc_open.py | 122 +++++++++++++++- openvixdisklib/openvixdisklib.py | 13 ++ tests/integration/test_nfc_open.py | 17 ++- tests/integration/test_openvixdisklib.py | 26 ++++ tests/unit/test_nfc_open.py | 173 +++++++++++++++++++++++ 9 files changed, 487 insertions(+), 22 deletions(-) create mode 100644 tests/unit/test_nfc_open.py diff --git a/README.md b/README.md index 254794b..565ffe6 100644 --- a/README.md +++ b/README.md @@ -24,10 +24,13 @@ Implemented against vCenter 8 / ESXi 8. Default transport is `nbdssl` - `VixDiskLib_Open` (datastore path, read-only or read-write) - `VixDiskLib_Read` (optional ``skip_decompression`` packs FastLZ extras) - `VixDiskLib_Write` +- `VixDiskLib_GetInfo` (capacity and physical geometry from the `Open` + reply; `biosGeo`/`adapterType`/`uuid` from `DDB_GET`, matching real + VDDK's cost and behavior) Not implemented: compression open flags other than FastLZ, CBT / -allocated-block queries, disk geometry (`DDB_GET`), encrypted disks, -and direct ESXi `ha-nfc` without vCenter `vpxa-nfc`. +allocated-block queries, encrypted disks, and direct ESXi `ha-nfc` +without vCenter `vpxa-nfc`. Requires Python 3.10 or later. diff --git a/docs/nfc_open.md b/docs/nfc_open.md index bf61550..96bf647 100644 --- a/docs/nfc_open.md +++ b/docs/nfc_open.md @@ -143,7 +143,7 @@ AIO types used for Open / Read / Close, correlated with the consecutive | 9 | `SET_SOCK_OPTS` | 12 | | | 22 | `SET_RES_POOL` | 4 | | | 4 | `OPEN_FILE` | 60 | path string | -| 11 | `DDB_GET` | 16 | key name (VDDK only) | +| 11 | `DDB_GET` | 16 | key name | | 7 | `IO` | 44 | sector bytes (read reply / write request) | | 5 | `CLOSE_FILE` | 8 | | | 3 | `CLOSE_SESSION` | 4 | | @@ -156,6 +156,65 @@ VDDK Open also issues several `DDB_GET` queries (`resumeConsolidateSector`, (16 zero bytes) on this unencrypted disk. They are not required to obtain a file handle or to read sector 0. +`VixDiskLib_GetInfo` (Step 14) triggers ~20 more `DDB_GET` calls right +after `OPEN_FILE`, for these keys (captured in request-order, key name +is the extra string after the 16-byte payload, no `ddb.` prefix on the +wire): `resumeConsolidateSector`, `isDigest` (×3), `iofilters` (×2), +`logicalSectorSize` (×2), `physicalSectorSize` (×2), +`isNativeLinkedClone` (×2), `KMFilters`, `sidecars`, `adapterType`, +`uuid`, `geometry.cylinders`, `geometry.heads`, `geometry.sectors`, +`geometry.biosCylinders`, `geometry.biosHeads`, `geometry.biosSectors`. +On this lab disk, `biosGeo` came back all zeros (key not found) and +`logicalSectorSize`/`physicalSectorSize`/the non-bios `geometry.*` keys +duplicate what OPEN_FILE already returned — only `adapterType` and +`uuid` are genuinely new information from this burst. + +### `DDB_GET` request/reply layout + +Decoded from the same capture (request/reply pairs matched by `opId` +across all ~28 calls seen in one `GetInfo`). + +Request: 16-byte fixed payload plus the key name as a raw ASCII extra +(no NUL terminator, not counted in `size` — same convention as +`OPEN_FILE`'s path): + +| Offset | Type | Meaning | +| ------ | -------- | --------------------------------------- | +| 0 | `uint64` | File handle (same value as `OPEN_FILE`) | +| 8 | `uint32` | Key name length in bytes | +| 12 | `uint32` | 0 | +| 16 | — | Key name (ASCII, no `ddb.` prefix) | + +Reply: 16 bytes plus a value extra, **not** padded (unlike +`QueryAllocatedBlocks`'s bitmap — verified by decoding all 28 replies +in sequence with no desync): + +| Offset | Type | Meaning | +| ------ | -------- | --------------------------------------- | +| 0-11 | — | Zero/unused in every capture | +| 12 | `uint32` | Value length in bytes (`0` = not found) | +| 16 | — | Value (ASCII **text**, not binary) | + +Values are ASCII text even for keys that sound numeric — +`geometry.cylinders` comes back as the literal bytes `b"2088"`, not a +binary `uint32`. This matches how a VMDK descriptor file's DDB (disk +database) section stores keys as plain-text `ddb. = ""` +lines; `adapterType` comes back as `b"lsilogic"` (a string), not +VDDK's numeric `VIXDISKLIB_ADAPTER_SCSI_LSILOGIC` enum value — VDDK's +own client does that string-to-enum mapping internally, which +OpenVixDiskLib does not reproduce (`DiskInfo.adapter_type` is the raw +DDB string). + +Implemented as `openvixdisklib.nfc_open.NfcDisk.ddb_get(key) -> str | +None` and `NfcDisk.query_full_info() -> DiskInfo` (5 round trips: +`geometry.biosCylinders`/`biosHeads`/`biosSectors`, `adapterType`, +`uuid`), wired into `VixDiskLibHandle.get_info`, which now matches +real VDDK's `VixDiskLib_GetInfo` exactly — capacity/physGeo free from +`OPEN_FILE`, the rest costing the same 5 round trips VDDK itself pays. +Validated against the live ESXi lab: matches native VDDK's `GetInfo` +output on the same disk (`adapterType=3` ↔ `"lsilogic"`, same `uuid` +string, same zeroed `biosGeo`). + ### OPEN_SESSION / sockopts / resource pool `OPEN_SESSION` payload is 16 bytes, little-endian: @@ -202,12 +261,22 @@ Reply payload (60 bytes), fields that matter: | Offset | Type | Meaning | | ------ | -------- | ------------------------------- | -| 8 | `uint64` | File handle (opaque, per open) | -| 16 | `uint32` | File type (`2` = `NFC_DISK`) | -| 20 | `uint32` | Flags echoed (`0x1e` or `0x1a`) | -| 36 | `uint32` | Sector size (`512` on this VM) | - -Later AIO messages pass that handle as a `uint64`. +| 8 | `uint64` | File handle (opaque, per open) | +| 16 | `uint32` | File type (`2` = `NFC_DISK`) | +| 20 | `uint32` | Flags echoed (`0x1e` or `0x1a`) | +| 28 | `uint64` | Disk capacity in **bytes** | +| 36 | `uint32` | Sector size (`512` on this VM) | +| 40 | `uint32` | Physical geometry cylinders | +| 44 | `uint32` | Physical geometry heads | +| 48 | `uint32` | Physical geometry sectors | + +Later AIO messages pass that handle as a `uint64`. Offset 28 was found +by capturing `VixDiskLib_GetInfo` (Step 14, +`docs/reverse_engineering_procedure.md`): it matches +`VixDiskLibInfo.capacity` converted to bytes, and offsets 40/44/48 +match `VixDiskLibInfo.physGeo` exactly — both already arrive with this +reply, no separate `GetInfo` wire call exists. `biosGeo`, `adapterType`, +and `uuid` are **not** here; VDDK gets those from `DDB_GET` (below). ### IO (read / write) @@ -232,6 +301,8 @@ classic type 4 `NFC_SESSION_COMPLETE`. | Handshake + AIO + OPEN_FILE | `openvixdisklib.nfc_open.open_disk` | | AIO extra size / pool count | `open_disk(..., aio_buffer_size=, aio_buffer_count=)` | | Sector read / write / close | `openvixdisklib.nfc_open.NfcDisk` | +| Full disk info (`GetInfo`) | `openvixdisklib.openvixdisklib.VixDiskLibHandle.get_info` | +| VMDK descriptor DDB lookup | `openvixdisklib.nfc_open.NfcDisk.ddb_get` | Run: @@ -246,7 +317,8 @@ I/O: `docs/nfc_read.md`, `docs/nfc_write.md`, and ## What is still VDDK-only -- `DDB_GET` / geometry / zlib and skipz compression / encryption keys +- zlib and skipz compression / encryption keys (`DDB_GET` is + implemented for the plain, non-encrypted keys covered above) - `NFC_DELTA_DISK`, change-block tracking - Host-switch (`NFC_AIO_SWITCH_HOST_*`) - Direct ESXi `ha-nfc` without vCenter `vpxa-nfc` diff --git a/docs/nfc_read.md b/docs/nfc_read.md index ee67652..5c9c2a7 100644 --- a/docs/nfc_read.md +++ b/docs/nfc_read.md @@ -194,4 +194,3 @@ buf when skip_decompression=True: extras packed densely from offset 0 - zlib and skipz NBD compression flags - `VixDiskLib_ReadAsync` (same IO messages, different client threading) - `VixDiskLib_QueryAllocatedBlocks` / allocation bitmaps -- `VixDiskLib_GetInfo` capacity (not required to read a known range) diff --git a/docs/reverse_engineering_procedure.md b/docs/reverse_engineering_procedure.md index b988f5c..268192d 100644 --- a/docs/reverse_engineering_procedure.md +++ b/docs/reverse_engineering_procedure.md @@ -8,8 +8,9 @@ This file is the **sequence of steps**, including dead ends, so later NFC work can follow the same loop instead of rediscovering it. Scope so far: `VixDiskLib_ConnectEx` + `VixDiskLib_Open` + -`VixDiskLib_Read` + `VixDiskLib_Write` against lab vCenter 8.0.1 / -ESXi 8, transports `nbd` and `nbdssl`. Validation method: +`VixDiskLib_Read` + `VixDiskLib_Write` + `VixDiskLib_GetInfo` against +lab vCenter 8.0.1 / ESXi 8, transports `nbd` and `nbdssl`. Validation +method: `tests/integration/` (the session-scoped `lab` fixture creates a temporary empty VM with a 10 GiB disk and destroys it when the pytest session ends). @@ -381,8 +382,58 @@ not an OPEN_FILE bit. Capture VDDK with that flag (NBD + the port-902 Replay: pip `pyfastlz` via `openvixdisklib/fastlz.py` (NFC extra is raw FastLZ, without the wrapper's 4-byte length prefix) plus `NfcDisk` compression on each IO. Proof: -`tests/integration/test_nfc_read_write.py` (`fastlz`) and -`tests/perf/test_compare.py`. +## Step 14 — `VixDiskLib_GetInfo` capacity + +Extended the ctypes probe from Step 13 to call `VixDiskLib_GetInfo` +after `Open`, under the SSL hook plus a `write`/`read` interceptor on +fd 902 (Step 7), to see what wire traffic `GetInfo` adds. + +Result: **no new SOAP or authd traffic** — the same `RetrieveContent` ++ `Login` + `NfcGetVmFiles` + authd sequence as a plain `Open`. All the +extra traffic is inside the already-open NFC/AIO session: ~20 more +`DDB_GET` (type 11) requests right after `OPEN_FILE`, for keys like +`adapterType`, `uuid`, `geometry.cylinders`, `geometry.biosCylinders`, +etc. (full list in `docs/nfc_open.md`). + +Dumping every byte of the `OPEN_FILE` reply (not just the fields the +earlier Open-only capture had labeled) found `capacity` (offset 28, +`uint64` bytes) and `physGeo` (offsets 40/44/48) already present — +verified they match `VixDiskLibInfo.capacity`/`physGeo` from the same +`GetInfo` call exactly. Only `biosGeo`, `adapterType`, and `uuid` are +genuinely `DDB_GET`-only; `biosGeo` came back "key not found" (zeros) +on this unencrypted lab disk. + +Fix: extended `_parse_open_reply` in `openvixdisklib/nfc_open.py` to +also read those offsets, added `nfc_open.DiskInfo`/`DiskGeometry`, and +exposed `VixDiskLibHandle.get_info()`. No new NFC message type was +needed — `DDB_GET` (`adapterType`/`uuid`/`biosGeo`) is still open work. +Validated against the live host: `capacity_sectors=33554432` +(16 GiB), `phys_geo=(2088, 255, 63)`, matching native VDDK's +`GetInfo` on the same disk. + + +## Step 16 — `DDB_GET` (AIO type 11) + +Already partly captured as a side effect of Step 14 (`VixDiskLib_GetInfo` +triggers ~28 `DDB_GET` calls); no new capture was needed, just decoding +the request/reply pairs from that saved log by matching `opId` across +both directions. Confirmed the request's first 8 bytes equal the +`OPEN_FILE` handle from the same capture, and that a "found" reply's +extra is plain ASCII text (`b"lsilogic"`, `b"2088"`, ...), not binary — +matching how a VMDK descriptor's DDB section stores key/value pairs as +text. No padding on the reply extra (unlike Step 15's bitmap), +confirmed by decoding all 28 request/reply pairs from one capture in +sequence without desync. + +Implemented as `NfcDisk.ddb_get(key) -> str | None` and +`NfcDisk.query_full_info() -> DiskInfo` (the 5 keys needed for +`bios_geo`/`adapter_type`/`uuid`), wired into +`VixDiskLibHandle.get_info` in place of the OPEN_FILE-only version +from Step 14 — `get_info` now matches real VDDK's `VixDiskLib_GetInfo` +completely, including paying the same round-trip cost. Full layout: +`docs/nfc_open.md`. Validated against the live ESXi lab: matches +native VDDK's `GetInfo` output on the same disk exactly. + ## What to write down @@ -404,8 +455,7 @@ OpenVixDiskLib. Not yet reversed, same loop as above: -- `DDB_GET` / disk geometry, zlib/skipz compression, encrypted disks +- zlib/skipz compression, encrypted disks - `NFC_DELTA_DISK`, CBT / `QueryAllocatedBlocks` -- `VixDiskLib_GetInfo` capacity - Host-switch AIO messages - Direct ESXi `ha-nfc` without vCenter `vpxa-nfc` diff --git a/openvixdisklib/nfc_open.py b/openvixdisklib/nfc_open.py index 1b25bd5..39f2a69 100644 --- a/openvixdisklib/nfc_open.py +++ b/openvixdisklib/nfc_open.py @@ -82,6 +82,37 @@ NFC_COMPRESSION_FASTLZ = 2 +@dataclass(frozen=True, slots=True) +class DiskGeometry: + """CHS geometry, matching VDDK's ``VixDiskLibGeometry``.""" + + cylinders: int + heads: int + sectors: int + + +@dataclass(frozen=True, slots=True) +class DiskInfo: + """Matches VDDK's ``VixDiskLibInfo``. + + ``phys_geo`` and ``capacity_sectors`` are read directly off + OPEN_FILE (offsets 40/44/48 and 28 respectively) — free, no extra + NFC round trip. ``bios_geo``, ``adapter_type``, and ``uuid`` come + from ``DDB_GET`` (see ``NfcDisk.ddb_get`` / ``query_full_info``, + ``docs/nfc_open.md``): each is a real round trip, matching what + real VDDK's ``VixDiskLib_GetInfo`` does. ``bios_geo`` defaults to + all zeros and ``adapter_type``/``uuid`` to ``None`` when the disk + has no snapshots or predates that DDB key (VDDK does the same for + a missing key). + """ + + capacity_sectors: int + phys_geo: DiskGeometry + bios_geo: DiskGeometry = DiskGeometry(cylinders=0, heads=0, sectors=0) + adapter_type: str | None = None + uuid: str | None = None + + @dataclass(frozen=True, slots=True) class ReadFragment: """One NFC AIO extra in a packed skip-decompression ``buf``. @@ -260,6 +291,7 @@ def __init__( compression: int = NFC_COMPRESSION_NONE, aio_buffer_size: int = NFC_AIO_BUFFER_SIZE, aio_buffer_count: int = NFC_AIO_BUFFER_COUNT, + info: DiskInfo | None = None, ) -> None: """Wrap an AIO session that already has ``path`` open. @@ -275,6 +307,8 @@ def __init__( at most this large. aio_buffer_count: OPEN_SESSION buffer pool count (default ``NFC_AIO_BUFFER_COUNT``). + info: Capacity/geometry from the OPEN_FILE reply. ``None`` + before the reply arrives. """ self._sock = sock self._op_id = 0 @@ -284,6 +318,7 @@ def __init__( self.compression = compression self.aio_buffer_size = aio_buffer_size self.aio_buffer_count = aio_buffer_count + self.info = info self._closed = False def _next_op_id(self) -> int: @@ -516,6 +551,78 @@ def write(self, start_sector: int, num_sectors: int, data: bytes) -> None: f"expected type={NFC_AIO_MSG_IO} opId={op_id}" ) + def ddb_get(self, key: str) -> str | None: + """Return a VMDK descriptor DDB value, or ``None`` if unset. + + Captured from VDDK: request is a 16-byte fixed payload plus the + key name as a raw ASCII extra (no NUL terminator, not counted + in ``size``, same convention as ``OPEN_FILE``'s path):: + + uint64 handle + uint32 key_name_length + uint32 reserved (0) + + + Reply is 16 bytes plus a value extra, **not** padded (unlike + ``QueryAllocatedBlocks``'s bitmap):: + + 96 bits reserved/unused (always zero in this lab) + uint32 value_length (0 = key not found) + + + Values are ASCII text even for keys that sound numeric + (``geometry.cylinders`` comes back as the bytes ``b"2088"``, + not a binary int) — this matches how a VMDK descriptor file's + DDB (disk database) section stores keys as plain text + ``ddb. = ""`` lines. See ``docs/nfc_open.md``. + + Args: + key: DDB key name without the ``ddb.`` prefix (for example + ``"adapterType"``, ``"uuid"``, ``"geometry.cylinders"``). + """ + key_bytes = key.encode("ascii") + request = struct.pack(" DiskInfo: + """Return a ``DiskInfo`` with ``bios_geo``/``adapter_type``/``uuid`` filled in. + + ``self.info`` (from OPEN_FILE) already has ``capacity_sectors`` + and ``phys_geo`` for free; this issues 5 ``DDB_GET`` round trips + for the rest, matching what real VDDK's ``VixDiskLib_GetInfo`` + does on every call. DDB values are ASCII text; geometry fields + are parsed as decimal integers, and any missing key falls back + to ``DiskInfo``'s defaults (matches VDDK: a disk with no + snapshots, or from before this DDB key existed, has none of + these set). + """ + assert self.info is not None + bios_cylinders = self.ddb_get("geometry.biosCylinders") + bios_heads = self.ddb_get("geometry.biosHeads") + bios_sectors = self.ddb_get("geometry.biosSectors") + bios_geo = DiskGeometry( + cylinders=int(bios_cylinders) if bios_cylinders else 0, + heads=int(bios_heads) if bios_heads else 0, + sectors=int(bios_sectors) if bios_sectors else 0, + ) + return DiskInfo( + capacity_sectors=self.info.capacity_sectors, + phys_geo=self.info.phys_geo, + bios_geo=bios_geo, + adapter_type=self.ddb_get("adapterType"), + uuid=self.ddb_get("uuid"), + ) + def close(self) -> None: """Close the VMDK, the AIO session, and the classic NFC session.""" if self._closed: @@ -591,16 +698,22 @@ def _aio_prepare(disk: NfcDisk) -> None: disk._aio_roundtrip(NFC_AIO_MSG_SET_RES_POOL, struct.pack(" tuple[int, int]: - if len(body) < 40: +def _parse_open_reply(body: bytes) -> tuple[int, int, DiskInfo]: + if len(body) < 52: raise NfcProtocolError(f"OPEN_FILE reply too short: {len(body)}") handle, file_type, _flags = struct.unpack_from(" nfc_open.DiskInfo: + """Return disk info. Matches ``VixDiskLib_GetInfo``. + + ``capacity_sectors``/``phys_geo`` are free (already in the + ``OPEN_FILE`` reply from ``open()``); ``bios_geo``/ + ``adapter_type``/``uuid`` cost 5 ``DDB_GET`` round trips, same + as real VDDK pays on every ``GetInfo`` call. See + ``docs/nfc_open.md``. + """ + return disk_handle.disk.query_full_info() + def read( self, disk_handle: _DiskHandle, diff --git a/tests/integration/test_nfc_open.py b/tests/integration/test_nfc_open.py index 9d9beb1..08fe813 100644 --- a/tests/integration/test_nfc_open.py +++ b/tests/integration/test_nfc_open.py @@ -6,7 +6,7 @@ import pytest from openvixdisklib import nfc_open -from tests.integration.base import SECTOR_SIZE, LabEnv, pattern_bytes +from tests.integration.base import _DISK_CAPACITY_KB, SECTOR_SIZE, LabEnv, pattern_bytes class TestNfcOpen: @@ -34,3 +34,18 @@ def test_open_disk_and_read_first_sector( got = disk.read(0, 1) assert got is not expected assert got == expected + + def test_open_disk_reports_capacity_and_geometry(self, lab: LabEnv) -> None: + """OPEN_FILE's reply carries capacity and physical geometry (GetInfo).""" + with ( + lab.authenticate() as session, + nfc_open.open_disk(session, lab.disk_path) as disk, + ): + assert disk.info is not None + assert disk.info.capacity_sectors == ( + _DISK_CAPACITY_KB * 1024 // SECTOR_SIZE + ) + geo = disk.info.phys_geo + assert geo.cylinders > 0 + assert geo.heads > 0 + assert geo.sectors > 0 diff --git a/tests/integration/test_openvixdisklib.py b/tests/integration/test_openvixdisklib.py index 6c853d6..5e1ac42 100644 --- a/tests/integration/test_openvixdisklib.py +++ b/tests/integration/test_openvixdisklib.py @@ -13,6 +13,7 @@ from openvixdisklib import openvixdisklib as vixdisklib from openvixdisklib.openvixdisklib import ReadResult from tests.integration.base import ( + _DISK_CAPACITY_KB, SECTOR_AT_1GB, SECTOR_SIZE, LabEnv, @@ -94,6 +95,31 @@ def test_write_and_read_sector_zero_and_one_gib( handle.read(disk, start, 1, read_buf) assert read_buf.raw[:SECTOR_SIZE] == expected + def test_get_info(self, lab: LabEnv) -> None: + """get_info returns the lab VM's known disk capacity and geometry.""" + handle = vixdisklib.VixDiskLibHandle(vixdisklib_compatibility_version="8.0") + connect_kwargs = lab.vixdisklib_connect_kwargs( + {"allow_untrusted": lab.allow_untrusted, "read_only": True} + ) + with ( + handle.connect(**connect_kwargs) as conn, + handle.open( + conn, lab.disk_path, flags=vixdisklib.VIXDISKLIB_FLAG_OPEN_READ_ONLY + ) as disk, + ): + info = handle.get_info(disk) + assert info.capacity_sectors == _DISK_CAPACITY_KB * 1024 // SECTOR_SIZE + assert info.phys_geo.cylinders > 0 + assert info.phys_geo.heads > 0 + assert info.phys_geo.sectors > 0 + # bios_geo is DDB-derived and unset (all zero) on a disk with + # no snapshots yet, matching VDDK's own default for a missing key. + assert info.bios_geo == vixdisklib.DiskGeometry( + cylinders=0, heads=0, sectors=0 + ) + assert info.adapter_type # non-empty DDB string, e.g. "lsilogic" + assert info.uuid # non-empty DDB string + def test_read_only_open_snapshot_parent(self, lab: LabEnv) -> None: """Read-only Open uses NfcGetVmFiles, including a snapshot parent path. diff --git a/tests/unit/test_nfc_open.py b/tests/unit/test_nfc_open.py new file mode 100644 index 0000000..59d6708 --- /dev/null +++ b/tests/unit/test_nfc_open.py @@ -0,0 +1,173 @@ +# Copyright 2026 Cloudbase Solutions Srl +# All Rights Reserved. + +"""Unit tests for the OPEN_FILE reply parsing in ``nfc_open``.""" + +import struct + +import pytest + +from openvixdisklib import nfc_open + + +class _FakeSocket: + """A minimal socket stand-in that replays scripted bytes for recv_into.""" + + def __init__(self, replies: bytes) -> None: + self._replies = replies + self.sent: list[bytes] = [] + + def sendall(self, data: bytes) -> None: + self.sent.append(bytes(data)) + + def recv_into(self, buf: memoryview) -> int: + n = min(len(buf), len(self._replies)) + buf[:n] = self._replies[:n] + self._replies = self._replies[n:] + return n + + + + +def _open_reply_body( + handle: int = 0x1234, + file_type: int = nfc_open.NFC_DISK, + flags: int = nfc_open.NFC_OPEN_FLAGS_READ_ONLY, + capacity_bytes: int = 17179869184, + sector_size: int = 512, + cylinders: int = 2088, + heads: int = 255, + sectors: int = 63, +) -> bytes: + """Build a synthetic 60-byte OPEN_FILE reply payload.""" + body = bytearray(60) + struct.pack_into(" bytes: + """Build a scripted DDB_GET reply: header + 16-byte body + value extra.""" + value_length = len(value) if value is not None else 0 + body = bytes(12) + struct.pack(" nfc_open.NfcDisk: + return nfc_open.NfcDisk( + sock=_FakeSocket(replies), path="[ds] a.vmdk", handle=0x1234, sector_size=512 + ) + + def test_found_key_returns_decoded_value(self) -> None: + disk = self._disk(_ddb_get_reply(op_id=0, value=b"lsilogic")) + assert disk.ddb_get("adapterType") == "lsilogic" + + def test_missing_key_returns_none(self) -> None: + disk = self._disk(_ddb_get_reply(op_id=0, value=None)) + assert disk.ddb_get("resumeConsolidateSector") is None + + def test_sends_handle_and_key_length_in_request(self) -> None: + sock = _FakeSocket(_ddb_get_reply(op_id=0, value=b"63")) + disk = nfc_open.NfcDisk( + sock=sock, path="[ds] a.vmdk", handle=0x1234, sector_size=512 + ) + disk.ddb_get("geometry.sectors") + (sent,) = sock.sent + # header(16) + handle(8) + key_len(4) + reserved(4) + key bytes + handle, key_len, reserved = struct.unpack_from(" nfc_open.NfcDisk: + # query_full_info calls ddb_get for biosCylinders, biosHeads, + # biosSectors, adapterType, uuid, in that order. + keys = [ + "geometry.biosCylinders", + "geometry.biosHeads", + "geometry.biosSectors", + "adapterType", + "uuid", + ] + replies = b"".join( + _ddb_get_reply(op_id=i, value=values.get(k)) for i, k in enumerate(keys) + ) + disk = nfc_open.NfcDisk( + sock=_FakeSocket(replies), path="[ds] a.vmdk", handle=1, sector_size=512 + ) + disk.info = nfc_open.DiskInfo( + capacity_sectors=1024, + phys_geo=nfc_open.DiskGeometry(cylinders=10, heads=20, sectors=30), + ) + return disk + + def test_combines_open_file_info_with_ddb_values(self) -> None: + disk = self._disk_with_replies( + { + "geometry.biosCylinders": b"100", + "geometry.biosHeads": b"200", + "geometry.biosSectors": b"63", + "adapterType": b"lsilogic", + "uuid": b"some-uuid", + } + ) + info = disk.query_full_info() + assert info.capacity_sectors == 1024 + assert info.phys_geo == nfc_open.DiskGeometry(cylinders=10, heads=20, sectors=30) + assert info.bios_geo == nfc_open.DiskGeometry(cylinders=100, heads=200, sectors=63) + assert info.adapter_type == "lsilogic" + assert info.uuid == "some-uuid" + + def test_missing_ddb_keys_fall_back_to_defaults(self) -> None: + disk = self._disk_with_replies({}) + info = disk.query_full_info() + assert info.bios_geo == nfc_open.DiskGeometry(cylinders=0, heads=0, sectors=0) + assert info.adapter_type is None + assert info.uuid is None + + + + +class TestParseOpenReply: + def test_parses_handle_capacity_and_geometry(self) -> None: + """Capacity (offset 28, bytes) and physGeo (40/44/48) are extracted.""" + body = _open_reply_body() + handle, sector_size, info = nfc_open._parse_open_reply(body) + assert handle == 0x1234 + assert sector_size == 512 + assert info.capacity_sectors == 17179869184 // 512 + assert info.phys_geo == nfc_open.DiskGeometry( + cylinders=2088, heads=255, sectors=63 + ) + + def test_zero_sector_size_falls_back_and_still_divides_capacity(self) -> None: + """A zero sector_size falls back to NFC_SECTOR_SIZE for both uses.""" + body = _open_reply_body(sector_size=0, capacity_bytes=1024 * 512) + _handle, sector_size, info = nfc_open._parse_open_reply(body) + assert sector_size == nfc_open.NFC_SECTOR_SIZE + assert info.capacity_sectors == (1024 * 512) // nfc_open.NFC_SECTOR_SIZE + + def test_wrong_file_type_raises(self) -> None: + """A non-NFC_DISK file type is rejected.""" + body = _open_reply_body(file_type=99) + with pytest.raises(nfc_open.NfcProtocolError, match="file type 99"): + nfc_open._parse_open_reply(body) + + def test_short_body_raises(self) -> None: + """A reply shorter than the physGeo fields is rejected.""" + with pytest.raises(nfc_open.NfcProtocolError, match="too short"): + nfc_open._parse_open_reply(bytes(40)) From f5368ab984165eec9923e98974bf7bcc5437fd8d Mon Sep 17 00:00:00 2001 From: Lucian Petrut Date: Wed, 23 Sep 2026 10:58:43 +0000 Subject: [PATCH 2/3] Fix linter errors --- tests/unit/test_nfc_open.py | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/tests/unit/test_nfc_open.py b/tests/unit/test_nfc_open.py index 59d6708..1558b5c 100644 --- a/tests/unit/test_nfc_open.py +++ b/tests/unit/test_nfc_open.py @@ -27,8 +27,6 @@ def recv_into(self, buf: memoryview) -> int: return n - - def _open_reply_body( handle: int = 0x1234, file_type: int = nfc_open.NFC_DISK, @@ -48,23 +46,24 @@ def _open_reply_body( return bytes(body) - - def _ddb_get_reply(op_id: int, value: bytes | None) -> bytes: """Build a scripted DDB_GET reply: header + 16-byte body + value extra.""" value_length = len(value) if value is not None else 0 body = bytes(12) + struct.pack(" nfc_open.NfcDisk: return nfc_open.NfcDisk( - sock=_FakeSocket(replies), path="[ds] a.vmdk", handle=0x1234, sector_size=512 + sock=_FakeSocket(replies), + path="[ds] a.vmdk", + handle=0x1234, + sector_size=512, ) def test_found_key_returns_decoded_value(self) -> None: @@ -90,8 +89,6 @@ def test_sends_handle_and_key_length_in_request(self) -> None: assert sent[16 + 16 :] == b"geometry.sectors" - - class TestQueryFullInfo: def _disk_with_replies(self, values: dict[str, bytes | None]) -> nfc_open.NfcDisk: # query_full_info calls ddb_get for biosCylinders, biosHeads, @@ -127,8 +124,12 @@ def test_combines_open_file_info_with_ddb_values(self) -> None: ) info = disk.query_full_info() assert info.capacity_sectors == 1024 - assert info.phys_geo == nfc_open.DiskGeometry(cylinders=10, heads=20, sectors=30) - assert info.bios_geo == nfc_open.DiskGeometry(cylinders=100, heads=200, sectors=63) + assert info.phys_geo == nfc_open.DiskGeometry( + cylinders=10, heads=20, sectors=30 + ) + assert info.bios_geo == nfc_open.DiskGeometry( + cylinders=100, heads=200, sectors=63 + ) assert info.adapter_type == "lsilogic" assert info.uuid == "some-uuid" @@ -140,8 +141,6 @@ def test_missing_ddb_keys_fall_back_to_defaults(self) -> None: assert info.uuid is None - - class TestParseOpenReply: def test_parses_handle_capacity_and_geometry(self) -> None: """Capacity (offset 28, bytes) and physGeo (40/44/48) are extracted.""" From 6f46d0229686e5613146788f9807c113d8ba8b3d Mon Sep 17 00:00:00 2001 From: Lucian Petrut Date: Wed, 23 Sep 2026 11:37:39 +0000 Subject: [PATCH 3/3] Fix mypy errors We're currently hitting the following mypy error: ``` tests/unit/test_nfc_open.py:107: error: Argument "sock" to "NfcDisk" has incompatible type "_FakeSocket"; expected "socket" [arg-type] ``` `typing.Protocol` is a convenient way of addressing this. https://typing.python.org/en/latest/spec/protocol.html --- openvixdisklib/nfc_open.py | 22 ++++++++++++++++++---- tests/unit/test_nfc_open.py | 10 +++++++--- 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/openvixdisklib/nfc_open.py b/openvixdisklib/nfc_open.py index 39f2a69..575ede8 100644 --- a/openvixdisklib/nfc_open.py +++ b/openvixdisklib/nfc_open.py @@ -26,6 +26,7 @@ import ssl import struct from dataclasses import dataclass +from typing import Protocol from openvixdisklib import fastlz from openvixdisklib.nfc_auth import NfcAuthSession, _ssl_client_context @@ -201,18 +202,31 @@ def wrap_nfcssl_socket(ssock: ssl.SSLSocket, server_hostname: str) -> ssl.SSLSoc raise +class NfcTransport(Protocol): + """Byte pipe used after the NFC handshake (TCP, TLS, or a test fake).""" + + def sendall(self, data: bytes) -> None: + """Send ``data`` in full.""" + + def recv_into(self, buffer: memoryview, nbytes: int = 0, flags: int = 0) -> int: + """Read into ``buffer`` and return the number of bytes stored.""" + + def close(self) -> None: + """Close the underlying connection.""" + + 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: +def _recvn(sock: NfcTransport, size: int) -> bytes: buf = bytearray(size) _recvn_into(sock, memoryview(buf)) return bytes(buf) -def _recvn_into(sock: socket.socket, buf: memoryview) -> None: +def _recvn_into(sock: NfcTransport, buf: memoryview) -> None: """Read exactly ``len(buf)`` bytes into ``buf``.""" view = buf.cast("B") if buf.format != "B" else buf filled = 0 @@ -251,7 +265,7 @@ def _aio_extra_len(ctype: int, body: bytes, chunk_len: int) -> int: raise NfcProtocolError(f"unsupported NFC IO compression type {ctype}") -def _send_nfc_msg(sock: socket.socket, msg_type: int, body: bytes = b"") -> None: +def _send_nfc_msg(sock: NfcTransport, 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(" None: def sendall(self, data: bytes) -> None: self.sent.append(bytes(data)) - def recv_into(self, buf: memoryview) -> int: - n = min(len(buf), len(self._replies)) - buf[:n] = self._replies[:n] + def recv_into(self, buffer: memoryview, nbytes: int = 0, flags: int = 0) -> int: + del nbytes, flags + n = min(len(buffer), len(self._replies)) + buffer[:n] = self._replies[:n] self._replies = self._replies[n:] return n + def close(self) -> None: + pass + def _open_reply_body( handle: int = 0x1234,