reader

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

flavor.psp.format_2025.reader ¶

PSPF 2025 Format Reader.

Classes¶

PSPFReader ¶

PSPFReader(
    bundle_path: Path | str, mode: int = ACCESS_AUTO
)

Read PSPF bundles with backend support.

Initialize reader with specified backend mode.

Parameters:

Name	Type	Description	Default
`bundle_path`	`Path \| str`	Path to PSPF bundle	required
`mode`	`int`	Backend mode (ACCESS_AUTO, ACCESS_MMAP, ACCESS_FILE, etc.)	`ACCESS_AUTO`

Source code in flavor/psp/format_2025/reader.py

def __init__(self, bundle_path: Path | str, mode: int = ACCESS_AUTO) -> None:
    """Initialize reader with specified backend mode.

    Args:
        bundle_path: Path to PSPF bundle
        mode: Backend mode (ACCESS_AUTO, ACCESS_MMAP, ACCESS_FILE, etc.)
    """
    self.bundle_path = Path(bundle_path) if isinstance(bundle_path, str) else bundle_path
    self._backend: Backend | None = None
    self._index: PSPFIndex | None = None
    self._metadata: dict[str, Any] | None = None
    self._launcher_size: int | None = None
    self._slot_descriptors: list[SlotDescriptor] | None = None
    self.mode = mode

    # Slot extractor for extraction operations
    from flavor.psp.format_2025.extraction import SlotExtractor

    self._extractor = SlotExtractor(self)

Functions¶

del ¶

__del__() -> None

Ensure backend is closed when object is garbage collected (important on Windows).

Source code in flavor/psp/format_2025/reader.py

def __del__(self) -> None:
    """Ensure backend is closed when object is garbage collected (important on Windows)."""
    with contextlib.suppress(Exception):
        self.close()

enter ¶

__enter__() -> Self

Context manager entry.

Source code in flavor/psp/format_2025/reader.py

def __enter__(self) -> Self:
    """Context manager entry."""
    self.open()
    return self

exit ¶

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Context manager exit.

Source code in flavor/psp/format_2025/reader.py

