diff --git a/4-Infrastructure/shim/ivshmem_client.py b/4-Infrastructure/shim/ivshmem_client.py new file mode 100644 index 00000000..97b49183 --- /dev/null +++ b/4-Infrastructure/shim/ivshmem_client.py @@ -0,0 +1,691 @@ +#!/usr/bin/env python3 +""" +Ivshmem Client — Zero-copy VM-to-VM communication via shared memory. + +Implements the ivshmem (Inter-VM Shared Memory) protocol for QEMU: + 1. IvshmemClient — mmap /dev/shm/ivshmem_bar0 for shared memory access + 2. IvshmemRing — doorbell ring for peer notification + 3. IvshmemTransform — encode data as shared memory write + notify peer + 4. IvshmemReceipt — machine-readable receipt for shared memory operations + +This is NOT a full ivshmem driver. It is a computation substrate that uses +QEMU's ivshmem device as a zero-copy shared memory mailbox. The peer VM +reads data from the same physical memory region without copying. + +Shim boundary: I/O only. No decision logic. All computation routing +decisions belong in Lean (future MeshRouting or SharedMemory modules). + +Reference: QEMU docs/interop/ivshmem-spec.txt, Linux ivshmem-client library + +Usage: + python3 ivshmem_client.py --example + python3 ivshmem_client.py --analyze + python3 ivshmem_client.py write [--offset ] [--slot ] + python3 ivshmem_client.py read --offset --length + python3 ivshmem_client.py ring --doorbell + python3 ivshmem_client.py receipt + python3 ivshmem_client.py --test-devshm +""" + +from __future__ import annotations + +import hashlib +import json +import mmap +import os +import struct +import sys +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +# ── Ivshmem Constants ────────────────────────────────────────────────────── + +# Default BAR0 size (64 MiB shared memory region) +IVSHMEM_BAR0_DEFAULT_SIZE: int = 64 * 1024 * 1024 + +# Doorbell register offset within BAR0 (QEMU ivshmem-spec) +IVSHMEM_BAR0_DOORBELL_OFFSET: int = 0x00 + +# Maximum payload size per write (1 MiB, to avoid single-write bottlenecks) +IVSHMEM_MAX_PAYLOAD_SIZE: int = 1 * 1024 * 1024 + +# Slot header size: 8 bytes (uint32 length + uint32 sequence) +IVSHMEM_SLOT_HEADER_SIZE: int = 8 + +# Default slot count in shared memory region +IVSHMEM_DEFAULT_SLOT_COUNT: int = 64 + +# Slot size: payload region per slot (1 MiB) +IVSHMEM_SLOT_SIZE: int = IVSHMEM_MAX_PAYLOAD_SIZE + +# Total region layout: +# [0x0000_0000 - 0x0000_FFFF] BAR0 registers (doorbell, interrupt mask) +# [0x0001_0000 - 0x0001_FFFF] Slot metadata (sequence numbers, flags) +# [0x0002_0000 - ...] Slot data regions +IVSHMEM_REGISTER_REGION_SIZE: int = 0x10000 +IVSHMEM_METADATA_REGION_OFFSET: int = 0x10000 +IVSHMEM_DATA_REGION_OFFSET: int = 0x20000 + +# ── Data Classes ─────────────────────────────────────────────────────────── + + +@dataclass +class IvshmemSlotHeader: + """Header for a single shared memory slot (8 bytes).""" + length: int = 0 # uint32: payload length in bytes + sequence: int = 0 # uint32: monotonically increasing sequence number + + def pack(self) -> bytes: + return struct.pack("!II", self.length, self.sequence) + + @staticmethod + def unpack(data: bytes) -> "IvshmemSlotHeader": + length, sequence = struct.unpack("!II", data[:IVSHMEM_SLOT_HEADER_SIZE]) + return IvshmemSlotHeader(length=length, sequence=sequence) + + def sizeof(self) -> int: + return IVSHMEM_SLOT_HEADER_SIZE + + +@dataclass +class IvshmemMetadata: + """Metadata for one slot in the metadata region (16 bytes).""" + sequence: int = 0 # uint32: last written sequence + length: int = 0 # uint32: last written length + flags: int = 0 # uint32: status flags (bit 0 = valid, bit 1 = read) + timestamp_lo: int = 0 # uint32: low 32 bits of write timestamp + + FLAG_VALID: int = 1 + FLAG_READ: int = 2 + + def pack(self) -> bytes: + return struct.pack("!IIII", self.sequence, self.length, self.flags, self.timestamp_lo) + + @staticmethod + def unpack(data: bytes) -> "IvshmemMetadata": + seq, length, flags, ts = struct.unpack("!IIII", data[:16]) + return IvshmemMetadata(sequence=seq, length=length, flags=flags, timestamp_lo=ts) + + def sizeof(self) -> int: + return 16 + + +@dataclass +class IvshmemState: + """State snapshot of the shared memory region. Used for receipts.""" + bar0_size: int = IVSHMEM_BAR0_DEFAULT_SIZE + slot_count: int = IVSHMEM_DEFAULT_SLOT_COUNT + slot_size: int = IVSHMEM_SLOT_SIZE + active_slot: int = 0 + sequence_counter: int = 0 + bytes_written: int = 0 + doorbell_rang: bool = False + + +@dataclass +class IvshmemReceipt: + """Machine-readable receipt for a shared memory operation.""" + schema: str = "ivshmem_receipt_v1" + transform_type: str = "shared_memory" + offset: int = 0 + length: int = 0 + slot: int = 0 + sequence: int = 0 + doorbell: bool = False + witness_hash: str = "" + bar0_size: int = 0 + slot_count: int = 0 + payload_crc32: str = "" + peer_notified: bool = False + + def to_dict(self) -> Dict[str, Any]: + return { + "schema": self.schema, + "transform_type": self.transform_type, + "offset": self.offset, + "length": self.length, + "slot": self.slot, + "sequence": self.sequence, + "doorbell": self.doorbell, + "witness_hash": self.witness_hash, + "bar0_size": self.bar0_size, + "slot_count": self.slot_count, + "payload_crc32": self.payload_crc32, + "peer_notified": self.peer_notified, + } + + +# ── IvshmemClient ────────────────────────────────────────────────────────── + + +class IvshmemClient: + """Client for QEMU ivshmem shared memory BAR0. + + Maps /dev/shm/ivshmem_bar0 (or a fallback file) into process memory. + Provides read/write access to the shared memory region without copying + data through kernel buffers — true zero-copy between VMs on the same host. + + If the ivshmem device is not available, falls back to a memory-mapped + file in /dev/shm for testing. + """ + + def __init__( + self, + device_path: str = "/dev/shm/ivshmem_bar0", + bar0_size: int = IVSHMEM_BAR0_DEFAULT_SIZE, + slot_count: int = IVSHMEM_DEFAULT_SLOT_COUNT, + fallback_to_devshm: bool = True, + ): + self.device_path = device_path + self.bar0_size = bar0_size + self.slot_count = slot_count + self.state = IvshmemState( + bar0_size=bar0_size, + slot_count=slot_count, + slot_size=IVSHMEM_SLOT_SIZE, + ) + self._mm: Optional[mmap.mmap] = None + self._fd: Optional[int] = None + self._fallback_path: Optional[str] = None + self._is_fallback = False + + def open(self) -> None: + """Open and memory-map the ivshmem BAR0 region.""" + if os.path.exists(self.device_path): + self._fd = os.open(self.device_path, os.O_RDWR | os.O_CREAT) + self._mm = mmap.mmap(self._fd, self.bar0_size) + self._is_fallback = False + elif self.device_path.startswith("/dev/shm/"): + # Fallback: create a shared memory file for testing + if not self.device_path.startswith("/dev/shm/"): + raise FileNotFoundError( + f"ivshmem device not found: {self.device_path}" + ) + self._fallback_path = self.device_path + self._fd = os.open( + self._fallback_path, os.O_RDWR | os.O_CREAT, 0o666 + ) + os.ftruncate(self._fd, self.bar0_size) + self._mm = mmap.mmap(self._fd, self.bar0_size) + self._is_fallback = True + else: + raise FileNotFoundError( + f"ivshmem device not found: {self.device_path}" + ) + + def close(self) -> None: + """Unmap and close the shared memory region.""" + if self._mm is not None: + self._mm.close() + self._mm = None + if self._fd is not None: + os.close(self._fd) + self._fd = None + + def __enter__(self) -> "IvshmemClient": + self.open() + return self + + def __exit__(self, *args: Any) -> None: + self.close() + + @property + def is_open(self) -> bool: + return self._mm is not None + + @property + def is_fallback(self) -> bool: + return self._is_fallback + + def read_region(self, offset: int, length: int) -> bytes: + """Read bytes from the shared memory region at offset.""" + if self._mm is None: + raise RuntimeError("IvshmemClient not opened") + if offset + length > self.bar0_size: + raise ValueError( + f"Read beyond BAR0: offset={offset} length={length} " + f"bar0_size={self.bar0_size}" + ) + self._mm.seek(offset) + return self._mm.read(length) + + def write_region(self, offset: int, data: bytes) -> None: + """Write bytes to the shared memory region at offset.""" + if self._mm is None: + raise RuntimeError("IvshmemClient not opened") + if offset + len(data) > self.bar0_size: + raise ValueError( + f"Write beyond BAR0: offset={offset} length={len(data)} " + f"bar0_size={self.bar0_size}" + ) + self._mm.seek(offset) + self._mm.write(data) + + def read_slot_header(self, slot: int) -> IvshmemSlotHeader: + """Read the header for a given slot.""" + offset = IVSHMEM_DATA_REGION_OFFSET + slot * IVSHMEM_SLOT_SIZE + header_bytes = self.read_region(offset, IVSHMEM_SLOT_HEADER_SIZE) + return IvshmemSlotHeader.unpack(header_bytes) + + def write_slot_header(self, slot: int, header: IvshmemSlotHeader) -> None: + """Write the header for a given slot.""" + offset = IVSHMEM_DATA_REGION_OFFSET + slot * IVSHMEM_SLOT_SIZE + self.write_region(offset, header.pack()) + + def read_slot_payload(self, slot: int, length: int) -> bytes: + """Read payload from a slot (after the header).""" + offset = ( + IVSHMEM_DATA_REGION_OFFSET + + slot * IVSHMEM_SLOT_SIZE + + IVSHMEM_SLOT_HEADER_SIZE + ) + return self.read_region(offset, length) + + def write_slot_payload(self, slot: int, data: bytes) -> None: + """Write payload to a slot (after the header).""" + offset = ( + IVSHMEM_DATA_REGION_OFFSET + + slot * IVSHMEM_SLOT_SIZE + + IVSHMEM_SLOT_HEADER_SIZE + ) + self.write_region(offset, data) + + def read_metadata(self, slot: int) -> IvshmemMetadata: + """Read metadata for a slot from the metadata region.""" + offset = IVSHMEM_METADATA_REGION_OFFSET + slot * IvshmemMetadata().sizeof() + data = self.read_region(offset, IvshmemMetadata().sizeof()) + return IvshmemMetadata.unpack(data) + + def write_metadata(self, slot: int, meta: IvshmemMetadata) -> None: + """Write metadata for a slot to the metadata region.""" + offset = IVSHMEM_METADATA_REGION_OFFSET + slot * meta.sizeof() + self.write_region(offset, meta.pack()) + + def ring_doorbell(self, peer_id: int = 0) -> None: + """Ring the doorbell register to notify the peer VM. + + In a real ivshmem setup, writing to BAR0 offset 0 triggers an + interrupt on the peer VM. In fallback mode, this is a no-op + (the peer reads from the same mmap). + """ + if self._mm is None: + raise RuntimeError("IvshmemClient not opened") + # Write peer_id to doorbell register (4 bytes at offset 0) + self._mm.seek(IVSHMEM_BAR0_DOORBELL_OFFSET) + self._mm.write(struct.pack("!I", peer_id)) + self.state.doorbell_rang = True + + def get_state(self) -> IvshmemState: + """Return a snapshot of the current client state.""" + return IvshmemState( + bar0_size=self.state.bar0_size, + slot_count=self.state.slot_count, + slot_size=self.state.slot_size, + active_slot=self.state.active_slot, + sequence_counter=self.state.sequence_counter, + bytes_written=self.state.bytes_written, + doorbell_rang=self.state.doorbell_rang, + ) + + +# ── IvshmemRing ──────────────────────────────────────────────────────────── + + +class IvshmemRing: + """Doorbell ring abstraction for ivshmem notification. + + Wraps the doorbell write primitive into a ring buffer pattern. + Each ring increments a sequence counter and writes to the doorbell + register, allowing the peer to track notification order. + """ + + def __init__(self, client: IvshmemClient, peer_id: int = 0): + self.client = client + self.peer_id = peer_id + self.ring_count: int = 0 + + def ring(self, slot: int = 0) -> int: + """Ring the doorbell and return the ring sequence number. + + Args: + slot: Which slot the peer should read from. + + Returns: + The ring sequence number (monotonically increasing). + """ + self.ring_count += 1 + self.client.ring_doorbell(self.peer_id) + return self.ring_count + + def ring_with_payload_length(self, slot: int, length: int) -> int: + """Ring the doorbell with metadata about the payload. + + Before ringing, writes the payload length to the slot header + so the peer knows how many bytes to read. + + Returns: + The ring sequence number. + """ + header = IvshmemSlotHeader(length=length, sequence=self.ring_count + 1) + self.client.write_slot_header(slot, header) + return self.ring(slot) + + +# ── IvshmemTransform ─────────────────────────────────────────────────────── + + +class IvshmemTransform: + """Encode data as a shared memory write and notify the peer. + + This is the primary transform primitive for ivshmem computation: + 1. Write payload to a shared memory slot + 2. Write metadata (length, sequence, flags) + 3. Ring doorbell to notify peer + 4. Generate a receipt + + The peer reads from the same slot offset — zero copy. + """ + + def __init__( + self, + client: IvshmemClient, + ring: Optional[IvshmemRing] = None, + ): + self.client = client + self.ring = ring or IvshmemRing(client) + self._sequence: int = 0 + + def transform( + self, + data: bytes, + slot: Optional[int] = None, + notify: bool = True, + ) -> Tuple[int, IvshmemReceipt]: + """Write data to shared memory and optionally notify the peer. + + Args: + data: Payload bytes to write. + slot: Slot index (auto-selected if None). + notify: Whether to ring the doorbell after writing. + + Returns: + (offset, receipt) where offset is the byte offset of the payload. + """ + if len(data) > IVSHMEM_MAX_PAYLOAD_SIZE: + raise ValueError( + f"Payload too large: {len(data)} > {IVSHMEM_MAX_PAYLOAD_SIZE}" + ) + + # Auto-select slot + if slot is None: + slot = self.client.state.active_slot + self.client.state.active_slot = (slot + 1) % self.client.state.slot_count + + self._sequence += 1 + + # Write payload to slot + payload_offset = ( + IVSHMEM_DATA_REGION_OFFSET + + slot * IVSHMEM_SLOT_SIZE + + IVSHMEM_SLOT_HEADER_SIZE + ) + self.client.write_slot_payload(slot, data) + + # Write slot header + header = IvshmemSlotHeader(length=len(data), sequence=self._sequence) + self.client.write_slot_header(slot, header) + + # Write metadata + import time + ts_lo = int(time.time() * 1000) & 0xFFFFFFFF + meta = IvshmemMetadata( + sequence=self._sequence, + length=len(data), + flags=IvshmemMetadata.FLAG_VALID, + timestamp_lo=ts_lo, + ) + self.client.write_metadata(slot, meta) + + # Ring doorbell + peer_notified = False + if notify: + self.ring.ring(slot) + peer_notified = True + + # Update state + self.client.state.sequence_counter = self._sequence + self.client.state.bytes_written += len(data) + + # Generate receipt + witness = self._compute_witness(data, slot, self._sequence) + receipt = IvshmemReceipt( + offset=payload_offset, + length=len(data), + slot=slot, + sequence=self._sequence, + doorbell=notify, + witness_hash=witness, + bar0_size=self.client.state.bar0_size, + slot_count=self.client.state.slot_count, + payload_crc32=f"{_crc32(data):08x}", + peer_notified=peer_notified, + ) + + return payload_offset, receipt + + def read_from_slot(self, slot: int) -> Tuple[bytes, IvshmemSlotHeader]: + """Read payload and header from a slot (peer-side operation).""" + header = self.client.read_slot_header(slot) + if header.length == 0: + return b"", header + payload = self.client.read_slot_payload(slot, header.length) + return payload, header + + def _compute_witness(self, data: bytes, slot: int, sequence: int) -> str: + """Compute a witness hash over the transform inputs.""" + h = hashlib.sha256() + h.update(data) + h.update(struct.pack("!II", slot, sequence)) + return h.hexdigest()[:16] + + +# ── Utility Functions ────────────────────────────────────────────────────── + + +def _crc32(data: bytes) -> int: + """Compute CRC32 of data.""" + import zlib + return zlib.crc32(data) & 0xFFFFFFFF + + +# ── CLI ──────────────────────────────────────────────────────────────────── + + +def main() -> None: + args = sys.argv[1:] + + if not args or args[0] == "--example": + print("=== Ivshmem Client — Zero-Copy VM-to-VM Communication ===") + print() + print("Three primitives for shared-memory computation:") + print() + print("1. IvshmemClient (BAR0 mmap):") + print(" Maps /dev/shm/ivshmem_bar0 into process memory") + print(" Read/write at arbitrary offsets, zero kernel copies") + print() + print("2. IvshmemRing (doorbell):") + print(" Write to BAR0 doorbell register → triggers peer interrupt") + print(" Sequence-numbered notifications for ordering") + print() + print("3. IvshmemTransform (write + notify):") + print(" payload → slot write → metadata → doorbell → peer reads") + print(" Receipt: {schema, transform_type: 'shared_memory', offset, length, witness_hash}") + print() + print("=== Memory Layout ===") + print(f" BAR0 registers: 0x{0:08X} - 0x{IVSHMEM_REGISTER_REGION_SIZE - 1:08X}") + print(f" Slot metadata: 0x{IVSHMEM_METADATA_REGION_OFFSET:08X} - " + f"0x{IVSHMEM_METADATA_REGION_OFFSET + IVSHMEM_DEFAULT_SLOT_COUNT * 16 - 1:08X}") + print(f" Slot data: 0x{IVSHMEM_DATA_REGION_OFFSET:08X} - ...") + print(f" Slot count: {IVSHMEM_DEFAULT_SLOT_COUNT}") + print(f" Slot size: {IVSHMEM_SLOT_SIZE} bytes (header + payload)") + print() + print("=== Demo ===") + + test_payload = b"\\xde\\xad\\xbe\\xef" * 16 + print(f"Payload ({len(test_payload)} bytes): {test_payload[:16].hex()}...") + print() + + # Create client with fallback + devshm_path = "/dev/shm/ivshmem_test_bar0" + print(f"Creating IvshmemClient (fallback: {devshm_path})...") + with IvshmemClient(device_path=devshm_path, bar0_size=IVSHMEM_BAR0_DEFAULT_SIZE) as client: + transform = IvshmemTransform(client) + offset, receipt = transform.transform(test_payload, slot=0, notify=True) + print(f"Write offset: 0x{offset:08X}") + print(f"Receipt: {json.dumps(receipt.to_dict(), indent=2)}") + print() + + # Read back + data, header = transform.read_from_slot(slot=0) + print(f"Read back: {data[:16].hex()}... ({len(data)} bytes)") + print(f"Header: length={header.length}, sequence={header.sequence}") + print() + + # Ring-only + ring = IvshmemRing(client, peer_id=1) + seq = ring.ring(slot=0) + print(f"Doorbell ring #{seq} to peer 1") + print() + + # Cleanup + if os.path.exists(devshm_path): + os.unlink(devshm_path) + return + + if args[0] == "--test-devshm": + print("Testing with /dev/shm fallback...") + test_path = "/dev/shm/ivshmem_test" + try: + with IvshmemClient(device_path=test_path) as client: + print(f" Opened: {test_path} (fallback={client.is_fallback})") + print(f" BAR0 size: {client.bar0_size} bytes") + + transform = IvshmemTransform(client) + payload = b"Hello, ivshmem!" + b"\\x00" * 48 + offset, receipt = transform.transform(payload, slot=0, notify=False) + print(f" Wrote {len(payload)} bytes at offset 0x{offset:08X}") + print(f" Receipt: {json.dumps(receipt.to_dict(), indent=2)}") + + data, header = transform.read_from_slot(slot=0) + assert data[:len(payload.rstrip(b'\\x00'))] == payload.rstrip(b'\\x00'), \ + "Read-back mismatch!" + print(f" Read-back verified: {len(data)} bytes") + print(" PASS") + finally: + if os.path.exists(test_path): + os.unlink(test_path) + return + + if args[0] == "--analyze": + if len(args) < 2: + print("Usage: ivshmem_client.py --analyze ", file=sys.stderr) + sys.exit(1) + payload = Path(args[1]).read_bytes() + print(f"=== {args[1]} ===") + print(f" Size: {len(payload)} bytes") + print(f" CRC32: {_crc32(payload):08x}") + print(f" SHA256: {hashlib.sha256(payload).hexdigest()[:16]}") + print(f" Fits in 1 slot: {len(payload) <= IVSHMEM_MAX_PAYLOAD_SIZE}") + print(f" Slots needed: {-(-len(payload) // IVSHMEM_MAX_PAYLOAD_SIZE)}") + print(f" Max slot payload: {IVSHMEM_MAX_PAYLOAD_SIZE} bytes") + return + + if args[0] == "write": + if len(args) < 2: + print("Usage: ivshmem_client.py write [--offset ] [--slot ]", file=sys.stderr) + sys.exit(1) + payload_path = args[1] + payload = Path(payload_path).read_bytes() + offset_arg = None + slot_arg = None + if "--offset" in args: + oi = args.index("--offset") + offset_arg = int(args[oi + 1]) + if "--slot" in args: + si = args.index("--slot") + slot_arg = int(args[si + 1]) + + devshm_path = "/dev/shm/ivshmem_bar0" + with IvshmemClient(device_path=devshm_path) as client: + transform = IvshmemTransform(client) + if slot_arg is not None: + offset, receipt = transform.transform(payload, slot=slot_arg, notify=True) + else: + offset, receipt = transform.transform(payload, notify=True) + print(json.dumps(receipt.to_dict(), indent=2)) + return + + if args[0] == "read": + offset_arg = 0 + length_arg = 64 + if "--offset" in args: + oi = args.index("--offset") + offset_arg = int(args[oi + 1]) + if "--length" in args: + li = args.index("--length") + length_arg = int(args[li + 1]) + if "--slot" in args: + si = args.index("--slot") + slot = int(args[si + 1]) + devshm_path = "/dev/shm/ivshmem_bar0" + with IvshmemClient(device_path=devshm_path) as client: + transform = IvshmemTransform(client) + data, header = transform.read_from_slot(slot) + result = { + "slot": slot, + "length": header.length, + "sequence": header.sequence, + "data_hex": data[:min(64, header.length)].hex(), + } + print(json.dumps(result, indent=2)) + else: + devshm_path = "/dev/shm/ivshmem_bar0" + with IvshmemClient(device_path=devshm_path) as client: + data = client.read_region(offset_arg, length_arg) + result = { + "offset": offset_arg, + "length": length_arg, + "data_hex": data.hex(), + } + print(json.dumps(result, indent=2)) + return + + if args[0] == "ring": + doorbell_peer = 0 + if "--doorbell" in args: + di = args.index("--doorbell") + doorbell_peer = int(args[di + 1]) + devshm_path = "/dev/shm/ivshmem_bar0" + with IvshmemClient(device_path=devshm_path) as client: + ring = IvshmemRing(client, peer_id=doorbell_peer) + seq = ring.ring(slot=0) + print(json.dumps({"peer_id": doorbell_peer, "sequence": seq}, indent=2)) + return + + if args[0] == "receipt": + if len(args) < 2: + print("Usage: ivshmem_client.py receipt ", file=sys.stderr) + sys.exit(1) + data = json.loads(Path(args[1]).read_text()) + print(json.dumps(data, indent=2)) + return + + print("Usage: ivshmem_client.py [args]", file=sys.stderr) + print("Commands: --example, --test-devshm, --analyze, write, read, ring, receipt", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/4-Infrastructure/shim/virtio_crypto_transform.py b/4-Infrastructure/shim/virtio_crypto_transform.py new file mode 100644 index 00000000..f4cad31a --- /dev/null +++ b/4-Infrastructure/shim/virtio_crypto_transform.py @@ -0,0 +1,608 @@ +#!/usr/bin/env python3 +""" +Virtio-Crypto Transform — Use virtio-crypto sessions as a hash computation pipeline. + +Implements the HASH session primitive from the virtio-crypto spec: + 1. VirtioCryptoSession — create a HASH session (SHA-256, SHA-512, MD5) + 2. VirtioCryptoHashTransform — encode data as a HASH request, produce receipt + 3. VirtioCryptoTransformReceipt — machine-readable result with witness + +This is NOT a crypto driver. It is a computation substrate that models the +virtio-crypto control queue and data queue protocol. The host QEMU backend +(gcrypt/AES-NI) processes hash requests without knowing it is being used +for general computation. + +Shim boundary: I/O only. No decision logic. All computation mapping decisions +belong in Lean (future VirtioCrypto modules). + +Reference: virtio/virtio_crypto.h (Linux kernel headers), RFC 6234 (SHA test vectors) + +Usage: + python3 virtio_crypto_transform.py --example + python3 virtio_crypto_transform.py --analyze + python3 virtio_crypto_transform.py hash [--algo sha256|sha512|md5] + python3 virtio_crypto_transform.py session-info [--algo sha256] + python3 virtio_crypto_transform.py receipt + python3 virtio_crypto_transform.py --test-vectors +""" + +from __future__ import annotations + +import hashlib +import json +import struct +import sys +from dataclasses import dataclass, field +from typing import Dict, Any, List, Optional, Tuple +from pathlib import Path + +# ── Virtio-Crypto Constants ────────────────────────────────────────────────── + +# Opcodes (from linux/virtio_crypto.h) +VIRTIO_CRYPTO_OPCODE_HASH: int = 0x06 +VIRTIO_CRYPTO_OPCODE_SYMCIPHER: int = 0x07 +VIRTIO_CRYPTO_OPCODE_AKCIPHER: int = 0x08 + +# Session types +VIRTIO_CRYPTO_OP_SESSION_CREATE: int = 0x01 +VIRTIO_CRYPTO_OP_SESSION_DESTROY: int = 0x02 + +# Hash algorithms (from linux/virtio_crypto.h) +VIRTIO_CRYPTO_HASH_SHA1: int = 0 +VIRTIO_CRYPTO_HASH_SHA224: int = 1 +VIRTIO_CRYPTO_HASH_SHA256: int = 2 +VIRTIO_CRYPTO_HASH_SHA384: int = 3 +VIRTIO_CRYPTO_HASH_SHA512: int = 4 +VIRTIO_CRYPTO_HASH_MD5: int = 5 +VIRTIO_CRYPTO_HASH_SHA3_224: int = 6 +VIRTIO_CRYPTO_HASH_SHA3_256: int = 7 +VIRTIO_CRYPTO_HASH_SHA3_384: int = 8 +VIRTIO_CRYPTO_HASH_SHA3_512: int = 9 + +# Status codes +VIRTIO_CRYPTO_OK: int = 0 +VIRTIO_CRYPTO_ERR: int = 1 +VIRTIO_CRYPTO_BADMSG: int = 2 +VIRTIO_CRYPTO_NOTSUPP: int = 3 +VIRTIO_CRYPTO_INVSESS: int = 4 + +# Feature bits +VIRTIO_CRYPTO_F_CIPHER_STATELESS: int = 0 +VIRTIO_CRYPTO_F_HASH_STATELESS: int = 1 + +# Algorithm name mapping +ALGO_MAP: Dict[str, int] = { + "md5": VIRTIO_CRYPTO_HASH_MD5, + "sha1": VIRTIO_CRYPTO_HASH_SHA1, + "sha224": VIRTIO_CRYPTO_HASH_SHA224, + "sha256": VIRTIO_CRYPTO_HASH_SHA256, + "sha384": VIRTIO_CRYPTO_HASH_SHA384, + "sha512": VIRTIO_CRYPTO_HASH_SHA512, + "sha3_224": VIRTIO_CRYPTO_HASH_SHA3_224, + "sha3_256": VIRTIO_CRYPTO_HASH_SHA3_256, + "sha3_384": VIRTIO_CRYPTO_HASH_SHA3_384, + "sha3_512": VIRTIO_CRYPTO_HASH_SHA3_512, +} + +ALGO_DIGEST_SIZE: Dict[str, int] = { + "md5": 16, + "sha1": 20, + "sha224": 28, + "sha256": 32, + "sha384": 48, + "sha512": 64, + "sha3_224": 28, + "sha3_256": 32, + "sha3_384": 48, + "sha3_512": 64, +} + +ALGO_VIRTIO_CODE: Dict[str, int] = { + "md5": VIRTIO_CRYPTO_HASH_MD5, + "sha1": VIRTIO_CRYPTO_HASH_SHA1, + "sha256": VIRTIO_CRYPTO_HASH_SHA256, + "sha384": VIRTIO_CRYPTO_HASH_SHA384, + "sha512": VIRTIO_CRYPTO_HASH_SHA512, +} + +# ── Virtio-Crypto Header Structures ───────────────────────────────────────── + + +@dataclass +class VirtioCryptoCtrlHdr: + """struct virtio_crypto_ctrl_header — control queue header. + + Layout (from linux/virtio_crypto.h): + __le32 opcode + __le32 algo + __le32 flag + __le32 session_id + __u8 padding[4] + Total: 20 bytes + """ + opcode: int = 0 + algo: int = 0 + flag: int = 0 + session_id: int = 0 + + CTRL_SIZE: int = 20 # 4 + 4 + 4 + 4 + 4 padding + + def pack(self) -> bytes: + return struct.pack( + "!IIIII", + self.opcode, + self.algo, + self.flag, + self.session_id, + 0, # padding + ) + + @staticmethod + def unpack(data: bytes) -> "VirtioCryptoCtrlHdr": + opcode, algo, flag, session_id, _ = struct.unpack("!IIIII", data[:20]) + return VirtioCryptoCtrlHdr( + opcode=opcode, + algo=algo, + flag=flag, + session_id=session_id, + ) + + +@dataclass +class VirtioCryptoSessionInput: + """struct virtio_crypto_session_input — session creation result. + + Layout: + __le32 session_id + __le32 status + Total: 8 bytes + """ + session_id: int = 0 + status: int = VIRTIO_CRYPTO_OK + + def pack(self) -> bytes: + return struct.pack("!II", self.session_id, self.status) + + @staticmethod + def unpack(data: bytes) -> "VirtioCryptoSessionInput": + session_id, status = struct.unpack("!II", data[:8]) + return VirtioCryptoSessionInput(session_id=session_id, status=status) + + +@dataclass +class VirtioCryptoHashSessionPara: + """struct virtio_crypto_hash_session_para — hash session parameters. + + Layout: + __le32 algo + __le32 hash_result_len + Total: 8 bytes + """ + algo: int = VIRTIO_CRYPTO_HASH_SHA256 + hash_result_len: int = 32 + + def pack(self) -> bytes: + return struct.pack("!II", self.algo, self.hash_result_len) + + @staticmethod + def unpack(data: bytes) -> "VirtioCryptoHashSessionPara": + algo, hash_result_len = struct.unpack("!II", data[:8]) + return VirtioCryptoHashSessionPara(algo=algo, hash_result_len=hash_result_len) + + +@dataclass +class VirtioCryptoHashDataReq: + """struct virtio_crypto_hash_data_req — data queue request for hash. + + Layout: + struct virtio_crypto_ctrl_hdr header + __le32 src_data_len + __le32 dst_data_len + Total: 28 bytes (20 + 4 + 4) + """ + header: VirtioCryptoCtrlHdr = field(default_factory=VirtioCryptoCtrlHdr) + src_data_len: int = 0 + dst_data_len: int = 0 + + HASH_REQ_SIZE: int = 28 + + def pack(self) -> bytes: + return self.header.pack() + struct.pack( + "!II", self.src_data_len, self.dst_data_len + ) + + @staticmethod + def unpack(data: bytes) -> "VirtioCryptoHashDataReq": + header = VirtioCryptoCtrlHdr.unpack(data[:20]) + src_data_len, dst_data_len = struct.unpack("!II", data[20:28]) + return VirtioCryptoHashDataReq( + header=header, + src_data_len=src_data_len, + dst_data_len=dst_data_len, + ) + + +# ── Session and Transform ──────────────────────────────────────────────────── + + +@dataclass +class VirtioCryptoSession: + """A HASH session in virtio-crypto. + + Models the session creation handshake: + Guest → control queue: VirtioCryptoCtrlHdr + VirtioCryptoHashSessionPara + Host → control queue: VirtioCryptoSessionInput (session_id, status) + + The session_id is then used for all subsequent hash requests on this session. + """ + algo: str = "sha256" + algo_code: int = VIRTIO_CRYPTO_HASH_SHA256 + digest_size: int = 32 + session_id: int = 0 + status: int = VIRTIO_CRYPTO_OK + + @staticmethod + def create(algo: str = "sha256") -> "VirtioCryptoSession": + """Create a new hash session for the given algorithm.""" + algo_lower = algo.lower() + if algo_lower not in ALGO_MAP: + raise ValueError(f"Unknown algorithm: {algo}. Supported: {list(ALGO_MAP.keys())}") + algo_code = ALGO_MAP[algo_lower] + digest_size = ALGO_DIGEST_SIZE[algo_lower] + return VirtioCryptoSession( + algo=algo_lower, + algo_code=algo_code, + digest_size=digest_size, + ) + + def session_create_request(self) -> bytes: + """Encode the control queue session creation request.""" + ctrl = VirtioCryptoCtrlHdr( + opcode=VIRTIO_CRYPTO_OP_SESSION_CREATE, + algo=self.algo_code, + flag=VIRTIO_CRYPTO_F_HASH_STATELESS, + session_id=0, + ) + para = VirtioCryptoHashSessionPara( + algo=self.algo_code, + hash_result_len=self.digest_size, + ) + return ctrl.pack() + para.pack() + + def session_create_response(self) -> bytes: + """Encode the host's session creation response.""" + resp = VirtioCryptoSessionInput( + session_id=self.session_id, + status=self.status, + ) + return resp.pack() + + def destroy_request(self) -> bytes: + """Encode the session destroy request.""" + ctrl = VirtioCryptoCtrlHdr( + opcode=VIRTIO_CRYPTO_OP_SESSION_DESTROY, + algo=self.algo_code, + flag=0, + session_id=self.session_id, + ) + return ctrl.pack() + + +@dataclass +class VirtioCryptoTransformReceipt: + """Machine-readable receipt for a virtio-crypto hash transform.""" + schema: str = "virtio_crypto_transform_receipt_v1" + transform_type: str = "hash" + algo: str = "sha256" + algo_code: int = VIRTIO_CRYPTO_HASH_SHA256 + payload_bytes: int = 0 + result_hex: str = "" + result_bytes: int = 0 + session_id: int = 0 + witness_hash: str = "" + request_header_hex: str = "" + + def to_dict(self) -> Dict[str, Any]: + return { + "schema": self.schema, + "transform_type": self.transform_type, + "algo": self.algo, + "algo_code": self.algo_code, + "payload_bytes": self.payload_bytes, + "result_hex": self.result_hex, + "result_bytes": self.result_bytes, + "session_id": self.session_id, + "witness_hash": self.witness_hash, + "request_header_hex": self.request_header_hex, + } + + +def _compute_hash(algo: str, data: bytes) -> bytes: + """Compute a hash using Python's hashlib (models QEMU's crypto backend).""" + import hashlib + h = hashlib.new(algo) + h.update(data) + return h.digest() + + +def _witness(data: bytes) -> str: + """SHA-256 of the full request+result bundle for receipt verification.""" + import hashlib + return hashlib.sha256(data).hexdigest() + + +# ── Transform Implementation ───────────────────────────────────────────────── + + +def transform_hash( + payload: bytes, + algo: str = "sha256", + session_id: int = 1, +) -> Tuple[bytes, VirtioCryptoTransformReceipt]: + """HASH transform: encode data as a virtio-crypto HASH request, compute hash. + + Protocol flow: + 1. Guest creates HASH session on control queue → session_id + 2. Guest posts VirtioCryptoHashDataReq + payload on data queue + 3. Host (QEMU crypto backend) computes hash + 4. Host writes hash result to dst buffer on data queue + + This function models the full round-trip. The hash is computed locally + (matching what QEMU's gcrypt backend would produce). + + Returns (result_digest, receipt). + """ + session = VirtioCryptoSession.create(algo) + session.session_id = session_id + + # Encode the data queue request header + data_req = VirtioCryptoHashDataReq( + header=VirtioCryptoCtrlHdr( + opcode=VIRTIO_CRYPTO_OPCODE_HASH, + algo=session.algo_code, + flag=VIRTIO_CRYPTO_F_HASH_STATELESS, + session_id=session_id, + ), + src_data_len=len(payload), + dst_data_len=session.digest_size, + ) + + # Compute hash (models QEMU crypto backend) + digest = _compute_hash(algo, payload) + + # Build receipt + request_bytes = data_req.pack() + payload + receipt = VirtioCryptoTransformReceipt( + transform_type="hash", + algo=algo, + algo_code=session.algo_code, + payload_bytes=len(payload), + result_hex=digest.hex(), + result_bytes=len(digest), + session_id=session_id, + witness_hash=_witness(request_bytes + digest), + request_header_hex=data_req.pack().hex(), + ) + + return digest, receipt + + +# ── RFC 6234 Test Vectors ──────────────────────────────────────────────────── + +SHA256_TEST_VECTORS: List[Tuple[str, str]] = [ + # (input, expected_sha256_hex) + ("", "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"), + ("abc", "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"), + ("abcdbcdecdefdefgefghfghighijhijkijkljklmklmnlmnomnopnopq", + "248d6a61d20638b8e5c026930c3e6039a33ce45964ff2167f6ecedd419db06c1"), + ("abcdefghbcdefghicdefghijdefghijkefghijklfghijklmghijklmnhijklmnoijklmnopjklmnopqklmnopqrlmnopqrsmnopqrstnopqrstu", + "cf5b16a778af8380036ce59e7b0492370b249b11e8f07a51afac45037afee9d1"), +] + +SHA512_TEST_VECTORS: List[Tuple[str, str]] = [ + ("", "cf83e1357eefb8bdf1542850d66d8007d620e4050b5715dc83f4a921d36ce9ce47d0d13c5d85f2b0ff8318d2877eec2f63b931bd47417a81a538327af927da3e"), + ("abc", "ddaf35a193617abacc417349ae20413112e6fa4e89a97ea20a9eeee64b55d39a2192992a274fc1a836ba3c23a3feebbd454d4423643ce80e2a9ac94fa54ca49f"), +] + +MD5_TEST_VECTORS: List[Tuple[str, str]] = [ + ("", "d41d8cd98f00b204e9800998ecf8427e"), + ("abc", "900150983cd24fb0d6963f7d28e17f72"), + ("message digest", "f96b697d7cb7938d525a2f31aaf161d0"), + ("abcdefghijklmnopqrstuvwxyz", "c3fcd3d76192e4007dfb496cca67e13b"), +] + + +# ── CLI ────────────────────────────────────────────────────────────────────── + + +def main(): + args = sys.argv[1:] + + if not args or args[0] == "--example": + print("=== Virtio-Crypto Transform ===") + print() + print("HASH computation via virtio-crypto sessions:") + print() + print(" payload → HASH session (SHA-256/SHA-512/MD5)") + print(" → QEMU crypto backend computes hash") + print(" → result digest + receipt") + print() + print("=== Protocol ===") + print() + print(" 1. Control queue: CREATE_SESSION(algo) → session_id") + print(" 2. Data queue: HASH_REQUEST(session_id, payload) → digest") + print(" 3. Control queue: DESTROY_SESSION(session_id)") + print() + print("=== Demo ===") + print() + + # SHA-256 + payload = b"abc" + digest, receipt = transform_hash(payload, algo="sha256") + print(f"SHA-256({payload!r}):") + print(f" result: {digest.hex()}") + print(f" receipt: {json.dumps(receipt.to_dict(), indent=2)}") + print() + + # SHA-512 + digest, receipt = transform_hash(payload, algo="sha512") + print(f"SHA-512({payload!r}):") + print(f" result: {digest.hex()}") + print() + + # MD5 + digest, receipt = transform_hash(payload, algo="md5") + print(f"MD5({payload!r}):") + print(f" result: {digest.hex()}") + print() + + # Session creation wire format + session = VirtioCryptoSession.create("sha256") + session.session_id = 1 + req = session.session_create_request() + print(f"Session create request ({len(req)} bytes): {req.hex()}") + resp = session.session_create_response() + print(f"Session create response ({len(resp)} bytes): {resp.hex()}") + print() + + # Data request wire format + data_req = VirtioCryptoHashDataReq( + header=VirtioCryptoCtrlHdr( + opcode=VIRTIO_CRYPTO_OPCODE_HASH, + algo=VIRTIO_CRYPTO_HASH_SHA256, + flag=VIRTIO_CRYPTO_F_HASH_STATELESS, + session_id=1, + ), + src_data_len=3, + dst_data_len=32, + ) + print(f"Data request header ({len(data_req.pack())} bytes): {data_req.pack().hex()}") + return + + if args[0] == "--analyze": + if len(args) < 2: + print("Usage: virtio_crypto_transform.py --analyze ", file=sys.stderr) + sys.exit(1) + payload = Path(args[1]).read_bytes() + print(f"=== {args[1]} ===") + print(f" Size: {len(payload)} bytes") + print(f" SHA-256: {hashlib.sha256(payload).hexdigest()}") + print(f" SHA-512: {hashlib.sha512(payload).hexdigest()}") + print(f" MD5: {hashlib.md5(payload).hexdigest()}") + print(f" Digest sizes: sha256=32B, sha512=64B, md5=16B") + return + + if args[0] == "--test-vectors": + print("=== RFC 6234 Test Vector Verification ===") + print() + all_pass = True + + # SHA-256 + print("--- SHA-256 ---") + for i, (inp, expected) in enumerate(SHA256_TEST_VECTORS): + digest, receipt = transform_hash(inp.encode(), algo="sha256") + actual = digest.hex() + ok = actual == expected + status = "PASS" if ok else "FAIL" + print(f" [{status}] vector {i}: input={inp[:40]!r}{'...' if len(inp) > 40 else ''}") + if not ok: + print(f" expected: {expected}") + print(f" actual: {actual}") + all_pass = False + print() + + # SHA-512 + print("--- SHA-512 ---") + for i, (inp, expected) in enumerate(SHA512_TEST_VECTORS): + digest, receipt = transform_hash(inp.encode(), algo="sha512") + actual = digest.hex() + ok = actual == expected + status = "PASS" if ok else "FAIL" + print(f" [{status}] vector {i}: input={inp[:40]!r}{'...' if len(inp) > 40 else ''}") + if not ok: + print(f" expected: {expected}") + print(f" actual: {actual}") + all_pass = False + print() + + # MD5 + print("--- MD5 ---") + for i, (inp, expected) in enumerate(MD5_TEST_VECTORS): + digest, receipt = transform_hash(inp.encode(), algo="md5") + actual = digest.hex() + ok = actual == expected + status = "PASS" if ok else "FAIL" + print(f" [{status}] vector {i}: input={inp[:40]!r}{'...' if len(inp) > 40 else ''}") + if not ok: + print(f" expected: {expected}") + print(f" actual: {actual}") + all_pass = False + print() + + # Receipt schema check + print("--- Receipt Schema ---") + _, receipt = transform_hash(b"abc", algo="sha256") + d = receipt.to_dict() + required_keys = {"schema", "transform_type", "algo", "payload_bytes", + "result_hex", "witness_hash"} + missing = required_keys - set(d.keys()) + if missing: + print(f" [FAIL] Missing keys in receipt: {missing}") + all_pass = False + else: + print(f" [PASS] Receipt has all required keys: {sorted(required_keys)}") + print(f" schema={d['schema']}, transform_type={d['transform_type']}") + print() + + if all_pass: + print("ALL TEST VECTORS PASSED") + else: + print("SOME TESTS FAILED") + sys.exit(1) + return + + if args[0] == "hash": + if len(args) < 2: + print("Usage: virtio_crypto_transform.py hash [--algo sha256|sha512|md5]", file=sys.stderr) + sys.exit(1) + payload = Path(args[1]).read_bytes() + algo = "sha256" + if "--algo" in args: + ai = args.index("--algo") + algo = args[ai + 1] + digest, receipt = transform_hash(payload, algo=algo) + print(json.dumps(receipt.to_dict(), indent=2)) + return + + if args[0] == "session-info": + algo = "sha256" + if "--algo" in args: + ai = args.index("--algo") + algo = args[ai + 1] + session = VirtioCryptoSession.create(algo) + info = { + "algo": session.algo, + "algo_code": session.algo_code, + "digest_size": session.digest_size, + "supported_algorithms": list(ALGO_MAP.keys()), + "session_create_request_size": len(session.session_create_request()), + "session_create_response_size": len(session.session_create_response()), + } + print(json.dumps(info, indent=2)) + return + + if args[0] == "receipt": + if len(args) < 2: + print("Usage: virtio_crypto_transform.py receipt ", file=sys.stderr) + sys.exit(1) + data = json.loads(Path(args[1]).read_text()) + print(json.dumps(data, indent=2)) + return + + print("Usage: virtio_crypto_transform.py [args]", file=sys.stderr) + print("Commands: --example, --analyze, --test-vectors, hash, session-info, receipt", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main()