mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-23 12:19:13 -05:00
Abdessamad Derraz
b8baf92f1d
mkstemp creates its scratch 0600 and os.replace kept that mode, so database.json, the manifests, the packs, the recipe snapshots and the large-file cache were readable by their owner only. set_plain_mode gives the scratch 0o666 less the umask, as open() would, before each rename. A test walks every function that renames a mkstemp file.
219 lines
8.7 KiB
Python
219 lines
8.7 KiB
Python
"""Generated artifacts are written whole or not at all.
|
|
|
|
write_if_changed got the scratch-and-rename treatment, but the install
|
|
manifests, the target manifests and release.json were still truncated at
|
|
their final name, and restore_large_files copied an asset straight to its
|
|
path in bios/: a scan running beside the copy hashed the truncated file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import artifacts # noqa: E402
|
|
|
|
|
|
class WholeOrNothing(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.dir = Path(self._tmp.name)
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def test_an_interrupted_write_keeps_the_previous_file(self):
|
|
target = self.dir / "manifest.json"
|
|
target.write_text("{}")
|
|
with mock.patch("os.replace", side_effect=OSError("cut")), self.assertRaises(OSError):
|
|
artifacts.write_text_atomic(str(target), "{\"files\": []}")
|
|
self.assertEqual(target.read_text(), "{}")
|
|
self.assertEqual(os.listdir(self.dir), ["manifest.json"])
|
|
|
|
def test_an_interrupted_copy_leaves_no_file(self):
|
|
source = self.dir / "asset.bin"
|
|
source.write_bytes(b"x" * 1000)
|
|
target = self.dir / "bios" / "asset.bin"
|
|
target.parent.mkdir()
|
|
with mock.patch("shutil.copy2", side_effect=OSError("cut")), self.assertRaises(OSError):
|
|
artifacts.copy_file_atomic(str(source), str(target))
|
|
self.assertEqual(os.listdir(target.parent), [])
|
|
|
|
|
|
class EveryWriterGoesThroughIt(unittest.TestCase):
|
|
def test_manifests_record_and_restore_use_the_atomic_helpers(self):
|
|
scripts = REPO_ROOT / "scripts"
|
|
for name, helper in (
|
|
("generate_pack.py", "write_text_atomic("),
|
|
("release_record.py", "write_text_atomic("),
|
|
("restore_large_files.py", "copy_file_atomic("),
|
|
):
|
|
source = (scripts / name).read_text(encoding="utf-8")
|
|
with self.subTest(script=name):
|
|
self.assertIn(helper, source)
|
|
pack = (scripts / "generate_pack.py").read_text(encoding="utf-8")
|
|
self.assertNotIn('with open(path, "w") as f:\n f.write(new_json)', pack)
|
|
self.assertNotIn("json.dump(result, f", pack)
|
|
self.assertNotIn("args.output.write_text(", (scripts / "release_record.py").read_text())
|
|
self.assertNotIn("shutil.copy2(source, path)", (scripts / "restore_large_files.py").read_text())
|
|
|
|
|
|
def test_scrapers_write_whole_files(self):
|
|
"""A streamed yaml.dump into platforms/<x>.yml let a concurrent reader
|
|
parse a valid platform holding a fraction of its systems."""
|
|
import re
|
|
|
|
streamed = re.compile(
|
|
r"open\([^)]*['\"](?:w|wb|a)['\"]|\.write_text\(|\.write_bytes\("
|
|
)
|
|
offenders = [
|
|
f"{path.relative_to(REPO_ROOT)}:{number}"
|
|
for path in sorted((REPO_ROOT / "scripts" / "scraper").rglob("*.py"))
|
|
for number, line in enumerate(
|
|
path.read_text(encoding="utf-8").splitlines(), 1
|
|
)
|
|
if streamed.search(line)
|
|
]
|
|
self.assertEqual(offenders, [])
|
|
|
|
def test_the_text_is_utf8_whatever_the_locale(self):
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
target = Path(directory) / "out.yml"
|
|
artifacts.write_text_atomic(str(target), "name: \u00e9mulateur \u2014 \u30d5\n")
|
|
self.assertEqual(
|
|
target.read_bytes().decode("utf-8"), "name: \u00e9mulateur \u2014 \u30d5\n"
|
|
)
|
|
|
|
|
|
class TheNativeIndexKeepsEveryPlatform(unittest.TestCase):
|
|
"""Native originals refresh platform by platform, in parallel. Each run
|
|
rewrote the whole URL index it had read before downloading, and the last
|
|
writer erased the others' entries."""
|
|
|
|
def test_an_entry_recorded_during_a_download_survives(self):
|
|
import io
|
|
import json
|
|
|
|
import export_native
|
|
|
|
with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as directory:
|
|
root = Path(directory)
|
|
index = root / export_native.SOURCES_INDEX
|
|
|
|
class Response(io.BytesIO):
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def other_run_records_meanwhile(request, timeout):
|
|
index.write_text(json.dumps({"retrobat/file.json": "https://b"}))
|
|
return Response(b"payload")
|
|
|
|
with mock.patch.object(
|
|
export_native.urllib.request, "urlopen", other_run_records_meanwhile
|
|
):
|
|
export_native.fetch("https://a", root / "batocera" / "file", index)
|
|
recorded = json.loads(index.read_text())
|
|
self.assertEqual(recorded.get("retrobat/file.json"), "https://b")
|
|
self.assertEqual(recorded.get("batocera/file"), "https://a")
|
|
self.assertEqual((root / "batocera" / "file").read_bytes(), b"payload")
|
|
|
|
class ScratchFilesGetThePlainMode(unittest.TestCase):
|
|
"""mkstemp makes its file 0600 and the rename kept it: database.json,
|
|
the manifests and the packs came out readable by their owner only."""
|
|
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self._tmp.cleanup)
|
|
self.dir = Path(self._tmp.name)
|
|
previous = os.umask(0o022)
|
|
self.addCleanup(os.umask, previous)
|
|
|
|
def _mode(self, path: Path) -> int:
|
|
return path.stat().st_mode & 0o777
|
|
|
|
def test_text_and_bytes_writers(self):
|
|
artifacts.write_text_atomic(str(self.dir / "a.json"), "{}")
|
|
artifacts.write_bytes_atomic(str(self.dir / "b.bin"), b"x")
|
|
self.assertEqual(self._mode(self.dir / "a.json"), 0o644)
|
|
self.assertEqual(self._mode(self.dir / "b.bin"), 0o644)
|
|
|
|
def test_the_umask_decides(self):
|
|
os.umask(0o002)
|
|
artifacts.write_text_atomic(str(self.dir / "a.json"), "{}")
|
|
self.assertEqual(self._mode(self.dir / "a.json"), 0o664)
|
|
|
|
def test_packs_and_their_manifest(self):
|
|
import zipfile # noqa: PLC0415
|
|
|
|
import generate_pack # noqa: PLC0415
|
|
|
|
pack = self.dir / "Pack.zip"
|
|
with generate_pack._writing_pack(str(pack)) as zf:
|
|
zf.writestr("a.bin", b"x")
|
|
self.assertEqual(self._mode(pack), 0o644)
|
|
os.chmod(pack, 0o600)
|
|
generate_pack.inject_manifest(str(pack), {"files": []})
|
|
self.assertEqual(self._mode(pack), 0o644)
|
|
with zipfile.ZipFile(pack) as zf:
|
|
self.assertIn("manifest.json", zf.namelist())
|
|
|
|
def test_every_renamed_scratch_gets_the_mode(self):
|
|
"""A function that renames a mkstemp file into place sets its mode.
|
|
|
|
copy_file_atomic is the one exception: copy2 gives the scratch the
|
|
source file's own mode.
|
|
"""
|
|
import ast # noqa: PLC0415
|
|
|
|
def name_of(node: ast.AST) -> str | None:
|
|
return node.id if isinstance(node, ast.Name) else None
|
|
|
|
def callee(call: ast.Call) -> str:
|
|
func = call.func
|
|
return func.attr if isinstance(func, ast.Attribute) else name_of(func) or ""
|
|
|
|
unset = []
|
|
for path in sorted((REPO_ROOT / "scripts").rglob("*.py")):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
for func in ast.walk(tree):
|
|
if not isinstance(func, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
continue
|
|
if func.name == "copy_file_atomic":
|
|
continue
|
|
scratches: set[str] = set()
|
|
for node in ast.walk(func):
|
|
if (isinstance(node, ast.Assign) and isinstance(node.value, ast.Call)
|
|
and callee(node.value) == "mkstemp"
|
|
and isinstance(node.targets[0], ast.Tuple)):
|
|
scratch = name_of(node.targets[0].elts[1])
|
|
if scratch:
|
|
scratches.add(scratch)
|
|
calls = [n for n in ast.walk(func) if isinstance(n, ast.Call)]
|
|
moded = {
|
|
name_of(c.args[0]) for c in calls
|
|
if callee(c) in ("set_plain_mode", "chmod") and c.args
|
|
}
|
|
for call in calls:
|
|
if callee(call) != "replace" or not call.args:
|
|
continue
|
|
first = name_of(call.args[0])
|
|
if first in scratches and first not in moded:
|
|
unset.append(f"{path.name}:{call.lineno} {func.name}")
|
|
self.assertEqual(unset, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|