def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Context manager exit."""
    self.close()

close ¶

close() -> None

Close the backend.

Source code in flavor/psp/format_2025/reader.py

def close(self) -> None:
    """Close the backend."""
    if self._backend:
        self._backend.close()
        self._backend = None

extract_slot ¶

extract_slot(slot_index: int, dest_dir: Path) -> Path

Extract a slot to a directory.

Source code in flavor/psp/format_2025/reader.py

def extract_slot(self, slot_index: int, dest_dir: Path) -> Path:
    """Extract a slot to a directory."""
    return self._extractor.extract_slot(slot_index, dest_dir)

extraction_lock ¶

extraction_lock(
    extract_dir: Path, timeout: float = 30.0
) -> Generator[Path, None, None]

Acquire an extraction lock for a given directory.

Source code in flavor/psp/format_2025/reader.py

@contextmanager
def extraction_lock(self, extract_dir: Path, timeout: float = 30.0) -> Generator[Path, None, None]:
    """Acquire an extraction lock for a given directory."""
    from flavor.locking import default_lock_manager

    lock_file = extract_dir / ".extraction.lock"
    with default_lock_manager.lock(lock_file.name, timeout=timeout) as lock:
        yield lock

get_backend ¶

get_backend() -> Backend

Get the backend for advanced operations.

Source code in flavor/psp/format_2025/reader.py

def get_backend(self) -> Backend:
    """Get the backend for advanced operations."""
    if not self._backend:
        self.open()
    assert self._backend is not None
    return self._backend

get_slot_view ¶

get_slot_view(slot_index: int) -> SlotView

Get a lazy view of a slot.

Source code in flavor/psp/format_2025/reader.py

def get_slot_view(self, slot_index: int) -> SlotView:
    """Get a lazy view of a slot."""
    return self._extractor.get_slot_view(slot_index)

open ¶

open() -> None

Open the bundle with appropriate backend.

Source code in flavor/psp/format_2025/reader.py

def open(self) -> None:
    """Open the bundle with appropriate backend."""
    if self._backend is None:
        self._backend = create_backend(self.mode, self.bundle_path)
        self._backend.open(self.bundle_path)

read_index ¶

read_index() -> PSPFIndex

Read and verify index block.

Source code in flavor/psp/format_2025/reader.py

def read_index(self) -> PSPFIndex:
    """Read and verify index block."""
    if self._index:
        return self._index

    if not self._backend:
        self.open()

    # Read index from MagicTrailer
    index_data = self.read_magic_trailer()

    # Convert to bytes if memoryview
    if isinstance(index_data, memoryview):
        index_data = bytes(index_data)

    self._index = PSPFIndex.unpack(index_data)

    # Debug log the parsed index values
    logger.debug(
        "📊 Parsed index values",
        package_size=self._index.package_size,
        launcher_size=self._index.launcher_size,
        metadata_offset=f"0x{self._index.metadata_offset:016x}",
        metadata_size=self._index.metadata_size,
        slot_table_offset=f"0x{self._index.slot_table_offset:016x}",
        slot_count=self._index.slot_count,
    )

    # Verify checksum (Adler-32 with checksum field as 0)
    expected_checksum = self._index.index_checksum
    if expected_checksum != 0:  # Only verify if checksum is set
        data_for_check = bytearray(index_data)
        data_for_check[4:8] = (
            b"\x00\x00\x00\x00"  # Zero out checksum field at offset 4 (after format_version)
        )
        actual_checksum = zlib.adler32(data_for_check) & 0xFFFFFFFF

        if expected_checksum != actual_checksum:
            raise ValueError(
                f"Index checksum mismatch: expected {expected_checksum}, got {actual_checksum}"
            )

    return self._index

read_magic_trailer ¶

read_magic_trailer() -> bytes

Read MagicTrailer and extract index data.

Source code in flavor/psp/format_2025/reader.py

def read_magic_trailer(self) -> bytes:
    """Read MagicTrailer and extract index data."""
    if not self._backend:
        self.open()

    assert self._backend is not None
    file_size = self.bundle_path.stat().st_size

    # Read MagicTrailer (last 8200 bytes)
    trailer = self._backend.read_at(file_size - DEFAULT_MAGIC_TRAILER_SIZE, DEFAULT_MAGIC_TRAILER_SIZE)

    # Convert to bytes if memoryview
    if isinstance(trailer, memoryview):
        trailer = bytes(trailer)

    # Verify magic bytes
    if trailer[:4] != TRAILER_START_MAGIC:
        raise ValueError("Invalid MagicTrailer: missing start marker")
    if trailer[-4:] != TRAILER_END_MAGIC:
        raise ValueError("Invalid MagicTrailer: missing end marker")

    # Extract index from between magic markers
    index_data = trailer[4 : 4 + DEFAULT_HEADER_SIZE]

    logger.debug(
        "🔍 Found index in MagicTrailer",
        trailer_size=DEFAULT_MAGIC_TRAILER_SIZE,
        file_size=file_size,
    )

    return index_data

read_metadata ¶

read_metadata() -> dict[str, Any]

Read and parse metadata.

Source code in flavor/psp/format_2025/reader.py

def read_metadata(self) -> dict[str, Any]:
    """Read and parse metadata."""
    if self._metadata:
        return self._metadata

    if not self._backend:
        self.open()

    assert self._backend is not None
    index = self.read_index()

    # Read metadata using backend
    metadata_data = self._backend.read_at(index.metadata_offset, index.metadata_size)

    # Convert to bytes if memoryview
    if isinstance(metadata_data, memoryview):
        metadata_data = bytes(metadata_data)

    # Verify metadata checksum (full SHA-256 - 32 bytes)
    import hashlib

    actual_checksum = hashlib.sha256(metadata_data).digest()
    expected_checksum = index.metadata_checksum if index.metadata_checksum else b"\x00" * 32
    if expected_checksum != b"\x00" * 32 and actual_checksum != expected_checksum:
        raise ValueError(
            f"Metadata checksum mismatch: expected {expected_checksum.hex()[:16]}..., got {actual_checksum.hex()[:16]}..."
        )

    # Parse metadata (always gzipped JSON in current implementation)
    # Decompress first
    try:
        metadata_data = gzip.decompress(metadata_data)
    except gzip.BadGzipFile as exc:
        raise ValueError("Metadata section is not valid gzip data; bundle may be corrupt") from exc

    # Parse JSON
    self._metadata = json_loads(metadata_data.decode("utf-8"))

    return self._metadata

read_slot ¶

read_slot(slot_index: int) -> bytes

Read a specific slot.

Parameters:

Name	Type	Description	Default
`slot_index`	`int`	Index of the slot to read	required

Returns:

Name	Type	Description
`bytes`	`bytes`	Decompressed slot data

Raises:

Type	Description
`ValueError`	If slot index is invalid

Source code in flavor/psp/format_2025/reader.py

def read_slot(self, slot_index: int) -> bytes:
    """Read a specific slot.

    Args:
        slot_index: Index of the slot to read

    Returns:
        bytes: Decompressed slot data

    Raises:
        ValueError: If slot index is invalid
    """
    if not self._backend:
        self.open()

    assert self._backend is not None
    descriptors = self.read_slot_descriptors()

    if slot_index < 0 or slot_index >= len(descriptors):
        raise ValueError(f"Invalid slot index: {slot_index} (have {len(descriptors)} slots)")

    descriptor = descriptors[slot_index]

    # Read slot data using backend
    slot_data = self._backend.read_slot(descriptor)

    # Convert to bytes if memoryview
    if isinstance(slot_data, memoryview):
        slot_data = bytes(slot_data)

    # Verify checksum (SHA-256 first 8 bytes)
    import hashlib

    hash_bytes = hashlib.sha256(slot_data).digest()[:8]
    actual_checksum = int.from_bytes(hash_bytes, byteorder="little")

    # DEBUG: Log checksum details for troubleshooting
    logger.debug(
        "🔍 Verifying slot checksum",
        slot_index=slot_index,
        expected=f"{descriptor.checksum:016x}",
        actual=f"{actual_checksum:016x}",
        data_size=len(slot_data),
    )

    if actual_checksum != descriptor.checksum:
        logger.error(
            f"❌ Slot {slot_index} checksum mismatch: expected {descriptor.checksum:016x}, got {actual_checksum:016x}, size={len(slot_data)}"
        )
        raise ValueError(
            f"Slot {slot_index} checksum mismatch: expected {descriptor.checksum:016x}, got {actual_checksum:016x}"
        )

    # Decompress if needed based on operations
    from flavor.psp.format_2025.operations import OP_GZIP, OP_TAR, unpack_operations

    ops = unpack_operations(descriptor.operations)

    if ops == [OP_GZIP]:
        return gzip.decompress(slot_data)
    elif ops == [OP_TAR, OP_GZIP]:
        # For tar.gz, decompress gzip layer (tar extraction happens later)
        return gzip.decompress(slot_data)
    elif ops == [OP_TAR]:
        # Uncompressed tar, no decompression needed
        return slot_data
    else:
        return slot_data

read_slot_descriptors ¶

read_slot_descriptors() -> list[SlotDescriptor]

Read all slot descriptors.

Source code in flavor/psp/format_2025/reader.py

def read_slot_descriptors(self) -> list[SlotDescriptor]:
    """Read all slot descriptors."""
    if self._slot_descriptors:
        return self._slot_descriptors

    if not self._backend:
        self.open()

    assert self._backend is not None
    index = self.read_index()
    descriptors = []

    # Read all slot descriptors
    for i in range(index.slot_count):
        offset = index.slot_table_offset + (i * DEFAULT_SLOT_DESCRIPTOR_SIZE)
        data = self._backend.read_at(offset, DEFAULT_SLOT_DESCRIPTOR_SIZE)

        # Convert to bytes if memoryview
        if isinstance(data, memoryview):
            data = bytes(data)

        descriptor = SlotDescriptor.unpack(data)
        descriptors.append(descriptor)

    self._slot_descriptors = descriptors
    return descriptors

stream_slot ¶

stream_slot(slot_index: int, chunk_size: int = 8192) -> Any

Stream a slot in chunks.

Source code in flavor/psp/format_2025/reader.py

def stream_slot(self, slot_index: int, chunk_size: int = 8192) -> Any:
    """Stream a slot in chunks."""
    return self._extractor.stream_slot(slot_index, chunk_size)

use_mmap ¶

use_mmap() -> None

Switch to memory-mapped backend for efficiency.

Source code in flavor/psp/format_2025/reader.py

def use_mmap(self) -> None:
    """Switch to memory-mapped backend for efficiency."""
    self.close()
    self.mode = ACCESS_MMAP
    self.open()

use_streaming ¶

use_streaming(chunk_size: int = 64 * 1024) -> None

Switch to streaming backend for large files.

Source code in flavor/psp/format_2025/reader.py

def use_streaming(self, chunk_size: int = 64 * 1024) -> None:
    """Switch to streaming backend for large files."""
    self.close()
    self._backend = StreamBackend(chunk_size)
    self._backend.open(self.bundle_path)

verify_all_checksums ¶

verify_all_checksums() -> bool

Verify all slot checksums.

Source code in flavor/psp/format_2025/reader.py

def verify_all_checksums(self) -> bool:
    """Verify all slot checksums."""
    return self._extractor.verify_all_checksums()

verify_attestation_policy_hash ¶

verify_attestation_policy_hash() -> None

Verify attestation_policy_hash in the index matches the package-declared policy.

If the index field is zero-filled (absent), this is a no-op. Raises ValueError on mismatch or if field is non-zero but no policy in metadata.

Source code in flavor/psp/format_2025/reader.py

def verify_attestation_policy_hash(self) -> None:
    """Verify attestation_policy_hash in the index matches the package-declared policy.

    If the index field is zero-filled (absent), this is a no-op.
    Raises ValueError on mismatch or if field is non-zero but no policy in metadata.
    """
    import hashlib
    import json

    index = self.read_index()
    stored = index.attestation_policy_hash.rstrip(b"\x00")
    if not stored:
        return

    metadata = self.read_metadata()
    policy_raw = metadata.get("policy", {})
    if not policy_raw:
        raise ValueError("attestation_policy_hash is non-zero but package metadata has no 'policy' key")

    canonical = json.dumps(policy_raw, sort_keys=True, separators=(",", ":"))
    computed = hashlib.sha256(canonical.encode()).hexdigest()
    if computed != stored.decode("ascii"):
        raise ValueError(
            f"attestation_policy_hash mismatch: index={stored.decode()!r} computed={computed!r}"
        )

verify_integrity ¶

verify_integrity() -> dict[str, Any]

Verify complete package integrity.

Returns:

Name	Type	Description
`dict`	`dict[str, Any]`	Verification result with standard keys

Source code in flavor/psp/format_2025/reader.py

def verify_integrity(self) -> dict[str, Any]:
    """Verify complete package integrity.

    Returns:
        dict: Verification result with standard keys
    """
    try:
        # Verify individual components
        magic_valid = self.verify_magic_trailer()
        checksums_valid = self.verify_all_checksums()
        signature_valid = self.verify_signature()
        valid = magic_valid and checksums_valid and signature_valid

        return {
            "valid": valid,
            "magic_valid": magic_valid,
            "checksums_valid": checksums_valid,
            "signature_valid": signature_valid,
            "tamper_detected": not valid,
            "error": None if valid else "Verification failed",
        }
    except Exception as e:
        logger.error(f"Integrity verification failed: {e}")
        return {
            "valid": False,
            "magic_valid": False,
            "checksums_valid": False,
            "signature_valid": False,
            "tamper_detected": True,
            "error": str(e),
        }

verify_magic_trailer ¶

verify_magic_trailer() -> bool

Verify MagicTrailer emoji bookends at end of file.

Source code in flavor/psp/format_2025/reader.py

def verify_magic_trailer(self) -> bool:
    """Verify MagicTrailer emoji bookends at end of file."""
    if not self._backend:
        self.open()

    assert self._backend is not None
    # Read MagicTrailer at end of file
    file_size = self.bundle_path.stat().st_size
    trailer = self._backend.read_at(file_size - DEFAULT_MAGIC_TRAILER_SIZE, DEFAULT_MAGIC_TRAILER_SIZE)

    # Convert to bytes if memoryview
    if isinstance(trailer, memoryview):
        trailer = bytes(trailer)

    # Verify magic bytes at start and end
    return trailer[:4] == TRAILER_START_MAGIC and trailer[-4:] == TRAILER_END_MAGIC

verify_signature ¶

verify_signature() -> bool

Verify bundle signature.

Per PSPF/2025 spec: signature covers the uncompressed JSON metadata.

Returns:

Name	Type	Description
`bool`	`bool`	True if signature is valid

Source code in flavor/psp/format_2025/reader.py

def verify_signature(self) -> bool:
    """Verify bundle signature.

    Per PSPF/2025 spec: signature covers the uncompressed JSON metadata.

    Returns:
        bool: True if signature is valid
    """
    if not self._backend:
        self.open()

    assert self._backend is not None
    index = self.read_index()

    # Get the signature from the index block
    signature = index.integrity_signature[:64]  # First 64 bytes, rest is padding

    # Get the metadata to verify (uncompressed JSON)
    metadata_compressed = self._backend.read_at(index.metadata_offset, index.metadata_size)

    # Convert to bytes if memoryview
    if isinstance(metadata_compressed, memoryview):
        metadata_compressed = bytes(metadata_compressed)

    # Decompress to get the original JSON that was signed
    import gzip

    metadata_json = gzip.decompress(metadata_compressed)

    verifier = Ed25519Verifier(index.public_key)
    return verifier.verify(metadata_json, signature)  # type: ignore[no-any-return]

verify_slot_integrity ¶

verify_slot_integrity(slot_index: int) -> bool

Verify integrity of a specific slot.

Source code in flavor/psp/format_2025/reader.py

def verify_slot_integrity(self, slot_index: int) -> bool:
    """Verify integrity of a specific slot."""
    return self._extractor.verify_slot_integrity(slot_index)

Functions¶

read_bundle ¶

read_bundle(
    bundle_path: Path, use_mmap: bool = True
) -> PSPFReader

Open a bundle for reading.

Parameters:

Name	Type	Description	Default
`bundle_path`	`Path`	Path to bundle	required
`use_mmap`	`bool`	Whether to use memory mapping	`True`

Returns:

Name	Type	Description
`PSPFReader`	`PSPFReader`	Reader instance

Source code in flavor/psp/format_2025/reader.py

def read_bundle(bundle_path: Path, use_mmap: bool = True) -> PSPFReader:
    """Open a bundle for reading.

    Args:
        bundle_path: Path to bundle
        use_mmap: Whether to use memory mapping

    Returns:
        PSPFReader: Reader instance
    """
    mode = ACCESS_MMAP if use_mmap else ACCESS_AUTO
    reader = PSPFReader(bundle_path, mode)
    reader.open()
    return reader

verify_bundle ¶

verify_bundle(bundle_path: Path) -> bool

Verify a bundle's integrity.

Parameters:

Name	Type	Description	Default
`bundle_path`	`Path`	Path to bundle	required

Returns:

Name	Type	Description
`bool`	`bool`	True if bundle is valid

Source code in flavor/psp/format_2025/reader.py

def verify_bundle(bundle_path: Path) -> bool:
    """Verify a bundle's integrity.

    Args:
        bundle_path: Path to bundle

    Returns:
        bool: True if bundle is valid
    """
    with PSPFReader(bundle_path, ACCESS_MMAP) as reader:
        # Check magic
        if not reader.verify_magic_trailer():
            logger.error("❌ Invalid magic ending")
            return False

        # Check index
        try:
            reader.read_index()
        except Exception as e:
            logger.error(f"❌ Invalid index: {e}")
            return False

        # Check all checksums
        if not reader.verify_all_checksums():
            return False

        # Check signature if present
        try:
            if not reader.verify_signature():
                logger.error("❌ Signature verification failed")
                return False
        except Exception as e:
            logger.warning(f"⚠️ Signature verification skipped: {e}")

        return True

reader

flavor.psp.format_2025.reader ¶

Classes¶

PSPFReader ¶

Functions¶

__del__ ¶

__enter__ ¶

__exit__ ¶

close ¶

extract_slot ¶

extraction_lock ¶

get_backend ¶

get_slot_view ¶

open ¶

read_index ¶

read_magic_trailer ¶

read_metadata ¶

read_slot ¶

read_slot_descriptors ¶

stream_slot ¶

use_mmap ¶

use_streaming ¶

verify_all_checksums ¶

verify_attestation_policy_hash ¶

verify_integrity ¶

verify_magic_trailer ¶

verify_signature ¶

verify_slot_integrity ¶

Functions¶

read_bundle ¶

verify_bundle ¶

del ¶

enter ¶

exit ¶