fix(music): keep detached audio readable on ACL mounts (#6961)

This commit is contained in:
jxxghp
2026-10-11 13:24:29 +08:00
parent 3fc4a5cb5c
commit 453a3479aa
2 changed files with 49 additions and 3 deletions

View File

@@ -7,7 +7,7 @@ from contextvars import ContextVar
from copy import deepcopy
from dataclasses import asdict, dataclass, field
from pathlib import Path, PurePosixPath, PureWindowsPath
from stat import S_IWUSR
from stat import S_IRUSR, S_IWUSR
from tempfile import NamedTemporaryFile
from threading import RLock
from typing import Any, Iterator, Optional, Union
@@ -119,6 +119,7 @@ def _audio_write_target(path: Path) -> Iterator[Path]:
不解析并改写符号链接指向的源文件。临时文件与目标同目录,以便原子替换;
任一写入失败或期间目标发生变化时保留原链接,并清理未提交副本。
独立副本至少保留所有者读写权限,避免 NAS ACL 挂载映射的 mode 产生不可读文件。
"""
stat = path.stat()
link_inode = path.lstat().st_ino
@@ -130,8 +131,8 @@ def _audio_write_target(path: Path) -> Iterator[Path]:
temporary = Path(stream.name)
try:
shutil.copy2(path, temporary)
# 做种源可以只读;只给独立副本增加写权限,源权限保持不变。
temporary.chmod(stat.st_mode | S_IWUSR)
# NAS ACL 挂载可将可读源文件映射为 000;只给独立副本补齐读写权限,源权限保持不变。
temporary.chmod(stat.st_mode | S_IRUSR | S_IWUSR)
initial_copy = _audio_signature(temporary)
yield temporary
if _audio_signature(temporary) == initial_copy:

View File

@@ -164,6 +164,51 @@ def test_read_only_seed_permissions_are_preserved(tmp_path):
assert FLAC(target)["title"] == ["New"]
@pytest.mark.parametrize("visible_mode", [0o000, 0o200, 0o040, 0o444, 0o640, 0o777])
@pytest.mark.parametrize("link_type", ["hardlink", "symlink"])
@pytest.mark.parametrize("container", ["flac", "m4a"])
@pytest.mark.parametrize("write_tags", [False, True])
def test_mapped_seed_permissions_leave_detached_audio_readable(
tmp_path, monkeypatch, visible_mode, link_type, container, write_tags):
"""模拟挂载层映射的权限,标签或封面写入后的副本至少可读写且不改源权限。"""
source = _audio_file(tmp_path, container)
source.chmod(0o640)
target = tmp_path / f"library.{container}"
os.link(source, target) if link_type == "hardlink" else target.symlink_to(source)
before = source.read_bytes()
original_stat = Path.stat
source_stat = source.stat()
def mapped_stat(path, *args, **kwargs):
"""只映射源 inode 的可见 mode,实际读取仍由宿主权限模拟 NAS ACL 放行。"""
result = original_stat(path, *args, **kwargs)
if (result.st_dev, result.st_ino) == (source_stat.st_dev, source_stat.st_ino):
values = {name: getattr(result, name) for name in dir(result) if name.startswith("st_")}
values["st_mode"] = (result.st_mode & ~0o777) | visible_mode
return SimpleNamespace(**values)
return result
monkeypatch.setattr(Path, "stat", mapped_stat)
assert AudioMetadataHelper.write(
target, MusicInfo(title="New"), write_tags=write_tags,
cover_data=None if write_tags else _COVER, cover_mime="image/png")
assert source.read_bytes() == before
assert original_stat(source).st_mode & 0o777 == 0o640
assert target.stat().st_mode & 0o777 == visible_mode | 0o600
assert not target.is_symlink() and not source.samefile(target)
with target.open("rb") as stream:
audio = MutagenFile(fileobj=stream, filename="audio")
if write_tags:
assert AudioMetadataHelper.read_tags(target).title == "New"
elif isinstance(audio, FLAC):
assert audio.pictures[0].data == _COVER
else:
assert bytes(audio.tags["covr"][0]) == _COVER
assert not list(tmp_path.glob(".*.mp-audio-partial"))
def test_concurrent_destination_change_cancels_replacement(tmp_path, monkeypatch):
"""目标被另一整理过程替换时不提交过时副本,也不覆盖另一过程的新内容。"""
source = _audio_file(tmp_path, "flac")