Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 26 additions & 6 deletions src/pullbox/utilities/comicinfo.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,10 @@

from __future__ import annotations

import contextlib
import os
import shutil
import stat
import tempfile
import xml.etree.ElementTree as ET
import zipfile
Expand Down Expand Up @@ -278,6 +281,26 @@ def _find_comicinfo_member(source: zipfile.ZipFile) -> zipfile.ZipInfo | None:
return None


def _copy_permission_bits(reference: Path, target: Path) -> None:
"""Give a rewritten archive the permission bits of the file it stands in for.

Rewrites go through ``tempfile.mkstemp``, which creates the file as 0600. Left
alone, an archive that was readable by a media server before the rewrite is
owner-only afterwards. Best effort: filesystems that reject chmod, such as some
SMB/CIFS mounts, must not fail the rewrite, but the failure is logged so an
owner-only archive can be traced back to it.
"""
try:
os.chmod(target, stat.S_IMODE(reference.stat().st_mode))
except OSError as exc:
logger.warning(
"comicinfo_permission_bits_not_restored",
reference=str(reference),
target=str(target),
error=str(exc),
)


def embed_comicinfo_in_cbz(
cbz_path: Path,
data: dict[str, Any],
Expand Down Expand Up @@ -371,6 +394,7 @@ def embed_comicinfo_in_cbz(
)

# Atomic replace
_copy_permission_bits(cbz_path, tmp_path)
shutil.move(str(tmp_path), str(cbz_path))

logger.info(
Expand All @@ -388,9 +412,6 @@ def embed_comicinfo_in_cbz(
tmp_path.unlink()
raise
finally:
import contextlib
import os

# Close the file descriptor if still open
if tmp_fd >= 0:
with contextlib.suppress(OSError):
Expand Down Expand Up @@ -481,6 +502,8 @@ def materialize_cbz_with_comicinfo(
"entries",
)

# A plain move or copy would carry the source's mode to the target.
_copy_permission_bits(source_path, temp_output_path)
shutil.move(str(temp_output_path), str(target_path))
if transfer_method == "move" and source_path.resolve(strict=False) != target_path.resolve(
strict=False
Expand Down Expand Up @@ -516,9 +539,6 @@ def materialize_cbz_with_comicinfo(
temp_output_path.unlink()
raise
finally:
import contextlib
import os

if tmp_fd >= 0:
with contextlib.suppress(OSError):
os.close(tmp_fd)
Expand Down
77 changes: 77 additions & 0 deletions tests/unit/test_comicinfo_embed.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,9 @@
import zipfile
from typing import TYPE_CHECKING

import structlog
from structlog.testing import capture_logs

from pullbox.utilities.comicinfo import embed_comicinfo_in_cbz, materialize_cbz_with_comicinfo

if TYPE_CHECKING:
Expand Down Expand Up @@ -286,3 +289,77 @@ def test_materialize_cbz_with_comicinfo_move_deletes_source_after_success(
assert not source.exists()
assert target.exists()
assert _read_comicinfo(target)["Series"] == "Moved Series"


def _mode(path: Path) -> int:
return path.stat().st_mode & 0o777


def test_embed_comicinfo_keeps_the_archive_permission_bits(tmp_path: Path) -> None:
archive_path = tmp_path / "issue.cbz"
_write_cbz(archive_path, "<ComicInfo><Series>Old Series</Series></ComicInfo>")
archive_path.chmod(0o664)

assert embed_comicinfo_in_cbz(archive_path, {"Series": "New Series"}) is True

assert _read_comicinfo(archive_path)["Series"] == "New Series"
assert _mode(archive_path) == 0o664


def test_embed_comicinfo_keeps_permission_bits_with_caller_temp_path(tmp_path: Path) -> None:
archive_path = tmp_path / "issue.cbz"
_write_cbz(archive_path, "<ComicInfo><Series>Old Series</Series></ComicInfo>")
archive_path.chmod(0o600)

embed_comicinfo_in_cbz(
archive_path,
{"Series": "New Series"},
temp_path=tmp_path / "work" / "issue.tmp.cbz",
)

assert _mode(archive_path) == 0o600


def test_materialize_cbz_with_comicinfo_carries_source_permission_bits(tmp_path: Path) -> None:
source_path = tmp_path / "downloads" / "issue.cbz"
source_path.parent.mkdir()
_write_cbz(source_path, "<ComicInfo><Series>Old Series</Series></ComicInfo>")
source_path.chmod(0o644)
target_path = tmp_path / "library" / "Series" / "issue.cbz"

materialize_cbz_with_comicinfo(
source_path,
target_path,
{"Series": "New Series"},
transfer_method="copy",
)

assert _read_comicinfo(target_path)["Series"] == "New Series"
assert _mode(target_path) == 0o644


def test_embed_comicinfo_survives_a_filesystem_that_rejects_chmod(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
archive_path = tmp_path / "issue.cbz"
_write_cbz(archive_path, "<ComicInfo><Series>Old Series</Series></ComicInfo>")

def reject_chmod(*_args: object, **_kwargs: object) -> None:
raise PermissionError("chmod is not permitted on this mount")

monkeypatch.setattr("pullbox.utilities.comicinfo.os.chmod", reject_chmod)
# A logger cached before app reconfiguration would bypass capture_logs.
monkeypatch.setattr(
"pullbox.utilities.comicinfo.logger",
structlog.wrap_logger(None, cache_logger_on_first_use=False),
)

with capture_logs() as logs:
assert embed_comicinfo_in_cbz(archive_path, {"Series": "New Series"}) is True
assert _read_comicinfo(archive_path)["Series"] == "New Series"
warnings = [log for log in logs if log["event"] == "comicinfo_permission_bits_not_restored"]
assert len(warnings) == 1
assert warnings[0]["log_level"] == "warning"
assert warnings[0]["reference"] == str(archive_path)
assert "chmod is not permitted" in warnings[0]["error"]