iso-scalpel/iso_scalpel/iso_handler.py

839 lines
34 KiB
Python

"""High-level wrapper around :mod:`pycdlib`.
This module is the bridge between the GUI and the pycdlib library. It
exposes a single :class:`IsoHandler` that knows how to:
* create / open / save ISO images (ISO9660 levels 1-3 with optional
Joliet, Rock Ridge and UDF extensions),
* list directory contents in any of the supported naming conventions,
* add files and directories from the host filesystem into the image
(transparently writing them into every enabled convention),
* remove and rename entries,
* extract files and whole directory trees back to the host,
* inspect and edit volume-descriptor metadata (label, publisher, ...),
* inspect and configure El Torito boot records.
The handler is GUI-agnostic: it never imports PySide6, so it can be
unit-tested headlessly (and is, via the dev harness in ``tests``).
"""
# This file is part of ISO Scalpel.
# Copyright (C) 2025 Jeremy Anderson <info@dcos.net>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, see <https://www.gnu.org/licenses/>
# or write to the Free Software Foundation, Inc., 51 Franklin Street,
# Fifth Floor, Boston, MA 02110-1301 USA.
from __future__ import annotations
import contextlib
import os
import sys
from collections.abc import Callable
from dataclasses import dataclass, field
import pycdlib
from pycdlib.pycdlibexception import PyCdlibException
from .iso_record import IsoRecord, NameType, _decode
# Exception tuple covering every failure mode a pycdlib operation can raise
# that we treat as "the image is malformed / the entry is missing" rather
# than a programming error. Used by the defensive wrappers below so we
# never silence a real bug (e.g. AttributeError, TypeError) along with the
# expected I/O and parse errors.
_PYCDLIB_ERRORS: tuple[type[BaseException], ...] = (
PyCdlibException,
OSError,
ValueError,
KeyError,
)
# --------------------------------------------------------------------------
# Public dataclasses describing image metadata
# --------------------------------------------------------------------------
@dataclass
class NewIsoOptions:
"""Options for :meth:`IsoHandler.new`."""
volume_label: str = "CDROM"
interchange_level: int = 1 # 1, 2 or 3
block_size: int = 2048
joliet: int | None = None # None / 1 / 2 / 3
rock_ridge: str | None = None # None / '1.09' / '1.12'
udf: str | None = None # None / '2.50' / '2.60'
publisher: str = ""
preparer: str = ""
application: str = "ISO Scalpel"
system_id: str = ""
volume_set_id: str = " "
copyright_file: str = ""
abstract_file: str = ""
bibliographic_file: str = ""
xa: bool = False
@dataclass
class VolumeProperties:
"""Snapshot of the volume descriptors for the properties dialog."""
volume_label: str = ""
system_id: str = ""
volume_set_id: str = ""
publisher: str = ""
preparer: str = ""
application: str = ""
copyright_file: str = ""
abstract_file: str = ""
bibliographic_file: str = ""
interchange_level: int = 1
block_size: int = 2048
has_joliet: bool = False
has_rock_ridge: bool = False
has_udf: bool = False
joliet_level: int | None = None
rock_ridge_version: str | None = None
udf_version: str | None = None
total_size: int = 0
@property
def extensions(self) -> list[str]:
"""Human-readable extension labels (e.g. ``["Joliet 3", "UDF 2.60"]``).
Centralised so the CLI, the Properties dialog, and the status bar all
report the same string instead of each re-implementing the if-chain.
"""
rows = (
(self.has_joliet, f"Joliet {self.joliet_level or ''}".strip()),
(self.has_rock_ridge, f"Rock Ridge {self.rock_ridge_version or ''}".strip()),
(self.has_udf, f"UDF {self.udf_version or ''}".strip()),
)
return [label for enabled, label in rows if enabled]
@dataclass
class BootInfo:
"""El Torito boot configuration."""
bootable: bool = True
boot_image_path: str = "" # path inside the ISO of the boot file
boot_catalog_path: str = "BOOT.CAT;1"
platform_id: int = 0 # 0=x86, 1=PowerPC, 2=Mac, 0xEF=EFI
media_name: str = "noemul" # noemul / 1200 / 1440 / 2880 / harddisk
load_size: int | None = None # sectors; None = auto (whole file)
load_segment: int = 0x07C0
boot_info_table: bool = False
efi: bool = False
enabled: bool = False
# --------------------------------------------------------------------------
# Internal directory tree
# --------------------------------------------------------------------------
@dataclass
class _DirNode:
"""An in-memory directory node tracking its path in every convention.
pycdlib requires the *full path* in each enabled convention when adding
an entry (e.g. both ``iso_path`` and ``joliet_path``). Because the
spelling of a directory name differs between conventions (ISO9660 is
upper-case 8.3, Joliet/UDF preserve case), we cannot derive one from
the other reliably. Instead we remember, per directory, the exact path
in each convention so that subsequent add operations are exact.
"""
name: str # display name (primary convention)
iso_path: str # always set (ISO9660 path)
joliet_path: str | None = None
udf_path: str | None = None
rr_name: str | None = None # rock-ridge *relative* name
children: dict = field(default_factory=dict) # name -> _DirNode
loaded: bool = False
# --------------------------------------------------------------------------
# Name mangling helpers
# --------------------------------------------------------------------------
_ISO9660_CHARS = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_")
# Map a NameType to the pycdlib keyword that selects its directory tree.
# Rock Ridge has no dedicated kwarg -- it lives on the ISO9660 tree and is
# disambiguated by ``rr_name`` on add operations, so navigation falls
# through to ``iso_path`` like plain ISO9660.
_PATH_KWARG: dict[NameType, str] = {
NameType.ISO9660: "iso_path",
NameType.JOLIET: "joliet_path",
NameType.UDF: "udf_path",
NameType.ROCK_RIDGE: "iso_path",
}
def mangle_iso9660_name(name: str, is_dir: bool = False, interchange_level: int = 1) -> str:
"""Convert an arbitrary filename into a valid ISO9660 identifier.
Level 1 enforces 8.3 with a restricted character set; levels 2 and 3
allow longer names (30 chars) but keep the same charset and upper case.
A ``;1`` version suffix is appended to file identifiers.
"""
base, ext = os.path.splitext(name)
base = base.upper()
ext = ext.upper().lstrip(".")
# sanitise charset
base = "".join(c if c in _ISO9660_CHARS else "_" for c in base)
ext = "".join(c if c in _ISO9660_CHARS else "_" for c in ext)
if interchange_level <= 1:
base = base[:8] or "_"
ext = ext[:3]
else:
base = base[:30] or "_"
ext = ext[:30]
if is_dir:
return base
if ext:
return f"{base}.{ext};1"
return f"{base};1"
def _safe_join(parent: str, child: str) -> str:
"""Join an ISO-style path (always absolute, '/' separated)."""
if parent == "/":
return f"/{child}"
return f"{parent}/{child}"
# --------------------------------------------------------------------------
# Main handler
# --------------------------------------------------------------------------
class IsoHandler:
"""Stateful wrapper around a single :class:`pycdlib.PyCdlib` instance."""
def __init__(self) -> None:
self._iso: pycdlib.PyCdlib | None = None
self._path: str | None = None # on-disk filename
self._dirty: bool = False
self._options: NewIsoOptions = NewIsoOptions()
self._boot: BootInfo = BootInfo()
self._tree: _DirNode | None = None
# progress callback: (op:str, current:int, total:int) -> None
self.progress_cb: Callable[[str, int, int], None] | None = None
self._cancel = False
# ------------------------------------------------------------------ state
@property
def is_open(self) -> bool:
return self._iso is not None
@property
def is_dirty(self) -> bool:
return self._dirty
@property
def filename(self) -> str | None:
return self._path
@property
def has_joliet(self) -> bool:
return bool(self._iso and self._iso.has_joliet())
@property
def has_rock_ridge(self) -> bool:
return bool(self._iso and self._iso.has_rock_ridge())
@property
def has_udf(self) -> bool:
return bool(self._iso and self._iso.has_udf())
def cancel(self) -> None:
"""Request cancellation of a long-running operation."""
self._cancel = True
def reset_cancel(self) -> None:
self._cancel = False
# ------------------------------------------------------------------ create
def new(self, options: NewIsoOptions) -> None:
"""Create a fresh, empty ISO image."""
if self._iso is not None:
self.close()
iso = pycdlib.PyCdlib()
iso.new(
interchange_level=options.interchange_level,
sys_ident=options.system_id,
vol_ident=options.volume_label or "CDROM",
set_size=1,
seqnum=1,
log_block_size=options.block_size,
vol_set_ident=options.volume_set_id or " ",
pub_ident_str=options.publisher,
preparer_ident_str=options.preparer,
app_ident_str=options.application or "ISO Scalpel",
copyright_file=options.copyright_file,
abstract_file=options.abstract_file,
bibli_file=options.bibliographic_file,
vol_expire_date=None,
app_use="",
joliet=options.joliet,
rock_ridge=options.rock_ridge,
xa=options.xa,
udf=options.udf,
)
self._iso = iso
self._options = options
self._path = None
self._dirty = True
self._boot = BootInfo()
self._tree = _DirNode(name="/", iso_path="/", joliet_path="/", udf_path="/", rr_name=None)
self._tree.loaded = True
# ------------------------------------------------------------------ open
def open(self, filename: str) -> None:
"""Open an existing ISO image from ``filename``."""
if self._iso is not None:
self.close()
iso = pycdlib.PyCdlib()
iso.open(filename, mode="rb")
self._iso = iso
self._path = os.path.abspath(filename)
self._dirty = False
self._options = NewIsoOptions(volume_label=self._read_vol_ident() or "CDROM")
self._boot = self._read_boot_info()
# Build a fresh (lazy) tree rooted at "/".
self._tree = _DirNode(
name="/",
iso_path="/",
joliet_path="/" if self.has_joliet else None,
udf_path="/" if self.has_udf else None,
)
self._tree.loaded = False
# ------------------------------------------------------------------ save
def save(self, filename: str | None = None) -> None:
"""Write the image to ``filename`` (or to the previously-opened file)."""
if self._iso is None:
raise RuntimeError("No image is open")
target = filename or self._path
if not target:
raise ValueError("No filename supplied")
self._iso.write(target, progress_cb=self._pycdlib_progress)
self._path = os.path.abspath(target)
self._dirty = False
# ------------------------------------------------------------------ close
def close(self) -> None:
if self._iso is not None:
try:
self._iso.close()
except _PYCDLIB_ERRORS as exc:
# The image handle is discarded regardless; log the close
# failure so a corrupt write surfaces instead of vanishing.
sys.stderr.write(f"warning: pycdlib close failed: {exc}\n")
self._iso = None
self._path = None
self._dirty = False
self._tree = None
self._boot = BootInfo()
# ------------------------------------------------------------------ nav
# Conventions are always probed richest-first so the UI lists UDF before
# Rock Ridge before Joliet before plain ISO9660. A single ordered table
# drives both available_name_types() and default_name_type().
_NAME_TYPE_PROBES: tuple[tuple[NameType, str], ...] = (
(NameType.UDF, "has_udf"),
(NameType.ROCK_RIDGE, "has_rock_ridge"),
(NameType.JOLIET, "has_joliet"),
)
def available_name_types(self) -> list[NameType]:
"""Naming conventions present in this image (ISO9660 always first)."""
extras = [nt for nt, attr in self._NAME_TYPE_PROBES if getattr(self, attr)]
return [NameType.ISO9660, *extras]
def default_name_type(self) -> NameType:
"""Preferred convention for display (richest available first)."""
for nt, attr in self._NAME_TYPE_PROBES:
if getattr(self, attr):
return nt
return NameType.ISO9660
def list_dir(self, path: str, name_type: NameType) -> list[IsoRecord]:
"""Return the (sorted) children of ``path`` in the given convention."""
if self._iso is None:
return []
kw = self._path_kwarg(path, name_type)
# Comprehension over the live pycdlib iterator, skipping the
# ``.``/``..`` self/references that every ISO directory carries.
records = [
IsoRecord.from_pycdlib(child, _safe_join(path, name), name_type)
for child in self._iso.list_children(**kw)
if child is not None
for ident in (child.file_identifier(),)
if ident not in (b".", b"..")
for name in (_decode(ident),)
if name not in (".", "..")
]
# Directories first, then alphabetical (case-insensitive).
records.sort(key=lambda r: (not r.is_dir, r.name.lower()))
return records
# ------------------------------------------------------------------ add
def add_file(self, local_path: str, dest_dir: str, name_type: NameType,
nice_name: str | None = None) -> None:
"""Add a host file into ``dest_dir`` of the image."""
if self._iso is None:
raise RuntimeError("No image is open")
nice_name = nice_name or os.path.basename(local_path)
node = self._ensure_node(dest_dir, name_type)
iso_name = self._unique_iso9660_name(nice_name, node, is_dir=False)
kwargs = {"filename": local_path, "iso_path": _safe_join(node.iso_path, iso_name)}
if self.has_rock_ridge:
kwargs["rr_name"] = nice_name
if self.has_joliet and node.joliet_path is not None:
kwargs["joliet_path"] = _safe_join(node.joliet_path, nice_name)
if self.has_udf and node.udf_path is not None:
kwargs["udf_path"] = _safe_join(node.udf_path, nice_name)
self._iso.add_file(**kwargs)
self._mark_dirty()
# Invalidate children cache for the parent.
node.children.clear()
node.loaded = False
def add_directory(self, dest_dir: str, name_type: NameType,
nice_name: str) -> str:
"""Create a new directory inside ``dest_dir``; returns its primary path."""
if self._iso is None:
raise RuntimeError("No image is open")
node = self._ensure_node(dest_dir, name_type)
nice_name = self._unique_nice_name(nice_name, node)
iso_name = self._unique_iso9660_name(nice_name, node, is_dir=True)
kwargs = {"iso_path": _safe_join(node.iso_path, iso_name)}
if self.has_rock_ridge:
kwargs["rr_name"] = nice_name
if self.has_joliet and node.joliet_path is not None:
kwargs["joliet_path"] = _safe_join(node.joliet_path, nice_name)
if self.has_udf and node.udf_path is not None:
kwargs["udf_path"] = _safe_join(node.udf_path, nice_name)
self._iso.add_directory(**kwargs)
# register the new directory in our tree
child = _DirNode(
name=nice_name,
iso_path=_safe_join(node.iso_path, iso_name),
joliet_path=_safe_join(node.joliet_path, nice_name) if node.joliet_path else None,
udf_path=_safe_join(node.udf_path, nice_name) if node.udf_path else None,
rr_name=nice_name if self.has_rock_ridge else None,
loaded=True,
)
node.children[nice_name] = child
self._mark_dirty()
return child.name
# ------------------------------------------------------------------ remove
def remove(self, path: str, name_type: NameType, is_dir: bool) -> None:
"""Remove a file or directory from the image."""
if self._iso is None:
raise RuntimeError("No image is open")
kw = self._path_kwarg(path, name_type)
if is_dir:
self._iso.rm_directory(**kw)
else:
self._iso.rm_file(**kw)
# drop from tree
parent_path, _, leaf = path.rpartition("/")
node = self._find_node(parent_path or "/", name_type)
if node is not None:
node.children.pop(leaf, None)
self._mark_dirty()
# ------------------------------------------------------------------ rename
def rename(self, path: str, name_type: NameType, new_name: str,
is_dir: bool) -> str:
"""Rename an entry.
pycdlib has no direct rename, so we remove and re-add the entry
(copying file data through a temporary on the host). For
directories we recurse.
"""
import tempfile
if self._iso is None:
raise RuntimeError("No image is open")
parent_path = path.rpartition("/")[0] or "/"
with tempfile.TemporaryDirectory() as tmp:
if is_dir:
local_dir = os.path.join(tmp, new_name)
self.extract_dir(path, name_type, local_dir)
self.remove(path, name_type, is_dir=True)
self.add_directory(parent_path, name_type, new_name)
# Re-import the extracted subtree under the new name.
self._import_tree(local_dir, _safe_join(parent_path, new_name), name_type)
else:
local_file = os.path.join(tmp, new_name)
self.extract_file(path, name_type, local_file)
self.remove(path, name_type, is_dir=False)
self.add_file(local_file, parent_path, name_type, nice_name=new_name)
self._mark_dirty()
return new_name
# ------------------------------------------------------------------ extract
def extract_file(self, iso_path: str, name_type: NameType,
local_path: str) -> None:
"""Extract a single file to ``local_path``."""
if self._iso is None:
raise RuntimeError("No image is open")
kw = self._path_kwarg(iso_path, name_type)
os.makedirs(os.path.dirname(os.path.abspath(local_path)), exist_ok=True)
self._iso.get_file_from_iso(local_path, **kw)
def extract_dir(self, iso_path: str, name_type: NameType,
local_dir: str) -> None:
"""Recursively extract a directory tree to ``local_dir``."""
if self._iso is None:
raise RuntimeError("No image is open")
os.makedirs(local_dir, exist_ok=True)
for rec in self.list_dir(iso_path, name_type):
dest = os.path.join(local_dir, rec.name)
if rec.is_dir:
self.extract_dir(rec.path, name_type, dest)
else:
self.extract_file(rec.path, name_type, dest)
# ------------------------------------------------------------------ props
def get_properties(self) -> VolumeProperties:
"""Read the current volume descriptors into a snapshot."""
if self._iso is None:
return VolumeProperties()
p = VolumeProperties(
has_joliet=self.has_joliet,
has_rock_ridge=self.has_rock_ridge,
has_udf=self.has_udf,
interchange_level=self._options.interchange_level,
block_size=self._options.block_size,
total_size=self._image_size(),
)
# pycdlib exposes the primary volume descriptor (PVD). Some
# identifier fields are ``FileOrTextIdentifier`` objects (which
# expose ``.text`` -> bytes); others are plain bytes.
def _id(val) -> str:
if val is None:
return ""
if hasattr(val, "text"):
val = val.text
return _decode(val).strip()
try:
pvd = self._iso.pvd
p.volume_label = _id(getattr(pvd, "volume_identifier", b""))
p.system_id = _id(getattr(pvd, "system_identifier", b""))
p.volume_set_id = _id(getattr(pvd, "volume_set_identifier", b""))
p.publisher = _id(getattr(pvd, "publisher_identifier", b""))
p.preparer = _id(getattr(pvd, "preparer_identifier", b""))
p.application = _id(getattr(pvd, "application_identifier", b""))
p.copyright_file = _id(getattr(pvd, "copyright_file_identifier", b""))
p.abstract_file = _id(getattr(pvd, "abstract_file_identifier", b""))
p.bibliographic_file = _id(getattr(pvd, "bibliographic_file_identifier", b""))
except _PYCDLIB_ERRORS:
# A minimal or damaged image may have no readable PVD; the
# default empty strings remain.
pass
if self.has_joliet:
p.joliet_level = 3
if self.has_rock_ridge:
p.rock_ridge_version = self._options.rock_ridge or "1.09"
if self.has_udf:
p.udf_version = self._options.udf or "2.60"
return p
def set_volume_label(self, label: str) -> None:
"""Update the volume identifier on the primary descriptor."""
if self._iso is None:
return
label = (label or "").upper()[:32]
try:
self._iso.pvd.volume_identifier = label.encode("ascii", "replace")
except _PYCDLIB_ERRORS as exc:
sys.stderr.write(f"warning: could not set volume label: {exc}\n")
self._options.volume_label = label
self._mark_dirty()
# ------------------------------------------------------------------ boot
def get_boot_info(self) -> BootInfo:
return self._boot
def set_boot(self, info: BootInfo, boot_file_local: str | None) -> None:
"""Configure El Torito boot.
``boot_file_local`` is a host path to the boot image; it will be
added into the ISO first, then referenced by the boot catalog.
"""
if self._iso is None:
raise RuntimeError("No image is open")
if not boot_file_local and not info.boot_image_path:
raise ValueError("A boot image is required")
# add the boot file into the image if a local file was given
if boot_file_local:
boot_iso_name = mangle_iso9660_name(os.path.basename(boot_file_local), is_dir=False)
self.add_file(boot_file_local, "/", self.default_name_type(),
nice_name=os.path.basename(boot_file_local))
info.boot_image_path = f"/{boot_iso_name}"
kwargs = {
"bootfile_path": info.boot_image_path,
"bootcatfile": self._ensure_abs(info.boot_catalog_path or "BOOT.CAT;1"),
"platform_id": info.platform_id,
"boot_info_table": info.boot_info_table,
"efi": info.efi,
"media_name": info.media_name,
"bootable": info.bootable,
"boot_load_seg": info.load_segment,
}
if info.load_size is not None:
kwargs["boot_load_size"] = info.load_size
# Joliet / UDF / Rock-Ridge boot-catalog names derive from the ISO
# catalog name, lower-cased and stripped of version suffix.
cat_iso = info.boot_catalog_path or "BOOT.CAT;1"
cat_stem = cat_iso.lstrip("/").split(";")[0] or "boot.cat"
bootcat = cat_stem.lower()
# Dispatch table: each enabled extension gets its convention-specific
# catalog path key.
ext_cat_keys = (
(self.has_rock_ridge, "rr_bootcatname", bootcat),
(self.has_joliet, "joliet_bootcatfile", "/" + bootcat),
(self.has_udf, "udf_bootcatfile", "/" + bootcat),
)
for enabled, key, value in ext_cat_keys:
if enabled:
kwargs[key] = value
self._iso.add_eltorito(**kwargs)
info.enabled = True
self._boot = info
self._mark_dirty()
def clear_boot(self) -> None:
"""Remove an existing El Torito boot configuration."""
if self._iso is None:
return
try:
self._iso.rm_eltorito()
except _PYCDLIB_ERRORS as exc:
# No boot record present, or pycdlib refused -- either way the
# caller wants a clean slate; surface the reason on stderr.
sys.stderr.write(f"warning: rm_eltorito failed: {exc}\n")
self._boot = BootInfo()
self._mark_dirty()
# ------------------------------------------------------------------ internals
def _path_kwarg(self, path: str, name_type: NameType) -> dict[str, str]:
"""Translate (path, name_type) into the right pycdlib keyword."""
return {_PATH_KWARG[name_type]: path}
def _ensure_node(self, dir_path: str, name_type: NameType) -> _DirNode:
"""Return the :class:`_DirNode` for ``dir_path``, loading if needed."""
if self._tree is None:
raise RuntimeError("No tree")
if dir_path in ("/", ""):
self._load_node(self._tree, name_type)
return self._tree
# walk the tree, loading lazily
parts = [p for p in dir_path.split("/") if p]
node = self._tree
for part in parts:
self._load_node(node, name_type)
if part not in node.children:
# not in our cache: try to resolve by listing & matching
node = self._resolve_child(node, part, name_type)
else:
node = node.children[part]
return node
def _load_node(self, node: _DirNode, name_type: NameType) -> None:
"""Populate ``node.children`` from the image (once)."""
if node.loaded:
return
path = self._node_primary_path(node, name_type)
for rec in self.list_dir(path, name_type):
if not rec.is_dir:
continue
child = _DirNode(
name=rec.name,
iso_path=self._resolve_iso_path(node, rec),
joliet_path=self._resolve_joliet_path(node, rec),
udf_path=self._resolve_udf_path(node, rec),
rr_name=rec.name if self.has_rock_ridge else None,
)
node.children[rec.name] = child
node.loaded = True
def _node_primary_path(self, node: _DirNode, name_type: NameType) -> str:
"""Pick the convention-specific path stored on ``node`` for ``name_type``.
Falls back to the ISO9660 path when the requested convention's path
was never recorded (e.g. a Joliet-only entry probed via ISO9660).
"""
attr = {
NameType.JOLIET: "joliet_path",
NameType.UDF: "udf_path",
}.get(name_type)
if attr is not None:
convention_path = getattr(node, attr)
if convention_path:
return convention_path
return node.iso_path
def _resolve_child(self, node: _DirNode, part: str, name_type: NameType) -> _DirNode:
"""Fallback when a child isn't cached: locate it by listing."""
self._load_node(node, name_type)
if part in node.children:
return node.children[part]
raise KeyError(f"Directory '{part}' not found in '{node.name}'")
def _resolve_iso_path(self, parent: _DirNode, rec: IsoRecord) -> str:
"""Best-effort ISO9660 path for a child listed in another convention."""
if rec.name_type == NameType.ISO9660:
return rec.path
return _safe_join(parent.iso_path, mangle_iso9660_name(rec.name, is_dir=True))
def _resolve_joliet_path(self, parent: _DirNode, rec: IsoRecord) -> str | None:
if not self.has_joliet:
return None
if rec.name_type == NameType.JOLIET:
return rec.path
return _safe_join(parent.joliet_path or "/", rec.name) if parent.joliet_path else None
def _resolve_udf_path(self, parent: _DirNode, rec: IsoRecord) -> str | None:
if not self.has_udf:
return None
if rec.name_type == NameType.UDF:
return rec.path
return _safe_join(parent.udf_path or "/", rec.name) if parent.udf_path else None
def _find_node(self, dir_path: str, name_type: NameType) -> _DirNode | None:
try:
return self._ensure_node(dir_path, name_type)
except KeyError:
return None
def _unique_iso9660_name(self, nice_name: str, node: _DirNode,
is_dir: bool) -> str:
"""Generate a collision-free ISO9660 identifier for a new entry."""
base = mangle_iso9660_name(nice_name, is_dir=is_dir,
interchange_level=self._options.interchange_level)
# Collect existing ISO9660 names so the new one does not collide.
try:
existing = {rec.raw_name.split(";")[0]
for rec in self.list_dir(node.iso_path, NameType.ISO9660)}
except _PYCDLIB_ERRORS:
existing = set()
candidate = base
cand_base = candidate.split(";")[0]
n = 1
while cand_base in existing:
n += 1
if is_dir:
candidate = f"{base[:7]}_{n}"
else:
stem = base.split(";")[0]
if "." in stem:
s, e = stem.rsplit(".", 1)
candidate = f"{s[:6]}_{n}.{e};1"
else:
candidate = f"{stem[:7]}_{n};1"
cand_base = candidate.split(";")[0]
return candidate
def _unique_nice_name(self, nice_name: str, node: _DirNode) -> str:
try:
existing = {rec.name.lower()
for rec in self.list_dir(
self._node_primary_path(node, self.default_name_type()),
self.default_name_type())}
except _PYCDLIB_ERRORS:
existing = set()
candidate = nice_name
n = 1
stem, ext = os.path.splitext(nice_name)
while candidate.lower() in existing:
n += 1
candidate = f"{stem}_{n}{ext}"
return candidate
def _import_tree(self, local_dir: str, iso_dir: str,
name_type: NameType) -> None:
"""Recursively import a host directory tree into ``iso_dir``."""
for entry in sorted(os.listdir(local_dir)):
full = os.path.join(local_dir, entry)
if os.path.isdir(full):
new_dir = self.add_directory(iso_dir, name_type, entry)
self._import_tree(full, _safe_join(iso_dir, new_dir), name_type)
else:
self.add_file(full, iso_dir, name_type, nice_name=entry)
def _read_vol_ident(self) -> str:
try:
return _decode(self._iso.pvd.volume_identifier).strip()
except _PYCDLIB_ERRORS:
return ""
def _read_boot_info(self) -> BootInfo:
"""Probe whether the open image carries an El Torito boot catalog.
``pycdlib.PyCdlib.eltorito_boot_catalog`` returns ``None`` when no
boot record is present (it does not raise), so the check is a plain
truthiness test rather than a try/except existence probe.
"""
info = BootInfo()
try:
info.enabled = self._iso.eltorito_boot_catalog is not None
except _PYCDLIB_ERRORS:
info.enabled = False
return info
def _image_size(self) -> int:
"""Total image size in bytes = ``space_size`` blocks * block size.
``space_size`` is an int attribute on the PVD; ``logical_block_size``
is a *method* on the PVD (pycdlib's API is inconsistent here). The
fallback to a 2048-byte block covers images where the PVD is
unreadable but ``space_size`` survived.
"""
try:
return int(self._iso.pvd.space_size) * int(self._iso.pvd.logical_block_size())
except _PYCDLIB_ERRORS:
try:
return int(self._iso.pvd.space_size) * 2048
except _PYCDLIB_ERRORS:
return 0
def _mark_dirty(self) -> None:
self._dirty = True
@staticmethod
def _ensure_abs(path: str) -> str:
"""Ensure an ISO-style path starts with '/'."""
if not path:
return "/"
return path if path.startswith("/") else "/" + path
def _pycdlib_progress(self, done, total, *args) -> None:
if self.progress_cb:
# A non-numeric or missing progress value is not fatal; suppress
# the conversion error so the write continues uninterrupted.
with contextlib.suppress(TypeError, ValueError):
self.progress_cb("write", int(done), int(total))